Backend Runtime

2026-09-28 artificial-intelligence / agentic 3.3K 字 13 分钟

Backend Runtime：服务端会话与运行边界

s02 通过工具定义、注册表和统一分发，让 Agent Loop 能够使用多个工具。 但能够调用工具，不等于已经具备可供多个客户端接入的后端运行时。

如果浏览器每次请求都提交完整消息历史，FastAPI 只是一次无状态的函数调用： 服务端无法判断历史是否被删改，两个并发请求也可能从同一个旧历史开始执行。 后续加入权限审批时，后端还需要暂停一次尚未完成的执行，并在用户决定后继续， 仅靠“请求携带消息列表”无法稳定表达这些状态。

s02.5 是本地工程化中间章，它保留 s02 的工具注册与分发，新增唯一机制：

后端持有 Conversation，并用 Run 表达一次用户输入触发的 Agent 执行。

FastAPI 是 Agent 的正式服务入口，Web 只是当前接入器。浏览器只保存 conversation_id，模型消息、工具调用结果和执行顺序都由后端维护。

flowchart LR
  accTitle: Backend Runtime 的职责边界
  accDescr: Web 或其他接入器只提交 Conversation 标识和本次输入，FastAPI 将请求交给 ConversationService，服务加载历史后调用 AgentRunner，AgentRunner 通过模型和工具运行时完成执行，成功后由服务保存新快照。

  web["Web 接入器"]
  other["其他接入器"]
  api["FastAPI"]
  service["ConversationService"]
  store[("ConversationStore")]
  runner["AgentRunner"]
  model["AsyncOpenAI"]
  runtime["ToolRuntime"]

  web -->|"conversation_id + input"| api
  other -->|"conversation_id + input"| api
  api --> service
  service -->|"读取 / 保存"| store
  service --> runner
  runner --> model
  runner --> runtime
  runtime --> runner
  runner --> service
  service --> api

  class api,service,store added
  class runner,runtime attention

Conversation、Run 和认证 Session

概念表达的内容生命周期
Conversation一段对话的模型历史和公开 Turn跨越多次用户输入
Run一次输入触发的完整 Agent Loop 执行从接收输入到完成或暂停
认证 Session用户身份和登录状态由认证系统决定

Conversation 解决“下一轮模型调用需要哪些历史”，Run 解决“这一次执行进行到哪里”。 认证 Session 则回答“调用者是谁”。把它们都叫作 session，会让消息状态、执行状态 和身份状态相互耦合。

本章只实现 Conversation 和完成态 Run，不实现用户认证。客户端拿到 conversation_id 不代表它已经获得了安全授权；正式系统仍需在 API 边界验证 调用者是否有权访问该 Conversation。

项目目录

s02-5-backend-runtime/
├── s02-5.md
├── README.md
├── main.py
├── config.py
├── bootstrap.py
├── workspace.py
├── agent/
│   ├── contract.py
│   ├── loop.py
│   └── service.py
├── api/
│   ├── app.py
│   └── schemas.py
├── conversations/
│   ├── contract.py
│   └── memory.py
├── tooling/
├── tools/
├── web/
│   └── agent-chat.html
└── tests/

各层职责：

模块职责
api/外部请求和响应
agent/service.pyRun 读取、执行并提交 Conversation
agent/loop.py模型和工具循环
conversations/Conversation 表示和保存
tooling/、tools/工具注册、分发和执行
web/浏览器接入器调用后端

依赖从协议边界指向应用服务，再指向 Agent 与存储契约。 Agent Loop 不导入 FastAPI，也不判断客户端类型；Store 不依赖模型或工具。 因此更换 Web、CLI、数据库或模型客户端时，不需要把全部职责重新组合一次。

项目代码

agentic-s02-5-backend-runtime 
"""s02.5 后端服务入口。

运行：
    uv run s02-5-backend-runtime/main.py
"""

from __future__ import annotations

import uvicorn
from api.app import app
from config import load_settings


def main() -> None:
    settings = load_settings()
    uvicorn.run(app, host=settings.host, port=settings.port)


if __name__ == "__main__":
    main()

Conversation 保存两种历史

Conversation 内部同时保存 messages 和 turns：

@dataclass(frozen=True)
class ConversationTurn:
    run_id: UUID
    user_input: str
    assistant_output: str


@dataclass(frozen=True)
class Conversation:
    conversation_id: UUID
    messages: tuple[ChatCompletionMessageParam, ...] = ()
    turns: tuple[ConversationTurn, ...] = ()

messages 是模型协议历史，包含 user、assistant、assistant tool calls 和 role="tool" 结果。下一次调用模型时必须保留完整的调用标识配对，不能只保存 页面上看到的问答文本。

turns 是面向客户端的公开历史，只包含每次 Run 的输入、最终回答和 run_id。 浏览器恢复页面时不需要理解 OpenAI tool call 协议，也不会看到内部工具结果。

这两份数据不是两个独立事实来源。它们在一次成功提交中由同一个 Run 同时产生： messages 服务于模型续写，turns 服务于外部展示。只更新其中一份会破坏一致性， 因此写入动作集中在 ConversationService。

模型历史使用元组保存，Conversation 使用 frozen=True，用于限制普通调用方 直接追加或重新绑定字段。这仍然不是深度不可变：消息本身是字典。 Store 在读写时执行 deepcopy()，避免调用方修改嵌套消息后绕过 save()。

AgentRunner 隔离模型循环

应用服务不应该知道 OpenAI 请求格式、工具 Schema 或 tool_call_id。 它只依赖一个窄接口：

@dataclass(frozen=True)
class AgentResult:
    answer: str
    messages: tuple[ChatCompletionMessageParam, ...]


class AgentRunner(Protocol):
    async def run(
        self,
        history: Sequence[ChatCompletionMessageParam],
    ) -> AgentResult:
        ...

ToolCallingAgent 实现这个接口。它为本次执行创建消息副本，在开头加入系统提示， 然后继续使用 s02 已建立的 OpenAI-compatible 工具循环：

  1. 将消息和 runtime.schemas() 发送给模型。
  2. 保存完整 assistant 消息，包括所有 tool calls 和调用标识。
  3. 按顺序执行每个工具。
  4. 将结果追加为对应 tool_call_id 的 tool 消息。
  5. 模型不再调用工具时返回最终文本和完整新历史。

系统提示属于当前后端部署配置，不写入 Conversation。这样每次模型请求都会获得 系统提示，但公开历史和持久化数据不会把运行配置伪装成用户消息。

AgentRunner 还让应用服务测试不必启动真实模型。测试替身可以记录收到的历史、 控制并发时机或主动抛出异常，从而验证会话语义而不是 HTTP SDK 的实现细节。

一次 Run 的提交过程

ConversationService.run() 是后端状态变更的中心：

async def run(self, conversation_id: UUID, user_input: str) -> CompletedRun:
    await self.get(conversation_id)
    lock = self._locks.setdefault(conversation_id, asyncio.Lock())

    async with lock:
        conversation = await self.get(conversation_id)
        run_id = self._run_id_factory()
        user_message: ChatCompletionMessageParam = {
            "role": "user",
            "content": user_input,
        }
        history = (*conversation.messages, user_message)

        result = await self._runner.run(history)
        turn = ConversationTurn(
            run_id=run_id,
            user_input=user_input,
            assistant_output=result.answer,
        )
        updated = Conversation(
            conversation_id=conversation_id,
            messages=result.messages,
            turns=(*conversation.turns, turn),
        )
        await self._store.save(updated)

    return CompletedRun(
        conversation_id=conversation_id,
        run_id=run_id,
        message=result.answer,
    )

执行顺序不能随意交换：

flowchart LR
  accTitle: Run 的串行执行与提交
  accDescr: 后端先确认 Conversation 存在并获取该会话的锁，再重新加载最新快照、追加本次用户消息并运行 Agent；只有 Agent 正常完成才保存新历史，失败则保留原快照。

  request["接收 input"]
  exists{"Conversation 存在"}
  lock["获取该会话的锁"]
  load["重新加载最新快照"]
  append["在本地副本追加 user"]
  execute["AgentRunner.run"]
  outcome{"正常完成"}
  save["同时保存 messages 与 turn"]
  completed(["completed"])
  unchanged(["原快照不变"])

  request --> exists
  exists -->|"否"| missing(["404"])
  exists -->|"是"| lock
  lock --> load
  load --> append
  append --> execute
  execute --> outcome
  outcome -->|"是"| save
  save --> completed
  outcome -->|"否"| unchanged

  class lock,load,append,save added
  class outcome,unchanged attention

第一次存在性检查发生在锁外，用于避免任意无效 UUID 持续占用锁表。 进入锁后必须重新加载 Conversation，因为等待期间前一个 Run 可能已经提交了 新历史。后一个 Run 应基于这份最新快照执行。

为什么锁必须按 Conversation 划分

如果没有锁，两个请求可能同时读取历史 H0：

Run A: H0 + input_A -> H1
Run B: H0 + input_B -> H2

最后一次 save() 会覆盖另一次结果。数据库写入本身即使是线程安全的， 也不能保证“读取旧历史、执行模型、写入新历史”这一整段业务操作不会竞争。

如果使用一个全局锁，虽然不会覆盖，但所有 Conversation 都必须排队。 一次耗时工具调用会阻塞其他用户，吞吐量退化为整个进程一次只能执行一个 Run。

实现使用 dict[UUID, asyncio.Lock]：

  • 同一 Conversation 的 Run 串行，后一个请求读取前一个请求的结果。
  • 不同 Conversation 使用不同锁，可以并行调用模型和工具。

这是单进程协调，不是分布式锁。多 worker 或多实例部署时，每个进程都有独立锁表， 必须由持久化层的版本检查、事务或跨进程协调机制替代。

失败不提交的准确含义

Agent 在本地消息副本上执行。只有 _runner.run() 正常返回后， 服务才创建公开 Turn 并调用 store.save()。模型请求失败、工具 handler 出现 编程错误或循环被取消时，原 Conversation 快照保持不变。

这项保证只覆盖会话状态，不是完整事务。工具可能已经执行真实副作用：

  • 文件已经写入。
  • Shell 命令已经运行。
  • 外部 API 已经接收请求。

后续模型调用失败时，删除未提交的消息历史无法撤销这些操作。 要实现副作用回滚，需要工具自身提供幂等键、补偿操作或事务能力， 不能由 Conversation Store 自动完成。

ConversationStore 是可替换边界

应用服务依赖 ConversationStore 协议：

class ConversationStore(Protocol):
    async def create(self) -> Conversation: ...

    async def get(
        self,
        conversation_id: UUID,
    ) -> Conversation | None: ...

    async def save(self, conversation: Conversation) -> None: ...

本章使用 InMemoryConversationStore，因为要先明确服务端状态所有权， 不把数据库选型混入核心机制。它通过一个短时间持有的内部锁保护字典读写， 并在边界复制快照。

内存实现有明确限制：

  1. 进程重启后 Conversation 全部丢失。
  2. 多 worker 各自持有不同字典，请求落到另一 worker 时会得到 404。
  3. 会话锁只在当前进程有效。
  4. 没有过期回收、容量限制和所有权校验。

因此当前服务只能使用单 worker。生产持久化不是把字典替换成数据库调用就结束： Store 还需要版本号或事务，防止多个进程从同一旧版本生成并覆盖新快照。

HTTP 协议只暴露领域状态

创建 Conversation：

POST /api/conversations
{
  "conversation_id": "6dd71267-1c81-4d31-a16b-cfdbcbd86d1b"
}

执行一次 Run：

POST /api/conversations/6dd71267-1c81-4d31-a16b-cfdbcbd86d1b/runs
Content-Type: application/json

{"input": "读取 README.md 并总结"}
{
  "status": "completed",
  "conversation_id": "6dd71267-1c81-4d31-a16b-cfdbcbd86d1b",
  "run_id": "02bf49da-f12a-402f-aee2-0aa75f47d9bf",
  "message": {
    "role": "assistant",
    "content": "..."
  }
}

读取公开历史：

GET /api/conversations/6dd71267-1c81-4d31-a16b-cfdbcbd86d1b

客户端不提交 messages。请求模型使用的完整历史由后端根据 conversation_id 读取，避免客户端成为权威状态来源。

CompletedRunResponse 使用字面量状态：

class CompletedRunResponse(ApiModel):
    status: Literal["completed"] = "completed"
    conversation_id: UUID
    run_id: UUID
    message: MessageResponse

status 是 Run 响应的判别字段。当前只有 completed，但客户端不需要根据 HTTP 响应形状猜测执行结果。s03 可以增加 approval_required 变体， 同时保留已完成响应的字段语义。

未知 Conversation 返回结构化错误：

{
  "error": {
    "code": "conversation_not_found",
    "message": "Conversation not found: ..."
  }
}

错误码供接入器判断分支，message 用于日志或界面展示。 它比把异常文本塞进成功响应更容易扩展，也不会把“Agent 的工具结果” 与“HTTP 请求本身失败”混为一谈。

FastAPI 管理进程级资源

模型客户端和工具运行时属于后端进程，而不是单个 HTTP 请求。 FastAPI lifespan 负责组装和释放它们：

@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncGenerator[None]:
    settings = load_settings()
    runtime = build_runtime()
    store = InMemoryConversationStore()

    async with create_client(settings) as client:
        runner = ToolCallingAgent(
            runtime,
            client=client,
            model=settings.model,
            system_prompt=system_prompt,
        )
        application.state.conversation_service = ConversationService(
            store,
            runner,
        )
        yield

这样安排有三个结果：

  1. 多次请求复用 AsyncOpenAI 的连接池，不为每个 Run 重建客户端。
  2. 服务关闭时明确释放 HTTP 连接。
  3. 导入 api.app 不会立刻创建模型客户端，测试可以替换依赖。

bootstrap.py 继续负责选择本进程启用的工具，FastAPI 不直接导入具体 handler。 资源生命周期、应用组装和工具实现分别位于不同边界。

Web 只是接入器

单文件 Web 页面只在 localStorage 中保存一个 conversation_id：

  1. 首次打开页面时调用 POST /api/conversations。
  2. 发送消息时只提交当前 ID 和 input。
  3. 刷新页面时调用 GET /api/conversations/{id} 恢复公开 Turn。
  4. 新建对话时申请新 ID，不清空或伪造服务端历史。
  5. 服务重启导致旧 ID 不存在时，客户端创建新的 Conversation。

页面不理解 OpenAI 消息类型、工具调用或内部错误结果。 未来的 CLI、飞书接入器或其他前端只要遵守相同 HTTP 契约，就能共享后端语义。