企业官网建设流程全解析

1. 为什么我选择 Flutter + LLM 多智能体这条技术路线

移动端做 AI 应用，最容易踩的坑就是"前端一套、后端一套、模型调用再一套"，三套东西各写各的，最后联调的时候互相甩锅。我去年接手一个智能助手类项目，最初用原生 Android 写界面、Python 写推理服务，结果光是流式输出的状态同步就折腾了两周。后来换成 Flutter 做客户端、FastAPI 做智能体编排层，整个链路才顺下来。这篇就聊聊 Flutter + LLM 多智能体这套组合到底怎么落地，适合谁看，以及我踩过的那些坑。

先说清楚这套东西是什么。Flutter 负责跨端 UI 和交互，LLM 是底层大模型能力，多智能体（Multi-Agent）是把一个复杂任务拆给多个各司其职的 Agent 去协作完成，FastAPI 则是把这些 Agent 串起来、对外暴露接口的服务层。能做什么？典型场景就是：用户一句话丢进来，系统自动判断该查资料、该算数、该调工具，然后几个 Agent 分工干完，把结果流式吐回手机屏幕。适合谁？有 Flutter 基础想往 AI 方向走的移动端开发，或者会 Python 想补前端展示层的后端同学，两边都能上手。

我特别想强调一点：多智能体不是"多个模型一起回答"这么简单。它的核心价值在于职责隔离和可观测性。单个 Agent 什么都干，prompt 会越写越长，出错根本不知道是哪一步崩的；拆成多个之后，每个 Agent 的输入输出边界清晰，调试和替换都容易得多。这也是我坚持用这套架构的根本原因。

2. 整体架构设计与技术选型拆解

2.1 三层架构：客户端、编排层、模型层

我把整个系统切成三层，边界划得很死，这样任何一层出问题都能快速定位。

• 客户端层（Flutter）：只负责 UI 渲染、用户输入、流式内容展示、本地状态管理。它不关心背后有几个 Agent，只认一个统一的 SSE 或 WebSocket 接口。
• 编排层（FastAPI + Agent 框架）：接收请求，做意图识别，调度多个 Agent 协作，管理会话上下文，最后把结果流式推回客户端。
• 模型层（LLM 服务）：可以是云端 API，也可以是本地部署的推理服务，对编排层暴露统一的调用接口。

这么分的好处是，客户端换框架不影响后端，模型换供应商不影响前端。我实测下来，这种解耦在迭代期能省掉大量返工。

2.2 为什么编排层选 FastAPI 而不是 Flask 或 Django

FastAPI 在这个场景里有三个硬优势，是我选它的直接原因。

第一是原生异步。LLM 调用是典型的 IO 密集型操作，一个请求可能要等好几秒甚至几十秒。Flask 的同步模型在这种场景下并发能力很差，而 FastAPI 基于 asyncio，一个 worker 就能扛住大量并发等待。第二是自动生成 OpenAPI 文档，前后端联调时 Flutter 这边直接照着 schema 写模型类，省去手写接口文档的麻烦。第三是Pydantic 数据校验，Agent 之间传递的结构化数据用它定义，类型不对直接报错，比手动 if-else 校验靠谱得多。

Django 太重，自带 ORM、Admin 那一套在这个项目里用不上，反而拖慢启动。Flask 轻是轻，但异步支持要靠额外扩展，不如 FastAPI 原生来得干净。

2.3 多智能体的拆分策略

Agent 怎么拆，是整个项目最考验设计功力的地方。我的原则是按能力边界拆，不按对话轮次拆。

一个典型的三 Agent 结构是这样的：

Router 只做分类，Tool 只做执行，Summary 只做整合。每个 Agent 的 prompt 都能写得很短很聚焦，准确率自然就上去了。我试过把所有逻辑塞进一个 Agent，prompt 写到两千多字，结果模型经常"忘记"其中某条规则，拆开之后这个问题基本消失。

2.4 Flutter 端的状态管理选型

Flutter 这边我推荐用Riverpod而不是 Provider 或 Bloc。原因很实际：流式输出场景下，状态更新非常频繁，Riverpod 的细粒度监听能避免整棵树重建。Bloc 的 event-state 模式在这种高频更新下写起来很啰嗦，Provider 又容易在复杂依赖下出现 rebuild 失控。

具体做法是把"会话消息列表"和"当前流式片段"分成两个独立的 provider。流式片段每来一个 token 就更新一次，只触发对应气泡的重建，历史消息列表不动。这个细节对流畅度影响很大，后面实操部分会展开。

3. 核心细节解析与实操要点

3.1 FastAPI 项目目录结构怎么组织

很多人写 FastAPI 项目，所有路由堆在一个 main.py 里，几百行之后自己都找不到。我用的结构是这样的：

app/ ├── main.py # 应用入口，注册路由和中间件 ├── api/ │ ├── __init__.py │ └── chat.py # 对话相关接口 ├── agents/ │ ├── base.py # Agent 基类，定义统一接口 │ ├── router_agent.py │ ├── tool_agent.py │ └── summary_agent.py ├── core/ │ ├── config.py # 配置管理 │ └── llm_client.py # 模型调用封装 ├── schemas/ │ └── chat.py # Pydantic 模型 └── services/ └── orchestrator.py # 多智能体编排逻辑

关键点是agents/base.py里定义一个抽象基类，所有 Agent 继承它，统一async def run(self, input) -> output接口。这样编排层调度的时候不用关心具体是哪个 Agent，直接遍历调用就行。这个设计我强烈推荐，后期加新 Agent 只要继承基类、注册到编排器，其他代码一行不用改。

3.2 Agent 基类与统一接口设计

基类长这样：

from abc import ABC, abstractmethod from pydantic import BaseModel class AgentInput(BaseModel): query: str context: dict = {} class AgentOutput(BaseModel): result: str metadata: dict = {} class BaseAgent(ABC): def __init__(self, name: str, llm_client): self.name = name self.llm = llm_client @abstractmethod async def run(self, data: AgentInput) -> AgentOutput: pass

为什么用 Pydantic 定义输入输出？因为 Agent 之间传数据最容易出问题。有一次 Tool Agent 返回的字段名和 Summary Agent 期望的不一致，排查了半天。用了 Pydantic 之后，字段对不上直接抛 ValidationError，错误信息还告诉你缺哪个字段，效率提升非常明显。

3.3 Flutter 端流式渲染的关键实现

Flutter 接收流式数据，我用的是http包的StreamedRequest，配合utf8.decoder逐块解析 SSE 格式。核心逻辑是：

final request = http.Request('POST', uri); request.body = jsonEncode(payload); final response = await client.send(request); response.stream .transform(utf8.decoder) .transform(const LineSplitter()) .listen((line) { if (line.startsWith('data: ')) { final chunk = line.substring(6); if (chunk == '[DONE]') return; ref.read(streamingProvider.notifier).append(chunk); } });

这里有个坑：LineSplitter遇到不完整的行会缓存，但如果服务端发送的 chunk 边界刚好切在 UTF-8 多字节字符中间，utf8.decoder会报错。解决办法是用utf8.decoder的allowMalformed: true参数，或者确保服务端按完整字符发送。我踩过一次中文乱码的坑，就是因为这个。

3.4 多智能体协同的编排逻辑

编排器是整个系统的大脑，核心是一个状态机式的流程：

class Orchestrator: def __init__(self, agents: dict): self.agents = agents async def execute(self, query: str): # 第一步：路由 route_result = await self.agents['router'].run( AgentInput(query=query) ) task_type = route_result.metadata['task_type'] # 第二步：按类型调度 if task_type == 'tool_call': tool_result = await self.agents['tool'].run( AgentInput(query=query, context=route_result.metadata) ) context = {'tool_output': tool_result.result} else: context = {} # 第三步：整合输出 final = await self.agents['summary'].run( AgentInput(query=query, context=context) ) return final

这个流程看起来简单，但每一步都有讲究。Router 的输出必须是结构化的（比如 JSON），否则后面没法程序化判断。Tool Agent 要处理超时和失败重试。Summary Agent 的 prompt 里要明确告诉它"基于以下事实回答，不要编造"。

提示：Router Agent 的输出格式一定要用 JSON mode 或者 function calling 强制约束，纯文本解析非常脆弱，模型稍微换个措辞你的解析就崩了。

4. 实操过程与核心环节实现

4.1 从零创建 Flutter 项目并配置依赖

创建项目这一步，命令行和 IDE 都行，我习惯用命令行：

flutter create --org com.example --platforms=android,ios multi_agent_app cd multi_agent_app

--platforms参数指定只生成需要的平台，能减少项目体积。生成之后，pubspec.yaml里加这几个依赖：

dependencies: flutter: sdk: flutter flutter_riverpod: ^2.4.0 http: ^1.1.0 freezed_annotation: ^2.4.1 dev_dependencies: build_runner: ^2.4.0 freezed: ^2.4.1

Riverpod 管状态，http 发请求，freezed 生成不可变数据类。freezed 这个可能有人觉得多余，但在消息列表这种频繁增删的场景下，不可变数据类能避免很多"改了引用但没触发重建"的诡异 bug。

4.2 FastAPI 服务搭建与依赖安装

后端这边，先建虚拟环境再装依赖，别直接往全局环境装：

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install fastapi uvicorn[standard] pydantic httpx python-dotenv

uvicorn[standard]带上了 uvloop 和 httptools，性能比裸 uvicorn 好不少。启动命令：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

--reload只在开发时用，生产环境一定要去掉，否则文件监听会拖慢性能。

4.3 模型调用封装与超时重试

模型调用这块，我单独封装了一个 client，把超时、重试、错误处理都收进去：

import httpx from tenacity import retry, stop_after_attempt, wait_exponential class LLMClient: def __init__(self, base_url: str, api_key: str, timeout: float = 30.0): self.client = httpx.AsyncClient( base_url=base_url, timeout=httpx.Timeout(timeout, connect=5.0), headers={'Authorization': f'Bearer {api_key}'} ) @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)) async def chat(self, messages: list, stream: bool = False): resp = await self.client.post('/v1/chat/completions', json={ 'messages': messages, 'stream': stream }) resp.raise_for_status() return resp

超时设置有个经验值：连接超时 5 秒，读取超时 30 秒。读取超时不能太短，模型生成长文本时可能超过 10 秒；也不能太长，否则用户等半天没反应。重试用指数退避，第一次等 1 秒，第二次 2 秒，第三次 4 秒，避免瞬间打爆服务端。

4.4 流式接口的完整实现

FastAPI 返回流式响应，用StreamingResponse：

from fastapi.responses import StreamingResponse @router.post('/chat/stream') async def chat_stream(req: ChatRequest): async def event_generator(): async for chunk in orchestrator.stream_execute(req.query): yield f'data: {json.dumps(chunk, 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这个头很关键。如果你前面挂了 Nginx 之类的反向代理，不加这个头，代理会缓冲整个响应再一次性发出去，流式效果就没了。我第一次部署时被这个坑了整整一个下午，本地测试好好的，一上服务器就变成"转圈半天然后一次性出结果"。

4.5 Flutter 端消息列表与流式气泡

客户端这边，消息模型用 freezed 定义：

@freezed class ChatMessage with _$ChatMessage { const factory ChatMessage({ required String id, required String role, required String content, @Default(false) bool isStreaming, }) = _ChatMessage; }

流式气泡的渲染逻辑是：收到第一个 chunk 时创建一条isStreaming: true的消息，后续 chunk 追加到这条消息的 content 上，收到[DONE]时把isStreaming置为 false。这样用户看到的就是文字一个个蹦出来的效果。

注意：追加 content 时不要每次都创建新对象，用 Riverpod 的state = state.copyWith(...)，freezed 生成的 copyWith 会做结构共享，性能比全量重建好很多。

5. 常见问题与排查技巧实录

5.1 流式输出中断或乱码

这是最高频的问题。排查顺序我总结成一张表：

我遇到最多的是代理缓冲，其次是客户端解码。这两个占了流式问题的八成以上。

5.2 Agent 之间数据传递失败

前面提过，字段名不一致是重灾区。我的解决办法是所有 Agent 的输入输出都用 Pydantic 模型定义，并且在编排器里做一次显式校验。宁可多写几行校验代码，也不要让错误数据流到下一个 Agent。另外，Agent 的 metadata 字段建议统一用字符串键，避免嵌套过深，调试时打印出来一目了然。

5.3 并发场景下的性能问题

AI Agent 怎么扛并发，这是很多人关心的。我的经验是：瓶颈几乎永远在模型调用，不在你的服务代码。所以优化方向是：

• 用异步 IO，别用同步阻塞调用
• 给模型调用加连接池，复用 TCP 连接
• 对相同 query 做结果缓存，命中直接返回
• 设置合理的并发上限，超过就排队，别让请求雪崩

我实测过，一个 4 核 8G 的机器，纯转发场景下 FastAPI 能扛住几百并发，但一旦每个请求都要调模型，并发数就取决于模型服务的吞吐了。所以别在服务层过度优化，先把模型侧的并发能力搞清楚。

5.4 Flutter 页面切换后状态丢失

flutter navigator 切换页面后会不会丢失状态，这个问题的答案是：取决于你怎么管理状态。用Navigator.push压栈，原页面还在栈里，状态保留；用pushReplacement替换，原页面销毁，状态丢失。如果状态放在页面内部的StatefulWidget里，页面销毁就没了；放在 Riverpod 的全局 provider 里，页面销毁状态还在。

我的做法是：会话数据放全局 provider，UI 临时状态（比如输入框内容、滚动位置）放页面内部。这样既不会丢数据，也不会让全局状态膨胀。

5.5 模型返回格式不符合预期

llm request failed: provider rejected the request schema or tool payload这类报错，通常是请求体结构不对。排查要点：

• 检查 messages 数组格式，role 只能是 system/user/assistant
• tool 定义要符合对应模型的 schema 规范
• 有些模型不支持某些参数，传了会直接拒绝

我的习惯是先用 curl 或 Postman 手动发一次请求，确认请求体没问题，再往代码里搬。这样能把"代码问题"和"请求格式问题"分开排查。

6. 我在这套架构上的一些实战心得

先说一个反直觉的结论：多智能体系统的复杂度，八成来自编排逻辑，而不是 Agent 本身。每个 Agent 单独写都很简单，难的是它们之间怎么协作、怎么传递上下文、怎么处理某个 Agent 失败的情况。所以我在项目里会花大量时间设计编排器的状态流转，而不是纠结单个 Agent 的 prompt 怎么写。

第二个心得是关于 prompt 的。Router Agent 的 prompt 我改了不下二十版，最后发现最有效的写法是给几个明确的例子，而不是写一堆规则。比如"用户问天气属于 tool_call，用户问概念属于 direct_answer"，配上两三个具体例子，比写十条抽象规则准确率高得多。

第三个是关于调试的。多智能体系统出问题时，光看最终输出根本不知道哪一步错了。我的做法是在编排器里加一个 debug 模式，把每个 Agent 的输入输出都打到日志里，用请求 ID 串起来。这样一次请求的完整链路一目了然。这个功能开发时多花半小时，调试时能省几十小时。

最后分享一个 Flutter 端的小技巧：流式渲染时，如果 token 来得太快，每来一个就 setState 会导致界面卡顿。我的做法是加一个 16ms 的节流（差不多一帧的时间），把这段时间内到达的 token 合并成一次更新。实测下来流畅度提升非常明显，尤其是模型输出速度快的时候。

这套架构我前后迭代了三个版本，从最初的单 Agent 硬扛，到现在的三 Agent 协作，每一步都是被实际问题逼出来的。如果你正准备入这个坑，我的建议是：先把单 Agent 跑通，再拆多 Agent。别一上来就设计复杂的协作流程，那样你连问题出在哪都找不到。等单 Agent 的链路稳定了，再按能力边界逐步拆分，每一步都保证可回滚，这样最稳。