汉字谜盒-核心功能开发 | 团子和蛋糕的笔记

五个接口总览

接口	方式	参数	说明
`/api/sessions`	POST	无	新建会话，返回会话标识
`/api/chat`	POST	`{session_id, message}`	与 AI 交互，返回 AI 的回复
`/api/sessions`	GET	无	获取会话列表（按时间倒序）
`/api/sessions/{session_id}`	GET	会话 ID（路径参数）	加载指定会话
`/api/sessions/{session_id}`	DELETE	会话 ID（路径参数）	删除指定会话

统一响应格式（三个字段）：

json

{ "code": 200, "message": "创建会话成功", "data": "2026-04-12_16-16-13" }

用 Pydantic 定义数据模型

python

from typing import Any
from pydantic import BaseModel

# 统一响应模型
class ApiResponse(BaseModel):
    code: int
    message: str
    data: Any          # 任意类型的数据

# 与 AI 交互的请求模型
class ChatRequest(BaseModel):
    session_id: str
    message: str

IMPORTANT

BaseModel 是 Pydantic 库提供的父类（FastAPI 深度集成了 Pydantic），用于定义数据模型和数据验证规则。定义好 ChatRequest 后，接口函数写成 def chat(request: ChatRequest)，FastAPI 就会自动把请求体里的 JSON 解析成这个对象，并校验字段类型——不用自己 json.loads。（前端传的就是 {"session_id": "...", "message": "你好"}。）

接口一：新建会话

打开页面时如果没有会话，就自动创建一个，会话标识形如 2026-04-15_12-00-05。

python

def generate_session_id():
    return datetime.now().strftime("%Y-%m-%d_%H-%M-%S")

@app.post("/api/sessions")
def create_session() -> ApiResponse:
    session_id = generate_session_id()

    session_data = {"current_session": session_id, "messages": []}
    with open(f"sessions/{session_id}.json", "w", encoding="utf-8") as f:
        json.dump(session_data, f, ensure_ascii=False, indent=2)

    return ApiResponse(code=200, message="创建会话成功", data=session_id)

接口二：与 AI 交互（核心）

请求体：{"session_id": "2026-04-15_12-00-05", "message": "你好"}

处理流程七步：

python

@app.post("/api/chat")
def chat(request: ChatRequest) -> ApiResponse:
    # 1. 加载 json 文件中的会话数据
    session_path = f"sessions/{request.session_id}.json"
    with open(session_path, "r", encoding="utf-8") as f:
        session_data = json.load(f)

    # 2. 构建给大模型的消息列表（system + 历史 + 本轮提问）
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for message in session_data["messages"]:
        messages.append(message)
    messages.append({"role": "user", "content": request.message})

    # 3. 调用 AI 大模型
    response = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=messages,
        stream=False,
        temperature=1.0,
    )

    # 4. 获取响应的数据
    ai_response = response.choices[0].message.content

    # 5. 更新消息列表（把 system 摘掉，把 AI 回复追加进去）
    messages.pop(0)
    messages.append({"role": "assistant", "content": ai_response})
    session_data["messages"] = messages

    # 6. 保存会话信息到 json 文件
    with open(session_path, "w", encoding="utf-8") as f:
        json.dump(session_data, f, ensure_ascii=False, indent=2)

    # 7. 返回数据
    return ApiResponse(code=200, message="请求成功", data=ai_response)

NOTE

第 5 步为什么要 messages.pop(0)？ 因为落盘保存的是"会话历史"，只需要 user 和 assistant 的往来内容；system 是每次请求时临时加上去的（第 2 步），不该混进历史里，否则下次加载会重复叠加。

WARNING

课程的代码里 model 写的是 "deepseek-v4-flash"。按 DeepSeek 官方文档（2026-09 查证），当前可用的是 deepseek-flash（对应 DeepSeek-V4.1-Flash）和 deepseek-v4-pro；deepseek-v4-flash 属于已被兼容的旧 id，仍能调用但会被路由到新模型；而更早的 deepseek-chat / deepseek-reasoner 已经不在文档的模型列表里。写新代码时建议用文档上的当前 id，具体以官方文档为准。

接口三~五：会话列表 / 加载 / 删除

python

# 会话列表：取出 sessions 目录下所有文件名，去掉后缀，倒序
@app.get("/api/sessions")
def get_sessions() -> ApiResponse:
    session_files = os.listdir("sessions")
    session_ids = [file.split(".")[0] for file in session_files]
    session_ids.sort(reverse=True)
    return ApiResponse(code=200, message="获取会话列表成功", data=session_ids)

# 加载指定会话：路径参数 session_id
@app.get("/api/sessions/{session_id}")
def get_session(session_id: str) -> ApiResponse:
    with open(f"sessions/{session_id}.json", "r", encoding="utf-8") as f:
        session_data = json.load(f)
    return ApiResponse(code=200, message="获取会话信息成功", data=session_data)

# 删除指定会话
@app.delete("/api/sessions/{session_id}")
def delete_session(session_id: str) -> ApiResponse:
    session_file = f"sessions/{session_id}.json"
    if os.path.exists(session_file):
        os.remove(session_file)
    return ApiResponse(code=200, message="删除会话成功", data=None)

TIP

路径参数：/api/sessions/{session_id} 里的花括号部分是变量，要原样写进函数形参（def get_session(session_id: str)），FastAPI 会自动把 URL 里那一段取出来传给你。这也是 REST 风格"URL 定位资源"的落地：同一个路径，靠不同的请求方式（GET/DELETE）做不同的事。

练习题

一、知识回顾（读完直接做下面的实践题）

统一响应格式的三个字段：code、message、data
定义数据模型需要继承 Pydantic 的 BaseModel 类；FastAPI 会据此自动解析请求体 JSON 并做类型校验（数据验证）
与 AI 交互的请求模型 ChatRequest 有两个字段：session_id 和 message
与 AI 交互七步：加载会话 json → 构建消息列表 → 调用大模型 → 获取响应 → 更新消息列表 → 保存 json → 返回数据
第 5 步用 messages.pop(0) 把 system 消息摘掉，因为它只是本次请求临时加上去的
路径参数写在 URL 的花括号 {} 里，并且要写进函数形参
会话列表接口用 os.listdir("sessions") 列出文件，再用 sort(reverse=True) 按时间倒序
删除会话前先判断文件是否存在：os.path.exists(path)，存在才 os.remove(path)

二、裸写题

2-1 新建会话接口 写 POST /api/sessions：生成以秒为单位的会话标识，创建 sessions/{id}.json（内含 current_session 与空 messages），返回统一格式。

- 提示（先自己想，实在想不出再点开）

一级 · 思路：生成标识 → 写文件 → 返回 二级 · 方法：datetime.now().strftime("%Y-%m-%d_%H-%M-%S") / json.dump(..., ensure_ascii=False, indent=2)三级 · 骨架：return ApiResponse(code=200, message="创建会话成功", data=session_id)

- 参考答案（做完再点开）

python

import os
import json
from datetime import datetime
from typing import Any
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="会话接口练习")

if not os.path.exists("sessions"):
    os.mkdir("sessions")

class ApiResponse(BaseModel):
    code: int
    message: str
    data: Any

def generate_session_id():
    return datetime.now().strftime("%Y-%m-%d_%H-%M-%S")

# 2-1
@app.post("/api/sessions")
def create_session() -> ApiResponse:
    session_id = generate_session_id()
    session_data = {"current_session": session_id, "messages": []}
    with open(f"sessions/{session_id}.json", "w", encoding="utf-8") as f:
        json.dump(session_data, f, ensure_ascii=False, indent=2)
    return ApiResponse(code=200, message="创建会话成功", data=session_id)

2-2 会话列表接口 写 GET /api/sessions：列出 sessions 目录下所有会话标识（去掉 .json 后缀），按时间倒序返回。

- 提示（先自己想，实在想不出再点开）

一级 · 思路：列目录 → 去后缀 → 排序 二级 · 方法：os.listdir / file.split(".")[0] / sort(reverse=True)三级 · 骨架：因为会话标识本身就是时间戳格式，字符串倒序就等于时间倒序

- 参考答案（做完再点开）

python

# 2-2
@app.get("/api/sessions")
def get_sessions() -> ApiResponse:
    session_ids = [f.split(".")[0] for f in os.listdir("sessions")]
    session_ids.sort(reverse=True)
    return ApiResponse(code=200, message="获取会话列表成功", data=session_ids)

@app.delete("/api/sessions/{session_id}")
def delete_session(session_id: str) -> ApiResponse:
    path = f"sessions/{session_id}.json"
    if os.path.exists(path):
        os.remove(path)
    return ApiResponse(code=200, message="删除会话成功", data=None)

2-3 用一个测试请求走通流程 用 FastAPI 自带的测试客户端：先 POST 新建会话拿到 id，再用这个 id 调 GET 列表（确认列表里有它），最后 DELETE 掉。
- 提示（先自己想，实在想不出再点开）
一级 · 思路：三个接口串成一条链路，不启动服务器也能测 二级 · 方法：client.post("/api/sessions") / client.get("/api/sessions") / client.delete(...)三级 · 骨架：session_id = resp.json()["data"]，再拼进 URL
- 参考答案（做完再点开）
python
```
# 2-3 链路测试
from fastapi.testclient import TestClient

if __name__ == "__main__":
    client = TestClient(app)
    resp = client.post("/api/sessions")
    session_id = resp.json()["data"]
    print("新建:", resp.json())

    print("列表:", client.get("/api/sessions").json())
    print("删除:", client.delete(f"/api/sessions/{session_id}").json())
    print("删除后列表:", client.get("/api/sessions").json())
```

五个接口总览 ​

用 Pydantic 定义数据模型 ​

接口一：新建会话 ​

接口二：与 AI 交互（核心） ​

接口三~五：会话列表 / 加载 / 删除 ​

相关 ​

练习题 ​

一、知识回顾（读完直接做下面的实践题） ​

二、裸写题 ​

五个接口总览

用 Pydantic 定义数据模型

接口一：新建会话

接口二：与 AI 交互（核心）

接口三~五：会话列表 / 加载 / 删除

相关

练习题

一、知识回顾（读完直接做下面的实践题）

二、裸写题