3.1 REST 设计：session / run / message 资源建模
第 2 章把 ResearchCopilot 的配置分层与模型路由定了下来，但那时它仍然只是一个「能在本机跑起来的库」。要变成产品，第一步是给它一张对外的脸——一组别人能调的 HTTP 接口。这一节就把这张脸画出来：定义会话、任务、消息三类资源，以及它们之间的 REST 语义。
本节把 ResearchCopilot 推进到「有稳定对外契约的 API」：新增
researchcopilot/api.py，把POST /sessions、POST /sessions/{id}/runs、GET /runs/{id}、GET /runs/{id}/messages四条路由立起来，真跑一遍并贴出真实响应。
3.1.1 先定资源，再定路由
REST 的第一条纪律是「先想清楚有哪些名词，再想动词」。研究助理这个场景里，名词其实很少：
| 资源 | 是什么 | 生命周期 |
|---|---|---|
session | 一次研究对话的容器，绑定一个 workspace | 用户创建，长期存活 |
run | 对某个 session 的一次执行（一次提问 → 多跳检索 → 一份报告） | 可多次创建，有终态 |
message | run 内部产生的对话条目（user / assistant / tool） | 随 run 产生，只读 |
把这三者定死之后，路由几乎是自动推导出来的：session 用 /sessions，run 挂在父资源下用 /sessions/{id}/runs，message 再挂在 run 下用 /runs/{id}/messages。动词交给 HTTP method——POST 创建、GET 读取。这就是资源导向：URL 里只有名词，动作用方法表达。
3.1.2 为什么 run 必须独立成资源
初学者最容易写出的版本是 POST /sessions/{id}/ask——「向会话提个问题」，一次性返回答案。它能把 demo 跑通，却会在三件事上撞墙：
| 需求 | 「ask 动作」写法 | 「run 资源」写法 |
|---|---|---|
| 任务跑很久，客户端要轮询进度 | 没有可查询的对象 | GET /runs/{id} 随时读状态 |
| 一次执行要能取消 | 无法寻址到那次执行 | POST /runs/{id}/cancel |
| 同一次执行要能重放事件流 | 事件流没有归属 | GET /runs/{id}/events（3.2 的 SSE） |
| 同一会话并发多个提问 | 状态混在 session 上 | 每个 run 独立，互不干扰 |
| 事后审计「第几次、跑了几步」 | 无从记录 | run 自带 id、状态、步数 |
核心区别在于：「执行」本身是有状态、有生命周期、需要被寻址的一等公民。把它降级成 session 上的一个动作，就等于把它的状态塞回 session，并发、取消、审计全部失去落点。所以 run 不是「session 的一次方法调用」，而是一个子资源：它有 id、有状态机、能被单独查询和取消。这也是为什么本卷后面的队列（第 7 章）、SSE（3.2）、可观测性（第 13 章）都围绕 run 这个 id 展开。
3.1.3 资源关系图与状态字段
三者的嵌套关系与 run 的状态迁移如下（纯文本，不依赖任何绘图工具）：
Session ──1:N──▶ Run ──1:N──▶ Message
│ │ role ∈ {user, assistant, tool}
│ │
│ └──status──▶ queued ──▶ running ──▶ succeeded
│ │
│ ├──▶ failed
└──workspace_id └──▶ canceled
Run.status 是这张 API 里唯一「会变」的字段，值域固定为五个：queued（已创建、等 worker）、running（正在执行）、succeeded、failed、canceled。终态是后三个，不可逆；只有 queued → running 和 running → 终态 两种迁移被允许。把状态收窄成一个封闭枚举，是让客户端能用 if 而不是正则去判断进度的前提。
Message 则完全只读，它的 role 沿用卷一第 1 章的 Role 口径（system / user / assistant / tool）。消息不是「写」出来的，是 run 执行过程的副产品，所以只有 GET，没有 POST——想加消息就去创建一个新 run。
3.1.4 四条路由与它们的 pydantic 模型
模型和路由落进 researchcopilot/api.py。为了聚焦资源建模，这里用内存字典存状态（第 5 章会换成 Postgres），并把 run 的执行留成一个 execute_run() 钩子，代表 worker 侧的活：
# researchcopilot/api.py
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Literal
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
def now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
class SessionCreate(BaseModel):
title: str = "untitled"
class Session(BaseModel):
id: str
title: str
created_at: str
class RunCreate(BaseModel):
input: str
class Run(BaseModel):
id: str
session_id: str
status: Literal["queued", "running", "succeeded", "failed", "canceled"] = "queued"
input: str
output: str | None = None
created_at: str
class MessageOut(BaseModel):
role: str
content: str
app = FastAPI(title="ResearchCopilot", version="0.3.0")
_sessions: dict[str, Session] = {}
_runs: dict[str, Run] = {}
_messages: dict[str, list[MessageOut]] = {}
@app.post("/sessions", status_code=201)
def create_session(body: SessionCreate) -> Session:
sid = "s_" + uuid.uuid4().hex[:8]
s = Session(id=sid, title=body.title, created_at=now())
_sessions[sid] = s
return s
@app.post("/sessions/{sid}/runs", status_code=202)
def create_run(sid: str, body: RunCreate) -> Run:
if sid not in _sessions:
raise HTTPException(status_code=404, detail="session not found")
rid = "r_" + uuid.uuid4().hex[:8]
r = Run(id=rid, session_id=sid, input=body.input, created_at=now())
_runs[rid] = r
return r
@app.get("/runs/{rid}")
def get_run(rid: str) -> Run:
if rid not in _runs:
raise HTTPException(status_code=404, detail="run not found")
return _runs[rid]
@app.get("/runs/{rid}/messages")
def list_messages(rid: str) -> list[MessageOut]:
if rid not in _runs:
raise HTTPException(status_code=404, detail="run not found")
return _messages.get(rid, [])
两处状态码是刻意选的。POST /sessions 返回 201，因为「创建」语义下资源诞生了；POST /sessions/{id}/runs 返回 202，因为此时 run 只是 queued——创建和执行是两件事，202 诚实地说「我收下了，但还没跑完」。返回体一律是 pydantic 模型，FastAPI 负责序列化，字段形状即契约。
3.1.5 真跑：四条路由走一遍
补上 worker 钩子和一段驱动脚本。execute_run() 模拟 worker：把 queued 翻成 running，跑完（此处用确定性假回答代替 minagent.run，真实模型路径见第 2 章的真/假路由），再翻成 succeeded 并写入消息：
def execute_run(rid: str) -> None:
"""worker 侧：把 queued 的 run 跑完（此处用确定性假回答代替 minagent.run）。"""
r = _runs[rid]
r.status = "running"
answer = f"围绕「{r.input}」，命中 3 条本地语料，结论见引用。"
r.output = answer
r.status = "succeeded"
_messages[rid] = [
MessageOut(role="user", content=r.input),
MessageOut(role="assistant", content=answer),
]
驱动脚本用 TestClient 直接打这个 app，无需真的起 uvicorn（TestClient 走的是 ASGI 进程内调用，语义与真 HTTP 一致）：
from fastapi.testclient import TestClient
client = TestClient(app)
s = client.post("/sessions", json={"title": "研究：向量数据库选型"}).json()
print("session :", s)
resp = client.post(f"/sessions/{s['id']}/runs", json={"input": "对比 pgvector 与 Qdrant"})
print("create status:", resp.status_code)
r = resp.json()
print("run(queued) :", r)
execute_run(r["id"])
print("run(worker) :", client.get(f"/runs/{r['id']}").json())
print("messages :", client.get(f"/runs/{r['id']}/messages").json())
missing = client.get("/runs/r_deadbeef")
print("404 :", missing.status_code, missing.json())
用 venv 里的 Python 3.14.6 与 FastAPI 0.143.0 跑出来的真实输出：
session : {'id': 's_98ad195c', 'title': '研究：向量数据库选型', 'created_at': '2026-10-10T18:51:53+00:00'}
create status: 202
run(queued) : {'id': 'r_d98c8f3c', 'session_id': 's_98ad195c', 'status': 'queued', 'input': '对比 pgvector 与 Qdrant', 'output': None, 'created_at': '2026-10-10T18:51:53+00:00'}
run(worker) : {'id': 'r_d98c8f3c', 'session_id': 's_98ad195c', 'status': 'succeeded', 'input': '对比 pgvector 与 Qdrant', 'output': '围绕「对比 pgvector 与 Qdrant」，命中 3 条本地语料，结论见引用。', 'created_at': '2026-10-10T18:51:53+00:00'}
messages : [{'role': 'user', 'content': '对比 pgvector 与 Qdrant'}, {'role': 'assistant', 'content': '围绕「对比 pgvector 与 Qdrant」，命中 3 条本地语料，结论见引用。'}]
404 : 404 {'detail': 'run not found'}
读这份输出：create status 是 202（不是 201），对应 3.1.4 里那个「收下但没跑完」的语义；创建瞬间 status 是 queued、output 是 null；execute_run() 跑完后同一个 GET /runs/{id} 就变成了 succeeded 且 output 有值。注意 GET /runs/{id} 返回的 run id 与创建时完全一致——这正是「run 是可寻址资源」的落地。最后那个 404 也说明了父资源校验：run 不存在时直接 404，而不是返回一个空对象。
3.1.6 OpenAPI 契约的形态
FastAPI 会依据路由签名自动生成 OpenAPI。同一次运行里读 /openapi.json 实测拿到：
openapi paths: ['/runs/{rid}', '/runs/{rid}/messages', '/sessions', '/sessions/{sid}/runs']
RunCreate : ['input']
paths 里四条路由齐了，且路径参数被标成 {rid} / {sid}；RunCreate 的 schema 里只有 input 一个必填字段。这份 schema 就是契约的机器可读形态：前端不必等接口文档，直接读 /openapi.json 就能生成 TS 客户端类型；第 9 章共享事件协议时，同一份 schema 还会被前后端一起消费。
不过要如实说明一个边界：OpenAPI 描述得了「请求长什么样、响应是什么模型」，描述不了「run 的状态会怎么变、事件流里会出现哪些类型」。前者是静态形状，后者是运行时行为——所以 3.2 的 SSE 事件契约，仍然得靠一张人工维护的事件表来钉住。
3.1.7 真实环境 / 真实模型下会怎样
真实部署里，这套路由的形状一行不改，但四处会变重：
- 内存字典换成 Postgres（第 5 章）。run 和 message 落库后，
GET /runs/{id}变成一次带索引的点查，session与workspace的归属由数据库约束保证。 - 202 后面真的挂队列（第 7 章）。现在
execute_run()是进程内直接调；生产里创建 run 只是往队列里塞一条消息，worker 在另一个进程消费。 - 每个资源都要校验租户（第 4 章）。
GET /runs/{id}之前必须先确认这个 run 属于调用者的 workspace，否则就是越权读别人的研究结果。 - 要能取消。真实 run 可能跑几分钟，用户关页面就该停；
POST /runs/{id}/cancel与第 8 章的超时取消配套。
本节验收清单：
- 三类资源各自有独立 id，run 能通过 id 单独寻址
-
POST /sessions/{id}/runs返回 202，且新 run 初始状态为queued -
GET /runs/{id}在 worker 跑完后能反映succeeded -
GET /runs/{id}/messages返回该 run 的消息，且role与卷一Role口径一致 - 不存在的 run 返回 404，而非空对象
3.1.8 小结
- 先定名词（session / run / message），再让 HTTP method 表达动词。
- run 是子资源不是 session 的动作：它有 id、状态机、能被轮询 / 取消 / 流式 / 审计。
Run.status是唯一可变的封闭枚举：queued → running → succeeded / failed / canceled。- 201 表示「资源已创建」，202 表示「已受理但未完成」——创建与执行是两件事。
- OpenAPI 给的是静态契约；运行时行为（状态迁移、事件流）要靠文档与代码共同维护。
到这里 ResearchCopilot 有了一张能调的脸，但它还是「一问一答」的。下一节 3.2 流式接口：SSE 推送事件 会把 run 的执行过程实时推给客户端。想回看这一节依赖的模型路由，翻 2.3 模型路由与降级：真/假模型开关，离线默认假模型 。
阅读导航：上一节：2.3 模型路由与降级：真/假模型开关，离线默认假模型 · 下一节：3.2 流式接口：SSE 推送事件 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。