基于FastAPI搭建AI智能体对话平台：从项目骨架到Agent引擎实现

上一篇我们把整个智能体对话平台的架构、模块边界和演进路线捋清楚了，当时很多人留言催更，问“什么时候开始写代码”。这一篇就直接开工，从初始化一个空项目开始，把地基一寸一寸打起来。这篇的核心关键词是“项目搭建”“FastAPI”“流式响应”“Agent 引擎”“会话存储”，适合那些准备做 AI Agent 应用但不知道怎么优雅起项目的人，也适合已经写过不少 CRUD 但第一次接触 LLM 项目的开发者。读完你手里会有一套完整的、可运行的最小骨架，后面无论是接新模型、加工具还是换前端，都在这套骨架上做加法。

把话说在前面：这个“地基”不是搭个能跑的 Hello World 就完了，而是要搭一个能承载多轮对话、工具调用、流式输出、会话持久化的骨架。脚手架如果立歪了，后面每一次加功能都是在歪墙上砌砖，越砌越心惊。所以这一篇会有大量“为什么这么写”的思考，代码反而是其次。

1. 地基要打在哪——先想清楚再动手

很多项目死在第一步，死法不是“不会写代码”，而是“没想清楚就写代码”。尤其是 AI 智能体对话平台，表面看就是一个聊天框来回来去发消息，但底下要扛的事远比普通 Web CRUD 复杂。

1.1 智能体对话平台的“地基”到底指什么

我理解的对话平台地基，至少要有六根柱子撑住整个系统，缺一根后面都会塌。第一根是LLM 接入层，统一封装各家大模型接口，让业务代码不需要感知底层是哪个厂商的模型；第二根是会话管理，要知道每一轮对话属于谁、上下文是什么、历史上聊过什么；第三根是Agent 引擎，这是智能体区别于普通聊天机器人的关键，它决定大模型的输出是否触发工具调用；第四根是消息存储，对话要落库，不能服务一重启就失忆；第五根是流式传输，大模型生成动辄几百上千字，一次性等结果用户体验太差，必须走 SSE 或 WebSocket；第六根是可观测性，要能随时知道哪一轮调用花了多少 Token、哪一步工具报错了。

这六根柱子里，前面五根是功能性的，第六根是保命性的。很多项目一开始图省事，先不加日志和链路追踪，等上线后用户报问题完全没法复现，那叫一个酸爽。地基阶段就把日志规范立起来，成本极低，收益极高。

1.2 技术选型的底层逻辑：为什么还是 Python + FastAPI

选型这件事，我不喜欢追新，更看重“生态成熟度”和“团队认知成本”。这套平台的底座我选 Python，理由不用赘述，大模型 SDK、数据处理、AI 生态全是 Python 的天下。Web 框架选 FastAPI 而不是 Flask 或 Django，有三个实际原因。

第一，FastAPI 原生支持异步，而大模型接口动辄几秒到几十秒，异步是刚需，底层库搞成同步请求会直接拖垮整个服务的并发能力。Django 的异步支持这些年虽然有进步，但生态心智还是偏向同步，Flask 更是基本同步模型为主。第二，FastAPI 自带 OpenAPI 文档，调接口联调时直接把 /docs 甩给前端，省掉手写接口文档这一步。第三，FastAPI 的依赖注入系统很顺手，写鉴权、写通用参数抽取、写可测试的代码结构都比 Flask 手工磨依赖来得利落。

ORM 这块，新项目我推荐 SQLModel，它是社区里某个知名大佬在 SQLAlchemy 和 Pydantic 之间做了一层缝合，既能用到 Pydantic 的模型校验和能力复用，又保留 SQLAlchemy 的全部能力。如果过去你只熟 SQLAlchemy，直接用也行，这块不是生死线。数据库先上 SQLite，够用、零运维，等项目确认并发上来了再平滑切 PostgreSQL，这套地基会从一开始就把 ORM 模型定义和具体数据库解耦。

1.3 和纯聊天机器人有什么本质不同

普通聊天机器人是“你问我答”，本质就是一个在线函数：输入用户消息，拼接 system prompt，调用模型，返回答案，完事。智能体对话平台多了一层“自主决策”，模型不再是回答完就结束，而是可能调工具、查数据库、读外部 API，甚至多轮思考后才给出最终答案。

这个差异直接影响了代码结构。普通聊天机器人你把 LLM 调用写在路由里也行，但智能体平台必须把“调模型”“跑循环”“执行工具”分离成独立模块。否则三个工具一接入，路由里的代码就成了屎山。这也是地基阶段我们就要把核心引擎抽象出来的原因，不是为了架构炫技,而是工具一多，不分层真的会把自己绕进去。

2. 项目初始化——从空目录到第一个可运行的服务

地基的第一步是搭一个“能跑但不丑陋”的基础工程。这一步不复杂，但细节不少，每一个选择都在为后面省时间。

2.1 环境准备与依赖锁定

先确认机器上的 Python 版本，建议 3.11 及以上。低于 3.10 的话，很多新语法糖用不了，写起来憋屈。创建虚拟环境是第一条死命令：

python3.11 -m venv .venv source .venv/bin/activate pip install --upgrade pip

依赖不建议一次性装一大堆，用多少装多少，装多了容易版本冲突。下面是第一批初始化依赖，全部是必要项：

把依赖写进 requirements.txt，用 pip 一次性安装。这是地基阶段最没技术含量但最不能省的动作，后面每一行代码都踩在依赖上。

需要注意一个坑：openai 这个 SDK 的版本更新极快，接口经常微调。装完之后先跑一行pip show openai看一眼版本，我见过太多人照着老教程写代码，结果版本差异导致参数报错，卡了一个晚上。

2.2 目录结构：不要把所有代码堆在 main.py 里

单文件教学适合 Demo，不适合工程。这个项目从一开始就分模块，目录是长这样的：

agent-platform/ ├── app/ │ ├── main.py # 应用入口，创建 FastAPI 实例 │ ├── core/ │ │ ├── config.py # 全局配置读取 │ │ └── logging.py # 日志初始化 │ ├── models/ # SQLModel 数据模型 │ │ ├── conversation.py │ │ └── message.py │ ├── schemas/ # Pydantic 请求/响应模型 │ │ └── chat.py │ ├── llm/ # LLM 接入层 │ │ ├── base.py # 抽象基类 │ │ ├── openai_client.py # 兼容 OpenAI 接口规范的实现 │ │ └── manager.py # 模型路由与实例管理 │ ├── agent/ # Agent 引擎 │ │ ├── engine.py # 推理主循环 │ │ └── tools.py # 工具注册与调度 │ ├── services/ # 业务服务层，路由只做薄转发 │ │ └── chat_service.py │ ├── routers/ # 路由层 │ │ └── chat.py │ └── db/ │ ├── session.py # 数据库会话 │ └── init_db.py # 建表逻辑 ├── tests/ # 测试目录 ├── .env.example ├── .gitignore └── requirements.txt

这套结构的核心逻辑是“路由、服务、模型、基础设施”四级分层。路由层只负责解析 HTTP 请求参数、调服务层、把结果封装成响应；服务层负责业务流程编排，比如调 Agent 引擎、处理会话 ID 逻辑；模型层管数据实体；基础设施层管数据库、配置、日志。谁也不要越级调用。

分层带来的直接好处是测试友好。你想测 Agent 引擎，直接 import engine，不用临时造 HTTP 请求；你想换数据库，只动 db 目录，业务代码一行不改。我见过太多把业务逻辑写在路由函数里的项目，后期给路由加一个鉴权中间件都得小心翼翼，一不留神就把业务搞挂。

2.3 配置管理：写死在代码里的秘密都要进环境变量

模型 API Key、数据库路径、模型名称这些配置，最忌讳写死在代码里。原因不单是安全隐患，还有多人协作和部署环境差异。本地一条 Key，测试环境一条 Key，生产环境一条 Key，代码里写死就只能每到一个环境改一次代码，迟早要出事。

配置文件放在app/core/config.py：

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): app_name: str = "Agent Platform" database_url: str = "sqlite:///./agent_platform.db" llm_api_key: str = "" llm_base_url: str = "https://api.openai.com/v1" llm_model_name: str = "gpt-4o-mini" llm_temperature: float = 0.7 llm_max_tokens: int = 2048 model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore" ) settings = Settings()

pydantic-settings 会自动从.env文件里加载同名配置项。.env文件加入.gitignore，仓库里只放一个.env.example模板，把键名列出来、值留空或者填占位符。这样任何人拉下代码，复制一份.env.example成.env，填上自己的 Key 就能跑。

这个方案被无数项目验证过，基本属于业界标准操作。别自创一套读取配置文件的方法，除非你有特殊场景，否则都是在重复造轮子且造得更差。

2.4 第一个能跑起来的接口：健康检查比 Hello World 更有用

很多人第一步写 Hello World，我更建议写一个健康检查接口。它不只是“我通了”，而是给部署探活、监控报警留的入口。以后上云原生，K8s 的 livenessProbe 和 readinessProbe 都要打这个接口。

from fastapi import FastAPI from app.core.config import settings app = FastAPI(title=settings.app_name) @app.get("/health", tags=["system"]) async def health_check(): return {"status": "ok", "app": settings.app_name}

启动命令：

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

浏览器打开http://localhost:8000/health，看到{"status":"ok"}就说明地基最底下的土夯实了。--reload是开发模式专用，改代码自动重启，但生产环境千万别带这个参数，性能和稳定性都会受影响。

3. LLM 接入层——地基里最核心的一根承重柱

这是整个地基里最不能含糊的部分。LLM 接入层就是一道防火墙，把“大模型厂商的 SDK 差异”全部挡在这层后面，上面所有模块只面向一个统一的抽象接口。

3.1 为什么要单独封装一层，而不是直接到处 import SDK

直接在大模型厂商的 SDK 散落在业务代码里，短期写起来爽，长期全是债。最典型的场景是：你今天写完”调模型A“的逻辑，明天老板说换个模型B，你发现模型B的 SDK 接口风格完全不同，参数名不一样，流式方式不一样，然后你就在全项目里做“用工”式的替换，改几十个文件，一边改一边担心漏改。

封装的核心思路是抽象一个BaseLLMClient基类，定义好两个方法：generate（非流式）和generate_stream（流式）。不同模型厂商各自有实现类，都继承这个基类。业务代码永远只依赖基类，不感知具体实现。

这和电源插座是一个道理：插座定义的是统一标准，你的手机充电器不用关心电厂烧煤还是烧气。换模型，就是换插头。

3.2 统一的接口定义：流式与非流式一把梭

先看非流式接口的抽象基类长什么样：

from abc import ABC, abstractmethod from typing import AsyncIterator from pydantic import BaseModel class LLMResponse(BaseModel): content: str model: str usage_total_tokens: int = 0 usage_prompt_tokens: int = 0 usage_completion_tokens: int = 0 class BaseLLMClient(ABC): def __init__(self, api_key: str, base_url: str, model: str): self.api_key = api_key self.base_url = base_url self.model = model @abstractmethod async def generate(self, messages: list[dict], **kwargs) -> LLMResponse: """非流式生成完整响应""" @abstractmethod def generate_stream(self, messages: list[dict], **kwargs) -> AsyncIterator[str]: """流式生成，逐个产出增量文本片段"""

字段LLMResponse里把 Token 用量也收纳进来，这一步很重要。Token 是成本指标，后面要做成本统计、异常排查，都靠这几个数字。很多初级开发者在设计接口时只关注返回的文本内容，忽略 Token 计量，后面想算成本就得重写接口。

再看基于兼容 OpenAI 接口规范的 SDK 的实现：

from openai import AsyncOpenAI from app.llm.base import BaseLLMClient, LLMResponse class OpenAILLMClient(BaseLLMClient): def __init__(self, api_key: str, base_url: str, model: str): super().__init__(api_key, base_url, model) self._client = AsyncOpenAI(api_key=api_key, base_url=base_url) async def generate(self, messages: list[dict], **kwargs) -> LLMResponse: resp = await self._client.chat.completions.create( model=self.model, messages=messages, stream=False, temperature=kwargs.get("temperature", 0.7), max_tokens=kwargs.get("max_tokens", 2048), ) return LLMResponse( content=resp.choices[0].message.content or "", model=resp.model, usage_total_tokens=resp.usage.total_tokens if resp.usage else 0, usage_prompt_tokens=resp.usage.prompt_tokens if resp.usage else 0, usage_completion_tokens=resp.usage.completion_tokens if resp.usage else 0, ) async def generate_stream(self, messages: list[dict], **kwargs) -> AsyncIterator[str]: stream = await self._client.chat.completions.create( model=self.model, messages=messages, stream=True, temperature=kwargs.get("temperature", 0.7), max_tokens=kwargs.get("max_tokens", 2048), ) async for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: yield chunk.choices[0].delta.content

这里用的是异步客户端AsyncOpenAI，配合 FastAPI 的异步能力，整个链路全程异步，不会出现“接口卡住但服务器线程被占死”的问题。

多模型厂商的接入就变成了添加新类的问题，比如接某国内模型厂商，它兼容 OpenAI 接口规范的话，直接复用OpenAILLMClient只要改base_url和model即可。如果不兼容，就再写一个继承BaseLLMClient的新实现类，业务代码零改动。这就是封装的价值。

3.3 模型管理：别在代码里散落模型名

业务代码里直接写"gpt-4o-mini"这种字符串是不好的习惯。一个平台很可能同时用多个模型，大模型当主脑、小模型当功能分类器、开源模型当备用。在manager.py里做一层模型路由：

from app.llm.base import BaseLLMClient from app.llm.openai_client import OpenAILLMClient from app.core.config import settings class LLMManager: def __init__(self): self._clients: dict[str, BaseLLMClient] = {} self._register_default_clients() def _register_default_clients(self): # 主对话模型 self._clients["default"] = OpenAILLMClient( api_key=settings.llm_api_key, base_url=settings.llm_base_url, model=settings.llm_model_name, ) def get_client(self, name: str = "default") -> BaseLLMClient: return self._clients.get(name, self._clients["default"]) llm_manager = LLMManager()

以后要在不同的对话场景里用不同的模型，只在 manager 里多注册一个实例就行。模型名、Base URL 都走配置，不写死在业务里。

3.4 接入层还要管好成本与容错

之前我在某个项目里接过一个模型，它偶尔会在流式输出中途断掉，错误信息还特别隐晦，是乱七八糟的连接错误。如果没有容错重试机制，用户看着对话窗口停住不动，体验极其糟糕。所以在接入层加一定次数的重试逻辑：

from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential async def generate_with_retry(client, messages, max_attempts=3): async for attempt in AsyncRetrying( stop=stop_after_attempt(max_attempts), wait=wait_exponential(multiplier=1, min=1, max=10), reraise=True ): with attempt: return await client.generate(messages)

指数退避是重试的核心原则：第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒，把压力错开，而不是疯狂重试把下游打爆。这个依赖tenacity库，顺手加进 requirements。

4. 智能体核心引擎起步——Agent 不只是套壳对话

到这里，真正拉开智能体和普通聊天机器人差距的部分开始了。Agent 引擎的核心是一个“思考-行动-观察”循环：模型生成决策，如果是工具调用就执行工具并返回结果，让模型继续思考，直到模型给出最终回答。

4.1 会话上下文：Agent 的记忆胶囊

Agent 不能只有“当前一句话”，它必须携带完整上下文。这里的上下文不只是历史消息，还包括系统提示、工具定义、可选的用户画像。我设计一个 Context 组装模块，核心是把会话里的历史消息翻译成大模型能理解的消息列表：

from app.models.conversation import Conversation from app.models.message import Message def build_messages(conversation: Conversation, user_input: str) -> list[dict]: messages = [{"role": "system", "content": conversation.system_prompt}] for msg in conversation.messages: messages.append({"role": msg.role, "content": msg.content}) messages.append({"role": "user", "content": user_input}) return messages

这里有几个细节值得注意。第一，system_prompt要放在第一条，这是模型理解全局行为准则的地方，顺序错了效果会打折。第二，历史消息如果过长，要做截断或摘要，否则一下子就顶爆模型的上下文窗口。简单方案是只保留最近 N 轮，进阶方案是上一轮内容先做摘要压缩再拼接。第一版先把“最近 N 轮”策略跑通，摘要后面迭代。

4.2 工具注册机制：让 Agent 拥有一双手

工具是智能体区别于普通聊天的关键。工具的本质是一个“可以被大模型凭名字和参数描述调用”的函数。在tools.py里做一套注册器：

from typing import Callable, Awaitable class ToolRegistry: def __init__(self): self._tools: dict[str, dict] = {} def register(self, name: str, description: str, parameters: dict, handler: Callable[..., Awaitable[str]]): self._tools[name] = { "spec": { "type": "function", "function": { "name": name, "description": description, "parameters": parameters, } }, "handler": handler, } def list_specs(self) -> list[dict]: return [tool["spec"] for tool in self._tools.values()] async def call(self, name: str, arguments: dict) -> str: handler = self._tools[name]["handler"] return await handler(**arguments)

每接入一个工具，就写一个异步函数，然后register一行注册。工具的 spec 遵循大模型厂商定义的 function calling 格式。注意第二个细节：工具执行结果必须返回字符串，因为大模型 API 规定工具结果要以字符串形式回传给模型。handler 返回结构化对象也行，但要记得在整体上把它们转成纯文本。

4.3 第一版 Agent 推理循环：ReAct 模式的迷你实现

ReAct（Reasoning + Acting）是当前大部分 Agent 框架的底子。放到代码里长这样：

class AgentEngine: def __init__(self, llm_client: BaseLLMClient, registry: ToolRegistry): self.llm = llm_client self.tools = registry async def run(self, messages: list[dict], max_iterations: int = 5): for _ in range(max_iterations): response = await self.llm.generate(messages, tools=self.tools.list_specs()) if response.tool_calls: for tool_call in response.tool_calls: result = await self.tools.call( tool_call.name, json.loads(tool_call.arguments), ) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result, }) continue # 没有工具调用，说明模型准备好了最终回复 messages.append({"role": "assistant", "content": response.content}) return response.content, messages return "到达最大迭代次数，强制结束。", messages

这个循环看着简单，但它是几乎所有 Agent 产品的内核。模型每一轮先输出是否需要工具调用，代码看情况决定是执行工具还是返回最终回答。max_iterations是循环的安全阀，防止模型陷入无限调工具的死循环，这也是实际运营中一定会遇到的坑。第一版先粗暴限制次数，后面可以升级为“预算管理”——按照 Token 消耗和工具调用次数双维度控制。

4.4 工具调用失败不能让它裸奔

工具执行大概率有异常：外部 API 超时、参数 JSON 解析失败、业务逻辑报错。这些异常不应该直接抛回给大模型接口，而应该转换成模型能理解的错误文本，并回贴到上下文中，让模型自我纠正。否则一个工具调用失败，整个对话直接崩掉，用户只看到一个错误页面。

async def safe_call(self, name: str, arguments: dict) -> str: try: return await self.tools.call(name, arguments) except Exception as e: return f"工具 {name} 执行失败，错误信息：{str(e)}。请修正参数后重试，或放弃此工具直接回答。"

这个设计极其实用。模型看到错误后，通常会自动调整参数重新尝试，或者干脆放弃工具直接回答，而不是整个进程崩溃。

5. 会话与消息存储——地基的第二层承重墙

Agent 引擎跑起来了，但还缺一样东西：记忆必须落到硬盘上。没有数据库的对话平台，就像失忆症患者，聊完上一句就忘了下一句。

5.1 数据模型设计：会话和消息不分家就是自找麻烦

先定义会话表和消息表。会话表存“一段对话的元信息”，消息表存“每一条具体的消息”。它们是一对多的关系。

from datetime import datetime, timezone from sqlmodel import SQLModel, Field, Relationship class Conversation(SQLModel, table=True): id: str = Field(primary_key=True) title: str = Field(default="新会话") system_prompt: str = Field(default="你是一个乐于助人的智能助手。") created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) messages: list["Message"] = Relationship(back_populates="conversation") class Message(SQLModel, table=True): id: str = Field(primary_key=True) conversation_id: str = Field(foreign_key="conversation.id") role: str = Field() # user / assistant / system / tool content: str = Field() tool_call_id: str | None = Field(default=None) tool_name: str | None = Field(default=None) extra_usage: str | None = Field(default=None) # JSON 字符串，存 Token 消费 created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) conversation: Conversation | None = Relationship(back_populates="messages")

消息表里特意放了tool_call_id、tool_name和extra_usage。这三个字段是给 Agent 调试和成本分析留的位置。没有tool_call_id，多轮工具调用的消息链路就对不上；没有extra_usage，每个对话花多少钱就是一笔糊涂账。

主键用字符串 UUID 而不是自增 ID，好处是分布式环境不用中央发号器，前端回传 ID 也方便。

5.2 数据库会话的初始化与依赖管理

数据库初始化走app/db/session.py和app/db/init_db.py：

from sqlmodel import SQLModel, create_engine from app.core.config import settings engine = create_engine(settings.database_url, echo=False) def init_db(): SQLModel.metadata.create_all(engine)

建表动作在 FastAPI 启动事件里执行一次：

from contextlib import asynccontextmanager from fastapi import FastAPI from app.db.init_db import init_db @asynccontextmanager async def lifespan(app: FastAPI): init_db() yield app = FastAPI(title=settings.app_name, lifespan=lifespan)

注意，init_db()用的create_all只会新建不存在的表，不会改已有表。1.0 阶段可以这么搞，等模型字段稳定了，上正式团队版本再换迁移工具。

5.3 并发问题：SQLite 不是让你全家桶线程硬怼

SQLite 的并发写能力弱，这是众所周知的事实。对话平台的消息写入比较频繁，如果同时多个请求写同一个 SQLite 文件，会出现database is locked报错。我的第一版直接给 SQLite 加了个 workaround 参数：

engine = create_engine( settings.database_url, echo=False, connect_args={"check_same_thread": False, "timeout": 30} )

check_same_thread=False是让 SQLite 允许 SQLModel 在异步线程里使用连接池，timeout=30是等锁时间拉长一点。这能缓解问题但根治不了，等并发量真上来了，还是要把 PostgreSQL 队列排上日程。

6. 从第一行代码到第一次完整对话——把地基跑起来

模块都垒好了，但光有砖头不是房子。这一节把上面所有碎片拼起来，让用户通过一个 HTTP 接口和 Agent 完成一次带流式输出与工具调用的完整对话。

6.1 定义请求与响应模型

前端传什么进来，后端回什么出去，必须先定义清楚，否则前后端各自瞎猜。

from pydantic import BaseModel class ChatRequest(BaseModel): conversation_id: str | None = None message: str stream: bool = True class ChatResponse(BaseModel): conversation_id: str reply: str

前端第一次调接口时不传conversation_id，表示开新会话；后续调接口把上一轮返回的conversation_id带上，系统就能接上上下文。

6.2 业务服务层：拼装完整链路

服务层把会话查询/创建、消息入库、Agent 运行、消息落库这几个动作串起来。这一层的代码是整个地基里最能体现“业务顺序”的位置。

import uuid from app.models.conversation import Conversation from app.models.message import Message from app.agent.engine import AgentEngine from app.llm.manager import llm_manager from app.agent.tools import registry from app.db.session import engine from sqlmodel import Session, select async def run_chat(conversation_id: str | None, user_message: str): with Session(engine) as db: if conversation_id: conversation = db.get(Conversation, conversation_id) else: conversation = Conversation(id=str(uuid.uuid4())) db.add(conversation) db.add(Message( id=str(uuid.uuid4()), conversation_id=conversation.id, role="user", content=user_message, )) db.commit() messages = build_messages(conversation, user_message) engine = AgentEngine(llm_manager.get_client(), registry) reply, full_messages = await engine.run(messages) db.add(Message( id=str(uuid.uuid4()), conversation_id=conversation.id, role="assistant", content=reply, )) db.commit() return conversation.id, reply

这里有一个特别容易踩的坑：build_messages必须在用户消息入库之后执行，否则查不到刚刚插入的 user 消息。或者更稳的办法是直接用变量拼接，显式地把你刚刚 insert 的那条消息加进 messages，而不依赖数据库查询。实际项目里我会选择显式传参方式，省掉一次查询，少一轮数据库往返。

6.3 流式输出：把 HTTP 转发体验做对

非流式接口一锤子返回已经能跑通了，但面向用户的体验还是得上流式。SSE（Server-Sent Events）是最简单的方案，FastAPI 原生支持：

from fastapi.responses import StreamingResponse @router.post("/chat/stream") async def chat_stream(req: ChatRequest): conversation_id, reply = await run_chat(req.conversation_id, req.message) # 简化版：先攒完整回复再流式回放 async def event_generator(): for chunk in reply: yield f"data: {chunk}\n\n" yield "data: [DONE]\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, )

这是最简实现，真正生产环境要改造成“Agent 引擎每产出一点内容就 yield 一次”，而不是先攒完整回复再回放，那叫伪流式。真正的直连流式我在后面系列讲“优化篇”时会展开细写。

6.4 用 pytest 给地基做一个快速体检

地基稳不稳，不能靠人肉 curl 几次就算数，要建一套自动化验证。先写一个最小测试：

import pytest from app.llm.manager import llm_manager from app.agent.engine import AgentEngine from app.agent.tools import registry @pytest.mark.asyncio async def test_agent_engine_basic(): client = llm_manager.get_client() engine = AgentEngine(client, registry) messages = [{"role": "user", "content": "你好，请介绍一下你自己。"}] reply, _ = await engine.run(messages) assert isinstance(reply, str) assert len(reply) > 0

注意这个测试会真实调用大模型接口，所以它依赖.env里的 API Key 是有效的。如果不想在 CI 里消耗真实 Token，就 mock 掉BaseLLMClient.generate。单元测试和目标库的边界要清楚，测的是逻辑编排，不是模型本身。

7. 地基阶段实测遇到的高频问题与排查清单

这一步直接把我在真实开发中遇到的坑列成清单，每一个都见过身边的同事或者自己在项目里真实翻过车。

问题一：装了最新版 FastAPI 但还是报lifespan参数错误。

坦白说这个坑主要出现在 FastAPI 和 Uvicorn 版本跨度特别大的时候。旧版 FastAPI 用的是@app.on_event("startup")，新版推荐lifespan。解决方案是统一升级依赖到最新版，不要新旧风格混用。

问题二：openai SDK 发送 tools 参数报错。

SDK 版本和模型兼容性是最常见的翻车点。新版 SDK 要求tools字段必须符合厂商定义的严格格式，所以我在tools.py里特意做了spec规范化。如果还是报错，先跑一下pip show openai确认版本，再看官方文档里chat.completions的最新参数签名。

问题三：SQLite 报database is locked。

这个前面说过，先用连接参数缓解，但更根本的方案是将来切 PostgreSQL。对话类应用写入频繁，SQLite 的锁粒度扛不住高并发，早做规划，晚吃苦头。

问题四：流式响应在浏览器端收不到第二个 chunk。

这个问题十有八九是 Nginx 开启了缓冲。SSE 响应必须显式关掉 Nginx 的 proxy_buffering，或者后端响应头带上X-Accel-Buffering: no。第一个 chunk 能发出去，后面的卡住，绝大多数就是这层缓冲捣的鬼。

问题五：Agent 循环里工具调用参数被错误地塞给了字符串。

JSON 解析失败是 Agent 开发高频问题。模型的工具调用 arguments 本质是 JSON 字符串，解析时要包一层 try-except。如果解析失败，把报错信息回贴给模型，大部分时候模型会自我修正。

下面是排查顺序速查表，按出现概率排序：

8. 这张地基图后面往哪里长

代码全跑通之后，下面的扩展路线就非常清晰了。第一优先做的是把伪流式改成真流式，让 Agent 引擎在思考过程中就逐字吐出输出，而不是等全部生成完再回放。第二是给会话管理加“自动摘要压缩”，当历史消息超过上下文窗口时，用一句概述替代旧消息，保住长期记忆。第三是引入多租户的数据隔离，为平台化做准备。第四是加一套完整的链路追踪，把每次对话消耗的模型、工具调用、Token 分布可视化出来。

我个人在这段地基搭建过程中体会最深的一点是：智能体项目最难的永远是边界画在哪。LLM SDK 该封装到哪一层、工具调度该放多深、数据库和业务逻辑的耦合度怎么控制，这些决定了一个项目做到后期是越做越清爽还是越做越拧巴。地基阶段的“慢”，换的是后面每一层迭代的“快”。如果你正在搭或者准备搭类似的智能体对话平台，照着这套骨架走一遍，先把最小闭环跑通，再谈加功能，路会顺很多。