이 글은 누구를 위한 것인가
- LLM 프로덕션에서 무슨 일이 일어나는지 전혀 모르는 팀
- 어떤 프롬프트 버전이 더 좋은 결과를 내는지 추적하고 싶은 엔지니어
- 사용자별, 기능별 LLM 비용을 분리해서 보고 싶은 팀
들어가며
LLM을 프로덕션에 올리면 블랙박스가 된다. "저 사용자가 왜 나쁜 답변을 받았을까?", "어제 비용이 갑자기 왜 늘었을까?", "프롬프트를 바꿨는데 실제로 좋아졌는지 어떻게 아나?" — 옵저버빌리티 없이는 이 질문들에 답할 수 없다.
Langfuse는 오픈소스 LLM 옵저버빌리티 플랫폼으로, Trace→Span 구조로 LLM 호출 전체를 추적할 수 있다.
이 글은 bluefoxdev.kr의 LLM 운영 모니터링 가이드 를 참고하고, Langfuse 실전 통합 관점에서 확장하여 작성했습니다.
1. Langfuse 트레이싱 구조
[Langfuse 계층 구조]
Trace (최상위 단위)
= 하나의 사용자 요청 전체
속성: user_id, session_id, name, metadata
└── Span (중간 단계)
= 파이프라인의 각 단계
예: "문서 검색", "컨텍스트 조합", "LLM 호출"
└── Generation (LLM 호출)
= 실제 API 호출
속성: model, input, output, usage (tokens, cost)
예시 — RAG 파이프라인:
Trace: "사용자 질문 처리"
├── Span: "의도 파악" (LLM)
├── Span: "벡터 검색" (DB)
├── Span: "컨텍스트 재랭킹"
└── Generation: "최종 답변 생성" (LLM)
2. Langfuse 기본 설정
# pip install langfuse anthropic
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
import anthropic
# Langfuse 초기화
langfuse = Langfuse(
public_key="lf_pk_...",
secret_key="lf_sk_...",
host="https://cloud.langfuse.com" # 또는 self-hosted
)
client = anthropic.Anthropic()
# 데코레이터로 자동 트레이싱
@observe(name="answer_question")
def answer_question(user_id: str, question: str) -> str:
# 현재 트레이스에 사용자 정보 추가
langfuse_context.update_current_trace(
user_id=user_id,
tags=["production", "v2"],
metadata={"channel": "web"}
)
# LLM 호출 — 자동으로 Generation 스팬 생성
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": question}]
)
return response.content[0].text
3. 멀티스텝 RAG 파이프라인 트레이싱
from langfuse.decorators import observe
import time
@observe(name="rag_pipeline")
def rag_pipeline(user_id: str, query: str) -> str:
"""RAG 파이프라인 전체 트레이스"""
# Step 1: 쿼리 임베딩
embedding = get_embedding(query)
# Step 2: 벡터 검색
docs = vector_search(embedding, top_k=5)
# Step 3: 재랭킹
ranked_docs = rerank(query, docs)
# Step 4: 컨텍스트 조합
context = "\n\n".join(d["content"] for d in ranked_docs[:3])
# Step 5: LLM 호출 (자동 추적)
answer = generate_answer(query, context)
return answer
@observe(name="vector_search")
def vector_search(embedding: list[float], top_k: int = 5) -> list[dict]:
"""벡터 검색 스팬"""
start = time.time()
results = db.similarity_search(embedding, top_k=top_k)
# 검색 메타데이터 추가
langfuse_context.update_current_observation(
metadata={
"top_k": top_k,
"result_count": len(results),
"search_time_ms": round((time.time() - start) * 1000),
}
)
return results
@observe(name="generate_answer", as_type="generation")
def generate_answer(query: str, context: str) -> str:
"""LLM 생성 스팬 — 비용 자동 추적"""
prompt = f"""주어진 컨텍스트를 바탕으로 질문에 답하세요.
컨텍스트:
{context}
질문: {query}"""
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
# 토큰 사용량 기록 (비용 계산용)
langfuse_context.update_current_observation(
usage={
"input": response.usage.input_tokens,
"output": response.usage.output_tokens,
},
model="claude-sonnet-4-6",
)
return response.content[0].text
4. 프롬프트 버전 관리
# Langfuse에서 프롬프트 버전 관리
# 1. 프롬프트 등록 (Langfuse UI 또는 API)
langfuse.create_prompt(
name="customer_support",
prompt="당신은 {{company}} 고객지원 전문가입니다.\n\n{{context}}\n\n사용자: {{question}}\n전문가:",
config={
"model": "claude-sonnet-4-6",
"max_tokens": 512,
"temperature": 0.3,
},
labels=["production"], # 프로덕션 태그
)
# 2. 프로덕션에서 사용
def get_support_answer(question: str, context: str) -> str:
# 최신 프로덕션 프롬프트 로드
prompt_obj = langfuse.get_prompt("customer_support", label="production")
# 변수 치환
prompt = prompt_obj.compile(
company="우리 회사",
context=context,
question=question,
)
response = client.messages.create(
model=prompt_obj.config.get("model", "claude-sonnet-4-6"),
max_tokens=prompt_obj.config.get("max_tokens", 512),
messages=[{"role": "user", "content": prompt}]
)
# 어떤 프롬프트 버전을 사용했는지 추적
langfuse_context.update_current_observation(
prompt=prompt_obj, # 버전 자동 연결
)
return response.content[0].text
# 3. A/B 테스트 (프롬프트 버전 비교)
import random
def get_answer_ab_test(question: str) -> tuple[str, str]:
variant = "v2" if random.random() > 0.5 else "v1"
prompt_obj = langfuse.get_prompt("customer_support", label=variant)
# ... 동일한 로직
langfuse_context.update_current_trace(
metadata={"ab_variant": variant}
)
return answer, variant
5. 비용·품질 대시보드 쿼리
# Langfuse API로 분석 데이터 조회
from langfuse import Langfuse
from datetime import datetime, timedelta
langfuse = Langfuse()
# 최근 7일 사용자별 비용
def get_user_costs(days: int = 7) -> list[dict]:
traces = langfuse.fetch_traces(
from_timestamp=datetime.now() - timedelta(days=days),
limit=1000
)
user_costs = {}
for trace in traces.data:
user_id = trace.user_id or "anonymous"
cost = trace.total_cost or 0
user_costs[user_id] = user_costs.get(user_id, 0) + cost
return sorted(
[{"user_id": k, "cost_usd": round(v, 4)} for k, v in user_costs.items()],
key=lambda x: -x["cost_usd"]
)
# 프롬프트 버전별 품질 비교
def compare_prompt_versions(prompt_name: str) -> dict:
generations = langfuse.fetch_observations(
type="GENERATION",
name=prompt_name,
)
version_scores = {}
for gen in generations.data:
version = gen.prompt_version or "unknown"
scores = gen.scores or []
if version not in version_scores:
version_scores[version] = {"scores": [], "count": 0}
version_scores[version]["count"] += 1
for score in scores:
version_scores[version]["scores"].append(score.value)
return {
version: {
"count": data["count"],
"avg_score": round(sum(data["scores"]) / len(data["scores"]), 2) if data["scores"] else None
}
for version, data in version_scores.items()
}
마무리
LLM 옵저버빌리티 없이 프로덕션을 운영하는 것은 로그 없이 서버를 운영하는 것과 같다. Langfuse의 Trace→Span 구조는 복잡한 RAG 파이프라인도 단계별로 가시화할 수 있다.
프롬프트 버전 관리를 Langfuse에서 하면 코드 배포 없이 프롬프트를 바꿀 수 있고, A/B 테스트 결과도 대시보드에서 바로 확인된다. 처음부터 통합하는 것이 나중에 후처리하는 것보다 훨씬 쉽다.