ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Flutter + LLM 多智能体实战:FastAPI 编排与流式输出

Flutter + LLM 多智能体实战:FastAPI 编排与流式输出 1. 为什么我选择 Flutter LLM 多智能体这条技术路线移动端做 AI 应用最容易踩的坑就是“前端演示很惊艳后端一上并发就崩”。我去年帮一个团队做智能客服的移动端原型最初用纯 Flutter 调云端大模型接口单机测试没问题十个人同时用就开始出现请求堆积、界面卡死、Token 消耗失控。后来把架构改成 Flutter 负责交互层、FastAPI 负责 Agent 编排层、LLM 负责推理层整个系统才真正稳下来。这套 Flutter LLM 多智能体实战入门的路线就是从那几次翻车经历里总结出来的。这篇文章面向的是有一定 Flutter 基础、想往 AI 应用方向走的移动端开发者也适合后端同学了解移动端 Agent 的交互特点。核心讲清楚三件事多智能体在移动端到底怎么分工、FastAPI 怎么做 Agent 的调度中枢、Flutter 怎么把流式输出和多轮状态管理做扎实。我不会只讲概念每个环节都会给出可复现的目录结构、关键代码和参数选择依据。先明确一个认知多智能体不是“多个模型同时说话”而是“多个职责单一的 Agent 通过消息传递协作完成一个任务”。在移动端场景里这个区别尤其重要因为手机算力和网络都不稳定Agent 数量一旦失控用户体验会断崖式下跌。我实测下来移动端 3 到 5 个 Agent 是比较舒服的区间超过 7 个协调开销就会吃掉大部分收益。2. 整体架构设计与技术选型逻辑2.1 三层架构的职责边界怎么划这套系统我采用的是清晰的三层结构。Flutter 层只做三件事收集用户输入、展示 Agent 执行过程、管理本地会话状态。它不直接调用 LLM也不做任何 Prompt 拼接。FastAPI 层是调度中枢负责 Agent 注册、任务分发、上下文管理和流式转发。LLM 层可以是本地模型也可以是云端 API通过统一接口封装方便替换。为什么不让 Flutter 直接调 LLM我试过问题很明显。第一API Key 放在客户端等于裸奔安全上过不去。第二移动端网络抖动频繁直连大模型容易出现半截响应状态很难恢复。第三多智能体的协调逻辑放在客户端iOS 和 Android 要各写一遍维护成本翻倍。把编排放在 FastAPIFlutter 只做展示职责清晰后续换模型、加 Agent 都不用动客户端。2.2 为什么选 FastAPI 而不是 Flask 或 DjangoFastAPI 在这个场景里有三个硬优势。原生异步支持Agent 调用 LLM 是典型的 IO 密集操作async/await 能让单进程扛住更多并发连接。自动生成 OpenAPI 文档Flutter 端可以直接根据 schema 生成请求模型减少手写解析代码。Pydantic 数据校验Agent 之间的消息格式用 Pydantic 模型定义传参错误在入口就被拦截不会带着脏数据跑到 LLM 那里。对比一下常见选择框架异步支持数据校验适合场景FastAPI原生 asyncPydantic 内置Agent 编排、流式接口Flask需扩展手动或第三方简单接口、原型Django3.1 部分支持Form/DRF全栈 Web、ORM 重我踩过的坑是早期用 Flask 写 Agent 调度每个请求开线程并发一上来线程池就满了日志里全是超时。换成 FastAPI 的 async 路由后同样的机器配置并发能力大概提升了三到四倍。这不是说 Flask 不好而是场景匹配问题。2.3 多智能体的角色划分原则移动端多智能体我建议按“输入理解、任务规划、执行、校验”四个环节来切。一个典型的入门配置是三个 AgentRouter Agent判断用户意图Executor Agent调用工具或生成内容Critic Agent检查输出质量。如果任务简单可以砍掉 Critic只留两个。这里有个关键原则每个 Agent 的 System Prompt 只描述自己的职责不要写其他 Agent 的事。我见过有人把整个流程写进每个 Agent 的 Prompt 里结果模型经常“越权”回答Router 开始干 Executor 的活。职责边界清晰协作才稳定。3. FastAPI 后端核心实现细节3.1 项目目录结构怎么组织FastAPI 项目最怕所有代码堆在 main.py 里。我推荐按职责分层下面是我实际在用的结构agent_backend/ ├── app/ │ ├── main.py # 应用入口注册路由和中间件 │ ├── config.py # 环境变量和模型配置 │ ├── models/ │ │ ├── request.py # 请求体 Pydantic 模型 │ │ └── message.py # Agent 间消息模型 │ ├── agents/ │ │ ├── base.py # Agent 基类定义统一接口 │ │ ├── router.py # 意图路由 Agent │ │ ├── executor.py # 任务执行 Agent │ │ └── critic.py # 质量校验 Agent │ ├── services/ │ │ ├── llm_client.py # LLM 调用封装 │ │ └── orchestrator.py # 多智能体协调器 │ └── routers/ │ └── chat.py # 对话接口支持流式 ├── requirements.txt └── .env这个结构的好处是加一个新 Agent 只需要在 agents 目录下新建文件在 orchestrator 里注册不用动路由层。LLM 客户端单独封装换模型只改一个文件。3.2 Agent 基类与消息协议设计所有 Agent 继承同一个基类保证接口一致。消息协议用 Pydantic 定义字段包括role、content、metadata。metadata 里放 Agent 名称、耗时、Token 用量方便后续排查。from pydantic import BaseModel from typing import Optional, Dict, Any class AgentMessage(BaseModel): role: str content: str agent_name: Optional[str] None metadata: Dict[str, Any] {} class BaseAgent: def __init__(self, name: str, system_prompt: str): self.name name self.system_prompt system_prompt async def run(self, messages: list[AgentMessage]) - AgentMessage: raise NotImplementedError这里有个细节metadata默认用空字典不要用 None。我早期用 None结果在拼接日志时频繁判空代码很啰嗦。统一用空字典取值时用.get()干净很多。3.3 流式接口的实现要点移动端最忌讳等半天没反应。FastAPI 的StreamingResponse配合 LLM 的流式输出可以让用户看到逐字生成的效果。关键点是设置正确的media_type和响应头避免中间层缓冲。from fastapi.responses import StreamingResponse router.post(/chat/stream) async def chat_stream(req: ChatRequest): async def event_generator(): async for chunk in orchestrator.run_stream(req.message): yield fdata: {chunk}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no} )X-Accel-Buffering: no这个头很关键。我遇到过本地测试流式正常部署到有反向代理的环境后变成一次性返回排查半天就是代理层开了缓冲。加上这个头问题解决。注意流式接口的异常处理要和普通接口区分。一旦开始 yieldHTTP 状态码已经发出去了后面出错只能通过发送错误事件来通知客户端不能再改状态码。4. Flutter 客户端的关键实现4.1 项目创建与依赖选择创建 Flutter 项目用flutter create agent_app然后重点加两个依赖http或dio做网络请求provider或riverpod做状态管理。我选dio加riverpod原因是 dio 对流的处理更灵活riverpod 的编译期安全能减少状态相关的运行时错误。网络层封装一个AgentApiClient统一处理 baseUrl、超时和错误映射。超时我设的是连接 10 秒、接收 60 秒。接收超时给长一点因为 LLM 生成长文本时最后一个 chunk 可能来得比较晚。4.2 流式响应的解析与 UI 更新Flutter 端接收 SSE 流用dio的ResponseType.stream然后逐行解析data:前缀。每收到一个 chunk通过 riverpod 的 notifier 更新状态UI 自动刷新。final response await dio.post( /chat/stream, data: {message: userInput}, options: Options(responseType: ResponseType.stream), ); response.data.stream.listen((chunk) { final text utf8.decode(chunk); for (final line in text.split(\n)) { if (line.startsWith(data: )) { final content line.substring(6); ref.read(chatProvider.notifier).appendContent(content); } } });这里有个容易忽略的点chunk 边界可能切断一个多字节字符。直接utf8.decode会抛异常。正确做法是用Utf8Decoder的流式模式或者维护一个缓冲区把不完整的字节留到下一个 chunk 一起解码。我一开始没处理中文偶尔出现乱码排查了很久。4.3 多轮会话的状态管理多智能体场景下会话状态不只是消息列表还要记录每个 Agent 的执行状态。我设计的状态模型包含messages对话历史、agentStatus各 Agent 当前状态、isStreaming是否正在生成。用 riverpod 的StateNotifier管理每次 Agent 状态变化都触发 UI 更新。界面上用一个横向的步骤条展示 Router、Executor、Critic 的执行进度用户能直观看到“现在轮到谁干活”。这个体验比单纯转圈好太多用户知道系统在推进不会以为卡死了。提示会话历史不要无限增长。我设的上限是最近 20 轮超过后丢弃最早的。原因是 LLM 的上下文窗口有限历史太长会导致 Token 超限而且旧消息对当前任务的参考价值递减。5. 多智能体协同的编排逻辑5.1 顺序编排与条件跳转入门阶段先用顺序编排Router 判断意图根据结果决定走哪条分支。比如用户问“帮我查天气”Router 输出intent: weatherorchestrator 就把请求转给天气 Executor如果是“帮我写一段文案”就转给文案 Executor。条件跳转用简单的 if-else 实现不要一上来就搞复杂的图结构。我见过新手直接用 LangGraph 那种状态机结果调试困难一个节点出错整个流程卡住。先用最直白的代码把流程跑通再考虑抽象。5.2 Agent 间的上下文传递Agent 之间传递的不只是用户消息还有前序 Agent 的输出。我的做法是维护一个context字典每个 Agent 执行完后把结果写进去下一个 Agent 从 context 里取自己需要的字段。context {user_input: message} router_result await router.run(context) context[intent] router_result.content executor_result await executor.run(context) context[draft] executor_result.content这样设计的好处是每个 Agent 只依赖 context 里的特定字段不关心其他 Agent 怎么实现的。加一个新 Agent 时只要约定好它读哪些字段、写哪些字段就行。5.3 超时与降级策略移动端网络不稳定每个 Agent 调用都要设超时。我的配置是单个 Agent 超时 30 秒整个流程超时 90 秒。超时后不是直接报错而是走降级如果 Executor 超时返回一个“正在处理中请稍后重试”的提示同时把任务放入后台队列重试。这里有个经验降级提示要具体。不要只说“出错了”要告诉用户“天气服务暂时不可用已为你保留查询记录”。用户对具体信息的容忍度远高于模糊报错。6. 常见问题与排查技巧实录6.1 LLM 请求失败的典型原因现象可能原因排查方法请求被拒绝schema 或 tool payload 不合法打印完整请求体对照 API 文档检查字段响应截断max_tokens 设置过小调大 max_tokens或检查是否触发长度限制超时网络或模型负载高加超时重试检查服务端日志返回空内容Prompt 格式问题简化 Prompt逐步加回条件定位我遇到最多的是 schema 不匹配。LLM 的 tool 调用对参数格式要求很严少一个必填字段就直接拒绝。解决办法是在 Pydantic 模型里把必填字段标清楚调用前先做一次本地校验。6.2 Flutter 端的典型报错flutter/runtime/dart_vm_initializer.cc相关的错误多数是原生插件初始化顺序问题。我遇到过一次是因为在main()里过早调用了需要平台通道的方法。解决办法是把这类调用放到WidgetsBinding.instance.addPostFrameCallback里等首帧渲染完再执行。另一个常见问题是 Navigator 切换页面后状态丢失。原因是页面被销毁State 跟着没了。如果会话状态需要跨页面保留用 riverpod 的全局 Provider不要放在页面级的 StatefulWidget 里。6.3 并发场景下的注意事项AI Agent 扛并发核心不是模型本身而是你的调度层。我总结了几条限制单用户并发请求数防止一个用户开多个窗口把配额打满对 LLM 调用做队列化超出并发上限的请求排队而不是直接拒绝缓存高频相同请求比如相同的意图判断结果可以复用。注意缓存要设短过期时间我一般设 5 分钟。LLM 的输出有随机性缓存太久会导致用户觉得“怎么每次回答都一样”。7. 我在这套架构上踩过的坑和最终建议第一个坑是过早引入复杂框架。一开始就想用 LangGraph 做状态机编排结果光是理解它的状态传递机制就花了两天实际业务逻辑还没写。后来退回到最朴素的 async 函数调用半天就把流程跑通了。框架是解决复杂问题的不是用来炫技的。第二个坑是忽略移动端的电量消耗。流式输出频繁触发 UI 重绘加上网络长连接手机发热明显。后来做了两个优化chunk 合并每 100 毫秒批量更新一次 UI非活跃会话自动断开连接。发热问题明显改善。第三个坑是日志丢失。用 uvicorn 跑 FastAPI默认日志配置下异步任务里的异常有时候不打印。解决办法是显式配置 logging把 Agent 执行的关键节点都打上日志包括输入、输出、耗时。排查问题时这些日志就是救命稻草。如果你刚开始做这个方向我的建议是先用两个 Agent 把“输入理解 执行”跑通Flutter 端先把流式展示做顺再逐步加 Agent 和复杂编排。不要一上来就追求多智能体的“智能”先把工程稳定性做扎实。移动端用户对卡顿和报错的容忍度比你想的要低得多。
返回列表