说句实话市面上的多 Agent 教程大多停在“两个 Agent 互相传一句话”的玩具阶段真正能放进生产目录、敢让同事接手、加新 Agent 不推倒重来的工程级示例反而非常少。我最近用 FastAPI LangGraph 搭了一个完整的多 Agent 网关把 Gateway、Agent、Tool、Memory 四层彻底拆开跑通了从用户请求到意图识别、再到子 Agent 调用工具、最后带记忆返回的完整链路。这篇文章不聊概念直接给真实目录结构和完整源代码顺便把我在这个过程中踩过的坑一并写清楚。适合正在从单 Agent 玩具代码往多 Agent 工程架构迁移的团队也适合想一晚上看懂 LangGraph 工程化落地方式的后端同学。1. 为什么必须拆四层单体 Agent 的扩容困境1.1 单体 Demo 看起来很美好直到需要加第二个业务我第一次写多 Agent 应用的时候也用过一个“万能 Agent”一个 system prompt 里塞客服话术、订单查询规则、商品推荐逻辑下面挂十几个 tool。写 Demo 的时候一切正常测两个轮次也看不出问题但一旦真实业务进来问题马上暴露。最典型的现象是工具选择开始乱跳。LLM 面对十几个工具描述经常把“查询订单状态”误判成“商品库存查询”因为两个工具名里都带“查询”prompt 里客服语气和订单规则互相冲撞今天改了客服的售后话术明天订单 Agent 的回答语气也变了。更麻烦的是上下文所有业务共享一份历史消息客服会话里的闲聊会污染订单查询的判断。这时候你自然想拆。但怎么拆直接复制一个 Agent、改改 prompt 加两个路由路由写死了Agent 一多代码就开始失控。我后来意识到问题不是 Agent 数量不够而是所有功能代码没有分层。没有分层任何一次改动都要掀翻全局。1.2 四层各管一件事数据流是这样走的拆完层之后整个系统变清楚了很多。四层职责非常明确层级核心职责典型依赖生命周期Gateway请求入口、认证、意图路由、会话编排FastAPI、编排器每次请求Agent业务策略、prompt 管理、LangGraph 状态图LangGraph、LLM客户端每次任务Tool原子能力查订单、查库存、调外部HTTP业务 SDK、HTTP 客户端每次调用Memory会话历史存取、跨 Agent 上下文共享文件系统 / Redis会话级数据流是一条很清晰的链路用户消息到 GatewayGateway 先做意图路由决定交给哪个子 Agent子 Agent 内部跑 LangGraph 状态图过程中需要外部能力就去 Tool 注册表取工具执行工具结果回到 Agent 的 LLM 调用循环每个轮次的上下文从 Memory 加载、结束后写回 Memory。一个 session_id 贯穿全链路所有状态跟着会话走。这套拆法解决了我的三个核心痛点首先是扩展性新业务只要新增一个 Agent 类并注册路由和工具可以复用其次是可测试性每一层都能单独 mock比如测 Gateway 不依赖真实 LLM最后是可靠性Memory 的存储实现可以随时从本地 JSONL 切到 RedisTool 的失败有兜底不会因为单个工具异常拖垮整个 Agent。1.3 这套分层到底解决哪几个真实问题说白了分层不是为了代码好看是为了降低“改一个东西炸一片”的概率。单体 Agent 里改一个工具的入参格式所有调用它的 Agent 都会受影响分层之后Tool 内部改签名只要协议不变上层完全不感知。同样你想从 OpenAI 兼容接口换到别的模型提供商也只需要在 Gateway 的 LLM 客户端层改一处Agent 层不用动。记忆的切换也是典型场景。开发阶段用 JSONL 落盘最方便部署之后换 Redis只需要实现同一个 BaseMemory 接口。这就是分层带来的自由度每一层都在为“将来会变”的部分留好替换窗口。2. 工程骨架从零搭建可运行的多 Agent 项目2.1 目录结构逐项说明项目我起了个名字就叫multi_agent_gateway核心思想是“Gateway 在外面Agent 在中间Tool 和 Memory 垫底”。真实目录结构如下multi_agent_gateway/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # pydantic-settings 配置 │ ├── gateway/ │ │ ├── __init__.py │ │ ├── schemas.py # 请求/响应模型 │ │ ├── router.py # /v1/chat 路由 │ │ ├── orchestrator.py # 编排器记忆加载、意图路由、Agent 调度 │ │ └── intent_router.py # 意图识别与 Agent 选择 │ ├── agents/ │ │ ├── __init__.py │ │ ├── base.py # Agent 基类 AgentState │ │ ├── registry.py # Agent 注册表 │ │ ├── workflow.py # LangGraph 状态图构建 │ │ ├── customer_service.py # 客服 Agent │ │ ├── order_agent.py # 订单 Agent │ │ └── product_agent.py # 商品 Agent │ ├── tools/ │ │ ├── __init__.py │ │ ├── base.py # BaseTool 工具 schema │ │ ├── registry.py # 工具注册表 │ │ ├── http_tool.py # 通用 HTTP 工具 │ │ └── order_tool.py # 订单查询工具 │ └── memory/ │ ├── __init__.py │ ├── base.py # BaseMemory 接口 │ ├── file_memory.py # JSONL 文件实现 │ └── redis_memory.py # Redis 实现生产推荐 ├── tests/ │ ├── test_workflow.py │ └── test_gateway.py ├── .env.example └── pyproject.toml每个目录都有明确边界gateway只关心 HTTP 层和编排不写任何业务 promptagents只关心 LangGraph 图和业务策略不直接碰存储tools是纯能力层不感知会话memory只做持久化。互相之间通过类和注册表交互不搞隐式全局变量。2.2 依赖与配置pydantic-settings 管好你的环境变量依赖没有多余的东西核心就这几个[project] name multi-agent-gateway version 0.1.0 requires-python 3.11 dependencies [ fastapi0.110, uvicorn[standard]0.29, langgraph0.2, openai1.30, pydantic-settings2.2, httpx0.27, redis5.0, ] [project.optional-dependencies] dev [pytest, pytest-asyncio, ruff]配置我统一用 pydantic-settings 管理避免散落一堆os.getenv。.env.example长这样LLM_API_KEYsk-xxx LLM_BASE_URLhttps://your-llm-provider.example.com/v1 LLM_MODELgpt-4o-mini DEFAULT_AGENTgeneral MEMORY_BACKENDfile REDIS_URLredis://localhost:6379/0config.py 对应实现from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) llm_api_key: str sk-xxx llm_base_url: str https://your-llm-provider.example.com/v1 llm_model: str gpt-4o-mini default_agent: str general memory_backend: str file redis_url: str redis://localhost:6379/0 request_timeout: float 60.0 max_history_rounds: int 10 settings Settings()注意这里把llm_base_url设成占位地址是因为现在不少团队用的是 OpenAI 兼容协议的自建模型服务这样配置能直接切换不绑定任何具体厂商。2.3 FastAPI 入口与 lifespan 初始化顺序main.py的初始化顺序是个容易出错的地方。我踩过一坑工具没注册完就急着构建 Agent 图结果运行时报工具找不到。正确的做法是放在 lifespan 里按固定顺序初始化from contextlib import asynccontextmanager from fastapi import FastAPI from app.agents.registry import agent_registry from app.agents.customer_service import CustomerServiceAgent from app.agents.order_agent import OrderAgent from app.agents.product_agent import ProductAgent from app.gateway.router import router as chat_router from app.tools.http_tool import HttpTool from app.tools.order_tool import OrderTool from app.tools.registry import tool_registry asynccontextmanager async def lifespan(app: FastAPI): # 顺序先工具再 Agent tool_registry.register(HttpTool) tool_registry.register(OrderTool) agent_registry.register(CustomerServiceAgent) agent_registry.register(OrderAgent) agent_registry.register(ProductAgent) yield tool_registry.clear() agent_registry.clear() app FastAPI(titleMulti-Agent Gateway, version0.1.0, lifespanlifespan) app.include_router(chat_router, prefix/v1)顺序背后的原因是Agent 构建图时会把当前注册好的工具列表注入 system prompt顺序反了 Agent 拿到的工具列表就是空的。而且 build 流程只在启动时做一次不要在请求路径里重复 compile 同一个图LangGraph 编译有开销每次请求都编译会白白增加延迟。3. 核心源码拆解Tool / Memory / Agent / Gateway 的落地写法3.1 Tool 层协议、注册表、HTTP 工具与异常兜底Tool 层的核心思想是“每个工具都要能被 LLM 理解也要能安全失败”。BaseTool 定义了两件事给 LLM 看的 schema以及真正执行的 run 方法。from abc import ABC, abstractmethod from typing import Any class BaseTool(ABC): name: str description: str parameters: dict {} abstractmethod async def run(self, **kwargs: Any) - Any: 执行工具逻辑返回可 JSON 序列化的结果 def schema(self) - dict: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, }注册表就是一个简单的字典提供 register、get、all_tools 三个方法from typing import Any from app.tools.base import BaseTool class ToolRegistry: def __init__(self) - None: self._tools: dict[str, BaseTool] {} def register(self, tool_cls: type[BaseTool]) - None: inst tool_cls() self._tools[inst.name] inst def get(self, name: str) - BaseTool | None: return self._tools.get(name) def all_tools(self) - list[dict]: return [tool.schema() for tool in self._tools.values()] def clear(self) - None: self._tools.clear() tool_registry ToolRegistry()具体工具我以订单查询为例import httpx from app.tools.base import BaseTool class OrderTool(BaseTool): name query_order description 根据订单号查询订单状态和物流信息 parameters { type: object, properties: { order_id: {type: string, description: 订单号}, }, required: [order_id], } async def run(self, **kwargs) - dict: order_id kwargs.get(order_id, ) # 实际项目这里会请求内部订单服务 async with httpx.AsyncClient(timeout5) as client: resp await client.get(fhttps://internal-order.example.com/api/orders/{order_id}) resp.raise_for_status() return resp.json()关于异常兜底我后来单独写了一篇文章的经验都浓缩在这句话里工具内部一定要自己 catch 所有异常并返回结构化错误信息而不是把异常抛给 LangGraph 节点。因为 LLM 需要“看到”工具错误才能决定下一步比如改参数重试或者告知用户如果异常直接冒泡整个图会崩体验很差。3.2 Memory 层JSONL 落盘实现到 Redis 扩展Memory 接口非常简单只有 append 和 load 两个操作。这个接口设计得越窄后面换存储越轻松from abc import ABC, abstractmethod class BaseMemory(ABC): abstractmethod async def append(self, session_id: str, role: str, content: str) - None: 写入一条消息 abstractmethod async def load(self, session_id: str, limit: int 10) - list[dict]: 读取最近 N 条历史消息开发阶段先用 JSONL 文件实现每个 session 一个文件方便调试时直接打开看import asyncio import json from pathlib import Path from app.memory.base import BaseMemory class FileMemory(BaseMemory): def __init__(self, base_dir: str data/memory) - None: self.base_dir Path(base_dir) self.base_dir.mkdir(parentsTrue, exist_okTrue) self._locks: dict[str, asyncio.Lock] {} def _path(self, session_id: str) - Path: return self.base_dir / f{session_id}.jsonl def _lock(self, session_id: str) - asyncio.Lock: if session_id not in self._locks: self._locks[session_id] asyncio.Lock() return self._locks[session_id] async def append(self, session_id: str, role: str, content: str) - None: async with self._lock(session_id): with self._path(session_id).open(a, encodingutf-8) as f: f.write(json.dumps({role: role, content: content}, ensure_asciiFalse) \n) async def load(self, session_id: str, limit: int 10) - list[dict]: path self._path(session_id) if not path.exists(): return [] lines path.read_text(encodingutf-8).strip().splitlines() return [json.loads(line) for line in lines[-limit:]]注意我在文件实现里加了asyncio.Lock按 session_id 粒度锁住写入。这不是过度设计——我遇到过两个协程同时写同一个 JSONL 文件结果内容交错成脏数据。开发环境并发不高但一旦同时处理多个用户这个问题就会炸。线上部署建议用 Redis 实现代码更简洁import json import redis.asyncio as redis from app.memory.base import BaseMemory class RedisMemory(BaseMemory): def __init__(self, redis_url: str, ttl: int 3600) - None: self.client redis.from_url(redis_url, decode_responsesTrue) self.ttl ttl def _key(self, session_id: str) - str: return fmemory:{session_id} async def append(self, session_id: str, role: str, content: str) - None: key self._key(session_id) await self.client.rpush(key, json.dumps({role: role, content: content}, ensure_asciiFalse)) await self.client.expire(key, self.ttl) async def load(self, session_id: str, limit: int 10) - list[dict]: raw_list await self.client.lrange(self._key(session_id), -limit, -1) return [json.loads(item) for item in raw_list]Redis 的 list 天然适合做消息队列式存储rpush 裁剪读取刚好符合“最近 N 轮”的语义。TTL 也解决了文件实现没做清理导致磁盘膨胀的问题。3.3 Agent 层LangGraph 状态图与工具调用循环这是整个工程的核心。Agent 基类负责定义一个标准的 LangGraph 图子类只需要提供 system_prompt 和 name。状态定义用 TypedDict其中messages用operator.add做累积这个细节后面会说。import operator from typing import Annotated, Any, TypedDict from langgraph.graph import END, StateGraph class AgentState(TypedDict): messages: Annotated[list[dict], operator.add] history: list[dict] session_id: str pending_tool_calls: list[Any] tool_results: Annotated[list[dict], operator.add] final_answer: str class BaseAgent: name: str base description: str system_prompt: str def build_graph(self): graph StateGraph(AgentState) graph.add_node(call_llm, self.call_llm) graph.add_node(exec_tool, self.exec_tool) graph.set_entry_point(call_llm) graph.add_conditional_edges( call_llm, self.need_tool, {tool: exec_tool, end: END}, ) graph.add_edge(exec_tool, call_llm) return graph.compile()call_llm节点负责把系统 prompt、历史记忆、当前轮 messages 拼好调用 LLM。如果模型返回 tool_call就存到pending_tool_calls并置need_toolTrue否则直接产出 final_answerimport json from openai import AsyncOpenAI from app.config import settings from app.tools.registry import tool_registry client AsyncOpenAI(api_keysettings.llm_api_key, base_urlsettings.llm_base_url, timeoutsettings.request_timeout) class BaseAgent: # ... 上面省略 ... async def call_llm(self, state: AgentState) - dict: messages self._build_messages(state) kwargs { model: settings.llm_model, messages: messages, } tools tool_registry.all_tools() if tools: kwargs[tools] tools kwargs[tool_choice] auto resp await client.chat.completions.create(**kwargs) choice resp.choices[0].message if choice.tool_calls: return { messages: [choice.model_dump()], pending_tool_calls: choice.tool_calls, need_tool: True, } return { messages: [{role: assistant, content: choice.content or }], final_answer: choice.content or , need_tool: False, } def _build_messages(self, state: AgentState) - list[dict]: messages [{role: system, content: self.system_prompt}] messages.extend(state.get(history, [])) messages.extend(state.get(messages, [])) for result in state.get(tool_results, []): messages.append({ role: tool, tool_call_id: result[tool_call_id], content: json.dumps(result[content], ensure_asciiFalse), }) return messagesexec_tool节点遍历待执行的工具调用逐个执行并把结果累积进tool_resultsclass BaseAgent: # ... 上面省略 ... async def exec_tool(self, state: AgentState) - dict: results [] for tc in state.get(pending_tool_calls, []): tool tool_registry.get(tc.function.name) if tool is None: results.append({ tool_call_id: tc.id, tool: tc.function.name, content: {error: ftool {tc.function.name} not found}, }) continue try: args json.loads(tc.function.arguments or {}) res await tool.run(**args) results.append({tool_call_id: tc.id, tool: tc.function.name, content: res}) except Exception as e: results.append({ tool_call_id: tc.id, tool: tc.function.name, content: {error: str(e)}, }) return { pending_tool_calls: [], need_tool: False, tool_results: results, } def need_tool(self, state: AgentState) - str: return tool if state.get(need_tool) else end这里的关键是把工具结果作为role: tool的消息回填给 LLM而且必须带上tool_call_id。OpenAI 兼容接口对这块校验很严格tool_call_id对不上会直接报错。我把工具失败也当成普通消息返回LLM 会自然回复“抱歉我现在无法查询该订单”而不是让图崩溃。子 Agent 的写法就很简单了以订单 Agent 为例from app.agents.base import BaseAgent class OrderAgent(BaseAgent): name order description 处理订单查询、物流跟踪相关咨询 system_prompt 你是订单助手。你可以通过 query_order 工具查询订单状态。 规则 1. 用户必须提供订单号才能查询 2. 查询失败时先确认订单号是否正确 3. 不要编造物流信息。 3.4 Gateway 层意图路由与编排器Gateway 是外层门面负责把 HTTP 请求变成 Agent 能理解的输入然后把结果包装回响应。请求和响应模型用 Pydantic 定义from pydantic import BaseModel, Field class ChatRequest(BaseModel): message: str Field(..., min_length1, max_length4000) session_id: str | None None agent: str | None None user_id: str | None None class ChatResponse(BaseModel): session_id: str agent: str answer: str tool_used: list[str] []路由层只做一件事接收请求调用编排器返回结果不写任何业务逻辑。from fastapi import APIRouter from app.gateway.orchestrator import run_agent_task from app.gateway.schemas import ChatRequest, ChatResponse router APIRouter(tags[chat]) router.post(/chat, response_modelChatResponse) async def chat(req: ChatRequest) - ChatResponse: return await run_agent_task(req)编排器是 Gateway 层最值得看的部分它把记忆加载、意图路由、Agent 调度、记忆写回串成一条流水线import uuid from app.agents.registry import agent_registry from app.agents.base import BaseAgent from app.config import settings from app.gateway.intent_router import route_intent from app.gateway.schemas import ChatRequest, ChatResponse from app.memory.file_memory import FileMemory from app.memory.redis_memory import RedisMemory memory RedisMemory(settings.redis_url) if settings.memory_backend redis else FileMemory() async def run_agent_task(req: ChatRequest) - ChatResponse: session_id req.session_id or str(uuid.uuid4()) agent_name req.agent or await route_intent(req.message) if agent_name not in agent_registry.list_names(): agent_name settings.default_agent agent_cls agent_registry.get(agent_name) if agent_cls is None: agent_cls BaseAgent agent: BaseAgent agent_cls() history await memory.load(session_id, limitsettings.max_history_rounds) initial_state { messages: [{role: user, content: req.message}], history: history, session_id: session_id, pending_tool_calls: [], need_tool: False, final_answer: , } graph agent.build_graph() result await graph.ainvoke( initial_state, config{recursion_limit: 30, configurable: {thread_id: session_id}}, ) await memory.append(session_id, user, req.message) await memory.append(session_id, assistant, result.get(final_answer, )) tool_names [item[tool] for item in result.get(tool_results, [])] return ChatResponse( session_idsession_id, agentagent_name, answerresult.get(final_answer, ), tool_usedtool_names, )意图路由我用了一个轻量方案让 LLM 做一次分类只输出 Agent 名称再接一层白名单校验。解析失败就落到 general Agent保证系统永远不会因为路由问题直接 500from openai import AsyncOpenAI from app.agents.registry import agent_registry from app.config import settings client AsyncOpenAI(api_keysettings.llm_api_key, base_urlsettings.llm_base_url) async def route_intent(message: str) - str: agent_names ,.join(agent_registry.list_names()) prompt ( f根据用户问题选择一个最合适的 Agent。可选 Agent{agent_names}。 f只输出一个名称不要解释。\n用户问题{message} ) try: resp await client.chat.completions.create( modelsettings.llm_model, messages[{role: user, content: prompt}], max_tokens10, temperature0, ) name resp.choices[0].message.content.strip() if name in agent_registry.list_names(): return name except Exception: pass return settings.default_agent这套代码跑起来后一次完整请求是用户发消息 → 路由选 Agent → Agent 自己的 LangGraph 图跑 1~2 轮 LLM 工具 → 记忆写回 → 返回结构化响应。链路清晰每一段都能单独 debug。4. 多 Agent 协作的编排细节状态传递与上下文隔离4.1 消息状态如何累积LangGraph reducer 细节LangGraph 的 State 是一个 TypedDict节点返回什么就更新什么。但它有一个隐晦的规则如果你用了Annotated[list, operator.add]节点返回的这个字段的值是“追加”进原列表的而不是整体替换。刚开始我没理解这一点直接在节点里返回{messages: all_messages}结果每次 LLM 调用后 messages 翻倍因为老消息被追加了一遍新消息。后来我改成节点只返回“新增的那几条消息”比如 call_llm 只返回当前 LLM 的响应消息operator.add会自动帮我把新消息追加到 existing 列表之后。这样做有个明显好处tool_results 天然累积。第一次工具调用结果、第二次工具调用结果会按顺序全部沉淀在状态里最后 LLM 能看到完整的工具执行记录。如果整体替换后一轮的 tool_results 会覆盖前一轮多轮工具调用的上下文就丢了。class AgentState(TypedDict): messages: Annotated[list[dict], operator.add] history: list[dict] session_id: str pending_tool_calls: list[Any] tool_results: Annotated[list[dict], operator.add] final_answer: str4.2 意图路由和子 Agent 之间怎么避免串话多 Agent 最容易翻车的地方是上下文串话。我有三个专门的手段第一每个子 Agent 都有独立的 system_prompt互不共享。订单 Agent 的规则不会飘到客服 Agent 那里。第二会话历史虽然共享同一份 Memory但历史消息本身不带 Agent 标识所以我在写回记忆时只存 user 和 assistant 两种角色具体这条回答是哪个 Agent 出的不做区分因为下一轮可能由另一个 Agent 处理强行打 Agent 标签反而会让 prompt 混乱。第三Agent 之间不做直接通信所有交互都通过 Gateway 编排器转发。谁发起的谁收尾不搞隐性的互相调用。这样一个用户在多轮对话里可能经历“客服 Agent 处理 → 订单 Agent 处理”两段上下文都基于同一份 Memory但每一轮的系统提示都是当前 Agent 自己的不会产生“客服规则里混进订单查询逻辑”的情况。4.3 从 router 模式扩展到 Supervisor 模式的思路这套代码用的是“意图路由 子 Agent”模式也是工程里最好落地的一种。如果后面想升级到 Supervisor 模式其实改动不大在 Gateway 层把route_intent替换成一个 supervisor Agent让它拥有工具delegate_to(order_agent)、delegate_to(customer_service_agent)等supervisor 自己就是一个 LangGraph 图负责分拆任务并汇总结果。这种模式适合“一个复杂任务需要多个 Agent 接力”的场景。但模式升级的核心骨架不用变Tool 协议、Memory 接口、Agent 注册表都是现成的你只需要新增一个 supervisor Agent 类。我的建议是第一版先跑通 router 模式把稳定性做出来再考虑 supervisor。因为 supervisor 模式对 LLM 的指令遵循能力要求高很多一旦路由逻辑出错排查链路的成本直线上升。5. 从 Demo 到生产并发、超时、重试、可观测性一起补上5.1 并发请求下的会话隔离FastAPI 的 async 接口天然支持高并发但并发一上来第一个坑就是“全局变量污染”。比如有人在模块里写了一个current_user_order变量用来暂存当前查询的订单号两个用户同时请求数据就串了。解决方式很朴素状态只存在于 request scope 里。我的方案是把一切与用户相关的数据都明确封装在session_id和initial_state里Agent 图内部不引用任何模块级可变变量。LLM 客户端可以全局共享但那是只读的配置工具调用可以每次 new 一个实例Memory 的写入锁定按 session_id 隔离。做到这些并发再多也不会串数据。5.2 LLM 不稳定是常态超时与重试策略接入过真实模型服务的人都知道一个请求等 30 秒才响应是常有的事偶尔还会直接失败。我见过太多人一上线就遇到“请求超时”因为他们根本没想到要给 LLM 调用设置超时。我给所有 LLM 调用统一设置了timeout60同时给工具执行加了一层保护用asyncio.wait_for包住工具调用单工具最长 10 秒超时就按工具错误返回不让整个图卡死。重试则用最简单的指数退避只对网络类和 5xx 错误重试对参数错误不重试import asyncio async def run_with_retry(coro_factory, retries3): for attempt in range(retries): try: return await coro_factory() except Exception: if attempt retries - 1: raise await asyncio.sleep(0.5 * (2 ** attempt))5.3 加日志与链路 ID出了问题能定位多 Agent 链路很长没有链路 ID排查问题就像大海捞针。我在 FastAPI 加了一个简单中间件请求进来时从 Header 读X-Request-ID没有就生成一个 UUID塞进 contextvars然后把日志里都带上这个 ID。工具调用、意图路由、Agent 名称、LLM 耗时全部记进去。import contextvars import uuid from fastapi import Request request_id_var: contextvars.ContextVar[str] contextvars.ContextVar(request_id, default-) async def request_id_middleware(request: Request, call_next): rid request.headers.get(X-Request-ID, str(uuid.uuid4())) token request_id_var.set(rid) try: response await call_next(request) response.headers[X-Request-ID] rid return response finally: request_id_var.reset(token)日志格式我推荐结构化输出每行一个 JSON线上 grep 起来很快。别小看这一步没有它后面 Agent 报错你都不知道是哪一次请求、哪个 Agent 出的问题。5.4 最小测试集跑通网关到 Agent 的闭环生产代码不能没有测试但多 Agent 工程的测试不能依赖真实 LLM。我的做法是测试时用 mock 替换 LLM 客户端固定返回一段包含 tool_call 的响应验证图会正确走进 exec_tool 节点并回填结果。import pytest class FakeLLM: async def chat_completions_create(self, **kwargs): return fake_response_with_tool_call() pytest.mark.asyncio async def test_workflow_tool_loop(monkeypatch): monkeypatch.setattr(app.agents.base.client.chat.completions.create, FakeLLM().chat_completions_create) agent OrderAgent() graph agent.build_graph() result await graph.ainvoke({messages: [], history: [], session_id: test}) assert final_answer in result测试的意义在于保护分层边界。以后谁改坏了 Tool 注册表、谁破坏了 Memory 接口测试能在 CI 阶段暴露而不是上生产了才发现。6. 踩坑记录LangGraph 工程的常见事故曲线6.1 图节点返回不生效状态原地修改失效这是我踩得很痛的一个坑。一开始我以为 LangGraph 节点是“传引用”直接在 state 的 dict 里加字段就行结果运行完发现状态根本没变化。查完文档才明白LangGraph 节点必须返回一个 dict返回的键会覆盖或追加到 State 中你在节点内部对 state 做任何原地修改图都不会感知。曾经写这样的代码async def bad_node(state): state[final_answer] 假装写了 return {} # 什么都没返回final_answer 永远不会出现在最终状态里。正确写法必须显式返回更新项。排查方法很简单在节点入口和出口各打一次日志对比 state 的变化一眼就看出有没有 Return。6.2 循环导入与 lifespan 初始化顺序工具注册表、Agent 注册表、Gateway 编排器这三者之间天然存在相互引用Agent 要用 ToolRegistry编排器要用 AgentRegistryTool 的 schema 又要被 Agent 注入。项目一变大循环导入就来了。我的规避心得是注册表只保存类和实例不保存“来自哪个模块的引用链”。具体说就是创建注册表时不做任何 import真正 fill 注册表的工作全部放到 FastAPI lifespan 里做。这样模块 import 阶段不会触发依赖链运行阶段再由 lifespan 控制顺序。初始化顺序也很讲究工具注册永远先于 Agent 注册因为 Agent build 图时会把工具 schema 注入 prompt。6.3 工具异常把 Agent 推进死循环有一次测试工具接口报错我没有 catch结果 LangGraph 图卡在 call_llm 和 exec_tool 之间来回跑直到recursion_limit触发异常。原因是LLM 发现工具返回了报错文本它以为是自己参数传错了于是换一个参数再调一次又失败又换参数……有 3 个可重试的 tool_call 就能把 recursion_limit 打满。现在我做了三重防护。第一工具内部 try/except 全部 catch返回结构化错误内容但内容里明确写上“工具执行失败请告知用户稍后重试”。第二exec_tool 节点里也再包一层防止工具实现没按约定 catch。第三conditional edge 里加一个重入次数计数超过 5 次强制走 end不让图穷举。6.4 记忆并发冲突与上下文超窗文件 Memory 的并发问题前面提过解决方式是 asyncio.Lock 按 session 粒度加锁。上下文超窗问题则来自另一个方向用户聊了几百轮之后历史消息全部塞进 promptToken 溢出模型乱答。方案是只保留最近 N 轮而且截断的时候优先保留“与当前意图相关的近三轮完整消息”而不是盲目截前 N 条。我这版用max_history_rounds 10实际可以根据模型上下文窗口动态调。还有一个小坑差点忘了OpenAI 兼容接口要求 tool 消息必须带tool_call_id而且 assistant 的 tool_call 消息必须紧跟对应 tool 结果。如果你在某轮把历史里的 tool_call 消息传给 LLM但没有回填 tool 结果接口直接报错。我在_build_messages里用tool_results动态拼接就是为了保证永远不会出现“有调用无结果”的情况。按我个人经验把这套四层架构跑通之后最大的感受是新需求来了不用再“掀桌子”了。加一个业务 Agent 只需要写一个类、注册一行加一个工具只需要实现 BaseTool、注册一行换记忆存储只需要替掉 Memory 实现。踩过的坑也都在前面积累了状态的返回语义、lifespan 的顺序、工具异常的兜底、上下文窗口的管理这些在官方文档和示例代码里几乎找不到现成答案。如果你正准备把一个单 Agent Demo 改造成多 Agent 工程直接照这套骨架改一版大概率能少熬好几个夜。