3. Langfuse 上报映射

SDK：Langfuse Python v2 有状态 API（langfuse>=2,<3）

---

1. 上报层级结构

一轮主会话 Turn 在 Langfuse 中的典型结构：

Trace: "Claude Code - Turn 3"
├── Generation: "Claude Response 1"     # 第一次模型调用
├── Span: "Tool: Bash"                  # 工具调用
├── Generation: "Claude Response 2"     # 拿到工具结果后的模型调用
├── Span: "Tool: Agent"                 # 子代理调用
│   ├── Span: "Subagent[Explore] Turn 1"
│   │   ├── Generation: "Claude Response 1"
│   │   └── Span: "Tool: Grep"
│   └── Span: "Subagent[Explore] Turn 2"
│       └── Generation: "Claude Response 1"
└── Generation: "Claude Response 3"     # 最终回复

生活类比：一张订单（Trace）里，先做凉菜（Gen 1）→ 去取食材（Tool）→ 再炒菜（Gen 2）→ 叫帮厨（Agent Span）→ 最后装盘（Gen 3）。

---

2. 主 Turn 上报：emit_main_turn()

def emit_main_turn(langfuse, session_id, turn_num, turn, transcript_path, sub_files, used_subagents):
    trace = langfuse.trace(
        name=f"Claude Code - Turn {turn_num}",
        session_id=session_id,
        tags=["claude-code"],
        timestamp=msg_ts(turn.user_msg),
        input={"role": "user", "content": user_text},
        metadata={
            "session_id": session_id,
            "turn_number": turn_num,
            "transcript_path": str(transcript_path),
            "source": "claude-code",
        },
    )
    _emit_turn(container=trace, turn=turn, sub_files=sub_files, used_subagents=used_subagents)

字段	来源
session_id	jsonl / 文件路径 stem
timestamp	用户消息的 timestamp
input	用户文本（脱敏 + 截断后）
output	最后一个有文本的 Generation 输出（由 _emit_turn 设置）

---

3. 核心发射：emit_turn()

_emit_turn(container, turn, sub_files, used_subagents) 在任意 Langfuse 容器（trace 或 span）下发射子 observation。

3.1 Generation 发射

每个 message.id 分组 → 一个 Generation：

container.generation(
    name="Claude Response" if len(calls) == 1 else f"Claude Response {gi + 1}",
    start_time=g["start_dt"],
    end_time=g["end_dt"],
    model=g["model"],
    input=gen_input,          # 第一次=user 文本；后续=tool results
    output={"role": "assistant", "content": gtext},
    usage_details=g["usage"], # {"input": N, "output": M}
    metadata={"message_id": g["mid"], "call_index": gi + 1},
)

Input 逻辑：

序号	gen_input 内容
第 1 次	{"role": "user", "content": "用户原文"}
第 2+ 次	{"role": "tool", "content": [tool_result 列表]}

这样 Langfuse 里能看到模型每次调用时实际收到了什么。

3.2 Tool Span 发射

tool_span = container.span(
    name=f"Tool: {tool_name}",
    start_time=tu_ts,         # tool_use 时间
    end_time=tr_ts or tu_ts,  # tool_result 时间
    input=in_obj,             # tool_use.input（脱敏截断）
    metadata={"tool_name": ..., "tool_id": ...},
)

有结果后 tool_span.end(output=..., metadata=...)。

3.3 子代理嵌套

当 tool_name in {"Task", "Agent"} 且 INCLUDE_SUBAGENTS=true：

sf = match_subagent(sub_files, used, tool_use_input, tool_use_ts)
if sf:
    sub_summary = _emit_subagent_inline(parent_span=tool_span, sub_file=sf)

_emit_subagent_inline() 读取子代理 jsonl → build_turns() → 每个子 Turn 建 Span → 递归 _emit_turn()。

turn_span = parent_span.span(
    name=f"Subagent[{agent_type}] Turn {idx}",
    input={"role": "user", "content": u_text},
)
_emit_turn(container=turn_span, turn=st, sub_files=[], ...)

子代理内部不再递归发现子代理（sub_files=[]），避免无限嵌套。

---

4. 时间线与排序

Langfuse 按 start_time 排列同级 observation。本方案为 Generation 和 Tool Span 都设置了真实时间戳，所以 UI 时间线能正确反映：

10:00:01  Generation 1（模型决定调工具）
10:00:02  Tool: Bash（执行命令）
10:00:03  Tool Result 回来
10:00:04  Generation 2（模型看到结果后继续）

---

5. Token 与成本

def get_usage(msg):
    u = msg["message"]["usage"]
    return {
        "input": u["input_tokens"],
        "output": u["output_tokens"],
        "cache_creation_input": u.get("cache_creation_input_tokens"),
        "cache_read_input": u.get("cache_read_input_tokens"),
    }

• 每个 Generation 带自己的 usage_details
• Trace metadata 汇总 usage_total（所有 Generation 之和）

---

6. 容器模式（v2 API 特点）

Langfuse v2 没有隐式「当前 span 上下文」，必须显式传递父对象：

trace = langfuse.trace(...)
trace.generation(...)   # 子 observation
trace.span(...)         # 子 observation

parent_span = trace.span(...)
parent_span.generation(...)  # 嵌套子 observation

流式模式用 _StreamingTraceHandle / _StreamingSpanHandle 包装 trace_id + parent_observation_id，实现 resume 时不重复创建 Trace。

---

7. 完整映射表

jsonl 字段	Langfuse 字段	处理
type=user content	Trace.input	脱敏 + 截断
type=assistant text	Generation.output	按 message.id 分组
message.model	Generation.model	默认 "claude"
message.usage	Generation.usage_details	跳过全零占位
message.id	Generation.metadata.message_id	分组键
tool_use.name	Span.name = Tool: {name}
tool_use.input	Span.input	脱敏截断
tool_result.content	Span.output
timestamp	start_time / end_time	UTC 解析
stop_reason=end_turn	Turn 完成标志
sessionId	Trace.session_id
subagent meta agentType	Span.name 中的标签

---

8. 上报后 Trace metadata 示例

{
  "source": "claude-code",
  "turn_number": 3,
  "transcript_path": "/home/user/.claude/projects/.../abc.jsonl",
  "tool_count": 4,
  "model_call_count": 3,
  "usage_total": {"input": 5200, "output": 380},
  "user_text_meta": {"truncated": false, "orig_len": 42},
  "final_text_meta": {"truncated": false, "orig_len": 256}
}

下一章：按你的方案阅读 4_hook_scheme.md 或 5_watcher_scheme.md。