1. 为什么我选择 Flutter LLM 多智能体这条技术路线移动端接入大模型这件事从 2023 年下半年开始就不断有人问我到底值不值得做我的回答一直是——看你想要什么。如果你只是想做个聊天框套壳那确实没什么意思一个 WebView 就够了。但如果你想做的是让手机端真正具备多个角色协同干活的能力那 Flutter LLM 多智能体这套组合是我目前试下来最顺手的一条路。先说清楚这套东西是什么。Flutter负责跨平台 UI 和原生能力调用LLM负责语言理解和生成多智能体Multi-Agent负责把一个大任务拆成若干角色分工协作FastAPI负责在后端把模型调用、工具执行、状态管理串起来。四者拼在一起你得到的是一个手机上的 AI 工作台而不是一个只会聊天的玩具。它适合谁我的判断是三类人一是已经会 Flutter、想往 AI 方向延伸的移动端开发者二是会 Python 后端、想补上移动端入口的 AI 工程师三是想做一个能落地的 AI Agent 项目、但不想一上来就啃分布式框架的独立开发者。如果你属于这三类中的任何一类这篇内容应该能帮你少走至少两周弯路。我踩过的第一个坑就是很多人把多智能体理解成多个模型实例。不是的。多智能体的核心不是模型数量而是角色划分 消息路由 状态共享。你完全可以用同一个模型通过不同的 system prompt 和工具集扮演 Planner、Executor、Critic 三个角色效果比堆三个不同模型还好。这个认知如果不先建立起来后面代码会写得很乱。还有一个反直觉的结论移动端做 Agent瓶颈从来不在模型而在通信和状态同步。Flutter 和原生之间、Flutter 和 FastAPI 之间、Agent 和 Agent 之间的消息传递才是真正吃掉你 70% 调试时间的地方。所以这篇我会把大量篇幅放在通信机制和状态管理上而不是教你调 API。2. 多智能体的角色设计别急着写代码先把谁干什么想清楚2.1 三个核心角色Planner、Executor、Critic我在实际项目里最稳定的角色划分是三个多一个都嫌乱Planner规划者接收用户原始需求输出结构化的任务列表。它的输出必须是 JSON不能是自然语言否则下游解析会崩溃。Executor执行者拿到单个子任务调用具体工具搜索、计算、数据库查询、Flutter 端能力完成它。Critic评审者检查 Executor 的结果是否满足 Planner 的预期不满足就打回重做最多重试 2 次。为什么是三个而不是两个因为没有 Critic 的 Agent 系统错误会一路传播到最终输出。我做过对比测试同样一个帮我整理本周会议纪要并生成待办的任务两角色方案Planner Executor的最终可用率大概在 62%加上 Critic 之后能到 85% 以上。这个提升非常值得多花的那点 token。2.2 用 system prompt 而不是多个模型来区分角色很多人第一反应是Planner 用 GPT-4Executor 用便宜的小模型。这个思路在成本上没错但在工程上会带来一个麻烦不同模型的输出格式不一致解析逻辑要写两套。我的做法是同一个模型三套 system prompt统一输出 JSON schema。这样解析层只需要一套代码。下面是我实际用的 Planner prompt 骨架PLANNER_PROMPT 你是一个任务规划专家。用户会给你一个需求你需要把它拆解成 1-5 个可独立执行的子任务。 输出必须是严格的 JSON格式如下 { tasks: [ {id: 1, action: 动作类型, params: {...}, depends_on: []}, ... ] } 可用动作类型search, calculate, query_db, summarize, notify 不要输出任何 JSON 之外的内容。注意最后那句不要输出任何 JSON 之外的内容。这句话不加模型有 30% 概率会在 JSON 前后加一句好的我来帮你规划然后你的json.loads就炸了。这是血泪教训。2.3 消息路由Agent 之间怎么说话多智能体最容易写乱的地方就是消息传递。我的建议是不要用自由文本传递用结构化消息。定义一个统一的消息格式class AgentMessage: def __init__(self, sender, receiver, msg_type, payload, trace_id): self.sender sender # 谁发的 self.receiver receiver # 发给谁 self.msg_type msg_type # task / result / critique / error self.payload payload # 具体内容 self.trace_id trace_id # 同一次请求的追踪 IDtrace_id这个字段看起来不起眼但没有它你根本没法调试。一次用户请求可能触发十几个 Agent 消息日志里没有 trace_id 串起来你就是在看天书。我建议用uuid4()生成在 FastAPI 的中间件里统一注入。2.4 状态共享用黑板模式而不是消息队列Agent 之间共享状态有两种常见做法消息队列和黑板模式。移动端场景我强烈推荐黑板模式。原因很简单消息队列适合异步、解耦、高并发的场景但移动端 Agent 的任务通常是短链路、强顺序的。用消息队列反而增加复杂度。黑板模式就是一个共享的 dict所有 Agent 都能读写class Blackboard: def __init__(self): self.data {} self.lock asyncio.Lock() async def write(self, key, value, agent_id): async with self.lock: self.data[key] {value: value, by: agent_id, ts: time.time()} async def read(self, key): return self.data.get(key, {}).get(value)这个asyncio.Lock不能省。FastAPI 是异步的多个 Agent 协程同时写黑板不加锁会出现数据覆盖。我一开始图省事没加结果在并发测试时出现了任务结果丢失查了整整一个下午。3. FastAPI 后端的目录结构别把 Agent 逻辑塞进路由里3.1 我实际用的目录结构FastAPI 项目最容易犯的错就是把所有逻辑写在main.py或者路由函数里。Agent 项目尤其致命因为 Agent 逻辑本身就复杂再和 HTTP 层混在一起两周后你自己都看不懂。我用了大半年的结构是这样的app/ ├── main.py # 只做 app 初始化和路由注册 ├── api/ │ ├── routes_agent.py # Agent 相关接口 │ └── routes_health.py # 健康检查 ├── agents/ │ ├── base.py # Agent 基类 │ ├── planner.py │ ├── executor.py │ └── critic.py ├── core/ │ ├── blackboard.py # 黑板 │ ├── router.py # 消息路由 │ └── llm_client.py # 模型调用封装 ├── schemas/ │ └── agent_schema.py # Pydantic 模型 └── config.py # 配置这个结构的关键在于agents/和api/完全解耦。路由层只负责接收请求、调用 Agent 编排器、返回结果不关心 Agent 内部怎么跑。这样你换 Web 框架比如换成 Gradio 做 demo时Agent 层一行都不用改。3.2 为什么用 Pydantic 定义所有消息FastAPI 自带 Pydantic很多人只拿它做请求体校验太浪费了。Agent 之间的消息也应该用 Pydantic 模型好处有三个自动校验字段缺失或类型错误立刻报错而不是等到下游才炸自动生成 JSON schema可以直接塞给 LLM 做结构化输出约束IDE 有补全写代码时不容易拼错字段名from pydantic import BaseModel, Field from typing import List, Optional class Task(BaseModel): id: int action: str params: dict Field(default_factorydict) depends_on: List[int] Field(default_factorylist) class Plan(BaseModel): tasks: List[Task] reasoning: Optional[str] Nonereasoning字段是可选的我加它是为了调试——让 Planner 顺便说一下为什么这么拆出问题时能快速定位是规划错了还是执行错了。3.3 LLM 客户端封装重试、超时、降级一个都不能少直接调模型 API 的代码在生产环境活不过一天。我的llm_client.py里至少包含这几层保护import asyncio from tenacity import retry, stop_after_attempt, wait_exponential class LLMClient: def __init__(self, model, timeout30): self.model model self.timeout timeout retry(stopstop_after_attempt(3), waitwait_exponential(min1, max8)) async def call(self, messages, json_modeFalse): try: resp await asyncio.wait_for( self._raw_call(messages, json_mode), timeoutself.timeout ) return resp except asyncio.TimeoutError: raise LLMTimeoutError(f模型调用超时 {self.timeout}s)wait_exponential是指数退避第一次等 1 秒第二次 2 秒第三次 4 秒。不要用固定间隔重试模型服务端限流时固定间隔重试只会加剧拥堵。json_mode参数对应模型的结构化输出能力开启后模型会强制返回合法 JSON能省掉大量解析容错代码。3.4 并发处理Agent 怎么扛住同时来的请求AI Agent 怎么扛并发这个问题我被问过很多次。答案分两层第一层是 FastAPI 层。FastAPI 本身是异步的用async def定义路由配合uvicorn的多 worker 模式单机扛几百 QPS 没问题。但要注意Agent 编排逻辑里如果有同步阻塞调用比如某些 SDK一定要用run_in_executor包起来否则会阻塞整个事件循环。第二层是 Agent 层。同一个用户的多个请求之间要隔离不同用户的请求之间更要隔离。我的做法是每个请求创建一个独立的 Blackboard 实例而不是全局共享一个。这样天然隔离不需要复杂的锁竞争。app.post(/agent/run) async def run_agent(req: AgentRequest): trace_id str(uuid4()) board Blackboard() # 每个请求独立 orchestrator Orchestrator(board, trace_id) result await orchestrator.run(req.query) return {trace_id: trace_id, result: result}如果并发量真的很大再考虑引入任务队列比如 Celery 或 arq把 Agent 执行异步化接口立刻返回 task_id客户端轮询结果。但在日活几千的量级上面这套同步方案完全够用不要过早优化。4. Flutter 端的通信与状态管理EventChannel 和 PlatformView 怎么选4.1 Flutter 调 FastAPI别用 http 包裸调Flutter 调后端接口很多人直接用http包。能用但在 Agent 场景下不够用因为 Agent 请求往往耗时长几秒到几十秒需要流式返回中间状态。我的方案是SSEServer-Sent Events dio。FastAPI 侧用StreamingResponse推送 Agent 的每一步进展Flutter 侧用 dio 的ResponseType.stream接收final response await dio.post( /agent/run, data: {query: userQuery}, options: Options(responseType: ResponseType.stream), ); response.data.stream.listen((chunk) { final text utf8.decode(chunk); // 解析 SSE 格式更新 UI });这样用户能看到正在规划...正在搜索...正在总结...的实时进度体验比转圈等待好太多。这是移动端 Agent 应用和 Web 端最大的体验差异点一定要利用好。4.2 EventChannel vs PlatformView什么时候用哪个热词里出现了flutter eventchannel和flutter platformview这两个是 Flutter 和原生通信的两条路用途完全不同机制适用场景数据方向性能MethodChannel一次性调用双向高EventChannel持续事件流原生→Flutter高PlatformView嵌入原生 UI 组件双向较低我的经验是Agent 相关的原生能力比如调用系统语音识别、读取本地文件用 EventChannel因为语音识别结果是持续流式返回的正好匹配 EventChannel 的模型。只有当你需要嵌入一个原生地图、原生相机预览这种 UI 组件时才用 PlatformView。PlatformView 的性能开销明显更大因为它要在 Flutter 的渲染树里挖个洞放原生 View合成成本高。我见过有人用 PlatformView 做一个纯文本显示这是典型的杀鸡用牛刀。4.3 页面切换丢状态Navigator 的坑flutter navigator切换页面后会丢失状态吗这个问题答案是取决于你怎么切。用Navigator.push压栈原页面还在栈里状态保留用Navigator.pushReplacement替换原页面被销毁状态丢失用Navigator.pop返回栈顶页面销毁下层页面状态保留Agent 应用里对话页面千万不要用 pushReplacement。用户切到设置页再回来发现对话历史没了这是灾难性体验。正确做法是用push并且把对话状态提到页面之上用 Provider 或 Riverpod 管理。// 用 Riverpod 管理对话状态页面重建也不丢 final conversationProvider StateNotifierProviderConversationNotifier, ListMessage( (ref) ConversationNotifier(), );4.4 Impeller 渲染引擎对 Agent 应用的实际影响flutter impeller是 Flutter 新的渲染引擎替代了老的 Skia。对 Agent 应用来说最直观的影响是长列表滚动的流畅度。Agent 对话往往很长消息列表动辄几百条Impeller 在这方面的表现明显更稳掉帧少。但要注意一个坑Impeller 在部分低端安卓设备上还有兼容问题尤其是老款 GPU。我的做法是在AndroidManifest.xml里保留切换开关出问题时能快速回退meta-data android:nameio.flutter.embedding.android.EnableImpeller android:valuefalse /上线前一定要在目标机型上实测别只看模拟器。5. 从零跑通一个最小可用 Agent完整实操链路5.1 环境准备版本对齐是第一道坎Flutter 和 Python 的版本管理我建议Flutter 用FVM管理别用全局版本。Agent 项目经常要试新特性全局版本一升级其他项目全炸。Python 用uv或poetry别用全局 pip。FastAPI 生态更新快依赖冲突很常见。# Flutter 侧 fvm install 3.22.0 fvm use 3.22.0 # Python 侧 uv venv uv pip install fastapi uvicorn httpx pydantic tenacity如何 as 创建 flutter 项目这个热搜词其实 Android Studio 创建 Flutter 项目很简单File → New → New Flutter Project选好 SDK 路径即可。但我建议用命令行创建因为 AS 的模板会塞一堆你不需要的注释和示例代码fvm flutter create --org com.yourcompany --platformsandroid,ios agent_app--platforms参数很重要不指定的话会生成 web、linux、windows 一堆你不需要的目录编译时白白浪费时间。5.2 后端最小闭环一个接口跑通三角色先不追求功能完整把 Planner → Executor → Critic 的链路跑通# orchestrator.py class Orchestrator: def __init__(self, board, trace_id): self.board board self.trace_id trace_id self.planner PlannerAgent() self.executor ExecutorAgent() self.critic CriticAgent() async def run(self, query): plan await self.planner.plan(query) await self.board.write(plan, plan, planner) results [] for task in plan.tasks: result await self.executor.execute(task) critique await self.critic.review(task, result) if not critique.passed: result await self.executor.execute(task, feedbackcritique.feedback) results.append(result) return await self.summarize(results)这段代码里Critic 的重试逻辑只做一次。为什么不做多次因为实测下来第二次还不过的任务第三次大概率也过不了纯粹浪费 token。这时候应该把问题抛给用户而不是死循环。5.3 Flutter 端最小闭环一个页面 一个流式请求Flutter 侧先做一个最简单的页面输入框 发送按钮 结果列表。关键是流式接收class AgentPage extends StatefulWidget { override StateAgentPage createState() _AgentPageState(); } class _AgentPageState extends StateAgentPage { final ListString _messages []; final _controller TextEditingController(); Futurevoid _send() async { final query _controller.text; setState(() _messages.add(我: $query)); final response await dio.post( http://10.0.2.2:8000/agent/run, data: {query: query}, options: Options(responseType: ResponseType.stream), ); response.data.stream.listen((chunk) { final text utf8.decode(chunk); setState(() _messages.add(Agent: $text)); }); } }注意10.0.2.2这个地址——这是安卓模拟器访问宿主机的专用 IP用localhost是连不上的。这个坑我见过太多人踩包括我自己第一次做的时候。5.4 联调时最容易卡住的三个点第一CORS。Flutter Web 调试时会有跨域问题FastAPI 要加 CORSMiddleware。但Flutter 原生Android/iOS没有 CORS 概念如果你只在原生上跑不用加。很多人加了 CORS 还是连不上就是因为搞混了运行环境。第二网络权限。Android 需要在AndroidManifest.xml里加INTERNET权限而且Android 9 以后默认禁止明文 HTTP本地调试要么用 HTTPS要么在 manifest 里开usesCleartextTraffic。第三超时设置。dio 默认超时是 5 秒Agent 请求经常超过这个时间。一定要手动设置final dio Dio(BaseOptions( connectTimeout: Duration(seconds: 10), receiveTimeout: Duration(minutes: 2), ));receiveTimeout设成 2 分钟因为流式响应是持续接收的超时时间要覆盖整个 Agent 执行周期。6. 实测中的性能与安全边界这些坑我替你踩过了6.1 Token 成本控制三个立竿见影的手段Agent 应用烧 token 是常态我总结了三个最有效的控制手段手段一Critic 用小模型。Critic 的判断逻辑相对简单结果是否满足要求用便宜的小模型完全够用成本能降 60% 以上。手段二黑板数据裁剪。Agent 之间传递的上下文不要全量传只传当前任务需要的字段。我见过有人把整个对话历史塞给每个 Agenttoken 消耗直接翻十倍。手段三缓存 Planner 结果。相似的需求Planner 的输出往往高度相似。用 query 的 embedding 做缓存 key命中率能到 30% 左右。6.2 Agent 安全输入过滤和工具白名单agent安全这个热搜词不是危言耸听。Agent 能调用工具就意味着如果输入被恶意构造可能触发非预期的工具调用。我的防护措施有两层第一层输入过滤。用户输入里如果包含明显的指令注入模式比如忽略之前的指令直接拒绝。第二层工具白名单。Executor 能调用的工具是硬编码的白名单LLM 输出的 action 如果不在白名单里直接报错绝不动态执行。ALLOWED_ACTIONS {search, calculate, query_db, summarize, notify} def validate_action(action): if action not in ALLOWED_ACTIONS: raise SecurityError(f非法动作: {action})永远不要用eval或exec执行 LLM 生成的代码。这句话我说一百遍都不嫌多。6.3 移动端的特殊约束内存和电量移动端跑 Agent有两个桌面端不用考虑的问题内存。Flutter 的 Dart 堆和原生堆是分开的Agent 返回的大 JSON 如果直接解析成对象树很容易触发 OOM。我的做法是流式解析边接收边处理不攒完整响应。电量。持续的网络请求和模型推理如果端侧跑小模型会显著耗电。长任务一定要支持后台执行 通知回调不要让用户盯着屏幕等。6.4 我踩过的最大的一个坑状态不一致最后分享一个我踩得最惨的坑。有一次线上出现Agent 显示任务完成但结果为空的问题。查了两天才发现Flutter 端的流式接收和 FastAPI 端的流式发送对 SSE 的event字段处理不一致。FastAPI 发送时用了自定义 event 类型Flutter 端的解析代码只处理了默认的message类型其他类型全被丢弃了。结果就是完成事件收到了结果事件丢了。修复很简单但发现很难。我的教训是SSE 的解析一定要写单元测试把各种 event 类型都覆盖到。别指望联调时能发现因为联调时你往往只测了 happy path。// 正确的 SSE 解析要处理所有 event 类型 void parseSSE(String raw) { final lines raw.split(\n); String? eventType; for (final line in lines) { if (line.startsWith(event:)) { eventType line.substring(6).trim(); } else if (line.startsWith(data:)) { final data line.substring(5).trim(); _handleEvent(eventType ?? message, data); } } }这套东西我从零跑通大概花了两周其中一周都在调通信和状态同步。如果你按这篇的路径走应该能压缩到三到四天。核心就一句话先把角色划分和消息格式定死再动手写代码。角色和消息格式定好了剩下的都是体力活定不好后面全是返工。