浅谈升级Multi-Agent

前言

本地已经跑通了LangGraph，现在要在此基础上搭建 Multi-Agent 架构。

受限于本地硬件环境决定用 3B 级别的参数去驱动复杂 Agent 图谱，但是痛点非常密集，意图路由漂移、冷门实体识别缺失、工具调用犹疑，以及随着 Context 增长导致的推理延迟暴增。

经过不断的链路重构与调优，系统目前已能实现毫秒级准确响应。

1. Supervisor 节点：从结构化提取退回到纯分类器

最初的方案是让 Supervisor 节点使用 with_structured_output 动态提取参数并返回完整的 JSON 对象。但在 3B 模型上，一旦用户输入存在歧义，模型极易产生语法破损或路由漂移。

工程解法：

• 裁剪 Prompt 上下文：路由决策阶段剥离对话历史，仅向模型喂入用户最新一条 HumanMessage。

• 收窄决策空间：严禁模型自主推理参数，仅需对文本打上标签（如 car_expert / weather_expert）。

• 结果二次解析：摒弃复杂的 Pydantic 强校验，采用朴素的正则提取 JSON 文本块。

在降低模型的认知负担后，Supervisor 的路由命中率从约 70% 提升到了 90% 以上。

2. 参数提取降级：字典拦截与 LLM 泛化兜底

在测试天气 Agent 时，发现 qwen2.5:3b 存在明显的实体识别盲区。例如处理“北京天气”正常，但处理“吐鲁番天气”时，模型因未能将“吐鲁番”识别为 Valid City，直接触发安全拒答，跳过了 tool_calls 的生成。

工程解法（双层调度机制）：

• 确定性拦截（Fast-Path）：在 Weather 节点内嵌入 city_matcher 模块，优先对输入文本与本地 cities.txt 进行前缀/子串匹配。一旦命中，跳过 LLM 提取，直接在 Python 侧拼装 AIMessage(tool_calls=[...]) 送入 ToolNode。

• LLM 泛化兜底（Slow-Path）：未命中本地词库的边缘场景（如非标准地名），再回退由模型自主决策。

事实证明，在小模型场景下，确定性的规则代码是最好的防爆网。

3. 响应链路优化：工具结果直接透传 (0s 延迟)

LangGraph 的标准 Pipeline 在工具执行完毕后，会将 ToolMessage 重新扔给 Agent 节点，让 LLM 做一次文本汇总后再输出给客户端。

对于天气查询这种 API 已吐出结构化中文（如 【吐鲁番】天气：晴朗，气温 38℃）的场景，二次总结会导致额外的 3000+ Token KV Cache 计算，首 Token 延迟高达 10 秒以上。

工程解法：

在 Agent 节点入口增加类型拦截：

Python

if messages and messages[-1].type == "tool":
    return {"messages": [AIMessage(content=messages[-1].content)]}

通过该改动，系统跳过了无意义的二次 LLM 推理。响应耗时降至 3 秒以内（包含网络 API 请求时间），并彻底消除了 LLM 转述带来的信息幻觉风险。

4. Redis 历史消息滑动窗口裁剪

3B 模型在 Prompt Token 突破 4000 时，推理吞吐率会发生断崖式下跌。

工程解法：

从 Redis 获取会话历史时，使用列表切片强行限制最大历史深度为 10 条（即 5 轮对话）：

Python

saved_messages = history_store.messages[-10:]

保持上下文在极低量级，这是保障本地笔记本 GPU 推理流畅度的基础条件。

5. 数据边界清洗：避免脏数据流入前端

第三方 API（如 wttr.in）返回的中文天气字段经常包含 "Partly cloudy 局部多云" 这类中英混杂的字符串。如果未经处理直接透传，会破坏前端的呈现体验。

工程解法： 将数据清洗逻辑下沉至工具函数内部，利用正则提取 Unicode 中文字符集（[\u4e00-\u9fff]+）。将清洗工作收拢在工具层，确保流向前端的只有纯净的数据串。

6. 流式事件 (SSE) 管道的特殊兼容

在引入“工具结果直接透传”优化后，遇到了一个工程坑点：透传生成的 AIMessage 是纯 Python 对象，无法触发 LangChain 的 on_chat_model_stream 事件，导致前端 SSE 管道陷入假死等待状态。

工程解法：

在 FastAPI 的 astream_events 监听循环中，补充对 on_chain_end 事件的捕获：

Python

elif event_type == "on_chain_end" and node_name in ("weather_expert", "car_expert"):
    output = event.get("data", {}).get("output", {})
    if isinstance(output, dict) and "messages" in output:
        last_msg = output["messages"][-1]
        if isinstance(last_msg, AIMessage) and last_msg.content and not last_msg.tool_calls:
            yield f"data: {json.dumps({'content': last_msg.content})}\n\n"

确保了无论数据是 LLM 实时生成还是代码直出，前端都能准确获取 Chunk 并渲染。

7. 实战总结与思考

在算力受限的环境下搭建生产级 Agent，需要明确代码与 AI 的职责边界：

1. 确定性逻辑归 Python，不确定性逻辑归 LLM：不要寄希望于用 Prompt 去纠正小模型的认知缺陷，本地字典和硬编码是性价比最高的解法。

2. 避免过早引入复杂度：在简单的业务场景下，无需盲目挂载 HITL（人工介入）或 Checkpointer 回滚机制。先确保基础链路在端到端的时延与准确率上达到可用状态。

环境上下文：Qwen2.5-3B-Instruct / LangGraph 0.2+ / Langfuse / Redis / FastAPI