LLM 프로덕션 모니터링 플랫폼 비교: LangSmith·LangFuse·Arize Phoenix 실전 운영 가이드

import os
import openai
from langsmith import traceable, Client
from langsmith.wrappers import wrap_openai

# 1. 환경 변수 설정
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_xxxxxxxxxxxx"
os.environ["LANGSMITH_PROJECT"] = "production-chatbot"
os.environ["LANGSMITH_ENDPOINT"] = "https://api.smith.langchain.com"

# 2. OpenAI 클라이언트를 LangSmith로 래핑 (자동 트레이싱)
client = wrap_openai(openai.Client())

# 3. @traceable 데코레이터로 커스텀 함수 트레이싱
@traceable(
    name="RAGPipeline",
    run_type="chain",
    tags=["production", "rag"],
    metadata={"version": "2.1.0"}
)
def rag_pipeline(user_query: str) -> dict:
    """RAG 파이프라인: 검색 -> 컨텍스트 구성 -> LLM 호출"""

    # 검색 단계 (자동으로 하위 Span으로 기록)
    context_docs = retrieve_documents(user_query)

    # 프롬프트 구성
    system_prompt = """당신은 기술 문서 기반 Q&A 어시스턴트입니다.
    주어진 컨텍스트만을 바탕으로 답변하세요.
    컨텍스트에 없는 내용은 "해당 정보를 찾을 수 없습니다"라고 답하세요."""

    context_text = "\n\n".join([doc["content"] for doc in context_docs])
    user_message = f"컨텍스트:\n{context_text}\n\n질문: {user_query}"

    # LLM 호출 (wrap_openai로 자동 트레이싱)
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message}
        ],
        temperature=0.1,
        max_tokens=1024
    )

    result = {
        "answer": response.choices[0].message.content,
        "sources": [doc["source"] for doc in context_docs],
        "model": "gpt-4o",
        "token_usage": {
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
            "total": response.usage.total_tokens
        }
    }

    return result


@traceable(name="DocumentRetrieval", run_type="retriever")
def retrieve_documents(query: str) -> list:
    """벡터 DB에서 관련 문서 검색"""
    # 실제 구현에서는 Pinecone, Weaviate 등 사용
    # 여기서는 예시를 위해 간소화
    from langsmith import get_current_run_tree
    run = get_current_run_tree()
    run.metadata["retriever_type"] = "pinecone"
    run.metadata["top_k"] = 5

    # ... 벡터 검색 로직 ...
    return [{"content": "검색된 문서 내용", "source": "docs/guide.md", "score": 0.95}]


# 4. LangSmith 클라이언트로 피드백 기록
ls_client = Client()

def record_user_feedback(run_id: str, score: float, comment: str = ""):
    """사용자 피드백을 LangSmith에 기록"""
    ls_client.create_feedback(
        run_id=run_id,
        key="user_satisfaction",
        score=score,          # 0.0 ~ 1.0
        comment=comment
    )

# 5. 실행
if __name__ == "__main__":
    result = rag_pipeline("Kubernetes Pod의 OOMKill 원인은?")
    print(f"답변: {result['answer']}")
    print(f"토큰 사용: {result['token_usage']}")

import os
from langfuse import observe, get_client
from langfuse.openai import openai  # LangFuse 래핑된 OpenAI

# 1. 환경 변수 설정 (셀프 호스팅 시 LANGFUSE_HOST 변경)
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxxxxxxxxxxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxxxxxxxxxxx"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"  # 셀프 호스팅: http://localhost:3000

# 2. @observe 데코레이터로 트레이싱
@observe()
def chatbot_pipeline(user_message: str, session_id: str) -> dict:
    """
    LangFuse @observe 데코레이터는 함수의 입출력, 실행 시간을
    자동으로 트레이스로 기록한다.
    """
    # 대화 이력 로드 (자동으로 하위 Span 생성)
    history = load_conversation_history(session_id)

    # LLM 호출 (langfuse.openai 모듈 사용 시 자동 트레이싱)
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "당신은 친절한 고객 지원 챗봇입니다."},
            *history,
            {"role": "user", "content": user_message}
        ],
        temperature=0.3,
        langfuse_prompt_name="customer-support-v3",  # 프롬프트 버전 추적
    )

    answer = response.choices[0].message.content

    # 품질 점수 기록
    langfuse_client = get_client()
    langfuse_client.score(
        name="relevance",
        value=evaluate_relevance(user_message, answer),
        comment="자동 평가"
    )

    return {
        "answer": answer,
        "session_id": session_id,
        "tokens": response.usage.total_tokens
    }


@observe()
def load_conversation_history(session_id: str) -> list:
    """세션별 대화 이력 로드"""
    # Redis 또는 DB에서 대화 이력 조회
    # ...
    return []


@observe()
def evaluate_relevance(question: str, answer: str) -> float:
    """LLM-as-a-Judge로 관련성 평가"""
    judge_response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""다음 질문에 대한 답변의 관련성을 0.0~1.0 사이 숫자로만 평가하세요.
            질문: {question}
            답변: {answer}
            점수:"""
        }],
        temperature=0.0,
        max_tokens=5
    )
    try:
        return float(judge_response.choices[0].message.content.strip())
    except ValueError:
        return 0.5


# 셀프 호스팅 Docker Compose 실행:
# git clone https://github.com/langfuse/langfuse.git
# cd langfuse
# docker compose up -d

import os
import phoenix as px
from phoenix.otel import register
from openinference.instrumentation.openai import OpenAIInstrumentor
from openai import OpenAI

# 1. Phoenix 서버 실행 (로컬 개발 시)
# px.launch_app()  # 로컬에서 Phoenix UI 실행 (http://localhost:6006)

# 2. OpenTelemetry 기반 트레이싱 설정
tracer_provider = register(
    project_name="production-chatbot",
    endpoint="http://phoenix-server:6006/v1/traces",  # Phoenix 서버 엔드포인트
)

# 3. OpenAI Instrumentor 활성화 (자동 트레이싱)
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

# 4. 일반적인 OpenAI 호출 - 자동으로 트레이스 수집
client = OpenAI()

def generate_summary(document: str) -> dict:
    """문서 요약 생성 - Phoenix가 자동으로 트레이싱"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "주어진 문서를 3줄로 요약하세요."},
            {"role": "user", "content": document}
        ],
        temperature=0.2
    )

    return {
        "summary": response.choices[0].message.content,
        "input_tokens": response.usage.prompt_tokens,
        "output_tokens": response.usage.completion_tokens,
    }

# 5. Phoenix Evaluation: 환각 탐지
from phoenix.evals import (
    HallucinationEvaluator,
    OpenAIModel,
    run_evals,
)

# 평가 모델 설정
eval_model = OpenAIModel(model="gpt-4o-mini")

# 환각 평가기
hallucination_eval = HallucinationEvaluator(eval_model)

# 데이터셋 기반 평가 실행
# Phoenix UI에서 수집된 트레이스를 DataFrame으로 내보낸 뒤 평가
import pandas as pd

eval_df = pd.DataFrame({
    "input": ["Kubernetes의 최대 Pod 수는?"],
    "output": ["Kubernetes 클러스터당 최대 150,000개의 Pod를 지원합니다."],
    "reference": ["기본 설정에서 노드당 최대 110개의 Pod, 클러스터당 최대 150,000개의 Pod를 지원합니다."]
})

# 환각 평가 실행
hallucination_results = run_evals(
    dataframe=eval_df,
    evaluators=[hallucination_eval],
    provide_explanation=True
)
print(hallucination_results)

import json
from langfuse import observe
from langfuse.openai import openai
from typing import Literal

# 평가 기준 정의
EVAL_CRITERIA = {
    "relevance": "질문과 답변의 관련성 (0.0~1.0)",
    "completeness": "답변의 완전성 - 필요한 정보를 모두 포함하는가 (0.0~1.0)",
    "faithfulness": "주어진 컨텍스트에 충실한가 (0.0~1.0)",
    "conciseness": "불필요한 내용 없이 간결한가 (0.0~1.0)"
}

@observe(name="AutoEvalPipeline")
def auto_evaluate(
    question: str,
    answer: str,
    context: str = "",
    criteria: list[str] = None
) -> dict:
    """
    LLM-as-a-Judge 자동 평가 파이프라인.
    여러 품질 기준에 대해 동시에 평가한다.
    """
    if criteria is None:
        criteria = list(EVAL_CRITERIA.keys())

    criteria_text = "\n".join([
        f"- {name}: {desc}" for name, desc in EVAL_CRITERIA.items()
        if name in criteria
    ])

    eval_prompt = f"""당신은 LLM 응답 품질 평가 전문가입니다.
다음 질문-답변 쌍을 평가 기준에 따라 채점하세요.

## 질문
{question}

## 컨텍스트 (제공된 경우)
{context if context else "없음"}

## 답변
{answer}

## 평가 기준
{criteria_text}

## 출력 형식 (JSON)
각 기준에 대해 score(0.0~1.0)와 reasoning(한국어 1문장)을 포함하세요.
```json
{{
  "scores": {{
    "기준명": {{"score": 0.0, "reasoning": "이유"}}
  }},
  "overall_score": 0.0,
  "summary": "전체 평가 요약 (한국어)"
}}
```"""

    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": eval_prompt}],
        temperature=0.0,
        response_format={"type": "json_object"}
    )

    eval_result = json.loads(response.choices[0].message.content)

    # LangFuse에 평가 점수 기록
    from langfuse import get_client
    lf_client = get_client()
    for criterion, data in eval_result.get("scores", {}).items():
        lf_client.score(
            name=f"eval_{criterion}",
            value=data["score"],
            comment=data["reasoning"]
        )

    return eval_result


# CI/CD에서의 평가 게이트: 새 프롬프트 배포 전 자동 평가
def prompt_deployment_gate(
    new_prompt: str,
    test_dataset: list[dict],
    min_overall_score: float = 0.7
) -> bool:
    """
    새 프롬프트의 품질이 기준을 충족하는지 확인하는 배포 게이트.
    CI/CD 파이프라인에서 호출된다.
    """
    scores = []
    for test_case in test_dataset:
        # 새 프롬프트로 응답 생성
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": new_prompt},
                {"role": "user", "content": test_case["question"]}
            ],
            temperature=0.1
        )
        answer = response.choices[0].message.content

        # 자동 평가 실행
        eval_result = auto_evaluate(
            question=test_case["question"],
            answer=answer,
            context=test_case.get("context", "")
        )
        scores.append(eval_result["overall_score"])

    avg_score = sum(scores) / len(scores) if scores else 0
    passed = avg_score >= min_overall_score

    print(f"평가 결과: 평균 {avg_score:.2f} / 기준 {min_overall_score}")
    print(f"배포 게이트: {'PASS' if passed else 'FAIL'}")

    return passed

from dataclasses import dataclass
from datetime import datetime, timedelta
from collections import defaultdict

# 모델별 토큰 가격 (2026년 3월 기준, USD per 1M tokens)
MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4-turbo": {"input": 10.00, "output": 30.00},
    "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
    "claude-3-5-haiku": {"input": 0.80, "output": 4.00},
}

@dataclass
class UsageRecord:
    timestamp: datetime
    model: str
    input_tokens: int
    output_tokens: int
    endpoint: str
    user_id: str = ""

class LLMCostTracker:
    """LLM 비용 추적 및 예산 알림"""

    def __init__(self, monthly_budget_usd: float = 5000.0):
        self.monthly_budget = monthly_budget_usd
        self.records: list[UsageRecord] = []

    def record_usage(self, record: UsageRecord):
        self.records.append(record)
        cost = self._calculate_cost(record)

        # 예산 초과 경고 체크
        monthly_cost = self.get_monthly_cost()
        budget_pct = (monthly_cost / self.monthly_budget) * 100

        if budget_pct > 90:
            self._send_alert(
                f"LLM 비용 경고: 월 예산의 {budget_pct:.1f}% 소진 "
                f"(${monthly_cost:.2f} / ${self.monthly_budget:.2f})"
            )

        return cost

    def _calculate_cost(self, record: UsageRecord) -> float:
        pricing = MODEL_PRICING.get(record.model, {"input": 0, "output": 0})
        input_cost = (record.input_tokens / 1_000_000) * pricing["input"]
        output_cost = (record.output_tokens / 1_000_000) * pricing["output"]
        return input_cost + output_cost

    def get_monthly_cost(self) -> float:
        now = datetime.now()
        month_start = now.replace(day=1, hour=0, minute=0, second=0)
        monthly_records = [r for r in self.records if r.timestamp >= month_start]
        return sum(self._calculate_cost(r) for r in monthly_records)

    def get_cost_breakdown(self) -> dict:
        """모델별, 엔드포인트별 비용 분석"""
        by_model = defaultdict(float)
        by_endpoint = defaultdict(float)

        for record in self.records:
            cost = self._calculate_cost(record)
            by_model[record.model] += cost
            by_endpoint[record.endpoint] += cost

        return {
            "by_model": dict(by_model),
            "by_endpoint": dict(by_endpoint),
            "total": sum(by_model.values())
        }

    def _send_alert(self, message: str):
        # Slack 또는 PagerDuty로 알림 전송
        print(f"[ALERT] {message}")