多智能体系统这两年从论文里走出来落到实际项目里的速度比很多人预想的要快。但真到动手阶段大部分人卡住的地方其实不是Agent 是什么这种概念问题而是前端用什么撑住交互、后端怎么把 LLM 调用和工具编排串起来、多个 Agent 之间怎么协同而不互相打架。我这次拿 Flutter 做客户端、FastAPI 做服务端、LLM 做推理核心搭了一套能跑起来的多智能体小系统从环境搭建到协同调度踩了不少坑这里把完整过程和我自己的取舍逻辑摊开讲一遍。适合已经会一点 Flutter 或者 Python、想往 Agent 方向落地的朋友也适合纯粹想看看多智能体到底怎么组装的开发者。1. 为什么选 Flutter FastAPI LLM 这套组合1.1 三个技术栈各自承担的角色先把分工说清楚不然后面容易乱。这套系统里Flutter 负责的是人机交互层——用户输入任务、看到 Agent 的思考过程、观察多个 Agent 的协作状态。FastAPI 负责编排层——接收请求、调度 Agent、管理会话状态、调用 LLM 接口、执行工具函数。LLM 负责决策层——理解任务、拆解步骤、决定调用哪个工具、生成最终回复。为什么不让 Flutter 直接调 LLM因为 API Key 不能放在客户端而且多智能体的编排逻辑谁先谁后、结果怎么汇总放在客户端会非常难维护。为什么不用纯 Python 脚本跑因为多智能体最有价值的场景是人机协作你需要一个能实时展示 Agent 状态的界面Flutter 的跨平台能力在这里很划算。1.2 和单 Agent 网页前端方案的对比很多人第一反应是用 Gradio 或者 Streamlit 快速搭个界面确实快但有两个硬伤。第一流式输出的控制粒度不够多智能体场景下你往往需要同时展示多个 Agent 的输出流Gradio 的组件模型做这个很别扭。第二状态管理弱Agent 之间的消息传递、任务队列、中间结果缓存这些用 Gradio 的 session 很难优雅处理。Flutter 的优势在于它的状态管理生态成熟Provider、Riverpod、Bloc 随便挑而且StreamBuilder天然适合处理 LLM 的流式返回。FastAPI 这边则是异步性能好、类型提示友好、和 LangChain/LangGraph 这类 Agent 框架集成顺畅。这套组合不是唯一解但在要交互、要流式、要多 Agent这三个约束下是我实测下来最顺手的。1.3 多智能体的核心价值到底在哪单 Agent 也能干活为什么非要多个关键在于职责分离带来的可靠性提升。一个 Agent 既要做规划、又要调工具、还要写最终答案很容易在某一步跑偏而且出错后很难定位是哪一环的问题。拆成多个专职 Agent 后规划 Agent 只管拆任务执行 Agent 只管调工具审核 Agent 只管检查结果每个环节的输入输出都清晰可测。举个具体例子用户说帮我查一下北京今天天气然后根据天气推荐穿什么。单 Agent 可能直接编一个天气就开始推荐。多 Agent 的话规划 Agent 拆成查天气和给建议两步工具 Agent 必须真的调用天气接口拿到数据建议 Agent 只能基于真实数据给建议。中间任何一步失败你都能立刻看到是哪个 Agent 出的问题。2. 环境搭建Flutter 和 FastAPI 各自的坑2.1 Flutter 项目创建与依赖选择创建项目本身很简单flutter create multi_agent_app就行。但有几个细节新手容易忽略。第一包名不要用默认的 com.example后面如果要打包上架会返工。第二Flutter 版本建议锁在稳定版我用的 3.22 系列Impeller 渲染引擎默认开启后列表滚动性能比 Skia 时代好不少但个别老设备上有兼容问题如果目标用户设备比较杂可以在AndroidManifest.xml里临时关掉。依赖方面核心就几个dependencies: flutter: sdk: flutter http: ^1.2.0 # 基础网络请求 dio: ^5.4.0 # 需要拦截器、超时控制时用 provider: ^6.1.0 # 状态管理 web_socket_channel: ^2.4.0 # 流式通信这里有个选择用 http 还是 dio。如果只是简单的 POST 请求http 够了。但多智能体场景下你需要处理流式响应SSE 或 WebSocket、请求重试、统一错误处理dio 的拦截器机制会省很多事。我最后两个都装了普通请求用 http流式用 dio 配合ResponseType.stream。2.2 FastAPI 项目目录结构怎么设计FastAPI 官方文档给的目录结构偏简单实际做 Agent 项目需要更细的分层。我踩过的坑是一开始把所有逻辑塞在main.py里写到第三个 Agent 就彻底乱了。后来改成这样backend/ ├── app/ │ ├── main.py # 应用入口注册路由 │ ├── api/ │ │ ├── routes_agent.py # Agent 相关接口 │ │ └── routes_chat.py # 对话接口 │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── llm_client.py # LLM 调用封装 │ ├── agents/ │ │ ├── base.py # Agent 基类 │ │ ├── planner.py # 规划 Agent │ │ ├── executor.py # 执行 Agent │ │ └── reviewer.py # 审核 Agent │ ├── tools/ │ │ └── registry.py # 工具注册表 │ └── schemas/ │ └── agent_schema.py # Pydantic 模型 ├── requirements.txt └── .env这个结构的好处是每个 Agent 独立成文件改一个不影响其他。tools/registry.py用装饰器模式注册工具函数新增工具不用改调度逻辑。schemas/里定义所有请求响应的 Pydantic 模型FastAPI 会自动做参数校验和文档生成。2.3 依赖安装与常见报错处理安装 FastAPI 全家桶pip install fastapi uvicorn[standard] python-dotenv pydantic-settings httpx如果要用 LangChain 或 LangGraph 做编排再加pip install langchain langchain-openai langgraph这里有个高频报错uvicorn 启动后日志丢失。原因是 uvicorn 默认的日志配置和 Python logging 模块冲突解决办法是在main.py里显式配置import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s )另一个坑是Pydantic v1 和 v2 的语法不兼容。LangChain 早期版本依赖 v1新版依赖 v2混装会报pydantic.errors.PydanticImportError。我的做法是统一用 v2然后检查所有依赖的版本兼容性pip check能帮你发现冲突。3. 多智能体的核心设计从单 Agent 到协同3.1 Agent 基类应该包含什么不管你有几个 Agent它们都有共性接收输入、调用 LLM、可能调用工具、返回输出。所以先定义一个基类from abc import ABC, abstractmethod from typing import Any class BaseAgent(ABC): def __init__(self, name: str, llm_client, tools: list None): self.name name self.llm llm_client self.tools tools or [] self.memory [] # 短期记忆 abstractmethod async def run(self, task: str, context: dict) - dict: 每个 Agent 必须实现的核心方法 pass def add_memory(self, role: str, content: str): self.memory.append({role: role, content: content}) # 控制记忆长度避免 token 爆炸 if len(self.memory) 20: self.memory self.memory[-20:]这里有个关键设计memory 的长度控制。LLM 的上下文窗口有限多轮对话后如果不裁剪token 消耗会指数级增长。我设的是保留最近 20 条实际项目里可以根据任务复杂度调整。裁剪策略也有讲究简单粗暴地丢最早的可能丢掉关键的任务背景更好的做法是让 LLM 自己总结历史。3.2 规划 Agent 的任务拆解逻辑规划 Agent 的职责是把用户的模糊需求拆成可执行的步骤。核心是 prompt 设计PLANNER_PROMPT 你是一个任务规划专家。用户会给你一个任务 你需要把它拆解成 2-5 个可执行的子步骤。 要求 1. 每个步骤必须具体、可验证 2. 标注每个步骤需要的能力如查数据、计算、生成文本 3. 输出 JSON 格式包含 steps 数组 用户任务{task} 输出格式 {{steps: [{{id: 1, desc: ..., capability: ...}}]}} 实测下来强制 JSON 输出是保证后续解析稳定的关键。但 LLM 有时候会在 JSON 外面包一层 markdown 代码块所以解析时要先清洗import json, re def parse_llm_json(text: str) - dict: # 去掉可能的 markdown 包裹 text re.sub(r^(?:json)?\s*, , text.strip()) text re.sub(r\s*$, , text) return json.loads(text)3.3 执行 Agent 与工具调用执行 Agent 拿到规划结果后逐步执行。工具调用是这里的核心。我用装饰器注册工具TOOL_REGISTRY {} def register_tool(name: str, description: str): def decorator(func): TOOL_REGISTRY[name] { func: func, description: description } return func return decorator register_tool(get_weather, 查询指定城市的天气) async def get_weather(city: str) - dict: # 实际调用天气 API ...执行 Agent 的 prompt 里要把可用工具的描述塞进去让 LLM 决定调哪个。这里有个经验工具描述要写得像给新人看的文档说清楚输入输出LLM 才能正确调用。我一开始写得太简略LLM 经常传错参数类型。3.4 审核 Agent 的价值与实现审核 Agent 是很多人会省掉的一环但我觉得它恰恰是多智能体相比单 Agent 最大的优势。它的职责是检查执行结果是否满足原始需求。实现上可以很简单REVIEWER_PROMPT 你是质量审核员。原始任务{task} 执行结果{result} 请判断 1. 结果是否完整回答了任务 2. 是否有明显错误或遗漏 3. 给出改进建议如果有 输出 JSON{{passed: true/false, issues: [...], suggestion: ...}} 如果passed为 false就把 suggestion 回传给执行 Agent 重做。这个循环最多跑 2-3 次避免无限重试烧 token。4. Flutter 端如何展示多智能体协作过程4.1 状态管理方案选择多智能体界面的状态比普通 App 复杂每个 Agent 有自己的状态等待、运行中、完成、失败还有全局的任务状态。我对比了三个方案方案优势劣势适用场景Provider简单直观上手快复杂状态嵌套时代码啰嗦中小型项目Riverpod编译期安全测试友好学习曲线陡中大型项目Bloc事件驱动清晰适合复杂流程样板代码多状态机复杂的场景我最后选了Riverpod因为多 Agent 场景下每个 Agent 的状态可以独立成 provider互不干扰而且ref.watch的依赖追踪让 UI 更新很精准。4.2 流式响应的处理LLM 的流式返回是体验的关键。FastAPI 这边用StreamingResponsefrom fastapi.responses import StreamingResponse app.post(/agent/run) async def run_agent(request: AgentRequest): async def event_generator(): async for chunk in orchestrator.run_stream(request.task): yield fdata: {json.dumps(chunk)}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream )Flutter 这边用 dio 接收流final response await dio.post( /agent/run, data: {task: task}, options: Options(responseType: ResponseType.stream), ); response.data.stream.listen((chunk) { final text utf8.decode(chunk); // 解析 SSE 格式更新对应 Agent 的状态 });这里有个坑SSE 的数据可能被 TCP 分包一个完整的data: {...}可能分两次到达。所以不能假设每次 listen 拿到的都是完整消息要维护一个 buffer按\n\n分割。4.3 多 Agent 状态的可视化界面上我用了一个纵向的时间线每个 Agent 是一个卡片卡片状态随执行进度变化。规划 Agent 完成后展开显示拆解的步骤执行 Agent 每完成一步就更新审核 Agent 最后给出结论。这种可视化让用户能清楚看到系统在干什么而不是干等一个 loading。实现上每个 Agent 卡片监听自己的 providerclass AgentCard extends ConsumerWidget { final String agentId; override Widget build(BuildContext context, WidgetRef ref) { final state ref.watch(agentStateProvider(agentId)); return Card( child: Column( children: [ Text(state.name), _buildStatusIndicator(state.status), if (state.output ! null) Text(state.output!), ], ), ); } }5. 踩坑实录那些文档不会告诉你的事5.1 Flutter 页面切换导致状态丢失这个坑我卡了大半天。现象是用户从对话页切到设置页再切回来Agent 的执行状态全没了。原因是Navigator.push后原页面被 disposeRiverpod 的 provider 如果没设置keepAlive状态会被回收。解决办法有两种。一是给关键 provider 加keepAlivefinal agentStateProvider StateNotifierProvider.autoDispose .familyAgentNotifier, AgentState, String((ref, id) { ref.keepAlive(); // 关键 return AgentNotifier(id); });二是用IndexedStack代替页面切换让所有页面常驻。我最后用的是第一种因为内存占用更可控。5.2 LLM 返回格式不稳定即使你在 prompt 里千叮咛万嘱咐要输出 JSONLLM 还是可能给你返回带解释文字的、带 markdown 的、甚至字段名拼错的 JSON。我的应对策略是三层防护第一层prompt 里给明确的 schema 示例第二层解析失败时用正则提取 JSON 部分第三层还是失败就调用一次 LLM 让它把上面的内容转成合法 JSON。实测下来加了第三层兜底后解析成功率从 85% 提到了接近 100%。代价是偶尔多一次 LLM 调用但比整个流程崩掉划算。5.3 并发请求下的 Agent 状态串扰多用户同时使用时如果 Agent 实例是全局单例状态会串。我一开始就是把 Agent 定义成模块级变量结果两个用户的任务互相污染。正确做法是每个请求创建独立的 Agent 实例或者用 session_id 隔离状态class Orchestrator: def __init__(self): self.sessions {} # session_id - agents def get_agents(self, session_id: str): if session_id not in self.sessions: self.sessions[session_id] { planner: PlannerAgent(...), executor: ExecutorAgent(...), reviewer: ReviewerAgent(...), } return self.sessions[session_id]生产环境还要考虑 session 过期清理不然内存会一直涨。5.4 FastAPI 异步阻塞问题FastAPI 是异步框架但如果你在 async 函数里调用了同步的阻塞代码比如requests.get整个事件循环会被卡住。我一开始用requests调 LLM 接口并发一上来响应就变得极慢。换成httpx.AsyncClient后问题解决。同理工具函数如果是 CPU 密集型的要用run_in_executor丢到线程池别阻塞事件循环。6. 多智能体协同的进阶思路6.1 从串行到并行的调度优化我最初的实现是严格串行规划 → 执行 → 审核。但很多任务里执行阶段的多个步骤之间没有依赖关系完全可以并行。比如查天气和查汇率两个步骤互不相关串行跑就是浪费时间。改成并行后用asyncio.gatherasync def execute_parallel(steps: list): tasks [execute_step(step) for step in steps] results await asyncio.gather(*tasks, return_exceptionsTrue) return results但要注意有依赖关系的步骤不能并行。所以规划 Agent 输出时最好标注步骤间的依赖调度器据此决定并行还是串行。6.2 Agent 之间的消息传递机制简单的多智能体是流水线式的A 的输出给 BB 的输出给 C。但更复杂的场景需要 Agent 之间双向通信比如执行 Agent 遇到问题可以反问规划 Agent。这时候需要一个消息总线class MessageBus: def __init__(self): self.queues {} # agent_name - asyncio.Queue async def send(self, to: str, message: dict): await self.queues[to].put(message) async def receive(self, agent_name: str): return await self.queues[agent_name].get()这个机制让 Agent 可以异步协作但也带来了死锁风险——两个 Agent 互相等待对方的消息。所以实际用的时候要加超时。6.3 成本控制与 token 优化多智能体最容易被忽视的成本是 token 消耗。三个 Agent 各跑一轮加上重试一次任务可能烧掉几万 token。我的优化手段有几个第一给每个 Agent 设置独立的 max_tokens 上限规划 Agent 不需要长篇大论第二缓存重复的工具调用结果同一个城市天气查两次没必要第三审核 Agent 只在必要时触发简单任务跳过审核。还有一个技巧是用小模型做规划大模型做执行。规划任务相对简单小模型够用执行和生成需要更强的能力才用大模型。这样成本能降一半以上。6.4 可观测性日志与追踪多智能体系统出问题时最难的是定位是哪个 Agent 的哪一步出了错。我的做法是给每个 Agent 的每次调用打上 trace_id所有日志带上这个 id然后用结构化日志输出import structlog logger structlog.get_logger() logger.info(agent_start, trace_idtrace_id, agentself.name, tasktask[:100])这样出问题时用 trace_id 一搜就能看到完整的调用链路。生产环境可以接入 LangSmith 或类似的可观测性平台可视化效果更好。7. 我在这套系统里的一些个人取舍关于 Flutter 和 FastAPI 的通信方式我最终选了 SSE 而不是 WebSocket。原因是多智能体场景下通信模式基本是客户端发一次请求服务端持续推送单向流就够了SSE 实现更简单而且能自动重连。WebSocket 的双向能力在这里用不上反而增加了连接管理的复杂度。关于 Agent 框架的选择我试过 LangChain 和 LangGraph。LangChain 的链式调用适合简单流程但多智能体的状态管理用 LangGraph 更自然它把 Agent 协作建模成图节点是 Agent边是消息传递调试时能可视化整个流程。不过 LangGraph 的学习成本不低如果只是两三个 Agent 的简单协作自己手写调度器反而更可控。关于记忆机制我目前用的是简单的滑动窗口但实际用下来发现关键信息应该被钉住而不是随窗口滑走。比如用户一开始说的核心需求后面几轮对话都不该丢。下一步我打算引入向量检索把历史消息存进向量库每次只召回相关的几条这样既省 token 又不丢关键信息。最后分享一个调试技巧多智能体系统开发阶段我会加一个单步模式每个 Agent 执行完就暂停等我在界面上点继续才走下一步。这样能清楚地看到每一步的输入输出定位问题比看日志快得多。上线前把这个模式关掉就行。