모델에 구애받지 않는 관측성을 위한 Langfuse
어떤 LLM 제공업체와도 작동하는 오픈 소스 대안으로 Langfuse를 통합하고, 검색 및 도구 호출을 위한 사용자 지정 스팬을 수집하며, 비용 추적 대시보드를 설정합니다.
모델에 구애받지 않는 관측성을 위한 Langfuse은(는) CoddyKit의 무료 AI Engineering Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 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 spanIntegrating 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.contentCost 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 trendsAdding 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 negativelyAutomated 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 hourPrompt 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 scoresSelf-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 튜터), CoddyKit PRO로 업그레이드하면 AI Engineering Academy 강의 전체를 잠금 해제할 수 있습니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.
“모델에 구애받지 않는 관측성을 위한 Langfuse”에서 뭘 배우나요?
어떤 LLM 제공업체와도 작동하는 오픈 소스 대안으로 Langfuse를 통합하고, 검색 및 도구 호출을 위한 사용자 지정 스팬을 수집하며, 비용 추적 대시보드를 설정합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Engineering Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Engineering Academy을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Engineering Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.
“모델에 구애받지 않는 관측성을 위한 Langfuse” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Engineering Academy 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Engineering Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- LLM 애플리케이션을 디버깅하기 어려운 이유
- LangSmith를 활용한 추적
- 모델에 구애받지 않는 관측성을 위한 Langfuse
- 지연 시간, 비용, 품질 저하 알림