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}")