AI Engineering Academy · 课时

使用 Langfuse 实现与模型无关的可观测性

将 Langfuse 集成为适用于任何 LLM 提供商的开源替代方案，捕获检索和工具调用的自定义跨度，并设置成本跟踪仪表板。

使用 Langfuse 实现与模型无关的可观测性 是 CoddyKit 上的免费 AI Engineering Academy 课时。这是第 3 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

Langfuse: Open-Source LLM Observability

Langfuse is an open-source observability platform for LLM applications that works with any model provider: OpenAI, Anthropic, Mistral, local models via Ollama, or your own fine-tuned model. Unlike LangSmith which ties you to LangChain, Langfuse integrates with any Python code through a simple SDK. You can self-host Langfuse for free or use the managed cloud at cloud.langfuse.com.

# pip install langfuse
from langfuse import Langfuse

langfuse = Langfuse(
    public_key='pk-lf-...',
    secret_key='sk-lf-...',
    host='https://cloud.langfuse.com'  # or your self-hosted URL
)

print('Langfuse connected:', langfuse.auth_check())

Traces, Spans, and Generations

Langfuse uses a hierarchical data model with three levels. A trace represents one end-to-end user request. Within a trace, spans represent individual processing steps (retrieval, preprocessing, tool calls). Generations are a special type of span specifically for LLM calls: they capture the model, prompt tokens, completion tokens, and cost in a structured way that enables cost dashboards and quality metrics.

from langfuse import Langfuse

langfuse = Langfuse()

# Create a trace for one user request
trace = langfuse.trace(
    name='rag-query',
    user_id='user_123',
    session_id='session_abc',
    tags=['production', 'rag']
)

# Add a retrieval span
retrieval_span = trace.span(
    name='vector-retrieval',
    input={'query': 'What is RAG?'}
)
chunks = vector_db.search('What is RAG?')
retrieval_span.end(output={'chunks': [c['text'][:100] for c in chunks]})

# Add an LLM generation
generation = trace.generation(
    name='answer-generation',
    model='gpt-4o',
    model_parameters={'temperature': 0.0},
    input=[{'role': 'user', 'content': 'Context: ...\nQuestion: What is RAG?'}]
)
response = openai_client.chat.completions.create(model='gpt-4o', messages=[...])
generation.end(
    output=response.choices[0].message.content,
    usage={'input': response.usage.prompt_tokens, 'output': response.usage.completion_tokens}
)

The Decorator Integration Pattern

Langfuse provides function decorators that automatically wrap your functions with trace spans. The @observe() decorator captures inputs and outputs, timing, and any exceptions. This is the cleanest way to instrument existing code without restructuring it.

from langfuse.decorators import observe, langfuse_context

# @observe wraps the function as a span automatically
@observe()
def retrieve_chunks(query: str) -> list[dict]:
    return vector_db.search(query, top_k=5)

@observe()
def generate_answer(query: str, context: str) -> str:
    response = openai_client.chat.completions.create(
        model='gpt-4o',
        messages=[
            {'role': 'system', 'content': 'Answer using the context.'},
            {'role': 'user', 'content': f'Context: {context}\nQuestion: {query}'}
        ]
    )
    # Attach LLM usage data to the current span
    langfuse_context.update_current_observation(
        usage={'input': response.usage.prompt_tokens, 'output': response.usage.completion_tokens},
        model='gpt-4o'
    )
    return response.choices[0].message.content

@observe(name='rag-pipeline')  # top-level trace
def rag_pipeline(query: str) -> str:
    chunks = retrieve_chunks(query)  # becomes a nested span
    context = '\n'.join([c['text'] for c in chunks])
    return generate_answer(query, context)  # becomes another nested span

Integrating with Any LLM Provider

Unlike LangSmith's deep integration with LangChain, Langfuse works with any LLM provider using the same decorator-based approach. Whether you are calling Anthropic's API, a local Ollama model, a Hugging Face inference endpoint, or a custom model you fine-tuned, Langfuse traces the call the same way. This provider neutrality is essential when you run multiple models in the same application.

from langfuse.decorators import observe, langfuse_context
import anthropic
from openai import OpenAI

anthropic_client = anthropic.Anthropic()
openai_client = OpenAI()

@observe()
def call_claude(prompt: str) -> str:
    response = anthropic_client.messages.create(
        model='claude-3-5-sonnet-20241022',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': prompt}]
    )
    langfuse_context.update_current_observation(
        model='claude-3-5-sonnet-20241022',
        usage={'input': response.usage.input_tokens, 'output': response.usage.output_tokens}
    )
    return response.content[0].text

@observe()
def call_gpt4(prompt: str) -> str:
    response = openai_client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}]
    )
    langfuse_context.update_current_observation(model='gpt-4o',
        usage={'input': response.usage.prompt_tokens, 'output': response.usage.completion_tokens})
    return response.choices[0].message.content

Cost Tracking Dashboards

Langfuse automatically computes cost from model name and token counts using a built-in pricing table that covers OpenAI, Anthropic, Mistral, and dozens of other providers. The cost dashboard shows: total spend by time period, cost broken down by model, cost broken down by feature or user (using tags and metadata), and daily/weekly spend trends. This visibility prevents bill shock and helps identify expensive outlier requests.

# Cost data is automatically computed - no manual config
# Langfuse knows: gpt-4o input = $0.005/1K tokens, output = $0.015/1K tokens

# Add metadata to enable cost breakdown by feature
@observe(name='rag-query')
def handle_rag_query(query: str, feature: str, user_id: str) -> str:
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=[feature, 'rag'],
        metadata={'feature': feature, 'query_length': len(query)}
    )
    return rag_pipeline(query)

# In Langfuse dashboard you can now filter costs by:
# - feature: 'document_qa', 'chat', 'summarization'
# - user_id: to see which users are your most expensive
# - model: to compare gpt-4o vs gpt-4o-mini costs
# - date range: to see daily/weekly/monthly trends

Adding User Feedback Scores

Langfuse allows you to attach user feedback to traces after the fact. When a user clicks a thumbs up/down on a response, you can record this as a score on the corresponding trace. This connects real user satisfaction signals to the full trace context, enabling you to analyze what makes highly-rated responses different from poorly-rated ones.

from langfuse.decorators import observe, langfuse_context

@observe()
def generate_response(query: str) -> dict:
    answer = rag_pipeline(query)
    # Get the current trace ID to link feedback later
    trace_id = langfuse_context.get_current_trace_id()
    return {'answer': answer, 'trace_id': trace_id}

# Later, when user submits feedback:
def record_user_feedback(trace_id: str, score: int, comment: str):
    langfuse.score(
        trace_id=trace_id,
        name='user_satisfaction',  # score name
        value=score,               # 1 (thumbs up) or 0 (thumbs down)
        comment=comment,
        data_type='BOOLEAN'
    )

# Now in Langfuse: filter traces where user_satisfaction = 0
# to find the exact prompts and contexts that users rated negatively

Automated Scores with LLM-as-Judge

Beyond user feedback, Langfuse supports automated scoring using LLM-as-judge evaluators. You can define evaluators that run asynchronously against sampled traces and score them on criteria like relevance, faithfulness, toxicity, or format correctness. These automated scores populate the same score dashboard as human feedback, giving you continuous quality monitoring without human annotation at scale.

from langfuse import Langfuse

langfuse = Langfuse()

def auto_score_traces():
    # Get recent unscored traces
    traces = langfuse.fetch_traces(tags=['production'], limit=50)
    
    for trace in traces.data:
        question = trace.input.get('query', '')
        answer = trace.output.get('answer', '') if trace.output else ''
        
        if not question or not answer:
            continue
        
        # LLM-as-judge scoring
        score = evaluate_relevance(question, answer)  # returns 0.0-1.0
        
        langfuse.score(
            trace_id=trace.id,
            name='auto_relevance',
            value=score,
            data_type='NUMERIC',
            comment='Automated relevance score from LLM judge'
        )

# Run this as a scheduled job every hour

Prompt Management in Langfuse

Langfuse includes a prompt management feature that stores your prompts in the Langfuse cloud and lets you fetch them at runtime. This decouples prompt versions from code deployments — you can update a prompt in the Langfuse UI and the change takes effect immediately without a code deploy. Langfuse also tracks which prompt version each trace used, so you can compare performance across prompt versions.

from langfuse import Langfuse

langfuse = Langfuse()

# Fetch the current production prompt by name
# The prompt lives in Langfuse UI, not in your code
prompt = langfuse.get_prompt('rag-system-prompt', version='production')

# Use it in your pipeline
messages = [
    {'role': 'system', 'content': prompt.compile(context_limit=4000)},
    {'role': 'user', 'content': query}
]

response = openai_client.chat.completions.create(model='gpt-4o', messages=messages)

# The trace is automatically linked to the prompt version
# In Langfuse you can filter: show me traces using prompt v3 vs v4
# and compare their quality scores

Self-Hosting Langfuse

Langfuse can be self-hosted with a single Docker Compose command, using PostgreSQL for storage. Self-hosting means your trace data never leaves your infrastructure — essential for applications handling PII, medical data, or proprietary content. The self-hosted version has the same features as the managed cloud but requires you to manage the infrastructure (backups, scaling, upgrades).

# Self-host Langfuse with Docker Compose
# docker-compose.yml (simplified)
# version: '3'
# services:
#   langfuse:
#     image: langfuse/langfuse:2
#     ports:
#       - '3000:3000'
#     environment:
#       - DATABASE_URL=postgresql://langfuse:password@postgres/langfuse
#       - NEXTAUTH_SECRET=your-random-secret
#       - SALT=your-random-salt
#   postgres:
#     image: postgres:15
#     environment:
#       - POSTGRES_DB=langfuse
#       - POSTGRES_PASSWORD=password

# After docker-compose up, point your SDK to:
langfuse = Langfuse(
    public_key='pk-lf-your-key',
    secret_key='sk-lf-your-key',
    host='http://localhost:3000'  # your self-hosted instance
)

Langfuse vs LangSmith: When to Choose Which

Choose LangSmith when you use LangChain heavily and want zero-configuration automatic tracing, deep integration with LangChain evaluations, and you are comfortable with vendor dependency. Choose Langfuse when you use multiple LLM providers, need to self-host for data privacy compliance, want open-source transparency, or build with frameworks other than LangChain. Both are production-ready and both offer generous free tiers.

OpenTelemetry Integration for LLMs

For teams that already use OpenTelemetry for distributed tracing, Langfuse supports OTLP (OpenTelemetry Protocol) ingestion. You can send LLM trace data from your existing OTel exporters directly to Langfuse without changing your instrumentation. This enables a unified observability stack where LLM traces, database query spans, and HTTP request traces all live in the same system with consistent correlation IDs.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Configure OTel to send to Langfuse OTLP endpoint
exporter = OTLPSpanExporter(
    endpoint='https://cloud.langfuse.com/api/public/otel/v1/traces',
    headers={
        'Authorization': 'Basic ' + base64.b64encode(b'pk-lf-xxx:sk-lf-xxx').decode()
    }
)

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

# Now create spans as usual - they appear in Langfuse automatically
tracer = trace.get_tracer('my-llm-app')
with tracer.start_as_current_span('rag-query') as span:
    span.set_attribute('llm.model', 'gpt-4o')
    span.set_attribute('llm.prompt_tokens', 500)
    result = rag_pipeline(query)

Quick Check

Test your understanding of Langfuse for model-agnostic observability from this lesson.

Lesson Recap

In this lesson you learned: Langfuse provides open-source, model-agnostic LLM observability using a hierarchical traces-spans-generations data model, the @observe() decorator instruments existing code with minimal changes, and cost tracking, user feedback scores, and automated LLM-as-judge scoring make Langfuse a complete quality monitoring platform. Next up we set up alerting on latency, cost, and quality degradation.

常见问题解答

「使用 Langfuse 实现与模型无关的可观测性」课时是免费的吗？

是的 — 「使用 Langfuse 实现与模型无关的可观测性」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。

「使用 Langfuse 实现与模型无关的可观测性」这节课中我会学到什么？

将 Langfuse 集成为适用于任何 LLM 提供商的开源替代方案，捕获检索和工具调用的自定义跨度，并设置成本跟踪仪表板。你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Engineering Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 3 节课，共 4 节。

「使用 Langfuse 实现与模型无关的可观测性」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Engineering Academy 课中编写并运行代码吗？

能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Engineering Academy