0Pricing
AI Engineering Academy · Урок

Langfuse для наблюдаемости, независимой от модели

Интегрируйте Langfuse как альтернативу с открытым исходным кодом, работающую с любым поставщиком LLM, добавьте пользовательские интервалы для поиска и вызовов инструментов и настройте панели отслеживания расходов.

«Langfuse для наблюдаемости, независимой от модели» — бесплатный урок AI Engineering Academy на CoddyKit. Это урок 3 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения 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 для наблюдаемости, независимой от модели» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Engineering Academy, подпишись на CoddyKit PRO. Курс AI Engineering Academy содержит 4 уроков всего.

Чему я научусь в уроке «Langfuse для наблюдаемости, независимой от модели»?

Интегрируйте Langfuse как альтернативу с открытым исходным кодом, работающую с любым поставщиком LLM, добавьте пользовательские интервалы для поиска и вызовов инструментов и настройте панели отслежив… Ты практикуешь AI Engineering Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Engineering Academy?

Предыдущий опыт не требуется. AI Engineering Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 3 из 4.

Сколько времени занимает урок «Langfuse для наблюдаемости, независимой от модели»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Engineering Academy?

Да. Каждый урок AI Engineering Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. Почему приложения на LLM сложно отлаживать
  2. Трассировка с LangSmith
  3. Langfuse для наблюдаемости, независимой от модели
  4. Оповещения о задержке, расходах и ухудшении качества
← Назад к AI Engineering Academy