SmartBizTrip/chat_service.py

# 本文件接续 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 "暂无结果"