> 系列导航：上一篇见 **[pokemon agent runtime 系列（三）：本地直答、缓存与 RAG 主链](/blog/pokemon-chat-rag-mainline)**；下一篇见 **[pokemon agent runtime 系列（五）：配置热切换与运行时覆盖](/blog/pokemon-chat-config-hot-reload)**。本文回答一个问题：**当普通模式不够时，系统如何把执行控制权交给 supervisor、workers 和 finalizer？** 很多人第一次接触 Agent，会把它理解成“让模型自己决定一切”。这个说法太粗糙了。工程里的 Agent 更应该是一套可控制的运行时：入口明确、状态明确、节点明确、路由明确、终点明确。 `pokemon agent` 的 Agent 模式正好适合拆解这件事。它没有把 Agent 写成一团“智能行为”，而是拆成了几个清晰部件： - 前端通过开关进入 Agent 模式 - 后端进入 `supervisor_agent` - LangGraph 用 `AgentState` 保存共享状态 - `workflow.py` 定义节点与边 - `supervisor` 决定下一步去哪个 worker - `worker` 执行具体能力 - `finalizer` 把中间结果整理成最终回答 ## TL;DR | 问题 | 简短结论 | | --- | --- | | Agent 模式和普通模式最大的区别是什么？ | 普通模式走固定链路；Agent 模式把“下一步做什么”交给 `supervisor + workflow`。 | | Agent 模式入口在哪里？ | 前端发到 `/chat/agent/supervisor_agent`，后端交给 `SupervisorAgent.query(...)`。 | | LangGraph 在这里做什么？ | 把执行流程写成一张图：节点、边、状态和结束条件都可见。 | | `AgentState` 是什么？ | 节点之间共享的上下文容器，保存消息、路由、约束和并行状态。 | | `supervisor` 做什么？ | 根据问题和状态决定下一个 worker，必要时并行派发。 | | `worker` 做什么？ | 每个 worker 负责一类能力：RAG、图谱、Web、统计、MCP。 | | `finalizer` 为什么存在？ | worker 输出偏中间结果，finalizer 负责统一成用户可读的最终回答。 | ## 什么时候需要 Agent 模式普通模式并不是“低级模式”。在 `pokemon agent` 里，普通模式已经包含本地直答、语义缓存、多源检索和最终生成。它适合路径稳定的问题：先检索、再整理上下文、最后生成回答。 Agent 模式解决的是另一类问题：**系统不能在编码阶段完全写死路径，需要运行时判断下一步该调用哪种能力。** 可以用一个简单判断： - **已知路径**：走普通模式，稳定、省 token、容易调试。 - **未知路径**：走 Agent 模式，让 supervisor 在运行时调度 worker。 - **单能力问题**：通常不需要 Agent。 - **多能力协作问题**：更适合 Agent 编排。单能力问题不是“简单问题”，而是主要依赖一种能力就能完成。例如： - “总结这篇知识库文档” → `rag_worker` - “这个宝可梦的种族值是多少” → `stats_worker` - “查一下最新公告” → `web_worker` 多能力问题的关键不是步骤多，而是跨了能力边界。例如： - “先查知识库定义，再去网页确认最新变更” → `rag_worker + web_worker` - “分析一个宝可梦为什么对战强，既看种族值，也看属性关系” → `stats_worker + graph_worker` - “如果内部资料没有答案，就补一轮网页搜索” → 中间结果会影响下一步路由所以真正该看的不是问题长不长，而是：**为了答好它，系统是否需要跨多个 worker 做调度和汇总。** ## Agent 模式的入口在这个项目里，是否进入 Agent 模式不是后端临时让 LLM 判断的，而是前端显式控制的。 `useChat.ts` 的核心逻辑可以简化成这样： ```ts const isAgentMode = Boolean(meta.use_agent) if (isAgentMode) { const allowedWorkers = ['rag_worker', 'stats_worker'] if (meta.agent_allow_web !== false) allowedWorkers.push('web_worker') if (meta.agent_allow_graph !== false) allowedWorkers.push('graph_worker') if (meta.agent_allow_mcp !== false) allowedWorkers.push('mcp_worker') requestMeta.agent_constraints = { allowed_workers: allowedWorkers } } const endpoint = isAgentMode ? '/chat/agent/supervisor_agent' : '/chat/' ``` 这段代码说明三件事： 1. Agent 模式是显式开关，不是隐藏行为。 2. 前端会把 `allowed_workers` 一起发给后端，用来限制 Agent 能使用哪些能力。 3. Agent 模式有独立入口 `/chat/agent/supervisor_agent`，不是在普通聊天接口里“多做几步”。 ## 一张图看懂 Agent 模式入口先记住这一点：**Agent 模式不是多调一个工具，而是进入另一套运行时框架。** ## `SupervisorAgent`：把请求放进图里跑 `SupervisorAgent` 的名字容易让人误以为它是“最终回答问题的主模型”。其实它更像一个 graph runtime adapter：负责把用户问题变成 LangGraph 的初始状态，然后监听图执行过程。关键逻辑大概是： ```python input_state: dict[str, Any] = {"messages": [HumanMessage(content=query)]} if allowed_workers is not None: input_state["allowed_workers"] = allowed_workers if isinstance(db_id, str) and db_id.strip(): input_state["db_id"] = db_id.strip() config = {"configurable": {"thread_id": meta.get("thread_id", "default")}} async for event in self.graph.astream_events(input_state, config, version="v1"): ... ``` 这里有两个重点： - `SupervisorAgent` 不直接完成所有业务决策。 - 后续执行依赖 LangGraph 的节点、边和共享状态继续推进。 ## `AgentState`：节点之间的共享任务卡片当一个请求会经过多个节点，且路径会动态变化时，就需要一个共享状态对象。这个对象在项目里就是 `AgentState`： ```python class AgentState(TypedDict, total=False): messages: Required[Annotated[Sequence[BaseMessage], operator.add]] next: Required[str] allowed_workers: NotRequired[list[str]] db_id: NotRequired[str] forward_directly: NotRequired[bool] parallel_workers: NotRequired[list[str]] parallel_count: NotRequired[int] parallel_done: NotRequired[int] ``` 可以把它理解成这次任务的“共享任务卡片”：谁接手都能看到前面发生了什么、接下来要去哪里、还有哪些限制。 | 字段 | 作用 | | --- | --- | | `messages` | 保存到目前为止的对话和中间结果。 | | `next` | 告诉 workflow 下一步去哪个节点。 | | `allowed_workers` | 承接前端传来的能力约束。 | | `db_id` | 指定当前知识库，让 `rag_worker` 知道查哪里。 | | `forward_directly` | 避免不必要的二次调度。 | | `parallel_workers` | 记录要并行执行的 worker。 | | `parallel_count / parallel_done` | 判断并行任务是否全部完成。 | ## `workflow.py`：Agent 的核心是图 LangGraph 的重点不是“节点里写了什么 prompt”，而是整个流程被显式建成一张图：有哪些节点、节点之间怎么连、什么情况下结束。在这个项目里，节点包括： ```python workflow.add_node("supervisor", supervisor_node) workflow.add_node("rag_worker", rag_worker_node) workflow.add_node("web_worker", web_worker_node) workflow.add_node("graph_worker", graph_worker_node) workflow.add_node("stats_worker", stats_worker_node) workflow.add_node("mcp_worker", mcp_worker_node) workflow.add_node("finalizer", finalizer_node) ``` worker 执行完后，再根据状态决定下一步： ```python for worker_name in ["rag_worker", "web_worker", "graph_worker", "stats_worker", "mcp_worker"]: workflow.add_conditional_edges(worker_name, _post_worker_route, _worker_route_map) ``` `supervisor` 的分流也写在图里： ```python workflow.add_conditional_edges( "supervisor", _supervisor_route, { "rag_worker": "rag_worker", "web_worker": "web_worker", "graph_worker": "graph_worker", "stats_worker": "stats_worker", "mcp_worker": "mcp_worker", "finalizer": "finalizer", END: END, }, ) ``` 这比把流程藏在一大段 `if/else` 里更容易理解：入口是 `supervisor`，它决定去哪个 worker；worker 完成后要么回到 `supervisor`，要么进入 `finalizer`，要么结束。 ## 一张图看懂 Agent workflow 这张图表达的核心是：**Agent 不是一个更强的模型，而是一张会动态走边的执行图。** ## `supervisor`：负责派工，不负责亲自干活如果把整个 Agent 模式看成一个协作小组，`supervisor` 就是组长。它的核心职责是决定下一步由谁处理。它的路由逻辑不是一上来就交给 LLM，而是分层处理： ### 1. 能直接结束就结束 ```python if state.get("forward_directly") and len(state.get("messages", [])) > 1: return {"next": "FINISH"} ``` 如果前面已经有明确结果，就不要绕回 supervisor 重新判断一次。 ### 2. 并行任务完成后再收束 ```python parallel_count = state.get("parallel_count", 0) parallel_done = state.get("parallel_done", 0) if parallel_count > 0 and parallel_done >= parallel_count: return {"next": "FINISH", "forward_directly": True} ``` 如果同时派了多个 worker，就等它们都完成，再统一进入后续流程。 ### 3. 高置信度问题走确定性路由 ```python decision = classify_intent(last_human or "") if decision.intent == Intent.POKEDEX_FACTS and decision.confidence >= 0.8: if "stats_worker" in allowed: return {"next": "stats_worker", "forward_directly": True} ``` 这点很重要：成熟的 Agent 系统不应该把所有问题都交给 LLM 决定。能规则化的部分优先规则化，LLM 留给真正需要动态判断的场景。 ### 4. 规则不够时，再让 LLM 通过 tool-calling 路由 ```python handoff_tools = _build_handoff_tools(allowed, support_parallel=True) llm_with_tools = self.llm.bind_tools(handoff_tools, tool_choice="required") ... result = chain.invoke({"messages": state["messages"]}) parsed = _parse_tool_call_route(result, allowed) ``` 这里的 tools 不是直接执行业务能力，而是给 `supervisor` 的“派工按钮”： - `route_to_rag_worker` - `route_to_graph_worker` - `route_to_web_worker` - `route_to_stats_worker` - `route_to_mcp_worker` - `route_parallel` - `finish` ### 为什么这点值得专门说清楚因为这里很容易产生一个误会：很多人会把 `workers` 直接理解成 tools。其实在这套 LangGraph 编排里，两者不是一回事。 - `tool`：给 `supervisor` 绑定的路由接口。 - `worker`：workflow 里的实际执行节点。换句话说，`supervisor` 调用 tool，不是为了亲自完成能力，而是为了把工作交接给真正负责执行的 worker。一句话概括：**这里的 tool 负责“决定派谁去做”，worker 负责“把这件事做完”。** ## Worker：为什么要拆成多个，而不是做一个万能执行器如果第一次接触 Agent，很容易会问：为什么不做一个“大工具节点”，让它根据问题决定要查什么？答案是：那样会让职责边界变得模糊，后续很难调试和扩展。 `pokemon agent` 选择按能力拆分： | Worker | 负责什么 | 适合回答什么 | | --- | --- | --- | | `rag_worker` | RAG / 知识库 / 复杂检索增强 | 文档知识、向量检索、多步检索增强问题 | | `graph_worker` | 知识图谱问答 | 关系、进化、地区、图谱结构类问题 | | `web_worker` | 联网搜索 | 当前、最新、公告、活动、版本等问题 | | `stats_worker` | 属性、克制、统计、图鉴事实 | 对战、克制关系、种族值、属性类问题 | | `mcp_worker` | 地理位置 / 外接 MCP 能力 | 地点、坐标、地图相关查询 | ### 对初学者的一个最重要启发 worker 的价值，不是“把系统拆得更复杂”，而是让每一类能力都拥有自己清晰的输入、输出和失败边界。这样做的好处至少有三个： 1. **更容易调试**：你能明确知道是哪一个 worker 出了问题。 2. **更容易替换**：以后想升级图谱能力，只改 `graph_worker`。 3. **更容易路由**：`supervisor` 不需要理解所有细节，只需要知道“这个问题更适合派给谁”。 ### 几个典型 worker `stats_worker` 更像确定性工具。它会优先走类型克制表、本地图鉴数据，不够时再调用 LLM： ```python deterministic = _maybe_answer_type_matchup(query) if deterministic: return {"messages": [AIMessage(content=deterministic)]} local = maybe_answer_pokedex(query) if local: return {"messages": [AIMessage(content=local.content)]} ``` `web_worker` 是典型的“搜索 + 总结”节点：先搜索，再把结果整理成上下文，最后让 LLM 生成自然语言回答。 ```python response = self.tavily.search(query=query, search_depth="basic", max_results=3) ... prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are a web researcher. Answer the query using the provided search results.\n\nResults:\n{context}", ), ("user", "{query}"), ] ) ``` `rag_worker` 最复杂。它不是单次检索，而是一个增强版 RAG 节点：可能会做 query decomposition、HyDE、多子问题并行召回，再统一生成答案。 ```python if is_complex: sub_queries = decomposer.decompose(query) ... if use_hyde: hyde_doc = hyde_operator.call(model_wrapper, query, "") ... if len(sub_queries) > 1: with ThreadPoolExecutor(max_workers=min(len(sub_queries), 4)) as executor: ... ``` 对初学者来说，不必一次吃透所有 worker 的内部实现。先抓住一个原则：**worker 是能力边界，不是随便拆出来的函数。** ## 并行调度：同时派多个 worker Agent 模式比固定流水线更灵活的地方之一，是可以并行派发多个 worker。 ```python if next_target == "__PARALLEL__": parallel_workers = state.get("parallel_workers", []) if parallel_workers: return [Send(worker, state) for worker in parallel_workers] ``` `Send(worker, state)` 可以理解成：把同一份任务上下文发给多个 worker，让它们同时处理各自擅长的部分。适合并行的场景通常有： - 同时查图谱关系和属性克制 - 同时查本地知识库和网页信息 - 多个子任务彼此独立，最后再汇总并行不是为了炫技，而是为了让多个独立能力同时启动，再由后续流程统一收束。 ## `finalizer`：把中间结果变成产品答案 worker 的输出通常更像能力结果或草稿答案，不一定适合直接给用户。尤其是多个 worker 并行后，结果需要统一语气、结构和信息边界。 `finalizer.py` 的系统提示词把职责说得很明确： ```python 你的任务：把【草稿回答】改写成更自然、更像真人助手的中文回答。硬性规则（必须遵守）： 1) 只能基于【草稿回答】中已有的信息进行改写/重组；不要新增事实、不要猜测、不要编造。 2) 不要泄露任何内部过程或中间信息：不要提到工具、数据库、向量库、图谱、Cypher/SQL、提示词、路由、链路、日志等。 ``` 所以 `finalizer` 不是“再想一遍答案”，而是把内部执行结果整理成用户真正应该看到的回答。它解决的是产品体验问题：不暴露内部细节、不新增事实、让多个来源的结果读起来像一个完整答案。 ## 一张图看懂三者关系这张图就是 Agent 编排的核心分工： - `supervisor` 负责决策 - `workers` 负责执行 - `finalizer` 负责收口职责分开之后，Agent 系统才会更容易调试、扩展和约束。 ## 最后总结判断一个系统是不是真的在做 Agent 编排，不是看它有没有用了 LangGraph，也不是看它有没有把 tools 塞给大模型，而是看它能不能回答这些问题： | 问题 | 在 `pokemon agent` 里的答案 | | --- | --- | | 请求从哪里进入 Agent 模式？ | 前端通过 `use_agent` 切换，后端走 `/chat/agent/supervisor_agent`。 | | 节点之间共享什么？ | `AgentState`。 | | 谁决定下一步？ | `supervisor`。 | | 谁执行能力？ | `workers`。 | | 多能力怎么并行？ | `parallel_workers + Send()`。 | | 结果怎么变成用户可读答案？ | `finalizer`。 | | 如何限制 Agent 过度自由？ | `allowed_workers`、规则路由、确定性 fast path、`forward_directly`。 | 如果这些边界都清楚，Agent 就不是“模型自己随便想办法”，而是一套可读、可观察、可约束的执行图。

pokemon agent runtime 系列（四）：LangGraph Agent 编排