zhouhui.jiang / Test_LangGraph

Go to a project
Toggle navigation Toggle navigation pinning
zhouhui.jiang
5a8f3de7 1 parent 4c0a2afc

update

Inline Side-by-side
Showing 3 changed files with 117 additions and 4 deletions
......@@ -13,7 +13,7 @@ API_CONFIG = {
"headers": {
"accept": "*/*",
"Content-Type": "application/json",
"Authorization": os.getenv("API_AUTHORIZATION", "Bearer 2.787c2f8b84f1462a83abba60fbb5553d"),
"Authorization": os.getenv("API_AUTHORIZATION", "Bearer 2.21e3deecca904c7697ab891b8276cf36"),
"Ver": os.getenv("API_VER", "033BD94B1168D7E4F0D644C3C95E35BF.D73E33B659AD1D6B7D181D1DF8D05760"),
"Referer": os.getenv("API_REFERER", "http://192.168.1.251/")
}
......
......@@ -133,6 +133,110 @@ def create_waybill_d(consignmentCode: str, Authorization: str = None) -> str:
return f"❌ 创建D类运单失败: {str(e)}"
def query_waybill_info(consignmentCode: str, Authorization: str = None) -> str:
"""根据运单编号查询运单信息
Args:
consignmentCode: 运单编号（必填）
Authorization: 授权令牌，如果提供则使用该值，否则使用 API_CONFIG 中的默认值
Returns:
运单信息，格式与 create_waybill_d 返回的数据结构相同
"""
try:
# 构建 headers，优先使用传入的 Authorization
headers = dict(API_CONFIG['headers'])
if Authorization:
headers['Authorization'] = Authorization
# 初始化返回数据
result_data = {
"consignmentCode": consignmentCode,
"waybill_id": "",
"operate_unit_name": "",
"operate_unit_code": "",
"detailList": []
}
# 请求表头详情，使用运单编号（chmCode）查询
head_url = f"{API_CONFIG['base_url']}/Exp/bus-customer/consignment/headInfo"
waybill_id = None
cons_type_code = "consType_D" # 默认值
try:
head_resp = requests.post(
head_url,
headers=headers,
json={"chmCode": consignmentCode},
timeout=30,
)
if head_resp.status_code == 200:
head_json = head_resp.json()
head_extra = (head_json.get("data") or {}).get("extra") or {}
# 提取运单ID
waybill_id = head_extra.get("id")
# 提取运单编号（如果返回的与输入不一致，使用返回的）
returned_code = head_extra.get("consignmentCode") or consignmentCode
result_data["consignmentCode"] = returned_code
result_data["waybill_id"] = str(waybill_id) if waybill_id else ""
# 提取公司信息
result_data["operate_unit_name"] = head_extra.get("operateUnitName") or ""
result_data["operate_unit_code"] = head_extra.get("operateUnitCode") or ""
# 提取运单类型（用于后续查询明细）
cons_type_code = head_extra.get("consTypeDicCode") or "consType_D"
else:
return f"❌ 查询运单信息失败: HTTP {head_resp.status_code}, 错误信息: {head_resp.text}"
except Exception as e:
return f"❌ 查询表头信息失败: {str(e)}"
# 若拿不到ID，直接返回已有数据
if not waybill_id:
return json.dumps(result_data, ensure_ascii=False, indent=2)
# 请求表体明细获取所有商品明细（sku_code 和 sku_name）
detail_url = f"{API_CONFIG['base_url']}/Exp/bus-customer/consignment/detail/list"
try:
detail_payload = {
"consType": cons_type_code,
"consignmentId": waybill_id,
"pageIndex": 1,
"pageSize": 100, # 增加页面大小以获取更多记录
}
detail_resp = requests.post(
detail_url,
headers=headers,
json=detail_payload,
timeout=30,
)
if detail_resp.status_code == 200:
detail_json = detail_resp.json()
records = (detail_json.get("data") or {}).get("records") or []
if isinstance(records, list):
# 遍历所有记录，构建 detailList
for item in records:
if isinstance(item, dict):
detail_item = {
"sku_code": item.get("skuCode") or "",
"sku_name": item.get("skuName") or ""
}
result_data["detailList"].append(detail_item)
except Exception:
pass
# 返回 JSON 格式数据
return json.dumps(result_data, ensure_ascii=False, indent=2)
except requests.exceptions.RequestException as e:
return f"❌ 网络请求失败: {str(e)}"
except json.JSONDecodeError as e:
return f"❌ 响应解析失败: {str(e)}"
except Exception as e:
return f"❌ 查询运单信息失败: {str(e)}"
def query_waybill_list(initialize: int = 1, consStatusName: str = "等待录入", Authorization: str = None) -> str:
"""查询运单列表信息
......
......@@ -12,7 +12,7 @@ from langchain_core.messages import AnyMessage
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 导入 API 模块
from API.waybill_api import query_waybill_list, create_waybill_d, push_waybill_for_ocr
from API.waybill_api import query_waybill_list, create_waybill_d, push_waybill_for_ocr, query_waybill_info
from API.paperless_api import upload_clearance_file
# 导入工具类
......@@ -179,7 +179,9 @@ def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List
system_msg = f"""你是一个专业的出口物流系统智能助手，专门帮助用户处理运单相关的业务操作。
## 你的主要职责：
1. **运单查询**：根据用户需求查询运单列表，支持按状态、时间等条件筛选，结果以JSON形式展示，AI不用对返回数据JSON进行加工
1. **运单查询**：根据用户需求查询运单信息
- **查询运单列表**：当用户查询运单但不提供运单号时，使用 query_waybill_list 查询运单列表，支持按状态、时间等条件筛选，结果以JSON形式展示，AI不用对返回数据JSON进行加工
- **查询单个运单详情**：当用户提供运单号查询时，使用 query_waybill_info 查询该运单的详细信息（包括运单基本信息、公司信息、商品明细等），返回格式与 create_waybill_d 相同
2. **运单创建**：协助用户创建D类运单，确保信息完整准确
- **重要说明**：当用户输入"创建运单"或"申报"时，都理解为"创建运单"操作，应调用 create_waybill_d 工具
3. **业务咨询**：解答用户关于出口物流流程、运单状态、操作规范等问题
......@@ -197,6 +199,9 @@ def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List
## 可用工具：
- query_waybill_list: 查询运单列表，支持按状态筛选，需提供Authorization，结果以JSON形式展示，AI不用对返回数据JSON进行加工
- **使用场景**：当用户查询运单但不提供运单号时使用此工具
- query_waybill_info: 根据运单编号查询单个运单的详细信息（包括运单基本信息、公司信息、商品明细等），需要提供参数（运单编号、Authorization），返回格式与 create_waybill_d 相同
- **使用场景**：当用户提供运单号查询运单详情时使用此工具
- create_waybill_d: 根据运单号创建D类运单，需要提供参数（运单号、Authorization）
- **注意**：当用户说"创建运单"或"申报"时，都应调用此工具
- upload_clearance_file: 上传清关PDF文件，需要 code、slip_id、pdf_path、Authorization,上传时文件路径不用确认，路径肯定是完整的，文件肯定是存在的，不用确认文件索引，工具会自动设置索引为0，请直接调用工具接口
......@@ -207,6 +212,10 @@ def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List
- 空字段会显示为空单元格
- 运单创建结果会显示成功/失败状态和详细信息
## query_waybill_info数据展示说明：
- 返回格式与 create_waybill_d 相同，包含：运单编号、运单ID、公司名称、公司编码、商品明细列表
- 按照数据返回的原本格式进行展示，不得增删前后缀或改写内容
## create_waybill_d数据展示说明：
- 按照数据返回的原本格式进行展示，不得增删前后缀或改写内容。
......@@ -222,7 +231,7 @@ def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List
# 创建 ReAct 智能体
agent = create_react_agent(
model=model,
tools=[query_waybill_list, create_waybill_d, upload_clearance_file, push_waybill_for_ocr],
tools=[query_waybill_list, query_waybill_info, create_waybill_d, upload_clearance_file, push_waybill_for_ocr],
pre_model_hook=pre_model_inspect_attachments,
prompt=_create_system_prompt,
)
......