Claude Code Transcript → Langfuse 方案实现说明

目录：langfuse_trace/
样例数据：transcript_demo/b0dcb60c-5539-42a7-b71b-80ffc878ec0f.jsonl

一、方案概览

Claude Code 将会话过程以 JSONL 形式写入本地 transcript。本方案在不改 Claude Code 核心的前提下，将 transcript 增量解析为 Turn（用户轮次），并上报到 Langfuse，实现：

主会话每轮对话 → 独立 Trace
模型调用 → Generation（含 token、真实时间戳）
工具调用 → Tool Span
子代理（Agent/Task）→ 嵌套在 Tool: Agent 下的 Subagent Turn

提供 两种上报方式，共享 common/langfuse_hook_all_agents.py 中的解析与 emit 逻辑：

维度	方案 A：Hook	方案 B：Watcher
入口	Claude Code `Stop` 事件触发	独立守护进程轮询 jsonl
实时性	每轮结束上报	默认每 1s 轮询 + 流式折中（trace 即时建、tool 实时 span）
依赖	需配置 `settings.json` hooks	无需 Hook，后台运行即可
典型场景	与 Claude Code 深度集成	CI/旁路监听/不便改 Hook 时

二者勿同时启用，否则可能重复上报同一 Turn。

二、目录结构

langfuse_trace/
├── 方案实现说明.md              # 本文档
├── README.md                    # 快速入门
├── requirements.txt             # langfuse>=2,<3
├── common/
│   └── langfuse_hook_all_agents.py   # 共享核心：解析 transcript + 上报 Langfuse
├── hook/                        # 方案 A：Stop Hook
│   ├── langfuse_hook_all_agents.py   # Hook 入口（薄封装 → common）
│   ├── e2e_langfuse_test.py
│   └── run_e2e_langfuse_test.sh
├── watcher/                     # 方案 B：jsonl 轮询 Watcher
│   ├── claude_langfuse_watcher.py
│   ├── run_watcher.sh           # 常驻监听入口（推荐）
│   ├── e2e_langfuse_watcher_test.py
│   └── run_e2e_langfuse_watcher_test.sh
├── transcript_demo/             # 脱敏样例（主 jsonl + subagents）
├── tests/
│   └── test_usage_transcript.py
└── docs/                        # 历史文档

三、Transcript 简介

3.1 存储位置

官方 Claude Code：

~/.claude/projects/<项目哈希>/
├── <sessionId>.jsonl                    # 主会话
└── <sessionId>/subagents/               # 或同级 agent-*.jsonl
    ├── agent-<id>.jsonl
    └── agent-<id>.meta.json

trpc-claudecode（本环境常用）：

~/.trpc-claudecode/projects/<编码项目路径>/
├── <sessionId>.jsonl                    # 主会话
└── <sessionId>/subagents/
    ├── agent-<id>.jsonl
    └── agent-<id>.meta.json

Watcher 默认监听 ~/.trpc-claudecode/projects（见 run_watcher.sh），也可通过 --projects-root 指向 ~/.claude/projects。

3.2 每行事件类型

type	含义
`user`	用户输入，或 `tool_result` 回灌
`assistant`	模型文本 / `tool_use`
`attachment`	Skill 列表等上下文注入
`system`	系统元数据

trpc-claudecode 拆行：同一 API 响应可能占多行 jsonl，共享 message.id——一行 text、每个 tool_use 各占一行。解析时按 message.id 分组，不能简单按 id 去重。

3.3 Turn 划分规则

user（真人，非 synthetic）  → 新 Turn
assistant                   → 追加当前 Turn
user + tool_result          → 归档工具结果，不新开 Turn

Synthetic user（如 Skill 注入的 Base directory for this skill:）不拆 Turn。

四、核心架构（共享逻辑）

4.0 数据流总览

批量模式（CC_LANGFUSE_STREAMING=false，Hook 默认；Watcher 可关闭）：

┌──────────────────────────────────────────────────────────────┐
│  输入：主 session jsonl（增量 offset + 半行 buffer）           │
└────────────────────────────┬─────────────────────────────────┘
                             ▼
┌──────────────────────────────────────────────────────────────┐
│  read_new_jsonl() → build_turns() → completed_turns()        │
│  discover_subagent_files() → match_subagent()                │
└────────────────────────────┬─────────────────────────────────┘
                             ▼
┌──────────────────────────────────────────────────────────────┐
│  emit_main_turn()：Turn 全部完成后一次性上报                   │
│  Langfuse v2 API: trace → generation + tool span + 嵌套      │
└────────────────────────────┬─────────────────────────────────┘
                             ▼
┌──────────────────────────────────────────────────────────────┐
│  flush() 成功 → 推进 turn_count / offset；失败则下次重试        │
└──────────────────────────────────────────────────────────────┘

流式折中模式（CC_LANGFUSE_STREAMING=true，Watcher 默认）：

┌──────────────────────────────────────────────────────────────┐
│  read_new_jsonl() → build_turns(include_pending=True)        │
│  仅处理 all_turns[turn_count]（当前进行中的 Turn）            │
└────────────────────────────┬─────────────────────────────────┘
                             ▼
┌──────────────────────────────────────────────────────────────┐
│  process_turn_streaming()                                    │
│  · user 消息 → 立即建 trace                                   │
│  · tool_use  → 立即建 tool span                               │
│  · Agent span 打开期间 → 增量读 subagent jsonl 并嵌套上报      │
│  · end_turn  → 收尾 generation + usage，推进 turn_count       │
└────────────────────────────┬─────────────────────────────────┘
                             ▼
┌──────────────────────────────────────────────────────────────┐
│  所有 observation 必须带 trace_id（避免 SDK resume 建空 trace）│
│  flush() 成功 → 推进 offset；end_turn 后推进 turn_count        │
└──────────────────────────────────────────────────────────────┘

4.1 Langfuse 层级结构

Trace: Claude Code - Turn N
├── Generation: Claude Response 1     [start_time/end_time + usage]
├── Tool: Agent
│   ├── Subagent[Explore] Turn 1
│   │   ├── Generation: ...
│   │   └── Tool: ...
│   └── ...
└── Generation: Claude Response 2

4.2 子代理匹配与流式嵌套

主 Turn 遇到 tool_use(name ∈ {Task, Agent}) 时：

扫描 agent-*.jsonl + .meta.json（不把 subagent jsonl 当作独立根 session 跟踪）
按 subagent_type == agentType + 时间戳就近打分
批量模式：tool_result 到达后全量解析子 jsonl，嵌套进 Tool: Agent span
流式模式：Agent span 打开期间，_streaming_sync_subagent_open_tools() 对匹配的 agent-*.jsonl 做增量 read_new_jsonl，子 Turn 的 generation / tool span 实时出现在 Tool: Agent 下

4.3 可靠提交

turn_count 仅在 langfuse.flush() 成功后推进
失败时保留 offset，下次 at-least-once 重试（可能重复，优于静默丢失）

4.4 SDK 版本要求（重要）

本方案使用 Langfuse Python SDK v2.x 有状态 API：trace() / span() / generation()。

bash

pip install -r requirements.txt
# 等价于: pip install 'langfuse>=2.0.0,<3.0.0'

若安装了 langfuse 3.x / 4.x，会报错：

'Langfuse' object has no attribute 'trace'

启动时会检测 SDK 并给出提示。langfuse_hook.py（v3 OTel API）与本目录方案 不兼容，请勿混用。

五、方案 A：Hook

5.1 安装

bash

pip install -r requirements.txt
cp hook/langfuse_hook_all_agents.py ~/.claude/hooks/
chmod +x ~/.claude/hooks/langfuse_hook_all_agents.py

5.2 环境变量

变量	必填	说明
`TRACE_TO_LANGFUSE`	✅	必须为 `true`
`CC_LANGFUSE_PUBLIC_KEY` / `LANGFUSE_PUBLIC_KEY`	✅	公钥
`CC_LANGFUSE_SECRET_KEY` / `LANGFUSE_SECRET_KEY`	✅	私钥
`CC_LANGFUSE_BASE_URL`	❌	默认 `https://cloud.langfuse.com`
`CC_LANGFUSE_INCLUDE_SUBAGENTS`	❌	默认 `true`
`CC_LANGFUSE_REDACT`	❌	默认 `true`（密钥脱敏）
`CC_LANGFUSE_DEBUG`	❌	详细日志

5.3 注册 Hook

~/.claude/settings.json：

json

{
  "hooks": {
    "Stop": [
      {
        "matcher": "*",
        "hooks": [
          {
            "type": "command",
            "command": "python3 ~/.claude/hooks/langfuse_hook_all_agents.py"
          }
        ]
      }
    ]
  }
}

仅挂 Stop，不要挂 SubagentStop（子代理由主 Hook 内联处理，避免扁平重复 trace）。

5.4 状态文件

~/.claude/state/langfuse_hook_all_agents_state.json（offset / buffer / turn_count）

日志：~/.claude/state/langfuse_hook.log

六、方案 B：Watcher（实时轮询）

6.1 原理

独立守护进程按 poll_interval（默认 1s）轮询，对每个已跟踪的主 session jsonl 独立处理：

rescan()：发现目录下新增的 UUID 主 session jsonl，加入跟踪列表
read_new_jsonl(offset)：按文件 offset 读增量行（支持半行 buffer）
build_turns() 重建 Turn 列表
按模式上报：
- 流式（默认）：处理 all_turns[turn_count]，user → 建 trace，tool → span，end_turn → 收尾并推进 turn_count
- 批量：仅上报 completed_turns() 中 turn_count 之后的已完成 Turn
flush 成功后才推进 state；失败保留 offset，下次 at-least-once 重试

tool_use + tool_result 之后模型仍可能继续输出最终回答。批量模式必须等 end_turn；流式模式在 Turn 进行中即可上报中间 observation，但 turn_count 仍只在 end_turn 后推进。

6.2 启动方式

bash

cd watcher

# 推荐：常驻监听（已配置凭证 + idle_timeout=0 + 流式默认开启）
./run_watcher.sh

# 监听单个 session（开发/调试）
python3 claude_langfuse_watcher.py \
  --transcript ~/.trpc-claudecode/projects/.../abc-session.jsonl

# 监听整个 projects 目录
python3 claude_langfuse_watcher.py \
  --projects-root ~/.trpc-claudecode/projects \
  --idle-timeout 0

# 后台
nohup ./run_watcher.sh &

空目录启动：--projects-root / --watch-dir 模式下，启动时没有任何 jsonl 也会继续运行，按 poll_interval 定期 rescan 发现新 session。适用于「先启 watcher、后开 Claude Code 对话」的 daemon 场景。--once 仍要求启动时已有文件。

--transcript 等待：指定的 jsonl 尚未创建时，watcher 会等待该文件出现后再跟踪（不会立即退出）。

修改代码后需重启 watcher 才能加载最新逻辑。

6.3 Watcher 专用环境变量

变量	默认	说明
`CC_LANGFUSE_WATCH_POLL_S`	`1.0`	轮询间隔（秒）
`CC_LANGFUSE_WATCH_IDLE_S`	`300`	文件空闲多久停止跟踪（`run_watcher.sh` 覆盖为 `0`）
`CC_LANGFUSE_STREAMING`	`true`（Watcher）	流式折中上报；`false` 回退批量模式
`PROJECTS_ROOT`	`~/.trpc-claudecode/projects`	`run_watcher.sh` 监听根目录

6.4 状态与日志

State：~/.claude/state/langfuse_watcher_state.json（与 Hook state 隔离）
上报日志：~/.claude/state/langfuse_hook.log（复用 hook 的 info()）
进程日志：~/.claude/state/langfuse_watcher.log（启停信号等）

6.5 流式上报（默认 `CC_LANGFUSE_STREAMING=true`）

事件	上报时机
user 消息写入	立即创建 trace（Langfuse 可见）
`tool_use` 行	立即创建 tool span
`tool_result` 到达	结束 tool span
`Agent`/`Task` span 打开期间	增量读 `subagents/agent-*.jsonl`，嵌套上报子 Turn
assistant `end_turn`	收尾 generation + usage，推进 `turn_count`

关闭流式（回退旧行为，仅 end_turn 一次性上报）：CC_LANGFUSE_STREAMING=false。

阶段	延迟来源
jsonl 追加 → watcher 感知	≤ `poll_interval`（默认 1s）
Langfuse UI 可见	额外 ingestion 延迟（数秒～数十秒）

6.6 主 session 文件过滤（`is_main_session_jsonl`）

Watcher 不会扫描目录下所有 jsonl，只跟踪「主 session」：

文件	是否跟踪	说明
`<UUID>.jsonl`	✅	主 session，文件名 stem 须匹配 UUID 格式
`agent-*.jsonl`	❌	子代理 transcript，由主 session 内联处理
`subagents/` 目录下文件	❌	同上
非 UUID 文件名	❌	跳过（`--transcript` 强制指定时例外，仍会跟踪）

子代理 jsonl 从不作为独立根 trace 上报，避免与主 session 重复。

6.7 多 jsonl / 多 session / 多进程

常见疑问：目录里有很多 jsonl（旧的 + 正在写的），Watcher 怎么知道上报哪个？

答案：不是「选一个」，而是「全部纳入跟踪，谁有新增量谁就上报」。

projects/-data-workspace-demo-project/
├── d6671986-....jsonl   ← 旧 session，仍在跟踪列表
├── 93ba7d55-....jsonl   ← 旧 session，仍在跟踪列表
└── a0e702c2-....jsonl   ← 当前正在写的 session

每轮 poll 对 _tracked 中每个文件分别执行 process_file()：

文件状态	Watcher 行为
无新行写入	`read_new_jsonl` 为空 → 基本 no-op，不上报
有增量写入	解析新内容 → 流式更新 Langfuse
所有 Turn 已上报（`turn_count` 已追上）	继续跟踪，但不再上报
首次发现已写完的旧文件	从 Turn 1 起逐轮补报（每 poll 处理当前 Turn）

不会按 mtime「只选最新文件」；区分新旧靠的是 per-file 的 offset + turn_count，不是全局「当前活跃 session」选择器。

常见疑问：多个 Claude 进程写多个 jsonl，能分别上报 trace 吗？

可以。 每个主 session jsonl 完全独立：

Claude 进程 A  →  session-aaa.jsonl  →  Langfuse sessionId=aaa  →  Turn 1, 2, ...
Claude 进程 B  →  session-bbb.jsonl  →  Langfuse sessionId=bbb  →  Turn 1, 2, ...

一文件一 session：初始 session_id = 文件名 stem（UUID）；读到 jsonl 内 sessionId 字段后会校正
一文件一 state：state_key = SHA256(session_id + 文件绝对路径)，互不干扰
Langfuse UI：按 sessionId 过滤即可看到各会话各自的 trace 列表

场景	说明
两个进程各开新 session	两个 UUID jsonl，各自独立 trace 序列
同一 session 不应有两个主 jsonl	正常只有一个 `<sessionId>.jsonl`
子 Agent	嵌套在父 session 的 `Tool: Agent` 下，不单独占根 trace

6.8 状态隔离与 idle 行为

每个跟踪文件在 langfuse_watcher_state.json 中独立存储：

json

{
  "<state_key>": {
    "offset": 12345,
    "buffer": "",
    "turn_count": 3,
    "streaming": { "trace_id": "...", "turn_num": 4, "tools": {...} }
  }
}

offset / buffer：jsonl 增量读取进度
turn_count：已成功上报并 finalize 的 Turn 数
streaming：当前进行中 Turn 的 trace/span 句柄（仅流式模式，end_turn 后清空）

idle_timeout（CC_LANGFUSE_WATCH_IDLE_S，run_watcher.sh 默认 0）：

值	行为
`0`	永不停止跟踪；旧 session 留在列表，无新写入时开销极小
`>0`（如 300）	文件超过 N 秒无新写入 → 从 `_tracked` 移除；下次 rescan 发现时会 `_bootstrap_messages_if_needed()` 从磁盘恢复全量消息

七、测试

7.1 单元测试

bash

python3 -m unittest tests/test_usage_transcript.py

7.2 端到端测试

需配置 Langfuse 凭证。

bash

export CC_LANGFUSE_PUBLIC_KEY=pk-lf-xxx
export CC_LANGFUSE_SECRET_KEY=sk-lf-xxx
export CC_LANGFUSE_BASE_URL=https://your-langfuse-host

./hook/run_e2e_langfuse_test.sh            # Hook：完整 transcript 一次上报
./watcher/run_e2e_langfuse_watcher_test.sh  # Watcher：--once 批量回放

八、与原版 `langfuse_hook.py` 的差异

维度	原版（hooks/langfuse_hook.py）	本方案
Langfuse SDK	v3 `start_as_current_observation`	v2 `trace/span/generation`
子代理	黑盒	嵌套上报
时间戳	常显示 0.00s	transcript 真实时间
Token	缺失	`usage_details`
失败重试	易丢数据	flush 失败不推进 state
上报方式	仅 Hook	Hook + Watcher

九、已知限制

子代理 jsonl 内不再递归 discover 更深层 Task。
子代理匹配为启发式（type + 时间戳），meta 缺失时可能误匹配。
at-least-once 语义：极端情况下可能重复上报同一 Turn。
仅支持 Langfuse SDK 2.x；4.x 需降级或重写 emit 层。
Watcher 无「只跟踪最新 session」选项；历史 jsonl 会一直留在跟踪列表（idle_timeout=0 时）。
首次发现大量历史 session 时，补报按 Turn 逐轮进行，非瞬时完成。

十、故障排查

现象	排查
`'Langfuse' object has no attribute 'trace'`	`pip install 'langfuse>=2,<3'`
日志有 `emitted` 但 UI 无 trace	ingestion 延迟；查 API `GET /api/public/traces?sessionId=...`
只有 `signal 15, stopping`	看 `langfuse_hook.log`，`emitted` 写在该文件
Watcher 一直 `turn_count=0`	确认 turn 是否已到 `end_turn`；开 `CC_LANGFUSE_DEBUG=true`
重复 trace	Hook 与 Watcher 同时启用，或 at-least-once 重试
Langfuse 里大量空白 trace（仅 name、无内容）	流式 resume 时未传 `trace_id` 的历史脏数据；升级代码后重启 watcher；可手动清理旧 trace
`Tool: Agent` span 为空	确认 `CC_LANGFUSE_STREAMING=true` 且子 agent jsonl 路径可被 `discover_subagent_files` 发现
旧 session 也被轮询	正常行为；无新写入时不上报；若需释放可设 `--idle-timeout 300`

十一、常见问题（FAQ）

Q：Watcher 会重复上报已经上报过的 Turn 吗？

一般不会。turn_count 仅在 end_turn 且 flush() 成功后推进；流式模式下进行中的 Turn 通过 streaming 状态 resume 同一 trace，不会每 poll 新建 trace。极端网络失败时可能 at-least-once 重试。

Q：为什么不用子 agent jsonl 的文件 mtime 来判断「当前活跃 session」？

主 session 的活跃度应看主 jsonl 是否有新行。子 agent jsonl 只在 Agent/Task tool 执行期间写入，且由主 session 驱动匹配，不适合作为「选哪个主 session」的依据。

Q：Hook 和 Watcher 的 state 文件能共用吗？

不能。Hook 用 langfuse_hook_all_agents_state.json，Watcher 用 langfuse_watcher_state.json，避免互相覆盖 offset / turn_count。

Q：Langfuse 里同一 sessionId 下 Turn 编号怎么对应？

每个完成的 Turn 对应一条根 trace，命名形如 Claude Code - Turn N，N = turn_count + 1（从 1 递增）。

十二、参考

历史详细文档：docs/langfuse_hook_all_agents_README.md
原版 Hook 参考：../hooks/langfuse_hook.py
通用 Hook 机制：../hooks/hook.md

docs

hook

watcher

Claude Code Transcript → Langfuse 方案实现说明 ​

一、方案概览 ​

二、目录结构 ​

三、Transcript 简介 ​

3.1 存储位置 ​

3.2 每行事件类型 ​

3.3 Turn 划分规则 ​

四、核心架构（共享逻辑） ​

4.0 数据流总览 ​

4.1 Langfuse 层级结构 ​

4.2 子代理匹配与流式嵌套 ​

4.3 可靠提交 ​

4.4 SDK 版本要求（重要） ​

五、方案 A：Hook ​

5.1 安装 ​

5.2 环境变量 ​

5.3 注册 Hook ​

5.4 状态文件 ​

六、方案 B：Watcher（实时轮询） ​

6.1 原理 ​

6.2 启动方式 ​

6.3 Watcher 专用环境变量 ​

6.4 状态与日志 ​

6.5 流式上报（默认 CC_LANGFUSE_STREAMING=true） ​

6.6 主 session 文件过滤（is_main_session_jsonl） ​

6.7 多 jsonl / 多 session / 多进程 ​

6.8 状态隔离与 idle 行为 ​

七、测试 ​

7.1 单元测试 ​

7.2 端到端测试 ​

八、与原版 langfuse_hook.py 的差异 ​

九、已知限制 ​

十、故障排查 ​

十一、常见问题（FAQ） ​

十二、参考 ​

Claude Code Transcript → Langfuse 方案实现说明

一、方案概览

二、目录结构

三、Transcript 简介

3.1 存储位置

3.2 每行事件类型

3.3 Turn 划分规则

四、核心架构（共享逻辑）

4.0 数据流总览

4.1 Langfuse 层级结构

4.2 子代理匹配与流式嵌套

4.3 可靠提交

4.4 SDK 版本要求（重要）

五、方案 A：Hook

5.1 安装

5.2 环境变量

5.3 注册 Hook

5.4 状态文件

六、方案 B：Watcher（实时轮询）

6.1 原理

6.2 启动方式

6.3 Watcher 专用环境变量

6.4 状态与日志

6.5 流式上报（默认 `CC_LANGFUSE_STREAMING=true`）

6.6 主 session 文件过滤（`is_main_session_jsonl`）

6.7 多 jsonl / 多 session / 多进程

6.8 状态隔离与 idle 行为

七、测试

7.1 单元测试

7.2 端到端测试

八、与原版 `langfuse_hook.py` 的差异

九、已知限制

十、故障排查

十一、常见问题（FAQ）

十二、参考