汉字谜盒-核心功能开发
汉字谜盒-核心功能开发
五个接口的开发：新建会话、与 AI 交互（Pydantic 数据模型 + 调用大模型七步）、会话列表、加载与删除会话
五个接口总览
| 接口 | 方式 | 参数 | 说明 |
|---|---|---|---|
/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（路径参数） | 删除指定会话 |
统一响应格式（三个字段）：
{ "code": 200, "message": "创建会话成功", "data": "2026-04-12_16-16-13" }用 Pydantic 定义数据模型
from typing import Anyfrom pydantic import BaseModel
# 统一响应模型class ApiResponse(BaseModel): code: int message: str data: Any # 任意类型的数据
# 与 AI 交互的请求模型class ChatRequest(BaseModel): session_id: str message: strBaseModel 是 Pydantic 库提供的父类（FastAPI 深度集成了 Pydantic），用于定义数据模型和数据验证规则。
定义好 ChatRequest 后，接口函数写成 def chat(request: ChatRequest)，FastAPI 就会自动把请求体里的 JSON 解析成这个对象，并校验字段类型——不用自己 json.loads。
（前端传的就是 {"session_id": "...", "message": "你好"}。）
接口一：新建会话
打开页面时如果没有会话，就自动创建一个，会话标识形如 2026-04-15_12-00-05。
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": "你好"}
处理流程七步：
@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)第 5 步为什么要 messages.pop(0)？ 因为落盘保存的是”会话历史”，只需要 user 和 assistant 的往来内容；system 是每次请求时临时加上去的（第 2 步），不该混进历史里，否则下次加载会重复叠加。
课程的代码里 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，具体以官方文档为准。
接口三~五：会话列表 / 加载 / 删除
# 会话列表：取出 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)路径参数：/api/sessions/{session_id} 里的花括号部分是变量，要原样写进函数形参（def get_session(session_id: str)），FastAPI 会自动把 URL 里那一段取出来传给你。这也是 REST 风格”URL 定位资源”的落地：同一个路径，靠不同的请求方式（GET/DELETE）做不同的事。
相关
练习题
一、回忆填空（写完再展开对答案）
- 统一响应格式的三个字段：
code、____、____ - 定义数据模型需要继承 Pydantic 的
____类；FastAPI 会据此自动解析请求体 JSON 并做 ____ 校验 - 与 AI 交互的请求模型
ChatRequest有两个字段：____和____ - 与 AI 交互七步：加载会话 json → 构建 ____ → 调用 ____ → 获取响应 → 更新消息列表 → 保存 json → 返回数据
- 第 5 步用
messages.____(0)把system消息摘掉，因为它只是本次请求临时加上去的 - 路径参数写在 URL 的 ____ 里，并且要写进函数 ____
- 会话列表接口用
os.____("sessions")列出文件，再用sort(reverse=____)按时间倒序 - 删除会话前先判断文件是否存在：
os.path.____(path)，存在才os.____(path)
填空答案（做完再点开）
- message / data 2. BaseModel / 类型（数据验证） 3.
session_id/message4. 消息列表 / 大模型 5. pop 6. 花括号{}/ 形参 7. listdir / True 8. exists / remove
二、裸写题
-
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) -
2-2 会话列表接口 写
GET /api/sessions：列出sessions目录下所有会话标识（去掉.json后缀），按时间倒序返回。提示一级 · 思路：列目录 → 去后缀 → 排序 二级 · 方法：
os.listdir/file.split(".")[0]/sort(reverse=True)三级 · 骨架：因为会话标识本身就是时间戳格式，字符串倒序就等于时间倒序 -
2-3 用一个测试请求走通流程 用
TestClient：先 POST 新建会话拿到 id，再用这个 id 调 GET 列表（确认列表里有它），最后 DELETE 掉。提示一级 · 思路：三个接口串成一条链路，不启动服务器也能测 二级 · 方法：
client.post("/api/sessions")/client.get("/api/sessions")/client.delete(...)三级 · 骨架：session_id = resp.json()["data"]，再拼进 URL
参考答案（做完再点开）
import osimport jsonfrom datetime import datetimefrom typing import Anyfrom fastapi import FastAPIfrom fastapi.testclient import TestClientfrom 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@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 链路测试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())评论区
如果你喜欢，那么欢迎来到我的世界！
了解更多音乐
暂未播放
