FastAPI+LangGraph实战：多Agent网关四层架构与完整源码

说句实话，市面上的多 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，Gateway 先做意图路由，决定交给哪个子 Agent；子 Agent 内部跑 LangGraph 状态图，过程中需要外部能力就去 Tool 注册表取工具执行；工具结果回到 Agent 的 LLM 调用循环；每个轮次的上下文从 Memory 加载、结束后写回 Memory。一个 session_id 贯穿全链路，所有状态跟着会话走。

这套拆法解决了我的三个核心痛点：首先是扩展性，新业务只要新增一个 Agent 类并注册，路由和工具可以复用；其次是可测试性，每一层都能单独 mock，比如测 Gateway 不依赖真实 LLM；最后是可靠性，Memory 的存储实现可以随时从本地 JSONL 切到 Redis，Tool 的失败有兜底，不会因为单个工具异常拖垮整个 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 层和编排，不写任何业务 prompt；agents只关心 LangGraph 图和业务策略，不直接碰存储；tools是纯能力层，不感知会话；memory只做持久化。互相之间通过类和注册表交互，不搞隐式全局变量。

2.2 依赖与配置：pydantic-settings 管好你的环境变量

依赖没有多余的东西，核心就这几个：

[project] name = "multi-agent-gateway" version = "0.1.0" requires-python = ">=3.11" dependencies = [ "fastapi>=0.110", "uvicorn[standard]>=0.29", "langgraph>=0.2", "openai>=1.30", "pydantic-settings>=2.2", "httpx>=0.27", "redis>=5.0", ] [project.optional-dependencies] dev = ["pytest", "pytest-asyncio", "ruff"]

配置我统一用 pydantic-settings 管理，避免散落一堆os.getenv。.env.example长这样：

LLM_API_KEY=sk-xxx LLM_BASE_URL=https://your-llm-provider.example.com/v1 LLM_MODEL=gpt-4o-mini DEFAULT_AGENT=general MEMORY_BACKEND=file REDIS_URL=redis://localhost:6379/0

config.py 对应实现：

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-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(title="Multi-Agent Gateway", version="0.1.0", lifespan=lifespan) 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(timeout=5) as client: resp = await client.get(f"https://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(parents=True, exist_ok=True) 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", encoding="utf-8") as f: f.write(json.dumps({"role": role, "content": content}, ensure_ascii=False) + "\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(encoding="utf-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_responses=True) self.ttl = ttl def _key(self, session_id: str) -> str: return f"memory:{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_ascii=False)) 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_tool=True；否则直接产出 final_answer：

import json from openai import AsyncOpenAI from app.config import settings from app.tools.registry import tool_registry client = AsyncOpenAI(api_key=settings.llm_api_key, base_url=settings.llm_base_url, timeout=settings.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_ascii=False), }) return messages

exec_tool节点遍历待执行的工具调用，逐个执行并把结果累积进tool_results：

class 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": f"tool {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_length=1, max_length=4000) 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_model=ChatResponse) 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, limit=settings.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_id=session_id, agent=agent_name, answer=result.get("final_answer", ""), tool_used=tool_names, )

意图路由我用了一个轻量方案：让 LLM 做一次分类，只输出 Agent 名称，再接一层白名单校验。解析失败就落到 general Agent，保证系统永远不会因为路由问题直接 500：

from openai import AsyncOpenAI from app.agents.registry import agent_registry from app.config import settings client = AsyncOpenAI(api_key=settings.llm_api_key, base_url=settings.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( model=settings.llm_model, messages=[{"role": "user", "content": prompt}], max_tokens=10, temperature=0, ) 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: str

4.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 调用统一设置了timeout=60，同时给工具执行加了一层保护：用asyncio.wait_for包住工具调用，单工具最长 10 秒，超时就按工具错误返回，不让整个图卡死。重试则用最简单的指数退避，只对网络类和 5xx 错误重试，对参数错误不重试：

import asyncio async def run_with_retry(coro_factory, retries=3): 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，编排器要用 AgentRegistry，Tool 的 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 粒度加锁。上下文超窗问题则来自另一个方向：用户聊了几百轮之后，历史消息全部塞进 prompt，Token 溢出，模型乱答。方案是只保留最近 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 工程，直接照这套骨架改一版，大概率能少熬好几个夜。