很多人开始做AI智能体对话平台的时候都容易走反方向一上来就让模型开口说话。我第一次搭这类项目也是这样找一套大模型服务的示例代码改巴改巴跑通了一次对话兴奋劲儿刚过就发现这只是一个会回话的脚本离平台还差着一个完整的地基。AI智能体对话平台开发实战里最磨人的往往不是想什么酷炫的 Agent 玩法而是那些看起来平平无奇的地基项目怎么组织、配置放哪里、会话和消息怎么建模、接口怎么把上下文喂给模型、出错以后怎么定位。这些东西如果第一周不弄明白后面每加一个功能都要回头拆墙。这篇是系列第二节主题就是把地基完整搭起来——从一台只装了 Python 的空机器到一个能跑通真实对话的最小平台并且把每一步背后的为什么讲清楚。适合刚准备动手做同类项目的人也适合已经写了大量示例代码、现在想回头整理结构的老手。1. 地基搭建到底在搭什么先别急着让模型开口1.1 Demo 后遗症跑通了反而更难受先讲一个很常见的场景我估计不少人都经历过。项目刚开始的时候图快把代码全堆在一个文件里凭记忆拼出一次模型调用curl 一发有回复了然后就开始在上面继续加功能。加着加着问题会像连环雷一样冒出来——上下文整个存在内存列表里服务一重启用户的历史对话全部蒸发密钥直接写在文件顶部某次把代码发给别人看或者提交到仓库的时候顺手就把密钥透传出去了模型名、温度这些参数散落在各处想换一套模型得全文搜索替换接口一报错前端收到的是完整的调用栈用户看不明白你自己也懒得翻日志。这些现象本质上是同一个问题地基没有做。地基这个词听起来很虚但落到代码上就是几个具体的决策和文件。这一节先把它们拆开后面几节逐项落地。1.2 地基的四块砖骨架、配置、数据、链路我给地基划了个很具体的范围一共四层可运行的服务骨架。有明确的启动入口、路由组织方式、目录分层能让人一眼看出这个项目怎么跑起来、新功能往哪里塞。环境与配置管理。密钥、模型服务地址、模型名、数据库地址全部走配置不写死在代码里环境不同就换一份配置。核心数据模型。会话、消息、智能体这三张表的关系一开始就定清楚别等对话记录没法查的时候再重构。一条最小但完整的对话链路。用户发一句话系统带着历史上下文去调模型模型返回后再落库整条链路闭环。1.3 这篇的边界哪些事这次不做既然是地基就要克制。这次我不会去做工具调用、多智能体路由、会话上下文压缩、流式输出这些进阶功能——它们都是下一篇或者后面几篇的主角。如果现在就把这些复杂度塞进门槛第一行代码根本落不了地。但这不意味着后路不管。地基的每一块砖都会留出接口模型接入做成薄的封装以后换服务商不伤筋动骨消息表单独建以后做记忆、做压缩、做审计都有地方挂路由目录按模块分以后加功能只是往里添文件。先跑通再长胖但跑通之前想清楚它会长成什么样。2. 技术选型四个决定未来半年开发体验的选择2.1 语言与框架为什么是 Python FastAPI选 Python 基本没什么悬念。模型服务商的 SDK、示例代码、开源生态几乎全是 Python 优先写 AI 相关业务能省掉大量踩坑时间。版本上用 3.11 起步类型标注和性能都够用。框架我直接说结论FastAPI。理由不是它自动生成接口文档好看那么简单核心原因是异步。一次对话要等大模型输出这个等待时间在网络上常常是十几秒到几十秒同步阻塞模式下一个慢请求会把整个服务卡住并发一上来就麻了。FastAPI 基于异步 Web 框架处理这种高等待、低计算密度的场景天然合适。再加上它和 Pydantic 深度绑定请求参数校验、响应模型、配置模型用的都是同一套类型系统少学一个概念。有人会问为什么不选 Django。Django 的 admin、ORM、中间件生态确实强但它是同步模型硬上异步需要额外封装对对话平台这种模式比较简单的后端属于杀鸡用了牛刀。Flask 则太自由项目一旦变多文件路由和依赖注入全靠自己约定团队协作时风格容易乱。2.2 模型接入不绑定任何一家大模型服务商这是地基里最容易犯的错。很多示例代码直接用了某个服务商封装的 SDK看起来省事实际上把服务和某一家厂商的接口细节耦合在了一起后面想换或者想本地部署开源模型得重写业务代码。我的做法是只依赖一个大模型行业内普遍兼容的接入协议自己封装一个极薄的客户端。请求体结构大体上是三样模型名、消息列表、温度等生成参数响应结构里拿聊天内容也是固定路径。把这个协议封装在 app/llm_client.py 这个文件里业务代码只调用chat(messages, model, temperature)至于背后是哪个服务商由配置文件的 base_url 和 api_key 决定。这样一来切换模型服务商这件事从改业务代码降级成改环境变量地基期多花二十分钟后面能省出无数个下午。2.3 存储SQLite 起步SQLAlchemy 留后路对话数据要不要一开始就用 PostgreSQL我的建议是不要。开发期最重要的目标是快速迭代SQLite 零配置、单文件、随手可备份足够支撑开发和单人测试场景。真正的坑是因为图省事直接写 SQLite 专有的操作方式把代码和存储引擎绑死。解决绑定问题用 SQLAlchemy 2.0 的 ORM。sqlite 和 postgresql 之间的切换绝大多数场景只需要改一行 database_url 配置。所以选型策略是存储引擎可以先低配数据访问层必须按可迁移的规格来写。模型定义里尽量少用某个数据库专有的字段类型用标准的 String、Text、DateTime以后搬家不折腾。2.4 一张决策表把这次的选择定死决策点我选的东西替代方案为什么这么选编程语言Python 3.11Go、Node.js模型生态 Python 优先类型和异步都够用Web 框架FastAPIFlask、Django原生异步、Pydantic 集成、自动接口文档数据访问SQLAlchemy 2.0raw SQL、Tortoise后续可平滑切到 PostgreSQL出站 HTTPhttpxrequests、aiohttp异步风格统一也能做同步调用配置管理pydantic-settingsos.environ 裸读自动类型转换一个类管所有配置依赖管理uvpoetry、pip venv安装快、锁文件可靠、环境隔离顺手这张表就是地基的宪法以后加依赖之前先对照一下别随口安装。3. 从空目录到可运行服务项目骨架的落地全过程3.1 目录结构每一层放什么边界是什么建目录这件事看起来只是 mkdir实际上是在给未来的自己划地盘。我最终用的是下面这个结构比较克制但每一层的职责都清楚agent-platform/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口注册路由和启动事件 │ ├── config.py # 全局配置 │ ├── database.py # 数据库引擎和会话管理 │ ├── models.py # ORM 模型 │ ├── schemas.py # 请求/响应模型 │ ├── llm_client.py # 大模型接入客户端 │ └── routes/ │ ├── __init__.py │ └── dialogue.py # 对话相关接口 ├── .env.example # 配置模板入库 ├── .gitignore └── pyproject.toml注意一个原则路由里不放业务逻辑模型文件里不放接口字段定义。请求进来路由层负责取参数、调服务、组织响应模型层只负责表结构schema 层管数据校验。这个分层在项目小时看起来有些冗余但它是后面所有功能的地基尤其是多人协作的时候没有边界就是互相踩脚。3.2 初始化动作依赖、配置模板、忽略文件项目目录建好后先做依赖初始化。我用 uv 管理几个命令就完成了cd agent-platform uv init --name agent-platform --python 3.11 uv add fastapi uvicorn[standard] sqlalchemy pydantic pydantic-settings httpx如果你习惯 pip等价的做法是创建虚拟环境后把同样的包名装进去。uv 只是让这个过程更快选哪个取决于你的习惯不要为了工具再开一条学习曲线。接下来写.env.example一个配置模板它的作用是告诉后续接手的人这个项目需要哪些环境变量每个变量长什么样。# .env.example DATABASE_URLsqlite:///./agent_platform.db LLM_BASE_URLhttps://api.example-llm.com/v1 LLM_API_KEY LLM_MODELyour-model-name然后.gitignore里写上.env、.venv/、__pycache__/、*.db这几行。这两件事一起做比任何安全培训都有效。3.3 第一行代码让环境先亮起来我说过这篇的标题是从零开始写第一行代码那这行代码到底是什么我建议不是模型调用而是一个健康检查接口。因为地基阶段第一优先的验证目标是服务能起来、配置能读、依赖能导入。先写配置文件用 pydantic-settings 把环境变量变成类型化配置from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) app_name: str agent-platform database_url: str sqlite:///./agent_platform.db llm_base_url: str llm_api_key: str llm_model: str settings Settings()再看 app/main.py这才是名副其实的第一行代码from fastapi import FastAPI app FastAPI(titleAgent Dialogue Platform, version0.1.0) app.get(/health) async def health(): return {status: ok}启动服务uv run uvicorn app.main:app --reload --port 8000浏览器打开或者 curl 一下 http://127.0.0.1:8000/health 看到{status:ok}地基的第一盏灯就亮了。3.4 验证骨架的几个标准别急着往下写先花两分钟验收四件事。第一配置文件能不能读到 .env 里的值把 LLM_API_KEY 填上再打印 settings.llm_api_key确认非空第二--reload模式下改代码能自动重载第三访问 http://127.0.0.1:8000/docs 能看到自动生成的接口文档说明 FastAPI 的自省能力正常第四端口被占用时切换端口的命令要顺手uvicorn app.main:app --port 8001这种基础操作不用查文档。这四件事过了项目骨架才算真的立住。很多人跳过这步直接往下写数据模型后面遇到为什么代码改了不生效为什么密钥读不到这种问题再回头排查反而更耗时。4. 数据模型先行会话、消息和智能体之间的关系设计4.1 三个核心实体各管一摊对话平台的表结构第一版我建议就三张agents、conversations、messages。它们的关系一句话说清一个智能体可以开多个会话一个会话包含多条消息。agents 存智能体的静态配置名字、系统提示词、默认使用的模型名、温度参数。把这些从对话代码里抽出来是因为以后你可能要做多个角色不同的智能体把配置放表里以后做管理界面、做切换路由都有了落脚点。conversations 存会话属于哪个智能体、标题是什么、什么时候创建的。会话是用户每一次对话的容器也是上下文隔离的单位。messages 存消息属于哪个会话、是谁说的、内容是什么、什么时候说的。消息是流水记录它的顺序就是喂给模型的上下文顺序。4.2 为什么消息要单独建表而不是存一个数组新手最容易犯的错是在 conversations 表里放一个 text 字段把整个历史对话序列化成一个 JSON 字符串塞进去。第一周确实够用但后面问题会接踵而至想查某条消息的内容得先反序列化整段历史想统计用户说了多少话、模型回复了多少字无从下手想实现消息级别的操作比如删除某条不当内容更是折磨。消息单独建表本质是把流水和容器分开。对话是不断追加的记录流会话是它们的归属地这两件事混在一张表里是后续所有上下文相关功能的隐患。代价只是多写几句外键关系但换来的是对每一句话的完整掌控。4.3 用 SQLAlchemy 2.0 把模型写出来先写 database.py统一管理引擎和会话from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, sessionmaker from .config import settings connect_args {} if settings.database_url.startswith(sqlite): connect_args {check_same_thread: False} engine create_engine(settings.database_url, connect_argsconnect_args) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) class Base(DeclarativeBase): passsqlite 的check_same_thread: False是让多线程环境下也能访问同一个数据库连接本地开发必备。然后是 models.py用 SQLAlchemy 2.0 的 Mapped 风格定义from datetime import datetime, timezone from sqlalchemy import DateTime, ForeignKey, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship from .database import Base def utcnow() - datetime: return datetime.now(timezone.utc) class Agent(Base): __tablename__ agents id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] mapped_column(String(50), nullableFalse) system_prompt: Mapped[str] mapped_column(Text, default) model_name: Mapped[str] mapped_column(String(50), default) temperature: Mapped[float] mapped_column(default0.7) created_at: Mapped[datetime] mapped_column(DateTime(timezoneTrue), defaultutcnow) class Conversation(Base): __tablename__ conversations id: Mapped[int] mapped_column(primary_keyTrue) agent_id: Mapped[int] mapped_column(ForeignKey(agents.id), indexTrue) title: Mapped[str] mapped_column(String(100), default新对话) created_at: Mapped[datetime] mapped_column(DateTime(timezoneTrue), defaultutcnow) agent: Mapped[Agent] relationship() messages: Mapped[list[Message]] relationship(back_populatesconversation) class Message(Base): __tablename__ messages id: Mapped[int] mapped_column(primary_keyTrue) conversation_id: Mapped[int] mapped_column(ForeignKey(conversations.id), indexTrue) role: Mapped[str] mapped_column(String(20)) content: Mapped[str] mapped_column(Text) created_at: Mapped[datetime] mapped_column(DateTime(timezoneTrue), defaultutcnow, indexTrue) conversation: Mapped[Conversation] relationship(back_populatesmessages)role 字段目前只有 system、user、assistant 三种值我特意没用数据库级别的枚举因为后续如果接入工具调用消息类型会扩展字符串字段给后期留了余地。content 用 Text 而不是 String因为对话内容没有理由限定长度多轮回复拼起来可能很长。4.4 建表的时机与迁移的债表结构定义好了什么时候真正建表地基阶段我在应用启动事件里做。把 main.py 更新为from contextlib import asynccontextmanager from fastapi import FastAPI from .database import Base, engine from .models import Agent, Conversation, Message # 导入即注册create_all 才知道建哪些表 asynccontextmanager async def lifespan(app: FastAPI): Base.metadata.create_all(bindengine) yield app FastAPI(titleAgent Dialogue Platform, version0.1.0, lifespanlifespan)create_all只建不存在的表不会改已有的表。这个阶段够用但心里要清楚它是一笔债务——等表结构开始频繁改动就要引入迁移工具来管理版本。地基期先不做迁移工具是因为表结构不会大动等第三次结构调整时自然就换过去了。另外一个建议每张表都带上 created_at时间统一用 UTC。对话平台的审计、统计、上下文顺序都依赖时间字段现在嫌麻烦不写后面补数据比写代码痛苦得多。5. 第一个真正能对话的接口从请求到落库的完整链路5.1 一次对话请求流过哪些节点到这一节终于要把对话链路串起来了。用户发一条消息到 POST /api/v1/conversations/{id}/messages系统要依次做五件事校验请求参数取出会话 ID 和用户消息内容把用户消息写入 messages 表按时间顺序查出该会话的全部历史消息组装上下文——系统提示词放最前然后是完整历史最后是这条新消息——调大模型接口拿到回复把回复写入 messages 表返回给前端。这个顺序有一个关键细节用户消息必须先落库再拿去调模型。否则如果模型调用成功但服务随后崩溃用户的这句话就丢了后端与模型之间的一致性无从谈起。5.2 写一个不绑厂商的 LLM 客户端大模型接入做成一个薄薄的类所有厂商差异都被关在这个文件里import httpx class LLMClient: def __init__(self, base_url: str, api_key: str): self.base_url base_url.rstrip(/) self.api_key api_key async def chat(self, messages: list[dict], model: str, temperature: float 0.7) - str: url f{self.base_url}/chat/completions headers {Authorization: fBearer {self.api_key}} payload { model: model, messages: messages, temperature: temperature, } async with httpx.AsyncClient(timeout60.0) as client: response await client.post(url, headersheaders, jsonpayload) response.raise_for_status() data response.json() return data[choices][0][message][content]代码很少但三个点值得说。第一用 httpx 而不是 requests因为要保持 async 风格统一同步和异步场景都能用。第二超时设了 60 秒大模型接口慢是常态预设太短会频繁误报。第三只要服务商兼容主流对话补全协议换厂商只需要改 base_url 和 api_key业务代码对换了模型这件事毫无感知。5.3 对话接口的实现先定义请求和响应 schema用 Pydantic 卡住入参边界from pydantic import BaseModel, Field class MessageCreate(BaseModel): content: str Field(min_length1, max_length16000) class MessageResult(BaseModel): role: str assistant content: str然后是核心的对话路由完整代码在 app/routes/dialogue.pyfrom fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select from sqlalchemy.orm import Session from ..config import settings from ..database import SessionLocal from ..llm_client import llm_client from ..models import Agent, Conversation, Message from ..schemas import MessageCreate, MessageResult router APIRouter(prefix/api/v1, tags[dialogue]) def get_db(): db SessionLocal() try: yield db finally: db.close() router.post(/conversations, status_code201) def create_conversation(db: Session Depends(get_db)): agent db.scalar(select(Agent).limit(1)) if agent is None: raise HTTPException(status_code400, detail还没有可用智能体请先创建) conversation Conversation(agent_idagent.id) db.add(conversation) db.commit() db.refresh(conversation) return {id: conversation.id, agent_id: conversation.agent_id} router.post(/conversations/{conversation_id}/messages, response_modelMessageResult) async def send_message(conversation_id: int, payload: MessageCreate, db: Session Depends(get_db)): conversation db.get(Conversation, conversation_id) if conversation is None: raise HTTPException(status_code404, detail会话不存在) db.add(Message(conversation_idconversation.id, roleuser, contentpayload.content)) db.commit() history db.scalars( select(Message) .where(Message.conversation_id conversation.id) .order_by(Message.created_at.asc()) ).all() messages [] if conversation.agent and conversation.agent.system_prompt: messages.append({role: system, content: conversation.agent.system_prompt}) messages.extend({role: m.role, content: m.content} for m in history) model_name settings.llm_model or (conversation.agent.model_name if conversation.agent else ) if not model_name: raise HTTPException(status_code500, detail未配置模型名称) try: reply await llm_client.chat( messages, modelmodel_name, temperatureconversation.agent.temperature if conversation.agent else 0.7, ) except Exception: raise HTTPException(status_code502, detail模型服务暂时不可用请稍后重试) db.add(Message(conversation_idconversation.id, roleassistant, contentreply)) db.commit() return MessageResult(contentreply)细看会发现我调模型前先把用户消息 commit 了调完模型又把助手消息落库。这两次写库不能合并因为中间隔着一次不确定的模型调用。如果模型失败用户消息已经存在接口返回 502前端可以提示重试至少不会出现用户说了一句话但数据库里没记录的可怕一致性问题。还需要在 main.py 里把路由挂上去from .routes.dialogue import router app FastAPI(titleAgent Dialogue Platform, version0.1.0, lifespanlifespan) app.include_router(router)5.4 用 curl 真实跑一轮启动服务后先建一个会话curl -X POST http://127.0.0.1:8000/api/v1/conversations返回类似{id:1,agent_id:1}。然后发消息curl -X POST http://127.0.0.1:8000/api/v1/conversations/1/messages \ -H Content-Type: application/json \ -d {content:你好请用一句话介绍你自己}第一次调用可能等几秒到十几秒这是正常的。返回的结构类似{role: assistant, content: 你好我是一个由大模型驱动的对话助手可以陪你聊天并帮你处理问题。}这一步看到真实回复地基搭建才算真正完成了闭环。你再发第二条消息模型就能根据第一条的上下文回答了——因为每一次请求都会带上该会话的全部历史消息。这也验证了数据模型的设计历史是从 messages 表里按顺序读出来的不是写在代码里的局部变量。6. 接口通了以后马上补的三块短板6.1 错误处理别让堆栈裸奔到前端第一次用 curl 调通是快乐的但快乐通常维持不了多久因为下一秒你就会开始测试异常路径。模型超时、网络抖动、请求被限流、接口返回错误格式这些在你上线前总会遇到。现在这个版本的问题在于异常一旦裸抛FastAPI 会返回 500前端收到一大串堆栈。用户看不懂也不该看。我的处理办法是在 llm_client 里把非预期的错误统一包装成自定义异常路由层只对自定义异常做捕获对外返回统一的错误体。前面代码里已经用 except 兜了一层后续可以把它抽成一个自定义异常类加一个统一的异常处理器让所有模型调用失败都返回同一个格式状态码 502消息模型服务暂时不可用。这里有个实用的取舍超时和限流其实值得做一次简单重试但地基阶段我建议先不做自动重试。原因很简单对话场景里用户能接受失败以后手动再发一次自动重试引入的重复消息风险比它带来的收益大。等做批量或者定时任务场景时再单独设计重试机制。6.2 密钥进 .env连垃圾桶都不收这个坑我见过太多次了api_key 直接写在代码里然后一不留神提交到仓库配置外泄账单爆炸。地基期就把规矩立住任何密钥不进代码文件只进环境变量和 .env而 .env 文件从第一天起就在 .gitignore 里。.env.example只保留字段名和占位符比如LLM_API_KEY给人看格式但不给真实值。收到新机器或者同事克隆仓库第一步是cp .env.example .env然后填自己的真实配置。这样密钥的流转路径是受控的而不是跟着源代码到处传。配置读取还有个次序问题要清楚pydantic-settings 的优先级是真实环境变量 .env 文件 代码里的默认值。所以生产环境把密钥放在部署平台的环境变量里本地开发放在 .env 里两者互不干扰代码一行都不用改。6.3 结构化日志让每个请求有迹可循前端再来找你对账的时候你会感谢日志。地基阶段的日志要求不用高但至少每一条请求要能回答四个问题调的什么接口、哪个会话产生、模型调用花了多久、成功还是失败。我用标准 logging 就能满足关键是输出格式。别只打印字符串尽量打成结构化的一行将来接日志收集平台时直接按字段检索ts2025-01-15T10:23:45Z levelINFO eventmessage_sent conversation_id1 roleuser ts2025-01-15T10:24:05Z levelINFO eventllm_call conversation_id1 latency_ms20300 modelyour-model-name顺手还能把模型接口返回的 token 用量记录下来这是以后做成本统计和配额管理的原始数据。我建议在 llm_client 的返回里把 usage 一并带出来路由层负责写日志模型层不碰日志逻辑职责又分了一层。6.4 留一道简单的门访问鉴权对话平台是面向用户的但地基期可以先不搞复杂的用户体系不过也不能裸奔。一个成本最低的方案是加一道 API Key 校验请求头里带 X-API-Key和配置里的值一致才放行。FastAPI 里用一个依赖实现挂在路由上from fastapi import Header, HTTPException from ..config import settings def require_api_key(x_api_key: str Header(default)) - None: if settings.api_key and x_api_key ! settings.api_key: raise HTTPException(status_code401, detailunauthorized)health 接口不挂这个依赖方便运维探测。这只是一个过渡方案等真上了用户体系再把它换成登录态和权限校验。地基阶段留一道门是为了防止自己在开发过程中把带着真实数据的服务暴露到公网。这三块短板补齐之后这套地基基本就合格了。它现在还很朴素没有工具调用没有多智能体没有流式对话但它的骨架、数据、配置、链路、日志、鉴权都立住了后面任何高级功能都不是在废墟上打补丁而是往这套结构里加房间。我个人在两三个项目上反复打磨过这套地基最深的体会是第一次敲下的那行app FastAPI(...)看起来平平无奇但它把一堆悬而未决的决策变成了一台能跑起来的机器。后面每次加新功能回到 models 和 llm_client 的次数屈指可数大部分时间都在 routes 层写代码。这就是地基该有的样子——前期多花一小时理清边界后面少熬不知道多少个通宵。下一篇可以放心去聊工具调用了。