多智能体系统这两年从论文里走出来，落到实际项目里的速度比很多人预想的要快。但真到动手阶段，一个很现实的问题摆在面前：后端用 Python 写 Agent 编排逻辑，前端用什么？Web 页面在移动端体验拉胯，原生开发又要维护两套代码。Flutter 加 LLM 这个组合，恰好卡在了一个很舒服的位置上——一套代码跑通 iOS、Android、Web、桌面，后端用 FastAPI 把大模型能力和多智能体协同逻辑封装成接口，前端只管渲染和交互。这套架构我前后搭过几次，踩过的坑不算少，今天就把从零到跑通的完整链路拆开讲一遍，包括项目结构怎么设计、Agent 之间怎么通信、Flutter 这边状态管理怎么选、流式输出怎么接、以及那些文档里不会写的报错怎么处理。
这篇文章适合两类人：一是已经会一点 Flutter，想往 AI 应用方向靠的移动端开发者；二是后端熟悉 Python 和 FastAPI，但前端想快速出活的 AI 工程师。如果你两边都只懂个皮毛也没关系，我会把关键原理和操作步骤都铺开讲，代码可以直接抄。
1. 为什么是 Flutter 加 FastAPI 这套组合
1.1 多智能体应用对前端提出的真实需求
先想清楚一件事：多智能体应用的前端到底要干什么。它和普通的 CRUD 应用有本质区别。普通应用是"用户点一下，后端返回一个确定结果"，而多智能体应用是"用户丢一个任务进去，多个 Agent 轮流思考、调用工具、互相传递中间结果，最后汇总输出"。这意味着前端要处理的是持续变化的、非结构化的、可能中途出错重试的数据流。
具体来说，前端至少要满足这几个能力：第一，能接收流式文本，因为大模型的输出是一个 token 一个 token 吐出来的，如果等全部生成完再显示，用户会以为程序卡死了；第二，能展示多个 Agent 的并行状态，比如"检索 Agent 正在工作""分析 Agent 已完成""汇总 Agent 排队中"；第三，能处理长会话的状态保持，用户切到别的页面再回来，对话历史不能丢；第四，要能跨平台，因为这类工具往往既要在手机上随手用，又要在电脑上做复杂操作。
Flutter 在这几点上表现都不错。它的StreamBuilder天然适合处理流式数据，StatefulWidget配合状态管理库能很好地维护会话状态，一套代码编译到六个平台。相比之下，如果你用 React Native，流式渲染的性能在低端安卓机上会明显吃力；如果用原生，开发成本直接翻倍。
1.2 FastAPI 作为 Agent 后端的天然优势
后端选 FastAPI 不是跟风。多智能体系统的核心是异步编排——一个任务可能要同时调用三四个 Agent，每个 Agent 又要调外部工具、查数据库、请求大模型接口，这些都是 IO 密集型操作。FastAPI 基于 asyncio，天生支持async/await，一个 worker 就能扛住大量并发请求，不用像 Flask 那样靠多进程硬堆。
另一个关键点是SSE（Server-Sent Events）。大模型的流式输出，业界标准做法就是用 SSE 把 token 逐个推给前端。FastAPI 的StreamingResponse配合异步生成器，几行代码就能实现，比 WebSocket 轻量得多，而且浏览器和移动端都原生支持。我实测下来，用 SSE 做流式输出，在 4G 网络下首字延迟能控制在 300ms 以内，体验相当顺滑。
还有一点容易被忽略：FastAPI 自动生成的 OpenAPI 文档。多智能体系统的接口往往很复杂，参数多、返回结构嵌套深，有了自动文档，前端对接时不用反复问后端"这个字段到底叫啥"，直接看/docs就行，省下大量沟通成本。
1.3 这套架构的边界在哪里
得说清楚这套组合不适合什么场景。如果你的 Agent 逻辑极其复杂，涉及大量本地计算（比如跑本地大模型推理），那 Flutter 端只适合做展示层，重活必须放后端。如果要做实时性要求极高的场景（比如毫秒级响应的游戏 AI），SSE 的延迟可能不够，得考虑 WebSocket 甚至 gRPC。另外，Flutter 的包体积比原生大，如果对安装包大小极度敏感，需要做裁剪。
但对于绝大多数"任务型多智能体应用"——比如智能客服、文档分析助手、自动化工作流工具——这套架构的性价比是最高的。开发速度快、维护成本低、跨平台体验一致，这三点在实际项目里比什么花哨的技术都重要。
2. 后端骨架：FastAPI 里怎么把多智能体跑起来
2.1 项目目录结构的设计逻辑
很多人写 FastAPI 项目，所有代码堆在一个main.py里，几百行下去自己都找不到东西。多智能体系统的代码量只会更大，目录结构必须一开始就设计好。我常用的结构是这样的：
backend/ ├── app/ │ ├── main.py # 应用入口，注册路由和中间件 │ ├── config.py # 配置管理，读环境变量 │ ├── api/ │ │ ├── __init__.py │ │ ├── chat.py # 对话相关接口 │ │ └── agent.py # Agent 管理接口 │ ├── agents/ │ │ ├── base.py # Agent 基类，定义统一接口 │ │ ├── retriever.py # 检索 Agent │ │ ├── analyzer.py # 分析 Agent │ │ └── summarizer.py # 汇总 Agent │ ├── core/ │ │ ├── orchestrator.py # 编排器，负责调度多个 Agent │ │ └── memory.py # 会话记忆管理 │ ├── models/ │ │ └── schemas.py # Pydantic 数据模型 │ └── utils/ │ └── llm_client.py # 大模型调用封装 ├── requirements.txt └── .env这个结构的关键在于职责分离。agents/目录下每个 Agent 是一个独立模块，只关心自己的输入输出，不关心谁调用它。core/orchestrator.py是大脑，负责决定哪个任务交给哪个 Agent、按什么顺序执行、结果怎么合并。api/层只做参数校验和响应封装，不写业务逻辑。这样设计的好处是，加一个新 Agent 只需要在agents/下新建文件，在编排器里注册一下，其他代码完全不用动。
2.2 Agent 基类：统一接口是协同的前提
多智能体协同的前提是每个 Agent 长得一样，编排器才能用统一的方式调用它们。所以先定义一个基类：
from abc import ABC, abstractmethod from typing import AsyncGenerator, Any class BaseAgent(ABC): def __init__(self, name: str, description: str): self.name = name self.description = description @abstractmethod async def run(self, task: str, context: dict) -> AsyncGenerator[str, None]: """执行任务，流式返回结果""" pass @abstractmethod def can_handle(self, task: str) -> bool: """判断这个 Agent 是否能处理该任务""" pass这里有两个设计决策值得说。第一，run方法返回的是AsyncGenerator，也就是流式输出。为什么不让它返回一个完整字符串？因为多智能体系统里，一个 Agent 的输出往往是下一个 Agent 的输入，如果等全部生成完再传递，整个链路的延迟会累加得很难看。流式传递能让下游 Agent 尽早开始工作。第二，can_handle方法让 Agent 自己判断能不能接活，编排器只需要遍历所有 Agent 问一遍，符合条件的就是候选者。这比在编排器里写一堆if-else判断任务类型要优雅得多，也更容易扩展。
2.3 编排器：多智能体协同的核心逻辑
编排器是整个系统最烧脑的部分。它要解决三个问题：任务怎么分、Agent 怎么选、结果怎么合。我一般用"规划-执行-汇总"三段式：
class Orchestrator: def __init__(self, agents: list[BaseAgent], llm_client): self.agents = agents self.llm = llm_client async def execute(self, user_input: str, session_id: str): # 第一步：规划，让大模型拆解任务 plan = await self._plan(user_input) # 第二步：执行，按计划调度 Agent results = [] for step in plan: agent = self._select_agent(step) if agent: async for chunk in agent.run(step, {"history": results}): yield f"data: {chunk}\n\n" results.append(chunk) # 第三步：汇总 final = await self._summarize(user_input, results) yield f"data: {final}\n\n"_plan方法会调用大模型，把用户的自然语言输入拆成结构化的步骤列表。比如用户说"帮我分析这份财报并给出投资建议"，规划结果可能是["提取财报关键数据", "计算财务比率", "对比行业均值", "生成投资建议"]。然后编排器逐步执行，每一步选一个最合适的 Agent。
这里有个实战经验：规划步骤不要超过 5 步。我试过让大模型拆出十几步的详细计划，结果执行到一半就乱了，因为中间步骤的输出质量会累积误差，越往后越离谱。控制在 3 到 5 步，每步的职责足够清晰，整体成功率最高。
2.4 SSE 流式接口的实现细节
把编排器的输出推给前端，用 FastAPI 的StreamingResponse：
from fastapi import APIRouter from fastapi.responses import StreamingResponse router = APIRouter() @router.post("/chat/stream") async def chat_stream(request: ChatRequest): async def event_generator(): async for chunk in orchestrator.execute(request.message, request.session_id): yield chunk return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", } )X-Accel-Buffering: no这个头很关键。如果你用 Nginx 做反向代理，默认它会缓冲响应，导致流式输出变成"憋一大坨再吐出来"，完全失去流式的意义。加上这个头告诉 Nginx 别缓冲。这个坑我踩过一次，本地测试好好的，一上生产就变成一次性输出，排查了半天才发现是 Nginx 的锅。
另外，SSE 的数据格式必须是data: xxx\n\n，两个换行符是消息结束标志，少一个前端就收不到。如果传输的是 JSON，记得先json.dumps再拼进去，不然前端解析会报错。
3. Flutter 端：从建项目到接住流式数据
3.1 创建项目和依赖选型
创建 Flutter 项目用命令行最干净：
flutter create --org com.example --project-name multi_agent_app multi_agent_app cd multi_agent_app--org参数决定包名，上架应用商店前一定要改，不然就是默认的com.example，审核会被打回。这个细节很多人第一次上架时才发现，返工很麻烦。
依赖方面，核心就几个：
dependencies: flutter: sdk: flutter http: ^1.1.0 # 基础网络请求 provider: ^6.0.5 # 状态管理 flutter_markdown: ^0.6.18 # 渲染大模型返回的 Markdown shared_preferences: ^2.2.0 # 本地存会话 ID状态管理我选 Provider 而不是 Riverpod 或 Bloc，原因很简单：这个场景的状态逻辑不复杂，主要是"会话列表"和"当前对话流"两块，Provider 够用且上手快。如果你的项目要长期维护、团队多人协作，可以考虑 Riverpod，但对入门项目来说 Provider 的性价比最高。
flutter_markdown这个包值得单独说。大模型返回的内容经常带 Markdown 格式（标题、列表、代码块），如果直接用Text组件渲染，会显示成一堆##和**，体验很差。用flutter_markdown能正确渲染，代码块还能高亮。
3.2 用 StreamBuilder 接住 SSE 数据
Flutter 端接收 SSE 有两种方式：用http包的StreamedResponse，或者用专门的flutter_client_sse包。我倾向用http包自己处理，因为可控性更强，出问题好排查：
Stream<String> streamChat(String message, String sessionId) async* { final request = http.Request( 'POST', Uri.parse('$baseUrl/chat/stream'), ); request.headers['Content-Type'] = 'application/json'; request.body = jsonEncode({ 'message': message, 'session_id': sessionId, }); final response = await http.Client().send(request); await for (final chunk in response.stream.transform(utf8.decoder)) { // 按 SSE 格式解析 for (final line in chunk.split('\n')) { if (line.startsWith('data: ')) { yield line.substring(6); } } } }这里有个坑：SSE 的数据块不保证按行完整到达。网络传输时，一个data: xxx\n\n可能被拆成两个 TCP 包，第一个包只有data: xx，第二个包是x\n\n。如果直接按\n分割，会解析出半截数据。稳妥的做法是维护一个缓冲区，每次收到数据先追加到缓冲区，然后按\n\n分割出完整消息，剩下的留在缓冲区等下次。这个处理逻辑稍微麻烦点，但不做的话，长文本输出时会出现乱码或丢字。
在 UI 层用StreamBuilder消费这个流：
StreamBuilder<String>( stream: chatStream(message, sessionId), builder: (context, snapshot) { if (snapshot.hasError) { return Text('出错了: ${snapshot.error}'); } if (!snapshot.hasData) { return const CircularProgressIndicator(); } return MarkdownBody(data: snapshot.data!); }, )StreamBuilder会在每次流里有新数据时自动重建，所以文字会一个字一个字地冒出来，效果和 ChatGPT 一样。
3.3 会话状态保持：Navigator 切换页面不丢数据
热词里有个问题很典型："flutter navigator 切换页面后会丢失状态吗"。答案是默认会丢。Flutter 的Navigator.push会把当前页面压在栈里，但如果你用Navigator.pushReplacement或者页面被系统回收，状态就没了。
多智能体应用里，用户很可能在对话进行到一半时切到设置页改个参数再切回来，如果对话历史丢了，体验直接崩。解决方案是把会话状态提到页面之上，用 Provider 管理：
class ChatProvider extends ChangeNotifier { final List<Message> _messages = []; List<Message> get messages => _messages; void addMessage(Message msg) { _messages.add(msg); notifyListeners(); } }然后在main.dart里用ChangeNotifierProvider包住整个 App，这样无论怎么切页面，ChatProvider实例都在，数据不会丢。这个思路和 React 里把状态提到 Redux 是一个道理——状态的生命周期要长于组件的生命周期。
3.4 处理 Flutter 常见报错
热词里出现了e/flutter (31173): [error:flutter/runtime/dart_vm_initializer.cc(41)] unhand，这是 Flutter 运行时未捕获异常的典型报错。多数情况下是异步代码里抛了异常但没try-catch。在多智能体场景里，最常见的触发点是网络请求失败或 JSON 解析失败。
处理原则是：所有涉及网络和解析的异步操作都要包 try-catch，并且把错误转成用户能看懂的话。比如：
try { final data = jsonDecode(response.body); // ... } on FormatException catch (e) { throw Exception('服务器返回格式异常，请稍后重试'); } on SocketException catch (e) { throw Exception('网络连接失败，请检查网络'); }不要直接把原始异常抛给用户看，FormatException: Unexpected character这种信息对用户毫无意义。
另一个热词是you are applying flutter's main gradle plugin imperatively using the apply s，这是 Gradle 配置的警告，说明你在build.gradle里用了旧的apply plugin写法。新版 Flutter 推荐用plugins {}块声明插件。虽然只是警告不影响运行，但建议改掉，不然升级 Flutter 版本时可能直接报错。
4. 多智能体协同的实战细节
4.1 Agent 之间怎么传递上下文
多智能体系统最容易出问题的地方就是上下文传递。假设检索 Agent 找到了一段资料，分析 Agent 要用这段资料做分析，那检索结果怎么传给分析 Agent？
最直接的做法是把前一个 Agent 的输出拼进后一个 Agent 的 prompt 里。但这里有个陷阱：上下文会越滚越大。三个 Agent 串行下来，最后一个 Agent 的输入可能已经几千 token 了，既慢又贵，还容易超出模型上下文窗口。
我的做法是给每个 Agent 的输出做摘要压缩。检索 Agent 返回原始资料后，先让大模型压缩成 200 字以内的要点，再传给下游。这样上下文长度可控，信息密度也更高。实测下来，压缩后的信息对下游任务的影响很小，但 token 消耗能降 60% 以上。
另一个技巧是给上下文加结构化标签。不要直接把文本拼在一起，而是用明确的标记：
[检索结果] {retrieved_content} [分析要求] 基于上述检索结果，分析...这样大模型能清楚区分哪部分是资料、哪部分是指令，输出质量明显更稳定。
4.2 并发控制：多个 Agent 同时跑怎么不出乱子
有些任务可以并行，比如同时让三个 Agent 从不同角度分析同一份数据。用asyncio.gather就能并发执行：
results = await asyncio.gather( agent_a.run(task), agent_b.run(task), agent_c.run(task), return_exceptions=True, )return_exceptions=True这个参数很重要。默认情况下，gather里任何一个任务抛异常，整个gather就失败了，其他任务的结果也拿不到。加上这个参数后，异常会作为结果返回，你可以单独处理失败的任务，成功的任务结果照常用。在多智能体场景里，一个 Agent 失败不应该拖垮整个流程。
但并发也不是越多越好。如果同时调用大模型接口，很容易触发速率限制（rate limit）。我的经验是并发数控制在 3 到 5 个，超过这个数就要加信号量（asyncio.Semaphore）限流。另外，如果多个 Agent 要写同一个共享资源（比如同一个会话的记忆），必须加锁，否则会出现竞态条件。
4.3 记忆管理：让 Agent 记住之前说过什么
多轮对话的核心是记忆。最简单的做法是把所有历史消息都塞进 prompt，但这样 token 消耗会线性增长，聊到第 20 轮时成本已经很高了。
我一般用滑动窗口加摘要的方案：保留最近 5 轮完整对话，更早的对话用大模型压缩成一段摘要。这样既保留了近期细节，又不丢失长期上下文。实现上，每次对话结束后检查历史长度，超过阈值就触发摘要：
async def manage_memory(session_id: str, new_message: Message): history = await get_history(session_id) history.append(new_message) if len(history) > 10: old = history[:-5] summary = await llm.summarize(old) history = [Message(role="system", content=f"历史摘要: {summary}")] + history[-5:] await save_history(session_id, history)这个方案在实测中效果不错，聊 50 轮以上也不会明显变慢或变贵。
4.4 错误处理与重试策略
大模型接口不稳定是常态，超时、限流、返回格式错误都可能发生。多智能体系统里，一个环节失败可能导致整个链路崩溃，所以必须有重试机制。
重试要遵循指数退避原则：第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒。不要失败后立刻重试，那样只会加重服务端压力，还容易被判定为攻击。用tenacity库可以很方便地实现：
from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), ) async def call_llm(prompt: str): return await llm_client.chat(prompt)但要注意，不是所有错误都值得重试。如果是参数错误（比如 prompt 超长），重试多少次都一样，应该直接失败并提示用户。只有网络超时、限流这类临时性错误才重试。判断方法是看异常类型，TimeoutError和RateLimitError重试，ValidationError直接抛。
5. 从跑通到好用：性能与体验优化
5.1 首字延迟优化
用户对 AI 应用的第一印象就是"多久能看到第一个字"。如果超过 2 秒还没反应，用户就会觉得卡。优化首字延迟有几个方向：
第一，后端尽早开始流式输出。不要在编排器里等所有规划完成再返回，规划阶段就可以先推一个"正在思考..."的占位消息，让前端立刻有反馈。第二，前端做乐观更新，用户发送消息后立刻在界面上显示用户气泡和一个加载中的 AI 气泡，不用等后端响应。第三，预热连接，App 启动时就建立好后端连接，避免首次请求的握手开销。
我实测过，做好这三点后，首字延迟能从 1.5 秒降到 400ms 左右，体感差异非常明显。
5.2 长文本渲染的性能问题
大模型一次输出几千字很常见，如果每来一个 token 就重建整个MarkdownBody，低端机上会明显卡顿。优化方法是节流渲染：不要每个 token 都触发重建，而是攒够一定量（比如 20 个字符）或者每隔 50ms 才更新一次 UI。
Timer? _throttleTimer; String _buffer = ''; void onChunk(String chunk) { _buffer += chunk; _throttleTimer?.cancel(); _throttleTimer = Timer(const Duration(milliseconds: 50), () { setState(() { _displayText = _buffer; }); }); }这样渲染频率从每秒几十次降到每秒 20 次，流畅度提升明显，而用户几乎感觉不到延迟。
5.3 打包和部署的注意事项
Flutter 打包 Android 时，记得在build.gradle里配置签名和混淆规则。如果用了flutter_markdown这类依赖反射的包，混淆时要加 keep 规则，否则 release 包会崩溃而 debug 包正常，这种问题最难排查。
后端部署方面，FastAPI 用uvicorn启动时，workers数量建议设为 CPU 核心数的 2 倍加 1。但注意，如果你用了内存态的会话存储（比如把历史存在进程内存里），多 worker 会导致会话数据不一致——用户第一次请求打到 worker 1，第二次打到 worker 2，就找不到历史了。这种情况必须把会话存到 Redis 或数据库里，做成无状态服务。
跨域问题也别忘了。Flutter Web 端会受浏览器同源策略限制，FastAPI 要加CORSMiddleware：
from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境要改成具体域名 allow_methods=["*"], allow_headers=["*"], )allow_origins在生产环境千万别用*，要写具体的前端域名，不然有安全风险。
6. 几个我踩过的坑和对应的解法
6.1 SSE 连接被中间层掐断
现象是流式输出到一半突然断了，前端报连接错误。排查下来通常是两个原因：一是 Nginx 的proxy_read_timeout默认 60 秒，超过就断；二是某些云厂商的负载均衡对长连接有超时限制。解法是把 Nginx 的超时调大，同时在应用层加心跳——每隔 15 秒推一个空注释: heartbeat\n\n，保持连接活跃。
6.2 中文乱码
Flutter 端解析 SSE 时如果没指定 UTF-8，中文会变成乱码。http包的response.stream默认是字节流，必须用utf8.decoder转换。这个坑在纯英文测试时发现不了，一上中文就暴露。
6.3 Agent 死循环
编排器如果设计不当，可能出现 Agent A 调用 Agent B，B 又回头调用 A 的情况，无限循环。防御措施是给每个任务加一个最大步数限制，比如 10 步，超过就强制终止并返回当前结果。同时记录调用链，检测到重复调用同一个 Agent 处理同一个子任务时直接跳过。
6.4 大模型返回格式不稳定
你要求大模型返回 JSON，它有时候会返回带 Markdown 代码块包裹的 JSON，有时候会多写一句"好的，这是结果："。解析时要做容错：先用正则提取出 JSON 部分，再解析；解析失败就重试一次，并在 prompt 里强调"只返回 JSON，不要任何其他文字"。
7. 后续可以往哪些方向扩展
跑通基础版本后，有几个方向值得继续深挖。一是工具调用，让 Agent 能真正"下地干活"，比如调用搜索接口、读写文件、执行代码，这需要设计一套工具注册和调用的机制。二是多模态，让 Agent 能处理图片和语音，Flutter 端接摄像头和麦克风，后端接多模态大模型。三是可视化编排，做一个拖拽界面让用户自己定义 Agent 之间的协作流程，这个对非技术用户很有吸引力。
我个人在实际操作中的体会是，多智能体系统的难点从来不在单个 Agent 有多聪明，而在于协同的可靠性。一个 Agent 偶尔出错没关系，但如果错误在链路里被放大，整个系统就不可用了。所以与其花时间调 prompt 让单个 Agent 更聪明，不如把精力放在错误处理、重试、降级这些工程细节上。这些地方做扎实了，系统的实际可用性会有质的提升。
最后分享一个小技巧：开发阶段一定要把每个 Agent 的输入输出完整打日志，包括耗时和 token 消耗。多智能体系统的调试如果只看最终结果，根本不知道是哪一环出了问题。有了详细日志，定位问题的速度能快好几倍。
