企业官网建设流程全解析

最近一段时间我一直在忙一个内部项目：把"研究分析"这条流程做成多智能体协作服务。需求本身不复杂——用户提交一个主题，系统自动派出多个智能体分工干活——一个去收集信息、一个做数据整理、一个负责生成带观点的分析报告，最终把结果通过 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 确实够用，我也用了很多年。但这次接的是多智能体服务，情况不太一样。

对比下来我的选择逻辑是这样：

多智能体服务有个明显特点：接口往往需要长时间占用连接，因为 Agent 要跑好几轮大模型调用，每次都是秒级延迟，用户端还希望流式看到进度。FastAPI 的异步机制加上 SSE（Server-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 = [ "fastapi>=0.115", "uvicorn[standard]>=0.30", "langgraph>=0.2", "langchain-core>=0.3", "httpx>=0.27", "pydantic-settings>=2.4", ]

配置方面，创建一个 .env.example 文件，把关键配置都放进去，避免配置写死在代码里：

OLLAMA_BASE_URL=http://localhost:11434 OLLAMA_MODEL=qwen2.5:14b OLLAMA_TIMEOUT=120 LOG_LEVEL=INFO

我的一个实际经验是：LangGraph 的版本兼容性问题比 FastAPI 更麻烦，尤其当你同时用 LangChain 的 message 工具和自定义检查点存储时。建议一开始就把版本锁住，升级的时候单独拉分支测，别直接升最新版。

3. 编排 Multi-Agent 协作：StateGraph 是灵魂

3.1 场景设计：研究分析流水线

进入核心部分。我为这个项目设计的场景是"主题研究报告生成"：用户给一个主题，系统自动产出结构化报告。整个过程切分成三个 Agent，每个 Agent 只负责一件事：

光看这一字排开的流程，似乎顺序调用就够了。但真实场景里有个关键点：collect 节点不一定一次就能收集到足够的数据。如果模型判断收集的信息太少、覆盖维度不够，就要重新收集一轮；analyze 节点分析完如果发现数据质量不行，还得回退让 collect 再补。这种动态流转，才是用图的真正价值。

3.2 状态定义与自定义 reducer

LangGraph 里所有节点共享一份状态。状态类型用 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( model=settings.ollama_model, prompt=prompt, temperature=0.3, ) # 解析模型输出，格式化为列表 items = parse_result(result) return { "collected_data": items, "current_step": "collect", }

这里有一个值得注意的点：节点的返回结果只会更新 state 里那一个字段，其他字段不受影响。所以每个 Agent 只需要关注自己负责的字段。collect 节点返回collected_data，analyze 节点返回analysis_result，write 节点返回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}, config=config, ) return result agent_service = AgentService(compiled_graph)

这里用asyncio.to_thread是因为我们的图内部虽然用了 async 节点，但有些兼容性场景下 invoke 是同步调用的。为了避免阻塞事件循环，我用 to_thread 把 invoke 放到线程池里跑。如果你的图是纯异步的，直接await self.graph.ainvoke(...)也行。

4.2 HTTP 路由设计与参数校验

路由层的设计以简单为主。我定义了两个端点：一个非流式返回完整报告，一个流式返回过程事件。请求参数用 Pydantic 模型声明，FastAPI 会自动完成校验，非法参数直接返回 422：

from pydantic import BaseModel, Field class RunRequest(BaseModel): topic: str = Field(..., min_length=2, max_length=200) thread_id: str | None = Field(default=None, max_length=64) @router.post("/agent/run", response_model=RunResponse) 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_info=True) raise HTTPException(status_code=500, detail="Agent 执行失败，请稍后重试")

注意我把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 f"data: {json.dumps({'event': 'start', 'thread_id': config['configurable']['thread_id']}, ensure_ascii=False)}\n\n" try: async for event in agent_service.graph.astream( {"topic": req.topic}, config=config, ): payload = { "event": "node_update", "node": getattr(event, "node", "unknown"), "data": event, } yield f"data: {json.dumps(payload, ensure_ascii=False, default=str)}\n\n" yield f"data: {json.dumps({'event': 'done'}, ensure_ascii=False)}\n\n" except Exception as exc: logger.error("stream error: %s", exc, exc_info=True) yield f"data: {json.dumps({'event': 'error', 'message': 'Agent 执行异常'}, ensure_ascii=False)}\n\n" return StreamingResponse(event_gen(), media_type="text/event-stream")

这里有个细节：json.dumps需要传default=str，因为 astream 抛出来的事件对象里有大量自定义类型（比如 LangGraph 的 message 对象），直接序列化会炸。我在这上面栽过一次，返回给前端的 JSON 突然变成TypeError: Object of type Message is not JSON serializable。后来统一用default=str保障序列化不崩。

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(checkpointer=checkpointer)

但注意：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(timeout=self.timeout) as client: resp = await client.post(f"{self.base_url}/api/generate", json=payload) 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, ...)的方式启动，加载配件的顺序会让你的基本配置失效，日志直接不进文件。

我的排查链路是：

1. 先确认日志是否进了 stdout——发现部分日志丢，不是全部丢；
2. 查看是否有多个 handler 在抢同一个 logger——没有；
3. 看 uvicorn 启动日志配置，发现 uvicorn 加载了默认配置，我的 basicConfig 被覆盖；
4. 在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 转码，并且打包程序时明确设置PYTHONIOENCODING=utf-8，乱码才消失。

6.3 LangGraph 并发状态隔离问题的复现与修复

前面提过并发同一个 thread_id 的问题。我完整复现过一次：两个并行请求同时调用graph.astream，传相同的 thread_id，结果最后一条消息把之前对话历史的累积内容覆盖了，相当于"失忆"。去查了 LangGraph 的检查点机制，发现它的状态存储是"按 thread_id 的最后一个状态快照"为准，并发写同一个 key 时没有合并逻辑。

我的修复方案是三层：

1. 服务端维护一个thread_id -> 执行状态的字典，执行中标记为 running，新请求进来直接 409；
2. 不同任务强制生成新 thread_id，同一研究任务的多轮追问复用同一个 thread_id，但不允许并行；
3. 图层面增加一个重试函数，如果检查点读取失败会自动重新初始化状态。

这三层下去之后，再也没有复现过状态覆盖的问题。这件事给我的启示是：多智能体的"状态安全"不是一个框架功能，而是你自己的并发设计。

6.4 一次完整的验证流程

项目上线前我设计了一套验证流程，专门用来验证多智能体编排和 API 层是否真的稳：

1. 先跑图级单元测试，不经过 HTTP，直接调compiled_graph.ainvoke，验证各节点字段正确、路由判断正确；
2. 再跑接口测试，用TestClient调/agent/run，验证状态码、返回结构、异常处理；
3. 最后跑流式测试，故意让某个 Agent 超时，验证 SSE 的错误事件能正常推给前端；
4. 性能上做一个简单的压测：20 个并发请求，每个请求不同 thread_id，观察内存和日志是否稳定。

这套流程让我在正式部署前就发现了一个阶段性问题：当 Ollama 服务端负载过高时，某个节点会抛出 ConnectionError，而我的图没有进行节点级重试。给路由函数套了一层重试装饰器之后，系统稳定性明显提升。

我在实际项目中的体会是：多智能体服务真正难的不是模型效果，而是状态管理和并发设计。FastAPI 负责把复杂流程包装成简洁接口，LangGraph 负责让流程变得可维护、可回退、可恢复，Ollama 则让整个系统可以完全跑在本地。三者配合得当，团队里任何一个后端工程师接手这个项目，都能快速看懂流程、调试异常、扩展新 Agent。

最后再分享一个小技巧：你可以给图的每个节点名加上环境前缀，比如dev_collect、prod_collect，这样在日志和 SSE 事件里一眼能看出当前跑的代码环境，排查线上问题时很省事。多智能体的世界里，可观测性就是生产力。以上经验希望能给准备做类似项目的你提供一些参考。