プロダクション環境でLLM应用を構築する際、最大の問題の一つが呼び出しの可视化管理です。「APIがタイムアウトしたけど原因がわからない」「トークン消費をリアルタイムで追踪したい」「エラー発生時に即座にアラートを受け取りたい」——这样的需求在实际开发中极为普遍。
本稿では、LangChainのコールバック機構を活用した実践的なモニタリング・ログ記録の手法を、HolySheep AIを活用した具体的な実装例と共に解説します。
问题场景：错误日志缺失导致的障害
私の実際のプロジェクトでは、プロダクション環境にLLM統合をデプロイした直後、次のエラーに遭遇しました：
ConnectionError: HTTPSConnectionPool(host='api.openai.com', port=443):
Max retries exceeded with url: /v1/chat/completions
(Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7f...>:
Failed to establish a new connection: [Errno 110] Connection timed out'))
TimeoutError: Request timed out after 30 seconds
Total tokens: unknown | Prompt tokens: unknown | Completion tokens: unknown
このエラーから何も情報が得られず、どこで何が起きているのか全く追踪できませんでした。LangChainのコールバック機構を活用することで、このような地狱から解放されました。
LangChainコールバック机制详解
LangChainはBaseCallbackHandler抽象クラスを中心とした灵活的回调系统を持ちます。これにより、LLM调用の全过程をモニタリングできます。
基础回调处理器实现
import os
from datetime import datetime
from typing import Any, Dict, List, Optional
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult
from langchain_openai import ChatOpenAI
HolySheep AI設定
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
class ComprehensiveLoggingHandler(BaseCallbackHandler):
"""
包括 토큰消费、延迟测量、错误追踪的综合回调处理器
対応モデル: GPT-4.1 $8/MTok、Claude Sonnet 4.5 $15/MTok、Gemini 2.5 Flash $2.50/MTok
"""
def __init__(self, log_file: str = "llm_calls.log"):
self.log_file = log_file
self.call_count = 0
self.total_tokens = 0
self.total_cost = 0.0
# 各モデルの价格（$/MTok）
self.model_prices = {
"gpt-4": 30.0,
"gpt-4-turbo": 10.0,
"gpt-3.5-turbo": 2.0,
"gpt-4.1": 8.0,
"claude-3-sonnet": 15.0,
"claude-3-opus": 75.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3": 0.42,
}
def _log(self, message: str, level: str = "INFO"):
timestamp = datetime.now().isoformat()
log_entry = f"[{timestamp}] [{level}] {message}\n"
with open(self.log_file, "a", encoding="utf-8") as f:
f.write(log_entry)
print(log_entry.strip())
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs
) -> None:
self.call_count += 1
model = serialized.get("name", "unknown")
self._log(f"LLM调用开始 #{self.call_count} | 模型: {model} | プロンプト長: {len(prompts[0])}文字")
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
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", prompt_tokens + completion_tokens)
self.total_tokens += total_tokens
# 模型价格计算（支持HolySheep的全部モデル）
model = response.llm_output.get("model_name", "gpt-4")
price_per_mtok = self.model_prices.get(model, 30.0)
cost = (total_tokens / 1_000_000) * price_per_mtok
self.total_cost += cost
self._log(
f"LLM调用完成 | "
f"Prompt: {prompt_tokens}tok | "
f"Completion: {completion_tokens}tok | "
f"合計: {total_tokens}tok | "
f"コスト: ${cost:.4f} | "
f"累計コスト: ${self.total_cost:.4f}"
)
def on_llm_error(self, error: Exception, **kwargs) -> None:
error_type = type(error).__name__
error_message = str(error)
self._log(
f"LLMエラー発生 | 型: {error_type} | メッセージ: {error_message}",
level="ERROR"
)
实际应用例
handler = ComprehensiveLoggingHandler("production_llm.log")
llm = ChatOpenAI(
model="gpt-4.1",
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_API_BASE"],
callbacks=[handler],
request_timeout=60
)
response = llm.invoke("Hello, explain LangChain callbacks in Japanese.")
print(f"回应: {response.content}")
高级监控：自定义中间件与实时アラート
先ほどの基本ハンドラーに加え、プロダクション環境ではリアルタイム监控とアラートが不可欠です。以下は、WeChat Webhook統合と自定义アラート条件を持つ高级実装です：
import json
import time
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Optional
import requests
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
@dataclass
class MonitoringMetrics:
"""监控指标聚合格类"""
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
total_tokens: int = 0
total_latency_ms: float = 0.0
error_counts: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
latency_by_model: Dict[str, list] = field(default_factory=lambda: defaultdict(list))
class ProductionMonitoringHandler(BaseCallbackHandler):
"""
HolySheep AI対応 プロダクション级监控处理器
特点：
- 实时延迟追踪（<50msレイテンシ目标监控）
- エラー率アラート（閾値超过时通知）
- コスト上限アラート
- Webhook通知（WeChat/ Slack対応）
"""
def __init__(
self,
webhook_url: Optional[str] = None,
cost_alert_threshold: float = 10.0,
error_rate_alert_threshold: float = 0.1,
latency_alert_threshold_ms: float = 5000,
):
self.metrics = MonitoringMetrics()
self.webhook_url = webhook_url
self.cost_alert_threshold = cost_alert_threshold
self.error_rate_alert_threshold = error_rate_alert_threshold
self.latency_alert_threshold_ms = latency_alert_threshold_ms
self._start_time: Optional[float] = None
self._lock = threading.Lock()
# HolySheep推奨价格表（2026年1月更新）
self.cost_per_mtok = {
"gpt-4.1": 8.0,
"gpt-4-turbo": 10.0,
"claude-sonnet-4-5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3-2": 0.42,
}
self.total_cost = 0.0
def _send_alert(self, title: str, message: str, level: str = "warning"):
"""Webhook通知发送（支持企业微信等）"""
if not self.webhook_url:
return
payload = {
"msgtype": "markdown",
"markdown": {
"content": f"## 🚨 LLM监控アラート [{level.upper()}]\n\n"
f"**{title}**\n\n{message}\n\n"
f"---\n*HolySheep AI 监控通知*"
}
}
try:
response = requests.post(
self.webhook_url,
json=payload,
timeout=5
)
response.raise_for_status()
print(f"[ALERT] 发送成功: {title}")
except requests.RequestException as e:
print(f"[ALERT ERROR] 发送失败: {e}")
def _calculate_cost(self, tokens: int, model: str) -> float:
"""トークン消费コスト計算"""
price = self.cost_per_mtok.get(model, 8.0) # デフォルトはGPT-4.1
return (tokens / 1_000_000) * price
def on_llm_start(self, serialized: Dict, prompts: List, **kwargs):
self._start_time = time.time()
def on_llm_end(self, response: LLMResult, **kwargs):
if self._start_time is None:
return
latency_ms = (time.time() - self._start_time) * 1000
model = response.llm_output.get("model_name", "unknown")
with self._lock:
self.metrics.total_calls += 1
self.metrics.successful_calls += 1
self.metrics.total_latency_ms += latency_ms
self.metrics.latency_by_model[model].append(latency_ms)
# トークン消费記録
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
tokens = usage.get("total_tokens", 0)
self.metrics.total_tokens += tokens
# コスト計算
cost = self._calculate_cost(tokens, model)
self.total_cost += cost
# コスト上限チェック
if self.total_cost >= self.cost_alert_threshold:
self._send_alert(
"コスト上限接近",
f"現在のコスト: ${self.total_cost:.2f}\n"
f"閾値: ${self.cost_alert_threshold:.2f}\n"
f"モデル: {model}"
)
# 延迟监控（HolySheep目标<50ms）
if latency_ms > self.latency_alert_threshold_ms:
self._send_alert(
"高延迟警告",
f"延迟: {latency_ms:.2f}ms\n"
f"モデル: {model}\n"
f"目標: <50ms（HolySheep AI実績値）"
)
def on_llm_error(self, error: Exception, **kwargs):
with self._lock:
self.metrics.total_calls += 1
self.metrics.failed_calls += 1
error_type = type(error).__name__
self.metrics.error_counts[error_type] += 1
error_rate = self.metrics.failed_calls / self.metrics.total_calls
if error_rate > self.error_rate_alert_threshold:
self._send_alert(
"エラー率上昇",
f"エラー率: {error_rate*100:.1f}%\n"
f"失敗回数: {self.metrics.failed_calls}\n"
f"エラー型: {error_type}\n"
f"最新エラー: {str(error)[:100]}"
)
def get_summary(self) -> Dict:
"""监控サマリー取得"""
with self._lock:
avg_latency = (
self.metrics.total_latency_ms / self.metrics.successful_calls
if self.metrics.successful_calls > 0 else 0
)
return {
"総呼び出し数": self.metrics.total_calls,
"成功": self.metrics.successful_calls,
"失敗": self.metrics.failed_calls,
"エラー率": f"{self.metrics.failed_calls / max(1, self.metrics.total_calls) * 100:.2f}%",
"合計トークン": self.metrics.total_tokens,
"平均レイテンシ": f"{avg_latency:.2f}ms",
"総コスト": f"${self.total_cost:.4f}",
"エラー内訳": dict(self.metrics.error_counts),
}
使用例
monitor = ProductionMonitoringHandler(
cost_alert_threshold=5.0,
error_rate_alert_threshold=0.05,
latency_alert_threshold_ms=3000,
)
llm = ChatOpenAI(
model="deepseek-v3-2", # $0.42/MTok — 最安値
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
callbacks=[monitor],
)
批量调用监控
for i in range(10):
try:
response = llm.invoke(f"タスク {i} の結果を生成してください")
except Exception as e:
print(f"タスク {i} 失敗: {e}")
输出监控サマリー
print("\n📊 监控サマリー:")
for key, value in monitor.get_summary().items():
print(f" {key}: {value}")
LangChain Expression Language（LCEL）との統合
LangChain Expression Languageを使用する場合でも、コールバックは自然に統合できます：
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
HolySheep AI設定
llm = ChatOpenAI(
model="gpt-4.1",
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
)
监控用プロンプト
prompt = ChatPromptTemplate.from_messages([
("system", "あなたは专业的技術ライターです。"),
("human", "{topic}について简潔に説明してください。")
])
链式调用 with 监控
chain = prompt | llm | StrOutputParser()
回调处理器設定（LCELでも同样的接口）
monitor_handler = ComprehensiveLoggingHandler()
invoke時にコールバックを渡す
response = chain.invoke(
{"topic": "LangChain回调机制"},
config={"callbacks": [monitor_handler]}
)
print(f"\n回应内容:\n{response}")
HolySheep AIの监控最适合场景
私自身の实践では、HolySheep AIのプロキシ服务を活用することで、监控実装の suivantsな利点を実感しています：
- 低延迟监控：公式记录的<50msレイテンシにより、API応答延迟の监控精度が向上
- 统一的计费监控：レートが¥1=$1（公式¥7.3=$1の85%節約）のため、成本计算が简单
- 多通貨対応：WeChat Pay/Alipay対応で、チーム成员へのコスト配分管理が容易
- 多様なモデル対応：GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2を统一的に监控可能
よくあるエラーと対処法
エラー1: 401 Unauthorized — API鍵认证失败
# 错误示例
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
base_urlに/v1が含まれているか确认すること
正しい実装
import os
from langchain_openai import ChatOpenAI
必ず/v1で终了するURLを使用
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
llm = ChatOpenAI(
model="gpt-4.1",
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_API_BASE"],
)
认证確認
try:
response = llm.invoke("ping")
print("认证成功")
except Exception as e:
if "401" in str(e):
print("API键无效。请确认：")
print("1. HolySheep AIでの注册完了")
print("2. API键的正确复制")
print("3. 信用额的残量確認")
エラー2: ConnectionError — ネットワークタイムアウト
# タイムアウト设定の重要性
from langchain_openai import ChatOpenAI
import os
プロダクション环境では必ずタイムアウトを設定
llm = ChatOpenAI(
model="gpt-4.1",
api_key=os.environ.get("OPENAI_API_KEY"),
base_url="https://api.holysheep.ai/v1",
request_timeout=60, # 60秒タイムアウト
max_retries=3, # リトライ回数
)
リトライ逻辑を実装
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_llm_with_retry(prompt: str) -> str:
try:
return llm.invoke(prompt).content
except Exception as e:
print(f"呼び出し失敗: {e}")
raise
#HolySheepの低延迟を活かしたタイムアウト设定
#目安: 预计延迟の2倍 + 缓冲时间
エラー3: RateLimitError — APIレート制限
# レート制限对策：セマフォによる并发制御
import asyncio
from collections import deque
from time import time
class RateLimitHandler:
"""
简单的レートリミッター実装
HolySheep AIのレート制限に対応
"""
def __init__(self, max_calls: int = 60, window_seconds: int = 60):
self.max_calls = max_calls
self.window = window_seconds
self.calls = deque()
def acquire(self) -> bool:
"""呼び出し許可判定"""
now = time()
# 古い记录を削除
while self.calls and self.calls[0] < now - self.window:
self.calls.popleft()
if len(self.calls) < self.max_calls:
self.calls.append(now)
return True
return False
def wait_time(self) -> float:
"""次の呼び出し可能までの時間（秒）"""
if not self.calls:
return 0
return max(0, self.calls[0] + self.window - time())
使用例
rate_limiter = RateLimitHandler(max_calls=60, window_seconds=60)
async def call_llm_with_rate_limit(prompt: str):
while not rate_limiter.acquire():
wait = rate_limiter.wait_time()
print(f"レート制限待ち: {wait:.2f}秒")
await asyncio.sleep(wait)
return await llm.ainvoke(prompt)
エラー4: ContextWindowExceededError — コンテキスト長超過
# 長い对话のコンテキスト管理策略
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.chat_history import InMemoryChatHistory
class SlidingWindowChatHistory(InMemoryChatHistory):
"""
滑动窗口方式の聊天历史记录
古いメッセージを自动削除してコンテキスト長を管理
"""
def __init__(self, max_messages: int = 20, max_tokens: int = 8000):
super().__init__()
self.max_messages = max_messages
self.max_tokens = max_tokens
def add_user_message(self, message: str) -> None:
super().add_user_message(message)
self._prune_if_needed()
def add_ai_message(self, message: str) -> None:
super().add_ai_message(message)
self._prune_if_needed()
def _prune_if_needed(self) -> None:
# メッセージ数での修剪
while len(self.messages) > self.max_messages:
self.messages.pop(0)
# トークン数での修剪（概算）
total_chars = sum(len(m.content) for m in self.messages)
while total_chars > self.max_tokens * 4 and len(self.messages) > 2:
removed = self.messages.pop(0)
total_chars -= len(removed.content)
使用例
chat_history = SlidingWindowChatHistory(max_messages=10)
for i in range(50):
user_input = f"ユーザー入力 {i}"
chat_history.add_user_message(user_input)
# AI响应
response = llm.invoke(
[SystemMessage(content="你是助手。保持简洁。")]
+ chat_history.messages
)
chat_history.add_ai_message(response.content)
print(f"对话 {i}: 消息数={len(chat_history.messages)}")
まとめ：监控的最佳实践
本稿では、LangChainのコールバック机制を活用したLLM呼び出しの监控・ログ記録の手法を解説しました。关键的なポイント：
- BaseCallbackHandlerを継承してカスタム监控クラスを作成
- トークン消费・延迟・ ошибка率をリアルタイム追踪
- Webhook統合でアラート即时通知
- レート制限・タイムアウト対策を必ず実装
HolySheep AIを活用すれば、¥1=$1のレート（公式比85%節約）で经济的にプロダクション级监控を実現できます。WeChat Pay/Alipay対応により、日本円建てでのコスト管理も容易です。
特にDeepSeek V3.2（$0.42/MTok）のような低成本モデルを活用すれば、大量呼び出しの监控コストも大幅に削減できます。
👉 HolySheep AI に登録して無料クレジットを獲得