LLM可観測性2.0：ログ・トレース・アラート設計

┌─────────────────────────────────────────────────────┐
│                LLMアプリ (Python)                     │
│                                                      │
│  ┌──────────┐    ┌──────────┐    ┌──────────────┐  │
│  │ ユーザー  │    │  LLM Chain│    │ ツール呼び出し│  │
│  │ リクエスト│───▶│ (LangChain│───▶│  (検索/DB等) │  │
│  └──────────┘    │  /LlamaI.) └──────────────┘  │
│                   └────┬─────┘                      │
│                        │ OpenTelemetry Span          │
└────────────────────────┼────────────────────────────┘
                         │ OTLP (gRPC/HTTP)
          ┌──────────────┼──────────────────┐
          ▼              ▼                  ▼
    LangSmith       Langfuse           Grafana/Tempo
    (品質分析)     (コスト追跡)        (インフラ統合)

Python 3.11+
opentelemetry-sdk >= 1.24.0
opentelemetry-exporter-otlp-proto-grpc >= 1.24.0
opentelemetry-instrumentation-httpx >= 0.45b0

# otel_setup.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME

def setup_tracer(service_name: str, otlp_endpoint: str = "http://localhost:4317") -> trace.Tracer:
    """
    OpenTelemetry TracerProviderを初期化してグローバルに設定する。
    
    Args:
        service_name: サービス識別名（例: "llm-chat-api"）
        otlp_endpoint: OTLPコレクターのエンドポイント
    """
    resource = Resource(attributes={
        SERVICE_NAME: service_name,
        "deployment.environment": "production",
        "ai.framework": "langchain",  # gen-ai semconv に準拠
    })
    
    exporter = OTLPSpanExporter(
        endpoint=otlp_endpoint,
        insecure=False,  # 本番では必ずTLS
    )
    
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    
    return trace.get_tracer(service_name)

# llm_client.py
import time
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage

tracer = trace.get_tracer("llm-chat-api")

def chat_with_tracing(user_message: str, model: str = "gpt-4o") -> str:
    """
    LLMリクエストにOpenTelemetryスパンを付与する。
    gen-ai semconv (v1.28) の属性名を使用。
    参考: https://opentelemetry.io/docs/specs/semconv/gen-ai/
    """
    with tracer.start_as_current_span("llm.chat") as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.request.temperature", 0.7)
        
        start_time = time.monotonic()
        
        try:
            llm = ChatOpenAI(model=model, temperature=0.7)
            response = llm.invoke([HumanMessage(content=user_message)])
            
            latency_ms = (time.monotonic() - start_time) * 1000
            
            # gen-ai semconv のトークン属性
            span.set_attribute("gen_ai.usage.input_tokens", 
                               response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0))
            span.set_attribute("gen_ai.usage.output_tokens",
                               response.response_metadata.get("token_usage", {}).get("completion_tokens", 0))
            span.set_attribute("llm.latency_ms", latency_ms)
            span.set_status(Status(StatusCode.OK))
            
            return response.content
            
        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise

# .env（git管理外に置くこと）
LANGCHAIN_TRACING_V2=true
LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
LANGCHAIN_API_KEY="lsv2_pt_xxxxxxxxxxxxxxxx"
LANGCHAIN_PROJECT="production-llm-app"

# langsmith_callback.py
from langsmith import Client
from langsmith.run_helpers import traceable
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import os

client = Client()

@traceable(
    project_name=os.getenv("LANGCHAIN_PROJECT", "default"),
    tags=["production", "chat-endpoint"],
)
def traced_chat(user_message: str, session_id: str) -> dict:
    """
    LangSmithにトレースを送信する関数。
    @traceable デコレーターで自動的にrun_idが付与される。
    """
    llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
    response = llm.invoke([HumanMessage(content=user_message)])
    
    return {
        "output": response.content,
        "session_id": session_id,
        # カスタムメタデータ（LangSmith UIで検索・フィルター可能）
        "metadata": {
            "user_segment": "premium",
            "feature_flag": "v2-prompt",
        }
    }

# langfuse_otel_setup.py
import base64
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME

def setup_langfuse_tracer(service_name: str) -> trace.Tracer:
    """
    LangfuseのOTLPエンドポイントにトレースを送信する設定。
    Langfuseはベーシック認証をAuthorizationヘッダーで受け取る。
    参考: https://langfuse.com/docs/integrations/opentelemetry/python-otel
    """
    public_key = os.environ["LANGFUSE_PUBLIC_KEY"]
    secret_key = os.environ["LANGFUSE_SECRET_KEY"]
    host = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com")
    
    # Basic認証ヘッダーを生成
    auth_token = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
    
    exporter = OTLPSpanExporter(
        endpoint=f"{host}/api/public/otel/v1/traces",
        headers={"Authorization": f"Basic {auth_token}"},
    )
    
    resource = Resource(attributes={SERVICE_NAME: service_name})
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    
    return trace.get_tracer(service_name)

# langfuse_cost_tracking.py
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage

langfuse = Langfuse()

@observe()  # Langfuseの自動トレースデコレーター
def generate_response(prompt: str, model: str = "gpt-4o") -> str:
    """
    @observe デコレーターでLangfuseにトレースを自動送信する。
    トークン消費量はopenai_model_usageから自動取得される。
    """
    langfuse_context.update_current_observation(
        name="chat-generation",
        input=prompt,
        metadata={"model": model, "environment": "production"},
    )
    
    llm = ChatOpenAI(model=model)
    response = llm.invoke([HumanMessage(content=prompt)])
    
    langfuse_context.update_current_observation(
        output=response.content,
        usage={
            "input": response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0),
            "output": response.response_metadata.get("token_usage", {}).get("completion_tokens", 0),
        }
    )
    
    return response.content

# llm-alerts.yaml
groups:
  - name: llm_observability
    interval: 60s
    rules:
      # === レイテンシアラート ===
      - alert: LLMHighLatency
        expr: |
          histogram_quantile(0.95,
            rate(llm_request_duration_ms_bucket[5m])
          ) > 5000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLMレスポンスのP95レイテンシが5秒を超えています"
          description: "{{ $labels.service }} のP95レイテンシ: {{ $value }}ms"

      # === コストアラート ===
      - alert: LLMTokenCostSpike
        expr: |
          rate(llm_token_cost_usd_total[1h]) * 3600 > 10.0
        for: 15m
        labels:
          severity: critical
        annotations:
          summary: "LLMトークンコストが時間あたり$10を超えています"
          description: "現在のコスト予測: ${{ $value }}/hour"

      # === 品質アラート ===
      - alert: LLMQualityScoreDrop
        expr: |
          avg_over_time(llm_quality_score[30m]) < 0.65
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "LLM品質スコアが閾値を下回っています"
          description: "30分平均スコア: {{ $value }} (閾値: 0.65)"

# pii_sanitizer.py
import re
from typing import Optional

def sanitize_prompt_for_logging(prompt: str) -> str:
    """
    ログに記録する前にPIIをマスクする。
    経験則: email、電話番号、クレジットカード番号を対象とする。
    """
    # メールアドレスのマスク
    prompt = re.sub(r'[\w\.-]+@[\w\.-]+\.\w+', '[EMAIL]', prompt)
    # 電話番号のマスク（日本形式）
    prompt = re.sub(r'\d{2,4}-\d{2,4}-\d{4}', '[PHONE]', prompt)
    # クレジットカード番号のマスク
    prompt = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', prompt)
    
    return prompt

# sampling_setup.py
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased, ParentBased

# 本番: 10%サンプリング（公式推奨: 1-10%）
# 参考: https://opentelemetry.io/docs/concepts/sampling/
sampler = ParentBased(root=TraceIdRatioBased(0.1))

# LangSmithのみ使う場合
LANGCHAIN_TRACING_V2=true
# OTELは使わない（OTEL_EXPORTER_OTLP_ENDPOINTを設定しない）

# OTELのみ使う場合
LANGCHAIN_TRACING_V2=false
OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4317