DeepSeek Harness: 会话日志与 Trajectory
最后更新:2026-08-31
每次 Agent 对话都是一段不可重现的旅程——模型有随机性、工具有副作用、上下文会累积。DSH 的 Trajectory 系统用"仅追加日志"完整记录每一步,让你能回溯、审计、fork、恢复任何一个时间点的会话状态。
📋 前置知识:已完成 07-python-sdk.md,了解 SDK 基础
1. 你将学到
- 仅追加日志(append-only log)设计原理
- SessionEvent 事件流类型与结构
- Trajectory 视图的使用方法
- 会话 fork 与恢复机制
- 日志持久化与导出
2. 仅追加日志设计
(1) 为什么是仅追加?
传统日志系统允许修改和删除,但 Agent 会话日志必须是不可变的——就像飞行数据记录仪(黑匣子),记录一旦写入就不能更改:
graph LR
E1[Event 1] --> E2[Event 2] --> E3[Event 3] --> E4[Event 4] --> E5[Event 5]
E5 -.->|只追加| NEW[Event 6]
style E1 fill:#e8f5e9
style E2 fill:#e8f5e9
style E3 fill:#e8f5e9
style E4 fill:#e8f5e9
style E5 fill:#e8f5e9
style NEW fill:#fff3e0
仅追加设计的三大原则:
| 原则 | 说明 | 好处 |
|---|---|---|
| 不可变 | 日志一旦写入不可修改或删除 | 审计追踪完整 |
| 有序 | 事件严格按时间顺序排列 | 重放可复现 |
| 仅追加 | 只能新增事件,不能回删 | 无并发冲突 |
(2) 与传统日志的对比
| 维度 | 传统日志 | DSH 仅追加日志 |
|---|---|---|
| 可修改 | ✅ 可以修改/删除 | ❌ 不可修改 |
| 并发安全 | 需要锁机制 | 天然安全(只追加) |
| 回滚能力 | 依赖备份 | 从任意点恢复 |
| 审计能力 | 可能被篡改 | 不可篡改 |
| 存储效率 | 可压缩 | 持续增长(需定期归档) |
(3) 日志存储结构
.dsh/
└── sessions/
└── sess_abc123/
├── events.log # 事件日志(仅追加)
├── snapshots/ # 状态快照
│ ├── snap_001.json
│ ├── snap_002.json
│ └── snap_003.json
└── metadata.json # 会话元数据
3. SessionEvent 事件流
(1) 事件类型
DSH 会话中的每一个操作都被记录为 SessionEvent:
type SessionEventType =
| 'session.created'
| 'session.config_changed'
| 'user.message'
| 'agent.message'
| 'agent.thinking'
| 'tool.call'
| 'tool.result'
| 'tool.approval.requested'
| 'tool.approval.resolved'
| 'session.forked'
| 'session.restored'
| 'error.occurred';
(2) 事件结构
每个 SessionEvent 包含标准字段:
interface SessionEvent {
id: string; // 唯一事件 ID
type: SessionEventType; // 事件类型
timestamp: number; // Unix 时间戳(毫秒)
sessionId: string; // 所属会话 ID
data: Record<string, unknown>; // 事件负载数据
parentId?: string; // 父事件 ID(fork 时使用)
}
(3) 各事件详解
用户消息事件:
▶ 示例 1user.message 事件
{
"id": "evt_001",
"type": "user.message",
"timestamp": 1724486400000,
"sessionId": "sess_abc123",
"data": {
"content": "帮我重构 utils 目录",
"attachments": []
}
}
工具调用事件:
▶ 示例 2tool.call 事件
{
"id": "evt_002",
"type": "tool.call",
"timestamp": 1724486401500,
"sessionId": "sess_abc123",
"data": {
"tool": "search",
"params": {
"pattern": "utils/*",
"type": "file"
},
"mode": "standard"
}
}
工具结果事件:
▶ 示例 3tool.result 事件
{
"id": "evt_003",
"type": "tool.result",
"timestamp": 1724486402300,
"sessionId": "sess_abc123",
"data": {
"toolCallId": "evt_002",
"status": "success",
"result": {
"files": ["utils/format.ts", "utils/validate.ts", "utils/helpers.ts"]
},
"duration_ms": 800
}
}
审批事件:
▶ 示例 4tool.approval 事件
{
"id": "evt_004",
"type": "tool.approval.requested",
"timestamp": 1724486403000,
"sessionId": "sess_abc123",
"data": {
"tool": "file_edit",
"params": {
"action": "edit",
"path": "utils/format.ts"
},
"riskLevel": "high"
}
}
{
"id": "evt_005",
"type": "tool.approval.resolved",
"timestamp": 1724486405000,
"sessionId": "sess_abc123",
"data": {
"approvalId": "evt_004",
"decision": "allowed",
"decidedBy": "user"
}
}
(4) 完整事件流示例
时间线 事件类型
─────────────────────────────────────────
10:00:00.000 session.created
10:00:05.120 user.message "帮我重构 utils 目录"
10:00:06.300 tool.call search → utils/*
10:00:07.100 tool.result 找到 3 个文件
10:00:08.200 tool.call file_edit → read utils/format.ts
10:00:08.500 tool.result 文件内容返回
10:00:10.800 agent.thinking 分析重构方案...
10:00:12.000 tool.approval.requested file_edit → edit
10:00:15.000 tool.approval.resolved → allowed
10:00:15.200 tool.call file_edit → edit utils/format.ts
10:00:15.600 tool.result 编辑完成
10:00:17.000 agent.message "重构完成!"
4. Trajectory 视图
(1) 什么是 Trajectory?
Trajectory 是会话日志的可视化界面,展示 Agent 的完整"轨迹":
┌─ Trajectory View ──────────────────────────────────────┐
│ │
│ 10:00 👤 帮我重构 utils 目录 │
│ 10:00 🔍 search(utils/*) → 3 files 0.8s │
│ 10:00 📄 file_edit(read) → utils/format.ts 0.3s │
│ 10:00 📄 file_edit(read) → utils/validate.ts 0.2s │
│ 10:00 🤔 Thinking... 分析重构方案 │
│ 10:00 ⚠️ Approval: edit utils/format.ts │
│ 10:00 → ✅ Allowed │
│ 10:00 📝 file_edit(edit) → utils/format.ts 0.4s │
│ 10:00 ⚠️ Approval: edit utils/validate.ts │
│ 10:00 → ✅ Allowed │
│ 10:00 📝 file_edit(edit) → utils/validate.ts 0.3s │
│ 10:00 🤖 重构完成!提取了共享类型定义... │
│ │
│ [Fork from here] [Restore to here] [Export] │
└─────────────────────────────────────────────────────────┘
(2) Web UI 访问 Trajectory
在 Web UI 中,点击顶部控制栏的日志图标即可打开 Trajectory 视图:
顶部控制栏 → 📋 → Trajectory
(3) Trajectory 过滤与搜索
▶ 示例 5按事件类型过滤
Trajectory View 过滤器:
┌──────────────────────────────────────────┐
│ 过滤: │
│ ☑ user.message ☑ agent.message │
│ ☑ tool.call ☑ tool.result │
│ ☐ agent.thinking ☐ approval events │
│ │
│ 搜索:[输入关键词...] │
└──────────────────────────────────────────┘
(4) SDK 访问 Trajectory
▶ 示例 6通过 SDK 获取事件流
from dsh import DSHClient
client = DSHClient(base_url="http://127.0.0.1:3080")
session = client.get_session("sess_abc123")
# 获取所有事件
events = session.get_trajectory()
for event in events:
print(f"[{event.timestamp}] {event.type}: {event.data}")
# 按类型过滤
tool_events = session.get_trajectory(event_type="tool.call")
for event in tool_events:
print(f"Tool: {event.data['tool']}")
print(f"Params: {event.data['params']}")
交互流程:
[1724486400000] session.created: {...} [1724486405120] user.message: "帮我重构 utils 目录" [1724486406300] tool.call: search → utils/* [1724486407100] tool.result: 找到 3 个文件 ...(更多事件) Filtered (tool.call): 3 events Tool: search | Params: {'pattern': 'utils/*'} Tool: file_edit | Params: {'action': 'edit', 'path': 'utils/format.ts'}⚠️ 你的实际事件流取决于会话内容,但事件类型和结构应与示例一致。
5. 会话 Fork 与恢复
(1) Fork 的概念
Fork 是从某个时间点创建会话分支——主线继续前进,分支独立发展:
graph LR
E1[Event 1] --> E2[Event 2] --> E3[Event 3] --> E4[Event 4]
E3 -->|fork| F1[Fork Event 1] --> F2[Fork Event 2]
E4 --> E5[Event 5]
style E1 fill:#e8f5e9
style E2 fill:#e8f5e9
style E3 fill:#e8f5e9
style E4 fill:#e8f5e9
style E5 fill:#e8f5e9
style F1 fill:#e3f2fd
style F2 fill:#e3f2fd
(2) Fork 的使用场景
| 场景 | 说明 |
|---|---|
| 方案探索 | 从同一节点尝试不同方案,对比效果 |
| 安全回退 | 做破坏性操作前 fork,失败了回主线 |
| A/B 测试 | 同一任务用不同模型/模式对比 |
| 实验性修改 | 不确定效果时,先在分支试 |
(3) Web UI 中 Fork
在 Trajectory 视图中,点击任意事件旁的 "Fork from here" 按钮:
10:00 📝 file_edit(edit) → utils/format.ts [Fork from here]
10:00 ⚠️ Approval: edit utils/validate.ts [Fork from here]
10:00 🤖 重构完成! [Fork from here]
Fork 后会创建一个新会话,从选定事件点开始,复制该点之前的所有上下文。
(4) SDK 中 Fork
▶ 示例 7SDK Fork 操作
client = DSHClient(base_url="http://127.0.0.1:3080")
session = client.get_session("sess_abc123")
# 从第 5 个事件处 fork
forked = session.fork(after_event="evt_005")
print(f"Forked session: {forked.id}")
print(f"Parent: {forked.parent_id}")
print(f"Fork point: evt_005")
# 在 fork 分支中继续对话
response = forked.send("换一种方式重构,按功能分文件")
交互流程:
📋 Forking session sess_abc123 at evt_005... ✅ Forked session created: sess_xyz789 📎 Parent: sess_abc123 | Fork point: evt_005 🤖 Agent: [在分支中] 换一种方式重构,按功能分文件...⚠️ 分支中的 Agent 行为完全独立于主线,输出因模型而异。
(5) 恢复(Restore)
恢复与 Fork 不同——Restore 将当前会话回退到指定事件点,丢弃之后的事件:
▶ 示例 8SDK 恢复操作
# 恢复到第 3 个事件点
session.restore(to_event="evt_003")
# 此后 evt_004、evt_005 等事件被标记为 "restored-away"
# 新事件从 evt_003 之后继续追加
输出:
TEXT 📖 仅展示✅ Session restored to evt_003 ⏭️ Events evt_004..evt_005 marked as restored-away 📝 session.restored event appended
注意:恢复操作不会删除事件(仅追加原则),而是标记后续事件为无效,并添加一个
session.restored事件。
6. 日志持久化
(1) 默认存储
DSH 会话日志默认存储在项目目录的 .dsh/sessions/ 下:
.dsh/
├── sessions/
│ ├── sess_abc123/
│ │ ├── events.log # 事件日志
│ │ ├── snapshots/ # 状态快照
│ │ └── metadata.json # 元数据
│ └── sess_def456/
│ ├── events.log
│ └── ...
├── config.yaml # DSH 配置
└── plugins/ # 插件目录
(2) 持久化配置
# dsh.config.yaml
storage:
# 存储路径
base_path: ".dsh/sessions"
# 快照策略
snapshots:
enabled: true
interval: 10 # 每 10 个事件保存一次快照
max_snapshots: 5 # 最多保留 5 个快照
# 日志轮转
rotation:
max_size_mb: 100 # 单个日志文件最大 100MB
max_files: 50 # 最多保留 50 个会话
# 归档
archive:
enabled: true
path: ".dsh/archive/"
after_days: 30 # 30 天后自动归档
输出:
TEXT 📖 仅展示✅ dsh.config.yaml storage validated 📂 Base path: .dsh/sessions 📸 Snapshots: every 10 events, max 5 kept 🔄 Rotation: max 100MB/file, max 50 sessions 📦 Archive: after 30 days → .dsh/archive/
(3) 导出会话日志
▶ 示例 9导出为 JSON
# 导出完整会话日志
session = client.get_session("sess_abc123")
events = session.get_trajectory()
import json
with open("session_export.json", "w") as f:
json.dump([e.to_dict() for e in events], f, indent=2)
输出:
TEXT 📖 仅展示✅ Exported 15 events to session_export.json 📊 File size: 4.2KB
▶ 示例 10导出为 Markdown
# CLI 导出
dsh session export sess_abc123 --format markdown --output session.md
# 输出格式
# # Session: sess_abc123
# ## 10:00 - User
# 帮我重构 utils 目录
# ## 10:00 - Tool: search
# Pattern: utils/* → 3 files found
# ...
(4) 日志清理
# 列出所有会话(按大小排序)
dsh session list --sort size
# 归档旧会话
dsh session archive --older-than 30d
# 删除归档(不可逆)
dsh session clean --archived-only
7. Trajectory 与审计
(1) 操作审计
Trajectory 记录了 Agent 的每一个操作,天然适合审计:
graph TB
AUDIT[审计需求] --> T1[谁执行的?]
AUDIT --> T2[什么时候?]
AUDIT --> T3[做了什么?]
AUDIT --> T4[结果如何?]
T1 --> TRAJ[Trajectory 事件流]
T2 --> TRAJ
T3 --> TRAJ
T4 --> TRAJ
(2) 合规场景
| 合规要求 | Trajectory 如何满足 |
|---|---|
| 操作可追溯 | 每个事件有 ID、时间戳、操作者 |
| 变更不可篡改 | 仅追加日志,无法修改历史 |
| 审批记录 | approval.requested + approval.resolved 完整记录 |
| 回滚能力 | 从任意检查点恢复 |
(3) 生成审计报告
▶ 示例 11生成审计报告
session = client.get_session("sess_abc123")
events = session.get_trajectory()
report = {
"session_id": session.id,
"duration": events[-1].timestamp - events[0].timestamp,
"user_messages": len([e for e in events if e.type == "user.message"]),
"tool_calls": len([e for e in events if e.type == "tool.call"]),
"approvals_requested": len([e for e in events if e.type == "tool.approval.requested"]),
"approvals_denied": len([e for e in events if e.type == "tool.approval.resolved" and e.data.get("decision") == "denied"]),
"files_modified": list(set([
e.data.get("path") for e in events
if e.type == "tool.call" and e.data.get("tool") == "file_edit"
])),
"errors": len([e for e in events if e.type == "error.occurred"])
}
import json
print(json.dumps(report, indent=2))
输出:
TEXT 📖 仅展示{ "session_id": "sess_abc123", "duration": 17000, "user_messages": 3, "tool_calls": 5, "approvals_requested": 2, "approvals_denied": 0, "files_modified": ["utils/format.ts", "utils/validate.ts"], "errors": 0 }
❓ 常见问题
rotation.max_size_mb 和 archive.after_days 可自动管理存储。dsh session export,SDK 用 get_trajectory() 方法。📖 小节
- DSH 会话日志采用仅追加设计:不可变、有序、只增不减
- SessionEvent 包含 13 种事件类型,覆盖对话、工具、审批、错误等
- Trajectory 视图可视化展示 Agent 的完整执行轨迹
- Fork 从任意节点创建分支,不影响主线;Restore 回退到指定节点
- 日志持久化支持快照、轮转、归档、导出
- Trajectory 天然支持操作审计和合规需求
- SDK 和 CLI 都可以访问和操作 Trajectory 数据
📝 作业
1. ⭐ 基础题:完成一次 Agent 对话(至少包含 2 次工具调用),打开 Trajectory 视图,列出所有事件及其类型和时间戳。
2. ⭐⭐ 进阶题:从一次会话的第 3 个事件处 Fork 出一个分支,在分支中尝试不同的方案。对比主线和分支的最终结果差异。
3. ⭐⭐⭐ 挑战题:用 Python SDK 编写一个 Trajectory 分析工具——输入会话 ID,自动生成审计报告(包含操作统计、审批记录、修改文件列表、错误汇总),输出为 JSON 格式。