zhouhui.jiang

update

......@@ -13,7 +13,7 @@ API_CONFIG = {
"headers": {
"accept": "*/*",
"Content-Type": "application/json",
"Authorization": os.getenv("API_AUTHORIZATION", "Bearer 2.37cfc9a928d14d0186e6896fa080bc99"),
"Authorization": os.getenv("API_AUTHORIZATION", "Bearer 2.6adbae8492564f68b213f8d1e785b3d9"),
"Ver": os.getenv("API_VER", "033BD94B1168D7E4F0D644C3C95E35BF.D73E33B659AD1D6B7D181D1DF8D05760"),
"Referer": os.getenv("API_REFERER", "http://192.168.1.251/")
}
......
......@@ -4,16 +4,18 @@ import os
import sys
import json
from typing import Dict, Any, List, Optional
from contextvars import ContextVar
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import AnyMessage
# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 添加项目根目录到 Python 路径(避免重复添加)
_project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _project_root not in sys.path:
sys.path.insert(0, _project_root)
# 导入 API 模块
from API.waybill_api import query_waybill_list, create_waybill_d, create_waybill_c, push_waybill_for_ocr, query_waybill_info, create_waybill_d_with_id
from API.paperless_api import upload_clearance_file, upload_file_for_ocr
from API.api_config import API_CONFIG
# 导入工具类
from langgraph_examples.utils.message_processor import MessageProcessor
......@@ -163,34 +165,8 @@ def extract_token(state: Dict[str, Any]) -> Dict[str, Any]:
return result
def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List[AnyMessage]:
"""
创建动态系统提示词
Args:
state: LangGraph 状态字典
config: Runnable 配置
Returns:
包含系统消息和原始消息的列表
"""
# 添加调试信息,确认函数被调用
# print("\n=== _create_system_prompt 被调用 ===")
# print(f"state type: {type(state)}")
# print(f"state keys: {list(state.keys()) if isinstance(state, dict) else 'not a dict'}")
# 从 state 中提取动态参数
params = extract_token(state)
token = params.get("token", "")
# 如果从 state 中提取的 token 为空,则从 api_config.py 中获取 Authorization 作为备选
if not token:
from API.api_config import API_CONFIG
token = API_CONFIG.get("headers", {}).get("Authorization", "")
# 创建系统提示词(使用 f-string 以便插入 token)
system_msg = f"""你是一个专业的出口物流系统智能助手,专门帮助用户处理运单相关的业务操作。
# 系统提示词模板(提取为模块级常量,避免每次调用都创建大字符串)
_SYSTEM_PROMPT_TEMPLATE = """你是一个专业的出口物流系统智能助手,专门帮助用户处理运单相关的业务操作。
## 你的主要职责:
1. **运单查询**:根据用户需求查询运单信息
......@@ -313,6 +289,33 @@ def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List
请根据用户的具体需求,选择合适的工具并提供帮助。"""
def _create_system_prompt(state: Dict[str, Any], config: RunnableConfig) -> List[AnyMessage]:
"""
创建动态系统提示词
Args:
state: LangGraph 状态字典
config: Runnable 配置
Returns:
包含系统消息和原始消息的列表
"""
# 添加调试信息,确认函数被调用
# print("\n=== _create_system_prompt 被调用 ===")
# print(f"state type: {type(state)}")
# print(f"state keys: {list(state.keys()) if isinstance(state, dict) else 'not a dict'}")
# 从 state 中提取动态参数
params = extract_token(state)
token = params.get("token", "")
# 如果从 state 中提取的 token 为空,则从 api_config.py 中获取 Authorization 作为备选
if not token:
token = API_CONFIG.get("headers", {}).get("Authorization", "")
# 使用模板创建系统提示词(只动态替换 token,避免每次创建大字符串)
system_msg = _SYSTEM_PROMPT_TEMPLATE.format(token=token)
# 返回系统消息 + 原始消息
result = [{"role": "system", "content": system_msg}] + state.get("messages", [])
print(f"返回消息数量: {len(result)}")
......