第 14 章 · Langfuse + OpenTelemetry 集成

本章目标：

• 理解 Langfuse 作为 LLM 可观测性平台的作用
• 掌握 Python SDK 的基本用法
• 集成 LangGraph 实现自动追踪
• 使用 OpenTelemetry 扩展可观测性

14.1 Langfuse 简介

Langfuse 是开源的 LLM 工程平台，提供：

• Tracing：请求级追踪，可视化执行路径
• Evaluation：自动化评估和排名
• Analytics：成本、延迟、错误率统计
• Prompt Management：版本化管理提示词

部署方式：自托管（Docker）或 Langfuse Cloud。

14.2 Python SDK 基础

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse(
    secret_key="sk-lf-...",
    public_key="pk-lf-...",
    host="https://cloud.langfuse.com"  # 或自托管地址
)

# 创建 trace
trace = langfuse.trace(name="my-agent")

# 创建 span（子操作）
span = trace.span(name="llm-call")
span.end(output={"response": "Hello"})

# 创建 event（事件标记）
trace.event(name="user-input", input={"text": "Hi"})

14.3 @observe 装饰器

from langfuse.decorators import observe
from langfuse import Langfuse

langfuse = Langfuse()

@observe(name="rag_pipeline")
def rag_pipeline(question: str) -> str:
    """RAG 管道追踪"""
    # 自动创建 trace 和 span
    documents = retrieve_documents(question)
    answer = generate_answer(question, documents)
    return answer

@observe(name="retrieve")
def retrieve_documents(query: str) -> list:
    # 嵌入向量检索
    ...

@observe(name="generate")
def generate_answer(question: str, docs: list) -> str:
    # LLM 生成
    ...

# 调用自动追踪
result = rag_pipeline("什么是 LangGraph?")

14.4 LangGraph 集成

from langgraph.graph import StateGraph, MessagesState, START, END
from langfuse.decorators import observe Langfuse

langfuse = Langfuse()

@observe()
def llm_node(state: MessagesState) -> dict:
    # 自动追踪 LLM 调用
    return {"messages": [{"role": "assistant", "content": "回答..."}]}

graph = StateGraph(MessagesState)
graph.add_node("llm", llm_node)
graph.add_edge(START, "llm")
graph.add_edge("llm", END)
compiled = graph.compile()

# 执行自动追踪
result = compiled.invoke({"messages": [...]})

14.5 OpenTelemetry 集成

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 配置 Provider
provider = TracerProvider()
exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831
)
processor = BatchSpanProcessor(exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# 在 LangGraph 节点中使用
def observability_node(state: MessagesState) -> dict:
    with tracer.start_as_current_span("agent_processing"):
        # 业务逻辑
        result = process(state)
        
        # 记录指标
        tracer.get_meter("agent").create_counter(
            "processing_count"
        ).add(1)
        
        return result

14.6 关键指标追踪

import time
from langfuse import Langfuse

langfuse = Langfuse()

def tracked_llm_call(prompt: str) -> dict:
    start = time.time()
    
    # 调用 LLM
    result = call_model(prompt)
    
    # 记录指标
    duration = time.time() - start
    langfuse.score(
        trace_id=langfuse.get_current_trace_id(),
        name="latency_ms",
        value=duration * 1000
    )
    langfuse.score(
        trace_id=langfuse.get_current_trace_id(),
        name="token_usage",
        value=result["usage"]["total_tokens"]
    )
    
    return result

本章小结

• Langfuse 提供 LLM 应用的全链路可观测性
• @observe 装饰器简化追踪代码
• OpenTelemetry 扩展支持 Jaeger 等后端
• 关键指标：延迟、token 用量、错误率、成本

🛠️ 动手实践

1. 集成 Langfuse 到现有 LangGraph Agent，查看追踪面板
2. 添加自定义评分（accuracy、faithfulness）
3. 配置 Jaeger 后端，实现分布式追踪

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Langfuse 的核心功能不包括？

2. @observe 装饰器的作用是？

3. OpenTelemetry 主要用于什么？

4. Jaeger 是什么类型的后端？