Claude Code Transcript → Langfuse 方案实现说明

基于 transcript_demo/ 样例数据与 langfuse_hook_all_agents.py 改进版 Hook 编写。
配套文件：
样例 transcript：transcript_demo/3e594db1-f0f2-4b0a-bfbc-aa7bd9a8eb38.jsonl
改进版 Hook：langfuse_hook_all_agents.py
原版 Hook：langfuse_hook.py

一、Transcript 简介

1.1 什么是 Transcript

Transcript 是 Claude Code 在本地持久化的会话事件日志，以 JSONL（每行一个 JSON 对象）形式追加写入。Hook 脚本不直接监听 API 调用，而是在 Stop 等事件触发时，增量读取 transcript 文件，解析出「用户轮次 → 模型回复 → 工具调用 → 工具结果」的完整链路，再上报到 Langfuse。

典型存储路径（因版本/二开 fork 略有差异）：

~/.claude/projects/<项目哈希>/
├── <sessionId>.jsonl                          # 主会话
└── <sessionId>/subagents/                     # trpc-claudecode 二开布局
    ├── agent-<agentId>.jsonl                  # 子代理独立 transcript
    └── agent-<agentId>.meta.json              # 子代理元数据

transcript_demo/ 即一次真实会话的脱敏快照：主会话 3e594db1-...jsonl + 两个子代理 agent-a3368c...（编译检查）与 agent-ac14079...（代码生成）。

1.2 JSONL 每行包含哪些内容

每行是一个独立事件，通过 parentUuid 形成链式父子关系。常见 type 如下：

type	含义	典型用途
`queue-operation`	消息队列操作	`enqueue` / `dequeue`，记录用户原始输入
`user`	用户侧消息	真人输入，或 `tool_result` 回灌
`assistant`	助手侧消息	模型文本 / `tool_use` 工具调用
`attachment`	附件/上下文注入	如 skill 列表、文件引用
`last-prompt`	最后一轮 prompt 快照	会话收尾时的 prompt 状态

通用顶层字段（几乎所有对话行都有）：

字段	说明
`uuid`	本行唯一 ID
`parentUuid`	上一事件 ID，构成事件链
`timestamp`	ISO8601 UTC 时间，如 `2026-05-11T09:01:52.077Z`
`sessionId`	所属会话 ID
`type`	事件类型
`isSidechain`	是否子代理侧链（主 jsonl 中为 `false`；子 agent jsonl 为 `true`）
`cwd` / `gitBranch` / `version`	工作目录、Git 分支、Claude Code 版本
`entrypoint` / `userType`	入口与用户类型

对话内容字段（user / assistant 行）：

json

{
  "type": "assistant",
  "message": {
    "id": "msg_15e9699e419a2552bfd8bc82",
    "role": "assistant",
    "model": "glm-5.1-w4afp8",
    "content": [
      { "type": "text", "text": "我先读取原型文件..." },
      { "type": "tool_use", "id": "call_xxx", "name": "Bash", "input": { "command": "..." } }
    ],
    "usage": {
      "input_tokens": 0,
      "output_tokens": 237,
      "cache_read_input_tokens": 0,
      "cache_creation_input_tokens": 0
    },
    "stop_reason": "tool_use"
  }
}

工具结果行（type=user，content 为 tool_result 块）：

json

{
  "type": "user",
  "message": {
    "role": "user",
    "content": [{
      "type": "tool_result",
      "tool_use_id": "call_xxx",
      "content": "...",
      "is_error": false
    }]
  },
  "sourceToolAssistantUUID": "...",
  "toolUseResult": { "stdout": "...", "stderr": "", "interrupted": false }
}

trpc-claudecode fork 的特殊写法：同一 API 响应可能被拆成多行 jsonl，共享同一个 message.id——一行放 text，每行 tool_use 各占一行。Hook 必须按 message.id 分组而非简单去重，否则会丢失文本或工具调用。

子代理相关：

主 jsonl 中不含 isSidechain:true 的子代理内部步骤；子代理调用体现为主会话里的 tool_use(name="Agent"|"Task") + 一条摘要型 tool_result。
子代理完整对话在独立文件 agent-<id>.jsonl，首行通常带 agentId；trpc 版本另有 agent-<id>.meta.json：

json

{ "agentType": "kuikly-codefix", "description": "Kuikly项目编译检查" }

1.3 transcript_demo 样例会话结构

主会话 3e594db1-...
├── Turn 1: 用户要求生成 Kuikly 项目
│   ├── Generation(s): 读原型、分析
│   ├── Tool: Agent → 匹配 agent-ac14079... (Kuikly-codegen)
│   └── 最终回复: 项目已生成
├── Turn 2: 主 agent 发起编译检查
│   ├── Tool: Agent → 匹配 agent-a3368c... (kuikly-codefix)
│   └── 最终回复: BUILD SUCCESSFUL 汇报
└── Turn 3: 汇总回复用户

二、官方 `langfuse_hook.py` 存在的问题（简要）

原版 Hook 实现了「增量读主 jsonl → 组装 Turn → 上报 Langfuse」的基本链路，但在生产可观测场景存在以下核心缺陷：

#	问题	影响
1	Subagent 完全黑盒	主 jsonl 里的 `Tool: Agent/Task` 只上报 input/output 摘要，子代理内部的 Generation、Bash/Read 等步骤不可见；token 成本无法分摊到子任务
2	Latency 恒为 0.00s	未把 transcript 的 `timestamp` 回填到 Langfuse 的 `start_time/end_time`，性能分析失效
3	Token 用量缺失	未读取 `message.usage`，Langfuse 成本看板无数据
4	无可靠交付保证	`turn_count` 在 emit 后即累加，`flush()` 失败被静默吞掉 → 网络抖动时静默丢数据
5	文件截断/旋转无防护	offset 指向旧位置后永久卡死，该 session 不再上报
6	无数据脱敏	user/tool 内容原样上云，存在密钥/PII 泄露风险
7	Turn 组装与 fork 不兼容	按 `message.id` 去重 assistant 行，在 trpc 拆行写法下会丢失 text 或 tool_use
8	SDK API 选型	使用 Langfuse v3 的 `start_as_current_observation`；与部分自托管 v3 前端兼容性存疑

若额外配置 SubagentStop，子代理 trace 会与主 trace 扁平并列，无法形成父子嵌套，且 trace 命名混淆。

三、`langfuse_hook_all_agents` 架构设计与实现原理

3.1 设计目标

在保持 100% fail-open（任何异常 return 0，不阻塞 Claude Code）的前提下：

主会话 + 子代理 统一嵌套可观测
时间戳、token、脱敏、可靠提交等生产级补齐
兼容官方 Claude Code 与 trpc-claudecode 两种 on-disk 布局

3.2 整体架构

┌─────────────────────────────────────────────────────────────────┐
│  Claude Code Hook (Stop 事件)                                    │
│  stdin: { sessionId, transcriptPath, ... }                       │
└───────────────────────────┬─────────────────────────────────────┘
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│  main()                                                          │
│  ① 读取 hook payload → session_id + transcript_path              │
│  ② 跳过 agent-*.jsonl 直调（防 SubagentStop 重复）               │
│  ③ FileLock + 加载 state (offset/buffer/turn_count)              │
│  ④ read_new_jsonl() 增量读主 transcript                            │
│  ⑤ build_turns() 组装 Turn 列表                                  │
│  ⑥ discover_subagent_files() 扫描子代理 jsonl                      │
│  ⑦ 对每个 Turn → emit_main_turn()                                │
│  ⑧ flush 成功才 commit state                                     │
└───────────────────────────┬─────────────────────────────────────┘
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│  Langfuse v2 Stateful API                                        │
│  trace → generation(s) + tool span(s) → nested subagent spans    │
└─────────────────────────────────────────────────────────────────┘

Langfuse SDK 选型：使用 Langfuse Python SDK v2.x（langfuse>=2,<3）的有状态 API（trace/span/generation），显式传入 start_time/end_time，兼容自托管 v3 Web UI；避免 v3/v4 OTel SDK 在旧前端上的 observation 类型兼容问题。

3.3 核心模块说明

（1）增量读取 `read_new_jsonl`

维护 { offset, buffer, turn_count } per (session_id, transcript_path) 哈希键
半行缓冲：未读完的 JSON 行留在 buffer，下次追加
截断检测：file.size < offset 时重置 offset/buffer，防止旋转后永久卡死

（2）Turn 组装 `build_turns`

划分规则：

user（非 tool_result、非 synthetic）→ 新 Turn 开始
assistant                           → 追加到当前 Turn
user + tool_result                  → 按 tool_use_id 归档结果，不新开 Turn

特殊处理：

Synthetic user：Skill 注入的 "Base directory for this skill:" 等 harness 消息不拆 Turn
保留所有 assistant 行：trpc fork 拆行时不去重，避免丢内容
为每个 tool_use_id 记录 timestamp、input、name，供后续子代理匹配与时间戳回填

（3）子代理发现 `discover_subagent_files`

按优先级扫描目录：

1. <主jsonl父目录>/<sessionId>/subagents/
2. <主jsonl父目录>/<主jsonl stem>/subagents/    ← transcript_demo 使用此布局
3. <主jsonl父目录>/                              ← 官方 Claude Code 同级 agent-*.jsonl

对每个 agent-<id>.jsonl 提取：

agent_id、agent_type（meta.json 或首行）
description（meta.json）
first_ts（首行 timestamp，用于匹配）

（4）主→子匹配 `match_subagent`

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

score = 0
若 subagent_type ≠ agent_type:     score += 1000
score += |主 tool_use 时间 − 子首行时间|（秒）
若时间差 > SUBAGENT_TIME_WINDOW_S: score += 500

取 score 最低且 < 1500 的未使用子文件

transcript_demo 中两次 Agent 调用可分别匹配到 agent-ac14079...（Kuikly-codegen）与 agent-a3368c...（kuikly-codefix）。

（5）Langfuse 上报 `_emit_turn` / `emit_main_turn`

层级结构（改进版在 Langfuse 上的目标形态）：

Trace: Claude Code - Turn N                    [session_id 聚合]
├── Generation: Claude Response 1              [真实 start/end + usage]
├── Tool: Bash                                 [tool_use ts → tool_result ts]
├── Generation: Claude Response 2
├── Tool: Agent                                [主轮子代理调用]
│   ├── Subagent[kuikly-codefix] Turn 1
│   │   ├── Generation: Claude Response
│   │   ├── Tool: TodoWrite
│   │   ├── Tool: Glob
│   │   └── Tool: Bash
│   └── Subagent[...] Turn 2
│       └── ...
└── (trace output = 最后一轮有文本的 assistant 回复)

关键实现点：

能力	实现方式
多 Generation	`_group_assistant_calls()` 按 `message.id` 分组，一轮模型调用对应一个 generation
时间戳	`parse_ts(msg["timestamp"])` → 各 observation 的 `start_time/end_time`
Token	`get_usage()` → `usage_details={input, output, cache_*}`
嵌套	v2 显式 parenting：`tool_span = container.span(...)` → `_emit_turn(container=tool_span, ...)`
子代理	`_emit_subagent_inline()` 全量读子 jsonl → 每个 sub-turn 建 span → 递归 `_emit_turn`（不再向下 discover）
脱敏/截断	`redact()` + `truncate_text()`，metadata 记录 `truncated/sha256`
防重复	若 hook 直接指向 `agent-*.jsonl`，`main()` 开头 skip

（6）可靠提交

python

emitted_ok = 0
for t in turns:
    try:
        emit_main_turn(...)
        emitted_ok += 1
    except:
        break                    # 保持顺序，失败 turn 下次重试

flush_ok = langfuse.flush() 成功?

if flush_ok and emitted_ok == len(turns):
    turn_count += emitted_ok
    持久化 offset/buffer
else:
    不推进 state                  # 下次原样重读、重发

权衡：极端情况下可能 at-least-once 重复上报，但优于静默丢失。

3.4 数据流（以 transcript_demo Turn 2 为例）

3.5 状态与配置

State 文件：~/.claude/state/langfuse_hook_all_agents_state.json（与原版 langfuse_state.json 隔离，可安全回滚）

关键环境变量：

变量	默认	作用
`TRACE_TO_LANGFUSE`	—	必须为 `true`
`CC_LANGFUSE_INCLUDE_SUBAGENTS`	`true`	关闭则不上报子代理内部
`CC_LANGFUSE_SUBAGENT_TIME_WINDOW_S`	`10`	主→子时间匹配窗口
`CC_LANGFUSE_REDACT`	`true`	密钥/密码正则脱敏
`CC_LANGFUSE_MAX_CHARS`	`20000`	单字段截断阈值

Hook 注册建议：仅挂 Stop，不要再挂 SubagentStop；子代理由主 Hook 内联处理以保证嵌套关系正确。

3.6 与原版对比总结

维度	原版	langfuse_hook_all_agents
子代理内部步骤	黑盒	嵌套在 `Tool: Agent` 下
时间戳	0.00s	transcript 真实时间
Token	无	usage_details
交付语义	at-most-once（易丢）	at-least-once（可重试）
文件截断	卡死	自动 reset
脱敏	无	内置正则
trpc 拆行 transcript	可能丢数据	按 id 分组 + 保留全行
Langfuse SDK	v3 context API	v2 stateful API

四、已知限制

子代理不再递归：子 agent jsonl 内若再调 Task，当前不再向下 discover（Claude Code 现状极少出现）。
State 单文件：多 session 高并发时仍有锁等待；极致场景可拆分为 per-session state 文件。
匹配启发式：依赖 subagent_type + 时间戳；meta 缺失或并发多同类型 agent 时可能误匹配（可通过缩小 SUBAGENT_TIME_WINDOW_S 缓解）。
重复上报边界：emit 成功但 flush 失败时，下次可能重复同一 Turn（Langfuse 侧需按 session + turn_number 去重或接受重复）。

五、参考

样例数据目录：transcript_demo/
改进版实现：langfuse_hook_all_agents.py
使用说明：langfuse_hook_all_agents_README.md
问题深度分析：langfuse_hook_问题与方案对比.md

code index

hooks

langfuse trace

docs

hook

watcher

Claude Code Transcript → Langfuse 方案实现说明

一、Transcript 简介

1.1 什么是 Transcript

1.2 JSONL 每行包含哪些内容

1.3 transcript_demo 样例会话结构

二、官方 `langfuse_hook.py` 存在的问题（简要）

三、`langfuse_hook_all_agents` 架构设计与实现原理

3.1 设计目标

3.2 整体架构

3.3 核心模块说明

（1）增量读取 `read_new_jsonl`

（2）Turn 组装 `build_turns`

（3）子代理发现 `discover_subagent_files`

（4）主→子匹配 `match_subagent`

（5）Langfuse 上报 `_emit_turn` / `emit_main_turn`

（6）可靠提交

3.4 数据流（以 transcript_demo Turn 2 为例）

3.5 状态与配置

3.6 与原版对比总结

四、已知限制

五、参考

docs

hook

watcher

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

一、Transcript 简介 ​

1.1 什么是 Transcript ​

1.2 JSONL 每行包含哪些内容 ​

1.3 transcript_demo 样例会话结构 ​

二、官方 langfuse_hook.py 存在的问题（简要） ​

三、langfuse_hook_all_agents 架构设计与实现原理 ​

3.1 设计目标 ​

3.2 整体架构 ​

3.3 核心模块说明 ​

（1）增量读取 read_new_jsonl ​

（2）Turn 组装 build_turns ​

（3）子代理发现 discover_subagent_files ​

（4）主→子匹配 match_subagent ​

（5）Langfuse 上报 _emit_turn / emit_main_turn ​

（6）可靠提交 ​

3.4 数据流（以 transcript_demo Turn 2 为例） ​

3.5 状态与配置 ​

3.6 与原版对比总结 ​

四、已知限制 ​

五、参考 ​

Claude Code Transcript → Langfuse 方案实现说明

一、Transcript 简介

1.1 什么是 Transcript

1.2 JSONL 每行包含哪些内容

1.3 transcript_demo 样例会话结构

二、官方 `langfuse_hook.py` 存在的问题（简要）

三、`langfuse_hook_all_agents` 架构设计与实现原理

3.1 设计目标

3.2 整体架构

3.3 核心模块说明

（1）增量读取 `read_new_jsonl`

（2）Turn 组装 `build_turns`

（3）子代理发现 `discover_subagent_files`

（4）主→子匹配 `match_subagent`

（5）Langfuse 上报 `_emit_turn` / `emit_main_turn`

（6）可靠提交

3.4 数据流（以 transcript_demo Turn 2 为例）

3.5 状态与配置

3.6 与原版对比总结

四、已知限制

五、参考