hermes-agent × ELK で構築する LLM 異常検知とエンドツーエンド分散トレーシング

実装 1：hermes-agent のリクエストロギングミドルウェア

python
middleware/logging_mw.py
import time, uuid, json, logging
from typing import Callable

logger = logging.getLogger("hermes.access")

class TraceLoggingMiddleware:
    def __init__(self, service_name: str = "hermes-agent"):
        self.service_name = service_name

    async def __call__(self, request, call_next: Callable):
        trace_id = request.headers.get("x-trace-id") or str(uuid.uuid4())
        start = time.perf_counter()
        request.state.trace_id = trace_id

        log_entry = {
            "trace_id": trace_id,
            "service": self.service_name,
            "method": request.method,
            "path": request.url.path,
            "user_id": request.headers.get("x-user-id"),
            "ts_start_ms": int(start * 1000),
        }

        try:
            response = await call_next(request)
            elapsed_ms = (time.perf_counter() - start) * 1000
            log_entry.update({
                "status": response.status_code,
                "elapsed_ms": round(elapsed_ms, 2),
                "llm_calls": getattr(request.state, "llm_calls", []),
                "ts_end_ms": int(time.perf_counter() * 1000),
            })
            return response
        except Exception as e:
            log_entry.update({
                "status": 500,
                "error": str(e),
                "error_type": type(e).__name__,
                "elapsed_ms": round((time.perf_counter() - start) * 1000, 2),
            })
            raise
        finally:
            logger.info(json.dumps(log_entry, ensure_ascii=False))

python
middleware/llm_wrapper.py
import os, time
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["YOUR_HOLYSHEEP_API_KEY"],
    base_url="https://api.holysheep.ai/v1",
)

PRICE_OUT_USD_PER_MTOK = {
    "gpt-4.1": 8.00,
    "claude-sonnet-4.5": 15.00,
    "gemini-2.5-flash": 2.50,
    "deepseek-v3.2": 0.42,
}

class TracedLLM:
    def __init__(self, trace_collector):
        self.trace = trace_collector

    async def chat(self, model: str, messages, **kwargs):
        t0 = time.perf_counter()
        resp = await client.chat.completions.create(
            model=model, messages=messages, **kwargs
        )
        latency_ms = (time.perf_counter() - t0) * 1000

        usage = resp.usage
        cost_usd = (usage.completion_tokens / 1_000_000) * PRICE_OUT_USD_PER_MTOK[model]

        self.trace.record_llm_call({
            "trace_id": self.trace.current_trace_id,
            "model": model,
            "prompt_tokens": usage.prompt_tokens,
            "completion_tokens": usage.completion_tokens,
            "cost_usd": round(cost_usd, 6),
            "latency_ms": round(latency_ms, 2),
            "finish_reason": resp.choices[0].finish_reason,
        })
        return resp

conf
/etc/logstash/conf.d/hermes.conf
input {
  beats { port => 5044 }
}

filter {
  if [fields][service] == "hermes-agent" {
    json {
      source => "message"
      target => "parsed"
      remove_field => ["message"]
    }

    ruby {
      code => '
        parsed = event.get("parsed") || {}
        parsed.each { |k, v| event.set(k, v) unless event.get(k) }
        llm_calls = event.get("llm_calls") || []
        total = 0.0
        llm_calls.each { |c| total += c["cost_usd"].to_f }
        event.set("total_cost_usd", total.round(6))
      '
    }

    if [elapsed_ms] > 3000 {
      mutate { add_tag => ["slow_request"] }
    }
    if [status] >= 500 {
      mutate { add_tag => ["server_error"] }
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://es-node-1:9200"]
    index => "hermes-agent-%{+YYYY.MM.dd}"
  }
}

yaml
/etc/elastalert/rules/openai_claude_anomaly.yaml
name: openai_claude_anomaly
type: any
index: hermes-agent-*

filter:
  - terms:
      llm_calls.model: ["gpt-4.1", "claude-sonnet-4.5"]

異常 1: レイテンシ p95 が急上昇
- metric_aggregation:
    metric_agg: "percentiles"
    field: "llm_calls.latency_ms"
    percentile: 95
    min_doc_count: 50
    query_key: "llm_calls.model"
    bucket_interval: "1m"
    sync_interval: "1m"
    alert:
      - slack

異常 2: 5xx エラー率 10% 超
- percentage:
    bucket_interval: "1m"
    sync_interval: "1m"
    min_window: 5
    query:
      query_string:
        query: "status:[500 TO 599]"
    percentage_range: 10

alert:
  slack:
    slack_webhook_url: "https://hooks.slack.com/services/XXX"
    slack_msg_color: danger

alert_subject: "🚨 hermes-agent LLM 異常検知 ({0})"
alert_text: |-
  モデル: {0}
  p95 レイテンシ: {1}ms
  エラー率: {2}%
  5 分間のリクエスト: {3} 件
  直近の trace_id: {4}