Flutter 与 LLM 多智能体实战：FastAPI 编排与流式输出-平芜编程栈

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_factory=dict, description="结构化载荷") raw_text: Optional[str] = Field(default=None, description="原始文本，调试用") class PlanResult(BaseModel): steps: list[str] = Field(description="拆解后的执行步骤") required_tools: list[str] = Field(description="需要的工具列表") confidence: float = Field(ge=0, le=1, description="规划置信度")

这么做的直接好处是：当执行 Agent 收到规划 Agent 的消息时，PlanResult会强制校验steps是不是列表、confidence是不是在 0 到 1 之间。如果 LLM 返回的 JSON 里confidence写成了"high"，Pydantic 会直接抛错，而不是让这个错误悄悄流到下游，最后在用户面前炸开。

提示：让 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 └── .env

agents/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: """接收消息，返回消息""" pass

4.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 f"data: {json.dumps(event, ensure_ascii=False)}\n\n" yield "data: [DONE]\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"} )

这里有个坑：X-Accel-Buffering: no这个 header 一定要加。不加的话，如果前面挂了 Nginx，Nginx 会缓冲整个响应，流式效果就没了，用户要等所有 Agent 跑完才看到内容。我第一次部署时就被这个坑了半小时，本地测试好好的，一上服务器就变成“转圈十秒然后一次性出结果”。

4.3 Flutter 端的状态管理与流式接收

Flutter 这边接收 SSE 我推荐用http包的Client.send配合StreamedResponse，不要用普通的get，因为普通请求会把整个响应体读完才返回。

import 'package:http/http.dart' as http; import 'dart:convert'; Stream<AgentEvent> 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(sender="user", receiver="planner", intent="plan", 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(sender="planner", receiver="executor", intent="execute", 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(sender="executor", receiver="reviewer", intent="review", 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()) raise

5.2 多智能体并发时的 token 消耗失控

多智能体系统 token 消耗是单 Agent 的三到五倍，这是必然的。但有些消耗是可以省下来的。我做过一个统计，规划 Agent 的 prompt 占了总 token 的 40%，因为每次都要把完整的工具列表和约束条件塞进去。

优化手段有两个。一是把静态 prompt 部分做成缓存，如果用的模型支持 prompt caching，把工具列表和约束条件标记为可缓存，能省不少钱。二是限制上下文长度，执行 Agent 不需要知道完整的规划历史，只需要知道当前步骤和前面步骤的结果摘要。我一开始把整个对话历史都传给每个 Agent，token 直接爆了，后来改成只传必要字段，消耗降了六成。

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: no`
Agent 输出 JSON 解析失败	LLM 加了额外文字	打印原始返回	正则剥离 + 重试
执行 Agent 编造结果	工具失败未上报	检查工具错误处理	强制 status 字段
Flutter 流式渲染掉帧	高频 setState	用 DevTools 看帧率	事件节流 + ListView.builder
token 消耗过高	上下文冗余	统计各 Agent prompt 长度	缓存静态部分 + 精简上下文
规划步骤过多	prompt 约束不足	检查规划输出	限制步骤数 2-5

5.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”好维护得多，因为每个环节都是可测试、可替换的。