![[智能对话系统架构设计文档]](images/news1.jpg)
智能对话系统架构设计文档基于 LangGraph LangChain 的 Agent 框架技术栈详解文档版本: v1.0 · 2026-06-061. 综述这不是 Chatbot是 Agent 架构1.1 从 Chatbot 到 Agent 的范式转变维度Chatbot聊天机器人Agent智能体核心模式输入 → LLM → 文本输出感知 → 规划 → 工具 → 记忆 → 反思输出产物自然语言文本结构化指令命令/工具调用/动作状态管理上下文窗口塞历史层次化记忆槽位/栈/持久化存储控制流Prompt 暗示 LLM 怎么做确定性图编排决定流程安全机制靠 LLM “自觉”硬性边界最大步数/路由锁本框架的核心设计哲学LLM 是 Agent 的大脑但不是中枢神经大脑负责推理和理解中枢神经LangGraph Policy Action负责协调执行1.2 整体架构图┌──────────────────────────────────────┐ │ LangGraph 编排器 │ │ (MessageProcessingState 状态图) │ │ │ │ START → understand → policy → action │ │ ↕ ↕ │ │ ┌────────── guard ───┘ │ │ ↓ │ │ response → END │ └──────────┬───────────────────────────┘ │ ┌─────────────────────┼─────────────────────┐ │ │ │ ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ │ 感知层 │ │ 规划层 │ │ 工具层 │ │ LLM命令 │ │ Policy │ │ Action │ │ 生成器 │ │ Ensemble │ │ 注册表 │ └───────────┘ └───────────┘ └───────────┘ │ │ │ │ ┌─────▼─────┐ │ │ │ 记忆层 │ │ │ │ Tracker │ │ │ │ Stack │ │ │ └───────────┘ │ │ │ └─────────────────┬─────────────────────────┘ │ ┌──────▼──────┐ │ 安全层 │ │ Guard Node │ └─────────────┘2. 感知层LLM 命令生成核心命题LLM 的输出不应该是文本回复而应该是结构化指令2.1 技术栈组件选型版本LLM 调用框架LangChain (langchain-openai)0.3LLM 提供商阿里云 DashScope (通义千问) / OpenAI / Anthropic-输出解析CommandParserPromptBuilder自定义实现-结构化输出Pydantic 命令模型2.0环境配置python-dotenv YAML 配置-2.2 为什么不是让 LLM 直接说话最朴素的 Agent 写法# ❌ 错误做法LLM 直接回复 response llm.invoke(用户说: 我要查订单。请回复) # 回复: 好的我来帮您查订单请问订单号是多少问题回复是文本而不是动作指令——后续代码无法判断现在该干嘛LLM 的回复风格不稳定同样的语义可能每轮输出不同格式没有结构化信息谁也不知道查哪个订单“用什么查”2.3 本框架的做法# ✅ 正确做法LLM 生成结构化命令 prompt PromptBuilder.build( user_message我要查订单, trackercurrent_tracker, # 当前对话状态 domaindomain, # 可用的槽位定义 flowsavailable_flows, # 可触发的业务Flow ) commands: List[Command] llm_generator.generate(prompt) # 输出: [ # TriggerFlow(namequery_order_detail), # SetSlot(nameuser_id, value123) # ]2.4 命令类型体系所有 LLM 的输出被统一映射到这几种命令类型class Command(ABC): 所有命令的基类 class TriggerFlow(Command): 触发一个业务流程 flow_name: str slots: Dict[str, Any] class SetSlot(Command): 设置槽位值 name: str value: Any class ChitchatCommand(Command): 闲聊回复当无法匹配业务时 text: str class CannotHandleCommand(Command): 无法处理 text: str class ErrorCommand(Command): 系统错误 message: str2.5 架构价值问题LLM 直接回复本方案输出可解析性❌ 文本需 NLP 后处理✅ 结构化命令直接处理格式稳定性❌ 每轮可能不同✅ 统一类型体系模型可替换⚠️ 需微调 Prompt✅ 换模型不改逻辑可测试性❌ 难以断言文本✅assert isinstance(cmd, TriggerFlow)链路追踪❌ 全是字符串✅ 每个命令有类型和参数一句话感知层的核心价值是把LLM 从发言人降级为翻译官——它负责把用户自然语言翻译成系统内部可执行的统一指令格式。3. 规划层PolicyEnsemble Flow核心命题Agent 的下一步动作应该由策略算法决定而不是让 LLM 再猜一次3.1 技术栈组件选型版本策略框架自定义PolicyEnsembleChain of Responsibility 模式-流程引擎FlowPolicy YAML 定义-知识策略EnterpriseSearchPolicyRAG 式知识检索-配置语言YAML (ruamel.yaml/pyyaml)6.0状态持有DialogueStack栈式 Flow 管理-3.2 为什么不是让 LLM 无脑决策# ❌ 错误做法每轮都问 LLM 下一步做什么 while True: action llm.invoke(看对话历史决定下一步动作) execute(action)问题LLM 没有任务完成的概念——它不知道 Flow 何时结束LLM 没有确定性路由能力——同样的场景可能每次决策不同LLM 容易被带偏——用户说别查了它可能真的不查了没有优先级——业务逻辑和闲聊混在一起3.3 本框架的做法class PolicyEnsemble: 策略集成器 - 链式调用多个策略 def __init__(self, policies: List[Policy]): self.policies policies async def predict(self, tracker, domain, flows) - PolicyPrediction: for policy in self.policies: prediction await policy.predict(tracker, domain, flows) if prediction.confidence self.THRESHOLD: logger.info(f策略 {policy.name} 决策: {prediction.action}) return prediction # 所有策略都无法决策 → 默认等待用户输入 return PolicyPrediction(actionACTION_LISTEN, confidence1.0)3.4 内置策略体系Policy A: FlowPolicy业务流程驱动其工作原理类似于一个图遍历器当前 Flow: modify_order_receive_info 当前步骤索引: 3 (collect: receiver_name) ↓ FlowPolicy 读取第 3 步定义: - 动作: utter_ask_receiver_name (向用户问姓名) - 下一步: collect 接收用户输入 ↓ 决策输出: action utter_ask_receiver_name metadata {slot_to_collect: receiver_name}YAML 定义 Flow 步骤就是预编程的规划图flows: modify_order_receive_info: steps: - collect: order_id # 步骤0: 收集订单号 - action: action_get_order_detail # 步骤1: 查订单详情 - collect: receiver_name # 步骤2: 收集新姓名 ask_before_filling: true # 每次都重新问 - action: action_update_order # 步骤3: 更新数据库Policy B: EnterpriseSearchPolicy知识检索驱动当 FlowPolicy 无法匹配时接管无匹配 Flow ↓ EnterpriseSearchPolicy 工作: 1. 检查是否有知识库Neo4j 图/向量库 2. 有 → 检索相关内容生成回复 3. 没有 → 走闲聊或降级 ↓ 决策输出: action action_chitchat_response 或 action action_search_enterprise_knowledge链式决策流程图用户: 帮我改一下收货地址 │ ▼ PolicyEnsemble.predict() │ ├── FlowPolicy │ ├── 检查是否有匹配的 Flow → ✅ modify_order_receive_info │ ├── 检查当前 Flow 进度 → 步骤 0 (collect order_id) │ └── 决策 → action_ask_order_id_before_delivered (置信度 1.0) │ └── EnterpriseSearchPolicy (未执行因为 FlowPolicy 已高置信度决策)3.5 为什么用 YAML 定义 Flow优势说明业务人员可编辑不需要会 Python写 YAML 就能定义对话流程版本可追踪YAML 是文本文件可以 Git diff条件分支清晰if: slots.order_id ! false直观可读运行时热加载修改 YAML 理论上可以不用重启当前实现需重载 Agent工具链丰富YAML lint、schema 校验成熟一句话规划层的核心价值是确定性逻辑 LLM 弹性的混合架构——FlowPolicy 提供可靠的业务逻辑骨架EnterpriseSearchPolicy 提供灵活的 AI 弹性填充。4. 工具层Action 注册表核心命题Agent 的工具应该是可注册的而不是硬编码 if-else4.1 技术栈组件选型工具框架自定义Action基类 全局注册表发现机制importlib动态模块加载 目录扫描数据模型ActionResult(dataclass)对话访问DialogueStateTracker作为 Action 的上下文4.2 为什么不是硬编码 if-else# ❌ 错误做法if-else 硬编码动作 async def execute_action(action_name, tracker, domain): if action_name action_get_order_detail: # 查数据库... return 订单详情是... elif action_name action_cancel_order: # 取消订单... return 订单已取消 elif action_name utter_greet: return 你好 # ... 每新增一个动作加一个 elif问题每加一个功能就要改框架代码多人协作时 merge 冲突无法插件化分发动作个数膨胀后一个文件几千行4.3 本框架的做法全局注册表# 全局单例注册表 _action_registry: Dict[str, Action] {} def register_action(action: Action) - None: 注册动作到全局注册表 _action_registry[action.name] action def get_action(name: str) - Optional[Action]: 根据名称获取动作 return _action_registry.get(name)自动发现机制def _load_custom_actions(actions_path: Path) - List[str]: 自动扫描 actions/ 目录发现并注册 Action 子类 for py_file in actions_path.glob(*.py): if py_file.name.startswith(_): continue # 跳过 __init__.py # 动态导入 spec importlib.util.spec_from_file_location(...) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # 扫描类定义找到继承 Action 的子类 for name, obj in inspect.getmembers(module, inspect.isclass): if issubclass(obj, Action) and obj is not Action: action_instance obj() register_action(action_instance)Action 定义示例class ActionGetOrderDetail(Action): 查询订单详情 property def name(self) - str: return action_get_order_detail async def run(self, tracker, domain, **kwargs) - ActionResult: # 从 tracker 获取上下文 order_id tracker.get_slot(order_id) user_id tracker.get_slot(user_id) # 执行实际操作 order await db.query_order(order_id, user_id) # 返回结构化响应 result ActionResult() result.add_response( textf订单 #{order.id} 的详情\n f状态: {order.status}\n f金额: ¥{order.amount}\n f收货地址: {order.address} ) return result4.4 Action 类型一览类型命名规范示例作用模板响应utter_xxxutter_greet回复预定义模板文本收集槽位utter_ask_xxxutter_ask_order_id向用户询问槽位值自定义动作action_xxxaction_get_order_detail执行任意代码逻辑特殊动作action_listen-等待用户输入闲聊action_chitchat_response-LLM 生成自由回复4.5 架构价值能力硬编码 if-else注册表 自动发现扩展性改框架源码新建文件就行可插拔❌✅ 复制 actions/ 目录就能迁移测试性需测整个文件单测一个 Action 类团队协作改同一文件冲突各写各的文件复用性复制粘贴pip 包分发一句话工具层的核心价值是注册而非硬编码的插件化架构——让 Agent 的工具扩展从侵入式变成声明式。5. 记忆层Tracker DialogueStack Store核心命题Agent 的记忆不能依赖 LLM 的上下文窗口需要独立的层次化记忆系统5.1 技术栈组件选型作用会话状态DialogueStateTracker(自定义)管理槽位、事件、消息历史Flow 上下文DialogueStackStackFrame管理 Flow 嵌套和中断持久化TrackerStore(Memory / JSON / MySQL)跨会话存储外部记忆Neo4j 图数据库 / 向量数据库RAG 式企业知识检索5.2 为什么不是塞上下文窗口# ❌ 错误做法把历史消息全塞进 Prompt history \n.join([f{m.role}: {m.text} for m in last_50_messages]) response llm.invoke(f{history}\n用户: 我的订单是什么状态)pythonCopy问题不可靠LLM 可能忽略前面的重要信息成本高每轮都重复传所有历史token 消耗线性增长没有关键信息概念用户说过的我的订单号是 123和今天天气不错权重一样没有持久化会话结束就丢失再来要重问5.3 本框架的层次化记忆记忆层级金字塔: ┌─────────────────────┐ │ 企业知识库 (RAG) │ ← 外部知识 │ Neo4j / 向量搜索 │ 生命周期: 永久 └─────────────────────┘ ↑ ┌─────────────────────┐ │ TrackerStore (持久化)│ ← 跨会话 │ JSON / MySQL │ 生命周期: 永久 └─────────────────────┘ ↑ ┌─────────────────────┐ │ DialogueStateTracker │ ← 会话内 │ 槽位 事件 消息 │ 生命周期: 会话起止 └─────────────────────┘ ↑ ┌─────────────────────┐ │ DialogueStack │ ← Flow 内 │ Flow 嵌套/中断/恢复 │ 生命周期: Flow 起止 └─────────────────────┘ ↑ ┌─────────────────────┐ │ LangGraph State │ ← 单轮请求内 │ 中间结果缓存 │ 生命周期: 一次 invoke └─────────────────────┘5.4 关键设计槽位系统槽位是 Agent 的长时记忆单元slots: user_id: type: text mappings: - type: from_llm # LLM 从用户输入提取 description: 用户ID order_id: type: text mappings: - type: controlled # Action 代码控制设置 description: 订单编号 current_flow: type: categorical values: - query_order - modify_address - cancel_order mappings: - type: from_llm # LLM 自动识别当前流程设计价值用户: 我要查订单 123 │ ▼ LLM 生成命令: [SetSlot(user_idabc), SetSlot(order_id123), TriggerFlow(query_order)] │ ▼ Tracker 记录: user_id abc ← 从会话开始就记住了 order_id 123 ← 后续 Action 直接读取 current_flow query_order ← 不用每轮重新问 LLM │ ▼ 十轮对话后... 用户: 改成送到新地址 │ ▼ FlowPolicy 检查: 当前在 order_detail Flow 中 → 步骤 5: 修改地址 (直接取 user_id 和 order_id) → 无需 LLM 重新提取这些信息5.5 对话栈的安全中断与恢复class DialogueStack: 栈式 Flow 管理支持嵌套 frames: List[StackFrame] # 栈顶 当前活动的 Flow # # 例如: # [0] main_flow (query_order) # → 流程中触发了售后服务 # [1] postsale_flow (refund_request) # → 售后完成, pop() # [0] main_flow (继续原来的 query_order)class FlowStackFrame: 一个 Flow 的执行上下文 flow_name: str # 当前 Flow 名称 step_index: int # 执行到第几步 collected_slots: dict # 已收集的槽位 state: FrameState # ACTIVE / PAUSED / COMPLETED一句话记忆层的核心价值是“把记忆从 LLM 的上下文窗口解放出来”——Agent 的记忆是结构化的、分层的、可持久化的不是聊天历史的一堆文字。6. 安全层Guard Node 条件边核心命题Agent 的安全措施不能依赖 LLM 的自觉需要硬的工程化边界6.1 技术栈组件选型循环防护guard_nodemax_actions计数器流程控制LangGraphadd_conditional_edges确定性条件边限流action_count运行时计数异常回退try/exceptActionResult(successFalse)优雅降级6.2 为什么不是靠 LLM 自觉# ❌ 错误做法在 Prompt 里告诉 LLM 别死循环 prompt 你是客服助手。 注意不要无限循环每个问题最多回答3次。 如果用户说结束就结束。 问题LLM 可能忽略指令越狱、prompt 注入LLM 不知道已经回答了 3 次——它没有计数器LLM 的结束判断不可靠——用户说再说几句它可能就继续了6.3 本框架的硬性安全边界安全机制 1Guard Node —— 循环计数器async def guard_node(state: MessageProcessingState) - Dict[str, Any]: 保护节点检查循环次数超限强制终止 action_count state.get(action_count, 0) max_actions state.get(max_actions, 10) if action_count max_actions: # 强制终止不管 LLM 还想干嘛 logger.warning( f达到最大动作数限制 ({action_count}/{max_actions})强制终止 ) return { is_finished: True, error: f达到最大动作数限制: {max_actions}, } return {} # 正常继续安全机制 2条件边 —— 确定性路由# 这条路由是 Python 代码不是 LLM Prompt # 它决定了什么时候该结束 def should_continue(state: Dict[str, Any]) - Literal[policy, response]: 决定是继续循环还是结束 is_finished state.get(is_finished, False) action_count state.get(action_count, 0) max_actions state.get(max_actions, 10) # 条件1任务完成 if is_finished: return response # → 结束本轮 # 条件2超步数限制 if action_count max_actions: return response # → 强制结束 # 否则继续 return policy # → 回到规划层安全机制 3Action 级别的异常隔离async def action_node(state): try: action get_action(action_name) if action is None: # 动作不存在 → 优雅跳过不崩溃 return {current_action_result: ActionResult(successFalse)} result await action.run(tracker, domain, **kwargs) # 即使 Action 内部报错也不会使整个 Agent 崩溃 except Exception as e: logger.error(f动作执行失败: {e}) return { error: str(e), current_action_result: ActionResult(successFalse), # 继续执行不会挂掉整个流程 }6.4 多层安全体系安全层防护内容实现方式图级别死循环guard_nodemax_actions节点级别节点崩溃try/except包围每个节点逻辑动作级别Action 异常每个 Action 独立捕获异常调用级别LLM 超时LLMClient配置timeout参数外部级别API 调用失败HTTP 请求 重试机制一句话安全层的核心价值是“让 Agent 的每一次执行都在可控范围内”——有计数器、有边界、有优雅降级而不是让 LLM 自己控制自己。7. 各层技术栈总览架构层核心技术辅助技术设计模式编排层LangGraphStateGraph / CompiledStateGraph图编排DAG感知层LangChain 自定义 CommandParserPydantic / DashScope SDKStrategy策略模式规划层PolicyEnsemble / YAML Flowruamel.yaml / pyyamlChain of Responsibility工具层Action 注册表importlib 动态加载Registry Plugin记忆层DialogueStateTracker / DialogueStackSQLAlchemy / Neo4j DriverMemento Stack安全层Guard Node / 条件边-Guard (防御性编程)通信层FastAPI / uvicornpython-socketio / WebSocketChannel Adapter配置层YAML ${ENV_VAR} 替换python-dotenvConfiguration Injection部署层pip install / setup.pyClick CLICLI Facade技术栈选择原则选型决策树: 这个能力需要确定性吗 ├── 是 → 用 Python 代码条件边、Guard、Action └── 否 → 用 LLM命令生成、闲聊、知识检索 │ ▼ 这个数据需要持久化吗 ├── 是 → TrackerStore / MySQL / Neo4j └── 否 → 内存LangGraph State / DialogueStack │ ▼ 这个配置需要用户修改吗 ├── 是 → YAML 配置Domain / Flow / endpoints └── 否 → Python 常量constants.pyCopy8. 延伸通向 Multi-Agent 的架构基础该框架的架构设计使其天然适合扩展为Multi-Agent 系统。8.1 当前架构已具备的条件当前组件Multi-Agent 映射改动量PolicyEnsemble多策略链式调用→ Agent 路由分发器小DialogueStack栈式 Flow 管理→ 跨 Agent 调用栈零Action注册表工具查找→ Agent 服务发现小TrackerStore会话持久化→ 全局会话存储零YAML Flow 定义→ Agent 间的通信协议中8.2 可能的 Multi-Agent 架构┌─────────────────────────────┐ │ Supervisor Agent │ │ 意图识别 任务路由 │ │ (使用当前 PolicyEnsemble 机制) │ └──────────┬──────────────────┘ │ ┌────────────────────┼────────────────────┐ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ 订单 Agent │ │ 物流 Agent │ │ 售后 Agent │ │ │ │ │ │ │ │ 自身 Flow │ │ 自身 Flow │ │ 自身 Flow │ │ 自身 Action │ │ 自身 Action │ │ 自身 Action │ │ 自身 Domain │ │ 自身 Domain │ │ 自身 Domain │ │ 自身 Tracker │ │ 自身 Tracker │ │ 自身 Tracker │ └──────────────┘ └──────────────┘ └──────────────┘8.3 架构优势总结单一 Agent Multi-Agent ┌─────────────┐ ┌─────────────────────┐ │ Policy A │ │ Supervisor Policy │ │ Policy B │ → │ ├─ 订单子 Agent │ │ Action 1-N │ │ ├─ 物流子 Agent │ │ Tracker │ │ ├─ 售后子 Agent │ │ Store │ │ └─ 知识库子 Agent │ └─────────────┘ │ 全局 Store / 栈 │ └─────────────────────┘ 改动: 结构不变把策略和Action换成子 AgentCopy附录关键文件索引模块核心文件职责图编排agent/graph/builder.pyLangGraph 图构建图状态agent/graph/state.py状态定义条件边agent/graph/edges.py路由逻辑感知节点agent/graph/nodes/understand.pyLLM 命令生成规划节点agent/graph/nodes/policy.py策略预测工具节点agent/graph/nodes/action.pyAction 执行安全节点agent/graph/nodes/guard.py循环防护响应节点agent/graph/nodes/response.py结果收集Agent 主类agent/agent.pyAgent 生命周期管理Action 系统agent/actions.py注册表 内置动作消息处理器agent/message_processor.py旧版备用处理器LLM 生成器dialogue_understanding/generator/llm_generator.pyLLM 调用 命令解析命令定义dialogue_understanding/commands/所有命令类型对话栈dialogue_understanding/stack/dialogue_stack.pyFlow 嵌套管理策略集成policies/policy_ensemble.py链式策略调用Flow 策略policies/flow_policy.py业务流程推进知识策略policies/enterprise_search_policy.pyRAG 检索决策Trackercore/tracker.py会话状态追踪存储core/stores/持久化存储配置shared/config.py全量配置加载LLM 客户端shared/llm/多模型客户端抽象电商 Demoecs_demo/完整电商客服示例
[智能对话系统架构设计文档]
智能对话系统架构设计文档基于 LangGraph LangChain 的 Agent 框架技术栈详解文档版本: v1.0 · 2026-06-061. 综述这不是 Chatbot是 Agent 架构1.1 从 Chatbot 到 Agent 的范式转变维度Chatbot聊天机器人Agent智能体核心模式输入 → LLM → 文本输出感知 → 规划 → 工具 → 记忆 → 反思输出产物自然语言文本结构化指令命令/工具调用/动作状态管理上下文窗口塞历史层次化记忆槽位/栈/持久化存储控制流Prompt 暗示 LLM 怎么做确定性图编排决定流程安全机制靠 LLM “自觉”硬性边界最大步数/路由锁本框架的核心设计哲学LLM 是 Agent 的大脑但不是中枢神经大脑负责推理和理解中枢神经LangGraph Policy Action负责协调执行1.2 整体架构图┌──────────────────────────────────────┐ │ LangGraph 编排器 │ │ (MessageProcessingState 状态图) │ │ │ │ START → understand → policy → action │ │ ↕ ↕ │ │ ┌────────── guard ───┘ │ │ ↓ │ │ response → END │ └──────────┬───────────────────────────┘ │ ┌─────────────────────┼─────────────────────┐ │ │ │ ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ │ 感知层 │ │ 规划层 │ │ 工具层 │ │ LLM命令 │ │ Policy │ │ Action │ │ 生成器 │ │ Ensemble │ │ 注册表 │ └───────────┘ └───────────┘ └───────────┘ │ │ │ │ ┌─────▼─────┐ │ │ │ 记忆层 │ │ │ │ Tracker │ │ │ │ Stack │ │ │ └───────────┘ │ │ │ └─────────────────┬─────────────────────────┘ │ ┌──────▼──────┐ │ 安全层 │ │ Guard Node │ └─────────────┘2. 感知层LLM 命令生成核心命题LLM 的输出不应该是文本回复而应该是结构化指令2.1 技术栈组件选型版本LLM 调用框架LangChain (langchain-openai)0.3LLM 提供商阿里云 DashScope (通义千问) / OpenAI / Anthropic-输出解析CommandParserPromptBuilder自定义实现-结构化输出Pydantic 命令模型2.0环境配置python-dotenv YAML 配置-2.2 为什么不是让 LLM 直接说话最朴素的 Agent 写法# ❌ 错误做法LLM 直接回复 response llm.invoke(用户说: 我要查订单。请回复) # 回复: 好的我来帮您查订单请问订单号是多少问题回复是文本而不是动作指令——后续代码无法判断现在该干嘛LLM 的回复风格不稳定同样的语义可能每轮输出不同格式没有结构化信息谁也不知道查哪个订单“用什么查”2.3 本框架的做法# ✅ 正确做法LLM 生成结构化命令 prompt PromptBuilder.build( user_message我要查订单, trackercurrent_tracker, # 当前对话状态 domaindomain, # 可用的槽位定义 flowsavailable_flows, # 可触发的业务Flow ) commands: List[Command] llm_generator.generate(prompt) # 输出: [ # TriggerFlow(namequery_order_detail), # SetSlot(nameuser_id, value123) # ]2.4 命令类型体系所有 LLM 的输出被统一映射到这几种命令类型class Command(ABC): 所有命令的基类 class TriggerFlow(Command): 触发一个业务流程 flow_name: str slots: Dict[str, Any] class SetSlot(Command): 设置槽位值 name: str value: Any class ChitchatCommand(Command): 闲聊回复当无法匹配业务时 text: str class CannotHandleCommand(Command): 无法处理 text: str class ErrorCommand(Command): 系统错误 message: str2.5 架构价值问题LLM 直接回复本方案输出可解析性❌ 文本需 NLP 后处理✅ 结构化命令直接处理格式稳定性❌ 每轮可能不同✅ 统一类型体系模型可替换⚠️ 需微调 Prompt✅ 换模型不改逻辑可测试性❌ 难以断言文本✅assert isinstance(cmd, TriggerFlow)链路追踪❌ 全是字符串✅ 每个命令有类型和参数一句话感知层的核心价值是把LLM 从发言人降级为翻译官——它负责把用户自然语言翻译成系统内部可执行的统一指令格式。3. 规划层PolicyEnsemble Flow核心命题Agent 的下一步动作应该由策略算法决定而不是让 LLM 再猜一次3.1 技术栈组件选型版本策略框架自定义PolicyEnsembleChain of Responsibility 模式-流程引擎FlowPolicy YAML 定义-知识策略EnterpriseSearchPolicyRAG 式知识检索-配置语言YAML (ruamel.yaml/pyyaml)6.0状态持有DialogueStack栈式 Flow 管理-3.2 为什么不是让 LLM 无脑决策# ❌ 错误做法每轮都问 LLM 下一步做什么 while True: action llm.invoke(看对话历史决定下一步动作) execute(action)问题LLM 没有任务完成的概念——它不知道 Flow 何时结束LLM 没有确定性路由能力——同样的场景可能每次决策不同LLM 容易被带偏——用户说别查了它可能真的不查了没有优先级——业务逻辑和闲聊混在一起3.3 本框架的做法class PolicyEnsemble: 策略集成器 - 链式调用多个策略 def __init__(self, policies: List[Policy]): self.policies policies async def predict(self, tracker, domain, flows) - PolicyPrediction: for policy in self.policies: prediction await policy.predict(tracker, domain, flows) if prediction.confidence self.THRESHOLD: logger.info(f策略 {policy.name} 决策: {prediction.action}) return prediction # 所有策略都无法决策 → 默认等待用户输入 return PolicyPrediction(actionACTION_LISTEN, confidence1.0)3.4 内置策略体系Policy A: FlowPolicy业务流程驱动其工作原理类似于一个图遍历器当前 Flow: modify_order_receive_info 当前步骤索引: 3 (collect: receiver_name) ↓ FlowPolicy 读取第 3 步定义: - 动作: utter_ask_receiver_name (向用户问姓名) - 下一步: collect 接收用户输入 ↓ 决策输出: action utter_ask_receiver_name metadata {slot_to_collect: receiver_name}YAML 定义 Flow 步骤就是预编程的规划图flows: modify_order_receive_info: steps: - collect: order_id # 步骤0: 收集订单号 - action: action_get_order_detail # 步骤1: 查订单详情 - collect: receiver_name # 步骤2: 收集新姓名 ask_before_filling: true # 每次都重新问 - action: action_update_order # 步骤3: 更新数据库Policy B: EnterpriseSearchPolicy知识检索驱动当 FlowPolicy 无法匹配时接管无匹配 Flow ↓ EnterpriseSearchPolicy 工作: 1. 检查是否有知识库Neo4j 图/向量库 2. 有 → 检索相关内容生成回复 3. 没有 → 走闲聊或降级 ↓ 决策输出: action action_chitchat_response 或 action action_search_enterprise_knowledge链式决策流程图用户: 帮我改一下收货地址 │ ▼ PolicyEnsemble.predict() │ ├── FlowPolicy │ ├── 检查是否有匹配的 Flow → ✅ modify_order_receive_info │ ├── 检查当前 Flow 进度 → 步骤 0 (collect order_id) │ └── 决策 → action_ask_order_id_before_delivered (置信度 1.0) │ └── EnterpriseSearchPolicy (未执行因为 FlowPolicy 已高置信度决策)3.5 为什么用 YAML 定义 Flow优势说明业务人员可编辑不需要会 Python写 YAML 就能定义对话流程版本可追踪YAML 是文本文件可以 Git diff条件分支清晰if: slots.order_id ! false直观可读运行时热加载修改 YAML 理论上可以不用重启当前实现需重载 Agent工具链丰富YAML lint、schema 校验成熟一句话规划层的核心价值是确定性逻辑 LLM 弹性的混合架构——FlowPolicy 提供可靠的业务逻辑骨架EnterpriseSearchPolicy 提供灵活的 AI 弹性填充。4. 工具层Action 注册表核心命题Agent 的工具应该是可注册的而不是硬编码 if-else4.1 技术栈组件选型工具框架自定义Action基类 全局注册表发现机制importlib动态模块加载 目录扫描数据模型ActionResult(dataclass)对话访问DialogueStateTracker作为 Action 的上下文4.2 为什么不是硬编码 if-else# ❌ 错误做法if-else 硬编码动作 async def execute_action(action_name, tracker, domain): if action_name action_get_order_detail: # 查数据库... return 订单详情是... elif action_name action_cancel_order: # 取消订单... return 订单已取消 elif action_name utter_greet: return 你好 # ... 每新增一个动作加一个 elif问题每加一个功能就要改框架代码多人协作时 merge 冲突无法插件化分发动作个数膨胀后一个文件几千行4.3 本框架的做法全局注册表# 全局单例注册表 _action_registry: Dict[str, Action] {} def register_action(action: Action) - None: 注册动作到全局注册表 _action_registry[action.name] action def get_action(name: str) - Optional[Action]: 根据名称获取动作 return _action_registry.get(name)自动发现机制def _load_custom_actions(actions_path: Path) - List[str]: 自动扫描 actions/ 目录发现并注册 Action 子类 for py_file in actions_path.glob(*.py): if py_file.name.startswith(_): continue # 跳过 __init__.py # 动态导入 spec importlib.util.spec_from_file_location(...) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # 扫描类定义找到继承 Action 的子类 for name, obj in inspect.getmembers(module, inspect.isclass): if issubclass(obj, Action) and obj is not Action: action_instance obj() register_action(action_instance)Action 定义示例class ActionGetOrderDetail(Action): 查询订单详情 property def name(self) - str: return action_get_order_detail async def run(self, tracker, domain, **kwargs) - ActionResult: # 从 tracker 获取上下文 order_id tracker.get_slot(order_id) user_id tracker.get_slot(user_id) # 执行实际操作 order await db.query_order(order_id, user_id) # 返回结构化响应 result ActionResult() result.add_response( textf订单 #{order.id} 的详情\n f状态: {order.status}\n f金额: ¥{order.amount}\n f收货地址: {order.address} ) return result4.4 Action 类型一览类型命名规范示例作用模板响应utter_xxxutter_greet回复预定义模板文本收集槽位utter_ask_xxxutter_ask_order_id向用户询问槽位值自定义动作action_xxxaction_get_order_detail执行任意代码逻辑特殊动作action_listen-等待用户输入闲聊action_chitchat_response-LLM 生成自由回复4.5 架构价值能力硬编码 if-else注册表 自动发现扩展性改框架源码新建文件就行可插拔❌✅ 复制 actions/ 目录就能迁移测试性需测整个文件单测一个 Action 类团队协作改同一文件冲突各写各的文件复用性复制粘贴pip 包分发一句话工具层的核心价值是注册而非硬编码的插件化架构——让 Agent 的工具扩展从侵入式变成声明式。5. 记忆层Tracker DialogueStack Store核心命题Agent 的记忆不能依赖 LLM 的上下文窗口需要独立的层次化记忆系统5.1 技术栈组件选型作用会话状态DialogueStateTracker(自定义)管理槽位、事件、消息历史Flow 上下文DialogueStackStackFrame管理 Flow 嵌套和中断持久化TrackerStore(Memory / JSON / MySQL)跨会话存储外部记忆Neo4j 图数据库 / 向量数据库RAG 式企业知识检索5.2 为什么不是塞上下文窗口# ❌ 错误做法把历史消息全塞进 Prompt history \n.join([f{m.role}: {m.text} for m in last_50_messages]) response llm.invoke(f{history}\n用户: 我的订单是什么状态)pythonCopy问题不可靠LLM 可能忽略前面的重要信息成本高每轮都重复传所有历史token 消耗线性增长没有关键信息概念用户说过的我的订单号是 123和今天天气不错权重一样没有持久化会话结束就丢失再来要重问5.3 本框架的层次化记忆记忆层级金字塔: ┌─────────────────────┐ │ 企业知识库 (RAG) │ ← 外部知识 │ Neo4j / 向量搜索 │ 生命周期: 永久 └─────────────────────┘ ↑ ┌─────────────────────┐ │ TrackerStore (持久化)│ ← 跨会话 │ JSON / MySQL │ 生命周期: 永久 └─────────────────────┘ ↑ ┌─────────────────────┐ │ DialogueStateTracker │ ← 会话内 │ 槽位 事件 消息 │ 生命周期: 会话起止 └─────────────────────┘ ↑ ┌─────────────────────┐ │ DialogueStack │ ← Flow 内 │ Flow 嵌套/中断/恢复 │ 生命周期: Flow 起止 └─────────────────────┘ ↑ ┌─────────────────────┐ │ LangGraph State │ ← 单轮请求内 │ 中间结果缓存 │ 生命周期: 一次 invoke └─────────────────────┘5.4 关键设计槽位系统槽位是 Agent 的长时记忆单元slots: user_id: type: text mappings: - type: from_llm # LLM 从用户输入提取 description: 用户ID order_id: type: text mappings: - type: controlled # Action 代码控制设置 description: 订单编号 current_flow: type: categorical values: - query_order - modify_address - cancel_order mappings: - type: from_llm # LLM 自动识别当前流程设计价值用户: 我要查订单 123 │ ▼ LLM 生成命令: [SetSlot(user_idabc), SetSlot(order_id123), TriggerFlow(query_order)] │ ▼ Tracker 记录: user_id abc ← 从会话开始就记住了 order_id 123 ← 后续 Action 直接读取 current_flow query_order ← 不用每轮重新问 LLM │ ▼ 十轮对话后... 用户: 改成送到新地址 │ ▼ FlowPolicy 检查: 当前在 order_detail Flow 中 → 步骤 5: 修改地址 (直接取 user_id 和 order_id) → 无需 LLM 重新提取这些信息5.5 对话栈的安全中断与恢复class DialogueStack: 栈式 Flow 管理支持嵌套 frames: List[StackFrame] # 栈顶 当前活动的 Flow # # 例如: # [0] main_flow (query_order) # → 流程中触发了售后服务 # [1] postsale_flow (refund_request) # → 售后完成, pop() # [0] main_flow (继续原来的 query_order)class FlowStackFrame: 一个 Flow 的执行上下文 flow_name: str # 当前 Flow 名称 step_index: int # 执行到第几步 collected_slots: dict # 已收集的槽位 state: FrameState # ACTIVE / PAUSED / COMPLETED一句话记忆层的核心价值是“把记忆从 LLM 的上下文窗口解放出来”——Agent 的记忆是结构化的、分层的、可持久化的不是聊天历史的一堆文字。6. 安全层Guard Node 条件边核心命题Agent 的安全措施不能依赖 LLM 的自觉需要硬的工程化边界6.1 技术栈组件选型循环防护guard_nodemax_actions计数器流程控制LangGraphadd_conditional_edges确定性条件边限流action_count运行时计数异常回退try/exceptActionResult(successFalse)优雅降级6.2 为什么不是靠 LLM 自觉# ❌ 错误做法在 Prompt 里告诉 LLM 别死循环 prompt 你是客服助手。 注意不要无限循环每个问题最多回答3次。 如果用户说结束就结束。 问题LLM 可能忽略指令越狱、prompt 注入LLM 不知道已经回答了 3 次——它没有计数器LLM 的结束判断不可靠——用户说再说几句它可能就继续了6.3 本框架的硬性安全边界安全机制 1Guard Node —— 循环计数器async def guard_node(state: MessageProcessingState) - Dict[str, Any]: 保护节点检查循环次数超限强制终止 action_count state.get(action_count, 0) max_actions state.get(max_actions, 10) if action_count max_actions: # 强制终止不管 LLM 还想干嘛 logger.warning( f达到最大动作数限制 ({action_count}/{max_actions})强制终止 ) return { is_finished: True, error: f达到最大动作数限制: {max_actions}, } return {} # 正常继续安全机制 2条件边 —— 确定性路由# 这条路由是 Python 代码不是 LLM Prompt # 它决定了什么时候该结束 def should_continue(state: Dict[str, Any]) - Literal[policy, response]: 决定是继续循环还是结束 is_finished state.get(is_finished, False) action_count state.get(action_count, 0) max_actions state.get(max_actions, 10) # 条件1任务完成 if is_finished: return response # → 结束本轮 # 条件2超步数限制 if action_count max_actions: return response # → 强制结束 # 否则继续 return policy # → 回到规划层安全机制 3Action 级别的异常隔离async def action_node(state): try: action get_action(action_name) if action is None: # 动作不存在 → 优雅跳过不崩溃 return {current_action_result: ActionResult(successFalse)} result await action.run(tracker, domain, **kwargs) # 即使 Action 内部报错也不会使整个 Agent 崩溃 except Exception as e: logger.error(f动作执行失败: {e}) return { error: str(e), current_action_result: ActionResult(successFalse), # 继续执行不会挂掉整个流程 }6.4 多层安全体系安全层防护内容实现方式图级别死循环guard_nodemax_actions节点级别节点崩溃try/except包围每个节点逻辑动作级别Action 异常每个 Action 独立捕获异常调用级别LLM 超时LLMClient配置timeout参数外部级别API 调用失败HTTP 请求 重试机制一句话安全层的核心价值是“让 Agent 的每一次执行都在可控范围内”——有计数器、有边界、有优雅降级而不是让 LLM 自己控制自己。7. 各层技术栈总览架构层核心技术辅助技术设计模式编排层LangGraphStateGraph / CompiledStateGraph图编排DAG感知层LangChain 自定义 CommandParserPydantic / DashScope SDKStrategy策略模式规划层PolicyEnsemble / YAML Flowruamel.yaml / pyyamlChain of Responsibility工具层Action 注册表importlib 动态加载Registry Plugin记忆层DialogueStateTracker / DialogueStackSQLAlchemy / Neo4j DriverMemento Stack安全层Guard Node / 条件边-Guard (防御性编程)通信层FastAPI / uvicornpython-socketio / WebSocketChannel Adapter配置层YAML ${ENV_VAR} 替换python-dotenvConfiguration Injection部署层pip install / setup.pyClick CLICLI Facade技术栈选择原则选型决策树: 这个能力需要确定性吗 ├── 是 → 用 Python 代码条件边、Guard、Action └── 否 → 用 LLM命令生成、闲聊、知识检索 │ ▼ 这个数据需要持久化吗 ├── 是 → TrackerStore / MySQL / Neo4j └── 否 → 内存LangGraph State / DialogueStack │ ▼ 这个配置需要用户修改吗 ├── 是 → YAML 配置Domain / Flow / endpoints └── 否 → Python 常量constants.pyCopy8. 延伸通向 Multi-Agent 的架构基础该框架的架构设计使其天然适合扩展为Multi-Agent 系统。8.1 当前架构已具备的条件当前组件Multi-Agent 映射改动量PolicyEnsemble多策略链式调用→ Agent 路由分发器小DialogueStack栈式 Flow 管理→ 跨 Agent 调用栈零Action注册表工具查找→ Agent 服务发现小TrackerStore会话持久化→ 全局会话存储零YAML Flow 定义→ Agent 间的通信协议中8.2 可能的 Multi-Agent 架构┌─────────────────────────────┐ │ Supervisor Agent │ │ 意图识别 任务路由 │ │ (使用当前 PolicyEnsemble 机制) │ └──────────┬──────────────────┘ │ ┌────────────────────┼────────────────────┐ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ 订单 Agent │ │ 物流 Agent │ │ 售后 Agent │ │ │ │ │ │ │ │ 自身 Flow │ │ 自身 Flow │ │ 自身 Flow │ │ 自身 Action │ │ 自身 Action │ │ 自身 Action │ │ 自身 Domain │ │ 自身 Domain │ │ 自身 Domain │ │ 自身 Tracker │ │ 自身 Tracker │ │ 自身 Tracker │ └──────────────┘ └──────────────┘ └──────────────┘8.3 架构优势总结单一 Agent Multi-Agent ┌─────────────┐ ┌─────────────────────┐ │ Policy A │ │ Supervisor Policy │ │ Policy B │ → │ ├─ 订单子 Agent │ │ Action 1-N │ │ ├─ 物流子 Agent │ │ Tracker │ │ ├─ 售后子 Agent │ │ Store │ │ └─ 知识库子 Agent │ └─────────────┘ │ 全局 Store / 栈 │ └─────────────────────┘ 改动: 结构不变把策略和Action换成子 AgentCopy附录关键文件索引模块核心文件职责图编排agent/graph/builder.pyLangGraph 图构建图状态agent/graph/state.py状态定义条件边agent/graph/edges.py路由逻辑感知节点agent/graph/nodes/understand.pyLLM 命令生成规划节点agent/graph/nodes/policy.py策略预测工具节点agent/graph/nodes/action.pyAction 执行安全节点agent/graph/nodes/guard.py循环防护响应节点agent/graph/nodes/response.py结果收集Agent 主类agent/agent.pyAgent 生命周期管理Action 系统agent/actions.py注册表 内置动作消息处理器agent/message_processor.py旧版备用处理器LLM 生成器dialogue_understanding/generator/llm_generator.pyLLM 调用 命令解析命令定义dialogue_understanding/commands/所有命令类型对话栈dialogue_understanding/stack/dialogue_stack.pyFlow 嵌套管理策略集成policies/policy_ensemble.py链式策略调用Flow 策略policies/flow_policy.py业务流程推进知识策略policies/enterprise_search_policy.pyRAG 检索决策Trackercore/tracker.py会话状态追踪存储core/stores/持久化存储配置shared/config.py全量配置加载LLM 客户端shared/llm/多模型客户端抽象电商 Demoecs_demo/完整电商客服示例
