2026/10/9 16:18:32

FastAPI + LangGraph 构建多智能体协作服务:状态编排与并发实践

FastAPI + LangGraph 构建多智能体协作服务:状态编排与并发实践 最近一段时间我一直在忙一个内部项目把研究分析这条流程做成多智能体协作服务。需求本身不复杂——用户提交一个主题系统自动派出多个智能体分工干活——一个去收集信息、一个做数据整理、一个负责生成带观点的分析报告最终把结果通过 API 返回给前端页面。真正落地的时候才发现难点根本不在单个模型的聪明程度而在于如何把多个 Agent 放在一起编排、如何让状态在步骤之间可靠传递、又如何在这一堆 Agent 之上做一层稳定的 HTTP 服务。这篇文章围绕我踩过的坑和最终方案来写主角是 FastAPI、LangGraph 和本地模型服务 Ollama。如果你正准备用 Python 做多智能体后端服务或者已经在用 LangChain 但感觉拼 Prompt 写循环的方式撑不起复杂场景那这篇内容应该能帮你少走不少弯路。1. 为什么是 FastAPI LangGraph先把分工想清楚1.1 多智能体服务里真正难的部分很多人一开始把多智能体想得太简单不就是写几个 System Prompt、定义几个函数、按顺序调用吗单智能体确实可以这么干但一旦进入多智能体协作问题立刻变了。你需要处理的不只是模型说什么还有步骤走到哪了上一步产出的数据下一步怎么用某个 Agent 失败了要不要重试用户中途打断后状态怎么办。这些问题的本质是协作流程是有状态的图而不是一段线性脚本。收集信息的 Agent 可能要循环好几轮才能凑够资料分析数据的 Agent 分析完发现信息不足还得回退让收集 Agent 再补一轮。这种条件回退自环重试多分支汇聚是写 if-else 或 for 循环很难维护的。所以我把整个服务拆成两层API 层负责接收请求、返回结果、管理 HTTP 生命周期编排层负责 Agent 之间的流转、状态传递和重试逻辑。FastAPI 管前者LangGraph 管后者两者之间只通过一个编译好的图对象交互。1.2 API 框架选型为什么是 FastAPI 而不是 Flask后端框架我只对比过 FastAPI 和 Flask因为这是 Python 世界里最容易纠结的一组。如果你的项目只是内部小工具、接口数量一只手数得过来Flask 确实够用我也用了很多年。但这次接的是多智能体服务情况不太一样。对比下来我的选择逻辑是这样对比维度FlaskFastAPI异步支持需要额外配 asyncio 插件默认同步原生 async/await直接支持请求参数校验手写校验逻辑容易漏边界Pydantic 模型声明自动校验接口文档需要装 flasgger 之类自动生成 OpenAPI 文档并发性能同步线程池长任务占线程异步事件循环高并发 IO 友好WebSocket/SSE需要额外库支持原生支持 StreamingResponse多智能体服务有个明显特点接口往往需要长时间占用连接因为 Agent 要跑好几轮大模型调用每次都是秒级延迟用户端还希望流式看到进度。FastAPI 的异步机制加上 SSEServer-Sent Events支持正好是干这个的。实测下来 FastAPI 在长连接场景下资源占用明显更可控而且 Pydantic 做的请求校验帮我拦住了不少非法输入——比如用户传了一个 3000 字的 topic或者 thread_id 传了空字符串。如果你在 Flask 和 FastAPI 之间举棋不定我的建议很直接新项目、要异步、要自动文档、要做流式直接上 FastAPI别犹豫。Flask 的生态成熟度是个优势但多智能体服务这个场景太吃异步能力和长连接了。1.3 LangGraph 和手写循环调用 Agent的差别再来说编排层。LangChain 本身提供了 Chain 的抽象早期我做多 Agent 时也用过但很快发现几个问题Chain 偏向线性流程条件分支要靠 RouteChain 之类的小工具拼代码一多就成了意大利面条另一方面状态管理不够显式一个 Agent 的输出只要结构稍有变化下一个 Agent 解析的时候就容易崩。LangGraph 的解法是彻底换思路把流程画成一张图。节点是做什么调用哪个 Agent、执行什么工具边是下一步去哪状态是一份全局共享的数据结构每个节点读写这份状态。这个模型天然覆盖了我前面说的几种复杂情况自环重试一条边从节点 A 指回节点 A 自己条件回退节点 B 执行完根据状态里的某个字段选择去 C 还是回 A并行分支多个节点指向同一个下游节点自动汇聚LangGraph 还有检查点机制Checkpointer可以把每一步的状态存下来支持断点续跑、人工介入、多轮会话记忆。这是 LangChain 普通 Chain 给不了的能力。我用下来最大的感受是调试多智能体流程时你能清楚地看到每一步状态长什么样、哪一步出了问题、是哪条路走错了——这在手写循环里几乎做不到。2. 项目骨架一个能直接开工的目录结构2.1 目录结构总览网上聊 FastAPI 项目目录结构的文章不少但落到多智能体项目上的不多。我一开始也踩过把所有东西塞进 main.py的坑。这个项目里我先定义需求边界再拆目录API 层归 API 层Agent 层归 Agent 层图的编排单独放一块通用服务Ollama 客户端、日志、配置独立成模块。最终结构长这样research_agent/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── core/ │ │ ├── config.py # 配置管理环境变量、路径 │ │ └── logging.py # 日志初始化 │ ├── api/ │ │ ├── __init__.py │ │ └── agent_routes.py # /agent/run、/agent/stream 路由 │ ├── agents/ │ │ ├── __init__.py │ │ ├── collector.py # 信息收集 Agent │ │ ├── analyst.py # 数据分析 Agent │ │ └── writer.py # 报告撰写 Agent │ ├── graph/ │ │ ├── __init__.py │ │ ├── state.py # 状态数据结构定义 │ │ ├── build_graph.py # 图的构建与编译 │ │ └── router.py # 条件路由逻辑 │ └── services/ │ ├── __init__.py │ └── ollama_client.py # 本地模型调用封装 ├── tests/ │ ├── test_routes.py │ └── test_graph.py ├── pyproject.toml ├── .env.example └── README.md这个结构每个目录的职责很单纯。我坚持的原则是agents 目录里的文件只关心怎么调模型、怎么写 Prompt、怎么解析结果graph 目录只关心流程怎么编排、状态怎么流转api 目录只关心请求怎么进来、响应怎么回去。这样改一个 Agent 的逻辑不会动到路由代码改路由的结构也不会碰到底层的图定义。2.2 各模块职责和边界我把目录拆这么细不是追求形式上的整洁而是基于一次真实的事故。项目早期我把 Agent 的调用逻辑直接写在路由函数里结果调试的时候来回翻文件改一个参数要同时动三四个地方。后来才体会到多智能体项目的复杂度主要集中在图编排和状态流转上这部分必须独立出来否则后面加新 Agent、加新路由分支时你会被自己的代码乱到怀疑人生。具体拆法如下core/config.py集中管理配置项包括 Ollama 地址、模型名称、超时时间、日志级别等。用 Pydantic Settings 读取环境变量避免到处硬编码。agents/ 目录每个文件实现一个 Agent。Agent 的本质是一个接收 state、返回 state 更新的函数。它内部负责拼 Prompt、调模型、解析结果然后往 state 里写入自己负责的那部分字段。graph/ 目录state.py 定义全局状态数据结构build_graph.py 把各 Agent 函数组装成图router.py 放条件路由的判断逻辑。这个目录是整个服务的编排核心。api/ 目录负责暴露 HTTP 接口做参数校验、调用编译好的图、把结果格式化返回。它不关心图内部怎么流转。2.3 依赖锁定与环境配置依赖管理我用的是 pyproject.toml没有用 requirements.txt。原因很简单这个项目要兼顾开发、测试、部署还需要锁定 LangGraph 这种迭代比较快的库的版本。用 Poetry 管理依赖能精确锁定版本还能导出 requirements.txt部署的时候不慌。核心依赖就这些[project] dependencies [ fastapi0.115, uvicorn[standard]0.30, langgraph0.2, langchain-core0.3, httpx0.27, pydantic-settings2.4, ]配置方面创建一个 .env.example 文件把关键配置都放进去避免配置写死在代码里OLLAMA_BASE_URLhttp://localhost:11434 OLLAMA_MODELqwen2.5:14b OLLAMA_TIMEOUT120 LOG_LEVELINFO我的一个实际经验是LangGraph 的版本兼容性问题比 FastAPI 更麻烦尤其当你同时用 LangChain 的 message 工具和自定义检查点存储时。建议一开始就把版本锁住升级的时候单独拉分支测别直接升最新版。3. 编排 Multi-Agent 协作StateGraph 是灵魂3.1 场景设计研究分析流水线进入核心部分。我为这个项目设计的场景是主题研究报告生成用户给一个主题系统自动产出结构化报告。整个过程切分成三个 Agent每个 Agent 只负责一件事Agent 节点职责输入输出到 Statecollect收集与主题相关的资料、文献、数据点topiccollected_dataanalyze对收集到的数据做模式分析和质量评估collected_dataanalysis_resultwrite基于分析结果生成最终报告analysis_resultreport光看这一字排开的流程似乎顺序调用就够了。但真实场景里有个关键点collect 节点不一定一次就能收集到足够的数据。如果模型判断收集的信息太少、覆盖维度不够就要重新收集一轮analyze 节点分析完如果发现数据质量不行还得回退让 collect 再补。这种动态流转才是用图的真正价值。3.2 状态定义与自定义 reducerLangGraph 里所有节点共享一份状态。状态类型用 TypedDict 定义字段就是各个 Agent 之间的接口契约。我在 state.py 里这样定义from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class ResearchState(TypedDict): topic: str collected_data: list[dict] analysis_result: dict report: str current_step: str messages: Annotated[list, add_messages]这里有两个细节需要解释。第一messages 字段使用了Annotated[list, add_messages]这是 LangGraph 提供的一个 reducer 写法。reducer 的作用是定义多节点往同一个字段写值时怎么合并。默认情况下后写入的值会覆盖先写入的值但如果用的是add_messagesreducer新消息会追加到列表后面而不是覆盖。在多智能体协作场景里每个 Agent 都可能产生日志消息或中间思考内容这个追加行为非常关键。第二current_step字段是我额外加的。实际调试时我发现LangGraph 虽然能看到当前执行到哪个节点但状态里不一定有完整的流转轨迹。我让每个 Agent 节点都把步骤名写进 state这样既能在流式返回时告诉前端现在在做什么排查问题也多了一条线索。这个小字段帮我省了不少事。3.3 节点实现Agent 的封装方式LangGraph 的节点本质就是一个普通函数接收整个 state返回部分更新的 state。我不建议把 Agent 写成一堆类方法简单函数反而更好测试。以 collect 节点为例async def collect_node(state: ResearchState) - dict: topic state[topic] prompt ( 你是一名资深行业研究员。请围绕用户给出的主题 收集至少5条关键信息。每条信息需要包含来源类型、结论、数据依据。 f主题是{topic} ) result await ollama_client.generate( modelsettings.ollama_model, promptprompt, temperature0.3, ) # 解析模型输出格式化为列表 items parse_result(result) return { collected_data: items, current_step: collect, }这里有一个值得注意的点节点的返回结果只会更新 state 里那一个字段其他字段不受影响。所以每个 Agent 只需要关注自己负责的字段。collect 节点返回collected_dataanalyze 节点返回analysis_resultwrite 节点返回report。这种各人自扫门前雪的机制后续加新 Agent 时几乎不用改老代码。实际开发中我建议把模型调用从 Agent 函数里抽出来统一走 services/ollama_client.py。原因有两条一是后续换模型服务商只改 client 一个文件二是单元测试时可以直接 mock client不用真的调用本地模型。3.4 条件路由不靠硬编码靠图的结构真正的编排逻辑集中在路由函数里。我在 graph/router.py 中定义了是否继续收集和是否回退的判断def route_after_collect(state: ResearchState) - str: if len(state[collected_data]) 5: return need_more_data return analyze def route_after_analyze(state: ResearchState) - str: quality_score state[analysis_result].get(quality_score, 0) if quality_score 0.6: return collect return write然后在 build_graph.py 中把图组装起来from langgraph.graph import StateGraph, END graph StateGraph(ResearchState) graph.add_node(collect, collect_node) graph.add_node(analyze, analyze_node) graph.add_node(write, write_node) graph.set_entry_point(collect) graph.add_conditional_edges( collect, route_after_collect, { need_more_data: collect, analyze: analyze, } ) graph.add_conditional_edges( analyze, route_after_analyze, { collect: collect, write: write, } ) graph.add_edge(write, END) compiled_graph graph.compile()这个图表达的逻辑是收集节点完成后如果数据不足图会有一条边指回 collect 节点自己触发下一轮收集分析节点判断数据质量过低时同样有一条回退边。这就是多智能体里最常见的自组织流程。你不需要在代码里写while循环控制重试次数LangGraph 自己会在图里跑这些边。限制重试层数可以在单层节点内部实现也可以外面加一个 session 级超时。我对这类设计的体会是把路由判断集中到一个文件里太重要了。刚开始我图省事把条件判断直接写在节点函数里用返回值控制流程结果图的结构散落在一堆 Agent 代码里。后来全挪到 router.py整个图一眼就能看懂改流程也变成改一行字典的事。4. FastAPI 接入层把图安全地暴露成接口4.1 服务类封装与共享实例编译好的图是一个重量级对象需要在应用启动时创建一次多个请求共用。我通常在 build_graph.py 最后实例化一个全局的compiled_graph在路由里直接导入。很多人喜欢在 main.py 里用一个全局变量挂着图对象也能跑但导入路径一复杂就乱了。我用一个简单的服务类把图包一层class AgentService: def __init__(self, graph): self.graph graph async def run(self, topic: str, thread_id: str | None None): config {configurable: {thread_id: thread_id or uuid4().hex}} result await asyncio.to_thread( self.graph.invoke, {topic: topic}, configconfig, ) return result agent_service AgentService(compiled_graph)这里用asyncio.to_thread是因为我们的图内部虽然用了 async 节点但有些兼容性场景下 invoke 是同步调用的。为了避免阻塞事件循环我用 to_thread 把 invoke 放到线程池里跑。如果你的图是纯异步的直接await self.graph.ainvoke(...)也行。4.2 HTTP 路由设计与参数校验路由层的设计以简单为主。我定义了两个端点一个非流式返回完整报告一个流式返回过程事件。请求参数用 Pydantic 模型声明FastAPI 会自动完成校验非法参数直接返回 422from pydantic import BaseModel, Field class RunRequest(BaseModel): topic: str Field(..., min_length2, max_length200) thread_id: str | None Field(defaultNone, max_length64) router.post(/agent/run, response_modelRunResponse) async def run_agent(req: RunRequest): try: result await agent_service.run(req.topic, req.thread_id) return { thread_id: result[thread_id], report: result[report], } except Exception as exc: logger.error(agent run failed: %s, exc, exc_infoTrue) raise HTTPException(status_code500, detailAgent 执行失败请稍后重试)注意我把thread_id定义在请求里而不是依赖全局会话中间件。这是刻意的选择——多智能体服务的会话状态和 Web 登录态是两回事用户可能在一个页面同时开多个不同主题的研究任务各自要有独立的轨迹。4.3 SSE 流式输出让多智能体说话变实时多智能体跑一轮通常需要几十秒甚至更久前端不可能干等。我用 SSE 做流式输出StreamingResponse是 FastAPI 原生支持的方式配合astream事件逐段返回router.post(/agent/stream) async def stream_agent(req: RunRequest): config {configurable: {thread_id: req.thread_id or uuid4().hex}} async def event_gen(): # 先发送开始事件 yield fdata: {json.dumps({event: start, thread_id: config[configurable][thread_id]}, ensure_asciiFalse)}\n\n try: async for event in agent_service.graph.astream( {topic: req.topic}, configconfig, ): payload { event: node_update, node: getattr(event, node, unknown), data: event, } yield fdata: {json.dumps(payload, ensure_asciiFalse, defaultstr)}\n\n yield fdata: {json.dumps({event: done}, ensure_asciiFalse)}\n\n except Exception as exc: logger.error(stream error: %s, exc, exc_infoTrue) yield fdata: {json.dumps({event: error, message: Agent 执行异常}, ensure_asciiFalse)}\n\n return StreamingResponse(event_gen(), media_typetext/event-stream)这里有个细节json.dumps需要传defaultstr因为 astream 抛出来的事件对象里有大量自定义类型比如 LangGraph 的 message 对象直接序列化会炸。我在这上面栽过一次返回给前端的 JSON 突然变成TypeError: Object of type Message is not JSON serializable。后来统一用defaultstr保障序列化不崩。SSE 还有个容易被忽略的点连接会因为代理服务器或前端超时而中断。我在前端做了自动重连服务端这边给 SSE 响应加了Cache-Control: no-cache和Connection: keep-alive的 header实测稳定很多。4.4 会话状态管理thread_id 与检查点多轮对话场景下用户可能基于同一份研究报告追问再补充一点数据这时候图需要能接着上次的状态继续跑。LangGraph 提供了 Thread 的概念核心是检查点机制每次节点执行完状态被持久化带上同一个thread_id再调用图时可以从上次结束的位置继续。我的做法是thread_id 由前端生成每次请求带上。服务端只把它透传给 LangGraph 配置。为了让状态真正持久化需要给图配置一个 Checkpointer。最简单的方案是内存检查点from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() compiled_graph graph.compile(checkpointercheckpointer)但注意MemorySaver 是内存级的服务重启状态就丢了。如果生产环境要求会话跨进程、跨重启保持需要换成 SQLite 或 Postgres 检查点存储。我项目里用的 SQLite文件持久化单机够用不会因为进程重启丢掉所有会话。5. 接上本地模型Ollama 调用与并发性能细节5.1 Ollama 客户端封装思路这个项目的模型服务用的是 Ollama部署简单、本地运行、支持市面上主流的开源模型。封装思路和直接调 OpenAI SDK 不太一样Ollama 暴露的是一个 HTTP API我用 httpx 做了异步客户端class OllamaClient: def __init__(self, base_url: str, timeout: int 120): self.base_url base_url.rstrip(/) self.timeout timeout async def generate(self, model: str, prompt: str, **kwargs) - str: payload { model: model, prompt: prompt, stream: False, **kwargs, } async with httpx.AsyncClient(timeoutself.timeout) as client: resp await client.post(f{self.base_url}/api/generate, jsonpayload) resp.raise_for_status() data resp.json() return data[response] ollama_client OllamaClient(settings.ollama_base_url, settings.ollama_timeout)封装成单例的好处是所有 Agent 共享同一个客户端超时、重试逻辑只写一遍。这里我建议重点关注超时时间——本地模型在长上下文场景下生成速度可能很慢默认的 30 秒超时经常不够。我把超时设成了 120 秒实测跑一个 14B 模型的复杂任务偶尔还会到 90 秒以上。5.2 异步改造与性能参数设置多智能体流程里每个 Agent 都要调一次模型串行跑下来性能很吃亏。我用 LangGraph 的异步能力把节点函数都改成了 async同时需要注意如果某个节点的内部逻辑是同步阻塞的比如加载本地文件记得用await asyncio.to_thread包装否则会在事件循环里卡住。并发参数上我用的是 uvicorn 单进程 异步执行的方案。启动命令uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 1有人可能下意识想加--workers 4提升吞吐但对多智能体服务要冷静。因为 LangGraph 的内存检查点是在单进程内存里的多 worker 会让状态不共享——同一用户的 thread_id 在不同 worker 上查到不同的状态这是灾难性的。如果非要扩并发我会先升级为 SQLite/Postgres 检查点再逐步加 worker。单机小规模场景单 worker 异步已经能扛住不少并发。5.3 并发冲突最容易翻车的地方这里必须单独提一个坑。当多个请求同时跑同一个图的实例时LangGraph 的并发安全取决于检查点的隔离性。用 MemorySaver 时每个 thread_id 的状态是隔离的但同一个 thread_id 被并发调用就有问题了——两个请求同时写同一份状态后写的覆盖先写的数据丢失。我在项目中踩到的场景是前端用户连点了两次研究按钮两个请求带着同一个 thread_id 同时打进来结果第二次请求覆盖了第一次的状态返回的报告变成了一团乱码。解决办法有两个层面前端层面同一个 thread_id 的请求做互斥按钮加 loading 状态服务端层面给同一个 thread_id 的并发请求加锁或者干脆每次新开任务强制生成新 thread_id。我最终选择了折中方案服务端校验如果当前 thread_id 还在执行中直接返回 409 冲突让前端提示当前任务还在跑。6. 实战中踩过的坑日志、打包、状态隔离6.1 uvicorn 日志丢失问题的完整排查链路这个坑我是真没想到会坑这么久。项目跑起来后发现一个现象Agent 执行过程中打的一些关键日志有时候能看见有时候完全消失而且没有规律。一开始我怀疑是自己 logging 配置写错了反复检查 format、level 都没问题。后来我才意识到问题出在 uvicorn 的日志架构上。uvicorn 默认使用它自己的日志配置如果你在代码里调了logging.basicConfig很可能被 uvicorn 的--log-config覆盖掉。尤其当使用了uvicorn.run(app, ...)的方式启动加载配件的顺序会让你的基本配置失效日志直接不进文件。我的排查链路是先确认日志是否进了 stdout——发现部分日志丢不是全部丢查看是否有多个 handler 在抢同一个 logger——没有看 uvicorn 启动日志配置发现 uvicorn 加载了默认配置我的 basicConfig 被覆盖在logging.yml或启动参数里显式指定自己的 logger 配置问题解决。最终的解法是给 uvicorn 传--log-config logging.yml核心配置简化如下handlers: default: class: logging.FileHandler filename: ./logs/app.log formatter: default loggers: uvicorn: handlers: [default] level: INFO这样所有 uvicorn 和代码里的日志都统一进文件不再出现只打一半的问题。6.2 FastAPI 在 Windows 下打包的补充方案如果你是 Windows 环境开发、要打包给同事用会遇到一个 FastAPI 项目常见的麻烦相对路径、uvicorn 子进程、模型路径都容易出问题。我的处理方式是用--app-dir参数显式指定应用目录不依赖 cwd 位置把所有配置项读取改成Path(__file__).parent / ...这种基于代码文件位置的写法避免打包后运行目录变化导致找不到文件静态资源路径统一走 settings 配置不在代码里写死。如果你是打包成 exe 或服务给 Windows 机器用还有一个细节Putty 里跑 uvicorn 日志中文会乱码Windows 控制台编码是 GBK。我后来在日志输出端统一加了 UTF-8 转码并且打包程序时明确设置PYTHONIOENCODINGutf-8乱码才消失。6.3 LangGraph 并发状态隔离问题的复现与修复前面提过并发同一个 thread_id 的问题。我完整复现过一次两个并行请求同时调用graph.astream传相同的 thread_id结果最后一条消息把之前对话历史的累积内容覆盖了相当于失忆。去查了 LangGraph 的检查点机制发现它的状态存储是按 thread_id 的最后一个状态快照为准并发写同一个 key 时没有合并逻辑。我的修复方案是三层服务端维护一个thread_id - 执行状态的字典执行中标记为 running新请求进来直接 409不同任务强制生成新 thread_id同一研究任务的多轮追问复用同一个 thread_id但不允许并行图层面增加一个重试函数如果检查点读取失败会自动重新初始化状态。这三层下去之后再也没有复现过状态覆盖的问题。这件事给我的启示是多智能体的状态安全不是一个框架功能而是你自己的并发设计。6.4 一次完整的验证流程项目上线前我设计了一套验证流程专门用来验证多智能体编排和 API 层是否真的稳先跑图级单元测试不经过 HTTP直接调compiled_graph.ainvoke验证各节点字段正确、路由判断正确再跑接口测试用TestClient调/agent/run验证状态码、返回结构、异常处理最后跑流式测试故意让某个 Agent 超时验证 SSE 的错误事件能正常推给前端性能上做一个简单的压测20 个并发请求每个请求不同 thread_id观察内存和日志是否稳定。这套流程让我在正式部署前就发现了一个阶段性问题当 Ollama 服务端负载过高时某个节点会抛出 ConnectionError而我的图没有进行节点级重试。给路由函数套了一层重试装饰器之后系统稳定性明显提升。我在实际项目中的体会是多智能体服务真正难的不是模型效果而是状态管理和并发设计。FastAPI 负责把复杂流程包装成简洁接口LangGraph 负责让流程变得可维护、可回退、可恢复Ollama 则让整个系统可以完全跑在本地。三者配合得当团队里任何一个后端工程师接手这个项目都能快速看懂流程、调试异常、扩展新 Agent。最后再分享一个小技巧你可以给图的每个节点名加上环境前缀比如dev_collect、prod_collect这样在日志和 SSE 事件里一眼能看出当前跑的代码环境排查线上问题时很省事。多智能体的世界里可观测性就是生产力。以上经验希望能给准备做类似项目的你提供一些参考。