Files

weishen 1d2531a71f 完善spec：本地解析推送结构化数据 & Django Admin每日导出

- 架构调整：三节点本地解析JSONL，推送结构化JSON而非原始文件
- API改为bulk_upsert接口，接收sessions/messages/tool_calls数组
- 推送脚本改用纯标准库，由OpenClaw cron job每天凌晨触发
- 新增Admin按时间范围查询对话和Daily Markdown导出功能
- Markdown导出精简thinking，保留user/assistant对话和tool调用
- 非技术决策：原始JSONL保留在各节点本地

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

2026-04-05 13:32:24 +08:00

16 KiB

Raw Permalink Blame History

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. 架构总览

┌─────────────────┐    parse    push    ┌──────────────────┐
│ Ubuntu1         │ ──────────►         │  Mac Mini        │
└─────────────────┘   structured data    │                  │
┌─────────────────┐    (JSON)     push   │  Django + DRF    │
│ Ubuntu2         │ ──────────►         │                  │
└─────────────────┘   structured data    │  ├─ API接收       │
┌─────────────────┐    (JSON)     push   │  ├─ 数据写入      │
│ Mac Mini (本地)  │ ──────────────────►  │  ├─ Django Admin  │
└─────────────────┘   structured data    └────┬─────────────┘
                     (JSON)                   │
                                        ┌─────┴─────────────┐
                                        │ PostgreSQL +      │
                                        │ TimescaleDB (NAS) │
                                        └───────────────────┘

各节点解析脚本负责：读取 JSONL → 本地解析为结构化数据（sessions / messages / tool_calls）→ POST JSON 到 Django API → 服务端幂等写入数据库。Django 服务端不再做 JSONL 解析。

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. 数据模型

所有表结构通过 Django Model + Django Migration 进行定义和映射，不直接执行 DDL。TimescaleDB 的 hypertable 创建通过 RunSQL 操作嵌入 Django migration 中。

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 时序优化

在 Django 的最后一张数据表 migration 后，新增 migration 使用 RunSQL 创建 hypertables
Session.start_time 设为 hypertable 的分区列（按天分区）
Message.timestamp 设为 hypertable 的分区列（按天分区）
ToolCall 表使用关联 Message.timestamp 进行分区（不额外添加字段，通过消息时间推导）

5. 推送机制

5.1 解析与推送脚本

Python 脚本部署在所有三个节点（Mac Mini / Ubuntu1 / Ubuntu2）。

Usage: python sync_sessions.py --remote-url http://macmini:8000/api/sessions/bulk_upsert/

职责：

扫描本地 agents/{agent_name}/sessions/ 目录
读取 .sync_state 文件，找出新增/变更的 .jsonl 文件（排除 .deleted.）
本地解析每个 JSONL：逐行 JSON → 提取 session 元信息 / messages / tool_calls
POST 结构化 JSON 到 Django API（POST /api/sessions/bulk_upsert/）
上传成功后更新 .sync_state

脚本特点：

纯 Python 标准库，无外部依赖
本地完成解析，服务端只负责结构化写入
幂等：服务端根据 session_id + agent_name 判断重复，重复推送跳过

请求体示例：

{
  "agent_name": "xingyao",
  "source_node": "macmini",
  "sessions": [
    {
      "session_id": "xxx-xxx",
      "session_version": 1,
      "model_provider": "anthropic",
      "model_id": "claude-sonnet-4-6",
      ...
    }
  ],
  "messages": [
    {
      "session_id": "xxx-xxx",
      "message_id": "yyy",
      "role": "user",
      ...
    }
  ],
  "tool_calls": [
    {
      "session_id": "xxx-xxx",
      "message_id": "yyy",
      "tool_call_id": "call_function_xxx_0",
      "tool_name": "exec",
      ...
    }
  ]
}

5.2 定时任务

各节点由 OpenClaw cron job 每天凌晨 2:00 触发一次：

0 2 * * * cd /path/to/script && python sync_sessions.py

6. API 设计

POST /api/sessions/bulk_upsert/

接收结构化 JSON 批量写入。

Content-Type: application/json
Body:
  {
    "agent_name": "xingyao",
    "source_node": "macmini",
    "sessions": [...],
    "messages": [...],
    "tool_calls": [...]
  }

响应：

{
  "status": "ok",
  "sessions_upserted": 3,
  "messages_upserted": 42,
  "tool_calls_upserted": 15
}

幂等：根据 session_id + agent_name 组合判断重复，已存在则跳过。整个请求在一个事务中写入，任一失败则全部回滚。

7. Django Admin

7.1 Session 列表页

可排序列：时间、Agent、模型、token 总量、message 数量
侧边栏筛选：agent_name、source_node、model_id、日期范围
搜索框：session_id、cwd
查看完整对话原文（渲染为人类可读 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 消耗

7.5 按时间范围查询对话

Admin 增加自定义视图：/admin/openclaw/daily/
支持选择日期范围 + agent_name
按 session 分组，按时间顺序展示完整对话内容
每条消息区分 user / assistant / toolResult，去掉 thinking 部分
人类可读 HTML，支持折叠/展开

7.6 Daily 复盘导出（Markdown）

Admin Session 列表页增加自定义 Action：选中 sessions → 导出 Markdown
按天为单位生成 daily-report-{YYYY-MM-DD}.md 文件并下载
精简规则：
- 去掉所有 thinking 内容块
- 保留 agent 纯文本回复
- 保留 user 原始输入
- 保留 tool 调用（工具名 + 参数摘要 + 执行结果文本）
- 保留 token 消耗、模型信息等摘要统计

Markdown 格式示例：

# Daily Report: 2026-04-05

## Session: abc12345 (Agent: xingyao)
**模型**: claude-sonnet-4-6 | **Token**: 45,230

### 10:23 User
帮我看看这个 bug...

### 10:23 Assistant
问题出在第 45 行...

### 10:24 Assistant → [Tool: exec]
`grep -n "bug" file.py`
**结果**: 匹配到 3 行

8. 解析引擎

8.1 客户端解析（sync_sessions.py）

运行在各节点，负责读取 JSONL 并提取结构化数据：

逐行读取 JSONL → 
提取 session 事件（id、version、cwd）→ 
跟踪 model_change / thinking_level_change 状态 →
遍历 message 事件 →
  提取 assistant usage（tokens、cost）→
  提取 content 数组中的 text / toolCall 块 →
遍历 toolResult 事件 → 关联到 toolCall
构建 JSON 批量提交

8.2 服务端写入（Django API）

接收客户端推送的结构化 JSON：

验证 agent_name + source_node →
批量 upsert Session（幂等：session_id + agent_name 去重）→
批量写入 Message（外键关联 Session）→
批量写入 ToolCall（外键关联 Message）→
更新 Session 聚合字段（token 总数、cost、message count 等）→
返回写入结果

整个写入在一个事务中完成，任一失败则全部回滚。

9. 技术栈

组件	版本
Python	3.12+
Django	5.x
Django REST Framework	3.15+
PostgreSQL	16+
TimescaleDB	2.14+
psycopg	3.x
数据库分区	按天分区，保留 90 天热数据

10. Docker Compose 部署

10.1 架构决策

Docker Compose 仅管理 Django 应用服务（web），数据库连接远程 NAS 上的 PostgreSQL + TimescaleDB
Django 服务器与 NAS 处于同一局域网，通过内网 IP 直连
所有可变配置通过 .env 文件注入，该文件加入 .gitignore

10.2 项目结构

agent-base/
├── docker-compose.yml      # 编排配置
├── Dockerfile              # Django 镜像构建
├── nginx/
│   └── nginx.conf.placeholder  # Nginx 反代占位配置（预留）
├── .env                    # 环境变量（不提交）
├── .env.example            # 环境变量模板（提交）
├── .dockerignore
├── manage.py
├── src/                    # Django 项目源码
│   ├── openclaw/           # app
│   └── config/             # 项目配置
└── docs/                   # 文档

10.3 `.env` 配置

# .env.example（提交到仓库）
# Django
DJANGO_SECRET_KEY=CHANGE_ME
DJANGO_PORT=8000
DJANGO_ALLOWED_HOSTS=*

# Database (NAS)
DB_HOST=192.168.x.x
DB_PORT=5432
DB_NAME=openclaw_archive
DB_USER=openclaw
DB_PASSWORD=CHANGE_ME

.env 文件被加入 .gitignore，仅保留 .env.example 作为模板。

10.4 Dockerfile

FROM python:3.12-slim

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential libpq-dev \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE ${DJANGO_PORT:-8000}

CMD ["gunicorn", "--bind", "0.0.0.0:${DJANGO_PORT:-8000}", \
     "--workers", "4", "--timeout", "120", \
     "config.wsgi:application"]

Gunicorn 4 个 worker，120 秒超时（解析 JSONL 可能耗时的场景）。

10.5 docker-compose.yml

services:
  web:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: openclaw-archive
    env_file:
      - .env
    ports:
      - "${DJANGO_PORT:-8000}:${DJANGO_PORT:-8000}"
    volumes:
      - static_volume:/app/staticfiles
      - jsonl_archive:/app/archive
    restart: unless-stopped

  # nginx 占位（预留）
  # nginx:
  #   image: nginx:alpine
  #   container_name: openclaw-nginx
  #   ports:
  #     - "80:80"
  #   volumes:
  #     - ./nginx/nginx.conf:/etc/nginx/nginx.conf
  #     - static_volume:/app/staticfiles:ro
  #   depends_on:
  #     - web

volumes:
  static_volume:
  jsonl_archive:  # 原始 JSONL 归档存储

10.6 部署与运维命令

# 首次部署
cp .env.example .env
vim .env                     # 填写真实配置
docker compose build
docker compose run --rm web python manage.py migrate
docker compose run --rm web python manage.py createsuperuser
docker compose up -d

# 日常运维
docker compose ps              # 查看状态
docker compose logs -f web     # 查看日志
docker compose down            # 停止
docker compose up -d --build   # 重建并重启

# 数据库迁移
docker compose run --rm web python manage.py makemigrations
docker compose run --rm web python manage.py migrate

10.7 Nginx 反代（预留）

当前阶段不使用 Nginx。docker-compose.yml 中已预留 Nginx 配置（注释状态），生产环境需要 HTTPS、静态文件优化时取消注释即可启用。

启用前需将 nginx/nginx.conf.placeholder 重命名为 nginx.conf 并根据实际域名和证书路径修改配置。

12. 非技术决策

不接收 .deleted. 文件
幂等推送：同一 session 重复推送时跳过（基于 session_id + agent_name 组合键）
Mac Mini 本地 session 同样通过推送脚本入库（保证统一流程，不走特殊路径）
原始 JSONL 保留在各节点本地，不推送到 Django 服务端

16 KiB Raw Permalink Blame History Unescape Escape