7.2 会话状态对象
3.3 把 history 存进了 Agent 实例，多轮对话第一次跑通。当时那是个进步，但放到真实服务里，它就露馅了：一个 Agent 实例只承载一段对话的 history，要同时服务一百个用户就得造一百个 Agent；进程一重启，所有对话全部清零；更麻烦的是，history 和「这个 Agent 是谁」的配置混在一个对象里，想单独把对话存下来或搬走都无从下手。问题的根子是状态与配置被焊死在一起。
本节把 minagent 推进到「状态独立成对象」：新增
minagent/session.py定义Session（持有messages/session_id/metadata/created_at，提供append/snapshot/restore），并把history从Agent里剥出去——Agent退回无状态的配置载体，状态全部交给Session。
7.2.1 history 焊在 Agent 里的三个麻烦
把 3.3 的 Agent 想象成一个真实后端服务的构件，三个麻烦会依次冒出来：
| 麻烦 | 表现 | 根因 |
|---|---|---|
| 无法并发服务 | 一个 Agent 只有一份 history | 状态与实例一对一绑定 |
| 无法持久化 | 进程重启，对话全丢 | 状态只活在内存里，无序列化出口 |
| 无法迁移 | 想把会话挪到另一台机器做不到 | 状态没有独立身份，取不出来 |
这三个麻烦指向同一个动作：把「状态」从 Agent 里赶出去。赶出去之后，Agent 只剩配置（model / registry / system / max_steps），变成彻底无状态的东西——同一个 Agent 实例可以轮流服务任意多个会话；而状态装进一个新对象，它有独立的身份、能序列化、能跨进程搬。
这个「无状态实例 + 有状态数据」的分法，是所有 Web 服务的通用套路：Controller 是无状态的，Session 存状态；Connection 是无状态的，Transaction 存状态。Agent 运行时不过是把这个老套路再走一遍。
7.2.2 Session 的四个字段
Session 是一个 dataclass，四个字段各司其职：
# minagent/session.py
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
from .types import Message
@dataclass
class Session:
session_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
messages: list[Message] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
created_at: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
| 字段 | 类型 | 作用 |
|---|---|---|
session_id | str | 会话的稳定身份，跨轮次、跨进程不变 |
messages | list[Message] | 消息历史，多轮对话的全部内容 |
metadata | dict[str, Any] | 挂载业务信息：用户 id、租户、标签 |
created_at | str | 创建时刻（ISO 8601），用于排序与过期清理 |
两个 default_factory 值得停一下。session_id 用 uuid4().hex[:12]——取 12 位十六进制（48 位随机），碰撞概率低到可以忽略，长度又短到能塞进日志行。created_at 用 datetime.now(timezone.utc).isoformat()——带时区的 ISO 字符串，而不是裸 datetime 对象。原因在 7.3 会揭晓：这个对象要序列化成 JSON，而 datetime 默认不可 JSON 化；存成字符串，序列化零成本，还天然可读、可排序。
metadata 是刻意留的口子。核心字段只有 messages，但真实业务总想挂点别的——「这个会话属于哪个用户」「是不是 VIP」「上一轮用了什么模型」。把这些塞进 metadata，核心结构就永远干净，业务信息也不会污染消息历史。
7.2.3 append / snapshot / restore
Session 的接口只有三个动作，对应状态的一生：
def append(self, msg: Message) -> None:
self.messages.append(msg)
def snapshot(self) -> dict[str, Any]:
return {
"session_id": self.session_id,
"created_at": self.created_at,
"metadata": dict(self.metadata),
"messages": [message_to_dict(m) for m in self.messages],
}
@classmethod
def restore(cls, data: dict[str, Any]) -> "Session":
return cls(
session_id=data["session_id"],
created_at=data["created_at"],
metadata=dict(data.get("metadata", {})),
messages=[message_from_dict(m) for m in data.get("messages", [])],
)
append：追加一条消息，最简单的动作。snapshot：把整个会话压成一个纯 dict——注意messages里的Message/ToolCall是 dataclass，不能直接json.dumps，得先转成 dict。这一步是 7.3 存盘的地基。restore：snapshot的逆运算，从 dict 还原出Session。它是classmethod，因为还原时还没有实例。
消息的 dict 互转由两个自由函数承担，Message 里的嵌套 ToolCall 要递归处理：
def message_to_dict(msg: Message) -> dict[str, Any]:
return {
"role": msg.role,
"content": msg.content,
"tool_calls": [
{"id": tc.id, "name": tc.name, "arguments": tc.arguments}
for tc in msg.tool_calls
],
"tool_call_id": msg.tool_call_id,
}
def message_from_dict(data: dict[str, Any]) -> Message:
return Message(
role=data["role"],
content=data.get("content", ""),
tool_calls=[
ToolCall(id=tc["id"], name=tc["name"], arguments=tc["arguments"])
for tc in data.get("tool_calls", [])
],
tool_call_id=data.get("tool_call_id"),
)
为什么不用 dataclasses.asdict 一步到位？因为它会深度递归，把 arguments 里的一切也照搬，遇到不可 JSON 化的值会直接炸；而且它输出的字段名与 Message 字段强耦合，将来给消息加一个内部字段（比如 timestamp）会立刻漏进存盘格式。手写这两个函数虽然啰嗦，但存盘格式与内存结构解耦——这是持久化数据的通用纪律：别让磁盘上的格式跟着内存里的类字段走。
7.2.4 Agent 变无状态：run_turn(session, …)
状态搬进 Session 后，Agent 的 history 字段删掉，run() 换成接收 Session 的 run_turn()：
# minagent/agent.py
class Agent:
def __init__(self, model, registry, system: str = "", max_steps: int = 8) -> None:
self.model = model
self.registry = registry
self.system = system
self.max_steps = max_steps
# 注意：不再有 self.history
def run_turn(self, session: Session, user_input: str) -> RunResult:
messages = list(session.messages)
if self.system and not messages:
messages.append(Message(role="system", content=self.system))
messages.append(Message(role="user", content=user_input))
steps: list[Step] = []
for i in range(self.max_steps):
resp = self.model.complete(messages, self.registry.schemas())
msg = resp.message
messages.append(msg)
if not msg.tool_calls:
steps.append(Step(index=i, kind="final", detail=msg.content))
session.messages = messages
return RunResult(messages=messages, steps=steps, final=msg.content)
for tc in msg.tool_calls:
result = self.registry.call(tc)
messages.append(Message(role="tool", content=result, tool_call_id=tc.id))
steps.append(Step(index=i, kind="tool_call", detail=f"{tc.name} -> {result}"))
steps.append(Step(index=self.max_steps, kind="error", detail="达到 max_steps 仍未收敛"))
session.messages = messages
return RunResult(messages=messages, steps=steps, final="")
对比 3.3 的 run()，改动只有三处，但性质全变了：
| 位置 | 3.3 | 本节 |
|---|---|---|
| 读历史 | self.history | session.messages |
| 写历史 | self.history = messages | session.messages = messages |
| 身份 | 实例自带状态 | 实例无状态，状态随 session 传入传出 |
system 的注入条件依旧是 if not messages——但现在判的是这个会话的历史是否为空，而不是 Agent 的。这意味着：新会话第一次进来会注入 system，同一会话第二轮则不会。会话之间彼此隔离，各注入各的。
7.2.5 跑一遍：两轮对话与两个会话
用一个假模型服务两个会话：会话 A 跑两轮（第一轮走工具、第二轮直接答），会话 B 全新开始：
agent = Agent(build_model(), build_registry(), system="你是一个计算助手。")
session_a = Session(metadata={"user": "alice"})
sid_a = session_a.session_id
r1 = agent.run_turn(session_a, "先算 2+3，再把结果乘 4")
print("A 第 1 轮:", r1.final, "| history:", len(session_a.messages))
r2 = agent.run_turn(session_a, "再加 10 是多少")
print("A 第 2 轮:", r2.final, "| history:", len(session_a.messages))
session_b = Session(metadata={"user": "bob"})
r3 = agent.run_turn(session_b, "你好")
print("B 第 1 轮:", r3.final, "| history:", len(session_b.messages))
本机实测（Python 3.14.6）：
A 第 1 轮: 第一轮：结果是 20。 | history: 7
A 第 2 轮: 第二轮：在 20 基础上再加 10 是 30。 | history: 9
B 第 1 轮: 另一个会话：你好，我是新会话。 | history: 3
A 的 session_id 稳定: True | created_at 稳定: True
A 与 B 的 session_id 不同: True
B 的 history 3 条（新会话：system+user+assistant）: 3
三个数字印证了三件事：
- A 从 7 涨到 9：第二轮只新增
user与assistant两条，没有重复注入system——history在Session上正常累积。 session_id与created_at全程不变：同一个Session对象跨两轮保持身份稳定。这两条断言，正是「会话」这个概念成立的定义。- B 只有 3 条（system + user + assistant）：全新会话从零开始，且注入了自己的
system。同一个Agent实例先后服务了 A、B 两个会话——无状态实例的价值就在这里。
snapshot / restore 的往返也一并验证：
snap = session_a.snapshot()
restored = Session.restore(json.loads(json.dumps(snap, ensure_ascii=False)))
print("恢复后 session_id 相同:", restored.session_id == session_a.session_id)
print("恢复后消息逐条相等:", [m.content for m in restored.messages]
== [m.content for m in session_a.messages])
恢复后 session_id 相同: True
恢复后消息逐条相等: True
恢复后工具调用相等: True
json.dumps 再 json.loads 绕一圈，是为了模拟「真的过了一次磁盘」——如果 snapshot 的产物里有 datetime 这类不可序列化的值，这一步会立刻报错。往返后逐条相等，说明序列化是无损的，7.3 存盘才有底气。
7.2.6 为什么要把状态赶出 Agent
一句话概括这个分离的收益：Agent 变成纯函数式的配置，Session 变成可搬运的数据。收益拆开看有三条：
- 可复用：一个无状态
Agent实例可以并发服务任意多个Session。并发模型从「一个用户一个 Agent」变成「一个 Agent 管所有用户」，内存与初始化成本骤降。 - 可持久化：
Session能snapshot成纯 JSON，落盘、进 Redis、发到消息队列都行——7.3 正是靠这一点做检查点。 - 可测试：测试时直接构造一个带指定
messages的Session塞进去，不必去摸Agent的内部状态。状态是显式参数，测试就变成纯粹的数据驱动。
反过来，如果状态继续焊在 Agent 里，这三条一条也做不到。所以「无状态实例 + 有状态数据」不是洁癖，而是服务化的前提。第 17 章把 minagent 包成 HTTP 服务时，每个请求带着一个 session_id 进来、取出对应 Session、跑一轮、再存回去——这个流程能成立，全靠本节这次拆分。
7.2.7 真实模型下会怎样
假模型的 session_id 是本地 uuid4 生成的，真实服务里它通常由入口层（HTTP 网关、前端）分配并随请求带上，因为同一个会话的多次请求可能落到不同进程，必须有一个全局认得的标识把它们串起来。
另一个真实差异在 messages 的规模。假模型的 history 只有几条，snapshot 出来的 JSON 不过几百字节；真实多轮对话里，messages 会包含长系统提示、几十轮问答、每次工具调用的完整往返，snapshot 可能达到几百 KB 甚至 MB。这意味着 snapshot 不能太频繁（7.3 会讲频率取舍），也不能每次都全量写——大会话的持久化要考虑增量或分片。
还有一点：真实场景里 metadata 往往会承载并发控制信息，比如「这个会话正在被哪个进程处理」的锁标记。假模型的单线程演示看不出竞争，但多个请求同时落到同一个 session_id 时，messages 的读改写会互相覆盖。这是第 13、14 章并发话题的伏笔，本节先记下这个坑。
到这里，会话状态已经是一个能 snapshot 成 JSON 的独立对象。但「能序列化」离「崩溃后能续跑」还差一步——snapshot 只是一次内存快照，真正要做的，是在运行途中把它写进磁盘，并在重启后从断点接上。下一节 7.3 检查点与快照
就来补这一步。想回看事件流是怎么订阅的，翻 7.1 流式事件模型
。
阅读导航：上一节：7.1 流式事件模型 · 下一节：7.3 检查点与快照 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。