FastAPI+LangGraph实战：多Agent服务四层架构与工程化落地

最近我一直在搞一套多Agent服务，技术栈就是标题里这套组合：FastAPI 提供 HTTP 网关能力，LangGraph 负责 Agent 编排，底层按照 Gateway / Agent / Tool / Memory 四层彻底切开。说句实在话，Demo 阶段怎么玩都行，但一旦要让 Agent 真的在线上服务，鉴权、会话恢复、工具超时、上下文膨胀这些问题会一起涌过来。这篇就把我落地的完整工程结构、每一层为什么这么设计、关键代码怎么写的，一步步拆开讲，给正在从单 Agent Demo 往多 Agent 工程化方向走的人一个可以直接照抄的参考。

---

1. 整体设计思路与分层动机

1.1 为什么单一 Agent 方案撑不住复杂业务

我先说结论：一个 Agent 包打天下，在几个工具的范围内没问题，一旦业务域多起来就很痛苦。

举个直观例子。假设要做企业内部的知识问答加工单流转系统，用户可能问“合同审批流程是什么”，也可能说“帮我开一个网络故障工单”。这两类需求如果塞进同一个 Agent，Prompt 会变得极其臃肿，知识检索工具、工单查询工具、权限校验工具全部混在一起。模型面对几十个工具时，工具选择的准确率会明显下降，而且每个工具的描述还不能写得太长，否则上下文预算一下就吃光了。更麻烦的是，后续一旦要新增数据分析能力，你只能在原有 Prompt 上继续叠，最后变成一个谁都不敢动的老系统。

我最终选择了多条 Agent 线路协同的方式：一个路由 Agent 负责判断用户意图，再把请求分发到知识助手或工单助手。每个 Agent 只挂自己领域内的工具，Prompt 短、工具少、行为可预期。这就是分层拆解带来的最直接收益——职责单一，扩展成本从“修改整个系统”降级为“新增一个节点”。

1.2 四层架构是怎么定下来的

• Gateway 层：所有外部请求的入口。FastAPI 在这里接收 HTTP 请求，做鉴权、限流、参数校验、会话身份提取，然后把业务请求转给 Agent 编排层。这一层不关心 Agent 内部逻辑。
• Agent 层：决策与编排。使用 LangGraph 定义状态图，负责意图路由、多 Agent 调度、与工具的交互循环。这一层不关心请求从哪里来，也不关心记忆存在哪个数据库。
• Tool 层：能力的执行单元。每个工具是一个标准的封装类，暴露统一的输入输出结构，被 Agent 按需调用。这一层不关心 Prompt 怎么写。
• Memory 层：记忆的存取。短期记忆就是对话上下文，长期记忆则落到 Redis 或数据库，供跨会话恢复。这一层对上层屏蔽存储细节。

这个分层最大的价值在于每一层的依赖方向非常清晰：Gateway 依赖 Agent，Agent 依赖 Tool 和 Memory，Tool 和 Memory 彼此独立。任何一个模块替换或升级，都不会牵动其他层。比如我把知识检索从 Elasticsearch 换成向量数据库，只需要改 KnowledgeSearchTool 内部实现，Agent 和 Gateway 完全不用动。

1.3 为什么选 LangGraph 而不是隐式循环的 Agent 执行器

LangGraph 最核心的特点是“把 Agent 流程画成一张显式的图”。节点和边都是代码里看得见的状态转换，每一步都能打断、能调试、能插入人工确认。

而很多早期框架提供的 Agent 执行器是隐式的 while 循环：模型决定调用工具，执行工具，再回到模型，直到模型认为可以结束。这种模式跑简单任务很顺手，但复杂的场景里你会发现很难控制流程。比如你想在某个 Agent 执行完工具后强制走一个人工审批节点，隐式循环里很难优雅地表达。LangGraph 的 StateGraph 允许你直接定义节点、条件边、结束点，还能在任意节点中断并恢复，这对真实业务太重要了。

另外 LangGraph 的多 Agent 编排也方便，多个 Agent 就是多个节点，节点之间通过 State 共享数据，路由逻辑写在条件边里。相比自己手写一个分发循环，图的语义清晰很多，排查问题的时候也能看明白数据到底走到了哪个节点。

---

2. 工程目录结构与模块职责

2.1 完整目录树

我先把工程目录完整贴出来，后面每个部分都会对应到这个结构里。这套目录是我重构过三轮之后觉得最顺手的形态。

fastapi-langgraph-agent/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── config.py # 全局配置，基于 pydantic-settings │ ├── gateway/ │ │ ├── __init__.py │ │ ├── dependencies.py # 鉴权依赖、会话依赖 │ │ ├── middleware.py # 请求ID注入、耗时统计中间件 │ │ ├── schemas.py # 请求/响应模型 │ │ └── routes/ │ │ ├── __init__.py │ │ ├── chat.py # 对话与流式接口 │ │ ├── health.py # 健康检查 │ │ └── sessions.py # 会话管理接口 │ ├── agents/ │ │ ├── __init__.py │ │ ├── state.py # 图状态定义 │ │ ├── graph.py # 状态图构建 │ │ └── nodes/ │ │ ├── __init__.py │ │ ├── router_node.py # 意图路由节点 │ │ ├── knowledge_agent.py # 知识问答 Agent │ │ └── order_agent.py # 工单流转 Agent │ ├── tools/ │ │ ├── __init__.py │ │ ├── base.py # 工具基类与协议 │ │ ├── registry.py # 全局工具注册表 │ │ └── implementations/ │ │ ├── __init__.py │ │ ├── knowledge_search.py # 知识库检索工具 │ │ ├── order_create.py # 创建工单工具 │ │ └── order_query.py # 工单查询工具 │ └── memory/ │ ├── __init__.py │ ├── cache.py # Redis 短期状态操作 │ ├── store.py # 长期消息记录存储 │ └── schemas.py # 消息数据模型 ├── tests/ │ ├── conftest.py │ ├── test_gateway.py │ ├── test_graph.py │ └── test_tools.py ├── pyproject.toml └── docker-compose.yml

2.2 模块职责对照

路径	核心职责	关键类或函数
app/main.py	创建 FastAPI 实例、注册中间件和路由	create_app()
app/config.py	统一管理环境变量与配置项	Settings
app/gateway/dependencies.py	鉴权、会话 ID 提取、当前用户上下文	get_current_user()
app/gateway/routes/chat.py	对话接口、SSE 流式输出	chat()
app/agents/state.py	定义图的全局状态结构	GraphState
app/agents/graph.py	组装节点和边生成可执行图	build_graph()
app/agents/nodes/router_node.py	意图识别与分发决策	router_node()
app/tools/base.py	工具统一协议	BaseTool
app/tools/registry.py	维护工具名称到类的映射	TOOL_REGISTRY、register_tool()
app/memory/cache.py	Redis 读写短期对话记录	RedisMemoryCache.append() / load()
app/memory/store.py	长期消息归档到数据库	MessageStore.save_message()

我一直强调一个原则：路由文件只做一件事——把 HTTP 请求参数转换为内部调用，再把 Agent 层的返回值包装成 HTTP 响应。业务逻辑绝不能写在路由函数里，否则单元测试会非常痛苦。

2.3 配置管理与应用入口

配置这块我用 pydantic-settings 统一管理。好处是所有的密钥、超时时间、模型参数、Redis 地址都集中在一个 Settings 类里，写测试时还能直接替换配置对象。

from functools import lru_cache from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str = "agent-gateway" api_prefix: str = "/api/v1" secret_key: str = "change-me-in-prod" redis_url: str = "redis://localhost:6379/0" llm_model_name: str = "local-model" llm_temperature: float = 0.3 llm_max_tokens: int = 2048 session_ttl: int = 3600 tool_timeout: int = 30 enable_request_log: bool = True class Config: env_file = ".env" env_file_encoding = "utf-8" @lru_cache def get_settings() -> Settings: return Settings()

main.py 的组装逻辑也相对固定：

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.config import get_settings from app.gateway.middleware import RequestContextMiddleware from app.gateway.routes.chat import router as chat_router from app.gateway.routes.health import router as health_router from app.gateway.routes.sessions import router as session_router def create_app() -> FastAPI: settings = get_settings() app = FastAPI(title=settings.app_name, version="1.0.0") app.add_middleware(RequestContextMiddleware) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) app.include_router(health_router, prefix=settings.api_prefix) app.include_router(chat_router, prefix=settings.api_prefix) app.include_router(session_router, prefix=settings.api_prefix) return app

---

3. Gateway 层：FastAPI 如何承接外部请求

3.1 路由设计与统一响应结构

Gateway 层的路由设计我遵循“一个业务入口一个文件”的方式。以对话接口为例，路由函数只做四件事：解析请求、鉴权、调用下游编排器、返回响应。

我这里为所有接口统一了一个响应包装结构，包含 code、message、data 三个字段。虽然会增加一点代码量，但前端错误处理会非常舒服，日志追踪也一致。

from fastapi import APIRouter, Depends from app.gateway.schemas import ChatRequest, ChatResponse from app.gateway.dependencies import get_current_user, get_session_id from app.agents.graph import AgentGraphService router = APIRouter(prefix="/chat", tags=["chat"]) @router.post("") async def chat( request: ChatRequest, user=Depends(get_current_user), session_id: str = Depends(get_session_id), ): service = AgentGraphService() result = await service.run( session_id=session_id, user_input=request.message, user_info=user, ) return ChatResponse( code=0, message="ok", data=result, )

这里有个细节：get_session_id依赖从请求头里取X-Session-ID，取不到就生成一个新的 UUID，并在响应头上写回去。这样前端第一次请求之后就能拿到会话 ID，后续多轮对话都带着它，会话恢复就有了依据。

3.2 鉴权、限流与请求上下文

鉴权在 FastAPI 里用依赖注入是最舒服的方式。我封装了一个get_current_user依赖，从 Authorization 头解析 token，然后去用户服务或者本地配置里校验。这样做的好处是：任何一个需要登录的接口，只需要在参数里声明这个依赖，框架会自动执行鉴权，不需要在业务函数里重复写校验代码。

限流我最初直接写了一个基于内存滑动窗口的装饰器，后来并发测试发现单机内存在高并发下容易抖，就换成了基于 Redis 的固定窗口限流。核心逻辑也很简单：以用户 ID 加接口路径为 key，记录窗口内请求次数，超过阈值直接返回 429。

在中间件层面，我做了两件看似简单但很有用的事情。第一件事是给每个请求注入X-Request-ID，用 uuid4 生成，日志里全程透传。出了问题想排查链路，直接 grep 请求 ID 就能串起来。第二件事是记录每个请求的耗时和状态码，用异步日志队列写入收集端，方便后续观测。

3.3 流式输出：从 LangGraph 到 SSE

用户最直观的体验是逐字输出。FastAPI 实现流式输出很容易，关键是 LangGraph 执行过程中的事件要能一层层传到前端。

我在 AgentGraphService 里提供了一个异步生成器，用 LangGraph 的 astream 模式逐事件产出。Gateway 层只需要把这个事件流包装成 Server-Sent Events 格式。这里贴一段核心代码，标记处就是经过验证的写法：

from fastapi.responses import StreamingResponse import json @router.post("/stream") async def chat_stream(request: ChatRequest, user=Depends(get_current_user)): service = AgentGraphService() async def event_stream(): async for event in service.run_stream( session_id=request.session_id, user_input=request.message, ): if event.get("type") == "token": yield f"data: {json.dumps({'delta': event['content']})}\n\n" elif event.get("type") == "tool_call": yield f"data: {json.dumps({'tool': event['name'], 'args': event['args']})}\n\n" elif event.get("type") == "done": yield "data: [DONE]\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 防止反代缓冲 }, )

X-Accel-Buffering这个 header 值得单独说。很多反代服务器默认会缓冲响应，导致流式输出的第一个 token 迟迟到不了前端。加上no之后，反代就会把数据实时转发。我第一次上线时忘了这一行，结果前端等到整个响应结束才看到文字，排查了很久才发现是这个缓存头的问题。

---

4. Agent 层：LangGraph 状态图与多 Agent 编排

4.1 图的状态设计

LangGraph 的状态设计是整个 Agent 层的地基。每个节点读入状态，处理后更新状态，状态在节点之间单向流动。

我的 GraphState 长这样：

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class GraphState(TypedDict): messages: Annotated[list, add_messages] # 对话历史，add_messages 自动合并 session_id: str current_agent: str user_info: dict tool_results: dict final_answer: str

messages字段使用 add_messages 归约器，LangGraph 会自动把新消息追加到历史列表，而不是覆盖原来的值。这一点很关键，因为多个节点都会往 messages 里写内容，如果没有归约器，后写的节点会直接把前面的消息清掉。

tool_results用普通 dict，记录工具名到结果的映射，方便 Agent 节点在生成最终回复时引用。

4.2 构建状态图与条件路由

构建图的时候，我定义了一个入口路由节点、两个业务 Agent 节点和一个结束点。

from langgraph.graph import StateGraph, START, END def build_graph(): builder = StateGraph(GraphState) builder.add_node("router", router_node) builder.add_node("knowledge_agent", knowledge_agent_node) builder.add_node("order_agent", order_agent_node) builder.add_edge(START, "router") builder.add_conditional_edges( "router", decide_next_node, { "knowledge": "knowledge_agent", "order": "order_agent", "end": END, }, ) builder.add_edge("knowledge_agent", END) builder.add_edge("order_agent", END) return builder.compile()

路由节点的工作是让模型从预定义的意图列表里做选择。我强制模型返回一个结构化的 JSON，而不是直接生成自由文本，否则后续的条件边很难稳定判断。代码大致是这样：

async def router_node(state: GraphState) -> dict: system_prompt = ( "你是意图路由助手。根据用户的最近一条消息，判断属于哪个意图。\n" "可选意图：knowledge（知识库问答）、order（工单操作）。\n" "只输出 JSON，格式：{\"intent\": \"knowledge\"}\n" ) resp = await chat_model.invoke([ {"role": "system", "content": system_prompt}, *state["messages"][-3:], # 只看最近几条，够判断意图 ]) intent = parse_json(resp.content).get("intent", "end") return {"current_agent": intent}

decide_next_node是一个纯函数，读state["current_agent"]并返回对应的节点名称。这里有个容易踩的坑：路由节点返回的 key 必须和add_conditional_edges的映射完全一致，空格、大小写都不能差，否则图会直接走默认分支或者报错。

4.3 多 Agent 协同的三种模式与取舍

多 Agent 协同我不建议一开始就上复杂的协商机制，工程上常用的就三种，按复杂度排序：

• 顺序流水线：Agent A 的输出作为 Agent B 的输入，适合处理流程固定、阶段清晰的任务。
• 路由分发：一个路由节点把请求分发给不同业务的 Agent，各 Agent 独立完成后返回。我这个系统用的就是这种。
• 共享状态协商：多个 Agent 在同一个共享状态上反复读写、互相修正，适应高度动态的任务，但对状态设计的要求很高，调试成本也大。

大部分企业内部场景用路由分发就够用。顺序流水线和协商式架构会让链路变长，任何一个节点出问题都会拖慢整体响应。等业务真的复杂到需要多个 Agent 来回讨论的时候，再升级也不迟。

4.4 Agent 内部如何调用工具

业务 Agent 节点内部按“绑定工具 → 模型决策 → 执行工具 → 回填结果”的循环来执行。LangGraph 支持节点内部自行完成一个完整的 Agent 循环，也可以拆成多个节点。为了保持图逻辑简单，我在每个业务 Agent 节点内部完成循环，但限制最多迭代 3 轮，防止模型陷入无限调工具的循环。

核心逻辑如下：

async def knowledge_agent_node(state: GraphState) -> dict: tool_objs = [ KnowledgeSearchTool(), ] model_with_tools = chat_model.bind_tools(tool_objs) msgs = state["messages"][-6:] # 取最近几轮消息作为上下文 full_agent_messages = [] for _ in range(3): resp = await model_with_tools.invoke(msgs) tool_calls = getattr(resp, "tool_calls", []) if not tool_calls: return {"final_answer": resp.content} for call in tool_calls: result = await execute_tool(call["name"], call["args"]) msgs.append(ToolMessage(content=result, tool_call_id=call["id"])) return {"final_answer": "抱歉，我暂时无法处理这个问题。"}

这里要注意：bind_tools的时候传入的是工具实例的 schema 描述，执行工具时再通过注册表找到对应的实现。定义和执行的解耦，是 Tool 层存在的意义。

---

5. Tool 层：注册机制与协议统一

5.1 工具基类与协议

Tool 层我坚持自定义一个薄薄的基类，而不是直接用框架自带的工具装饰器。原因有两个。第一，自带的装饰器把函数直接变成框架内部的 Runnable，单元测试时依赖较重；第二，装饰器方式不利于把工具按业务域归类管理，我在工程里更希望有一个清晰的类层次。

工具基类长这样：

from abc import ABC, abstractmethod from pydantic import BaseModel class BaseTool(ABC): name: str = "" description: str = "" args_schema: type[BaseModel] | None = None def get_schema(self) -> dict: return { "name": self.name, "description": self.description, "parameters": self.args_schema.model_json_schema() if self.args_schema else {}, } @abstractmethod async def run(self, **kwargs) -> str: """工具实际执行逻辑，返回字符串结果""" ...

run方法必须返回字符串。我统一让所有工具把结果序列化成文本，这样模型可以直接消费工具输出，不需要再做一次类型转换。如果工具的原始结果是结构化数据，就在工具内部先转好。

5.2 注册表与自动发现

工具注册我用一个全局字典加装饰器实现。装饰器把类名注册进字典，key 是工具名，value 是工具类本身。

TOOL_REGISTRY: dict[str, type[BaseTool]] = {} def register_tool(cls: type[BaseTool]) -> type[BaseTool]: if not cls.name: raise ValueError(f"Tool class {cls.__name__} must define name") TOOL_REGISTRY[cls.name] = cls return cls async def execute_tool(name: str, args: dict) -> str: tool_cls = TOOL_REGISTRY.get(name) if not tool_cls: return f"未知工具: {name}" tool = tool_cls() try: return await tool.run(**args) except Exception as e: return f"工具执行失败: {e}"

每个业务 Agent 节点只暴露自己的工具列表。知识 Agent 只绑定 KnowledgeSearchTool，工单 Agent 只绑定 OrderCreateTool 和 OrderQueryTool。这样做模型面对的工具数量少，选错工具的概率显著下降。

5.3 工具执行的容错与输出截断

工具执行有三个问题躲不开：超时、异常、输出过长。

超时我用 asyncio 的 wait_for 包一层，工具调用超过设定秒数就直接返回失败提示，避免整个请求卡死在外部接口上。异常处理上面已经演示过，统一捕获后把错误信息返回给模型，让模型根据错误信息决定下一步。输出截断则是我踩过最深的坑之一。知识检索工具返回的文档片段往往非常长，如果直接塞给模型，很快就把上下文窗口占满，导致模型根本没有空间生成回复。我现在统一在工具层做截断，超过 800 字的部分用“...（内容过长已省略）”替代。

---

6. Memory 层：跨会话记忆与持久化

6.1 短期记忆与上下文窗口管理

短期记忆就是当前对话窗口内的消息列表。LangGraph 的 messages 字段天然承担了短期记忆的职责，节点往里面加消息就行。但 messages 不能无限增长，否则 token 成本会爆炸。我采用“滑动窗口 + 首轮摘要”的方式管理。

具体做法：当 messages 里的消息数量超过 20 条，就保留最开始的 system 提示和最近的 12 条，把中间的部分让模型生成一段摘要。这样用户聊到第 100 轮，每轮仍然只带最近 12 条消息和一个历史摘要，模型始终有一个大小稳定的上下文。

6.2 长期记忆与持久化存储

长期记忆解决的是跨会话恢复问题。用户退出再回来，系统需要知道之前的对话内容。

我在工程里同时用了 Redis 和数据库。Redis 存短期活跃会话，设置过期时间；数据库做消息的完整归档。网关层通过会话 ID 从 Redis 拉取最近的对话历史，如果 Redis 里没有，则从数据库恢复。

Redis 读写代码：

import json import redis.asyncio as redis class RedisMemoryCache: def __init__(self, redis_url: str): self.client = redis.from_url(redis_url, decode_responses=True) async def append_message(self, session_id: str, message: dict, ttl: int = 3600): key = f"session:{session_id}:messages" await self.client.rpush(key, json.dumps(message, ensure_ascii=False)) await self.client.expire(key, ttl) async def load_recent(self, session_id: str, limit: int = 20): raw_list = await self.client.lrange(f"session:{session_id}:messages", -limit, -1) return [json.loads(raw) for raw in raw_list] async def clear(self, session_id: str): await self.client.delete(f"session:{session_id}:messages")

注意 decode_responses=True 必须打开，否则从 Redis 取出来的都是字节串，每次都要手动 decode，非常麻烦。

6.3 会话恢复与多轮一致性

LangGraph 每次请求都是重新构建图去执行，所以恢复会话的核心是“先把历史消息灌进初始状态，再启动图”。在 AgentGraphService 里体现得很直接：

async def run(self, session_id: str, user_input: str, user_info: dict): history = await self.memory.load_recent(session_id, limit=20) initial_state = GraphState( messages=history + [{"role": "user", "content": user_input}], session_id=session_id, current_agent="", user_info=user_info, tool_results={}, final_answer="", ) result = await self.graph.ainvoke(initial_state) await self.memory.append_message(session_id, {"role": "user", "content": user_input}) await self.memory.append_message(session_id, {"role": "assistant", "content": result["final_answer"]}) return result

因为 add_messages 归约器会自动合并列表，所以可以直接把 Redis 加载的历史消息放进初始 messages。这里有一个非常重要但容易忽略的问题：直接往 Redis 里存原始消息，只适合内部系统。如果涉及用户隐私，消息存储前必须做脱敏处理，比如把姓名、手机号替换成占位符。

---

7. 完整调用链路与关键参数配置

7.1 一次请求的完整链路

把前面所有层串起来，一次请求的路径是这样的：

1. 前端发起 POST /api/v1/chat/stream，带上 Authorization 头和 X-Session-ID。
2. Gateway 中间件注入请求 ID，记录开始时间，并把请求体解析到 ChatRequest。
3. get_current_user 校验 token，get_session_id 解析或生成会话 ID。
4. AgentGraphService 从 Redis MemoryCache 加载最近 20 条消息。
5. LangGraph 从 START 进入 router 节点，模型判断意图。
6. 条件边把请求分发到 knowledge_agent 或 order_agent。
7. 业务 Agent 节点绑定对应工具，模型决定是否调用工具。
8. 工具执行完成后，Agent 节点生成最终回答。
9. 回答写入 final_answer，图执行结束。
10. 服务把最终消息写入 Redis，并返回结果。
11. SSE 流把事件逐步推给前端。

这段时间里，工具执行通常占了一半以上的耗时。路由判断和回复生成合并起来可能不到 3 秒，工具调用如果涉及外部接口查询，动辄 5 到 10 秒。所以日志里一定要记录每个节点的耗时，别等排查问题的时候两眼一抹黑。

7.2 部署配置与性能调优建议

我只说几个被验证过比较重要的参数：

参数	建议值	原因
模型温度	0.2 到 0.4	工程场景要求稳定，太高会出现随机行为
模型最大输出 token	2048	足够生成完整回答，又不会让单轮响应时间过长
工具超时	10 到 30 秒	低于 10 秒外部接口容易误报，高于 30 秒用户体验太差
会话过期	1 小时	一小时基本覆盖一次连续交互，过长浪费 Redis 内存
滑动窗口	最近 20 条	兼顾上下文与 token 成本

在我那台 8C16G 的测试服务器上，本地部署的模型参数量中等，并发 20 路请求时，平均首 token 延迟在 0.8 到 1.5 秒左右，完整回答时间 3 到 6 秒。其中一次工具调用约占 40% 到 50% 耗时，模型推理占 30%，剩余是网络开销。如果想优化，优先优化工具所需的外部接口响应速度，比换一个更大的模型收益更明显。

---

8. 常见问题与排查技巧实录

8.1 问题速查表

现象	常见原因	解决办法
条件边不生效，请求直接到 END	路由节点返回的 key 与分支映射不一致	在路由节点里额外输出一个 debug 字段，日志打印实际分支
工具一直不被模型调用	工具描述写得太模糊，或参数 schema 有字段没加描述	强化 description，给每个参数补充英文说明
工具输出过长导致模型报错	工具结果未截断，超出上下文预算	在工具层做 max_chars 截断，加省略标记
流式输出首 token 很慢	反代缓冲了 SSE 响应	加上 X-Accel-Buffering: no 响应头
高并发时会话内容串号	把 session_id 错误地存到了类实例变量	所有状态必须放 GraphState 或显式参数传递
Redis 报连接池耗尽	每个请求都新建连接	全局复用 Redis 客户端连接池
模型重复调用同一个工具	缺少最大迭代限制	Agent 循环设置最多 3 轮

8.2 两个踩坑过程实录

第一个坑是多 Agent 之间状态污染。我在第一版代码里图省事，把 GraphState 的消息列表定义成了一个模块级变量。单线程测试一切正常，并发一上来，A 用户的消息串到了 B 用户的上下文里。排查了很久才意识到：LangGraph 的 StateGraph 每次执行时会创建独立的 state 副本，但外部模块级变量不会跟着复制。解决办法就是彻底放弃所有全局可变数据结构，所有数据一律通过 State 显式传递，并通过 session_id 隔离。

第二个坑是工具并行执行导致的顺序错乱。早期版本里，如果模型一次调用多个工具，我直接并发执行，然后把结果按完成顺序写回 messages。结果模型的 tool_call_id 对不上执行结果，出现工具结果串位。后来统一按 tool_call_id 匹配，先收集工具调用列表，并发执行后按原始顺序重新排序回填，问题立刻解决。

8.3 如果让我重来一次

整个系统我重构了三次，最后沉淀下来的核心设计原则只有三条：状态必须显式、工具必须可观测、上下文必须受控。

显式状态的意思是所有跨节点传递的数据都走 GraphState，不要依赖隐式对象和全局变量。可观测的意思是每个工具调用都要有日志，包括入参、耗时、返回结果，这是排查 Agent 行为异常的保底手段。受控的上下文意味着消息列表永远有上限，Tool 结果永远有截断，agent 循环永远有最大轮数。这三条守住，多 Agent 系统的稳定性基本就有了保障。

如果你是第一次把 FastAPI 和 LangGraph 组合成一个多 Agent 工程，照着这个目录结构和分层思路搭一版绝对不会亏。先别急着加复杂功能，把一两个业务域完整跑通，再考虑扩展成更多 Agent 的协同网络。这套骨架我在实际项目里反复打磨过，结构本身不用怎么动，往里面填业务就行。