learnNote

主题

langfuse_hook_all_agents.py — 改进版使用说明 ​

这是 langfuse_hook.py 的进化版本。 修复了原版关于 subagent 不可观测、Latency 显示为 0、丢数据 等问题， 在保持完全 fail-open 的前提下做了 6 大改进。

配套文档：

  • ./langfuse_hook_精读.md（小白入门，原版逻辑详解）
  • ./langfuse_hook_问题与方案对比.md（问题清单与改进建议）

0. 一句话简介 ​

给主 jsonl 上报的同时，自动找到对应的 agent-*.jsonl 子代理 transcript，把它的内部步骤嵌套成主轮 Tool: Agent 的子节点；同时把每个 observation 的真实时间戳和 token 用量都填上，并加了基础脱敏与可靠提交。

1. 跟原版的差异速览 ​

维度原版 langfuse_hook.py改进版 langfuse_hook_all_agents.py
subagent 内部步骤❌ 完全黑盒✅ 自动发现 + 嵌套上报
兼容子代理目录布局❌ 不处理✅ 同时支持官方 agent-*.jsonl（与主 jsonl 同级）和 trpc 二开 <sessionId>/subagents/agent-*.jsonl
主→子匹配n/a✅ subagent_type == agentType + 时间戳就近
Trace/Tool 时间戳❌ 全部 0.00s✅ 用 transcript 里的 timestamp 回填 start_time/end_time
Token 用量❌ 缺失✅ 从 message.usage 提取并写入 usage_details
失败重试❌ turn_count 总是 +1，丢了就丢了✅ flush 失败时不推进 offset 与 turn_count，下次重试
文件被截断/旋转❌ 永久卡死✅ 检测到 size < offset 自动重置
数据脱敏❌ 裸传✅ 内置正则脱敏（API key/Bearer/password/私钥），可关闭
防止重复上报n/a✅ 若 hook 直接被 SubagentStop 触发到 agent-*.jsonl，自动跳过

2. 安装 ​

bash
pip install langfuse
mkdir -p ~/.claude/state

把 langfuse_hook_all_agents.py 放到任意位置，例如：

bash
cp langfuse_hook_all_agents.py ~/.claude/hooks/langfuse_hook_all_agents.py
chmod +x ~/.claude/hooks/langfuse_hook_all_agents.py

3. 环境变量 ​

变量必填默认说明
TRACE_TO_LANGFUSE✅—必须设为 true，否则脚本立即 return 0
CC_LANGFUSE_PUBLIC_KEY / LANGFUSE_PUBLIC_KEY✅—Langfuse 公钥
CC_LANGFUSE_SECRET_KEY / LANGFUSE_SECRET_KEY✅—Langfuse 私钥
CC_LANGFUSE_BASE_URL / LANGFUSE_BASE_URL❌https://cloud.langfuse.com自建 Langfuse 改这里
CC_LANGFUSE_DEBUG❌false设 true 写详细 debug 日志
CC_LANGFUSE_MAX_CHARS❌20000单字段截断阈值
CC_LANGFUSE_REDACT❌true设 false 关闭脱敏
CC_LANGFUSE_INCLUDE_SUBAGENTS❌true设 false 不展开子代理
CC_LANGFUSE_SUBAGENT_TIME_WINDOW_S❌10主→子匹配的时间窗（秒）

4. 在 Claude Code 中注册 ​

~/.claude/settings.json：

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

⚠️ 不要再单独挂 SubagentStop。 改进版的设计是：主 hook 一并处理子代理，这样能保证嵌套关系正确。 如果你不小心挂了 SubagentStop，脚本检测到 transcript 文件名以 agent- 开头会主动跳过，避免重复上报和扁平 trace。

5. Langfuse 上看到的效果 ​

之前（原版）：

Trace: Claude Code - Turn 2
├─ Claude Response
├─ Tool: Agent      ← 空壳叶子，里面什么都没有
└─ Tool: Agent      ← 空壳叶子
Latency: 0.00s

改进版：

Trace: Claude Code - Turn 2                       ⏱ 4m 12s
├─ Generation: Claude Response                    ⏱ 8s, in=2.1k tokens, out=512
├─ Tool: Agent  (subagent_type=trpc-go-codegen)  ⏱ 2m 34s
│  ├─ Subagent[trpc-go-codegen-by-proto] Turn 1
│  │  ├─ Generation: Claude Response  in=1.2k out=350
│  │  ├─ Tool: Bash  (ls /data/workspace/...)
│  │  └─ Tool: Read  (echo.proto)
│  ├─ Subagent[...] Turn 2
│  │  └─ ...
│  └─ output: "已生成 trpc-go 项目结构 ..."
└─ Tool: Agent  (subagent_type=codefix)           ⏱ 1m 38s
   └─ Subagent[codefix] Turn 1
      └─ ...

时间、token、子代理细节、父子嵌套关系——全都有了。

6. 工作原理（与原版不同的关键点） ​

6.1 子代理目录探测 ​

discover_subagent_files() 按以下顺序尝试：

候选 1（trpc-claudecode 二开）：<主jsonl目录>/<sessionId>/subagents/
候选 2（同上）：               <主jsonl目录>/<主jsonl文件名 stem>/subagents/
候选 3（官方 Claude Code）：    <主jsonl目录>/                   ← agent-*.jsonl 直接和主 jsonl 同级

发现到的每个 agent-<id>.jsonl 都会读取：

  • 同名 agent-<id>.meta.json（trpc 版本独有，含 agentType / description）
  • 文件第一行的 timestamp、agentType

6.2 主 → 子的匹配算法 ​

主 jsonl 中遇到 tool_use(name in {"Task","Agent"}) 时，记下它的：

  • tool_use.input.subagent_type
  • 该消息的 timestamp

匹配规则（match_subagent）：

score = 0
if subagent_type 匹配 agent_type:        score += 0
elif subagent_type 与 agent_type 不同:    score += 1000   (基本排除)
score += abs(主tool_use_ts − 子jsonl首行ts)秒数
if 时间差 > SUBAGENT_TIME_WINDOW_S:       score += 500    (退到次选)

挑选 score 最低且 < 1500 的子文件；找不到就放弃。

实测在真实数据上，两次 Agent 调用都能精确匹配到对应的 agent-*.jsonl：

Turn 1 req_type=trpc-go-codegen-by-proto -> matched ad15ea16142239653 (trpc-go-codegen-by-proto)
Turn 1 req_type=codefix                  -> matched a2f54dbfcf1aa4b7b (codefix)

6.3 嵌套上报 ​

利用 langfuse v3 的 with langfuse.start_as_current_observation(...) 上下文嵌套——任何在该 with 块内创建的 observation 都会自动成为它的子节点。所以：

python
with start_as_current_observation(name="Tool: Agent", ...) as tool_obs:
    # 在这里调用 _emit_subagent_inline -> 这些子 turn 的所有 generation/tool
    # 都会自然挂到 tool_obs 下面
    _emit_subagent_inline(...)
    tool_obs.update(output=主轮收到的tool_result)

6.4 时间戳回填 ​

每个 observation 都尝试从对应消息的 timestamp 字段取值：

Observationstart_timeend_time
Trace（主轮）user_msg tslast assistant ts
Generationfirst assistant tslast assistant ts
Tooltool_use 所在 assistant tstool 输出后 update 时自动收
Subagent Turn子 user_msg ts子 last assistant ts

→ Langfuse 上 Latency 不再是 0.00s。

6.5 可靠提交 ​

emitted_ok = 0
for t in turns:
    try:
        emit_main_turn(...); emitted_ok += 1
    except:
        break          # 出错就停，剩下的下次再来

flush_ok = try langfuse.flush() else False

if flush_ok and emitted_ok == len(turns):
    持久化 (offset, buffer, turn_count + emitted_ok)
else:
    # 不动状态文件，下次原样重读、重发
    pass

副作用：极少数情况下会重复上报（emit 成功但 flush 失败时），换来的是不丢数据——比静默丢失更可接受。

7. 验证方式 ​

执行一遍真实会话后，可以这样自查：

bash
# 1. 看 hook 自身日志
tail -f ~/.claude/state/langfuse_hook.log

# 2. 看 state 文件
cat ~/.claude/state/langfuse_hook_all_agents_state.json | python3 -m json.tool

# 3. 在 Langfuse Web 上：
#    Sessions → 点对应 sessionId → 应能看到嵌套结构和有时长的各 observation

8. 已知限制 / 待办 ​

  1. 极端嵌套：如果子代理内部又调用了子代理，本实现不会再向下递归（实测 Claude Code 当前版本没有这种用法；要支持也只是 _emit_subagent_inline 里也做 discover）。
  2. 跨进程并发：state 文件仍是单文件全量重写。多 session 高并发时锁等待依然存在。极致优化可拆成 state/sessions/<key>.json 分片。
  3. OTel 标准化：当前还是直接调 langfuse SDK；如果想接到 OpenTelemetry collector 上，把 _emit_turn_into_current_context 内部换成 OTel span 即可，整体结构不变。
  4. subagent 第一行 user 不再 dedup：build_turns 用 message.id 去重 assistant 已可以；但 subagent 第一行没有 id 的 user 不会引发任何问题（因为没有 assistant_latest 它就不会被 flush 成轮次）。

9. 回滚 ​

如果改进版出问题需要紧急回滚：

bash
# settings.json 把 command 改回原版
"command": "python3 ~/.claude/hooks/langfuse_hook.py"

两份脚本的 state 文件名不同：

  • 原版：~/.claude/state/langfuse_state.json
  • 改进版：~/.claude/state/langfuse_hook_all_agents_state.json

互不影响，切回去也不会重发数据。

10. 一句话总结 ​

改进版 = 原版 + 子代理嵌套 + 时间戳 + token + 脱敏 + 可靠提交。 默认替换原版即可获得完整的可观测性，无需额外配置 SubagentStop。