★ FEATURED ARTICLE 1. 为什么我选择 Flutter LLM 多智能体这条技术路线移动端做 AI 应用最容易踩的坑就是前端一套、后端一套、模型调用再一套三套东西各写各的最后联调的时候互相甩锅。我去年接手一个智能助手类项目最初用原生 Android 写界面、Python 写推理服务结果光是流式输出的状态同步就折腾了两周。后来换成 Flutter 做客户端、FastAPI 做智能体编排层整个链路才顺下来。这篇就聊聊 Flutter LLM 多智能体这套组合到底怎么落地适合谁看以及我踩过的那些坑。先说清楚这套东西是什么。Flutter 负责跨端 UI 和交互LLM 是底层大模型能力多智能体Multi-Agent是把一个复杂任务拆给多个各司其职的 Agent 去协作完成FastAPI 则是把这些 Agent 串起来、对外暴露接口的服务层。能做什么典型场景就是用户一句话丢进来系统自动判断该查资料、该算数、该调工具然后几个 Agent 分工干完把结果流式吐回手机屏幕。适合谁有 Flutter 基础想往 AI 方向走的移动端开发或者会 Python 想补前端展示层的后端同学两边都能上手。我特别想强调一点多智能体不是多个模型一起回答这么简单。它的核心价值在于职责隔离和可观测性。单个 Agent 什么都干prompt 会越写越长出错根本不知道是哪一步崩的拆成多个之后每个 Agent 的输入输出边界清晰调试和替换都容易得多。这也是我坚持用这套架构的根本原因。2. 整体架构设计与技术选型拆解2.1 三层架构客户端、编排层、模型层我把整个系统切成三层边界划得很死这样任何一层出问题都能快速定位。客户端层Flutter只负责 UI 渲染、用户输入、流式内容展示、本地状态管理。它不关心背后有几个 Agent只认一个统一的 SSE 或 WebSocket 接口。编排层FastAPI Agent 框架接收请求做意图识别调度多个 Agent 协作管理会话上下文最后把结果流式推回客户端。模型层LLM 服务可以是云端 API也可以是本地部署的推理服务对编排层暴露统一的调用接口。这么分的好处是客户端换框架不影响后端模型换供应商不影响前端。我实测下来这种解耦在迭代期能省掉大量返工。2.2 为什么编排层选 FastAPI 而不是 Flask 或 DjangoFastAPI 在这个场景里有三个硬优势是我选它的直接原因。第一是原生异步。LLM 调用是典型的 IO 密集型操作一个请求可能要等好几秒甚至几十秒。Flask 的同步模型在这种场景下并发能力很差而 FastAPI 基于 asyncio一个 worker 就能扛住大量并发等待。第二是自动生成 OpenAPI 文档前后端联调时 Flutter 这边直接照着 schema 写模型类省去手写接口文档的麻烦。第三是Pydantic 数据校验Agent 之间传递的结构化数据用它定义类型不对直接报错比手动 if-else 校验靠谱得多。Django 太重自带 ORM、Admin 那一套在这个项目里用不上反而拖慢启动。Flask 轻是轻但异步支持要靠额外扩展不如 FastAPI 原生来得干净。2.3 多智能体的拆分策略Agent 怎么拆是整个项目最考验设计功力的地方。我的原则是按能力边界拆不按对话轮次拆。一个典型的三 Agent 结构是这样的Agent 名称职责输入输出Router Agent意图识别与任务分发用户原始 query任务类型 目标 AgentTool Agent调用外部工具搜索、计算、数据库结构化任务参数工具执行结果Summary Agent整合结果并生成自然语言回复各 Agent 的中间结果最终回复文本Router 只做分类Tool 只做执行Summary 只做整合。每个 Agent 的 prompt 都能写得很短很聚焦准确率自然就上去了。我试过把所有逻辑塞进一个 Agentprompt 写到两千多字结果模型经常忘记其中某条规则拆开之后这个问题基本消失。2.4 Flutter 端的状态管理选型Flutter 这边我推荐用Riverpod而不是 Provider 或 Bloc。原因很实际流式输出场景下状态更新非常频繁Riverpod 的细粒度监听能避免整棵树重建。Bloc 的 event-state 模式在这种高频更新下写起来很啰嗦Provider 又容易在复杂依赖下出现 rebuild 失控。具体做法是把会话消息列表和当前流式片段分成两个独立的 provider。流式片段每来一个 token 就更新一次只触发对应气泡的重建历史消息列表不动。这个细节对流畅度影响很大后面实操部分会展开。3. 核心细节解析与实操要点3.1 FastAPI 项目目录结构怎么组织很多人写 FastAPI 项目所有路由堆在一个 main.py 里几百行之后自己都找不到。我用的结构是这样的app/ ├── main.py # 应用入口注册路由和中间件 ├── api/ │ ├── __init__.py │ └── chat.py # 对话相关接口 ├── agents/ │ ├── base.py # Agent 基类定义统一接口 │ ├── router_agent.py │ ├── tool_agent.py │ └── summary_agent.py ├── core/ │ ├── config.py # 配置管理 │ └── llm_client.py # 模型调用封装 ├── schemas/ │ └── chat.py # Pydantic 模型 └── services/ └── orchestrator.py # 多智能体编排逻辑关键点是agents/base.py里定义一个抽象基类所有 Agent 继承它统一async def run(self, input) - output接口。这样编排层调度的时候不用关心具体是哪个 Agent直接遍历调用就行。这个设计我强烈推荐后期加新 Agent 只要继承基类、注册到编排器其他代码一行不用改。3.2 Agent 基类与统一接口设计基类长这样from abc import ABC, abstractmethod from pydantic import BaseModel class AgentInput(BaseModel): query: str context: dict {} class AgentOutput(BaseModel): result: str metadata: dict {} class BaseAgent(ABC): def __init__(self, name: str, llm_client): self.name name self.llm llm_client abstractmethod async def run(self, data: AgentInput) - AgentOutput: pass为什么用 Pydantic 定义输入输出因为 Agent 之间传数据最容易出问题。有一次 Tool Agent 返回的字段名和 Summary Agent 期望的不一致排查了半天。用了 Pydantic 之后字段对不上直接抛 ValidationError错误信息还告诉你缺哪个字段效率提升非常明显。3.3 Flutter 端流式渲染的关键实现Flutter 接收流式数据我用的是http包的StreamedRequest配合utf8.decoder逐块解析 SSE 格式。核心逻辑是final request http.Request(POST, uri); request.body jsonEncode(payload); final response await client.send(request); response.stream .transform(utf8.decoder) .transform(const LineSplitter()) .listen((line) { if (line.startsWith(data: )) { final chunk line.substring(6); if (chunk [DONE]) return; ref.read(streamingProvider.notifier).append(chunk); } });这里有个坑LineSplitter遇到不完整的行会缓存但如果服务端发送的 chunk 边界刚好切在 UTF-8 多字节字符中间utf8.decoder会报错。解决办法是用utf8.decoder的allowMalformed: true参数或者确保服务端按完整字符发送。我踩过一次中文乱码的坑就是因为这个。3.4 多智能体协同的编排逻辑编排器是整个系统的大脑核心是一个状态机式的流程class Orchestrator: def __init__(self, agents: dict): self.agents agents async def execute(self, query: str): # 第一步路由 route_result await self.agents[router].run( AgentInput(queryquery) ) task_type route_result.metadata[task_type] # 第二步按类型调度 if task_type tool_call: tool_result await self.agents[tool].run( AgentInput(queryquery, contextroute_result.metadata) ) context {tool_output: tool_result.result} else: context {} # 第三步整合输出 final await self.agents[summary].run( AgentInput(queryquery, contextcontext) ) return final这个流程看起来简单但每一步都有讲究。Router 的输出必须是结构化的比如 JSON否则后面没法程序化判断。Tool Agent 要处理超时和失败重试。Summary Agent 的 prompt 里要明确告诉它基于以下事实回答不要编造。提示Router Agent 的输出格式一定要用 JSON mode 或者 function calling 强制约束纯文本解析非常脆弱模型稍微换个措辞你的解析就崩了。4. 实操过程与核心环节实现4.1 从零创建 Flutter 项目并配置依赖创建项目这一步命令行和 IDE 都行我习惯用命令行flutter create --org com.example --platformsandroid,ios multi_agent_app cd multi_agent_app--platforms参数指定只生成需要的平台能减少项目体积。生成之后pubspec.yaml里加这几个依赖dependencies: flutter: sdk: flutter flutter_riverpod: ^2.4.0 http: ^1.1.0 freezed_annotation: ^2.4.1 dev_dependencies: build_runner: ^2.4.0 freezed: ^2.4.1Riverpod 管状态http 发请求freezed 生成不可变数据类。freezed 这个可能有人觉得多余但在消息列表这种频繁增删的场景下不可变数据类能避免很多改了引用但没触发重建的诡异 bug。4.2 FastAPI 服务搭建与依赖安装后端这边先建虚拟环境再装依赖别直接往全局环境装python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install fastapi uvicorn[standard] pydantic httpx python-dotenvuvicorn[standard]带上了 uvloop 和 httptools性能比裸 uvicorn 好不少。启动命令uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload--reload只在开发时用生产环境一定要去掉否则文件监听会拖慢性能。4.3 模型调用封装与超时重试模型调用这块我单独封装了一个 client把超时、重试、错误处理都收进去import httpx from tenacity import retry, stop_after_attempt, wait_exponential class LLMClient: def __init__(self, base_url: str, api_key: str, timeout: float 30.0): self.client httpx.AsyncClient( base_urlbase_url, timeouthttpx.Timeout(timeout, connect5.0), headers{Authorization: fBearer {api_key}} ) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) async def chat(self, messages: list, stream: bool False): resp await self.client.post(/v1/chat/completions, json{ messages: messages, stream: stream }) resp.raise_for_status() return resp超时设置有个经验值连接超时 5 秒读取超时 30 秒。读取超时不能太短模型生成长文本时可能超过 10 秒也不能太长否则用户等半天没反应。重试用指数退避第一次等 1 秒第二次 2 秒第三次 4 秒避免瞬间打爆服务端。4.4 流式接口的完整实现FastAPI 返回流式响应用StreamingResponsefrom fastapi.responses import StreamingResponse router.post(/chat/stream) async def chat_stream(req: ChatRequest): async def event_generator(): async for chunk in orchestrator.stream_execute(req.query): yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no } )X-Accel-Buffering: no这个头很关键。如果你前面挂了 Nginx 之类的反向代理不加这个头代理会缓冲整个响应再一次性发出去流式效果就没了。我第一次部署时被这个坑了整整一个下午本地测试好好的一上服务器就变成转圈半天然后一次性出结果。4.5 Flutter 端消息列表与流式气泡客户端这边消息模型用 freezed 定义freezed class ChatMessage with _$ChatMessage { const factory ChatMessage({ required String id, required String role, required String content, Default(false) bool isStreaming, }) _ChatMessage; }流式气泡的渲染逻辑是收到第一个 chunk 时创建一条isStreaming: true的消息后续 chunk 追加到这条消息的 content 上收到[DONE]时把isStreaming置为 false。这样用户看到的就是文字一个个蹦出来的效果。注意追加 content 时不要每次都创建新对象用 Riverpod 的state state.copyWith(...)freezed 生成的 copyWith 会做结构共享性能比全量重建好很多。5. 常见问题与排查技巧实录5.1 流式输出中断或乱码这是最高频的问题。排查顺序我总结成一张表现象可能原因排查方法中文乱码UTF-8 解码边界问题检查 decoder 是否 allowMalformed输出到一半停住代理缓冲检查 X-Accel-Buffering 头完全没流式效果服务端未 flush确认 yield 后没有额外缓冲偶发连接断开超时设置过短调大 read timeout我遇到最多的是代理缓冲其次是客户端解码。这两个占了流式问题的八成以上。5.2 Agent 之间数据传递失败前面提过字段名不一致是重灾区。我的解决办法是所有 Agent 的输入输出都用 Pydantic 模型定义并且在编排器里做一次显式校验。宁可多写几行校验代码也不要让错误数据流到下一个 Agent。另外Agent 的 metadata 字段建议统一用字符串键避免嵌套过深调试时打印出来一目了然。5.3 并发场景下的性能问题AI Agent 怎么扛并发这是很多人关心的。我的经验是瓶颈几乎永远在模型调用不在你的服务代码。所以优化方向是用异步 IO别用同步阻塞调用给模型调用加连接池复用 TCP 连接对相同 query 做结果缓存命中直接返回设置合理的并发上限超过就排队别让请求雪崩我实测过一个 4 核 8G 的机器纯转发场景下 FastAPI 能扛住几百并发但一旦每个请求都要调模型并发数就取决于模型服务的吞吐了。所以别在服务层过度优化先把模型侧的并发能力搞清楚。5.4 Flutter 页面切换后状态丢失flutter navigator 切换页面后会不会丢失状态这个问题的答案是取决于你怎么管理状态。用Navigator.push压栈原页面还在栈里状态保留用pushReplacement替换原页面销毁状态丢失。如果状态放在页面内部的StatefulWidget里页面销毁就没了放在 Riverpod 的全局 provider 里页面销毁状态还在。我的做法是会话数据放全局 providerUI 临时状态比如输入框内容、滚动位置放页面内部。这样既不会丢数据也不会让全局状态膨胀。5.5 模型返回格式不符合预期llm request failed: provider rejected the request schema or tool payload这类报错通常是请求体结构不对。排查要点检查 messages 数组格式role 只能是 system/user/assistanttool 定义要符合对应模型的 schema 规范有些模型不支持某些参数传了会直接拒绝我的习惯是先用 curl 或 Postman 手动发一次请求确认请求体没问题再往代码里搬。这样能把代码问题和请求格式问题分开排查。6. 我在这套架构上的一些实战心得先说一个反直觉的结论多智能体系统的复杂度八成来自编排逻辑而不是 Agent 本身。每个 Agent 单独写都很简单难的是它们之间怎么协作、怎么传递上下文、怎么处理某个 Agent 失败的情况。所以我在项目里会花大量时间设计编排器的状态流转而不是纠结单个 Agent 的 prompt 怎么写。第二个心得是关于 prompt 的。Router Agent 的 prompt 我改了不下二十版最后发现最有效的写法是给几个明确的例子而不是写一堆规则。比如用户问天气属于 tool_call用户问概念属于 direct_answer配上两三个具体例子比写十条抽象规则准确率高得多。第三个是关于调试的。多智能体系统出问题时光看最终输出根本不知道哪一步错了。我的做法是在编排器里加一个 debug 模式把每个 Agent 的输入输出都打到日志里用请求 ID 串起来。这样一次请求的完整链路一目了然。这个功能开发时多花半小时调试时能省几十小时。最后分享一个 Flutter 端的小技巧流式渲染时如果 token 来得太快每来一个就 setState 会导致界面卡顿。我的做法是加一个 16ms 的节流差不多一帧的时间把这段时间内到达的 token 合并成一次更新。实测下来流畅度提升非常明显尤其是模型输出速度快的时候。这套架构我前后迭代了三个版本从最初的单 Agent 硬扛到现在的三 Agent 协作每一步都是被实际问题逼出来的。如果你正准备入这个坑我的建议是先把单 Agent 跑通再拆多 Agent。别一上来就设计复杂的协作流程那样你连问题出在哪都找不到。等单 Agent 的链路稳定了再按能力边界逐步拆分每一步都保证可回滚这样最稳。
阅读完成 · 觉得有帮助？
