# 本文件接续 api_server.py 顶部的主流程 3～11；网页默认入口为 ChatService.chat_stream。 # 阅读时先找 chat_stream，再顺着其调用进入 intent_agent、规划方法和代理调用方法。 # chat 是非流式入口，两者互为替代；它们不是先后调用关系。 # 主助手按计划分派任务；子助手中的 AgentExecutor 才执行“模型选工具 → 工具返回 → 再问模型”的循环。 """ 需求：SmartBizTrip核心对话服务，供CLI和API共同调用 架构说明： 本模块实现了企业差旅智能体的核心对话流程，采用优化的 Planning + ReAct 架构： 1. 意图识别（intent_agent）：分析用户输入，识别出用户想做什么（查天气/查航班/查接待资源等） 2. 启发式路由（_should_skip_planning）：快速判断任务复杂度 - 单意图或独立多意图：跳过规划，直接执行（省去 1 次 LLM 调用） - 含依赖关系的复杂任务：调用 planning_agent 生成执行计划 3. ReAct循环（react_loop）：针对复杂任务，逐步执行每个步骤 - Action（行动）：直接调用对应的 agent 执行查询（跳过 Thought 推理，省去每步 1 次 LLM 调用） - Observation（观察）：记录执行结果，供后续步骤参考 4. 结果汇总：所有步骤完成后，生成一条连贯的最终回复 整体流程： 用户输入 → 意图识别 → [跳过规划 / 调用planning_agent] → [直接执行 / ReAct循环] → 返回回复 A2A架构说明： Agent2Agent (A2A) 是一种代理间通信协议。本系统通过 AgentNetwork 管理多个子代理： - WeatherQueryAssistant（天气查询代理）：运行在 localhost:5005 - TicketAssistant（票务代理）：运行在 localhost:5006 - TripAssistant（差旅管家代理）：运行在 localhost:5007 主助手通过 A2A 协议向子代理发送任务请求，获取查询结果后整合回复用户。 架构图： +─────────────────────────────────────────────────────────────────────────────+ │ ChatService 类 │ │ │ │ ┌────────────────── 成员变量 ──────────────────┐ │ │ │ agent_urls{} ── 子代理地址映射 │ │ │ │ agent_network ── AgentNetwork 实例 │ │ │ │ llm ── ChatOpenAI (LLM) │ │ │ │ memory ── ConversationMemory │ │ │ │ messages[] ── 当前对话消息列表 │ │ │ │ conversation_history ── 完整对话字符串 │ │ │ │ _available_tools_text ── 工具列表缓存 │ │ │ └───────────────────────────────────────────────┘ │ │ │ │ ┌────────────────── 方法分层 ──────────────────┐ │ │ │ │ │ │ │ [入口层] │ │ │ │ chat() ── 标准对话入口 │ │ │ │ chat_stream() ── 流式对话入口 │ │ │ │ │ │ │ │ [智能决策层] │ │ │ │ intent_agent() ── 意图识别 (LLM) │ │ │ │ _should_skip_planning() ── 启发式路由规则 │ │ │ │ planning_agent() ── 任务规划 (LLM) │ │ │ │ │ │ │ │ [执行层] │ │ │ │ react_loop() ── ReAct 循环 (标准) │ │ │ │ _react_loop_stream() ── ReAct 循环 (流式) │ │ │ │ execute_step() ── 执行单个计划步骤 │ │ │ │ │ │ │ │ [代理调用层] │ │ │ │ _call_agent_intent() ── 统一调用 (标准) │ │ │ │ _call_agent_intent_stream() ── 统一调用 (流式)│ │ │ │ │ │ │ │ [辅助层] │ │ │ │ get_available_tools() ── 动态工具列表 │ │ │ │ get_agent_cards() ── 获取代理卡片 │ │ │ │ _init_db_and_load_memory() ── DB初始化+加载 │ │ │ │ update_user_profile() ── 更新用户偏好 │ │ │ │ get_memory_state() ── 记忆状态摘要 │ │ │ │ clear_memory() ── 清空所有记忆 │ │ │ └───────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────+ 请求处理流程 (chat / chat_stream，图中 1～6 为函数内部阶段，跨文件编号见 #【主流程】): 用户输入 │ ▼ ┌─────────────────┐ │ 1. 写入记忆 │ memory.add_message("user", input) └────────┬────────┘ ▼ ┌─────────────────┐ │ 2. 意图识别 │ intent_agent() → LLM → (intents, queries, follow_up) └────────┬────────┘ ▼ ┌──────────────────────────────────────────────┐ │ 3. 特殊情况判断 │ │ out_of_scope? → 返回追问 │ │ follow_up? → 返回追问 │ └────────────────────┬─────────────────────────┘ ▼ ┌──────────────────────────────────────────────┐ │ 4. 任务规划 │ │ _should_skip_planning()? │ │ ├─ YES → plan = {need_plan: false} │ │ └─ NO → planning_agent() → LLM 生成计划 │ └────────────────────┬─────────────────────────┘ ▼ ┌──────────────────────────────────────────────┐ │ 5. 执行 │ │ need_plan? │ │ ├─ YES → react_loop() │ │ │ ├─ 按 depends_on 分组 │ │ │ ├─ 组内并行 execute_step() │ │ │ │ → _call_agent_intent() │ │ │ └─ 汇总 → LLM 生成连贯回复 │ │ │ │ │ └─ NO → 遍历 intents │ │ → _call_agent_intent() 每个意图 │ │ → 拼接所有回复 │ └────────────────────┬─────────────────────────┘ ▼ ┌─────────────────┐ │ 6. 写入记忆 │ memory.add_message("assistant", response) └────────┬────────┘ ▼ 返回回复 _call_agent_intent() 内部路由: intent │ ├─ city_service → 尝试查询天气 → city_service_prompt() + LLM → 生成建议 │ ├─ flight/train/event/car_rental/... │ │ │ ├─ 提取实体 → memory.extract_entities() │ ├─ 更新上下文 → memory.update_task_context() │ ├─ A2A → agent.send_task_async(task) │ │ ├─ WeatherQueryAssistant → summarize_weather_prompt() + LLM │ │ ├─ TicketQueryAssistant/TripAssistant → 总结；TicketAssistant → 原文 │ │ │ （流式版本把 TicketAssistant 也纳入总结） │ │ └─ 其他 → 直接返回原始结果 │ └─ 返回总结结果 │ └─ 未识别 → "暂不支持此意图" 子代理调用关系: ChatService (由 API :8080 调用) │ ├── [A2A :5005] → WeatherQueryAssistant (天气查询) ├── [A2A :5006] → TicketAssistant (票务预定) ├── [A2A :5007] → TripAssistant (差旅管家: 商务用车/会务套餐/保险) │ └── [直连 LLM] → city_service (先尝试查询天气，再生成城市接待建议) 优化策略总结: 1. 启发式路由 → 简单任务跳过 planning_agent，省 1 次 LLM 调用 2. 省略 Thought → ReAct 循环中跳过 Thought 推理，每步省 1 次 LLM 调用 3. 并行执行 → 无依赖步骤通过 asyncio.gather() 并发执行 4. 工具说明 → get_available_tools 可整理代理卡片；当前对话主线未调用 """ # ==================== 导入依赖 ==================== import asyncio # 异步IO库，用于并发调用多个 agent（异步网络请求） import json # JSON处理，用于解析大模型的JSON输出和序列化数据 import re # 正则表达式，用于清理大模型返回的JSON（去掉代码块标记） import uuid # 唯一标识符生成，用于给每个任务分配唯一ID from datetime import datetime # 时间处理，用于获取当前日期和时间戳 import mysql.connector # MySQL 数据库驱动，用于记忆持久化 import pytz # 时区库，用于将时间转换到 Asia/Shanghai 时区 from langchain_core.prompts import ChatPromptTemplate # LangChain 的提示模板类 # AgentNetwork: 代理网络管理器，负责注册和获取子代理 # A2AClient: A2A 客户端，支持自定义超时 # TextContent: 消息文本内容包装 # Message: A2A 消息对象，包含角色和内容 # MessageRole: 消息角色枚举（USER/ASSISTANT） # Task: A2A 任务对象，用于向子代理发送请求 from langchain_openai import ChatOpenAI # LangChain 的大模型接口，兼容 OpenAI API 格式 # python_a2a 是 A2A 协议的 Python 客户端库 from python_a2a import AgentNetwork, Message, MessageRole, Task, TextContent from python_a2a.client import A2AClient # 导入项目内部模块 from config import Config # 配置模块（模型地址、API Key、意图映射等） from create_logger import logger # 日志模块，用于记录运行日志 from main_prompts import SmartBizTripPrompts # Prompt 模板管理 from memory import ConversationMemory # 记忆管理（短期记忆、用户偏好、任务上下文） conf = Config() # 全局配置实例，包含模型参数、意图映射等 class ChatService: """ SmartBizTrip 对话服务类 这是整个系统的核心类，负责处理用户的所有输入并生成回复。 内部流程为：意图识别 → 任务规划 → 执行（直接/ReAct循环）→ 返回回复 使用示例： service = ChatService() # 创建服务实例 response = await service.chat("北京明天天气怎么样？") # 发送用户输入 print(response) # 打印助手回复 """ def __init__(self): """ 初始化 ChatService 实例（启动 S4.2） 完成以下初始化工作： 1. 配置子代理网络（注册天气、票务和差旅代理的客户端） 2. 初始化大模型连接 3. 初始化记忆管理器 4. 初始化对话历史和工具列表缓存 """ # ========== 1. 配置 A2A 代理网络 ========== # 记录每个子代理的访问地址 # 调用者：api_server.py 创建全局 ChatService() 时自动进入。 # 输入：没有业务问题参数，使用模块级 conf 配置。输出：初始化好的 self，不返回聊天文本。 # self.xxx 是服务实例上保留的状态，后面的每次请求都能读取；方法里的普通局部变量只属于本次调用。 self.agent_urls = { "WeatherQueryAssistant": "http://localhost:5005", # 天气查询代理的地址 "TicketAssistant": "http://localhost:5006", # 票务代理的地址 "TripAssistant": "http://localhost:5007" # 差旅管家代理的地址 } # 创建代理网络实例，给它起个名字 self.agent_network = AgentNetwork(name="企业差旅助手网络") # 向网络中注册子代理，使用自定义超时（120秒），避免子代理 LLM+MCP 调用链超时 # 这里只登记通信客户端；不会在当前进程创建或启动远端子助手。 self.agent_network.add("WeatherQueryAssistant", A2AClient("http://localhost:5005", timeout=120)) self.agent_network.add("TicketAssistant", A2AClient("http://localhost:5006", timeout=120)) self.agent_network.add("TripAssistant", A2AClient("http://localhost:5007", timeout=120)) # ========== 2. 初始化大模型连接 ========== # 使用 LangChain 的 ChatOpenAI 接口连接大模型 # 实际模型与地址来自 Config；创建对象尚未发送模型推理请求。 self.llm = ChatOpenAI( model=conf.model_name, # 配置指定的模型名称 api_key=conf.api_key, # API 密钥 base_url=conf.base_url, # API 基础地址 temperature=0.1 # 温度参数：越小输出越确定、越稳定（0.1表示比较严谨） ) # ========== 3. 初始化记忆和对话状态 ========== # self.memory 负责结构化记忆与持久化；不是大模型的内部记忆。 # short_term_limit=10 限制消息条数，通常一轮用户+助手就占两条。 self.memory = ConversationMemory(short_term_limit=10) # 短期记忆最多保留10条对话 self.messages = [] # 对话消息列表，格式：[{"role": "user/assistant", "content": "..."}] self.conversation_history = "" # 完整对话历史字符串，用于调试和展示 self._available_tools_text = "" # 保存整理后的工具说明，当前对话主线未读取它 # 城市提取 prompt，用于从城市接待建议查询中提取目的地城市 # 把城市提取规则保存成模板，后续传入 conversation_history/query 后才能得到本轮模型输入。 # 创建模板是本地对象操作；实际推理发生在 _fetch_weather_for_city_service 的 chain.invoke。 self._city_extract_prompt = ChatPromptTemplate.from_template( """从以下用户查询和对话历史中提取目的地城市（用于天气查询）。规则： - 只输出城市名称，不要输出其他内容。 - 如果查询中没有明确城市，尝试从对话历史中推断。 - 如果无法确定城市，输出"未知"。 对话历史：{conversation_history} 用户查询：{query} """) # ========== 4. 初始化数据库连接，加载持久化记忆 ========== self._init_db_and_load_memory() def get_available_tools(self) -> str: """ 从 A2A 代理网络中动态获取可用工具列表，格式化为文本 这是 ReAct 架构的关键：告诉大模型当前有哪些工具/agent 可以调用， 让大模型在推理时从这些工具中选择合适的。 本方法从已注册的代理读取卡片；新增代理仍需注册和配置路由。 当前 chat/chat_stream 没有调用本方法，也没有把其结果注入主流程提示词。 返回值： str: 格式化的工具列表文本，例如： - Agent: WeatherQueryAssistant, 查询天气信息 Skill: 天气查询: 根据城市查询天气 - Agent: TicketQueryAssistant, 查询票务信息 Skill: 票务查询: 查询航班、高铁票等 """ # 【辅助】按需读取代理卡片并整理说明；当前 chat/chat_stream 主线没有调用此方法。 # 此方法赋值 _available_tools_text，但本方法自身没有“缓存非空就直接返回”的判断。 # 输入：self.agent_network 中已经注册的代理。输出：多行字符串，描述代理名、技能和说明。 # 函数名中的 tools 指整理出的能力说明文本，与 A2A 子助手加载到的 StructuredTool 对象不同。 try: # 调用 get_agent_cards() 获取所有在线 agent 的卡片信息 cards = self.get_agent_cards() lines = [] for card in cards: # 每个 agent 一行，包含名称和描述 lines.append(f"- Agent: {card['name']}, {card['description']}") # 每个 agent 的技能作为子项列出 for skill in card['skills']: lines.append(f" Skill: {skill}") self._available_tools_text = "\n".join(lines) except Exception as e: # 如果 agent server 没有启动或网络不通，使用默认的工具列表作为降级方案 logger.warning(f"获取可用工具列表失败: {e}") self._available_tools_text = ( "- WeatherQueryAssistant: 查询天气信息\n" "- TicketQueryAssistant: 查询票务信息（航班/高铁票/会展门票）\n" "- TicketAssistant: 预定票务\n" "- TripAssistant: 差旅管家（商务用车/会务套餐/保险的查询与预订）" ) return self._available_tools_text def _init_db_and_load_memory(self): """ 初始化数据库连接，并从数据库加载持久化记忆 启动流程： 1. 创建 MySQL 连接，注入到 memory 实例 2. 从数据库加载用户偏好、查询历史、短期对话 """ # 【启动 S4.2.1】尝试连接数据库，依次恢复偏好、查询历史、短期消息。 # 连接失败会保留内存模式；这不表示依赖数据库的业务查询也能运行。 # 调用者：__init__。输入：conf 中的数据库参数；输出：通过修改 self.memory/self.messages 恢复状态。 # 三种 load 方法各自读取对应表；此方法不创建表，不把 SQL 初始化文件自动执行一遍。 try: db_conn = mysql.connector.connect( host=conf.host, user=conf.user, password=conf.password, database=conf.database ) self.memory.set_db_connection(db_conn) logger.info("数据库连接初始化成功") # 从数据库恢复记忆数据 self.memory.load_profile_from_db() self.memory.load_entities_from_db() self.memory.load_messages_from_db() # 将加载的短期消息同步到 messages 列表 for msg in self.memory.short_term_messages: self.messages.append({"role": msg["role"], "content": msg["content"]}) logger.info(f"记忆加载完成: 偏好={len(self.memory.user_profile)}项, " f"历史={len(self.memory.entity_history)}条, 对话={len(self.memory.short_term_messages)}条") except Exception as e: logger.warning(f"数据库连接初始化失败: {e}") self.memory.set_db_connection(None) async def _fetch_weather_for_city_service(self, query_str: str) -> str: """ 为城市接待建议查询目的地天气，返回简要天气信息。 如果无法确定目的地，返回空字符串。 参数： query_str (str): 用户的接待资源查询内容 返回值： str: 天气信息摘要，或空字符串 """ # 【主流程 6：城市建议子分支，内部步骤 1】提取目的地；未知城市直接返回空天气。 # 调用者：两个 _call_agent_intent 版本中的 city_service 分支。 # 输入：用户接待需求文本；输出：天气代理产物文本或空字符串。 # 顺序：先让模型提取城市 → 判断城市是否存在 → 构造天气任务 → 等待代理 → 取回产物。 # 该函数自身不会生成接待建议，结果作为 weather_info 交给外层的建议模板。 try: # 从查询中提取目的地城市 chain = self._city_extract_prompt | self.llm city = chain.invoke({ "conversation_history": self.memory.get_short_term_text(), "query": query_str }).content.strip() if city == "未知" or not city: logger.info(f"无法从查询中提取目的地城市: {query_str}") return "" logger.info(f"从接待资源查询中提取到城市: {city}") # 通过天气代理查询该城市天气 agent = self.agent_network.get_agent("WeatherQueryAssistant") if agent is None: return "" # 保留接待需求里的日期，让天气助手按用户要求查询；没给日期才默认明天。 # 【内部步骤 2】保留需求中的日期，构造天气 Task，发给 WeatherQueryAssistant。 weather_query = ( f"请查询{city}的天气，日期以用户接待需求为准；仅在需求未提供日期时查询明天。\n" f"用户接待需求：{query_str}" ) msg = Message(content=TextContent(text=weather_query), role=MessageRole.USER) task = Task(id="task-" + str(uuid.uuid4()), message=msg.to_dict()) raw_response = await asyncio.get_event_loop().run_in_executor( None, lambda: asyncio.run(agent.send_task_async(task)) ) # 【内部步骤 3】任务完成且有产物时取回文本；未完成或没有产物时返回空串。 # 产物本身仍可能描述“无数据”，不能将非空文本等同于有效天气数据。 if raw_response.status.state == 'completed' and raw_response.artifacts: return raw_response.artifacts[0]['parts'][0]['text'] return "" except Exception as e: logger.warning(f"查询接待资源天气失败: {e}") return "" def update_user_profile(self, profile: dict, replace: bool = False): """ 更新用户偏好设置，由前端 API 直接调用 参数： profile (dict): 用户偏好键值对，如 {"seat_type": "二等座", "cabin_type": "经济舱"} 默认合并；replace=True 时以网页提交的完整偏好替换旧值 """ # 【旁路：偏好更新】交给 memory 合并或替换，并尝试持久化；完成后由 API 返回状态。 # 调用者：POST /api/memory/profile。输入：偏好字典和是否整体替换。 # 数据交给 ConversationMemory.update_profile；后续 intent_agent 再将偏好文本放入模型输入。 self.memory.update_profile(profile, replace=replace) logger.info(f"更新用户偏好: {profile}") def intent_agent(self, user_input: str): """ 意图识别：分析用户输入，判断用户想做什么 工作流程： 1. 将用户输入、对话历史、用户偏好等上下文组装后发送给大模型 2. 大模型返回 JSON，包含识别到的意图列表、改写后的查询、追问消息 支持的意图类型： - weather: 天气查询 - flight: 航班查询 - train: 高铁票查询 - event: 会展门票查询 - order: 票务预定 - city_service: 城市接待建议 - out_of_scope: 超出系统能力范围 参数： user_input (str): 用户本次的输入内容 返回值： tuple: (intents, user_queries, follow_up_message) - intents (list): 识别到的意图列表，如 ["weather", "flight"] - user_queries (dict): 改写后的查询，如 {"weather": "北京明天天气", "flight": "北京到上海航班"} - follow_up_message (str): 追问消息，当意图不明确时用于追问用户 """ # 组装 Prompt 模板 + 大模型，形成完整的处理链 # 【主流程 4.1】获取提示模板并与模型组成调用链；此时还没有查天气或票务。 # 调用者：chat 或 chat_stream，每轮正常处理先调用一次。 # 输入：user_input 字符串，例如“北京明天天气怎样”；输出：三个值组成的元组。 # 示意：(["weather"], {"weather": "北京某日期天气"}, "")，具体内容由模型生成。 # 调用方用同样顺序接到 intents、user_queries、follow_up_message 三个变量中。 chain = SmartBizTripPrompts.intent_prompt() | self.llm # 计算上海时区的当前日期，填入 intent_prompt 的 current_date 占位符。 current_date = datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d') # 调用大模型，传入所有上下文信息 # 【主流程 4.2】实际请求模型，结合模板引用的历史、偏好、任务上下文和本轮问题。 # 这是同步 invoke；即使外层是 async 方法，这次调用也不会自动变成非阻塞。 intent_response = chain.invoke({ "conversation_history": self.memory.get_short_term_text(), # 短期对话历史 "query": user_input, # 用户本次输入 "current_date": current_date, # 供模板解析今天、明天等相对日期 "user_profile": self.memory.get_profile_text(), # 用户偏好（如"二等座"、"经济舱"） "task_context": json.dumps(self.memory.current_task, ensure_ascii=False) # 当前任务上下文 }).content.strip() # .content 获取文本内容，.strip() 去掉首尾空白 logger.info(f"意图识别原始响应: {intent_response}") # 大模型有时会返回 ```json ... ``` 格式的代码块，需要用正则去掉 # 清理的是字符串边缘的 Markdown 代码围栏，便于 json.loads 解析。 # 这不校验 JSON 字段类型，也不会自动修复漏引号、漏逗号等结构错误。 intent_response = re.sub(r'^```json\s*|\s*```$', '', intent_response).strip() logger.info(f"清理后响应: {intent_response}") # 将 JSON 字符串解析为 Python 字典 # 【主流程 4.3】把模型文本解析成意图、改写查询和追问；无效 JSON 交由对话入口捕获。 intent_output = json.loads(intent_response) # dict.get(key, default) 在字段缺失时使用默认值；字段存在但类型不对时不会自动纠正类型。 # intents 用列表保留多个需求，user_queries 用字典按意图找到对应的查询内容。 intents = intent_output.get("intents", []) # 意图列表，如 ["weather", "flight"] user_queries = intent_output.get("user_queries", {}) # 改写后的查询字典 follow_up_message = intent_output.get("follow_up_message", "") # 追问消息 # 先确认是字符串才 strip 去空白；模型若给了其他类型，这里将追问归为空串。 # 这样外层可以统一用 follow_up_message != "" 判断是否需要先回复用户。 follow_up_message = follow_up_message.strip() if isinstance(follow_up_message, str) else "" # 两项条件都满足才补充默认问候：识别为范围外，而且模型没有生成可用回复。 # 已有追问时保留模型原文；不会覆盖每一条普通业务问题。 if "out_of_scope" in intents and not follow_up_message: follow_up_message = ( "你好！我是企业差旅助手，可以协助查询天气、航班、高铁票、会展门票，" "以及商务用车、会务套餐和保险，并提供城市接待建议。" "查询和预订功能需要相应服务及数据源可用。" "你可以先告诉我出发地、目的地和出行日期。" ) logger.info(f"intents: {intents}||user_queries: {user_queries}||follow_up_message: {follow_up_message} ") return intents, user_queries, follow_up_message def _should_skip_planning(self, intents: list) -> bool: """ 启发式判断：是否可以跳过 planning_agent，直接执行 判断规则（满足任一即跳过）： 1. 未包含 complex，且意图数不超过一个时直接执行 2. 多意图全部命中下方 independent_intents 集合时直接执行； 当前直接执行分支使用 for + await，按意图顺序串行调用 需要进入 planning 的情况： - complex 意图无论有几项都进入规划 - 多意图中出现不在 independent_intents 集合内的类型（如 order） - 单独的 order 或 trip_order 仍会跳过规划；本方法不直接分析自然语言依赖 优化收益：省去 1 次 LLM 调用（planning_agent），大幅降低简单请求的响应时间 参数： intents (list): 识别到的意图列表 返回值： bool: True 表示跳过 planning_agent，False 表示需要调用 planning_agent """ # complex 必须进入规划，即使意图列表只有一项。 # 【主流程 5.1】按意图标签做规则判断。规则返回 True 才跳过 planning_agent。 # 依赖是否表达为 complex 取决于前面的意图识别，不是这个函数再次理解用户原话。 # 输入：意图字符串列表。输出：布尔值，True=直接执行，False=还要询问规划模型。 # 注意这里不是在生成 plan，返回值也不表示任务已完成；它仅决定下一段代码走哪条路。 # 第一个判断优先级最高：即使列表只有 ["complex"]，也在这里返回 False。 # return 会直接离开整个函数，不会再去检查下面的“单意图”规则。 if "complex" in intents: return False # 规则1：单意图直接跳过 if len(intents) <= 1: return True # 规则2：多意图但全部为独立查询类，无需分步规划 independent_intents = {"weather", "flight", "train", "event", "city_service", "car_rental", "meeting_package", "insurance", "trip_order"} # 逐项检查：只要发现一个不在集合里的意图就返回 False。 # 所有项都检查完仍没提前返回，才执行循环后的 return True；集合成员检查不调用模型。 for intent in intents: if intent not in independent_intents: # 存在需要分步的意图（如 order），不跳过 # TODO 如果大模型认为意图非常复杂，使用complex。这里命中False return False return True def planning_agent(self, intents: list, user_queries: dict) -> dict: """ 任务规划：直接进行任务的拆解 注意：在调用此方法前会先通过 _should_skip_planning 做启发式判断， 只有复杂任务才会进入此方法，省去不必要的 LLM 调用。 参数： intents (list): 识别到的意图列表，如 ["weather", "flight", "city_service"] user_queries (dict): 改写后的查询字典，如 {"weather": "...", "flight": "..."} 返回值： dict: 规划结果，格式为： - 简单任务：{"need_plan": false, "reason": "单意图，直接查询即可", "steps": []} - 复杂任务：{"need_plan": true, "reason": "多意图需要分步执行", "steps": [{"step": 1, "action": "...", "intent": "weather", "depends_on": 0}, ...]} """ # 【主流程 5.2：可选】只有规则未跳过时才调用规划模型，返回 need_plan 和 steps。 # 即使调用了规划模型，模型仍可能返回 need_plan=False，随后走直接执行分支。 # 调用者：chat/chat_stream 的规划分支。输入：意图列表和按意图保存的改写查询。 # 输出：Python 字典，例如 {"need_plan": True, "steps": [{"step": 1, "intent": "weather", "depends_on": 0}]}。 # step 是步骤编号；intent 用来找代理；action 描述动作；depends_on 指要先拿到哪个步骤的结果。 chain = SmartBizTripPrompts.planning_prompt() | self.llm # 调用参数中的 query 从 messages 最后一条取；对话入口刚加入了本轮用户消息。 # json.dumps 把意图列表/查询字典变成提示词中的文本，并不是向子代理发送请求。 planning_response = chain.invoke({ "conversation_history": self.memory.get_short_term_text(), "query": self.messages[-1]["content"] if self.messages else "", "intents": json.dumps(intents, ensure_ascii=False), "user_queries": json.dumps(user_queries, ensure_ascii=False) }).content.strip() logger.info(f"规划响应: {planning_response}") planning_response = re.sub(r'^```json\s*|\s*```$', '', planning_response).strip() # 这里只将模型文本变成字典；步骤编号、依赖存在性和循环依赖要到 _execute_plan 再检查。 plan = json.loads(planning_response) return plan async def _call_agent_intent(self, intent: str, query_str: str) -> str: """ 统一调用 agent 的底层逻辑 这是一个私有方法（以 _ 开头），不直接对外暴露，供以下两个场景复用： 1. 简单任务直接执行时（chat 方法的 else 分支） 2. ReAct 循环中的步骤执行时（execute_step 方法） 这样做的好处：避免了代码重复，修改 agent 调用逻辑时只需改一处。 处理逻辑说明： - 城市接待建议先尝试调用天气子代理，再让主模型根据天气生成建议 - 如果是其他意图，通过 A2A 协议调用对应的子代理 - 收到结果后，根据代理类型做适当的总结（天气/票务需要总结，其他直接返回） 参数： intent (str): 意图类型，如 "weather"、"flight"、"city_service" query_str (str): 查询内容，如 "北京明天天气" 返回值： str: agent 返回的结果文本 """ # 根据意图类型，从配置中查找对应的 agent 名称 # 【主流程 6.1】把 intent 映射成子助手名称；city_service 走单独的城市建议分支。 # 调用者：非流式 chat 的直接分支，或复杂计划的 execute_step。 # 输入：intent 决定交给谁，query_str 告诉对方具体做什么；输出：一次性返回一个字符串。 # 例如 intent="train" 时，配置映射得到 TicketAssistant，再创建 Task 发给票务服务。 # 查的是本地映射字典。例如 weather→WeatherQueryAssistant，train/flight→TicketAssistant。 # 没有映射时得到 None；city_service 虽不在映射里，仍会被下方单独的 if 分支处理。 agent_name = conf.intent.get(intent) # ========== 情况1：城市接待建议 =========== # 城市建议没有独立的接待代理；先尝试查天气，再由主模型生成建议。 # 天气是否可用取决于子代理和数据源，不能仅凭调用就认定查询成功。 if intent == "city_service": chain = SmartBizTripPrompts.city_service_prompt() | self.llm weather_info = await self._fetch_weather_for_city_service(query_str) return chain.invoke({"query": query_str, "weather_info": weather_info}).content.strip() # ========== 情况2：需要调用子代理 =========== elif agent_name: # 对于这些查询意图，保存意图类型和整段查询文本到记忆中 # 这样后续对话可以引用这些实体（多轮对话场景） # extract_entities 内部已自动持久化到数据库 if intent in ["flight", "train", "event", "car_rental", "meeting_package", "insurance"]: # 这里先记查询历史，再更新当前任务，随后才实际查询。 # 因此“查询历史里有一条记录”不代表该次业务查询已成功。 self.memory.extract_entities(intent, query_str) self.memory.update_task_context({"type": intent, "query": query_str}) # 从代理网络中获取对应的 agent 实例 # 【主流程 6.2】取得 A2A 客户端，准备携带历史和本轮查询的 Task。 agent = self.agent_network.get_agent(agent_name) # 检查是否取到了客户端对象；即使对象存在，远端服务也可能未启动，通信时仍可能抛异常。 if agent is None: # agent 未注册或不可用时，返回友好提示 logger.warning(f"未找到代理：{agent_name}") return f"抱歉，{agent_name} 暂时不可用，请稍后重试。" # 构建发送给 agent 的消息：结构化传递对话历史和当前查询 # 给子助手的 TextContent 内部放一段 JSON 字符串，包含 history 和 query。 # 这两个字段是业务内容；外层 A2A Message/Task 另外负责协议封装。 agent_input = json.dumps({ "history": self.memory.get_short_term_text(), "query": query_str }, ensure_ascii=False) msg = Message(content=TextContent(text=agent_input), role=MessageRole.USER) # Message 用角色和 TextContent 包装文本，to_dict 将它转换为 Task 所接收的消息字典。 # UUID 给每次代理任务生成独立标识，便于区分请求；它不是订单号或用户会话号。 task = Task(id="task-" + str(uuid.uuid4()), message=msg.to_dict()) # 通过 A2A 协议异步发送任务给子代理 # 线程池运行独立事件循环来等待 A2A；当前协程用 await 等待其返回。 # 这里只把这段 A2A 调用移入线程池，并未把所有同步模型/数据库调用异步化。 try: # 【主流程 6.3 → 7】经 A2A 把任务送到独立子助手进程，等待其完成工具调用。 raw_response = await asyncio.get_event_loop().run_in_executor( # None 表示使用默认线程池；lambda 是稍后由线程执行的无参数函数。 # 线程内 asyncio.run 创建事件循环并运行 send_task_async；外层 await 等待线程结果。 # 两层返回最终都汇合到 raw_response，也就是带状态和产物的 A2A Task。 None, lambda: asyncio.run(agent.send_task_async(task)) ) logger.info(f"{agent_name} 原始响应: {raw_response}") # 解析 agent 返回结果 # 【主流程 9】取回 task.artifacts 中的文本，或读取状态里的追问/失败信息。 # COMPLETED 只代表代理处理结束；仍要看返回内容，不能据此认定有数据或已真实下单。 if raw_response.status.state == 'completed' and raw_response.artifacts: # 任务成功完成，从 artifacts 中提取结果文本 # 数据路径：artifacts 是产物列表 → 第一个产物的 parts 列表 → 第一个 part 的 text。 # 当前只取第一个文本 part；不会在这里遍历、合并所有产物。 agent_result = raw_response.artifacts[0]['parts'][0]['text'] elif raw_response.status.message: # 有消息但可能失败了，尝试从消息中提取文本 agent_result = raw_response.status.message.get('content', {}).get('text', str(raw_response.status.message)) else: agent_result = f"查询失败：{raw_response.status.message or '未知错误'}" except Exception as e: # agent 服务不可用（如网络不通、服务崩溃），返回友好错误提示 logger.error(f"{agent_name} 调用异常: {str(e)}") return f"{agent_name} 服务暂时不可用：{str(e)}" # 根据代理类型，使用对应的总结 prompt 处理结果 # 【主流程 10】按代理类型整理结果。流式版本在 astream 时持续产出文本片段。 if agent_name == "WeatherQueryAssistant": chain = SmartBizTripPrompts.summarize_weather_prompt() | self.llm return chain.invoke({"query": query_str, "raw_response": agent_result}).content.strip() # 非流式版本此处仅匹配这两个名称；当前 TicketAssistant 会走下面的原文返回分支。 # 流式版本则将 TicketAssistant 也纳入票务总结，两个入口的处理细节并不完全相同。 elif agent_name == "TicketQueryAssistant" or agent_name == "TripAssistant": chain = SmartBizTripPrompts.summarize_ticket_prompt() | self.llm return chain.invoke({"query": query_str, "raw_response": agent_result}).content.strip() else: # 其他代理直接返回原始结果 return agent_result # ========== 情况3：未识别的意图 =========== else: return "暂不支持此意图。" async def execute_step(self, step: dict, user_queries: dict, previous_results: list | None = None) -> str: """ 执行单个计划步骤（供 ReAct 循环调用） 这个方法负责将规划中的单个步骤转化为实际的执行： 1. 从步骤中提取意图和查询内容 2. 调用 _call_agent_intent 执行实际的查询 参数： step (dict): 计划中的单个步骤，格式如： {"step": 1, "action": "查询天气", "intent": "weather", "depends_on": 0} user_queries (dict): 改写后的查询字典 返回值： str: 该步骤的执行结果 """ # 【计划执行 P4】取出当前步骤的意图和查询；必要时将动作及前置结果一起传给子助手。 # 调用者：_execute_plan 批量调度时执行。 # 输入：step 是当前步骤字典，user_queries 是意图→查询的映射，previous_results 是祖先步骤结果列表。 # 输出：当前步骤的结果字符串。它不会自行决定再增加步骤，只执行规划里已有的一步。 intent = step.get("intent", "") # 优先从 user_queries 中获取改写后的查询，如果没有则用用户原始输入 # 按当前步骤 intent 查找改写查询；没有该键时退回最后一条消息文本，再没有则用空串。 # 同一意图的多个计划步骤可能取到同一段查询，后面的 action 用于区分各步要完成的动作。 query_str = user_queries.get(intent, self.messages[-1]["content"] if self.messages else "") logger.info(f"执行步骤：{step.get('action', '')}，意图：{intent}") # 将执行动作和前置结果传给子代理，不额外增加 Thought 模型调用。 # 有前置结果或有动作说明时，把查询升级为结构化 JSON 文本。 # previous_results 是已执行步骤的结果，不是尚未执行的计划；由子助手阅读并决定工具参数。 if previous_results or step.get("action"): query_str = json.dumps({ "query": query_str, "action": step.get("action", ""), "previous_results": previous_results or [], "instruction": "依据前置结果完成当前动作；结果不足或失败时请说明或追问，不得编造参数。", }, ensure_ascii=False) # 委托给统一的 agent 调用方法 # await 等待一个步骤完成，return 把它的文本交回 _execute_plan 的 gather 收集。 return await self._call_agent_intent(intent, query_str) async def _execute_plan(self, steps: list, user_queries: dict) -> list: """先校验依赖，再按就绪批次执行；后续步骤携带前置结果。""" # 【计划执行 P1】先校验所有步骤号：必须是唯一正整数；depends_on=0 表示没有前置步骤。 # 当前每步只接收一个直接前置步骤编号，后续会沿这条依赖链向前收集结果。 # 调用者：react_loop 或 _react_loop_stream；两种入口共用同一个执行计划的方法。 # 输入：steps 列表与 user_queries 字典；输出：含 step/description/result 的结果列表。 # 这里分为两个阶段：先完整排批次，再实际跑任务。pending/scheduled 属于排批次阶段，completed 属于执行阶段。 # 例：1 无依赖，2 依赖 1，3 无依赖 → 第一批 1 和 3，第二批 2；步骤 2 会收到步骤 1 的结果。 # pending 以步骤号为键，方便后面快速判断依赖是否存在，以及删除已经排入批次的步骤。 pending = {} for step in steps: number = step.get("step") # 用 type(...) is int 也会拒绝 True/False；在 Python 中 bool 是 int 子类，但这里不允许把它当步骤号。 # 编号小于等于 0 或重复也会直接抛错；0 专门保留为“无依赖”的标记。 if type(number) is not int or number <= 0 or number in pending: raise ValueError("计划步骤序号必须为不重复的正整数") pending[number] = step for step in steps: dep = step.get("depends_on", 0) # 依赖为 0 可以通过；非 0 依赖必须在本次计划里找到对应步骤。 # 此时 pending 已收集所有步骤，所以前置步骤写在输入列表后面也可以被找到。 if type(dep) is not int or (dep != 0 and dep not in pending): raise ValueError("计划包含不存在的前置步骤") # 完整校验后才执行，避免发现循环依赖前已执行部分动作。 # 【计划执行 P2】只计算执行批次，尚未调用代理。scheduled 记录已安排的步骤。 # 例如 1→2，另有独立步骤 3，则批次为 [1,3]、[2]；不存在的依赖和环会提前报错。 batches = [] # 集合先放 0：这样所有 depends_on=0 的步骤在第一轮就满足“前置已安排”。 scheduled = {0} # 每轮选出一批就绪步骤；只要还有未安排步骤就继续。 # 此阶段不执行业务，只修改 pending/scheduled/batches 三个排程容器。 while pending: # 列表推导式逐个查看尚未安排的步骤，只保留其直接前置已在 scheduled 中的项。 # 同一轮 ready 生成后才更新 scheduled，所以链条 1→2 不会在这一轮同时被加入 ready。 ready = [step for step in pending.values() if step.get("depends_on", 0) in scheduled] # pending 还有步骤却没有任何一步可安排，说明剩余依赖形成环，例如 1→2→1 或自己依赖自己。 # 抛出异常退出，避免一边执行一边才发现计划根本无法完成。 if not ready: raise ValueError("计划存在循环依赖") batches.append(ready) for step in ready: scheduled.add(step["step"]) # 删除的是“待安排字典”里的键，不删除原始 steps 列表，也不是删除业务结果。 del pending[step["step"]] # 【计划执行 P3】依赖全部验证通过后才真正执行；completed 保存已执行结果。 completed = {} results = [] # 重新建立完整步骤索引；前面的 pending 已经排空，追溯依赖时要用这份完整索引。 by_number = {step["step"]: step for step in steps} # 外层循环按批次串行；本批 await gather 返回后才会进入下一批。 for batch in batches: tasks = [] for step in batch: # 每个步骤单独创建祖先列表，避免把另一个并行步骤的前置结果混进来。 ancestors = [] dep = step.get("depends_on", 0) # 向前追溯所有祖先，再反转为“最早前置 → 直接前置”的顺序传给当前步骤。 while dep: # 先拿直接前置的结果，再把 dep 改为它的前置；一直追到 0 结束。 # 由于批次已正确排列，执行到这里时这些前置结果都已放进 completed。 ancestors.append(completed[dep]) dep = by_number[dep].get("depends_on", 0) # 调用 async def 先得到协程对象，append 只收集它；真正调度在后面的 gather。 # reversed 把“最近前置→最早前置”倒过来，list 将反向迭代器变成可传递的列表。 tasks.append(self.execute_step(step, user_queries, list(reversed(ancestors)))) # 【计划执行 P4 → 主流程 6～9】同一批并发调度，整批返回后才进入下一批。 # 列表中的协程在 gather 时才被调度；返回结果与 tasks 的位置对应，不按完成先后排序。 # return_exceptions=True：普通子任务异常作为结果项返回，便于同批其他任务完成。 # 它不是在此处自动重试，也不会把数据库/同步模型调用全部变成并行计算。 outputs = await asyncio.gather(*tasks, return_exceptions=True) # 【计划执行 P5】异常转成失败文本并记录；后续依赖步骤仍会收到该文本继续执行。 # 这里没有依据失败内容自动取消后续步骤，也没有动态重规划。 # zip 按位置把原步骤和对应返回值配对，即便步骤 3 比步骤 1 更早完成也不会配错。 for step, output in zip(batch, outputs): if isinstance(output, Exception): output = f"执行失败：{output}" result = {"step": step["step"], "description": step.get("description", step.get("action", "")), "result": output} # 同一份 result 一方面按步骤号保存给后续依赖读取，一方面加入 results 用于最终汇总。 completed[step["step"]] = result # completed 给后续步骤查找前置结果 results.append(result) return results async def react_loop(self, steps: list, user_queries: dict) -> str: """ ReAct 循环：按规划步骤逐步执行，直接行动（Action），跳过 Thought 推理 执行流程： 1. 按依赖关系分组步骤 2. 无依赖的步骤并行执行 3. 记录每步结果（Observation） 4. 最终汇总生成连贯回复 优化：省略 Thought LLM 调用（plan 已确定动作，Thought 无额外决策价值）， 省去每步 1 次 LLM 调用，多步骤场景下显著减少响应时间。 参数： steps (list): 计划步骤列表，如： [{"step": 1, "action": "...", "intent": "flight", "depends_on": 0}, {"step": 2, "action": "...", "intent": "weather", "depends_on": 0}] user_queries (dict): 改写后的查询字典 返回值： str: 综合所有步骤结果的最终回复 """ # 【主流程 5：复杂分支 → 6～9】先完成整个计划；每步通过非流式代理调用获得结果。 # 随后进入主流程 10：多个结果由模型汇总，单个结果直接返回；并非数据库逐条推送。 # 调用者：非流式 chat 的 need_plan=True 分支。 # 输入：计划步骤和查询字典；输出：完整最终回复字符串。执行步骤并不在本方法内逐个展开，而是交给 _execute_plan。 step_results = await self._execute_plan(steps, user_queries) # 提取每步文本供单步骤返回；step_results 本身仍保留编号和说明，供多步骤汇总使用。 observations = [item["result"] for item in step_results] # ========== 所有步骤完成后的最终汇总 ========== # 多个步骤才额外调用汇总模型；一步直接返回该结果，零步返回“暂无结果”。 if len(step_results) > 1: # 多个步骤：使用专门的汇总 prompt，将各步骤结果整合成一条连贯回复 summary_chain = SmartBizTripPrompts.react_summary_prompt() | self.llm # 把编号、动作说明和结果拼成多行文本，让汇总模型知道每段结果来自哪一步。 all_obs = "\n".join([f"步骤{s['step']} ({s['description']}): {s['result']}" for s in step_results]) final_response = summary_chain.invoke({ "query": self.messages[-1]["content"] if self.messages else "", "all_observations": all_obs }).content.strip() else: # 只有一个步骤：直接使用该步骤的结果 final_response = observations[0] if observations else "暂无结果" return final_response async def chat(self, user_input: str) -> str: """ 处理用户输入的主方法 —— 整个对话流程的入口 完整流程： 1. 记录用户消息到记忆中 2. 意图识别：判断用户想做什么 3. 处理特殊情况（超出范围 / 需要追问） 4. 任务规划：判断是简单任务还是复杂任务 5. 执行：简单任务直接执行，复杂任务进入 ReAct 循环 6. 记录回复到记忆中 7. 返回最终回复 参数： user_input (str): 用户的输入内容 返回值： str: 助手的回复内容 使用示例： service = ChatService() response = await service.chat("北京明天天气怎么样？") """ # 将用户消息保存到记忆中（短期记忆、对话历史等） # 【主流程 3】先保存本轮用户消息，让意图识别能读取最新历史。 # short_term_messages 有长度限制；messages 和 conversation_history 在此处持续追加。 # 调用者：POST /api/chat。输入：用户文本；输出：return 一个完整字符串。 # 正常路径中 response 最终总是汇总后的回复；异常路径改为返回 error_message。 # 与流式版本不同，这个方法不使用 yield，HTTP 调用方必须等全部业务处理结束。 # 这一步位于下方 try 之前；先记录用户消息，再做意图识别。 # 后续读取短期历史时已经包含当前问题；写入历史不意味着后面的模型调用一定成功。 self.memory.add_message("user", user_input) self.messages.append({"role": "user", "content": user_input}) self.conversation_history += f"\nUser: {user_input}" try: # ========== 步骤1：意图识别 ========== # 【主流程 4】调用意图识别，取得意图列表、各意图查询和追问。 intents, user_queries, follow_up_message = self.intent_agent(user_input) # ========== 步骤2：处理特殊情况 ========== if "out_of_scope" in intents: # 用户输入超出系统能力范围，返回追问消息 response = follow_up_message elif follow_up_message != "": # 意图不明确，需要追问用户 response = follow_up_message else: # ========== 步骤3：任务规划 ========== # 启发式判断：单意图或独立多意图跳过 planning_agent，省去1次LLM调用 # 【主流程 5】先用规则判断是否需要规划，再按 need_plan 选择执行分支。 if self._should_skip_planning(intents): plan = {"need_plan": False, "reason": "启发式判断：任务简单，可直接执行", "steps": []} logger.info(f"跳过规划: {plan['reason']}") else: plan = self.planning_agent(intents, user_queries) # 只读取规划结论；缺字段时按 False 处理。后面 if need_plan 才决定是否执行 steps。 need_plan = plan.get("need_plan", False) logger.info(f"规划结果: need_plan={need_plan}, reason={plan.get('reason', '')}") if need_plan: # ========== 复杂任务：进入 ReAct 循环 ========== # 获取计划步骤，逐步执行 steps = plan.get("steps", []) response = await self.react_loop(steps, user_queries) else: # ========== 简单任务：直接执行 ========== # 对每个识别到的意图，直接调用对应的处理逻辑 responses = [] # 收集每个意图的回复 routed_agents = [] # 记录路由到了哪些 agent # 【主流程 6：直接分支】逐个意图等待处理完成；这里是串行，不是并发。 for intent in intents: logger.info(f"处理意图：{intent}") # 获取改写后的查询内容 # 用当前意图取对应查询；当前代码缺键时退回空字典，并没有在这里自动补齐查询。 query_str = user_queries.get(intent, {}) # 统一调用 agent 处理 result = await self._call_agent_intent(intent, query_str) responses.append(result) # 记录配置直接映射的代理；城市建议内部调用的天气代理不计入此列表。 if intent != "city_service" and conf.intent.get(intent): routed_agents.append(conf.intent[intent]) # 将多个意图的回复用空行拼接成最终回复 response = "\n\n".join(responses) if routed_agents: logger.info(f"路由到代理：{routed_agents}") # 将助手回复也保存到记忆中 # 【主流程 11】已有完整回复后再存助手消息，为下一轮提供上下文。 # 成功走完分支后保存整段 response；只保存最终拼接文本，不为每个流式片段新增一条消息。 self.memory.add_message("assistant", response) self.conversation_history += f"\nAssistant: {response}" self.messages.append({"role": "assistant", "content": response}) return response # 【异常返回】意图或规划 JSON 无法解析时记录并返回提示，不继续后面的业务步骤。 except json.JSONDecodeError as e: # 大模型返回的不是有效 JSON（虽然概率低，但偶尔会发生） logger.error(f"意图识别JSON解析失败") error_message = f"意图识别JSON解析失败：{str(e)}。请重试。" self.memory.add_message("assistant", error_message) self.messages.append({"role": "assistant", "content": error_message}) return error_message except Exception as e: # 捕获其他所有异常（网络错误、服务不可用等） logger.error(f"处理异常: {str(e)}") error_message = f"处理失败：{str(e)}。请重试。" self.memory.add_message("assistant", error_message) self.messages.append({"role": "assistant", "content": error_message}) return error_message def get_agent_cards(self) -> list: """ 获取所有代理的卡片信息 Agent Card 是 A2A 协议中描述一个 agent 能力的方式， 包含：名称、技能列表、描述、地址、状态等。 返回值： list: 代理卡片列表，每个卡片格式为： { "name": "WeatherQueryAssistant", "skills": ["天气查询: 根据城市查询天气"], "description": "天气查询助手", "url": "http://localhost:5005", "status": "在线" } """ # 【旁路：代理卡片】读取每个已登记代理的卡片，转换成网页需要的字典列表。 # status 的“在线”是这里写入的展示文本，不能证明该代理下游的 MCP/数据库可用。 # 调用者：/api/agents 接口或 get_available_tools。 # 输入：已注册的代理名称；输出：可被 JSON 序列化的字典列表，供网页绘制卡片。 cards = [] for agent_name in self.agent_network.agents.keys(): agent_card = self.agent_network.get_agent_card(agent_name) agent_url = self.agent_urls.get(agent_name, "未知地址") cards.append({ "name": agent_name, # 将每个技能对象格式化为 "技能名: 技能描述" 的字符串 "skills": [s.name + ": " + s.description for s in agent_card.skills], "description": agent_card.description, "url": agent_url, "status": "在线" }) return cards def get_memory_state(self) -> dict: """ 获取当前记忆状态的摘要信息（用于调试和展示） 返回值： dict: 包含以下信息： - short_term_messages: 最近5条对话消息 - user_profile: 用户偏好 - current_task: 当前任务上下文 - entity_history: 最近5条实体记录 """ # 【旁路：记忆展示】只取最近 5 条消息和查询记录，返回给网页侧栏；不请求模型。 # 调用者：GET /api/memory。输出包含四个字段：最近对话、偏好、当前任务、最近查询历史。 # 这里把角色标记 user/assistant 转成“用户/助手”供展示，底层记忆中仍保留原始角色标记。 return { "short_term_messages": [ {"role": "用户" if m["role"] == "user" else "助手", "content": m["content"], "timestamp": m["timestamp"]} for m in self.memory.short_term_messages[-5:] ], "user_profile": self.memory.user_profile, "current_task": self.memory.current_task, "entity_history": self.memory.entity_history[-5:] } def clear_memory(self): """ 清空所有记忆（短期记忆、用户偏好、任务上下文等） 网页调用 /api/memory/clear 接口时调用，用于重置对话状态。 memory.clear() 内部会自动清理数据库中的持久化数据。 """ # 【旁路：清空】先清理 memory 管理的数据，再清理 ChatService 自身维护的两份历史。 # 调用者：POST /api/memory/clear。没有聊天文本输入，也不返回回复内容。 # self.memory、self.messages、self.conversation_history 是三处状态，所以需要分别清理。 self.memory.clear() self.messages.clear() self.conversation_history = "" async def chat_stream(self, user_input: str): """ 流式处理用户输入的主方法 —— 决策主线与 chat 相同，最终回复通过 yield 分片返回 完整流程： 1. 记录用户消息到记忆中 2. 意图识别：判断用户想做什么 3. 处理特殊情况（超出范围 / 需要追问） 4. 任务规划：判断是简单任务还是复杂任务 5. 执行：简单任务直接执行，复杂任务进入 ReAct 循环 6. 最终回复通过 yield 分片返回，再保存拼接后的完整回复 参数： user_input (str): 用户的输入内容 Yields： str: 助手回复的文本片段 """ # 【主流程 3】先保存本轮用户消息，让意图识别能读取最新历史。 # short_term_messages 有长度限制；messages 和 conversation_history 在此处持续追加。 # 调用者：api_server.sse_generator 的 async for。输入：本轮用户文本；输出：多次 yield 的字符串。 # 顺序：记用户消息 → 识别意图 → 追问或规划/执行 → 边产出边收集回复 → 记助手消息。 # yield 只是把当前片段交给调用者，不表示本函数结束；外层继续取值时会从下一行接着运行。 # 片段传递共三层：模型/下层生成器 → 本方法 → API 的 SSE 生成器 → 浏览器。 # 这一步位于下方 try 之前；先记录用户消息，再做意图识别。 # 后续读取短期历史时已经包含当前问题；写入历史不意味着后面的模型调用一定成功。 self.memory.add_message("user", user_input) self.messages.append({"role": "user", "content": user_input}) self.conversation_history += f"\nUser: {user_input}" try: # 【主流程 4】调用意图识别，取得意图列表、各意图查询和追问。 intents, user_queries, follow_up_message = self.intent_agent(user_input) # 【主流程 5：提前返回分支】超出范围或需要补信息时直接 yield 回复，跳过代理/工具。 if "out_of_scope" in intents or follow_up_message != "": response = follow_up_message # 追问分支只产出一次整段文字；继续取值时仍会向后运行到保存助手消息的部分。 yield response else: # 【主流程 5】先用规则判断是否需要规划，再按 need_plan 选择执行分支。 if self._should_skip_planning(intents): plan = {"need_plan": False, "reason": "启发式判断：任务简单，可直接执行", "steps": []} logger.info(f"跳过规划(stream): {plan['reason']}") else: plan = self.planning_agent(intents, user_queries) # 只读取规划结论；缺字段时按 False 处理。后面 if need_plan 才决定是否执行 steps。 need_plan = plan.get("need_plan", False) logger.info(f"规划结果: need_plan={need_plan}, reason={plan.get('reason', '')}") if need_plan: steps = plan.get("steps", []) # 【主流程 10】一边转发片段，一边保留副本；结束后拼成 response 供第 11 步保存。 parts = [] async for chunk in self._react_loop_stream(steps, user_queries): parts.append(chunk) yield chunk response = "".join(parts) else: responses = [] # 【主流程 6：直接分支】逐个意图等待处理完成；这里是串行，不是并发。 for intent in intents: # 已有前一个意图结果时先输出两个换行；这里的 if 不会跳出整个意图循环。 if responses: yield "\n\n" logger.info(f"处理意图：{intent}") # 用当前意图取对应查询；当前代码缺键时退回空字典，并没有在这里自动补齐查询。 query_str = user_queries.get(intent, {}) # parts 收集当前意图片段；responses 收集各意图全文。每次进入新意图都要新建 parts。 parts = [] async for chunk in self._call_agent_intent_stream(intent, query_str): parts.append(chunk) yield chunk # 空串 join 把当前意图的片段按顺序拼成全文，加入 responses 后才处理下一个意图。 responses.append("".join(parts)) # 所有意图结束后用空行连接，形成待保存的完整助手回复。 response = "\n\n".join(responses) # 【主流程 11】已有完整回复后再存助手消息，为下一轮提供上下文。 # 成功走完分支后保存整段 response；只保存最终拼接文本，不为每个流式片段新增一条消息。 self.memory.add_message("assistant", response) self.conversation_history += f"\nAssistant: {response}" self.messages.append({"role": "assistant", "content": response}) # 【异常返回】意图或规划 JSON 无法解析时记录并返回提示，不继续后面的业务步骤。 except json.JSONDecodeError as e: logger.error("意图识别JSON解析失败") error_message = f"意图识别JSON解析失败：{str(e)}。请重试。" self.memory.add_message("assistant", error_message) self.messages.append({"role": "assistant", "content": error_message}) yield error_message except Exception as e: logger.error(f"处理异常: {str(e)}") error_message = f"处理失败：{str(e)}。请重试。" self.memory.add_message("assistant", error_message) self.messages.append({"role": "assistant", "content": error_message}) yield error_message async def _call_agent_intent_stream(self, intent: str, query_str: str): """ 流式版本的 _call_agent_intent，仅在生成最终回复时使用流式输出 调用方需要自行收集 yield 的片段拼接为完整回复 """ # 【主流程 6.1】把 intent 映射成子助手名称；city_service 走单独的城市建议分支。 # 调用者：chat_stream 的直接分支。输入：意图与查询；输出：异步生成器产生的若干字符串片段。 # 这是含 yield 的 async def，调用后得到生成器；外层通过 async for 才让它逐段运行。 # A2A 请求在这里仍先等完整 Task 返回，随后主模型 astream 整理时才分片输出。 # 查的是本地映射字典。例如 weather→WeatherQueryAssistant，train/flight→TicketAssistant。 # 没有映射时得到 None；city_service 虽不在映射里，仍会被下方单独的 if 分支处理。 agent_name = conf.intent.get(intent) if intent == "city_service": chain = SmartBizTripPrompts.city_service_prompt() | self.llm weather_info = await self._fetch_weather_for_city_service(query_str) async for chunk in chain.astream({"query": query_str, "weather_info": weather_info}): yield chunk.content if hasattr(chunk, 'content') else str(chunk) elif agent_name: if intent in ["flight", "train", "event", "car_rental", "meeting_package", "insurance"]: # 这里先记查询历史，再更新当前任务，随后才实际查询。 # 因此“查询历史里有一条记录”不代表该次业务查询已成功。 self.memory.extract_entities(intent, query_str) self.memory.update_task_context({"type": intent, "query": query_str}) # 【主流程 6.2】取得 A2A 客户端，准备携带历史和本轮查询的 Task。 agent = self.agent_network.get_agent(agent_name) # 按名称取出已注册的 A2A 客户端 # 检查是否取到了客户端对象；即使对象存在，远端服务也可能未启动，通信时仍可能抛异常。 if agent is None: logger.warning(f"未找到代理：{agent_name}") yield f"抱歉，{agent_name} 暂时不可用，请稍后重试。" else: # 流式版本拼接历史文本；非流式版本使用包含 history/query 的 JSON 文本。 # 本路径把历史和当前改写查询直接拼成对话文本，而不是非流式版本的 history/query JSON。 # 原始用户消息已在记忆中，因此追加的是用于本意图处理的改写查询。 chat_history = self.memory.get_short_term_text() + f'\nUser: {query_str}' msg = Message(content=TextContent(text=chat_history), role=MessageRole.USER) # Message 用角色和 TextContent 包装文本，to_dict 将它转换为 Task 所接收的消息字典。 # UUID 给每次代理任务生成独立标识，便于区分请求；它不是订单号或用户会话号。 task = Task(id="task-" + str(uuid.uuid4()), message=msg.to_dict()) try: ## 异步调用的调度安排 # 【主流程 6.3 → 7】经 A2A 把任务送到独立子助手进程，等待其完成工具调用。 raw_response = await asyncio.get_event_loop().run_in_executor( # None 表示使用默认线程池；lambda 是稍后由线程执行的无参数函数。 # 线程内 asyncio.run 创建事件循环并运行 send_task_async；外层 await 等待线程结果。 # 两层返回最终都汇合到 raw_response，也就是带状态和产物的 A2A Task。 None, lambda: asyncio.run(agent.send_task_async(task)) ) logger.info(f"{agent_name} 原始响应: {raw_response}") # 【主流程 9】取回 task.artifacts 中的文本，或读取状态里的追问/失败信息。 # COMPLETED 只代表代理处理结束；仍要看返回内容，不能据此认定有数据或已真实下单。 if raw_response.status.state == 'completed' and raw_response.artifacts: # 数据路径：artifacts 是产物列表 → 第一个产物的 parts 列表 → 第一个 part 的 text。 # 当前只取第一个文本 part；不会在这里遍历、合并所有产物。 agent_result = raw_response.artifacts[0]['parts'][0]['text'] elif raw_response.status.message: agent_result = raw_response.status.message.get('content', {}).get('text', str(raw_response.status.message)) else: agent_result = f"查询失败：{raw_response.status.message or '未知错误'}" except Exception as e: logger.error(f"{agent_name} 调用异常: {str(e)}") agent_result = f"{agent_name} 服务暂时不可用：{str(e)}" # 【主流程 10】按代理类型整理结果。流式版本在 astream 时持续产出文本片段。 if agent_name == "WeatherQueryAssistant": chain = SmartBizTripPrompts.summarize_weather_prompt() | self.llm elif agent_name in ("TicketQueryAssistant", "TicketAssistant", "TripAssistant"): chain = SmartBizTripPrompts.summarize_ticket_prompt() | self.llm else: yield agent_result return # astream 返回模型消息片段；取 content 后交给外层，外层再封装 SSE。 # hasattr 用于判断片段是否有 content 属性，没有时才退回 str(chunk)。 async for chunk in chain.astream({"query": query_str, "raw_response": agent_result}): yield chunk.content if hasattr(chunk, 'content') else str(chunk) else: yield "暂不支持此意图。" async def _react_loop_stream(self, steps: list, user_queries: dict): """ 流式版本的 ReAct 循环，直接行动（跳过 Thought 推理），最终汇总时流式输出 优化：与 react_loop 一致，省略 Thought LLM 调用，减少每步 1 次 LLM 请求。 调用方需要自行收集 yield 的片段拼接为完整回复。 """ # 【主流程 5：复杂分支 → 6～9】先完成整个计划；每步通过非流式代理调用获得结果。 # 随后进入主流程 10：多个结果由模型汇总，单个结果直接返回；并非数据库逐条推送。 # 调用者：chat_stream 的 need_plan=True 分支。 # 输入：计划步骤和查询字典；输出：供外层消费的文本片段。 # 先 await 全部步骤结果，再由汇总模型输出；只有一步时直接 yield 那一步的完整文本。 step_results = await self._execute_plan(steps, user_queries) # 提取每步文本供单步骤返回；step_results 本身仍保留编号和说明，供多步骤汇总使用。 observations = [item["result"] for item in step_results] # 多个步骤才额外调用汇总模型；一步直接返回该结果，零步返回“暂无结果”。 if len(step_results) > 1: summary_chain = SmartBizTripPrompts.react_summary_prompt() | self.llm # 把编号、动作说明和结果拼成多行文本，让汇总模型知道每段结果来自哪一步。 all_obs = "\n".join([f"步骤{s['step']} ({s['description']}): {s['result']}" for s in step_results]) async for chunk in summary_chain.astream({ "query": self.messages[-1]["content"] if self.messages else "", "all_observations": all_obs }): yield chunk.content if hasattr(chunk, 'content') else str(chunk) else: yield observations[0] if observations else "暂无结果"