Langfuse는 LLM(Large Language Model) 애플리케이션을 위한 오픈소스 엔지니어링 플랫폼입니다. 팀이 협업하여 LLM 애플리케이션을 디버깅, 분석, 반복 개선할 수 있도록 돕는 통합 솔루션을 제공합니다.
LLM 애플리케이션의 모든 동작을 투명하게 추적하고 모니터링할 수 있는 기능입니다.
userId를 통해 개별 사용자의 비용과 사용량 모니터링Python SDK 예제:
from langfuse import Langfuse
# 초기화
langfuse = Langfuse(
public_key="pk-lf-...",
secret_key="sk-lf-..."
)
# Trace 생성
trace = langfuse.trace(
name="my-llm-app",
user_id="user-123",
metadata={"environment": "production"}
)
# Generation 추적
generation = trace.generation(
name="chat-completion",
model="gpt-4",
input={"messages": [{"role": "user", "content": "Hello"}]},
output={"content": "Hi there!"},
usage={"prompt_tokens": 10, "completion_tokens": 5}
)
JavaScript/TypeScript SDK 예제:
import { Langfuse } from "langfuse";
const langfuse = new Langfuse({
publicKey: "pk-lf-...",
secretKey: "sk-lf-..."
});
const trace = langfuse.trace({
name: "my-llm-app",
userId: "user-123",
metadata: { environment: "production" }
});
const generation = trace.generation({
name: "chat-completion",
model: "gpt-4",
input: { messages: [{ role: "user", content: "Hello" }] },
output: { content: "Hi there!" },
usage: { promptTokens: 10, completionTokens: 5 }
});
프롬프트를 체계적으로 관리하고 버전 관리하며, 팀과 협업할 수 있는 기능입니다.
production, staging, development 등 환경별로 프롬프트 배포# 프롬프트 템플릿
"""
당신은 {{role}}입니다.
사용자 질문: {{user_question}}
답변 형식: {{format}}
"""
# 사용 시
prompt = langfuse.get_prompt("assistant-template")
compiled = prompt.compile(
role="친절한 AI 어시스턴트",
user_question="Python이란?",
format="간단명료하게"
)
프롬프트 생성 및 가져오기:
from langfuse import Langfuse
langfuse = Langfuse()
# 프롬프트 가져오기
prompt = langfuse.get_prompt(
name="customer-support-agent",
label="production" # 또는 version=3
)
# 프롬프트 컴파일 (변수 치환)
compiled_prompt = prompt.compile(
customer_name="홍길동",
issue_type="결제 문제"
)
# LLM에 전달
response = openai.chat.completions.create(
model="gpt-4",
messages=compiled_prompt
)
LLM 애플리케이션의 품질을 측정하고 개선하기 위한 체계적인 평가 시스템입니다.
예제:
# LLM-as-a-Judge 평가 설정
langfuse.score(
trace_id="trace-123",
name="answer-quality",
value=0.85,
comment="답변이 정확하고 관련성이 높음"
)
API 예제:
POST /api/public/scores
{
"traceId": "trace-123",
"name": "user-satisfaction",
"value": 1,
"dataType": "BOOLEAN",
"comment": "사용자가 답변에 만족함"
}
SDK 예제:
langfuse.score(
trace_id="trace-123",
name="hallucination-check",
value=0,
data_type="NUMERIC",
comment="환각 없음"
)
브라우저 SDK 예제:
import { Langfuse } from "langfuse";
const langfuse = new Langfuse({
publicKey: "pk-lf-..."
});
// 사용자 피드백 전송
langfuse.score({
traceId: "trace-123",
name: "user-feedback",
value: 1, // 좋아요
comment: "정확한 답변이었어요!"
});
데이터셋 생성:
# 데이터셋 생성
dataset = langfuse.create_dataset(
name="customer-support-qa",
description="고객 지원 QA 테스트 케이스"
)
# 아이템 추가
dataset.create_item(
input={"question": "환불은 어떻게 하나요?"},
expected_output={"answer": "구매일로부터 14일 이내 환불 가능합니다."}
)
실험 실행:
from langfuse import Langfuse
langfuse = Langfuse()
# 데이터셋 로드
dataset = langfuse.get_dataset("customer-support-qa")
# 실험 실행
for item in dataset.items:
# 프롬프트 가져오기
prompt = langfuse.get_prompt("support-agent", version=2)
# LLM 호출
response = llm.generate(
prompt=prompt.compile(**item.input),
trace_id=item.id
)
# 결과 평가
langfuse.score(
trace_id=item.id,
name="accuracy",
value=evaluate(response, item.expected_output)
)
graph TD
A[프로덕션 트레이스] --> B{평가 필요}
B --> C[LLM-as-a-Judge]
B --> D[Annotation Queue]
B --> E[User Feedback]
C --> F[Score 생성]
D --> F
E --> F
F --> G[Analytics Dashboard]
G --> H{문제 발견}
H -->|Yes| I[Dataset 추가]
I --> J[Experiment 실행]
J --> K[개선된 버전]
K --> L[프로덕션 배포]
H -->|No| M[모니터링 계속]
# Metrics API 사용 예제
import requests
response = requests.get(
"https://cloud.langfuse.com/api/public/metrics",
headers={"Authorization": "Bearer sk-lf-..."},
params={
"from": "2024-01-01",
"to": "2024-01-31",
"groupBy": "model"
}
)
metrics = response.json()
UI에서 내보내기:
Blob Storage로 내보내기:
# 데이터 내보내기 설정
langfuse.configure_export(
destination="s3://my-bucket/langfuse-exports",
schedule="daily",
format="parquet"
)
API Reference: https://api.reference.langfuse.com
# SDK로 데이터 쿼리
traces = langfuse.fetch_traces(
name="chat-completion",
from_timestamp="2024-01-01",
to_timestamp="2024-01-31",
user_id="user-123"
)
for trace in traces:
print(f"Trace: {trace.id}, Cost: {trace.cost}")
# 데이터 보존 정책 설정
langfuse.set_retention_policy(
traces=90, # 90일 후 삭제
scores=365, # 1년 후 삭제
datasets="永久" # 영구 보존
)
from langfuse.langchain import LangfuseCallbackHandler
handler = LangfuseCallbackHandler(
public_key="pk-lf-...",
secret_key="sk-lf-..."
)
# LangChain 체인에 추가
chain.run(input="Hello", callbacks=[handler])
from llama_index.core import set_global_handler
set_global_handler("langfuse")
# 이후 모든 LlamaIndex 호출이 자동으로 추적됨
from langfuse.openai import openai
# OpenAI SDK를 래핑
client = openai.OpenAI()
# 기존 코드 그대로 사용하면 자동 추적
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Hello"}]
)
import { createOpenAI } from '@ai-sdk/openai';
import { observeVercelAI } from 'langfuse';
const openai = observeVercelAI(createOpenAI({
apiKey: process.env.OPENAI_API_KEY
}));
from instructor import patch
from langfuse.openai import openai
client = patch(openai.OpenAI())
# Pydantic 모델과 함께 사용
import litellm
from litellm.integrations.langfuse import LangfuseLogger
litellm.success_callback = ["langfuse"]
litellm.failure_callback = ["langfuse"]
# 이후 모든 LiteLLM 호출이 자동 추적
response = litellm.completion(
model="gpt-4",
messages=[{"role": "user", "content": "Hello"}]
)
PoC → 개발 → 테스트 → 프로덕션
↓ ↓ ↓ ↓
관찰 프롬프트 평가 모니터링
관리 실험
빠른 시작:
pip install langfuse
from langfuse import Langfuse
langfuse = Langfuse(
public_key="your-public-key",
secret_key="your-secret-key"
)
# 첫 번째 트레이스
trace = langfuse.trace(name="my-app")
권장사항:
시기: 프롬프트가 어느 정도 안정화되기 시작할 때
단계:
1. 기존 프롬프트를 Langfuse에 등록
2. 코드에서 프롬프트 가져오기로 변경
3. 팀과 함께 프롬프트 개선
4. 버전 관리 시작
# Before
prompt = "당신은 고객 지원 AI입니다..."
# After
prompt = langfuse.get_prompt("customer-support", label="production")
개발 환경:
1. 테스트 데이터셋 구축
2. 실험 실행
3. 품질 메트릭 정의
프로덕션 환경:
1. 사용자 피드백 수집
2. LLM-as-a-Judge 설정
3. 정기적인 어노테이션
과제:
Langfuse 활용:
# 1. 대화 추적
trace = langfuse.trace(
name="customer-support",
user_id=customer_id,
session_id=conversation_id,
metadata={"channel": "web", "language": "ko"}
)
# 2. 프롬프트 관리
prompt = langfuse.get_prompt("support-agent", label="production")
# 3. 의도 분류
classification = trace.span(
name="intent-classification",
input=user_message
)
# 4. 답변 생성
response = trace.generation(
name="response-generation",
model="gpt-4",
input=prompt.compile(
user_message=user_message,
user_history=history
)
)
# 5. 사용자 피드백 수집
langfuse.score(
trace_id=trace.id,
name="user-satisfaction",
value=user_rating
)
결과:
과제:
Langfuse 활용:
trace = langfuse.trace(name="rag-pipeline")
# 1. 문서 검색
retrieval = trace.span(
name="document-retrieval",
input={"query": user_question},
metadata={"top_k": 5}
)
docs = vector_db.search(user_question, k=5)
retrieval.end(output={"retrieved_docs": len(docs)})
# 2. 리랭킹
reranking = trace.span(name="reranking")
ranked_docs = reranker.rank(docs, user_question)
reranking.end(output={"top_docs": ranked_docs[:3]})
# 3. 답변 생성
generation = trace.generation(
name="answer-generation",
model="gpt-4",
input={
"question": user_question,
"context": ranked_docs
}
)
# 4. 환각 체크 (LLM-as-a-Judge)
langfuse.score(
trace_id=trace.id,
name="hallucination-check",
value=hallucination_score
)
결과:
과제:
Langfuse 활용:
# 세션으로 연속된 코딩 세션 추적
trace = langfuse.trace(
name="code-assistant",
session_id=coding_session_id,
user_id=developer_id
)
# 컨텍스트 수집
context_span = trace.span(name="context-gathering")
code_context = get_project_context(project_id)
context_span.end(output=code_context)
# 프롬프트 선택 (언어별)
prompt = langfuse.get_prompt(
f"code-gen-{programming_language}",
label="production"
)
# 코드 생성
generation = trace.generation(
name="code-generation",
model="gpt-4",
input=prompt.compile(
task=user_request,
context=code_context,
language=programming_language
)
)
# 코드 품질 평가
langfuse.score(
trace_id=trace.id,
name="code-quality",
value=static_analysis_score
)
결과:
과제:
Langfuse 활용:
# A/B 테스트를 위한 프롬프트 버전
import random
version = "A" if random.random() < 0.5 else "B"
prompt = langfuse.get_prompt(
"content-generator",
label=f"experiment-{version}"
)
trace = langfuse.trace(
name="content-generation",
metadata={"experiment_group": version}
)
# 콘텐츠 생성
generation = trace.generation(
name="generate-content",
model="gpt-4",
input=prompt.compile(
topic=content_topic,
tone=brand_tone,
format=content_format
)
)
# 사용자 참여도 추적
langfuse.score(
trace_id=trace.id,
name="engagement-rate",
value=calculate_engagement(content_id)
)
결과:
자체 호스팅 옵션:
최신 가격: https://langfuse.com/pricing
Langfuse는 활발하게 개발 중이며, 커뮤니티 피드백을 기반으로 계속 발전하고 있습니다.
최신 로드맵: https://langfuse.com/docs/roadmap
최근 추가된 기능 (Changelog):
계획된 기능:
Langfuse는 LLM 애플리케이션을 개발하고 운영하는 데 필요한 모든 도구를 제공하는 종합 플랫폼입니다.
핵심 가치:
시작이 간단합니다:
pip install langfuse
from langfuse import Langfuse
langfuse = Langfuse()
# 이제 시작입니다!
더 나은 LLM 애플리케이션을 만들기 위한 여정을 Langfuse와 함께하세요! 🚀