LangChainでLLMアプリケーションを本番運用する際、監視と告警は可用性を維持するために不可欠な要素です。本稿では、LangChainアプリケーションに堅牢な監視システムを導入し、HolySheep AI API（今すぐ登録）を活用した実践的な告警設定方法を解説します。
1. LangChain監視の重要性
LLMアプリケーションを本番環境にデプロイすると、以下のような問題が発生します：
- APIレイテンシ急上昇：通常100msが1秒以上に
- Rate Limit超過：429 Too Many Requestsエラー
- 認証失敗：401 Unauthorizedによるサービス停止
- トークン消費異常：予期せぬコスト増加
HolySheep AIは¥1=$1のレート（公式比85%節約）を提供し、<50msの低レイテンシを実現していますが、それでも監視なしでの運用は危険です。
2. LangChain監視の実装
2.1 Callbackを使った監視アーキテクチャ
LangChainのCallback機構を活用すると、LLM呼び出しの詳細をキャプチャできます。以下のコードは、HolySheep AI API向けの包括的な監視システムを実装しています：
"""
LangChain監視システム - HolySheep AI API対応
ファイル: langchain_monitor.py
"""
import os
import time
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field, asdict
from collections import defaultdict
import threading
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_core.messages import BaseMessage
HolySheep AI設定
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class LLMCallRecord:
"""LLM呼び出し記録"""
timestamp: str
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
latency_ms: float
status: str
error_message: Optional[str] = None
cost_usd: float = 0.0
class MonitoringCallback(BaseCallbackHandler):
"""LangChain監視用コールバック"""
def __init__(self):
self.records: List[LLMCallRecord] = []
self.metrics: Dict[str, Any] = defaultdict(list)
self.lock = threading.Lock()
# HolySheep AI 2026年価格表 (USD/1M tokens output)
self.price_table = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
"holy-default": 1.0 # デフォルト料金
}
def on_llm_start(
self,
serialized: Dict[str, Any],
prompts: List[str],
**kwargs: Any
) -> None:
"""LLM呼び出し開始時に実行"""
self.metrics["start_time"] = time.time()
self.metrics["last_prompt_length"] = len(prompts[0]) if prompts else 0
def on_llm_end(
self,
response: LLMResult,
**kwargs: Any
) -> None:
"""LLM呼び出し完了時に実行"""
start_time = self.metrics.pop("start_time", time.time())
latency_ms = (time.time() - start_time) * 1000
# トークン情報の取得
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
else:
prompt_tokens = completion_tokens = total_tokens = 0
# コスト計算
model_name = self._extract_model_name(response)
cost_usd = self._calculate_cost(model_name, completion_tokens)
# 記録の保存
record = LLMCallRecord(
timestamp=datetime.now().isoformat(),
model=model_name,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
latency_ms=round(latency_ms, 2),
status="success",
cost_usd=round(cost_usd, 6)
)
with self.lock:
self.records.append(record)
self.metrics["total_calls"] = len(self.records)
self.metrics["total_cost"] = sum(r.cost_usd for r in self.records)
def on_llm_error(
self,
error: Exception,
**kwargs: Any
) -> None:
"""LLM呼び出しエラー時に実行"""
start_time = self.metrics.pop("start_time", time.time())
latency_ms = (time.time() - start_time) * 1000
record = LLMCallRecord(
timestamp=datetime.now().isoformat(),
model="unknown",
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
latency_ms=round(latency_ms, 2),
status="error",
error_message=str(error)
)
with self.lock:
self.records.append(record)
def _extract_model_name(self, response: LLMResult) -> str:
"""モデル名の抽出"""
if response.llm_output and "model_name" in response.llm_output:
return response.llm_output["model_name"]
return "unknown"
def _calculate_cost(self, model: str, completion_tokens: int) -> float:
"""コスト計算（HolySheep AI料金体系）"""
price_per_million = self.price_table.get(
model.lower(),
self.price_table["holy-default"]
)
return (completion_tokens / 1_000_000) * price_per_million
def get_summary(self) -> Dict[str, Any]:
"""監視サマリーの取得"""
with self.lock:
if not self.records:
return {"message": "No records yet"}
successful = [r for r in self.records if r.status == "success"]
failed = [r for r in self.records if r.status == "error"]
successful_latencies = [r.latency_ms for r in successful]
return {
"total_calls": len(self.records),
"successful_calls": len(successful),
"failed_calls": len(failed),
"success_rate": len(successful) / len(self.records) * 100,
"avg_latency_ms": sum(successful_latencies) / len(successful_latencies) if successful_latencies else 0,
"min_latency_ms": min(successful_latencies) if successful_latencies else 0,
"max_latency_ms": max(successful_latencies) if successful_latencies else 0,
"total_cost_usd": sum(r.cost_usd for r in self.records),
"total_tokens": sum(r.total_tokens for r in self.records),
"error_rate": len(failed) / len(self.records) * 100
}
グローバル監視インスタンス
monitor = MonitoringCallback()
print("✅ LangChain監視システム初期化完了")
print(f"📡 APIエンドポイント: {HOLYSHEEP_BASE_URL}")2.2 AlertManagerによる異常検知
監視データに基づいて異常を検知し、通知を送るAlertManagerを実装します：
"""
LangChain AlertManager - 異常検知と通知
ファイル: alert_manager.py
"""
import os
import logging
from enum import Enum
from typing import Callable, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class Alert:
level: AlertLevel
title: str
message: str
timestamp: datetime
metadata: dict
class AlertRule:
"""告警ルール定義"""
def __init__(
self,
name: str,
condition: Callable[[dict], bool],
level: AlertLevel,
message_template: str
):
self.name = name
self.condition = condition
self.level = level
self.message_template = message_template
class AlertManager:
"""告警管理クラス"""
def __init__(self):
self.rules: list[AlertRule] = []
self.alerts: list[Alert] = []
self.logger = logging.getLogger(__name__)
# デフォルトルールの設定
self._setup_default_rules()
def _setup_default_rules(self):
"""デフォルト告警ルールの設定"""
# レイテンシ異常ルール（HolySheep AIの<50ms監視）
self.add_rule(AlertRule(
name="high_latency",
condition=lambda m: m.get("max_latency_ms", 0) > 5000, # 5秒超
level=AlertLevel.WARNING,
message_template="⚠️ レイテンシ異常: 最大 {max_latency_ms}ms (閾値: 5000ms)"
))
# エラー率上昇ルール
self.add_rule(AlertRule(
name="high_error_rate",
condition=lambda m: m.get("error_rate", 0) > 5.0, # 5%超
level=AlertLevel.CRITICAL,
message_template="🚨 エラー率急上昇: {error_rate}% (閾値: 5%)"
))
# コスト超過ルール（HolySheep AI ¥1=$1換算）
self.add_rule(AlertRule(
name="high_cost",
condition=lambda m: m.get("total_cost_usd", 0) > 100.0, # $100超
level=AlertLevel.WARNING,
message_template="💰 コスト警告: ${total_cost_usd} (日次上限: $100)"
))
# サービス停止ルール
self.add_rule(AlertRule(
name="service_down",
condition=lambda m: m.get("failed_calls", 0) > 10 and m.get("success_rate", 100) < 50,
level=AlertLevel.CRITICAL,
message_template="🔴 サービス異常: 成功率 {success_rate}%、失敗 {failed_calls}件"
))
# Rate Limit監視ルール
self.add_rule(AlertRule(
name="rate_limit_exceeded",
condition=lambda m: m.get("rate_limit_hits", 0) > 0,
level=AlertLevel.WARNING,
message_template="⏱️ Rate Limit超過: {rate_limit_hits}回"
))
def add_rule(self, rule: AlertRule):
"""告警ルールの追加"""
self.rules.append(rule)
self.logger.info(f"✅ 告警ルール追加: {rule.name}")
def evaluate(self, metrics: dict) -> list[Alert]:
"""メトリクスの評価と告警生成"""
new_alerts = []
for rule in self.rules:
try:
if rule.condition(metrics):
# フォーマット変数の置換
message = rule.message_template.format(**metrics)
alert = Alert(
level=rule.level,
title=f"[{rule.level.value.upper()}] {rule.name}",
message=message,
timestamp=datetime.now(),
metadata={"rule": rule.name, "metrics": metrics}
)
new_alerts.append(alert)
self.alerts.append(alert)
self.logger.warning(f"{alert.title}: {alert.message}")
except Exception as e:
self.logger.error(f"ルール評価エラー ({rule.name}): {e}")
return new_alerts
def get_recent_alerts(
self,
level: Optional[AlertLevel] = None,
minutes: int = 60
) -> list[Alert]:
"""最近の告警を取得"""
cutoff = datetime.now() - timedelta(minutes=minutes)
filtered = [a for a in self.alerts if a.timestamp > cutoff]
if level:
filtered = [a for a in filtered if a.level == level]
return filtered
def send_email_alert(self, alert: Alert, config: dict):
"""メールによる告警送信"""
try:
msg = MIMEMultipart()
msg['From'] = config['smtp_user']
msg['To'] = config['to_email']
msg['Subject'] = f"[{alert.level.value.upper()}] LLM Monitoring Alert"
body = f"""
LLM監視告警
レベル: {alert.level.value} 時刻: {alert.timestamp.isoformat()} タイトル: {alert.title}
メッセージ:
{alert.message}
"""
msg.attach(MIMEText(body, 'html'))
with smtplib.SMTP(config['smtp_host'], config['smtp_port']) as server:
server.starttls()
server.login(config['smtp_user'], config['smtp_password'])
server.send_message(msg)
self.logger.info(f"📧 メール告警送信完了: {alert.title}")
except Exception as e:
self.logger.error(f"メール送信失敗: {e}")
AlertManagerインスタンス生成
alert_manager = AlertManager()
print("✅ AlertManager初期化完了")
print(f"📋 設定済みルール数: {len(alert_manager.rules)}")3. HolySheep AI API統合監視システム
HolySheep AIのAPIを呼び出すLangChainチェーンに監視を統合します。以下のコードは、実際のAPI呼び出しで発生しうるエラーを適切に処理しながら監視を続けます：
"""
HolySheep AI API統合監視システム
ファイル: holysheep_monitoring_app.py
"""
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
環境変数読み込み
load_dotenv()
HolySheep AI設定（¥1=$1レート、85%節約）
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1" # HolySheep AI公式エンドポイント
監視システムインポート
from langchain_monitor import MonitoringCallback, monitor
from alert_manager import AlertManager, AlertLevel
class HolySheepLLMApplication:
"""HolySheep AI LLMアプリケーション"""
def __init__(self):
self.alert_manager = AlertManager()
self._setup_llm()
self._setup_chain()
def _setup_llm(self):
"""LLM初期化（HolySheep AI API使用）"""
self.llm = ChatOpenAI(
model="deepseek-v3.2", # $0.42/MTok - 最低コスト
base_url=BASE_URL,
api_key=HOLYSHEEP_API_KEY,
timeout=60.0,
max_retries=3,
callbacks=[monitor] # 監視コールバック登録
)
def _setup_chain(self):
"""チェーン設定"""
prompt = ChatPromptTemplate.from_messages([
("system", "あなたは役立つAIアシスタントです。"),
("user", "{question}")
])
self.chain = prompt | self.llm | StrOutputParser()
def invoke(self, question: str) -> str:
"""チェーン実行"""
try:
response = self.chain.invoke({"question": question})
# 実行後にメトリクス評価
metrics = monitor.get_summary()
alerts = self.alert_manager.evaluate(metrics)
return response
except Exception as e:
# エラーを監視システムに報告
monitor.metrics["last_error"] = str(e)
# 接続エラー処理
if "ConnectionError" in str(type(e).__name__) or "timeout" in str(e).lower():
raise ConnectionError(f"API接続タイムアウト: {e}")
# 認証エラー処理
if "401" in str(e) or "Unauthorized" in str(e):
raise PermissionError(f"API認証失敗: {e}")
# Rate Limit処理
if "429" in str(e) or "rate limit" in str(e).lower():
raise Exception(f"Rate Limit超過: {e}")
raise
def health_check(self) -> dict:
"""ヘルスチェック"""
metrics = monitor.get_summary()
return {
"status": "healthy" if metrics.get("error_rate", 0) < 5 else "degraded",
"latency_ms": metrics.get("avg_latency_ms", 0),
"success_rate": metrics.get("success_rate", 100),
"total_cost": metrics.get("total_cost_usd", 0)
}
アプリケーションインスタンス生成
app = HolySheepLLMApplication()
テスト実行
if __name__ == "__main__":
print("🏥 HolySheep AI監視システム テスト")
# 正常系テスト
try:
response = app.invoke("LangChainとは何ですか？")
print(f"✅ 応答: {response[:100]}...")
except Exception as e:
print(f"❌ エラー: {e}")
# ヘルスチェック
health = app.health_check()
print(f"\n📊 ヘルスチェック結果:")
print(f" ステータス: {health['status']}")
print(f" 平均レイテンシ: {health['latency_ms']:.2f}ms")
print(f" 成功率: {health['success_rate']:.1f}%")
print(f" 累計コスト: ${health['total_cost']:.4f}")4. 実践的な監視ダッシュボード
Prometheus + Grafanaを用いた監視ダッシュボードの設定例を示します：
# docker-compose.yml - 監視スタック
version: '3.8'
services:
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
volumes:
- grafana_data:/var/lib/grafana
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
alertmanager:
image: prom/alertmanager:latest
ports:
- "9093:9093"
volumes:
- ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
command:
- '--config.file=/etc/alertmanager/alertmanager.yml'# prometheus.yml
global:
scrape_interval: 15s
rule_files:
- "llm_alerts.rules"
alerting:
alertmanagers:
- static_configs:
- targets:
- alertmanager:9093
scrape_configs:
- job_name: 'holysheep-llm'
static_configs:
- targets: ['host.docker.internal:8000']
metrics_path: '/metrics'よくあるエラーと対処法
| エラータイプ | 原因 | 解決方法 |
|---|---|---|
ConnectionError: timeout |
APIエンドポイントへの接続失敗 または 応答遅延 | |
401 Unauthorized |
APIキーが無効または期限切れ | |
429 Too Many Requests |
Rate Limit超過（HolySheep AIの制限に達した） | |
JSONDecodeError |
API応答の形式が不正 | |
5. 監視ベストプラクティス
5.1 指標の収集と保存
HolySheep AIの¥1=$1コスト優位性を最大化するため、トークン消費をリアルタイムで追跡することが重要です：
- レイテンシ：P50、P95、P99パーセンタイルを追跡
- コスト：リアルタイムコスト計算（DeepSeek V3.2: $0.42/MTok）
- エラー率：5%超でWarning、10%超でCritical
- Token使用量：日次・月次での集計
5.2 自動復旧メカニズム
監視システムと連携した自動復旧を実装することで、MTTR（Mean Time To Recovery）を短縮できます：
# 自動復旧システム
class AutoRecoverySystem:
def __init__(self, max_consecutive_failures=5):
self.failures = 0
self.max_failures = max_consecutive_failures
def handle_failure(self, error: Exception):
"""障害発生時の自動処理"""
self.failures += 1
if self.failures >= self.max_failures:
print("🔄 自動復旧シーケンス開始")
# 1. キャッシュクリア
# cache.clear()
# 2. 接続プール再生成
# self.recreate_connections()
# 3. Fallback LLMへの切り替え
# self.switch_to_fallback_model()
# 4. 監視ダッシュボードに通知
# notify_dashboard("auto_recovery_triggered")
self.failures = 0
print("✅ 自動復旧完了")
def handle_success(self):
"""成功時のカウンタリセット"""
if self.failures > 0:
self.failures = 0まとめ
LangChainアプリケーションの本番運用において、監視と告警は可用性とコスト最適化の両面で重要な役割を果たします。本稿で解説した監視システムを導入することで、HolySheep AIの¥1=$1レートと50ms未満の低レイテンシという利点を最大限に活かしたLLMOpsを実現できます。
監視ダッシュボード、成本アラート、異常検知を組み合わせることで、問題の早期発見と自動復旧が可能となり、SREチーム，工科大学，研究者，工科大学関係者が安心してLLMアプリケーションを運用できるようになります。