AI 에이전트 모니터링 완전 가이드 — LangSmith vs Langfuse 실전 비교

프로덕션에서 AI 에이전트가 이상한 답을 내놨어요.

고객이 계좌 잔액을 물었는데 에이전트가 숫자를 지어냈어요. 4번의 툴 호출, 2개의 서브 에이전트. 어디서 망가졌는지 로그엔 최종 출력만 있어요.

일반 로그:
[ERROR] Response: "잔액은 1,250,000원입니다" ← 틀림. 근데 왜?

관측성 툴의 트레이스:
[Trace: user_query]
  ├─ [Tool: get_account_id] → 성공 (32ms)
  ├─ [Tool: get_balance] → 실패 (타임아웃) ← 여기서 망가짐
  ├─ [LLM: fallback_response] → 환각 발생
  └─ [Output] "잔액은 1,250,000원입니다"

관측성이 없으면 5단계 에이전트 디버깅은 추측이에요. 트레이스가 있으면 정확히 어느 단계에서 뭐가 잘못됐는지 바로 보여요.

---

LLM 관측성에서 추적해야 할 핵심 지표

일반 APM(Datadog, New Relic)과 달라요. LLM 전용 지표가 따로 있어요.

전통적인 APM:        LLM 관측성:
- 응답 시간          - 응답 시간 + TTFT(첫 토큰 시간)
- 에러율             - 에러율 + 환각 발생률
- CPU/메모리          - 토큰 사용량 (입력/출력)
- 요청 수            - 비용 (요청당 $)
                    - 검색 정확도 (RAG 사용 시)
                    - 답변 품질 점수
                    - 프롬프트 버전별 성능

이 지표들을 추적하는 두 대장이 LangSmith와 Langfuse예요.

---

LangSmith — LangChain 공식 관측성 플랫폼

개념

LangChain 팀이 만든 공식 LLMOps 플랫폼이에요. LangChain/LangGraph 앱에서 환경 변수 하나로 자동 트레이싱이 핵심이에요.

설치 및 기본 설정

# 환경 변수만 설정하면 LangChain 앱 전체 자동 트레이싱
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__xxxxxxxxxxxx"
os.environ["LANGCHAIN_PROJECT"] = "my-production-app"

# 이후 LangChain 코드는 변경 없이 자동 트레이싱됨
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

llm = ChatAnthropic(model="claude-sonnet-4-6")
prompt = ChatPromptTemplate.from_template("다음 질문에 답하세요: {question}")
chain = prompt | llm

# 이 호출이 LangSmith에 자동으로 기록됨
response = chain.invoke({"question": "파이썬 리스트 정렬 방법은?"})

LangGraph 에이전트 트레이싱

from langgraph.graph import StateGraph, END
from langchain_core.tracers.langchain import wait_for_all_tracers
from typing import TypedDict

class AgentState(TypedDict):
    query: str
    search_results: list
    answer: str

def search_node(state: AgentState) -> dict:
    # 이 함수 내부 LLM 호출도 자동 트레이싱
    results = web_search(state["query"])
    return {"search_results": results}

def answer_node(state: AgentState) -> dict:
    answer = llm.invoke(
        f"다음 정보를 바탕으로 답하세요:\n{state['search_results']}\n\n질문: {state['query']}"
    )
    return {"answer": answer.content}

# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("search", search_node)
workflow.add_node("answer", answer_node)
workflow.set_entry_point("search")
workflow.add_edge("search", "answer")
workflow.add_edge("answer", END)
app = workflow.compile()

# 실행 — 전체 그래프 실행이 하나의 트레이스로 기록됨
result = app.invoke({"query": "2026년 AI 트렌드는?"})

# 비동기 환경에서는 트레이서 완료 대기
wait_for_all_tracers()

수동 트레이싱 (LangChain 외 코드)

from langsmith import traceable
from langsmith.wrappers import wrap_openai
from openai import OpenAI

# OpenAI 클라이언트 래핑 — 자동 트레이싱
openai_client = wrap_openai(OpenAI())

@traceable(name="rag_pipeline", run_type="chain")
def rag_pipeline(query: str) -> str:
    # 검색 단계
    docs = retrieve_documents(query)

    # LLM 단계
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "다음 문서를 참고해서 답하세요."},
            {"role": "user", "content": f"문서:\n{docs}\n\n질문: {query}"}
        ]
    )

    return response.choices[0].message.content

@traceable(name="retrieve_documents", run_type="retriever")
def retrieve_documents(query: str) -> list:
    # 벡터 검색 — 이 단계도 별도 스팬으로 기록
    return vectorstore.similarity_search(query, k=3)

평가 파이프라인

from langsmith.evaluation import evaluate
from langsmith import Client

client = Client()

# 평가 데이터셋 생성
dataset = client.create_dataset(
    "rag_evaluation_v1",
    description="RAG 파이프라인 평가용 데이터셋"
)

# 테스트 케이스 추가
examples = [
    {
        "inputs": {"query": "환불 정책은?"},
        "outputs": {"answer": "구매 후 30일 이내 환불 가능합니다."}
    },
    {
        "inputs": {"query": "배송 기간은?"},
        "outputs": {"answer": "일반 배송은 3~5 영업일입니다."}
    }
]
client.create_examples(inputs=[e["inputs"] for e in examples],
                        outputs=[e["outputs"] for e in examples],
                        dataset_id=dataset.id)

# LLM-as-Judge 평가자 정의
def correctness_evaluator(run, example) -> dict:
    """정답과 비교해서 점수 계산"""
    prompt = f"""
    정답: {example.outputs["answer"]}
    모델 답변: {run.outputs["answer"]}

    모델 답변이 정답과 의미상 일치하나요?
    JSON으로만 답하세요: {{"score": 0~1, "reasoning": "이유"}}
    """
    import json
    result = json.loads(llm.invoke(prompt).content)
    return {"key": "correctness", "score": result["score"]}

# 평가 실행
results = evaluate(
    rag_pipeline,          # 평가할 함수
    data=dataset,          # 평가 데이터셋
    evaluators=[correctness_evaluator],
    experiment_prefix="rag-v2-test"
)

print(f"평균 정확도: {results.aggregate_feedback['correctness']:.2f}")

---

Langfuse — 오픈소스 프레임워크 무관 플랫폼

개념

MIT 라이선스 오픈소스예요. 셀프호스팅이 킬러 기능이에요. LangChain에 종속되지 않고 어떤 프레임워크든 다 트레이싱해요. 2025년 6월 Python SDK v3가 OpenTelemetry 기반으로 재작성됐어요.

설치

pip install langfuse

# 셀프호스팅 (Docker Compose 한 방에 실행)
git clone https://github.com/langfuse/langfuse
cd langfuse
docker compose up -d
# localhost:3000 에서 대시보드 접근

기본 트레이싱 — @observe 데코레이터

import os
from langfuse import observe, get_client
from langfuse.openai import openai  # OpenAI 자동 래핑

os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxxxxxxxxxxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxxxxxxxxxxx"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"
# 셀프호스팅이면: os.environ["LANGFUSE_HOST"] = "http://localhost:3000"

langfuse = get_client()

@observe()  # 이 데코레이터만 붙이면 자동 트레이싱
def rag_pipeline(query: str) -> str:
    docs = retrieve_docs(query)
    answer = generate_answer(query, docs)
    return answer

@observe(name="document_retrieval")
def retrieve_docs(query: str) -> list:
    return vectorstore.similarity_search(query, k=3)

@observe(name="answer_generation")
def generate_answer(query: str, docs: list) -> str:
    # langfuse.openai 래핑 시 토큰/비용 자동 기록
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"참고 문서:\n{docs}"},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

# 실행
result = rag_pipeline("환불 정책은?")
langfuse.flush()  # 비동기 전송 완료 대기

Anthropic 연동

from langfuse.decorators import observe, langfuse_context
import anthropic

client = anthropic.Anthropic()

@observe()
def claude_agent(query: str) -> str:
    # 현재 트레이스에 메타데이터 추가
    langfuse_context.update_current_observation(
        input=query,
        metadata={"version": "v2.1", "env": "production"}
    )

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": query}]
    )

    # 토큰 사용량 수동 기록 (Anthropic은 자동 래핑 없음)
    langfuse_context.update_current_observation(
        usage={
            "input": response.usage.input_tokens,
            "output": response.usage.output_tokens
        },
        output=response.content[0].text
    )

    return response.content[0].text

프롬프트 버전 관리

Langfuse의 강점 중 하나예요. 프롬프트를 코드에서 분리하고 버전을 관리해요.

from langfuse import get_client

langfuse = get_client()

# 프롬프트 생성 및 버전 관리 (대시보드에서도 가능)
langfuse.create_prompt(
    name="rag_system_prompt",
    prompt="다음 문서를 참고해서 답하세요. 모르면 '모른다'고 하세요.\n\n문서: {{context}}",
    labels=["production"],  # 프로덕션 태그
    config={"model": "claude-sonnet-4-6", "temperature": 0.1}
)

# 프로덕션 버전 불러오기
@observe()
def answer_with_versioned_prompt(query: str, context: str) -> str:
    # 항상 production 태그의 최신 버전 사용
    prompt = langfuse.get_prompt("rag_system_prompt", label="production")

    # 변수 치환
    system_message = prompt.compile(context=context)

    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

LLM-as-Judge 자동 평가

from langfuse import get_client

langfuse = get_client()

# 평가자 등록 (대시보드에서도 가능)
langfuse.create_score_config(
    name="answer_quality",
    data_type="NUMERIC",
    min=0,
    max=1
)

# 트레이스에 점수 부여
def evaluate_and_score(trace_id: str, query: str, answer: str):
    prompt = f"""
    질문: {query}
    답변: {answer}

    이 답변의 품질을 0~1 점수로 평가하세요.
    기준: 정확성, 완전성, 명확성
    JSON으로만: {{"score": 0.0~1.0, "comment": "평가 이유"}}
    """
    import json
    result = json.loads(llm.invoke(prompt).content)

    # 트레이스에 점수 기록
    langfuse.score(
        trace_id=trace_id,
        name="answer_quality",
        value=result["score"],
        comment=result["comment"]
    )

# 프로덕션에서 샘플링해서 평가
import random

@observe()
def monitored_rag(query: str) -> str:
    answer = rag_pipeline(query)

    # 5% 샘플링해서 자동 평가
    if random.random() < 0.05:
        trace_id = langfuse_context.get_current_trace_id()
        evaluate_and_score(trace_id, query, answer)

    return answer

---

두 플랫폼 전체 비교

항목 LangSmith Langfuse

오픈소스	❌ (독점 SaaS)	✅ (MIT 라이선스)
셀프호스팅	엔터프라이즈만	✅ 무료
프레임워크 독립성	LangChain 최적화	✅ 모든 프레임워크
자동 트레이싱	✅ LangChain 한 줄	데코레이터 필요
오버헤드	~0% (최저)	~15%
무료 티어	5K 트레이스/월	50K 유닛/월
프롬프트 관리	✅	✅
LLM-as-Judge	✅	✅
데이터 주권	❌ (클라우드만)	✅ 완전 통제
가격 모델	트레이스당	유닛당
GitHub 스타	비공개	19,000+

---

실전 선택 기준

LangSmith를 선택해야 할 때

✅ LangChain / LangGraph 100% 사용 중
✅ 코드 변경 없이 즉시 트레이싱 원함
✅ 관리형 SaaS가 편함
✅ 소규모 트래픽 (비용 예측 가능)

Langfuse를 선택해야 할 때

✅ 의료, 금융 등 데이터가 외부에 나가면 안 되는 업종
✅ LangChain 외 다른 프레임워크 혼용 (FastAPI + Anthropic 직접 호출 등)
✅ 대용량 트래픽 (셀프호스팅으로 비용 절감)
✅ 오픈소스 스택 선호
✅ 프롬프트 버전 관리 중요

---

프로덕션 모니터링 대시보드 구성

두 플랫폼 공통으로 설정해야 할 알림이에요.

# 비용 폭탄 방지 — 일별 비용 임계값 알림
def setup_cost_alerts():
    """Langfuse API 또는 LangSmith API로 알림 설정"""

    # 하루 $50 초과 시 슬랙 알림
    daily_cost = get_daily_cost()
    if daily_cost > 50:
        send_slack_alert(f"⚠️ LLM 비용 경고: 오늘 ${daily_cost:.2f} 사용됨")

# 품질 저하 감지 — 평균 점수 하락 알림
def monitor_quality_degradation():
    recent_scores = get_recent_scores(hours=1)
    avg_score = sum(recent_scores) / len(recent_scores)

    if avg_score < 0.7:  # 70% 미만이면 알림
        send_slack_alert(f"🚨 품질 저하 감지: 평균 점수 {avg_score:.2f}")

# 레이턴시 급등 감지
def monitor_latency():
    p95_latency = get_p95_latency(minutes=10)

    if p95_latency > 5000:  # 5초 초과
        send_slack_alert(f"⏱️ 레이턴시 급등: P95 {p95_latency}ms")

# cron으로 주기적 실행
# */5 * * * * python monitor.py

---

전체 관측성 파이프라인

사용자 요청
    ↓
[AI 에이전트]
    ├─ 모든 LLM 호출 → LangSmith / Langfuse 트레이스
    ├─ 툴 호출 결과 → 스팬으로 기록
    └─ 최종 응답 → 출력 기록
         ↓
[자동 평가] (5% 샘플링)
    ├─ LLM-as-Judge → 품질 점수
    └─ 저점수 트레이스 → 사람 검토 큐
         ↓
[알림]
    ├─ 비용 > 임계값 → Slack 알림
    ├─ 품질 < 임계값 → Slack 알림
    └─ 레이턴시 급등 → PagerDuty
         ↓
[대시보드]
    ├─ 일별/주별 비용 추이
    ├─ 품질 점수 트렌드
    ├─ 실패 트레이스 목록
    └─ 프롬프트 버전별 성능 비교

---

마무리

AI 에이전트 관측성의 핵심 원칙 세 가지예요.

첫째, 관측성 없는 AI 프로덕션은 없다. 트레이스가 있으면 디버깅이 3~5배 빨라요. 처음부터 넣으세요.

둘째, 비용은 반드시 추적해야 한다. LLM 비용은 예고 없이 폭발해요. 일별 임계값 알림 설정은 필수예요.

셋째, 모든 답변을 평가할 필요 없다. 5% 샘플링으로 LLM-as-Judge 자동 평가하고, 저점수만 사람이 검토해요. 😄

---