Langfuse Hook 从零到一精读

目标读者：从未写过 Hook、也没用过 Langfuse 的同学。文档会用大量生活化类比，把每一段晦涩的代码都讲到"哦，原来是这么回事"的程度。配套源码：./langfuse_hook.py

0. 先把背景讲清楚（非常重要）

0.1 这段代码到底想干什么？

一句话：

Claude Code 每说一句话、每调用一次工具，都偷偷把这次对话的"录像片段"发送到一个叫 Langfuse 的"监控大屏"上，让你之后能回放、分析、找问题。

类比：

角色	类比
Claude Code	一位正在帮你写代码的"程序员同事"
Transcript 文件	这位同事的"工作日志本"，他每说一句话就在本子上写一行
Hook 脚本（本文件）	一台装在他桌子旁边的"摄像头"
Langfuse	远程的"监控中心大屏"
你	老板，想随时回看同事都干了啥

摄像头不能影响同事工作 → 所以 Hook 必须永远不报错、永远不卡住（专业说法叫 fail-open，下文反复出现）。

0.2 什么是 Hook？

Hook = 钩子。可以理解为"事件触发的回调"。

生活类比：

你家门口装了门铃感应器（Hook 配置）。
每次有人按门铃（事件触发），感应器就会自动打开摄像头录 5 秒（Hook 脚本被执行）。
你不需要每次手动开摄像头。

Claude Code 里的 Hook 也一样：你只要在配置里写一句"每次对话结束后跑一下 langfuse_hook.py"，剩下的就交给 Claude Code 自己触发。

0.3 什么是 Langfuse？

类比：你给程序员同事配的"行车记录仪 + 后台分析系统"。

它能：

记录每一轮对话的输入输出
记录用了哪些工具、工具输入输出是什么
统计 token 用量、耗时
把同一个 session 里的多轮串成一条时间轴
以树状结构展示：trace（一次会话）→ generation（一次大模型调用）→ tool（一次工具调用）

记住这个树形结构，下文反复用到。

0.4 什么是 Transcript？

Claude Code 在你对话过程中，会把所有消息一条一条追加写到一个文件里，每行一个 JSON。这种格式叫 JSONL（JSON Lines）。

类比：

工作日志本.jsonl：

第1行：{"type":"user","message":{"content":"帮我写个登录页"}}
第2行：{"type":"assistant","message":{"content":"好的，我先看看项目结构"}}
第3行：{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Read","id":"abc"}]}}
第4行：{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"abc","content":"package.json内容"}]}}
第5行：{"type":"assistant","message":{"content":"我看到这是个 Vite 项目..."}}
...

注意一个反直觉的点：工具的执行结果（tool_result）是以 type:"user" 写进去的。这是因为在大模型协议里，"工具反馈"会以"用户提供的信息"的身份再喂给模型。

1. 整体运行流程（先看森林，再看树）

Claude Code 触发事件（比如一轮对话结束）
        │
        ▼
拉起 python langfuse_hook.py（短生命周期进程）
        │
        ▼
1. 读 stdin 拿到 payload（包含 session_id、transcript 路径）
2. 打开 ~/.claude/state/langfuse_state.json 看上次读到哪儿了
3. 从 transcript 文件的"上次位置"继续读新增内容
4. 把零散的 JSON 行组装成一个个"轮次（Turn）"
5. 把每个 Turn 上报到 Langfuse
6. 把"我读到哪儿了 + 上报了多少轮"写回 state.json
        │
        ▼
进程退出，等下次再被拉起

核心思想：增量、断点续传、绝不阻塞。

类比：你是一个抄日记的助理，每次老板叫你来一次，你就：

拿出小本本，看上次抄到第几页第几行
翻到那里，继续抄新写的内容
抄完发给云端
在小本本上更新"我抄到哪里了"
下班

2. 关键概念逐个攻破

2.1 fail-open 原则

代码里到处是这种写法：

python

try:
    from langfuse import Langfuse, propagate_attributes
except Exception:
    sys.exit(0)

意思是：Langfuse 包没装？不管，直接退出，绝不报错给 Claude Code。

类比：保安发现摄像头坏了，他不会冲进会议室打断老板，而是默默回去修。

整个脚本贯彻这个原则：

import 失败 → 退出 0
读 stdin 失败 → 返回空
读文件失败 → 返回空
上报失败 → 跳过这一轮，继续下一轮
任何意外 → 走 finally，安全退出

2.2 短生命周期 + 状态文件

Hook 脚本不是常驻进程，每次事件触发都会重新拉起一个 Python 进程。这意味着：

脚本自己什么也记不住，所有"上次读到哪、已经报了多少轮"必须存到磁盘文件里。

这就是为什么有：

~/.claude/state/
├── langfuse_state.json   # 全局状态：每个 session 读到哪里、已报几轮
├── langfuse_state.lock   # 文件锁：防止两个 Python 进程同时改 state
└── langfuse_hook.log     # 这个 hook 自己的日志

类比：你是个临时工，每天上班先打开桌上的"交接本"看上一班做到哪里，下班前把进度写回去。

2.3 增量读取（offset + buffer）

源码：

python

@dataclass
class SessionState:
    offset: int = 0       # 上次读到 transcript 的第几个字节
    buffer: str = ""      # 上次没读完的"半行"暂存
    turn_count: int = 0   # 已经上报到 Langfuse 多少轮了

为什么需要 buffer？

类比：你抄日记，刚抄到一半，写日记的人正在写新句子。你抄到了 {"type":"user","mess，后面还没写完。你不能现在就交差，只能：

先把这半句话装进口袋（buffer）
下次来抄的时候，把口袋里的半句 + 今天写的新内容拼起来再切行

代码体现：

python

combined = ss.buffer + text
lines = combined.split("\n")
ss.buffer = lines[-1]      # 最后一行可能不完整，留下次再拼
ss.offset = new_offset      # 记录字节位置

为什么需要 offset？

因为 transcript 文件可能已经几十兆，你不能每次都从头读。f.seek(offset) 直接跳到上次读完的位置，只读新增字节。

2.4 文件锁（FileLock）

python

fcntl.flock(self._fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)

fcntl.flock 是 Linux/Mac 上的"文件锁"系统调用。LOCK_EX = 独占锁，LOCK_NB = 拿不到立刻返回不阻塞。

类比：你和另一个临时工都被同时叫去抄日记，但小本本只有一本。先到的人锁上抽屉，后到的人等 2 秒，等不到就放弃这次。这样小本本永远不会被两个人同时改花。

代码实现的策略：

拿不到锁 → 等 50ms 重试，最多重试 2 秒
还拿不到 → 不锁了，硬上（best-effort）
Windows 没有 fcntl → 直接不锁

为什么这样宽容？还是那句话：fail-open，绝不阻塞 Claude Code。

2.5 原子写入

python

tmp = STATE_FILE.with_suffix(".tmp")
tmp.write_text(json.dumps(state, ...))
os.replace(tmp, STATE_FILE)

为什么不直接写 STATE_FILE？

类比：你给老板誊抄一份重要合同。如果你直接在合同上改，写到一半电脑死机，合同就废了。聪明做法是：

先在草稿纸上抄完整版
检查无误
把旧合同扔掉，把草稿换上去（这一步是原子的）

os.replace 在 POSIX 系统是原子操作，要么完整替换，要么完全不动，绝不会出现半截文件。

2.6 Turn（轮次）的概念

这是整个脚本最精妙的部分。

一段对话长这样：

用户: "帮我看看 package.json"           ← 用户提问
助手: "好，我读一下"                     ← 助手文字
助手: 调用工具 Read("package.json")     ← 助手工具调用（tool_use）
[工具结果]: "{ ... }"                   ← tool_result（披着 user 外衣）
助手: "我看到这是个 Vite 项目..."        ← 助手最终回答
─────────────────────────────────────  ← 一轮（Turn）结束
用户: "再看下 vite.config.ts"           ← 新一轮开始
助手: ...

代码把这一坨杂乱的 JSON 行整理成一个 Turn 对象：

python

@dataclass
class Turn:
    user_msg: Dict             # 那条真正的用户提问
    assistant_msgs: List       # 这一轮里所有助手消息
    tool_results_by_id: Dict   # 工具调用 id → 工具返回结果

划分轮次的规则

读到一条真·用户消息（不是 tool_result 那种伪 user）→ 上一轮结束，新一轮开始。

python

if is_tool_result(msg):
    # 不分新轮，把工具结果塞进当前轮
    ...
elif role == "user":
    flush_turn()         # 上一轮收尾
    current_user = msg   # 新一轮开始
elif role == "assistant":
    # 加进当前轮的 assistant 列表
    ...

为什么要按 message.id 去重？

Claude Code 是流式写 transcript 的。同一条助手消息，可能：

先写进去一个"还没写完的版本"
一会儿又写一个"完整版本"
两条的 message.id 是一样的

如果不去重，你会把同一段话报两次给 Langfuse，看起来像幻觉。

代码用了一个小技巧——最新覆盖：

python

if mid not in assistant_latest:
    assistant_order.append(mid)   # 第一次见，记下出现顺序
assistant_latest[mid] = msg       # 永远用最新版本覆盖

类比：抄日记时遇到涂改——按"最后一次涂改后的版本"算。

tool_use 和 tool_result 的配对

tool_use 在 assistant 消息里，告诉你"我要调一个工具"，带有一个 id。
tool_result 在伪 user 消息里，带 tool_use_id 指回去。

代码用 dict 一对：

python

tool_results_by_id[tool_use_id] = tr.content

后面 emit_turn 上报时，对每个 tool_use 查一下这个 dict，找得到就把输出带上。

类比：餐厅服务员下单（tool_use）和厨房上菜（tool_result）通过"订单编号"对应。

2.7 截断与指纹（truncate_text）

python

MAX_CHARS = int(os.environ.get("CC_LANGFUSE_MAX_CHARS", "20000"))

为什么需要截断？

类比：你抄日记交给云端打印。如果某天日记里夹了一本《红楼梦》，你硬抄过去——传输费爆炸、云端存储爆炸、查询界面也卡死。

策略：

超过 20000 字符就截断
同时记下原长度 + SHA256 指纹
想看完整内容？拿指纹去本地 transcript 里查

python

return head, {
    "truncated": True,
    "orig_len": orig_len,
    "kept_len": len(head),
    "sha256": hashlib.sha256(s.encode("utf-8")).hexdigest(),
}

SHA256 的作用就像身份证号：唯一指向某段内容。Langfuse 上看到指纹，就能去本地全量日志里精准捞到原文。

2.8 Langfuse 上报的树形结构

python

with propagate_attributes(session_id=..., trace_name=...):
    with langfuse.start_as_current_observation(name="Turn N", ...) as trace_span:
        with langfuse.start_as_current_observation(as_type="generation", ...):
            pass
        for tc in tool_calls:
            with langfuse.start_as_current_observation(as_type="tool", ...) as tool_obs:
                tool_obs.update(output=tc.get("output"))

最终在 Langfuse 上呈现成一棵树：

Session: abc123                         ← 通过 session_id 把所有 turn 串起来
├─ Trace: "Claude Code - Turn 1"
│  ├─ Generation: "Claude Response"      (model=claude-sonnet, input=用户问, output=助手答)
│  ├─ Tool: "Tool: Read"                 (input=文件路径, output=文件内容)
│  └─ Tool: "Tool: Bash"                 (input=命令, output=stdout)
├─ Trace: "Claude Code - Turn 2"
│  └─ ...

类比：B 站的视频列表（Session）→ 单个视频（Trace）→ 视频里的章节（Generation/Tool）。

with 语法体现"作用域"：嵌套在里面的 observation 自动归属到外层的 trace 下。这是 Langfuse SDK 的设计精髓。

3. 代码逐段解读（按从上到下顺序）

3.1 头部：导入与全局常量

python

try:
    from langfuse import Langfuse, propagate_attributes
except Exception as e:
    sys.exit(0)

STATE_DIR = Path.home() / ".claude" / "state"
LOG_FILE = STATE_DIR / "langfuse_hook.log"
STATE_FILE = STATE_DIR / "langfuse_state.json"
LOCK_FILE = STATE_DIR / "langfuse_state.lock"

DEBUG = os.environ.get("CC_LANGFUSE_DEBUG", "").lower() == "true"
MAX_CHARS = int(os.environ.get("CC_LANGFUSE_MAX_CHARS", "20000"))

要点：

包没装 → 静默退出。
三个状态文件都放在 ~/.claude/state/，便于备份和清理。
DEBUG 由环境变量控制，避免改代码。

3.2 日志四件套

python

def _log(level, message):
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            f.write(f"{ts} [{level}] {message}\n")
    except Exception:
        pass   # 日志写不出也不能崩

类比：摄像头自己也有一个小日记本，记录"几点几分我开机了/我没拍到东西/我崩了"。即使日记本写不进去（磁盘满了），摄像头本体也不能因此死机。

3.3 FileLock

参考 2.4 节，已详解。

3.4 状态读写

python

def load_state():
    try:
        if not STATE_FILE.exists():
            return {}
        return json.loads(STATE_FILE.read_text(encoding="utf-8"))
    except Exception:
        return {}

def save_state(state):
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        tmp = STATE_FILE.with_suffix(".tmp")
        tmp.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8")
        os.replace(tmp, STATE_FILE)
    except Exception as e:
        debug(f"save_state failed: {e}")

def state_key(session_id, transcript_path):
    raw = f"{session_id}::{transcript_path}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

state_key 的目的：多个 session 的状态都存在同一个 JSON 里，用 hash 当 key 互不干扰。

json

{
  "abc123hash": { "offset": 12345, "buffer": "", "turn_count": 7 },
  "def456hash": { "offset": 9999,  "buffer": "{\"ty", "turn_count": 3 }
}

3.5 读 stdin

python

def read_hook_payload():
    try:
        data = sys.stdin.read()
        if not data.strip():
            return {}
        return json.loads(data)
    except Exception:
        return {}

Claude Code 把 hook 上下文以 JSON 的形式从 标准输入 喂给脚本，不是命令行参数。

类比：门铃感应器触发时，门口的小喇叭会塞一张小纸条进摄像头的口袋（stdin），上面写着"刚才是张三按的，他在 3 楼摄像头下"。

3.6 提取 session_id 和 transcript 路径

python

session_id = (
    payload.get("sessionId")
    or payload.get("session_id")
    or payload.get("session", {}).get("id")
)

为什么要兼容多种字段名？因为 Claude Code 的 hook payload 格式不同版本/不同事件类型字段不一样。这种"兼容写法"叫防御式编程。

类比：你订快递，地址栏可能写"门牌号"也可能写"房间号"也可能写"单元号"，你都得能识别。

3.7 Transcript 解析助手函数

这一组函数把 JSONL 行抽象成统一接口：

函数	作用	类比
`get_role(msg)`	取 user/assistant 角色	看说话人是谁
`get_content(msg)`	取消息内容	看说了啥
`is_tool_result(msg)`	判断这条 user 是不是工具回填	区分"客人说话"和"厨房上菜"
`iter_tool_results(content)`	提取所有 tool_result 块	列出这条消息里有几道菜
`iter_tool_uses(content)`	提取所有 tool_use 块	列出这条消息里下了几个订单
`extract_text(content)`	把 content 拍平成纯文本	把图文混排的话变成纯文字摘要
`truncate_text(s)`	超长截断 + 留指纹	把长篇大论压缩成摘要+条形码
`get_model(msg)`	取模型名	哪个老师答的题
`get_message_id(msg)`	取 message.id	这条消息的身份证号

3.8 增量读 JSONL

python

def read_new_jsonl(transcript_path, ss):
    if not transcript_path.exists():
        return [], ss

    with open(transcript_path, "rb") as f:
        f.seek(ss.offset)         # 跳到上次的位置
        chunk = f.read()           # 只读新增字节
        new_offset = f.tell()      # 记下新位置

    text = chunk.decode("utf-8", errors="replace")
    combined = ss.buffer + text    # 拼上上次的半行
    lines = combined.split("\n")
    ss.buffer = lines[-1]          # 最后一行可能不完整，留下来
    ss.offset = new_offset

    msgs = []
    for line in lines[:-1]:
        line = line.strip()
        if not line:
            continue
        try:
            msgs.append(json.loads(line))
        except Exception:
            continue              # 解析失败的行跳过
    return msgs, ss

完整体现了 2.3 节讲的"offset + buffer"思想。errors="replace" 是怕乱码字节让 decode 崩，遇到坏字节用 ? 替代。

3.9 build_turns（核心业务逻辑）

参考 2.6 节，这里讲一个细节：

python

def flush_turn():
    nonlocal current_user, ...
    if current_user is None:
        return
    if not assistant_latest:
        return                # 没有助手回复，不算一轮
    ...
    turns.append(Turn(...))

为什么 assistant_latest 为空就不算一轮？

类比：用户刚说完话、模型还没开始回复，你就不能急着打卡，因为"对话没闭环"。

实际场景：用户输入"帮我重构这段代码" → 模型还在思考 → 此时如果 hook 被触发了 → 这一轮先不报，等下次 hook 触发，模型已经回了，再一并报。

这就是为什么需要 buffer 和 offset 把"半成品轮次"留住下次再处理。

3.10 emit_turn（上报到 Langfuse）

参考 2.8 节。补充一个细节：

python

last_assistant = turn.assistant_msgs[-1]
assistant_text_raw = extract_text(get_content(last_assistant))

只取最后一条助手消息作为最终回答。中间的那些"思考、调工具"消息都被合并到 tool_calls 里了。

model = get_model(turn.assistant_msgs[0])：从第一条助手消息取模型名（一轮里模型不会变）。

3.11 main 函数

python

def main():
    if os.environ.get("TRACE_TO_LANGFUSE", "").lower() != "true":
        return 0                              # 总开关
    
    public_key = ...                          # 取凭证
    if not public_key or not secret_key:
        return 0

    payload = read_hook_payload()
    session_id, transcript_path = extract_session_and_transcript(payload)
    if not session_id or not transcript_path:
        return 0                              # 缺关键信息就放弃

    if not transcript_path.exists():
        return 0

    langfuse = Langfuse(...)

    try:
        with FileLock(LOCK_FILE):              # 加锁保证状态文件不被并发改坏
            state = load_state()
            key = state_key(session_id, str(transcript_path))
            ss = load_session_state(state, key)

            msgs, ss = read_new_jsonl(transcript_path, ss)
            if not msgs:
                # 即便没新消息也要保存（offset 可能变了）
                write_session_state(state, key, ss)
                save_state(state)
                return 0

            turns = build_turns(msgs)
            if not turns:
                write_session_state(state, key, ss)
                save_state(state)
                return 0

            emitted = 0
            for t in turns:
                emitted += 1
                turn_num = ss.turn_count + emitted
                try:
                    emit_turn(langfuse, session_id, turn_num, t, transcript_path)
                except Exception as e:
                    debug(f"emit_turn failed: {e}")
                    # 单轮失败不影响其它轮

            ss.turn_count += emitted
            write_session_state(state, key, ss)
            save_state(state)

        try:
            langfuse.flush()                   # 强制把网络请求发出去
        except Exception:
            pass

        return 0

    except Exception as e:
        debug(f"Unexpected failure: {e}")
        return 0

    finally:
        try:
            langfuse.shutdown()                # 关闭 SDK，确保 buffer 清空
        except Exception:
            pass

注意几个关键点：

多重短路：开关、凭证、payload、文件、SDK 初始化任何一步出错都直接 return 0。
状态总要保存：哪怕没产生新轮次，offset/buffer 也可能变了，也要存。
flush + shutdown：因为脚本马上要退出，必须主动把缓冲里的网络请求发完，否则数据就丢了。

类比：摄像头快下班了，必须先把今天没传完的录像传完再断电。

4. 把所有概念串成一个完整故事

设想你在 Claude Code 里输入：

"帮我把 src/login.tsx 的样式改成 Tailwind"

接下来发生的事：

第 1 秒：用户消息写进 transcript（第 100 行）。

第 2 秒：Claude Code 触发 UserPromptSubmit 事件（假设配置了这个 hook）。

Python 脚本拉起。
看 state.json：{ "offset": 95, "turn_count": 5 }。
从 offset=95 读到现在，新增 1 行（用户提问）。
build_turns 返回 0 个完整 turn（因为助手还没回）。
保存 offset，退出。

第 3 秒：Claude 模型回复 "好，我先读一下这个文件" + 调用 Read 工具。Transcript 多了 2 行。

第 5 秒：Read 工具返回文件内容。Transcript 多了 1 行。

第 8 秒：模型继续回复 "我看到这是 React 17，建议这样改：..." + 一段代码。Transcript 多了 1 行。

第 9 秒：Claude Code 触发 Stop 事件（一次完整对话回合结束）。

Python 脚本再次拉起。
看 state.json：{ "offset": 已更新, "turn_count": 5 }。
从上次 offset 读到现在，新增 4 行。
build_turns 返回 1 个 Turn（用户问 + 4 条助手 + 1 个工具结果配对完整）。
调用 emit_turn(turn_num=6) → 在 Langfuse 上看到一个 "Claude Code - Turn 6" 的 trace，下挂 1 个 generation + 1 个 tool。
turn_count 更新为 6，保存。
退出。

第 10 秒：你打开 Langfuse 网页，看到这一轮完整记录。

整个过程，Claude Code 主进程完全没受影响，即使中途 Python 脚本崩了，下次它还能从断点继续。

5. 安装与使用

5.1 准备工作

bash

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

5.2 设置环境变量

bash

export TRACE_TO_LANGFUSE=true
export CC_LANGFUSE_PUBLIC_KEY=pk-lf-xxxx
export CC_LANGFUSE_SECRET_KEY=sk-lf-xxxx
export CC_LANGFUSE_BASE_URL=https://cloud.langfuse.com
# 可选：
export CC_LANGFUSE_DEBUG=true
export CC_LANGFUSE_MAX_CHARS=30000

5.3 在 Claude Code 中注册 Hook

在 ~/.claude/settings.json 里加：

json

{
  "hooks": {
    "Stop": [
      {
        "matcher": "*",
        "hooks": [
          {
            "type": "command",
            "command": "python3 /path/to/langfuse_hook.py"
          }
        ]
      }
    ]
  }
}

也可以同时挂在 UserPromptSubmit、PostToolUse 等事件上，让上报更实时。但要注意：事件越多触发越频繁，对磁盘和 Langfuse 调用次数都有影响。

5.4 验证

起一个 Claude Code 会话，跑几轮对话。
看 ~/.claude/state/langfuse_hook.log，应该有 INFO Processed N turns ... 的日志。
打开 Langfuse 网页，应该看到对应的 traces。

5.5 排错

现象	排查
日志没生成	检查脚本可执行权限，检查 `~/.claude/state` 目录权限
日志有但 Langfuse 没数据	检查凭证、检查 `langfuse.flush()` 没异常、检查网络
Turn 编号跳跃	多个客户端同时跑，建议每台机器独立 state
Buffer 越来越大	transcript 文件被截断或 Claude 写入异常，可清空对应 session 的 state 重置

6. 速查表

你想做的	改哪里
关掉上报	`unset TRACE_TO_LANGFUSE`
切到自建 Langfuse	改 `CC_LANGFUSE_BASE_URL`
不截断长文本	把 `CC_LANGFUSE_MAX_CHARS` 设很大（注意成本）
重置某个 session 的进度	编辑 `~/.claude/state/langfuse_state.json` 删除对应 key
调试	`export CC_LANGFUSE_DEBUG=true`
改上报内容（比如脱敏）	修改 `emit_turn` 里的 `user_text/assistant_text`
改 trace 树形结构	修改 `emit_turn` 里 `start_as_current_observation` 的嵌套

7. 总结与思考题

一句话总结

这是一个"短生命周期 + 增量读 + 状态文件"的经典模式，把 Claude Code 不可变的 JSONL 日志流，转成 Langfuse 上的可回放、可分析的对话轨迹。

思考题（建议自己想一想再去看下一篇分析文档）

如果两个 Claude Code 同时在两个项目里跑，会不会冲突？
如果 Langfuse 暂时挂了，几小时后恢复，能补传吗？
如果用户中途 Ctrl+C 掉 Claude Code，已经发生的对话还能上报吗？
如果某条助手消息的 message.id 缺失，会怎样？
transcript 文件如果被旋转（rotate）或换路径，offset 还有效吗？

想知道答案 + 这个方案有什么问题、业内还有什么别的方案？请看： ./langfuse_hook_问题与方案对比.md

8. 进阶专题：subagent / Task 子会话怎么被处理？

一句话结论：默认情况下根本不会被上报；即使额外挂 SubagentStop 也只是扁平地并列，无法嵌套到主 Task span 下面。

8.1 真相：subagent 有自己的独立 transcript

实测 Claude Code v2.0.70 的存储结构（在 ~/.claude/projects/<项目hash>/ 下）：

~/.claude/projects/-data-workspace-codegen-demo/
├── d95a1d96-….jsonl          ← 主 session
├── 86811fe7-….jsonl          ← 主 session
├── agent-a67f567.jsonl       ← 子代理 a67f567 的独立 transcript
├── agent-a7822e5.jsonl       ← 另一个子代理
├── agent-a81d4c2.jsonl
└── agent-adfe0b5.jsonl

验证：

bash

grep -c '"isSidechain":true' 主.jsonl       # → 0
grep -c '"isSidechain":true' agent-*.jsonl  # → 全是 true

agent-*.jsonl 第一行长这样：

json

{
  "parentUuid": null,
  "isSidechain": true,
  "sessionId": "d95a1d96-…",      ← 指回主 session
  "agentId": "a67f567",            ← 子代理 id
  "type": "user",
  "message": {"role": "user", "content": "Warmup"}
}

也就是说：

每个 subagent 调用 = 一个独立的 agent-<id>.jsonl 文件
subagent 通过 sessionId 字段挂回主 session
主 session 的 .jsonl 里不会出现 sidechain 行

8.2 hook 默认行为：subagent 完全是黑盒

回忆 hook 的输入：stdin payload 里只有一个 transcript_path。

触发事件	transcript_path 指向
主对话的 `Stop` / `UserPromptSubmit`	主 session 的 jsonl
`SubagentStart` / `SubagentStop`	对应的 `agent-<id>.jsonl`

如果你在 settings.json 里只挂了主对话事件（最常见的 Stop），hook 永远不会读到 agent-*.jsonl。

那主对话视角看到了什么？

主 jsonl 里只能看到一对：

[主] assistant   tool_use(name="Task", id="t1", input={prompt:"...", agentId:"a67f567"})
[主] user        tool_result(t1, "<子代理最终摘要>")

→ Langfuse 上呈现：

✅ 主轮 Tool: Task 的 input 是给子代理的 prompt
✅ 主轮 Tool: Task 的 output 是子代理的最终摘要
❌ 子代理内部的 Grep/Read/思考链/token 用量 完全不可见

一句话：主对话视角完整、子代理是黑盒。

⚠️ 这一节修正了本文早期版本的一个错误结论：之前担心 sidechain 行会污染主 jsonl 导致伪轮次和 Task output 丢失—— 不会。subagent 在物理上是另一个文件，根本不会进 build_turns。

8.3 如果挂了 `SubagentStop`，会发生什么？

json

{
  "hooks": {
    "Stop": [...],
    "SubagentStop": [
      { "matcher": "*", "hooks": [{ "type": "command", "command": "python3 .../langfuse_hook.py" }] }
    ]
  }
}

此时子代理结束时也会触发 hook，stdin payload 里 transcript_path 指向 agent-<id>.jsonl。

代码运行轨迹：

state_key = sha256(sessionId + agent-xxx.jsonl路径) → 这是一个全新的 key（路径不同），不会和主 session 的 state 冲突。
把 agent-xxx.jsonl 当作一个独立的"session"处理：build_turns、emit_turn、turn_count 全是这个文件独立的。
上报到 Langfuse 时 session_id 用的是 payload 里那个 sessionId 字段（等于主 sessionId），所以在 Langfuse 的 Session 视图里会和主对话聚合到一起。

实际效果：

Langfuse Session: d95a1d96-...
├─ Trace: "Claude Code - Turn 1"   ← 主对话
├─ Trace: "Claude Code - Turn 2"   ← 主对话
├─ Trace: "Claude Code - Turn 1"   ← agent-a67f567 内部第 1 轮  ← 跟主对话同名！
└─ Trace: "Claude Code - Turn 2"   ← agent-a67f567 内部第 2 轮

新问题出现：

❌ trace 名字撞车：主对话和子代理都叫 "Claude Code - Turn N"，看起来像同一个 session 跑了 4 轮，实际上有 2 轮是子代理。
❌ 没有父子嵌套关系：子代理的 trace 没有挂在主轮 Tool: Task 这个 span 下面，扁平地并列在 Session 里。
❌ 子代理的 trace 不知道自己属于哪一次 Task 调用：如果一次主对话调了 3 次 Task，3 个 agent 文件混在一个 Session 里完全分不清。
✅ 但好处是：子代理的内部步骤、token 用量、工具调用全都能看见了。

8.4 类比

可以这样类比 subagent 的 transcript 设计：

老板（主 session）让秘书（子代理）出去办件事。秘书在另一个房间（agent-xxx.jsonl）独立录音。老板的会议室录音（主 jsonl）只听得到："秘书你去办XX，办好后回来汇报。"——"老板这是结果。" 秘书房间的录音里才能听到她到底问了谁、查了什么资料。
当前 hook 默认只录会议室；如果再加一个秘书房间的录音机（SubagentStop），两份录音也是分开归档的，没有"剪辑成一段嵌套时间线"的机制。

8.5 真正的正确形态应该是什么样

理想呈现：

Trace: Turn N (主)
├─ Generation: Claude Response (主)
└─ Tool: Task                          ← 这个 span 应作为容器
   ├─ Generation: Subagent Resp 1     (来自 agent-a67f567.jsonl)
   ├─ Tool: Grep
   ├─ Generation: Subagent Resp 2
   └─ output: 子会话最终摘要

要实现它，hook 需要：

在处理主 jsonl 遇到 tool_use(name=Task) 时，主动去读对应的 agent-<agentId>.jsonl。
把子代理 transcript 也跑一遍 build_turns，但作为子 span 嵌套到主 Task 下，而不是独立 trace。
维护跨文件的 offset/state（避免重复读）。

8.6 修复思路

思路	描述	难度
A	主 hook 在 emit Task tool 时，按 `tool_use.input.agentId` 找到 `agent-*.jsonl`，读取并嵌套上报；不再单独挂 `SubagentStop`	推荐
B	同时挂 `SubagentStop`，但额外维护一张"子 trace 待挂载"表，等主 hook 处理时把子 trace 链接为孩子	复杂、需 Langfuse 异步合并能力
C	不深究子代理内部，但至少把 `agentId` 写进主轮 Task span 的 metadata，方便事后人工对照 `agent-*.jsonl`	最简单的兜底

最低成本的代码骨架（思路 C）：

python

# 在 _tool_calls_from_assistants 里
for tu in iter_tool_uses(get_content(am)):
    name = tu.get("name")
    inp = tu.get("input") or {}
    extra = {}
    if name == "Task" and isinstance(inp, dict):
        agent_id = inp.get("agentId") or inp.get("subagent_type")
        if agent_id:
            extra["agent_transcript"] = f"agent-{agent_id}.jsonl"
    calls.append({"id": ..., "name": ..., "input": ..., "extra": extra})

把 extra 放进 emit_turn 的 metadata 里，Langfuse 上点开主轮 Tool: Task 就能看到该子代理的 transcript 文件名，需要时去本地查阅。

思路 A 的关键代码（伪代码示意）：

python

def emit_subagent_inline(langfuse, parent_span, agent_jsonl_path, session_id):
    # 读子代理整个 jsonl（subagent 通常一次跑完，不大）
    msgs = list(_read_all_jsonl(agent_jsonl_path))
    sub_turns = build_turns(msgs)
    for st in sub_turns:
        # 在 parent_span 这个上下文里嵌套创建 generation/tool observation
        ...

8.7 速查：当前行为 vs 期望行为（更新版）

项目	默认 (只挂 Stop)	挂了 SubagentStop	期望（思路 A 修复后）
子代理内容是否上报	❌ 完全不报	✅ 单独报	✅ 嵌套报
是否独立 Trace	n/a	✅（独立但同名）	❌（应作为主 Task 的子 span）
主轮 Task 的 output	✅ 子代理摘要	✅ 子代理摘要	✅ 子代理摘要
子代理内部步骤可见	❌	✅	✅
父子关系可见	❌	❌	✅
trace 名字会不会撞车	n/a	❌ 会	✅ 不会

docs

hook

watcher

Langfuse Hook 从零到一精读 ​

0. 先把背景讲清楚（非常重要） ​

0.1 这段代码到底想干什么？ ​

0.2 什么是 Hook？ ​

0.3 什么是 Langfuse？ ​

0.4 什么是 Transcript？ ​

1. 整体运行流程（先看森林，再看树） ​

2. 关键概念逐个攻破 ​

2.1 fail-open 原则 ​

2.2 短生命周期 + 状态文件 ​

2.3 增量读取（offset + buffer） ​

2.4 文件锁（FileLock） ​

2.5 原子写入 ​

2.6 Turn（轮次）的概念 ​

划分轮次的规则 ​

为什么要按 message.id 去重？ ​

tool_use 和 tool_result 的配对 ​

2.7 截断与指纹（truncate_text） ​

2.8 Langfuse 上报的树形结构 ​

3. 代码逐段解读（按从上到下顺序） ​

3.1 头部：导入与全局常量 ​

3.2 日志四件套 ​

3.3 FileLock ​

3.4 状态读写 ​

3.5 读 stdin ​

3.6 提取 session_id 和 transcript 路径 ​

3.7 Transcript 解析助手函数 ​

3.8 增量读 JSONL ​

3.9 build_turns（核心业务逻辑） ​

3.10 emit_turn（上报到 Langfuse） ​

3.11 main 函数 ​

4. 把所有概念串成一个完整故事 ​

5. 安装与使用 ​

5.1 准备工作 ​

5.2 设置环境变量 ​

5.3 在 Claude Code 中注册 Hook ​

5.4 验证 ​

5.5 排错 ​

6. 速查表 ​

7. 总结与思考题 ​

一句话总结 ​

思考题（建议自己想一想再去看下一篇分析文档） ​

8. 进阶专题：subagent / Task 子会话怎么被处理？ ​

8.1 真相：subagent 有自己的独立 transcript ​

8.2 hook 默认行为：subagent 完全是黑盒 ​

8.3 如果挂了 SubagentStop，会发生什么？ ​

8.4 类比 ​

8.5 真正的正确形态应该是什么样 ​

8.6 修复思路 ​

8.7 速查：当前行为 vs 期望行为（更新版） ​

Langfuse Hook 从零到一精读

0. 先把背景讲清楚（非常重要）

0.1 这段代码到底想干什么？

0.2 什么是 Hook？

0.3 什么是 Langfuse？

0.4 什么是 Transcript？

1. 整体运行流程（先看森林，再看树）

2. 关键概念逐个攻破

2.1 fail-open 原则

2.2 短生命周期 + 状态文件

2.3 增量读取（offset + buffer）

2.4 文件锁（FileLock）

2.5 原子写入

2.6 Turn（轮次）的概念

划分轮次的规则

为什么要按 message.id 去重？

tool_use 和 tool_result 的配对

2.7 截断与指纹（truncate_text）

2.8 Langfuse 上报的树形结构

3. 代码逐段解读（按从上到下顺序）

3.1 头部：导入与全局常量

3.2 日志四件套

3.3 FileLock

3.4 状态读写

3.5 读 stdin

3.6 提取 session_id 和 transcript 路径

3.7 Transcript 解析助手函数

3.8 增量读 JSONL

3.9 build_turns（核心业务逻辑）

3.10 emit_turn（上报到 Langfuse）

3.11 main 函数

4. 把所有概念串成一个完整故事

5. 安装与使用

5.1 准备工作

5.2 设置环境变量

5.3 在 Claude Code 中注册 Hook

5.4 验证

5.5 排错

6. 速查表

7. 总结与思考题

一句话总结

思考题（建议自己想一想再去看下一篇分析文档）

8. 进阶专题：subagent / Task 子会话怎么被处理？

8.1 真相：subagent 有自己的独立 transcript

8.2 hook 默认行为：subagent 完全是黑盒

8.3 如果挂了 `SubagentStop`，会发生什么？

8.4 类比

8.5 真正的正确形态应该是什么样

8.6 修复思路

8.7 速查：当前行为 vs 期望行为（更新版）