Day16 别急着让AI开口：先给《AI智能伴侣》搭一套能跑的服务端

摘要： 这篇从 FastAPI 的最小接口一路搭到《AI智能伴侣》的完整聊天链路：先用 Pydantic 约束请求和响应，再补齐增删改查、Swagger 验证与同步异步的边界；接着整理项目目录，让 Nginx 或 FastAPI 把前端送到浏览器；最后从本地预设文件读到 /api/presets，把昵称和性格拼进系统提示词，通过 OpenAI 兼容接口调用 DeepSeek，并在页面上完成第一次单轮聊天。这一版先不做对话记忆，代码、验收和后续改造点都按这个边界写清楚。

做一个会聊天的 AI，表面上只差“调一下模型”这一步。真正把代码写起来，会遇到一连串更具体的问题：前端该向哪个地址取预设？请求字段漏了怎么办？Swagger 里为什么查得到数据，页面却不动？浏览器等模型回复时，服务端还能不能接收别的请求？

先把这些环节跑通，再谈记忆和长期陪伴。这篇文章就沿着“接口能测、页面能开、角色能选、模型能回”这条线，把《AI智能伴侣》的服务端搭出来。

1. 先看整套 Web 项目怎么转起来

这个项目分成三块。前端负责聊天界面、角色下拉框和消息展示；服务端用 FastAPI 处理业务、读取本地文件并调用模型；数据库暂时不接，伴侣预设先放在 data/companion_presets.json 里。以后要接会话持久化，再替换这一层存储即可。

项目入口不是一个孤零零的 HTML 文件。浏览器先访问前端首页，页面加载后再请求 /api/presets 获取角色；用户发出消息时，前端向 /api/chat 提交昵称、性格和当前消息；FastAPI 拼好提示词，调用 DeepSeek，把回复装进统一响应送回页面。

本地开发有两种前端启动方式。使用资料里的 Nginx 时，把页面放到静态目录，用 location /api 把接口请求转发到 8000 端口；也可以把前端资源收进项目，由 FastAPI 的 StaticFiles 一并提供。两种方式面对的是同一组后端接口，先把接口路径定稳，再换部署方式就不用重写业务。

2. 从一条最小接口开始

FastAPI 最直观的用法，是“装饰器绑定路径 + 函数返回数据”。下面这个 /users 接口只用来确认服务能启动、请求能到达函数、返回值能变成 JSON：

import uvicorn
from fastapi import FastAPI

app = FastAPI(title="AI智能伴侣")

@app.get("/users", summary="查询用户列表")
def get_users():
    return [
        {"id": 1, "name": "小王", "age": 18},
        {"id": 2, "name": "张三", "age": 28},
    ]

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

@app.get("/users") 把 GET /users 交给下面的函数处理，框架会负责把 Python 列表序列化成 JSON。开发时还可以执行 uvicorn main:app --reload，代码保存后让开发服务器重新加载。入口放进 if __name__ == "__main__":，是为了导入模块时不会顺手再启动一次服务器。

接口跑通以后，再回浏览器或 Swagger，确认 http://localhost:8000/users 真的返回数组。这个看似简单的小检查很重要，后面页面不动时，先分清是接口没通，还是前端没取对数据。

3. 把数据从字典搬进 Pydantic 模型

一条接口用字典最省事，但项目很快就会需要查询、新增、修改和删除用户。字典的键没有编译器约束，name 拼成 nmae、年龄传成字符串、某次漏传字段，都可能一路混进业务代码。

Pydantic 用类型注解描述数据结构，FastAPI 接上后会自动校验请求和返回值：

from pydantic import BaseModel, Field

class User(BaseModel):
    id: int
    name: str = Field(min_length=2, max_length=20)
    age: int = Field(ge=0, le=120)

这里不止告诉了读者 age 应该是什么，也告诉了框架“年龄必须在 0 到 120 之间”。Field(description=...) 还会写进 Swagger 文档。可选字段写成 str | None = None；创建模型实例时使用 User(id=1, name="小王", age=18)，关键字段一眼能看清。

接口开始变多时，可以保留同一个最小项目，按下面的路由约定继续扩展：

请求	路径	用途
GET	/users	查询用户列表
GET	/users/{user_id}	查询单个用户
POST	/users	新增用户
PUT	/users/{user_id}	修改用户
DELETE	/users/{user_id}	删除用户

URL 定位资源，HTTP 方法表达动作。复数和具体处理方式都属于接口约定，约定越稳定，前后端联调时越少来回猜。

4. 让接口重复、失败和增删改查都真正走通

用户列表适合演示“多个相同字段”的数据。下面用内存列表让整组接口能够运行，重点是观察路由、状态码和请求体的关系；真实项目再把这一层换成数据库：

from fastapi import FastAPI, HTTPException, Response, status

app = FastAPI(title="AI智能伴侣")
users = {
    1: {"id": 1, "name": "小王", "age": 18},
    2: {"id": 2, "name": "张三", "age": 28},
}

@app.get("/users/{user_id}", summary="查询单个用户")
def get_user(user_id: int):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="用户不存在")
    return users[user_id]

@app.post("/users", status_code=201, summary="新增用户")
def create_user(user: User):
    if user.id in users:
        raise HTTPException(status_code=409, detail="用户编号已存在")
    users[user.id] = user.model_dump()
    return users[user.id]

@app.put("/users/{user_id}", summary="修改用户")
def update_user(user_id: int, user: User):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="用户不存在")
    if user.id != user_id:
        raise HTTPException(status_code=400, detail="编号不能修改")
    users[user_id] = user.model_dump()
    return users[user_id]

@app.delete(
    "/users/{user_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="删除用户",
)
def delete_user(user_id: int):
    if users.pop(user_id, None) is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return Response(status_code=status.HTTP_204_NO_CONTENT)

新增成功用 201，删除成功用 204，资源不存在返回 404，编号冲突返回 409。Swagger 里逐个点开这些接口，能直接看到请求模型、状态码和错误信息。新增同一个编号、传入年龄 130、删除不存在的编号，都应该得到明确反馈，而不是服务器悄悄报错。

这里用 user.model_dump() 把模型转成普通字典，方便放进演示用的内存结构。真实业务中建议直接使用数据库模型，并让 Pydantic 只负责接口边界，避免把业务数据散落在多个层里。

5. 异步要写到等待点上，不能只改函数名

聊天接口需要等模型生成回复，这段时间很适合讲同步与异步。先把三件耗时任务写成同一个对照实验：

import asyncio
import time

def sync_task(task_id: int):
    time.sleep(1)
    return task_id

async def async_task(task_id: int):
    await asyncio.sleep(1)
    return task_id

def run_sync():
    start = time.perf_counter()
    for task_id in range(3):
        sync_task(task_id)
    return time.perf_counter() - start

async def run_async():
    start = time.perf_counter()
    await asyncio.gather(*(async_task(task_id) for task_id in range(3)))
    return time.perf_counter() - start

同步写法完成三个任务大约要 3 秒；asyncio.gather 能把三个等待重叠，总时长接近 1 秒。差别不在函数前面有没有 async，而在代码是否真的遇到 await 并让出执行权。

放到 FastAPI 里，async def 负责声明协程函数，await 负责标记可以等待的 I/O。如果协程里仍然调用了 time.sleep 或某个同步阻塞 SDK，事件循环还是会被占住。对模型 SDK，需要按当前版本的异步用法与部署方式实测；不能只看接口写了 async def，就断定整条调用链已经非阻塞。

6. 给每个接口装同一个响应外壳

伴侣预设、聊天回复、文件缺失和参数错误，如果各返回各的格式，前端就得为每个接口单独判断。项目约定一套响应结构：

{
  "code": 200,
  "message": "加载预设信息成功",
  "data": []
}

对应的 Pydantic 模型可以写成：

from typing import Any
from pydantic import BaseModel, Field

class ApiResponse(BaseModel):
    code: int = Field(default=200, description="业务状态码")
    message: str | None = Field(default=None, description="结果说明")
    data: Any = Field(default=None, description="业务数据")

data 使用 Any，是为了让一个外壳能够承载预设数组、聊天文本或空结果。若项目需要更强的类型安全，也可以为每个业务接口定义专用的 data 模型。统一重点是“前端先看结果说明，再读取业务数据”，错误分支同样返回 ApiResponse。

注意，HTTP 成功状态码和响应体里的业务码不是同一个概念。接口成功调用但预设文件不存在，HTTP 层仍可能返回 200，而 code 是 404；也可以直接把 HTTP 状态码设成 404。两种设计都能用，关键是在项目里保持一致，并让前端明确按哪种规则判断。

7. 搭好项目目录，分清前端从哪来

在 ai_partner 项目里，可以把代码、数据和资源放在清楚的边界内：

ai_partner/
├── main.py
├── data/
│   └── companion_presets.json
└── static/
    └── index.html

data 放后端读取的预设文件，static 放页面、样式和脚本。Path(__file__).resolve().parent 能把路径锚定到 Python 文件所在目录，不必依赖在哪个终端目录启动服务。

选择 Nginx 时，页面用 88 端口打开，/api 请求转发给 8000。核心配置类似这样：

server {
    listen 88;
    server_name localhost;

    location / {
        root html;
        index index.html;
    }

    location /api {
        proxy_pass http://localhost:8000;
    }
}

这样浏览器打开的是 http://localhost:88，并不会因为后端跑在 8000 就必须直接访问后端首页。Nginx 负责静态文件与代理，FastAPI 只管 /api 下的业务接口。

如果不想再开 Nginx，也可以把前端目录挂进 FastAPI：

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI(title="AI智能伴侣")
app.mount("/", StaticFiles(directory="static", html=True), name="static")

html=True 让访问 / 时找到 index.html。挂载静态资源后，分别打开页面和 /docs，确认页面资源与接口文档都能访问。

8. 第一个功能：把预设伴侣送回页面

预设文件中，每条角色通常包含 id、name、nick_name、nature 和 sort_order。页面需要按照 sort_order 显示角色，因此服务端读取后要先排序，再装进统一响应。

import json
from pathlib import Path

PRESETS_FILE = (
    Path(__file__).resolve().parent / "data" / "companion_presets.json"
)

@app.get(
    "/api/presets",
    summary="获取伴侣预设信息列表",
    response_model=ApiResponse,
)
def get_presets():
    try:
        presets = json.loads(PRESETS_FILE.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return ApiResponse(code=404, message="预设信息文件不存在", data=[])
    except json.JSONDecodeError:
        return ApiResponse(code=500, message="预设信息格式错误", data=[])

    presets.sort(key=lambda item: item["sort_order"])
    return ApiResponse(
        code=200,
        message="加载预设信息成功",
        data=presets,
    )

路径固定到项目目录，文件丢失时返回可读错误，JSON 写坏时也不会把原始堆栈直接甩给页面。先打开 http://localhost:8000/docs 调 /api/presets，确认数组顺序正确；再打开页面，确认下拉框真的读到了这些角色。只看到页面标题，不代表数据已经加载成功。

9. 选中的角色怎样变成系统提示词

客户端传来昵称和性格，服务端把这两项填进自己的模板。这样前端只负责选择角色，真正的回复规则仍然由服务端控制：

SYSTEM_PROMPT_TEMPLATE = """你叫 {nick_name}，正在和用户进行日常聊天。

规则：
1. 每次只回复一条完整消息。
2. 不描述场景，不替用户补写对话。
3. 回复简短、自然，像日常聊天。
4. 语气和内容要符合下面的伴侣性格。

伴侣性格：
{nature}
"""

检查提示词，不是看它写得多长，而是看“叫什么、怎么说、不要做什么”有没有写清。源资料中的“完全代入真实伴侣”属于演示设定；面向真实用户的产品应明确标注 AI 身份，不把虚拟角色包装成真人服务，并为不适合的情感依赖留出边界。

模型调用需要区分角色消息和用户消息。当前 OpenAI Python SDK 提供了类型化消息参数：

from openai.types.chat import (
    ChatCompletionSystemMessageParam,
    ChatCompletionUserMessageParam,
)

system_prompt = SYSTEM_PROMPT_TEMPLATE.format(
    nick_name=request.nick_name,
    nature=request.nature,
)

messages = [
    ChatCompletionSystemMessageParam(
        role="system",
        content=system_prompt,
    ),
    ChatCompletionUserMessageParam(
        role="user",
        content=request.message,
    ),
]

不要接受客户端直接传来的整段系统指令，否则人设和边界都可以被页面绕过。昵称和性格也应当做长度与内容校验；模板里的用户数据不是程序指令，不能让角色文件控制服务端的行为。客户端请求模型只保留当前对话需要的字段：

from pydantic import BaseModel, Field

class ChatRequest(BaseModel):
    session_name: str | None = Field(default=None, description="会话名称")
    message: str = Field(min_length=1, max_length=2000, description="用户消息")
    nick_name: str = Field(min_length=1, max_length=20, description="伴侣昵称")
    nature: str = Field(min_length=1, max_length=500, description="伴侣性格")

session_name 可以留给前端组织不同会话，但这一版没有按它读取或保存历史。

10. 接上 DeepSeek，但记住只发了两条消息

DeepSeek 提供 OpenAI 兼容接口，可以先沿用 OpenAI Python SDK，只更换服务地址和密钥。下面仍然只展示聊天主流程，模型参数应与当前服务端文档核对：

import os

from fastapi import HTTPException
from openai import OpenAI

def call_deepseek(request: ChatRequest):
    client = OpenAI(
        api_key=os.environ["DEEPSEEK_API_KEY"],
        base_url="https://api.deepseek.com",
        timeout=30.0,
    )
    system_prompt = SYSTEM_PROMPT_TEMPLATE.format(
        nick_name=request.nick_name,
        nature=request.nature,
    )
    messages = [
        ChatCompletionSystemMessageParam(
            role="system",
            content=system_prompt,
        ),
        ChatCompletionUserMessageParam(
            role="user",
            content=request.message,
        ),
    ]
    return client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        stream=False,
    )

@app.post(
    "/api/chat",
    summary="AI伴侣对话",
    response_model=ApiResponse,
)
async def chat(request: ChatRequest):
    if not os.environ.get("DEEPSEEK_API_KEY"):
        raise HTTPException(status_code=503, detail="DeepSeek 服务密钥未配置")

    try:
        completion = await asyncio.to_thread(call_deepseek, request)
    except Exception as exc:
        return ApiResponse(
            code=502,
            message=f"模型调用失败：{type(exc).__name__}",
        )

    return ApiResponse(
        code=200,
        message="AI对话成功",
        data=completion.choices[0].message.content,
    )

call_deepseek 使用同步 SDK 调用，交给 asyncio.to_thread 执行，避免把同步网络等待直接卡在事件循环里。这是这一版容易跟做的保底方案，不是对所有 SDK、并发规模和部署模式的统一答案。如果当前 SDK 提供经过验证的异步方法，可以优先使用；如果代码路径不适合异步，也可以让 FastAPI 的执行模型去处理同步函数，最终按 SDK 版本与真实并发测试决定。

messages 只有系统消息与当前用户消息。用户下一句问“那它呢？”时，上一轮的问题和回复都已经不在当前请求里。带上 session_name 不会自动得到上下文；要让伴侣接住上一句，必须另外保存历史，并在下一次请求里按会话加载回来。

模型名要以当前 DeepSeek 文档为准。源资料出现过 deepseek-flash、deepseek-v4-flash 等写法，文中使用兼容接口常见的 deepseek-chat 作为示例；可用模型、计费和思考模式都应当回到当前文档确认。密钥只放服务端环境变量，不能写进页面、Markdown 或仓库。

11. 最后沿着页面、接口和模型逐层验收

开发时按这个顺序检查，定位问题会快很多：

1. 打开 /docs 调 /api/presets，确认角色数组非空且已经按顺序排序。
2. 打开前端首页，确认下拉框选项与 JSON 中的角色一致。
3. 选一个角色发送明确的消息，确认页面拿到的是 ApiResponse.data 中的回复文本。
4. 换一个性格明显不同的角色再聊，观察口吻是否随系统提示词改变。
5. 再追问一句依赖上一轮的话，确认当前版本确实只按单轮回复。
6. 临时从请求中删掉 message、nick_name 或 nature，确认 Pydantic 在调用模型前就拦住错误。
7. 文件缺失、JSON 损坏和密钥缺失时，分别确认页面能收到可理解的失败结果。

再准备几条不适合直接重试的失败情况：密钥错误、模型超时、上游限流、SDK 异常。演示版本可以只返回错误类型，正式产品还要写日志、做有限重试与超时控制，不能把所有异常都包装成“操作成功”。

就这一版来说，前端有预设数据，后端有统一接口，聊天有人设，模型调用有清楚的失败边界。它还不是完整的陪伴产品：对话记忆、会话持久化、上下文裁剪、错误重试与更细的安全限制，都应当在下一步按顺序补上。先把这次单轮聊天跑稳，记忆接进来时，才不会一边修接口一边猜上下文的格式。

创作不易，坚持难熬，留下个赞再走呗！