第 6 篇：Langfuse 实战——从 Trace 接入到 Score、Dataset 和 Experiment

1
LangGraph
2
OpenAI Agents SDK
3
Claude Agent SDK
4
自定义 Agent Runtime
5
Temporal / Durable Workflow
6
业务编排器

1
Agent Framework / Runtime
2
        │
3
        │ 产生运行事实
4
        ▼
5
OpenTelemetry / Langfuse SDK
6
        │
7
        │ 采集和传播
8
        ▼
9
Langfuse
10
        │
11
        ├── Trace 与 Observation
12
        ├── Prompt 版本
13
        ├── Score
14
        ├── Annotation Queue
15
        ├── Dataset
16
        ├── Experiment
17
        └── Metrics / API

1
{
2
  "level": "INFO",
3
  "message": "tool call completed",
4
  "duration_ms": 438
5
}

1
一次完整任务
2
一次模型 Generation
3
一个 Tool Observation
4
一次 Retriever 调用
5
一个 Session
6
一个 Score
7
一个 Dataset Item
8
一次 Experiment Run

1
过去 24 小时所有失败的 Tool Observation
2
某个 Prompt 版本的平均输入 Token
3
某个 Agent 版本的 p95 延迟
4
某类 Retriever 的低分结果
5
所有带 user_feedback=negative 的 Trace
6
某个 Dataset Run 中失败的 Item

1
Observability
2
记录 Agent 实际发生了什么
3
        │
4
        ▼
5
Evaluation
6
判断执行得好不好
7
        │
8
        ▼
9
Dataset
10
保存值得重复验证的输入和期望结果
11
        │
12
        ▼
13
Experiment
14
比较 Prompt、模型和 Agent 版本
15
        │
16
        ▼
17
Prompt / Agent Release
18
上线更好的版本
19
        │
20
        └──────────────► 新的生产 Trace

1
跨函数上下文传播
2
跨线程和异步任务传播
3
跨服务 Trace Context
4
Parent-child 关系
5
Span 生命周期
6
OTLP 导出
7
Batch、Sampling 和 Collector

1
将 OTel Span 映射为 Observation
2
识别 Generation、Tool、Agent、Retriever
3
展示 Prompt、Token、Cost 与 TTFT
4
关联 User、Session、Environment 和 Release
5
保存 Score
6
管理 Dataset 和 Experiment
7
组织 Annotation Queue

1
所有共享同一 trace_id 的 Observations
2
+
3
传播到这些 Observations 的 Trace 级属性

1
一次用户 Prompt
2
一次 Coding Agent Turn
3
一次完整工具工作流
4
一次 Dataset Item 执行

1
Observation Type
2
Input
3
Output
4
Model
5
Model Parameters
6
Usage
7
Cost
8
Prompt Reference
9
Version
10
Metadata
11
Status
12
Level

1
agent
2
├── retriever
3
├── generation
4
├── tool
5
│   └── external API span
6
├── generation
7
└── evaluator

1
context.compose
2
parse_output
3
checkpoint.persist
4
approval.wait
5
git.diff.verify
6
artifact.upload
7
environment.restore

1
name
2
start_time
3
end_time
4
input
5
output
6
metadata
7
status
8
parent

1
from langfuse import get_client
2

3
langfuse = get_client()
4

5
with langfuse.start_as_current_observation(
6
    as_type="span",
7
    name="context.compose",
8
    input={"message_count": 12},
9
) as span:
10
    context = build_context()
11
    span.update(
12
        output={
13
            "estimated_tokens": context.estimated_tokens,
14
            "tool_count": len(context.tools),
15
        }
16
    )

1
model
2
model_parameters
3
input
4
output
5
usage_details
6
cost_details
7
completion_start_time
8
prompt reference

1
with langfuse.start_as_current_observation(
2
    as_type="generation",
3
    name="planner-model",
4
    model="model-x",
5
    model_parameters={
6
        "temperature": 0.2,
7
        "max_tokens": 2048,
8
    },
9
    input=messages,
10
) as generation:
11
    response = call_model(messages)
12

13
    generation.update(
14
        output=response.output,
15
        usage_details={
16
            "input": response.usage.input_tokens,
17
            "input_cached_tokens": response.usage.cached_input_tokens,
18
            "output": response.usage.output_tokens,
19
        },
20
        completion_start_time=response.first_token_time,
21
    )

1
规划
2
模型调用
3
工具选择
4
子 Agent
5
状态更新
6
最终输出

1
from langfuse import observe
2

3
@observe(as_type="agent", name="coding-agent")
4
def run_agent(task):
5
    ...

1
agent
2
├── generation: planner
3
├── retriever: repository-search
4
├── tool: read-file
5
├── generation: patch-author
6
├── tool: edit-file
7
├── tool: run-tests
8
└── generation: final-response

1
read_file
2
edit_file
3
run_tests
4
create_pull_request
5
send_message
6
database_query
7
MCP tool call

1
tool_call_id
2
tool_name
3
arguments
4
result
5
approval
6
side_effect_class
7
error
8
duration
9
artifact_ref

1
with langfuse.start_as_current_observation(
2
    as_type="tool",
3
    name="run-tests",
4
    input={
5
        "command": "pytest -q tests/orders",
6
        "tool_call_id": "tool_123",
7
    },
8
) as tool_obs:
9
    result = run_tests()
10
    tool_obs.update(
11
        output={
12
            "exit_code": result.exit_code,
13
            "summary": result.summary,
14
            "artifact_ref": result.log_ref,
15
        },
16
        metadata={
17
            "business_outcome": "tests_failed"
18
            if result.exit_code != 0
19
            else "tests_passed"
20
        },
21
    )

1
向量检索
2
BM25
3
SQL Read
4
代码搜索
5
文档搜索
6
Memory Recall

1
query
2
index_version
3
top_k
4
candidate_count
5
selected_chunks
6
scores
7
filter_reason

1
with langfuse.start_as_current_observation(
2
    as_type="retriever",
3
    name="repository-search",
4
    input={
5
        "query": "calculate_discount edge case",
6
        "index_version": "repo-abc123",
7
        "top_k": 8,
8
    },
9
) as retrieval:
10
    candidates = search_repository()
11

12
    retrieval.update(
13
        output={
14
            "selected": [
15
                {
16
                    "chunk_id": c.id,
17
                    "rank": c.rank,
18
                    "score": c.score,
19
                }
20
                for c in candidates[:4]
21
            ]
22
        }
23
    )

1
没召回
2
被过滤
3
Rerank 降级
4
Token Budget 淘汰
5
模型误读

1
Session
2
├── Trace 1：用户第一个问题
3
├── Trace 2：用户补充约束
4
├── Trace 3：Agent 继续执行
5
└── Trace 4：人工确认后的完成

1
from langfuse import observe, propagate_attributes
2

3
@observe(as_type="agent")
4
def process_turn(request):
5
    with propagate_attributes(
6
        session_id=request.session_id,
7
        user_id=request.user_id,
8
    ):
9
        return run_agent(request)

1
Trace
2
Observation
3
Session
4
Dataset Run

1
NUMERIC
2
CATEGORICAL
3
BOOLEAN
4
TEXT

1
用户反馈
2
人工标注
3
规则 Evaluator
4
LLM-as-a-Judge
5
Experiment Evaluator
6
Guardrail
7
外部评测流水线

1
{
2
  "name": "task_success",
3
  "value": 1,
4
  "data_type": "BOOLEAN",
5
  "comment": "Tests passed and no forbidden files changed",
6
  "trace_id": "trace_123"
7
}

1
id
2
dataset_id
3
input
4
expected_output
5
metadata
6
source_trace_id
7
source_observation_id
8
media_references
9
status

1
source_trace_id
2
source_observation_id

1
Dataset
2
├── Dataset Item A
3
├── Dataset Item B
4
└── Dataset Item C
5

6
Experiment / Dataset Run
7
├── Run Item A → Trace A → Scores
8
├── Run Item B → Trace B → Scores
9
└── Run Item C → Trace C → Scores

1
dataset_run_id
2
dataset_item_id
3
trace_id
4
observation_id（兼容字段）

1
import { NodeSDK } from "@opentelemetry/sdk-node";
2
import { LangfuseSpanProcessor } from "@langfuse/otel";
3

4
const sdk = new NodeSDK({
5
  spanProcessors: [
6
    new LangfuseSpanProcessor(),
7
  ],
8
});
9

10
sdk.start();

1
OTel Trace
2
OTel Span
3
GenAI Semantic Attributes

1
Langfuse Trace
2
Observation
3
Generation / Tool / Agent / Retriever

1
from langfuse.langchain import CallbackHandler
2
from langfuse import propagate_attributes
3

4
langfuse_handler = CallbackHandler()
5

6
with propagate_attributes(
7
    user_id="user_42",
8
    session_id="session_77",
9
    tags=["langgraph", "production"],
10
    metadata={
11
        "agent_version": "3.2.0",
12
        "graph_revision": "sha256:...",
13
    },
14
):
15
    result = graph.invoke(
16
        {"messages": [{"role": "user", "content": "Fix the bug"}]},
17
        config={
18
            "callbacks": [langfuse_handler],
19
        },
20
    )

1
chain
2
agent
3
tool
4
retriever
5
generation

1
from openinference.instrumentation.openai_agents import (
2
    OpenAIAgentsInstrumentor,
3
)
4

5
OpenAIAgentsInstrumentor().instrument()

1
from langfuse import observe, propagate_attributes
2

3
@observe(as_type="agent")
4
def run_workflow(input_data):
5
    with propagate_attributes(
6
        user_id="user_42",
7
        session_id="session_77",
8
        version="workflow-v5",
9
        metadata={"task_id": "task_123"},
10
    ):
11
        return Runner.run_sync(agent, input_data)

1
Agent
2
Generation
3
Function Tool
4
Handoff
5
Guardrail

1
Task ID
2
User / Session
3
Environment
4
Agent Version
5
Outcome
6
Score

1
from openinference.instrumentation.claude_agent_sdk import (
2
    ClaudeAgentSDKInstrumentor,
3
)
4

5
ClaudeAgentSDKInstrumentor().instrument()

1
async with ClaudeSDKClient(options=options) as client:
2
    await client.query("Fix the failing test")
3

4
    async for message in client.receive_response():
5
        handle_message(message)

1
user_id
2
session_id
3
task_id
4
agent_version
5
repository_revision
6
environment

1
langfuse.flush()

1
from __future__ import annotations
2

3
from dataclasses import dataclass
4
from typing import Any
5

6
from langfuse import get_client, propagate_attributes
7

8
langfuse = get_client()
9

10

11
@dataclass(frozen=True)
12
class Task:
13
    task_id: str
14
    user_id: str
15
    session_id: str
16
    repository: str
17
    prompt: str
18

19

20
def run_agent(task: Task) -> dict[str, Any]:
21
    with langfuse.start_as_current_observation(
22
        as_type="agent",
23
        name="coding-agent",
24
        input={
25
            "task_id": task.task_id,
26
            "prompt": task.prompt,
27
        },
28
    ) as agent_obs:
29

30
        with propagate_attributes(
31
            user_id=task.user_id,
32
            session_id=task.session_id,
33
            environment="production",
34
            tags=["coding-agent", "repo-fix"],
35
            metadata={
36
                "task_id": task.task_id,
37
                "repository": task.repository,
38
                "agent_version": "3.2.0",
39
                "prompt_version": "planner-v7",
40
            },
41
            version="3.2.0",
42
        ):
43
            retrieved = retrieve_repository_context(task)
44
            plan = generate_plan(task, retrieved)
45
            tool_results = execute_plan(plan)
46
            outcome = verify_outcome(task, tool_results)
47

48
            result = {
49
                "answer": outcome.answer,
50
                "tests_passed": outcome.tests_passed,
51
                "unexpected_changes": outcome.unexpected_changes,
52
                "diff_ref": outcome.diff_ref,
53
            }
54

55
            agent_obs.update(output=result)
56

57
            agent_obs.score_trace(
58
                name="task_success",
59
                value=1 if outcome.success else 0,
60
                data_type="BOOLEAN",
61
                comment=outcome.reason,
62
            )
63

64
            return result
65

66

67
def retrieve_repository_context(task: Task) -> list[dict]:
68
    with langfuse.start_as_current_observation(
69
        as_type="retriever",
70
        name="repository-search",
71
        input={
72
            "query": task.prompt,
73
            "index_version": "repo-index-abc123",
74
            "top_k": 8,
75
        },
76
    ) as retriever_obs:
77
        chunks = repository_search(task.prompt)
78

79
        output = [
80
            {
81
                "chunk_id": item.id,
82
                "rank": item.rank,
83
                "score": item.score,
84
                "path": item.path,
85
            }
86
            for item in chunks
87
        ]
88

89
        retriever_obs.update(output=output)
90
        return output
91

92

93
def generate_plan(task: Task, context: list[dict]) -> dict:
94
    with langfuse.start_as_current_observation(
95
        as_type="generation",
96
        name="planner",
97
        model="model-x",
98
        model_parameters={
99
            "temperature": 0.2,
100
            "max_tokens": 2048,
101
        },
102
        input={
103
            "task": task.prompt,
104
            "context": context,
105
        },
106
    ) as generation:
107
        response = planner_model(task.prompt, context)
108

109
        generation.update(
110
            output=response.plan,
111
            usage_details={
112
                "input": response.usage.input_tokens,
113
                "input_cached_tokens":
114
                    response.usage.cached_input_tokens,
115
                "output": response.usage.output_tokens,
116
            },
117
        )
118

119
        return response.plan
120

121

122
def execute_plan(plan: dict) -> list[dict]:
123
    results = []
124

125
    for tool_call in plan["tool_calls"]:
126
        with langfuse.start_as_current_observation(
127
            as_type="tool",
128
            name=tool_call["name"],
129
            input={
130
                "tool_call_id": tool_call["id"],
131
                "arguments": tool_call["arguments"],
132
            },
133
        ) as tool_obs:
134
            try:
135
                result = tool_runtime.execute(tool_call)
136
            except Exception as exc:
137
                tool_obs.update(
138
                    output={"error": type(exc).__name__},
139
                    level="ERROR",
140
                    status_message=str(exc),
141
                )
142
                raise
143

144
            normalized = {
145
                "tool_call_id": tool_call["id"],
146
                "success": result.success,
147
                "artifact_ref": result.artifact_ref,
148
                "business_outcome": result.business_outcome,
149
            }
150

151
            tool_obs.update(output=normalized)
152
            results.append(normalized)
153

154
    return results
155

156

157
def verify_outcome(task: Task, results: list[dict]):
158
    with langfuse.start_as_current_observation(
159
        as_type="evaluator",
160
        name="environment-outcome-verifier",
161
        input={
162
            "task_id": task.task_id,
163
            "tool_results": results,
164
        },
165
    ) as evaluator_obs:
166
        outcome = environment_verifier(task)
167

168
        evaluator_obs.update(
169
            output={
170
                "tests_passed": outcome.tests_passed,
171
                "unexpected_changes": outcome.unexpected_changes,
172
                "success": outcome.success,
173
            }
174
        )
175

176
        evaluator_obs.score(
177
            name="outcome_verified",
178
            value=1 if outcome.success else 0,
179
            data_type="BOOLEAN",
180
            comment=outcome.reason,
181
        )
182

183
        return outcome

1
with langfuse.start_as_current_observation(
2
    as_type="agent",
3
    name="root-agent",
4
):
5
    # 此处创建的新 Observation 默认成为 root-agent 的子节点
6
    ...

1
当前 Observation
2
→ 新 Observation 自动成为 Child

1
with langfuse.start_as_current_observation(
2
    as_type="agent",
3
    name="main-agent",
4
):
5
    with langfuse.start_as_current_observation(
6
        as_type="generation",
7
        name="planner",
8
    ):
9
        ...
10

11
    with langfuse.start_as_current_observation(
12
        as_type="tool",
13
        name="run-tests",
14
    ):
15
        ...

1
main-agent
2
├── planner
3
└── run-tests

1
32 位十六进制 trace_id
2
16 位十六进制 observation_id

1
from langfuse import get_client
2

3
langfuse = get_client()
4

5
trace_id = langfuse.create_trace_id(
6
    seed="task:task_123:run:run_456"
7
)

1
with langfuse.start_as_current_observation(
2
    as_type="agent",
3
    name="coding-agent",
4
    trace_context={
5
        "trace_id": "abcdef1234567890abcdef1234567890",
6
        "parent_span_id": "fedcba0987654321",
7
    },
8
):
9
    ...

1
HTTP request_id
2
Workflow run_id
3
Task run_id
4
Message delivery_id

1
traceparent
2
tracestate
3
baggage

1
API Gateway
2
→ Agent Service
3
→ Tool Service
4
→ MCP Server
5
→ External API

1
MCP Client
2
1. 获取当前 OTel Context
3
2. Inject traceparent / tracestate / baggage
4
3. 写入 Tool Call 的 _meta
5

6
MCP Server
7
1. 从 _meta Extract
8
2. Attach Context
9
3. 创建 Server / Tool Span

1
Separate Traces：
2
Client 和 Server 保持独立服务边界
3

4
Linked Trace：
5
通过 _meta 传播 OTel Context，形成一条完整 Trace

1
development
2
staging
3
production
4
canary
5
evaluation

1
with propagate_attributes(
2
    environment="production",
3
):
4
    ...

1
Trace
2
Observation
3
Score
4
Session

1
小写字母、数字、-、_
2
长度不超过 40
3
不能以 langfuse 开头

1
同一 Project
2
environment = tenant_a
3
environment = tenant_b

1
agent_version
2
model_route
3
request_model
4
response_model
5
prompt_name
6
prompt_version
7
tool_schema_set_version
8
retriever_version
9
memory_policy_version

1
release = Git Commit / Image Digest / Build ID
2
version = 某个 Observation、Agent 或 Prompt 的逻辑版本

1
release：
2
这次部署运行了哪份代码
3

4
version：
5
这个命名组件采用哪个逻辑实现

1
agent_version=3.2.0
2
prompt_version=planner-v7
3
retriever_version=rerank-v4

1
dataset_id
2
dataset_version
3
evaluator_name
4
evaluator_version
5
rubric_version
6
judge_model
7
judge_prompt_version

1
from datetime import datetime, timezone
2
from langfuse import get_client
3

4
langfuse = get_client()
5

6
dataset = langfuse.get_dataset(
7
    name="coding-agent-regression",
8
    version=datetime(
9
        2026, 7, 1, 0, 0, 0,
10
        tzinfo=timezone.utc,
11
    ),
12
)