始めに：为什么要监控LLM调用？
AIアプリケーション開発において、LangChain回调机制（Callback Mechanism）は非常に重要な機能です。私は実際に運用してから、この机制なしではコスト管理も品質管理も 불가능だと実感しています。
本ガイドでは、 완전 초보자でも理解できるように、ステップバイステップで説明します。専門用語はできるだけ避け、平易な言葉で書いています。
LangChain回调机制とは？
简单に言うと、LLM（大规模言語モデル）がどんな風に動いているかを見るための「監視カメラ」です。
- APIにどんなリクエストを送ったか記録できる
- レスポンス时间长さを测定できる
- エラー发生時に即座に気づくことができる
- コスト사용量도把握할 수 있다
第一步：事前准备
必要なものを揃えましょう。
# 必要なライブラリをインストール
pip install langchain langchain-openai langchain-anthropic
環境変数の設定
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_API_BASE="https://api.holysheep.ai/v1"💡 スクリーンショットヒント： ターミナルで上のコマンドを1行ずつ実行する 모습。エラーが出ないことを確認してください。
第二步：基礎的なCallbackHandlerの作成
监控用クラスを作成しましょう。これが核心部分です。
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
from datetime import datetime
import json
class MyCallbackHandler(BaseCallbackHandler):
"""自定义的监控处理器"""
def __init__(self):
self.total_tokens = 0
self.api_calls = []
self.start_time = None
def on_llm_start(self, serialized, prompts, **kwargs):
"""LLM调用开始时触发"""
self.start_time = datetime.now()
call_info = {
"event": "start",
"time": self.start_time.isoformat(),
"model": serialized.get("name", "unknown"),
"prompt_length": len(str(prompts))
}
self.api_calls.append(call_info)
print(f"📤 LLM调用开始: {call_info['time']}")
print(f" 模型: {call_info['model']}")
def on_llm_end(self, response: LLMResult, **kwargs):
"""LLM调用结束时触发"""
end_time = datetime.now()
duration = (end_time - self.start_time).total_seconds()
# 从响应中提取使用量信息
usage = {}
if hasattr(response, 'llm_output') and response.llm_output:
usage = response.llm_output.get('token_usage', {})
result = {
"event": "end",
"time": end_time.isoformat(),
"duration_ms": duration * 1000,
"tokens_used": usage
}
self.api_calls.append(result)
print(f"📥 LLM调用结束")
print(f" 耗时: {duration*1000:.2f}ms")
print(f" 使用量: {usage}")
def on_llm_error(self, error, **kwargs):
"""错误发生时触发"""
error_info = {
"event": "error",
"time": datetime.now().isoformat(),
"error": str(error)
}
self.api_calls.append(error_info)
print(f"❌ 错误发生: {error}")
def get_summary(self):
"""获取监控摘要"""
return {
"total_calls": len(self.api_calls),
"calls_detail": self.api_calls
}💡 スクリーンショットヒント： Pythonファイルに上のコードを貼り付けて保存する様子。ファイル名をmonitoring.pyにします。
第三步：实际应用
上で作成した监控处理器を使って、実際にLLMを呼び出してみましょう。
import os
from langchain_openai import ChatOpenAI
from monitoring import MyCallbackHandler
HolySheep AI的连接设置
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
创建监控处理器实例
monitor = MyCallbackHandler()
创建LLM实例并关联监控处理器
llm = ChatOpenAI(
model="gpt-4.1",
temperature=0.7,
callbacks=[monitor] # ← 关键！这里连接监控
)
执行调用
print("=== LLM调用开始 ===")
response = llm.invoke("LangChain回调机制是什么？用韩语回答。")
print(f"\n📝 回答: {response.content}")
输出监控结果
print("\n=== 监控结果 ===")
summary = monitor.get_summary()
print(json.dumps(summary, ensure_ascii=False, indent=2))💡 スクリーンショットヒント： 上のコードを実行した後のターミナル。呼び出し時間と使用量が記録されている様子。
第四步：成本监控实战
实际のコストを計算する実用的な例を見てみましょう。
import os
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from datetime import datetime
class CostTrackingHandler(BaseCallbackHandler):
"""成本追踪监控器"""
# HolySheep AI的定价（美元/百万トークン）
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"gpt-4.1-mini": {"input": 1.0, "output": 4.0},
"claude-sonnet-4": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.5, "output": 2.5},
"deepseek-v3.2": {"input": 0.42, "output": 2.10},
}
def __init__(self):
self.total_cost = 0.0
self.total_input_tokens = 0
self.total_output_tokens = 0
self.call_count = 0
def on_llm_end(self, response, **kwargs):
self.call_count += 1
usage = response.llm_output.get('token_usage', {}) if response.llm_output else {}
input_tokens = usage.get('prompt_tokens', 0)
output_tokens = usage.get('completion_tokens', 0)
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
# 从metadata获取model名称
model = "unknown"
if hasattr(response, 'llm_output') and response.llm_output:
model = response.llm_output.get('model_name', 'unknown')
# 计算成本
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
call_cost = input_cost + output_cost
self.total_cost += call_cost
print(f"📊 调用 #{self.call_count} | 模型: {model}")
print(f" 输入Token: {input_tokens:,} | 输出Token: {output_tokens:,}")
print(f" 本次成本: ${call_cost:.6f}")
print(f" 累计成本: ${self.total_cost:.6f}")
def get_report(self):
return {
"total_calls": self.call_count,
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"total_cost_usd": self.total_cost
}
使用示例
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
cost_tracker = CostTrackingHandler()
llm = ChatOpenAI(model="deepseek-v3.2", callbacks=[cost_tracker])
多次调用测试
questions = [
"你好",
"LangChain是什么？",
"回调机制有什么用？"
]
for q in questions:
response = llm.invoke(q)
print(f"Q: {q}")
print(f"A: {response.content[:50]}...")
print("-" * 50)
print("\n💰 最终成本报告:")
print(cost_tracker.get_report())💡 スクリーンショットヒント： 上のコードを実行后的コンソール。3回の呼び出しそれぞれのコストと累计コストが表示されている様子。
第五步：日志ファイルへの保存
监控データをファイルに保存して、後から分析できるようにしましょう。
import json
import logging
from datetime import datetime
from pathlib import Path
class FileLoggingHandler(BaseCallbackHandler):
"""日志文件监控处理器"""
def __init__(self, log_dir="logs"):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
self.log_file = self.log_dir / f"llm_calls_{datetime.now().strftime('%Y%m%d')}.jsonl"
# 设置文件日志
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.INFO)
if not self.logger.handlers:
handler = logging.FileHandler(self.log_file, encoding='utf-8')
formatter = logging.Formatter('%(message)s')
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def on_llm_start(self, serialized, prompts, **kwargs):
log_entry = {
"timestamp": datetime.now().isoformat(),
"event": "llm_start",
"model": serialized.get("name", "unknown"),
"prompt_preview": str(prompts)[:200]
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
def on_llm_end(self, response: LLMResult, **kwargs):
usage = {}
if response.llm_output:
usage = response.llm_output.get('token_usage', {})
model = response.llm_output.get('model_name', 'unknown')
else:
model = "unknown"
log_entry = {
"timestamp": datetime.now().isoformat(),
"event": "llm_end",
"model": model,
"usage": usage,
"duration_ms": getattr(response, 'latency', None)
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
def on_llm_error(self, error, **kwargs):
log_entry = {
"timestamp": datetime.now().isoformat(),
"event": "llm_error",
"error": str(error)
}
self.logger.error(json.dumps(log_entry, ensure_ascii=False))
使用方法
file_logger = FileLoggingHandler(log_dir="./my_logs")
llm = ChatOpenAI(model="gpt-4.1", callbacks=[file_logger])
response = llm.invoke("测试消息")
print(f"✅ 日志已保存到: {file_logger.log_file}")第六步：非同期应用
非同期処理で监控を使う珍しい例も見てみましょう。
import asyncio
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
class AsyncCallbackHandler(BaseCallbackHandler):
"""异步监控处理器"""
def __init__(self):
self.events = []
async def on_llm_start(self, serialized, prompts, **kwargs):
self.events.append(f"[异步] LLM开始: {datetime.now().isoformat()}")
print("🔄 异步调用开始...")
async def on_llm_end(self, response, **kwargs):
self.events.append(f"[异步] LLM结束: {datetime.now().isoformat()}")
print("✅ 异步调用完成")
async def main():
handler = AsyncCallbackHandler()
llm = ChatOpenAI(model="gemini-2.5-flash", callbacks=[handler])
tasks = [
llm.ainvoke("第一个问题"),
llm.ainvoke("第二个问题"),
llm.ainvoke("第三个问题")
]
responses = await asyncio.gather(*tasks)
print("\n📋 所有事件:")
for event in handler.events:
print(event)
return responses
运行
asyncio.run(main())よく会遇到するエラーと解决方案
エラー1：CallbackHandlersが動作しない
# ❌ 错误做法
llm = ChatOpenAI(model="gpt-4.1")
llm.invoke("hello", callbacks=[monitor]) # 这样callbacks不会生效
✅ 正确做法 - 在初始化时传入
llm = ChatOpenAI(model="gpt-4.1", callbacks=[monitor])
llm.invoke("hello") # 监控会自动生效原因： LangChain v0.3以降、callbacks引数はコンストラクタに直接渡す必要があります。
エラー2：token使用量信息が取得できない
# ❌ 错误做法
使用标准OpenAI客户端时不指定base_url
llm = ChatOpenAI(api_key="xxx", model="gpt-4.1")
✅ 正确做法 - 使用HolySheep AI的endpoint
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
openai_api_key="YOUR_HOLYSHEEP_API_KEY",
openai_api_base="https://api.holysheep.ai/v1", # ← 必须指定
model="gpt-4.1"
)原因： APIエンドポイントを直接指定しないと、使用量データが返回されません。
エラー3：非同期监控で顺序が乱れる
# ❌ 错误做法
async def wrong_way():
handler = AsyncCallbackHandler()
llm = ChatOpenAI(model="gpt-4.1")
# 直接调用ainvoke，不会复用handler的状态
await llm.ainvoke("prompt1")
await llm.ainvoke("prompt2")
✅ 正确做法 - 使用RunnableWithMessageHistory或显式传递callbacks
async def correct_way():
handler = AsyncCallbackHandler()
llm = ChatOpenAI(model="gpt-4.1", callbacks=[handler])
# 每个调用都会使用同一个handler
tasks = [llm.ainvoke(f"问题{i}") for i in range(3)]
results = await asyncio.gather(*tasks)
# handler.events现在包含所有调用的记录
print(f"总共记录了 {len(handler.events)} 个事件")原因： 非同期并发执行时、每个任务需要明确关联同一个handler实例。
实战经验分享
私の場合、LangChain回调机制を始めて使ったのは、本番環境のコストが予想外に高くなった时でした。监控を入れる前は、何のAPIがどれくらい呼ばれているかさっぱ分かりませんでした。
回调机制を導入してからは、:
- 无效な重复调用を即座に発見
- コスト使用量の_pattern_を把握
- 错误発生時の_root cause_解析が格段に速くなった
特にコスト监控は每月の 청구서 が大幅に减りました。DeepSeek V3.2の$0.42/MTokという低価格を活かせば、コスト最优化の效果は絶大です。
次のステップ
本ガイド学习了后、以下を試してみてください：
- 自定义CallbackHandlerを作成して、特定のキーワードを含む响应をフラグ
- LangSmith等の外部服务に监控データを連携
- コスト上限アラート機能を実装
HolySheep AIなら、单一API键で多个模型を切换でき、成本监控も一元管理できます。
まとめ
LangChain回调机制は、LLMアプリケーションの">可視性と制御性を大幅に向上させる重要な機能です。监控を導入することで、コスト管理、品質保证、デバッグ效率が剧的に改善されます。
まずは基础的なCallbackHandlerから试して、徐々に高度な监控機能を追加していきましょう。
👉 HolySheep AI 가입하고 무료 크레딧 받기