1. 为什么要在 Flutter 里塞进一个多智能体系统移动端做 AI 应用最容易踩的坑就是把它做成一个“套壳聊天框”。用户输入一句话App 转发给大模型拿到回复显示出来完事。这种结构在 Demo 阶段看着挺唬人一旦进入真实业务场景就立刻露馅模型不知道当前页面是什么状态、不知道用户历史操作、更没法在多个任务之间做协调。你让它“帮我订明天下午的会议室”它只能回你一句“好的请告诉我会议室名称”因为它根本没有能力去调用日历、查询空闲、发起确认这一连串动作。多智能体架构解决的正是这个问题。它的核心思路不是让一个大模型包办所有事而是把复杂任务拆成若干职责单一的 Agent每个 Agent 有自己的系统提示词、工具集和上下文边界再由一个调度层决定谁在什么时候出场。Flutter 负责的是前端交互和状态呈现FastAPI 负责后端 Agent 编排与大模型调用两者通过 HTTP 或流式接口通信。这套组合的好处在于前端保持轻量和跨平台后端可以独立迭代 Agent 逻辑不用每次改提示词都重新发版。我最初接触这个方向是因为一个内部工具的需求——要在移动端做一个能查数据、能生成报表、能根据自然语言调整筛选条件的助手。如果只用一个 Agent提示词会膨胀到三四千字模型经常顾此失彼拆成“意图识别 Agent”“数据查询 Agent”“报表生成 Agent”之后每个提示词控制在八百字以内准确率肉眼可见地提升。这也是我后来把 Flutter 和 FastAPI 这套组合固定下来的原因。这篇文章适合两类人一是已经会写 Flutter 页面、想往 AI 应用方向走的移动端开发者二是熟悉 Python 后端、想补上移动端交互这一环的 AI 工程师。我会从整体架构讲到具体代码把 Agent 编排、流式输出、状态同步这些关键环节都拆开说清楚尽量让每一步都能直接抄作业。2. 整体架构设计与技术选型思路2.1 三层结构UI 层、编排层、模型层整套系统我习惯分成三层来看。最上面是 Flutter 的 UI 层负责对话界面、Agent 状态指示、工具调用结果的可视化。中间是 FastAPI 的编排层它接收前端请求决定激活哪些 Agent管理对话历史和工具调用循环。最下面是模型层可以是云端 API也可以是本地部署的模型服务。这么分层的直接好处是职责清晰。UI 层不需要知道模型是哪个厂商的编排层不需要关心界面怎么渲染。我见过有人把模型调用直接写在 Flutter 里用 Dart 的 HTTP 客户端去请求短期看省了一个后端但一旦要做多 Agent 协调、要做工具调用的循环、要管理会话状态Dart 这边写起来会非常别扭。Python 生态里有 LangChain、LangGraph 这些成熟工具编排逻辑用 Python 写效率高得多。提示如果你的项目只是单轮问答确实可以省掉后端。但只要涉及多轮工具调用或者多 Agent 协作强烈建议把编排层独立出来。2.2 为什么选 FastAPI 而不是 Flask 或 DjangoFastAPI 在这个场景里有三个实打实的优势。第一是原生异步支持Agent 编排过程中大量时间花在等待模型响应和工具执行上异步能让多个 Agent 的调用并行起来。第二是自动生成的接口文档前后端联调时 Flutter 这边直接看/docs就能知道请求体长什么样省掉大量沟通成本。第三是 Pydantic 模型校验Agent 之间传递的结构化数据用 Pydantic 定义之后类型错误在入口就被拦住了不会等到模型解析那一步才报错。Flask 做同步接口没问题但异步支持要靠扩展写起来不够自然。Django 太重ORM 和 Admin 这些在这个场景里用不上反而增加启动开销。FastAPI 的轻量和异步特性刚好匹配 Agent 编排的需求。2.3 Flutter 端的定位与通信方式选择Flutter 在这套架构里不是简单的“显示器”。它需要处理流式输出的逐字渲染、Agent 执行状态的实时更新、工具调用结果的卡片式展示。这些都需要前端有足够的状态管理能力。通信方式上普通请求用 HTTP POST 就够了但流式输出必须用 SSEServer-Sent Events或者 WebSocket。我选 SSE 的原因是它实现简单FastAPI 这边用StreamingResponse就能推Flutter 这边用http包的流式响应就能接。WebSocket 虽然双向通信更灵活但在这个场景里前端不需要主动推消息给后端SSE 足够用而且断线重连的逻辑更简单。通信方式适用场景实现复杂度本项目选择HTTP POST单轮请求响应低用于非流式接口SSE服务端单向流式推送中用于对话流式输出WebSocket双向实时通信高暂不使用2.4 多智能体的职责划分原则Agent 怎么拆这是整个项目里最需要想清楚的问题。我的经验是遵循“单一职责 工具边界”两个原则。单一职责是说每个 Agent 只做一类事比如“查询 Agent”只负责把自然语言转成查询条件“执行 Agent”只负责调用具体工具。工具边界是说每个 Agent 能调用的工具集要明确不要让所有 Agent 都能访问所有工具否则模型会在工具选择上浪费大量 token。具体到这个项目我设计了四个 Agent路由 Agent 负责判断用户意图并决定后续流程查询 Agent 负责把用户问题转成结构化查询参数执行 Agent 负责调用具体的数据接口总结 Agent 负责把执行结果组织成自然语言回复。这四个 Agent 的提示词各自独立工具集也各不相同。3. 核心细节解析与实操要点3.1 Agent 提示词的结构化写法提示词写得好不好直接决定 Agent 的稳定性。我踩过的坑是早期用大段自然语言描述模型经常漏掉约束条件。后来改成结构化写法把角色、能力、约束、输出格式分开写稳定性提升明显。一个查询 Agent 的提示词大概长这样QUERY_AGENT_PROMPT 你是查询参数生成器。 ## 角色 将用户的自然语言问题转换为结构化查询参数。 ## 可用字段 - date_range: 日期范围格式 YYYY-MM-DD 到 YYYY-MM-DD - category: 分类可选值为 sales, inventory, finance - limit: 返回条数默认 20 ## 约束 1. 只输出 JSON不要有任何解释文字 2. 无法确定的字段留空不要编造 3. 日期必须补全为完整格式 ## 输出格式 {date_range: , category: , limit: 20} 这种写法的好处是模型有明确的“填空”目标而不是自由发挥。实测下来结构化提示词的字段填充准确率比自然语言描述高出不少。3.2 工具调用的循环控制Agent 调用工具不是一次就结束的可能需要多轮。比如查询 Agent 生成参数后执行 Agent 调用接口发现数据为空可能需要回到查询 Agent 调整参数。这个循环必须有终止条件否则会无限循环烧 token。我的做法是设置最大循环次数同时在每轮循环后检查是否已经满足退出条件。LangGraph 在这方面提供了很好的支持用状态图的方式定义节点和边循环逻辑一目了然。from langgraph.graph import StateGraph, END def should_continue(state): if state[iteration] 5: return summarize if state[task_completed]: return summarize return execute workflow StateGraph(AgentState) workflow.add_node(route, route_agent) workflow.add_node(query, query_agent) workflow.add_node(execute, execute_agent) workflow.add_node(summarize, summarize_agent) workflow.add_conditional_edges(execute, should_continue)注意最大循环次数不要设太大3 到 5 次足够。超过这个次数还没完成说明任务本身有问题继续循环只是浪费资源。3.3 Flutter 端的状态管理方案Flutter 这边我用的是 Riverpod。选它的原因是 Agent 执行过程中状态变化频繁需要细粒度的状态更新Riverpod 的 provider 机制能精确控制哪些组件需要重建。对话状态我定义成一个类包含消息列表、当前活跃 Agent、执行进度、错误信息这几个字段。每次 SSE 收到新数据就更新对应的字段UI 自动响应。class ChatState { final ListMessage messages; final String? activeAgent; final bool isStreaming; final String? error; ChatState({ this.messages const [], this.activeAgent, this.isStreaming false, this.error, }); ChatState copyWith({...}) ChatState(...); }这里有个细节要注意流式输出时每条消息的内容是逐步增长的如果每次都重建整个消息列表性能会很差。我的做法是把正在流式输出的那条消息单独用一个 provider 管理其他历史消息不动。3.4 SSE 流式接口的实现细节FastAPI 这边实现 SSE 要用StreamingResponse媒体类型设为text/event-stream。每个事件的数据格式要遵循 SSE 规范以data:开头以两个换行结尾。from fastapi.responses import StreamingResponse async def event_generator(query: str): async for chunk in agent_orchestrator.run(query): yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n app.post(/chat/stream) async def chat_stream(request: ChatRequest): return StreamingResponse( event_generator(request.query), media_typetext/event-stream )Flutter 这边接收 SSE 用http包的send方法拿到流之后逐行解析。要注意的是 Dart 的流是按字节来的需要自己处理粘包和拆包按\n\n分割事件。4. 实操过程与核心环节实现4.1 后端项目目录结构FastAPI 项目的目录结构直接影响后续维护成本。我习惯按职责分层而不是按文件类型分。下面是我在这个项目里用的结构backend/ ├── app/ │ ├── main.py │ ├── config.py │ ├── api/ │ │ ├── __init__.py │ │ └── chat.py │ ├── agents/ │ │ ├── __init__.py │ │ ├── base.py │ │ ├── router_agent.py │ │ ├── query_agent.py │ │ ├── execute_agent.py │ │ └── summarize_agent.py │ ├── orchestrator/ │ │ ├── __init__.py │ │ └── graph.py │ ├── tools/ │ │ ├── __init__.py │ │ └── data_tools.py │ ├── models/ │ │ ├── __init__.py │ │ └── schemas.py │ └── services/ │ ├── __init__.py │ └── llm_client.py ├── requirements.txt └── .envagents目录放各个 Agent 的定义orchestrator放编排逻辑tools放工具函数services放模型客户端这类外部依赖。这样分的好处是改某个 Agent 的提示词时不会影响到其他部分工具函数也能被多个 Agent 复用。4.2 模型客户端的封装模型调用我单独封装了一层目的是把厂商相关的细节隔离掉。这样以后换模型只需要改这一个文件。class LLMClient: def __init__(self, api_key: str, base_url: str, model: str): self.client AsyncOpenAI(api_keyapi_key, base_urlbase_url) self.model model async def chat(self, messages: list, temperature: float 0.1): response await self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, ) return response.choices[0].message.content async def stream_chat(self, messages: list): stream await self.client.chat.completions.create( modelself.model, messagesmessages, streamTrue, ) async for chunk in stream: delta chunk.choices[0].delta.content if delta: yield delta温度参数我设得比较低因为 Agent 场景需要的是稳定输出而不是创意发挥。查询参数生成这类任务温度设 0.1 甚至 0 都行。4.3 编排图的完整实现编排逻辑是整个后端的核心。我用 LangGraph 定义了一个状态图节点是各个 Agent边是流转条件。class AgentState(TypedDict): query: str intent: str query_params: dict tool_result: dict final_answer: str iteration: int task_completed: bool def build_graph(): graph StateGraph(AgentState) graph.add_node(route, route_agent) graph.add_node(query, query_agent) graph.add_node(execute, execute_agent) graph.add_node(summarize, summarize_agent) graph.set_entry_point(route) graph.add_edge(route, query) graph.add_edge(query, execute) graph.add_conditional_edges( execute, should_continue, { query: query, summarize: summarize, } ) graph.add_edge(summarize, END) return graph.compile()should_continue这个函数决定执行完工具之后是回到查询节点调整参数还是直接进入总结节点。判断依据是工具返回结果是否为空、是否达到最大迭代次数。4.4 Flutter 端 SSE 接收与渲染Flutter 这边接收 SSE 的代码需要处理几个细节。首先是请求头要设置Accept: text/event-stream其次是响应流要按事件分割。StreamString streamChat(String query) async* { final request http.Request( POST, Uri.parse($baseUrl/chat/stream), ); request.headers[Content-Type] application/json; request.headers[Accept] text/event-stream; request.body jsonEncode({query: query}); final response await http.Client().send(request); final stream response.stream.transform(utf8.decoder); String buffer ; await for (final chunk in stream) { buffer chunk; while (buffer.contains(\n\n)) { final index buffer.indexOf(\n\n); final event buffer.substring(0, index); buffer buffer.substring(index 2); if (event.startsWith(data: )) { yield event.substring(6); } } } }拿到数据之后解析 JSON根据事件类型更新对应的状态。事件类型我定义了三种agent_start表示某个 Agent 开始执行token表示流式输出的一个片段agent_end表示 Agent 执行结束。4.5 工具调用的具体实现工具函数我统一放在tools目录下每个工具用装饰器注册这样 Agent 可以通过名称查找。TOOL_REGISTRY {} def register_tool(name: str, description: str): def decorator(func): TOOL_REGISTRY[name] { function: func, description: description, } return func return decorator register_tool(query_sales_data, 查询销售数据) async def query_sales_data(date_range: str, category: str, limit: int 20): # 实际的数据查询逻辑 return {records: [...], total: 100}Agent 在生成工具调用请求时需要把可用工具的描述一起传给模型。模型返回工具名称和参数后编排层从注册表里找到对应函数执行。提示工具描述要写得具体包括参数含义和返回值格式。模型对工具描述的理解程度直接影响调用准确率。5. 常见问题与排查技巧实录5.1 Agent 不按预期调用工具怎么办这是最常见的问题。模型要么不调用工具直接编答案要么调用了错误的工具。排查思路分三步走。第一步检查工具描述是否清晰。如果描述太笼统模型无法判断什么时候该用这个工具。比如“查询数据”这种描述就不如“根据日期范围和分类查询销售记录返回记录列表和总数”来得明确。第二步检查提示词里有没有强制要求。在系统提示词里明确写“你必须先调用工具获取数据不允许直接回答”能显著降低模型偷懒的概率。第三步检查模型能力。有些小参数模型对工具调用的支持不好换一个工具调用能力强的模型往往能直接解决问题。5.2 流式输出中断或乱序SSE 流式输出偶尔会出现中断原因可能是网络波动也可能是后端生成器抛异常了。前端这边要做好断线处理收到不完整的事件时不要直接解析等下一个换行符到了再处理。后端这边要在生成器里加异常捕获出错时发送一个error类型的事件而不是直接断开连接。这样前端能知道发生了什么而不是一脸茫然地等在那里。async def event_generator(query: str): try: async for chunk in agent_orchestrator.run(query): yield fdata: {json.dumps(chunk)}\n\n except Exception as e: error_event {type: error, message: str(e)} yield fdata: {json.dumps(error_event)}\n\n5.3 多轮对话上下文丢失多轮对话需要维护历史消息。我的做法是在后端维护会话状态每个会话有一个 ID前端每次请求带上这个 ID。后端根据 ID 取出历史消息拼接到当前请求前面。要注意的是历史消息不能无限增长否则 token 消耗会失控。我的策略是保留最近十轮对话更早的做摘要压缩。摘要用一个单独的轻量模型生成成本很低。问题现象可能原因排查方向解决方案Agent 不调用工具工具描述不清检查工具描述补充参数和返回值说明流式输出中断后端异常查看后端日志生成器加异常捕获上下文丢失会话 ID 未传递检查请求头前端持久化会话 ID循环次数过多退出条件不明确检查 should_continue增加完成判断逻辑响应速度慢串行调用检查编排图无依赖节点并行执行5.4 性能优化的几个实操点Agent 编排最容易出现的性能问题是串行等待。路由 Agent 判断完意图之后查询 Agent 和执行 Agent 如果有依赖关系那必须串行但如果没有依赖完全可以并行。另一个优化点是模型选择。路由和总结这类任务用轻量模型就够了查询参数生成需要一定的理解能力可以用中等模型只有复杂推理才需要上大模型。混合使用能显著降低成本。还有一点是缓存。相同的查询参数对应的工具结果可以缓存一段时间避免重复调用。我用的是内存缓存加过期时间简单够用。5.5 Flutter 端渲染性能问题流式输出时如果每个 token 都触发一次 setState界面会卡。解决办法是用ValueNotifier或者 Riverpod 的StateNotifier只更新变化的部分。另外文本内容增长时用SelectableText.rich配合TextSpan增量更新比每次重建整个 Text 组件要流畅得多。列表滚动也要注意。消息多的时候用ListView.builder懒加载不要一次性构建所有消息组件。如果消息里有 Markdown 内容渲染开销更大可以考虑用flutter_markdown的懒渲染模式。6. 从能跑到好用几个值得投入的改进方向项目能跑起来只是第一步真正投入使用还需要在几个地方下功夫。第一个是错误恢复。Agent 执行过程中任何一步失败都应该有降级方案。比如查询 Agent 生成的参数不合法不应该直接报错给用户而是让执行 Agent 返回一个友好的提示同时记录日志供后续分析。第二个是可观测性。每个 Agent 的输入输出、耗时、token 消耗都要记录下来。我用的是结构化日志每条日志包含会话 ID、Agent 名称、耗时、状态。出问题时能快速定位是哪个环节拖了后腿。第三个是提示词版本管理。Agent 的提示词会随着使用不断调整每次调整都应该有记录方便回滚和对比效果。我习惯把提示词放在单独的配置文件里用版本号标记改之前先备份。这套架构我从最初的原型到现在稳定运行前后迭代了大概十几个版本。最大的体会是多智能体的难点不在模型本身而在编排逻辑和状态管理。把这两块理顺了剩下的都是工程问题。Flutter 和 FastAPI 这个组合在移动端 AI 应用场景里算是比较顺手的选择前端跨平台省事后端 Python 生态丰富两边通过清晰的接口约定各司其职后续扩展新 Agent 或者换模型都不会伤筋动骨。 
长河君 · 2026/10/15 19:53:58 · 阅读 1.2k Flutter+FastAPI多智能体架构实战：移动端AI应用开发指南
1. 为什么要在 Flutter 里塞进一个多智能体系统移动端做 AI 应用最容易踩的坑就是把它做成一个“套壳聊天框”。用户输入一句话App 转发给大模型拿到回复显示出来完事。这种结构在 Demo 阶段看着挺唬人一旦进入真实业务场景就立刻露馅模型不知道当前页面是什么状态、不知道用户历史操作、更没法在多个任务之间做协调。你让它“帮我订明天下午的会议室”它只能回你一句“好的请告诉我会议室名称”因为它根本没有能力去调用日历、查询空闲、发起确认这一连串动作。多智能体架构解决的正是这个问题。它的核心思路不是让一个大模型包办所有事而是把复杂任务拆成若干职责单一的 Agent每个 Agent 有自己的系统提示词、工具集和上下文边界再由一个调度层决定谁在什么时候出场。Flutter 负责的是前端交互和状态呈现FastAPI 负责后端 Agent 编排与大模型调用两者通过 HTTP 或流式接口通信。这套组合的好处在于前端保持轻量和跨平台后端可以独立迭代 Agent 逻辑不用每次改提示词都重新发版。我最初接触这个方向是因为一个内部工具的需求——要在移动端做一个能查数据、能生成报表、能根据自然语言调整筛选条件的助手。如果只用一个 Agent提示词会膨胀到三四千字模型经常顾此失彼拆成“意图识别 Agent”“数据查询 Agent”“报表生成 Agent”之后每个提示词控制在八百字以内准确率肉眼可见地提升。这也是我后来把 Flutter 和 FastAPI 这套组合固定下来的原因。这篇文章适合两类人一是已经会写 Flutter 页面、想往 AI 应用方向走的移动端开发者二是熟悉 Python 后端、想补上移动端交互这一环的 AI 工程师。我会从整体架构讲到具体代码把 Agent 编排、流式输出、状态同步这些关键环节都拆开说清楚尽量让每一步都能直接抄作业。2. 整体架构设计与技术选型思路2.1 三层结构UI 层、编排层、模型层整套系统我习惯分成三层来看。最上面是 Flutter 的 UI 层负责对话界面、Agent 状态指示、工具调用结果的可视化。中间是 FastAPI 的编排层它接收前端请求决定激活哪些 Agent管理对话历史和工具调用循环。最下面是模型层可以是云端 API也可以是本地部署的模型服务。这么分层的直接好处是职责清晰。UI 层不需要知道模型是哪个厂商的编排层不需要关心界面怎么渲染。我见过有人把模型调用直接写在 Flutter 里用 Dart 的 HTTP 客户端去请求短期看省了一个后端但一旦要做多 Agent 协调、要做工具调用的循环、要管理会话状态Dart 这边写起来会非常别扭。Python 生态里有 LangChain、LangGraph 这些成熟工具编排逻辑用 Python 写效率高得多。提示如果你的项目只是单轮问答确实可以省掉后端。但只要涉及多轮工具调用或者多 Agent 协作强烈建议把编排层独立出来。2.2 为什么选 FastAPI 而不是 Flask 或 DjangoFastAPI 在这个场景里有三个实打实的优势。第一是原生异步支持Agent 编排过程中大量时间花在等待模型响应和工具执行上异步能让多个 Agent 的调用并行起来。第二是自动生成的接口文档前后端联调时 Flutter 这边直接看/docs就能知道请求体长什么样省掉大量沟通成本。第三是 Pydantic 模型校验Agent 之间传递的结构化数据用 Pydantic 定义之后类型错误在入口就被拦住了不会等到模型解析那一步才报错。Flask 做同步接口没问题但异步支持要靠扩展写起来不够自然。Django 太重ORM 和 Admin 这些在这个场景里用不上反而增加启动开销。FastAPI 的轻量和异步特性刚好匹配 Agent 编排的需求。2.3 Flutter 端的定位与通信方式选择Flutter 在这套架构里不是简单的“显示器”。它需要处理流式输出的逐字渲染、Agent 执行状态的实时更新、工具调用结果的卡片式展示。这些都需要前端有足够的状态管理能力。通信方式上普通请求用 HTTP POST 就够了但流式输出必须用 SSEServer-Sent Events或者 WebSocket。我选 SSE 的原因是它实现简单FastAPI 这边用StreamingResponse就能推Flutter 这边用http包的流式响应就能接。WebSocket 虽然双向通信更灵活但在这个场景里前端不需要主动推消息给后端SSE 足够用而且断线重连的逻辑更简单。通信方式适用场景实现复杂度本项目选择HTTP POST单轮请求响应低用于非流式接口SSE服务端单向流式推送中用于对话流式输出WebSocket双向实时通信高暂不使用2.4 多智能体的职责划分原则Agent 怎么拆这是整个项目里最需要想清楚的问题。我的经验是遵循“单一职责 工具边界”两个原则。单一职责是说每个 Agent 只做一类事比如“查询 Agent”只负责把自然语言转成查询条件“执行 Agent”只负责调用具体工具。工具边界是说每个 Agent 能调用的工具集要明确不要让所有 Agent 都能访问所有工具否则模型会在工具选择上浪费大量 token。具体到这个项目我设计了四个 Agent路由 Agent 负责判断用户意图并决定后续流程查询 Agent 负责把用户问题转成结构化查询参数执行 Agent 负责调用具体的数据接口总结 Agent 负责把执行结果组织成自然语言回复。这四个 Agent 的提示词各自独立工具集也各不相同。3. 核心细节解析与实操要点3.1 Agent 提示词的结构化写法提示词写得好不好直接决定 Agent 的稳定性。我踩过的坑是早期用大段自然语言描述模型经常漏掉约束条件。后来改成结构化写法把角色、能力、约束、输出格式分开写稳定性提升明显。一个查询 Agent 的提示词大概长这样QUERY_AGENT_PROMPT 你是查询参数生成器。 ## 角色 将用户的自然语言问题转换为结构化查询参数。 ## 可用字段 - date_range: 日期范围格式 YYYY-MM-DD 到 YYYY-MM-DD - category: 分类可选值为 sales, inventory, finance - limit: 返回条数默认 20 ## 约束 1. 只输出 JSON不要有任何解释文字 2. 无法确定的字段留空不要编造 3. 日期必须补全为完整格式 ## 输出格式 {date_range: , category: , limit: 20} 这种写法的好处是模型有明确的“填空”目标而不是自由发挥。实测下来结构化提示词的字段填充准确率比自然语言描述高出不少。3.2 工具调用的循环控制Agent 调用工具不是一次就结束的可能需要多轮。比如查询 Agent 生成参数后执行 Agent 调用接口发现数据为空可能需要回到查询 Agent 调整参数。这个循环必须有终止条件否则会无限循环烧 token。我的做法是设置最大循环次数同时在每轮循环后检查是否已经满足退出条件。LangGraph 在这方面提供了很好的支持用状态图的方式定义节点和边循环逻辑一目了然。from langgraph.graph import StateGraph, END def should_continue(state): if state[iteration] 5: return summarize if state[task_completed]: return summarize return execute workflow StateGraph(AgentState) workflow.add_node(route, route_agent) workflow.add_node(query, query_agent) workflow.add_node(execute, execute_agent) workflow.add_node(summarize, summarize_agent) workflow.add_conditional_edges(execute, should_continue)注意最大循环次数不要设太大3 到 5 次足够。超过这个次数还没完成说明任务本身有问题继续循环只是浪费资源。3.3 Flutter 端的状态管理方案Flutter 这边我用的是 Riverpod。选它的原因是 Agent 执行过程中状态变化频繁需要细粒度的状态更新Riverpod 的 provider 机制能精确控制哪些组件需要重建。对话状态我定义成一个类包含消息列表、当前活跃 Agent、执行进度、错误信息这几个字段。每次 SSE 收到新数据就更新对应的字段UI 自动响应。class ChatState { final ListMessage messages; final String? activeAgent; final bool isStreaming; final String? error; ChatState({ this.messages const [], this.activeAgent, this.isStreaming false, this.error, }); ChatState copyWith({...}) ChatState(...); }这里有个细节要注意流式输出时每条消息的内容是逐步增长的如果每次都重建整个消息列表性能会很差。我的做法是把正在流式输出的那条消息单独用一个 provider 管理其他历史消息不动。3.4 SSE 流式接口的实现细节FastAPI 这边实现 SSE 要用StreamingResponse媒体类型设为text/event-stream。每个事件的数据格式要遵循 SSE 规范以data:开头以两个换行结尾。from fastapi.responses import StreamingResponse async def event_generator(query: str): async for chunk in agent_orchestrator.run(query): yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n app.post(/chat/stream) async def chat_stream(request: ChatRequest): return StreamingResponse( event_generator(request.query), media_typetext/event-stream )Flutter 这边接收 SSE 用http包的send方法拿到流之后逐行解析。要注意的是 Dart 的流是按字节来的需要自己处理粘包和拆包按\n\n分割事件。4. 实操过程与核心环节实现4.1 后端项目目录结构FastAPI 项目的目录结构直接影响后续维护成本。我习惯按职责分层而不是按文件类型分。下面是我在这个项目里用的结构backend/ ├── app/ │ ├── main.py │ ├── config.py │ ├── api/ │ │ ├── __init__.py │ │ └── chat.py │ ├── agents/ │ │ ├── __init__.py │ │ ├── base.py │ │ ├── router_agent.py │ │ ├── query_agent.py │ │ ├── execute_agent.py │ │ └── summarize_agent.py │ ├── orchestrator/ │ │ ├── __init__.py │ │ └── graph.py │ ├── tools/ │ │ ├── __init__.py │ │ └── data_tools.py │ ├── models/ │ │ ├── __init__.py │ │ └── schemas.py │ └── services/ │ ├── __init__.py │ └── llm_client.py ├── requirements.txt └── .envagents目录放各个 Agent 的定义orchestrator放编排逻辑tools放工具函数services放模型客户端这类外部依赖。这样分的好处是改某个 Agent 的提示词时不会影响到其他部分工具函数也能被多个 Agent 复用。4.2 模型客户端的封装模型调用我单独封装了一层目的是把厂商相关的细节隔离掉。这样以后换模型只需要改这一个文件。class LLMClient: def __init__(self, api_key: str, base_url: str, model: str): self.client AsyncOpenAI(api_keyapi_key, base_urlbase_url) self.model model async def chat(self, messages: list, temperature: float 0.1): response await self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, ) return response.choices[0].message.content async def stream_chat(self, messages: list): stream await self.client.chat.completions.create( modelself.model, messagesmessages, streamTrue, ) async for chunk in stream: delta chunk.choices[0].delta.content if delta: yield delta温度参数我设得比较低因为 Agent 场景需要的是稳定输出而不是创意发挥。查询参数生成这类任务温度设 0.1 甚至 0 都行。4.3 编排图的完整实现编排逻辑是整个后端的核心。我用 LangGraph 定义了一个状态图节点是各个 Agent边是流转条件。class AgentState(TypedDict): query: str intent: str query_params: dict tool_result: dict final_answer: str iteration: int task_completed: bool def build_graph(): graph StateGraph(AgentState) graph.add_node(route, route_agent) graph.add_node(query, query_agent) graph.add_node(execute, execute_agent) graph.add_node(summarize, summarize_agent) graph.set_entry_point(route) graph.add_edge(route, query) graph.add_edge(query, execute) graph.add_conditional_edges( execute, should_continue, { query: query, summarize: summarize, } ) graph.add_edge(summarize, END) return graph.compile()should_continue这个函数决定执行完工具之后是回到查询节点调整参数还是直接进入总结节点。判断依据是工具返回结果是否为空、是否达到最大迭代次数。4.4 Flutter 端 SSE 接收与渲染Flutter 这边接收 SSE 的代码需要处理几个细节。首先是请求头要设置Accept: text/event-stream其次是响应流要按事件分割。StreamString streamChat(String query) async* { final request http.Request( POST, Uri.parse($baseUrl/chat/stream), ); request.headers[Content-Type] application/json; request.headers[Accept] text/event-stream; request.body jsonEncode({query: query}); final response await http.Client().send(request); final stream response.stream.transform(utf8.decoder); String buffer ; await for (final chunk in stream) { buffer chunk; while (buffer.contains(\n\n)) { final index buffer.indexOf(\n\n); final event buffer.substring(0, index); buffer buffer.substring(index 2); if (event.startsWith(data: )) { yield event.substring(6); } } } }拿到数据之后解析 JSON根据事件类型更新对应的状态。事件类型我定义了三种agent_start表示某个 Agent 开始执行token表示流式输出的一个片段agent_end表示 Agent 执行结束。4.5 工具调用的具体实现工具函数我统一放在tools目录下每个工具用装饰器注册这样 Agent 可以通过名称查找。TOOL_REGISTRY {} def register_tool(name: str, description: str): def decorator(func): TOOL_REGISTRY[name] { function: func, description: description, } return func return decorator register_tool(query_sales_data, 查询销售数据) async def query_sales_data(date_range: str, category: str, limit: int 20): # 实际的数据查询逻辑 return {records: [...], total: 100}Agent 在生成工具调用请求时需要把可用工具的描述一起传给模型。模型返回工具名称和参数后编排层从注册表里找到对应函数执行。提示工具描述要写得具体包括参数含义和返回值格式。模型对工具描述的理解程度直接影响调用准确率。5. 常见问题与排查技巧实录5.1 Agent 不按预期调用工具怎么办这是最常见的问题。模型要么不调用工具直接编答案要么调用了错误的工具。排查思路分三步走。第一步检查工具描述是否清晰。如果描述太笼统模型无法判断什么时候该用这个工具。比如“查询数据”这种描述就不如“根据日期范围和分类查询销售记录返回记录列表和总数”来得明确。第二步检查提示词里有没有强制要求。在系统提示词里明确写“你必须先调用工具获取数据不允许直接回答”能显著降低模型偷懒的概率。第三步检查模型能力。有些小参数模型对工具调用的支持不好换一个工具调用能力强的模型往往能直接解决问题。5.2 流式输出中断或乱序SSE 流式输出偶尔会出现中断原因可能是网络波动也可能是后端生成器抛异常了。前端这边要做好断线处理收到不完整的事件时不要直接解析等下一个换行符到了再处理。后端这边要在生成器里加异常捕获出错时发送一个error类型的事件而不是直接断开连接。这样前端能知道发生了什么而不是一脸茫然地等在那里。async def event_generator(query: str): try: async for chunk in agent_orchestrator.run(query): yield fdata: {json.dumps(chunk)}\n\n except Exception as e: error_event {type: error, message: str(e)} yield fdata: {json.dumps(error_event)}\n\n5.3 多轮对话上下文丢失多轮对话需要维护历史消息。我的做法是在后端维护会话状态每个会话有一个 ID前端每次请求带上这个 ID。后端根据 ID 取出历史消息拼接到当前请求前面。要注意的是历史消息不能无限增长否则 token 消耗会失控。我的策略是保留最近十轮对话更早的做摘要压缩。摘要用一个单独的轻量模型生成成本很低。问题现象可能原因排查方向解决方案Agent 不调用工具工具描述不清检查工具描述补充参数和返回值说明流式输出中断后端异常查看后端日志生成器加异常捕获上下文丢失会话 ID 未传递检查请求头前端持久化会话 ID循环次数过多退出条件不明确检查 should_continue增加完成判断逻辑响应速度慢串行调用检查编排图无依赖节点并行执行5.4 性能优化的几个实操点Agent 编排最容易出现的性能问题是串行等待。路由 Agent 判断完意图之后查询 Agent 和执行 Agent 如果有依赖关系那必须串行但如果没有依赖完全可以并行。另一个优化点是模型选择。路由和总结这类任务用轻量模型就够了查询参数生成需要一定的理解能力可以用中等模型只有复杂推理才需要上大模型。混合使用能显著降低成本。还有一点是缓存。相同的查询参数对应的工具结果可以缓存一段时间避免重复调用。我用的是内存缓存加过期时间简单够用。5.5 Flutter 端渲染性能问题流式输出时如果每个 token 都触发一次 setState界面会卡。解决办法是用ValueNotifier或者 Riverpod 的StateNotifier只更新变化的部分。另外文本内容增长时用SelectableText.rich配合TextSpan增量更新比每次重建整个 Text 组件要流畅得多。列表滚动也要注意。消息多的时候用ListView.builder懒加载不要一次性构建所有消息组件。如果消息里有 Markdown 内容渲染开销更大可以考虑用flutter_markdown的懒渲染模式。6. 从能跑到好用几个值得投入的改进方向项目能跑起来只是第一步真正投入使用还需要在几个地方下功夫。第一个是错误恢复。Agent 执行过程中任何一步失败都应该有降级方案。比如查询 Agent 生成的参数不合法不应该直接报错给用户而是让执行 Agent 返回一个友好的提示同时记录日志供后续分析。第二个是可观测性。每个 Agent 的输入输出、耗时、token 消耗都要记录下来。我用的是结构化日志每条日志包含会话 ID、Agent 名称、耗时、状态。出问题时能快速定位是哪个环节拖了后腿。第三个是提示词版本管理。Agent 的提示词会随着使用不断调整每次调整都应该有记录方便回滚和对比效果。我习惯把提示词放在单独的配置文件里用版本号标记改之前先备份。这套架构我从最初的原型到现在稳定运行前后迭代了大概十几个版本。最大的体会是多智能体的难点不在模型本身而在编排逻辑和状态管理。把这两块理顺了剩下的都是工程问题。Flutter 和 FastAPI 这个组合在移动端 AI 应用场景里算是比较顺手的选择前端跨平台省事后端 Python 生态丰富两边通过清晰的接口约定各司其职后续扩展新 Agent 或者换模型都不会伤筋动骨。