1. 为什么要把 Flutter 和 LLM 多智能体放在一起折腾移动端做 AI 应用最容易踩的坑就是“把大模型当接口调”。一开始我也这么干Flutter 页面里塞一个输入框用户点一下后端转发给 LLM返回一段文字渲染出来收工。这种玩法做 Demo 没问题但只要业务稍微复杂一点比如用户问“帮我查一下上周的订单顺便看看物流到哪了如果延迟了就发个提醒”单个 LLM 调用根本扛不住——它既要做意图识别又要查订单又要判断物流状态还要生成提醒文案一个 prompt 里塞四件事结果就是每件事都做得稀碎。多智能体Multi-Agent这套思路就是来解决这个问题的。核心逻辑很朴素把一件复杂的事拆给几个“专职角色”每个角色只干自己最擅长的一段中间通过结构化的消息传递协作。放到 Flutter LLM 的场景里就是前端负责交互和状态呈现后端用 FastAPI 挂载一个多智能体编排层每个 Agent 各司其职最后把结果流式吐回客户端。这套组合适合谁如果你已经写过 Flutter 页面、能跑通一个 FastAPI 的 hello world但一想到“多智能体”就觉得是论文里的东西那这篇就是写给你的。我不会上来就讲什么协同群集运动控制而是从能跑起来的最小闭环开始一层层把 Agent 的职责划分、通信协议、流式输出、状态管理这些真正卡人的地方讲透。关键词里出现的 Flutter、LLM、多智能体、Agent、FastAPI会在后面每个环节里落到具体的代码和配置上而不是停留在概念层面。先说清楚一个前提多智能体不是“多个模型实例堆在一起”。我见过有人开三个 LLM 客户端让它们互相聊天美其名曰多智能体结果 token 烧得飞快输出还互相矛盾。真正的多智能体系统关键在于职责边界清晰和消息契约稳定。前者决定每个 Agent 的 prompt 怎么写后者决定它们之间怎么传数据不出错。这两点想明白了后面写代码就是体力活。2. 整体架构设计与技术选型思路2.1 三层结构客户端、编排层、模型层我最终落地的结构是三层从下往上说。最底下是模型层就是实际的 LLM 调用。这里不把模型调用散落在各个 Agent 里而是统一封装成一个LLMClient负责处理重试、超时、token 计数、流式解析。为什么这么做因为多智能体系统里模型调用次数是单 Agent 的好几倍如果每个 Agent 自己写一套调用逻辑改一个超时参数要改五个地方维护成本直接爆炸。中间是编排层跑在 FastAPI 上。这一层是整套系统的核心包含 Agent 定义、路由逻辑、状态机、以及和客户端通信的 WebSocket 或 SSE 接口。编排层不关心 Flutter 长什么样它只负责“收到一个用户请求决定交给哪些 Agent按什么顺序执行把中间结果和最终结果推出去”。最上面是客户端层也就是 Flutter 应用。它做三件事收集用户输入、展示 Agent 的执行过程这点很重要后面细说、维护会话状态。Flutter 这边我强烈建议用flutter_bloc或者riverpod来管状态因为多智能体的输出是流式的、分阶段的用setState硬扛会写到怀疑人生。2.2 为什么编排层选 FastAPI 而不是别的FastAPI 在这个场景里有三个实打实的优势。第一是原生异步多智能体执行时大量时间花在等 LLM 返回上异步能让多个 Agent 的调用真正并行而不是假并发。第二是SSE 和 WebSocket 支持干净StreamingResponse配合async generator几行代码就能把流式输出搭起来。第三是Pydantic 做消息契约Agent 之间传的数据结构用 Pydantic 模型定义序列化和校验一步到位这在多智能体系统里太重要了——一个 Agent 输出的字段名写错下游全崩。有人会问为什么不用 Gradio。Gradio 做单轮 Demo 确实快但它的定位是“快速展示模型能力”不是“构建有状态的、多轮的多智能体服务”。Gradio 和 FastAPI 不是替代关系我实际项目里经常是 FastAPI 做后端Gradio 单独挂一个调试页面两者共存。2.3 Agent 数量不是越多越好新手最容易犯的错是“角色崇拜”恨不得把系统提示词拆成十个 Agent。我的经验是从两个开始最多不超过五个。两个 Agent 的最小组合是“规划者 执行者”规划者负责理解意图、拆解任务执行者负责调用工具、生成结果。三个 Agent 可以加一个“审核者”检查执行结果是否符合要求。超过五个消息传递的复杂度会指数上升调试时你根本追不清是哪一环出了问题。Agent 的划分依据应该是能力边界不是话题边界。比如“订单查询 Agent”和“物流查询 Agent”如果都只是调一个 HTTP 接口那完全可以合并成一个“数据查询 Agent”用不同的工具函数区分。只有当两个角色的 prompt 逻辑、工具集、输出格式差异很大时才值得拆开。3. 核心细节解析Agent 职责划分与消息契约3.1 用 Pydantic 定义 Agent 之间的消息格式多智能体系统崩得最多的地方就是 Agent 之间传的数据格式对不上。我的做法是所有 Agent 的输入输出都用 Pydantic 模型定义放在一个schemas.py里统一管理。from pydantic import BaseModel, Field from typing import Literal, Optional class AgentMessage(BaseModel): sender: str Field(description发送方 Agent 名称) receiver: str Field(description接收方 Agent 名称) intent: Literal[plan, execute, review, final] Field( description消息意图类型 ) payload: dict Field(default_factorydict, description结构化载荷) raw_text: Optional[str] Field(defaultNone, description原始文本调试用) class PlanResult(BaseModel): steps: list[str] Field(description拆解后的执行步骤) required_tools: list[str] Field(description需要的工具列表) confidence: float Field(ge0, le1, description规划置信度)这么做的直接好处是当执行 Agent 收到规划 Agent 的消息时PlanResult会强制校验steps是不是列表、confidence是不是在 0 到 1 之间。如果 LLM 返回的 JSON 里confidence写成了highPydantic 会直接抛错而不是让这个错误悄悄流到下游最后在用户面前炸开。提示让 LLM 输出结构化数据时不要只在 prompt 里写“请返回 JSON”一定要在调用时用模型支持的 JSON mode 或者 function calling再叠加 Pydantic 校验。双重保险缺一不可。3.2 规划 Agent 的 prompt 怎么写才不飘规划 Agent 是整个系统的入口它的输出质量直接决定后面所有环节。我踩过的坑是一开始 prompt 写得太开放让它“自由发挥拆解任务”结果它有时候拆三步有时候拆八步有时候还自己把执行结果也编出来了。后来我把 prompt 收紧成固定模板核心是三条约束。第一只输出步骤不输出结果明确告诉它“你的职责是规划不是执行”。第二步骤数量限制在 2 到 5 步超过 5 步说明任务太复杂应该先反问用户澄清。第三每步必须标注需要的工具名工具名从预定义列表里选不允许自创。PLANNER_PROMPT 你是一个任务规划助手。用户会给你一个请求你需要把它拆解成可执行的步骤。 可用工具列表 - query_order: 查询订单信息 - query_logistics: 查询物流状态 - send_notification: 发送通知 - generate_text: 生成文本内容 要求 1. 只输出规划步骤不要执行任何步骤 2. 步骤数量在 2 到 5 之间 3. 每个步骤必须标注使用的工具名且只能从上面的列表中选择 4. 如果用户请求信息不足在 steps 里加入一步向用户澄清 输出格式为 JSON包含 steps 和 required_tools 两个字段。 这个 prompt 的关键在于把 LLM 的自由度锁死。多智能体系统里每个 Agent 都应该是一个“窄接口”输入输出范围越小系统越稳定。3.3 执行 Agent 的工具调用与错误处理执行 Agent 拿到规划结果后按步骤调用工具。这里有个细节工具调用失败时不要让 Agent 自己瞎猜。我见过有的实现工具报错了LLM 收到错误信息后自己编了一个“假设查询成功”的结果继续往下走最后给用户的答案是假的。正确做法是工具调用失败时把错误信息结构化返回给执行 Agent并明确指示“如果工具失败停止当前步骤将错误上报”。执行 Agent 的输出里要有一个status字段取值success或failed失败时附带error_detail。class ExecutionResult(BaseModel): step_index: int tool_name: str status: Literal[success, failed] result: Optional[dict] None error_detail: Optional[str] None编排层收到failed状态时有两个选择要么重试当前步骤适合网络抖动这类瞬时错误要么把错误抛给审核 Agent 决定是否降级处理。我一般设置最多重试两次两次都失败就直接返回用户“当前无法完成该操作”绝不硬编。4. 实操过程从零搭一个可运行的多智能体服务4.1 FastAPI 项目目录结构先给一个我实际在用的目录结构这个结构在 FastAPI 面经里也经常被问到因为它兼顾了清晰和可扩展。multi_agent_service/ ├── app/ │ ├── main.py # FastAPI 入口挂载路由 │ ├── config.py # 配置读环境变量 │ ├── schemas.py # Pydantic 模型 │ ├── llm_client.py # LLM 调用封装 │ ├── agents/ │ │ ├── base.py # Agent 基类 │ │ ├── planner.py # 规划 Agent │ │ ├── executor.py # 执行 Agent │ │ └── reviewer.py # 审核 Agent │ ├── orchestrator.py # 编排逻辑 │ ├── tools/ │ │ ├── registry.py # 工具注册表 │ │ └── order_tools.py # 具体工具实现 │ └── api/ │ └── chat.py # SSE 流式接口 ├── requirements.txt └── .envagents/base.py里定义一个抽象基类所有 Agent 继承它统一run方法的签名。这样编排层调用不同 Agent 时不用关心具体实现只认基类接口。from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, name: str, llm_client): self.name name self.llm llm_client abstractmethod async def run(self, message: AgentMessage) - AgentMessage: 接收消息返回消息 pass4.2 SSE 流式接口的实现Flutter 端要展示 Agent 的执行过程就必须用流式传输。FastAPI 的StreamingResponse配合async generator是最省事的方案。from fastapi import APIRouter from fastapi.responses import StreamingResponse import json router APIRouter() router.post(/chat/stream) async def chat_stream(request: ChatRequest): async def event_generator(): async for event in orchestrator.run(request.user_input): yield fdata: {json.dumps(event, 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这个 header 一定要加。不加的话如果前面挂了 NginxNginx 会缓冲整个响应流式效果就没了用户要等所有 Agent 跑完才看到内容。我第一次部署时就被这个坑了半小时本地测试好好的一上服务器就变成“转圈十秒然后一次性出结果”。4.3 Flutter 端的状态管理与流式接收Flutter 这边接收 SSE 我推荐用http包的Client.send配合StreamedResponse不要用普通的get因为普通请求会把整个响应体读完才返回。import package:http/http.dart as http; import dart:convert; StreamAgentEvent streamChat(String userInput) async* { final request http.Request( POST, Uri.parse($baseUrl/chat/stream), ); request.headers[Content-Type] application/json; request.body jsonEncode({user_input: userInput}); final response await http.Client().send(request); final stream response.stream.transform(utf8.decoder); await for (final chunk in stream) { final lines chunk.split(\n\n); for (final line in lines) { if (line.startsWith(data: )) { final data line.substring(6); if (data [DONE]) return; yield AgentEvent.fromJson(jsonDecode(data)); } } } }状态管理用flutter_bloc的话定义一个ChatBloc把AgentEvent按类型分发到不同的 state。这里的关键是区分“中间过程事件”和“最终结果事件”。中间过程事件用来更新 UI 上的执行进度条最终结果事件才渲染成聊天气泡。如果不区分用户会看到一堆“规划中...执行中...”的碎片文字混在对话里体验很糟糕。注意Flutter 的Future.then回调是放入微任务队列的而 SSE 的流式数据是宏任务级别的 IO 事件。如果你在then里做重计算会阻塞后续事件的处理。流式接收的回调里只做轻量的 JSON 解析和状态更新重计算丢给compute或者 isolate。4.4 编排层的执行循环编排层是整个系统的心脏它的逻辑其实不复杂就是一个带状态的循环。class Orchestrator: def __init__(self, planner, executor, reviewer): self.planner planner self.executor executor self.reviewer reviewer async def run(self, user_input: str): # 第一步规划 yield {type: status, stage: planning} plan_msg await self.planner.run( AgentMessage(senderuser, receiverplanner, intentplan, payload{input: user_input}) ) plan PlanResult(**plan_msg.payload) yield {type: plan, steps: plan.steps} # 第二步逐步执行 results [] for i, step in enumerate(plan.steps): yield {type: status, stage: executing, step: i} exec_msg await self.executor.run( AgentMessage(senderplanner, receiverexecutor, intentexecute, payload{step: step, index: i}) ) result ExecutionResult(**exec_msg.payload) results.append(result) yield {type: step_result, index: i, result: result.dict()} if result.status failed: yield {type: error, detail: result.error_detail} return # 第三步审核并生成最终回复 yield {type: status, stage: reviewing} final_msg await self.reviewer.run( AgentMessage(senderexecutor, receiverreviewer, intentreview, payload{results: [r.dict() for r in results]}) ) yield {type: final, content: final_msg.raw_text}这个循环里每个yield都会通过 SSE 推到 Flutter 端。用户看到的是先出现“正在规划”然后列出步骤然后一步步打勾最后出结果。这种“过程可见”的设计比单纯转圈等待的体验好太多而且当某一步失败时用户能清楚看到是哪一步出了问题。5. 常见问题与排查技巧实录5.1 Agent 输出 JSON 解析失败怎么办这是最高频的问题。LLM 返回的文本里混了 markdown 代码块标记或者多了一句“好的以下是结果”json.loads直接崩。我的处理是三层防御。第一层调用 LLM 时开启 JSON mode如果模型支持。第二层解析前先用正则把json 和剥掉。第三层解析失败时不要直接抛异常而是把原始文本回传给 LLM让它“重新输出纯 JSON不要任何额外文字”最多重试两次。import re, json def safe_parse_json(text: str) - dict: text text.strip() text re.sub(r^(?:json)?\s*, , text) text re.sub(r\s*$, , text) try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.DOTALL) if match: return json.loads(match.group()) raise5.2 多智能体并发时的 token 消耗失控多智能体系统 token 消耗是单 Agent 的三到五倍这是必然的。但有些消耗是可以省下来的。我做过一个统计规划 Agent 的 prompt 占了总 token 的 40%因为每次都要把完整的工具列表和约束条件塞进去。优化手段有两个。一是把静态 prompt 部分做成缓存如果用的模型支持 prompt caching把工具列表和约束条件标记为可缓存能省不少钱。二是限制上下文长度执行 Agent 不需要知道完整的规划历史只需要知道当前步骤和前面步骤的结果摘要。我一开始把整个对话历史都传给每个 Agenttoken 直接爆了后来改成只传必要字段消耗降了六成。5.3 Flutter 端流式渲染卡顿Flutter 收到 SSE 事件后如果每个事件都触发一次setState或者 bloc 的emit高频事件下会掉帧。解决办法是做事件节流状态类事件status合并处理比如 100ms 内的多个 status 只保留最后一个结果类事件step_result、final立即处理因为它们需要及时展示。另外聊天气泡的列表一定要用ListView.builder不要用Column包一堆 widget。多智能体场景下消息数量增长很快Column会一次性构建所有子 widget消息一多就卡。5.4 常见问题速查表问题现象可能原因排查方向解决手段SSE 流式变成一次性返回Nginx 缓冲检查响应头加X-Accel-Buffering: noAgent 输出 JSON 解析失败LLM 加了额外文字打印原始返回正则剥离 重试执行 Agent 编造结果工具失败未上报检查工具错误处理强制 status 字段Flutter 流式渲染掉帧高频 setState用 DevTools 看帧率事件节流 ListView.buildertoken 消耗过高上下文冗余统计各 Agent prompt 长度缓存静态部分 精简上下文规划步骤过多prompt 约束不足检查规划输出限制步骤数 2-55.5 一个容易被忽略的坑Agent 名称大小写这个坑很蠢但很致命。Agent 之间传消息时sender和receiver字段如果大小写不一致比如规划 Agent 写的是Planner执行 Agent 期望的是planner路由就会失败。我的做法是在BaseAgent初始化时强制把 name 转成小写并且在AgentMessage的 Pydantic 模型里加一个 validator 做归一化。这种问题在单元测试里很难发现因为测试时你总是用一致的命名但线上跑起来LLM 生成的字段值可能大小写随机。6. 工具选型与扩展方向6.1 LLM 客户端封装的关键参数llm_client.py里我固定了几个参数这些是踩坑后总结出来的。timeout设 30 秒因为多智能体场景下单个 Agent 的调用不应该超过这个时间超了说明 prompt 有问题。max_retries设 2配合指数退避。temperature规划 Agent 用 0.1执行 Agent 用 0.3审核 Agent 用 0.0。规划需要稳定执行可以稍微灵活审核必须严格。流式解析时要注意不同模型的流式返回格式不一样。有的按 token 返回有的按 chunk 返回有的会在最后附一个 usage 统计。封装层要把这些差异抹平对上层的 Agent 暴露统一的async generator接口。6.2 从两个 Agent 扩展到更多角色的时机什么时候该加 Agent我的判断标准是当某个 Agent 的 prompt 里出现了明显的“如果...则...”分支逻辑且分支之间的工具集完全不重叠时就该拆了。比如执行 Agent 一开始既查订单又发通知后来发现查订单需要严格的参数校验发通知需要处理模板渲染两者逻辑差异太大就拆成两个独立 Agent。但拆之前先问自己能不能用同一个 Agent 加不同的工具函数解决大部分情况下答案是能。多智能体系统的复杂度是乘法增长的每加一个 Agent消息路径就多一批调试难度上一个台阶。克制是美德。6.3 后续可以扩展的方向这套骨架搭起来后往上加东西就顺了。可以加一个记忆 Agent把历史会话的关键信息存起来下次规划时作为上下文注入。可以加一个降级策略当主模型调用失败时自动切到备用模型。还可以把工具调用从本地函数扩展到外部 API这时候要注意加超时和熔断别让一个慢接口拖垮整个编排链。Flutter 端可以做的扩展更多把 Agent 执行过程做成可视化的流程图每个节点实时更新状态加一个“中断”按钮用户可以在执行到一半时取消把历史会话做成可回溯的时间线点某一步能看到当时的完整输入输出。这些体验上的打磨才是让多智能体应用从“能跑”变成“好用”的关键。我个人在实际操作中的体会是多智能体系统的难点从来不在模型本身而在边界定义和错误传播。把每个 Agent 的输入输出锁死把错误在最早的地方拦住剩下的就是工程问题。这套东西跑通一次之后你会发现它比单 Agent 的“万能 prompt”好维护得多因为每个环节都是可测试、可替换的。