9.5 KiB
9.5 KiB
OpenClaw Session 解析与归档系统设计
1. 需求概述
1.1 背景
OpenClaw 多 Agent 系统运行于 Mac Mini + Ubuntu1 + Ubuntu2 三节点，每日产生大量 session JSONL 会话文件，散落于各服务器，缺乏统一查询和分析能力。
1.2 目标
- 各节点定时推送 JSONL 到 Mac Mini
- 解析后存入 PostgreSQL + TimescaleDB（三表关联：sessions / messages / tool_calls）
- Django Admin 提供可查询、可筛选的管理界面
1.3 核心分析维度
- Token 消耗趋势（按 Agent / 时间 / 模型）
- Tool 调用分析（最常用工具、失败率、平均耗时）
- 模型行为分析（thinking level 分布、模型切换频率、stopReason 分布）
2. 架构总览
┌─────────────────┐ push ┌──────────────────┐
│ Ubuntu1 │ ────────► │ Mac Mini │
└─────────────────┘ │ │
┌─────────────────┐ push │ Django + DRF │
│ Ubuntu2 │ ────────► │ │
└─────────────────┘ │ ├─ API接收 │
┌─────────────────┐ local │ ├─ 解析引擎 │
│ Mac Mini (本地) │ ────────► │ ├─ Django Admin │
└─────────────────┘ └────┬─────────────┘
│
┌─────┴─────────────┐
│ PostgreSQL + │
│ TimescaleDB │
└───────────────────┘
3. 数据源
3.1 JSONL 文件结构
每个 .jsonl 文件是一个 session，每行是一个 JSON 事件，事件类型：
| 类型 | 说明 |
|---|---|
session |
session 元信息（id、timestamp、cwd） |
model_change |
模型切换 |
thinking_level_change |
思考级别变化 |
custom |
自定义事件（如 model-snapshot） |
message |
对话消息（user / assistant / toolResult） |
message 的 content 是数组，每个 content 可能是：
text— 纯文本thinking— 推理过程toolCall— 工具调用
3.2 文件命名约定
- 普通 session:
{uuid}.jsonl - 带 topic:
{uuid}-topic-{timestamp}.jsonl - 被重置过的:
{uuid}.jsonl.reset.{timestamp} - 被删除的:
{uuid}.jsonl.deleted.{timestamp}
解析时需跳过 .deleted. 文件。
3.3 Agent 识别
session 文件位于 agents/{agent_name}/sessions/ 路径下，从路径提取 agent_name。
4. 数据模型
4.1 Session 表
每个 JSONL 文件一条记录。
class Session(models.Model):
session_id = models.UUIDField() # session UUID
agent_name = models.CharField() # 如 xingyao
source_node = models.CharField() # 来源节点: macmini / ubuntu1 / ubuntu2
session_version = models.IntegerField() # version from JSONL
model_provider = models.CharField() # 最后一个 model_change 的 provider
model_id = models.CharField() # 最后一个 model_change 的 modelId
thinking_level = models.CharField() # 最后一个 thinking_level_change
start_time = models.DateTimeField() # 第一个事件的 timestamp
end_time = models.DateTimeField() # 最后一个事件的 timestamp
cwd = models.CharField() # 工作目录
total_tokens = models.IntegerField() # 所有 assistant message 的 usage.totalTokens 之和
total_cost = models.FloatField() # 所有 assistant message 的 cost.total 之和
message_count = models.IntegerField() # message 数量
tool_call_count = models.IntegerField() # toolCall 数量
error_count = models.IntegerField() # isError=True 的 toolResult 数量
raw_file_path = models.TextField() # 原始 JSONL 在 Mac Mini 的本地路径
pushed_at = models.DateTimeField() # 推送/入库时间
status = models.CharField() # active / deleted
metadata = models.JSONField() # 原始 session 和 custom 事件的原始 JSON
4.2 Message 表
每个 type: message 的事件一条记录，外键关联 Session。
class Message(models.Model):
session = models.ForeignKey(Session)
message_id = models.CharField() # 事件 id
parent_id = models.CharField() # parentId
seq = models.IntegerField() # 在 session 中的顺序
role = models.CharField() # user / assistant / toolResult
content_text = models.TextField() # 提取的纯文本（去 JSON 结构，用于全文搜索）
raw_content = models.JSONField() # message.content 原始 JSON
raw_message = models.JSONField() # 整条 message 原始 JSON
timestamp = models.DateTimeField()
# assistant 专用
model = models.CharField()
provider = models.CharField()
stop_reason = models.CharField()
tokens_input = models.IntegerField()
tokens_output = models.IntegerField()
tokens_cache_read = models.IntegerField()
tokens_cache_write = models.IntegerField()
tokens_total = models.IntegerField()
cost_total = models.FloatField()
# toolResult 专用
tool_call_id = models.CharField() # 对应哪个 toolCall
tool_name = models.CharField()
is_error = models.BooleanField()
exit_code = models.IntegerField()
duration_ms = models.IntegerField()
4.3 ToolCall 表
每个 type: toolCall 的 content 块一条记录。
class ToolCall(models.Model):
session = models.ForeignKey(Session)
message = models.ForeignKey(Message)
tool_call_id = models.CharField() # call_function_xxx_N
tool_name = models.CharField() # 如 exec
arguments = models.JSONField() # 调用参数原始 JSON
# 执行结果（从对应的 toolResult 关联回来）
result_text = models.TextField() # toolResult 的提取文本
is_error = models.BooleanField()
exit_code = models.IntegerField()
duration_ms = models.IntegerField()
seq = models.IntegerField() # 在该 message 中的顺序
4.4 TimescaleDB 时序优化
Session.start_time设为 hypertable 的分区列（按天分区）Message.timestamp设为 hypertable 的分区列（按天分区）ToolCall.created_at设为 hypertable 的分区列（按天分区）
5. 推送机制
5.1 推送脚本
Python 脚本部署在所有三个节点。
Usage: python sync_sessions.py --remote-url http://macmini:8000/api/sessions/upload/
- 扫描各节点的 sessions 目录
- 维护本地
.sync_state文件，记录已推送文件的 mtime - 筛选条件：只推 mtime 较新的
.jsonl文件（排除.deleted.） - 用 multipart/form-data 上传，同时发送
agent_name和source_node - 上传成功后更新
.sync_state
5.2 定时任务
各节点 crontab 每天凌晨 2:00 执行一次：
0 2 * * * cd /path/to/script && python sync_sessions.py
6. API 设计
POST /api/sessions/upload/
接收 JSONL 文件上传。
Content-Type: multipart/form-data
Fields:
- file: JSONL 文件内容
- agent_name: str
- source_node: str
响应：
{
"status": "ok",
"session_id": "xxx",
"messages_parsed": 42,
"tool_calls_parsed": 15
}
幂等：根据 session_id + agent_name 组合判断重复，已存在则跳过。
7. Django Admin
7.1 Session 列表页
- 可排序列：时间、Agent、模型、token 总量、message 数量
- 侧边栏筛选：agent_name、source_node、model_id、日期范围
- 搜索框：session_id、cwd
- 自定义列表操作：查看完整对话原文（渲染 JSONL 为人类可读 HTML）
7.2 Session 详情页（Inline）
- Message 按 seq 排序的内联列表
- 每条 message 可折叠展开
- 区分 user / assistant / toolResult 的视觉样式
- assistant 显示 thinking、text、toolCall
- toolResult 显示执行结果和错误状态
- ToolCall 独立 tab，可排序
7.3 Message 列表页
- 按角色、时间范围、model_id、tool_name 筛选
- 搜索 content_text 全文
7.4 ToolCall 列表页
- 按 tool_name、is_error、exit_code 筛选
- 排序：duration_ms、token 消耗
8. 解析引擎
同步解析，流程：
接收文件 → 逐行 JSON → 构建事件流 →
提取 session 元信息 →
写入 Session →
按 parentId 树关系遍历 message →
写入 Message（提取 assistant usage、toolResult 元信息）→
提取 toolCall 内容块 →
写入 ToolCall（关联 message 和对应的 toolResult）
解析失败处理：单条 JSON 行解析失败记录日志但不中断，整文件解析失败时回滚整个事务。
9. 技术栈
| 组件 | 版本 |
|---|---|
| Python | 3.12+ |
| Django | 5.x |
| Django REST Framework | 3.15+ |
| PostgreSQL | 16+ |
| TimescaleDB | 2.14+ |
| psycopg | 3.x |
| 数据库分区 | 按天分区，保留 90 天热数据 |
10. 非技术决策
- 不接收
.deleted.文件 - 幂等上传：同一 session 重复推送时跳过（基于 session_id + agent_name 组合键）
- Mac Mini 本地 session 同样通过推送脚本入库（保证统一流程，不走特殊路径）
- 原始 JSONL 保留一份到指定归档路径，供原始数据追溯