필사 모드: 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. 用 LangSmith 包装 OpenAI 客户端(自动追踪)
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:
    """从向量数据库检索相关文档"""
    # 实际实现中会用到 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 装饰器会自动把函数的输入输出、
    执行时间记录为一条 Trace。
    """
    # 加载对话历史(自动生成子 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 或数据库查询对话历史
    # ...
    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

# 各模型的 token 价格(截至 2026 年 3 月，每 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}")