预计时间：60 分钟 ｜ 前置：完成第 04 章 ｜ 本章调用真实 DeepSeek 模型
第 03、04 章解决了智能体能力增多后怎样组织和协作。现在回到第 02 章留下的运行循环：它用一个 list[Message] 保存对话历史，循环直接修改这个列表，再把它发送给模型。这个设计适合最小示例，但随着智能体持续运行，会出现三个问题：
- 不是所有历史都该给模型看。一次真实运行会产生大量过程记录：流式分片、生命周期边界、压缩动作。它们要留存，但发给模型只会浪费 token、制造噪音。
- 同一份历史有多种用途。模型请求要读取它，持久化功能要把它写入磁盘，界面要展示它，程序恢复时还要重新载入它。如果每个功能各存一份，内容很容易不同步。
- 历史一旦写入就不应再改。第 02 章使用的普通列表允许任意调用方执行
pop或insert，一次误改就会影响后续上下文。
本章用一份只允许追加的事件日志统一解决这三个问题。用户消息、模型回复、工具调用和运行边界都依次写入日志；需要向模型发送历史时，再从日志中挑出相关事件并转换成消息。这种“保存事件，再根据事件还原当前状态”的方法称为事件溯源。
完成本章后，你将能够：
- 区分“完整事件日志”和“发送给模型的消息投影”；
- 实现只追加、可订阅且数据不可变的
Session； - 把系统提示词记录为模型可见的
system/message； - 把智能体循环中的用户消息、模型回复和工具调用记录为事件；
- 使用按位置读取和稳定快照 API 检查日志，并校验事件序号的连续性。
先想清楚日志和消息历史的区别。日志回答的问题是发生了什么：用户说了什么、模型回了什么、哪个工具被调用了、结果是什么、这一轮何时开始何时结束。消息历史回答的问题是模型该看到什么：把日志里对模型有用的部分挑出来，按协议格式排好。
两者的要求不同：日志要完整，消息历史则只保留模型真正需要的内容。事件溯源把它们分成两层：
┌─────────────────────────┐
追加事件 ────────▶ │ Session（事件日志） │
（唯一写入路径） │ #0 turn/start │
│ #1 step/start │
│ #2 system/message │
│ #3 user/message │───▶ derive_messages() ──▶ 发给模型
│ #4 request/header │ （生成消息）
│ #5 assistant/message │
│ #6 tool/call / result │
│ #7 step/end / turn/end │
└─────────────────────────┘
这个结构带来三个好处：运行过程能够完整保留；模型、界面和持久化功能都从同一份日志读取信息；已经写入的事件不会被后续代码改写。把事件转换成特定用途数据的过程称为“投影”，derive_messages() 生成的就是供模型使用的消息投影。
事件是日志的最小单位：
@dataclass(frozen=True)
class SessionEvent:
seq: int
type: str
time: int
data: Mapping[str, Any]
surface_op: str | Mapping[str, Any] | None = None
source_event_seqs: tuple[int, ...] = ()seq从 0 开始连续递增，5.6 节的重放校验靠它。type是事件类型。除消息和工具事件外，本章还记录request/header、step/start、step/end与轮次边界，用来说明每次模型请求属于哪一轮、哪一步。time是 Unix 毫秒时间戳，记录事件发生的时刻。data是事件内容，frozen=True保证事件对象本身不可改。surface_op说明消息事件追加到模型可见历史，还是替换已有表层节点；source_event_seqs记录替换依据。替换事件必须引用被覆盖的全部节点，第 09 章会用它裁剪工具结果。
Session 类只提供一个写入入口，append：
class Session:
def __init__(self) -> None:
self._log: list[SessionEvent] = []
self._snapshot: tuple[SessionEvent, ...] | None = None
self._listeners: list[Any] = []
def append(self, type, data, *, surface_op=None, source_event_seqs=()):
if type in SURFACE_EVENT_TYPES:
surface_op = "append" if surface_op is None else surface_op
frozen_data = _freeze_json(data)
event = SessionEvent(
seq=len(self._log), type=type, time=_now(), data=frozen_data,
surface_op=_freeze_json(surface_op),
source_event_seqs=_freeze_json(source_event_seqs),
)
_surface_events([*self._log, event]) # 提交前验证表层操作
self._log.append(event)
self._snapshot = None
for listener in list(self._listeners):
listener(event)
return event三个动作，各对应一个设计意图：
_freeze_json(data)在写入前校验并冻结。它拒绝循环引用、非字符串对象键、非有限数、负零、超出 JSON 安全范围的整数和非 JSON 类型；列表变成元组，字典变成仍可序列化的FrozenDict。调用方不能改写事件和快照。self._snapshot = None让snapshot_events()的全量缓存失效。只读取一个位置时使用event_at(seq)，需要稳定区间时再创建快照，避免每次查询都复制完整日志。- 通知订阅者。通过
subscribe注册的函数会立即收到新事件。第 08 章会实现文件存储并显式调用save()；也可以进一步让存储功能订阅事件，在需要时自动保存。
derive_messages() 是投影规则的实现：
def derive_messages(self) -> list[Message]:
messages: list[Message] = []
for event in _surface_events(self._log):
message = _derive_event_message(event)
if message is not None:
messages.append(message)
return messages
def _derive_event_message(event: SessionEvent) -> Message | None:
if event.type == "system/message":
content = event.data.get("content")
return Message(role="system", content=content) if content else None
if event.type == "user/message":
return Message(role="user", content=event.data["content"])
if event.type == "assistant/message":
raw_calls = event.data.get("tool_calls") or []
tool_calls = tuple(
ToolCall(id=c["id"], name=c["name"], arguments=c["arguments"])
for c in raw_calls
)
reasoning_content = event.data.get("reasoning_content")
if event.data.get("content") is None and not reasoning_content and not tool_calls:
return None
return Message(
role="assistant",
content=event.data.get("content"),
reasoning_content=reasoning_content,
tool_calls=tool_calls,
)
if event.type == "tool/result":
return Message(
role="tool",
content=event.data["content"],
tool_call_id=event.data["call_id"],
)
return None_surface_events() 先应用 append 与 replace，再把四种表层事件投影成消息；其余事件留在日志里，不进模型：
| 事件 | 投影结果 | 说明 |
|---|---|---|
system/message |
role="system" |
当前有效的系统提示词 |
user/message |
role="user" |
人的话 |
assistant/message |
role="assistant" |
模型的话，可能带 reasoning_content 与 tool_calls |
tool/result |
role="tool" |
工具结果，带 tool_call_id 对应 |
request/header、step/turn 边界、tool/call |
不投影 | 请求与过程记录，只进日志 |
tool/call 记录模型请求调用工具这件事本身，其中包含原始参数，但不会再次转换成模型消息。assistant/message 里的 tool_calls 已经携带了同一请求，重复发送没有意义。日志保留独立的 tool/call，是为了让界面和调试工具能够直接查看每次调用。
第 02 章的循环直接操作 history 列表。日志化之后，循环的每一步都变成追加一条事件：
def run_agent(client, tools, system_prompt, user_prompt, max_steps=10) -> Session:
tools_by_name = {tool.name: tool for tool in tools}
session = Session()
session.append("turn/start", {"turn": 1})
for step in range(1, max_steps + 1):
session.append("step/start", {"turn": 1, "step": step})
session.record_system_prompt(system_prompt, turn=1, step=step)
if step == 1:
session.append("user/message", {"content": user_prompt})
session.append("request/header", {"header": {"config": ..., "tools": ...}})
try:
# 投影历史 → 请求模型 → 记录 assistant 与工具事件
...
finally:
session.append("step/end", {"turn": 1, "step": step})
# ...与第 02 章相比，这里有两个变化：
- 系统提示词以
system/message进入消息历史，模型配置和工具说明写入request/header。两部分共同构成可从日志重建的请求；V3 请求头不再保存system副本。 turn/start与turn/end标记一轮任务的开始和结束，step/start与step/end标记一次模型调用及后续工具执行。finally保证发生异常时也会写入步骤结束事件。
日志全量留存后，可以通过重放重建会话状态。from_log 是重建入口：
@classmethod
def from_log(cls, events: list[SessionEvent]) -> "Session":
session = cls()
for index, event in enumerate(events):
if event.seq != index:
raise ValueError(
f"重放失败：第 {index} 个事件 seq 为 {event.seq}（应为 {index}）"
)
frozen_data = _freeze_json(event.data)
session._log.append(SessionEvent(..., data=frozen_data))
return session从磁盘读回的数据可能损坏，因此 from_log 除了检查连续编号，还会重新验证事件类型、时间戳和 JSON 数据，并再次冻结 data。更完整的实现还需要检查轮次与步骤是否闭合，以及每个工具调用是否都有对应结果。
重放有三个常见用途：程序中断后从磁盘日志恢复会话；为子智能体复制一段已经完成的对话，第 14 章会继续讲解；以及在调试时还原一次运行的完整过程。本章只演示最基础的情况：使用同一份日志重建出完全相同的消息历史。
uv run python chapters/05-session-log/src/demo.py下面是模型调用一次 calculator、经过两个 step 后作答时的代表性输出。回复文本、tool call id 和 step 数可能随模型行为变化，但每个 step 的开始与结束、请求头和 turn 边界都会被记录：
=== ① 订阅者视角 ===
[订阅者] 看到新事件 #0 user/message
=== ② 事件日志原文（唯一事实来源） ===
#0 turn/start {'turn': 1}
#1 step/start {'turn': 1, 'step': 1}
#2 system/message {'turn': 1, 'step': 1, 'content': '你是一个数学助手……'}
#3 user/message {'content': '1+2*3 等于几？'}
#4 request/header {'header': {'config': {...}, 'tools': (...)}, 'reason': 'initial'}
#5 assistant/message {'content': '', 'tool_calls': ({'id': 'call_00_...', 'name': 'calculator', ...},)}
#6 tool/call {'call_id': 'call_00_...', 'name': 'calculator', 'arguments': '{"expression": "1+2*3"}'}
#7 tool/result {'call_id': 'call_00_...', 'content': '7.0', 'is_error': False}
#8 step/end {'turn': 1, 'step': 1}
#9 step/start {'turn': 1, 'step': 2}
#10 assistant/message {'content': '根据计算，**1 + 2 × 3 = 7**。', 'tool_calls': ()}
#11 step/end {'turn': 1, 'step': 2}
#12 turn/end {'turn': 1, 'reason': 'completed'}
=== ③ 派生消息：模型看到的历史（derive_messages 投影） ===
[system] 你是一个数学助手……
[user] 1+2*3 等于几？
[assistant → 请求工具] calculator({"expression": "1+2*3"})
[tool → 结果] 7.0
[assistant] 根据计算，**1 + 2 × 3 = 7**。
运算顺序是先算乘法（2 × 3 = 6），再算加法（1 + 6 = 7）。
=== ④ 重放：同一份日志 → 新会话 → 完全相同的消息历史 ===
重放派生 5 条消息，与原始派生一致: True
← 日志是唯一事实来源：持久化、界面、重放都从它派生
第 ② 节完整记录了发生过的事情。在这次包含两个步骤的示例中，共有 13 条事件，其中请求头、步骤边界和工具调用记录不会转换成模型消息。第 ③ 节包含系统、用户、助手和工具共 5 条消息。第 ④ 节证明生成结果只依赖日志：同一份日志会得到相同的消息历史。事件 data 中的列表在冻结后变成元组，也能直观看到数据已经不可修改。
SessionEvent：日志中的一条事件，seq连续，写入后不可修改Session.append：统一的写入入口，负责校验、冻结和通知订阅者event_at/snapshot_events：按位置读取，或取得稳定区间快照derive_messages：从包含系统提示词的表层事件生成模型消息- 运行边界：用事件记录请求信息、轮次、步骤和工具调用
from_log：校验已有事件并重新建立会话
| 官方实现 | 我们对应实现 | 说明 |
|---|---|---|
packages/core/session/README.zh.md |
Session |
与官方一样使用事件重建状态，在追加事件时复制并冻结数据，外部代码不能修改事件快照 |
| 同上 | derive_messages |
对齐 V3 的 system/message 与表层替换语义；官方使用增量投影，教学版每次重新处理全部事件 |
| 同上 | subscribe |
官方持久化插件订阅 session/event，需要保存时再写入磁盘；第 08 章实现文件存储 |
packages/core/agent-loop/README.zh.md |
运行过程日志 | 官方同样把已经接收的消息、请求边界和工具调用写入日志，并在后续步骤重建模型输入 |
- 为什么“发生过什么”和“下一次应给模型看什么”不应共用一份可变消息列表？请分别从审计、界面展示、上下文成本和崩溃恢复的角度分析。
- 为一个代码助手设计事件日志。除用户、模型和工具消息外，再选择两类只用于检查过程或界面展示的事件，并说明它们为什么不应进入模型消息。
- 只追加日志便于回放，却也会长期保存错误或敏感信息。面对合规删除、数据更正和调试可追溯性之间的冲突，你会怎样设计更高一层的删除或遮蔽机制？
- 扩展本章 Session，使它能够记录并重放一种新的过程事件，同时为一个消费者提供派生视图。要求原始事件保持不可变，并用同一份日志验证重放前后的派生结果一致。