最近我一直在搞一套多Agent服务技术栈就是标题里这套组合FastAPI 提供 HTTP 网关能力LangGraph 负责 Agent 编排底层按照 Gateway / Agent / Tool / Memory 四层彻底切开。说句实在话Demo 阶段怎么玩都行但一旦要让 Agent 真的在线上服务鉴权、会话恢复、工具超时、上下文膨胀这些问题会一起涌过来。这篇就把我落地的完整工程结构、每一层为什么这么设计、关键代码怎么写的一步步拆开讲给正在从单 Agent Demo 往多 Agent 工程化方向走的人一个可以直接照抄的参考。1. 整体设计思路与分层动机1.1 为什么单一 Agent 方案撑不住复杂业务我先说结论一个 Agent 包打天下在几个工具的范围内没问题一旦业务域多起来就很痛苦。举个直观例子。假设要做企业内部的知识问答加工单流转系统用户可能问“合同审批流程是什么”也可能说“帮我开一个网络故障工单”。这两类需求如果塞进同一个 AgentPrompt 会变得极其臃肿知识检索工具、工单查询工具、权限校验工具全部混在一起。模型面对几十个工具时工具选择的准确率会明显下降而且每个工具的描述还不能写得太长否则上下文预算一下就吃光了。更麻烦的是后续一旦要新增数据分析能力你只能在原有 Prompt 上继续叠最后变成一个谁都不敢动的老系统。我最终选择了多条 Agent 线路协同的方式一个路由 Agent 负责判断用户意图再把请求分发到知识助手或工单助手。每个 Agent 只挂自己领域内的工具Prompt 短、工具少、行为可预期。这就是分层拆解带来的最直接收益——职责单一扩展成本从“修改整个系统”降级为“新增一个节点”。1.2 四层架构是怎么定下来的Gateway 层所有外部请求的入口。FastAPI 在这里接收 HTTP 请求做鉴权、限流、参数校验、会话身份提取然后把业务请求转给 Agent 编排层。这一层不关心 Agent 内部逻辑。Agent 层决策与编排。使用 LangGraph 定义状态图负责意图路由、多 Agent 调度、与工具的交互循环。这一层不关心请求从哪里来也不关心记忆存在哪个数据库。Tool 层能力的执行单元。每个工具是一个标准的封装类暴露统一的输入输出结构被 Agent 按需调用。这一层不关心 Prompt 怎么写。Memory 层记忆的存取。短期记忆就是对话上下文长期记忆则落到 Redis 或数据库供跨会话恢复。这一层对上层屏蔽存储细节。这个分层最大的价值在于每一层的依赖方向非常清晰Gateway 依赖 AgentAgent 依赖 Tool 和 MemoryTool 和 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.yml2.2 模块职责对照路径核心职责关键类或函数app/main.py创建 FastAPI 实例、注册中间件和路由create_app()app/config.py统一管理环境变量与配置项Settingsapp/gateway/dependencies.py鉴权、会话 ID 提取、当前用户上下文get_current_user()app/gateway/routes/chat.py对话接口、SSE 流式输出chat()app/agents/state.py定义图的全局状态结构GraphStateapp/agents/graph.py组装节点和边生成可执行图build_graph()app/agents/nodes/router_node.py意图识别与分发决策router_node()app/tools/base.py工具统一协议BaseToolapp/tools/registry.py维护工具名称到类的映射TOOL_REGISTRY、register_tool()app/memory/cache.pyRedis 读写短期对话记录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(titlesettings.app_name, version1.0.0) app.add_middleware(RequestContextMiddleware) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) app.include_router(health_router, prefixsettings.api_prefix) app.include_router(chat_router, prefixsettings.api_prefix) app.include_router(session_router, prefixsettings.api_prefix) return app3. 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, userDepends(get_current_user), session_id: str Depends(get_session_id), ): service AgentGraphService() result await service.run( session_idsession_id, user_inputrequest.message, user_infouser, ) return ChatResponse( code0, messageok, dataresult, )这里有个细节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, userDepends(get_current_user)): service AgentGraphService() async def event_stream(): async for event in service.run_stream( session_idrequest.session_id, user_inputrequest.message, ): if event.get(type) token: yield fdata: {json.dumps({delta: event[content]})}\n\n elif event.get(type) tool_call: yield fdata: {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_typetext/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: strmessages字段使用 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(contentresult, tool_call_idcall[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(fTool 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_responsesTrue) async def append_message(self, session_id: str, message: dict, ttl: int 3600): key fsession:{session_id}:messages await self.client.rpush(key, json.dumps(message, ensure_asciiFalse)) await self.client.expire(key, ttl) async def load_recent(self, session_id: str, limit: int 20): raw_list await self.client.lrange(fsession:{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(fsession:{session_id}:messages)注意 decode_responsesTrue 必须打开否则从 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, limit20) initial_state GraphState( messageshistory [{role: user, content: user_input}], session_idsession_id, current_agent, user_infouser_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 一次请求的完整链路把前面所有层串起来一次请求的路径是这样的前端发起 POST /api/v1/chat/stream带上 Authorization 头和 X-Session-ID。Gateway 中间件注入请求 ID记录开始时间并把请求体解析到 ChatRequest。get_current_user 校验 tokenget_session_id 解析或生成会话 ID。AgentGraphService 从 Redis MemoryCache 加载最近 20 条消息。LangGraph 从 START 进入 router 节点模型判断意图。条件边把请求分发到 knowledge_agent 或 order_agent。业务 Agent 节点绑定对应工具模型决定是否调用工具。工具执行完成后Agent 节点生成最终回答。回答写入 final_answer图执行结束。服务把最终消息写入 Redis并返回结果。SSE 流把事件逐步推给前端。这段时间里工具执行通常占了一半以上的耗时。路由判断和回复生成合并起来可能不到 3 秒工具调用如果涉及外部接口查询动辄 5 到 10 秒。所以日志里一定要记录每个节点的耗时别等排查问题的时候两眼一抹黑。7.2 部署配置与性能调优建议我只说几个被验证过比较重要的参数参数建议值原因模型温度0.2 到 0.4工程场景要求稳定太高会出现随机行为模型最大输出 token2048足够生成完整回答又不会让单轮响应时间过长工具超时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 的协同网络。这套骨架我在实际项目里反复打磨过结构本身不用怎么动往里面填业务就行。