Flutter 与 FastAPI 构建多智能体应用：流式输出与状态管理实战-北京尧图网络

多智能体系统这两年从论文里走出来落到实际项目里的速度比很多人预想的要快。但真到动手阶段一个很现实的问题摆在面前后端用 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 那样靠多进程硬堆。另一个关键点是SSEServer-Sent Events。大模型的流式输出业界标准做法就是用 SSE 把 token 逐个推给前端。FastAPI 的StreamingResponse配合异步生成器几行代码就能实现比 WebSocket 轻量得多而且浏览器和移动端都原生支持。我实测下来用 SSE 做流式输出在 4G 网络下首字延迟能控制在 300ms 以内体验相当顺滑。还有一点容易被忽略FastAPI 自动生成的 OpenAPI 文档。多智能体系统的接口往往很复杂参数多、返回结构嵌套深有了自动文档前端对接时不用反复问后端这个字段到底叫啥直接看/docs就行省下大量沟通成本。1.3 这套架构的边界在哪里得说清楚这套组合不适合什么场景。如果你的 Agent 逻辑极其复杂涉及大量本地计算比如跑本地大模型推理那 Flutter 端只适合做展示层重活必须放后端。如果要做实时性要求极高的场景比如毫秒级响应的游戏 AISSE 的延迟可能不够得考虑 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 fdata: {chunk}\n\n results.append(chunk) # 第三步汇总 final await self._summarize(user_input, results) yield fdata: {final}\n\n_plan方法会调用大模型把用户的自然语言输入拆成结构化的步骤列表。比如用户说帮我分析这份财报并给出投资建议规划结果可能是[提取财报关键数据, 计算财务比率, 对比行业均值, 生成投资建议]。然后编排器逐步执行每一步选一个最合适的 Agent。这里有个实战经验规划步骤不要超过 5 步。我试过让大模型拆出十几步的详细计划结果执行到一半就乱了因为中间步骤的输出质量会累积误差越往后越离谱。控制在 3 到 5 步每步的职责足够清晰整体成功率最高。2.4 SSE 流式接口的实现细节把编排器的输出推给前端用 FastAPI 的StreamingResponsefrom 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_typetext/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包自己处理因为可控性更强出问题好排查StreamString 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消费这个流StreamBuilderString( 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 ListMessage _messages []; ListMessage 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 flutters 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_exceptionsTrue, )return_exceptionsTrue这个参数很重要。默认情况下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(rolesystem, contentf历史摘要: {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( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10), ) 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 要加CORSMiddlewarefrom 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 BB 又回头调用 A 的情况无限循环。防御措施是给每个任务加一个最大步数限制比如 10 步超过就强制终止并返回当前结果。同时记录调用链检测到重复调用同一个 Agent 处理同一个子任务时直接跳过。6.4 大模型返回格式不稳定你要求大模型返回 JSON它有时候会返回带 Markdown 代码块包裹的 JSON有时候会多写一句好的这是结果。解析时要做容错先用正则提取出 JSON 部分再解析解析失败就重试一次并在 prompt 里强调只返回 JSON不要任何其他文字。7. 后续可以往哪些方向扩展跑通基础版本后有几个方向值得继续深挖。一是工具调用让 Agent 能真正下地干活比如调用搜索接口、读写文件、执行代码这需要设计一套工具注册和调用的机制。二是多模态让 Agent 能处理图片和语音Flutter 端接摄像头和麦克风后端接多模态大模型。三是可视化编排做一个拖拽界面让用户自己定义 Agent 之间的协作流程这个对非技术用户很有吸引力。我个人在实际操作中的体会是多智能体系统的难点从来不在单个 Agent 有多聪明而在于协同的可靠性。一个 Agent 偶尔出错没关系但如果错误在链路里被放大整个系统就不可用了。所以与其花时间调 prompt 让单个 Agent 更聪明不如把精力放在错误处理、重试、降级这些工程细节上。这些地方做扎实了系统的实际可用性会有质的提升。最后分享一个小技巧开发阶段一定要把每个 Agent 的输入输出完整打日志包括耗时和 token 消耗。多智能体系统的调试如果只看最终结果根本不知道是哪一环出了问题。有了详细日志定位问题的速度能快好几倍。

Flutter 与 FastAPI 构建多智能体应用：流式输出与状态管理实战

相关推荐

Sprite Atlas性能优化：从图集生成到GPU渲染的全链路实践

Android Studio Tab被顶掉与自动关闭？一文教你配置Editor Tabs

等保测评安全运维管理：环境、资产与流程要点解析

更多资讯

webpack打包纯静态商城站：多页面资源处理与工程化实践

思科华为组播配置与排错实战：从RPF原理到IGMP Snooping部署

Qt+CMake三大配置范式：从Demo到企业级项目的演进路径

吃透六自由度机器人D-H法：建模、正逆解与避坑全攻略

Python调用Simulink编译DLL：零MATLAB依赖的轻量部署方案

Oransim路线图前瞻：从v0.2到因果基础模型，闭环AI媒体投放与百万级智能体集群的完整规划

今日更新

防关联浏览器哪个好？三款产品实测差异解析

2026钢铁行业AI转型系统选型对比：多基地排产、大宗原燃料采购与质量追溯，六家厂商能力差异在哪里

影刀RPA深度实战：大模型Fine-tuning定制RPA专用Prompt与指令生成

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？