LangChainコールバック机制：AI API呼び出しチェーンの監視機能をゼロから実装

AIアプリケーションを構築する際、「API 호출이本当に正しく動作しているか」「トークン消费量はどうか」「応答时间是問題ないか」を常に監視したいですよね。本稿では、LangChainのコールバック机制を使って、HolySheep AIへのAPI呼び出しを 체계的に監視する方法を、超初心者向けに解説します。

コールバック机制とは？为什么要监控？

コールバック（Callback）とは、特定のイベントが発生した際に 자동으로呼び出される関数のことです。LangChainにおけるコールバックは、以下のような場面で重要な役割を果たします：

• APIへの要求开始時・終了時のログ記録
• トークン使用量のリアルタイム監視
• エラー発生時の詳細な情报収集
• 応答时间（レイテンシ）の測定
• マルチステップチェーン全体の実行追跡

実践的なメリット：私の場合、本番環境にAIサービスを導入する際、コールバックなしで運用したところ、問題発生時に何が起きているか全く分からず、的痛苦经历がありました。コールバックを導入後は、异常を即座に検知でき対応の工数が剧的に减りました。

始める前の准备：環境构筑

まず、必要なライブラリをインストールします。HolySheep AIは¥1=$1の超お得レートで、<50msの低レイテンシが魅力的です。以下のコマンドで必要な包をインストールしてください：

pip install langchain langchain-openai python-dotenv

スクリーンショットヒント：ターミナルで上のコマンドを実行すると、インストール.progressが棒グラフで表示されます。 Successful installation と表示されれば完了です。

次に、.envファイルを作成し、APIキーを設定します：

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

重要：YOUR_HOLYSHEEP_API_KEYの部分をご自身のAPIキーに置き換えてください。HolySheep AI に登録하면ダッシュボードからAPIキーを取得できます。登録するだけで無料クレジットが貰えるのも嬉しいです。

基本的なコールバック实战：カスタムハンドラーの作成

LangChainのコールバック机制の核心は、BaseCallbackHandlerを継承したカスタムクラスを作成することです。以下の例では、API呼び出しの詳細をリアルタイムでコンソールに出力するハンドラーを作ります：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult, AgentAction, AgentFinish

load_dotenv()

カスタムコールバックハンドラーの定義
class DebugCallbackHandler(BaseCallbackHandler):
    """デバッグ用のコールバックハンドラー"""
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        """LLM呼び出し開始時に実行"""
        print(f"🔵 LLM呼び出し開始")
        print(f"   プロンプト長: {len(prompts[0])} 文字")
        print(f"   入力トークン概算: {len(prompts[0].split()) * 1.3:.0f}")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        """LLM呼び出し終了時に実行"""
        generation = response.generations[0][0]
        print(f"🟢 LLM呼び出し完了")
        print(f"   応答テキスト: {generation.text[:100]}...")
        
        # トークン使用量の取得
        if hasattr(response, 'llm_output') and response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            print(f"   プロンプトトークン: {token_usage.get('prompt_tokens', 'N/A')}")
            print(f"   生成トークン: {token_usage.get('completion_tokens', 'N/A')}")
            print(f"   合計トークン: {token_usage.get('total_tokens', 'N/A')}")
    
    def on_llm_error(self, error, **kwargs):
        """エラー発生時に実行"""
        print(f"🔴 LLMエラー発生: {type(error).__name__}")
        print(f"   エラーメッセージ: {str(error)}")
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        """ツール呼び出し開始時に実行"""
        print(f"🟡 ツール実行開始: {serialized.get('name', 'unknown')}")
    
    def on_tool_end(self, output, **kwargs):
        """ツール呼び出し終了時に実行"""
        print(f"🟢 ツール実行完了")

HolySheep AI用のChatOpenAIクライアント設定
llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    callbacks=[DebugCallbackHandler()],  # コールバックを設定
    max_tokens=500
)

テスト実行
result = llm.invoke("LangChainのコールバックについて1文で説明してください")
print(f"\n最終結果: {result.content}")

このコードを実行すると、以下のような 출력이表示されます：

• LLM呼び出し開始時のプロンプト情報
• 応答完了時のトークン使用量
• 実際の応答テキスト

スクリーンショットヒント：実行結果のコンソールには、🔵🟢🟡の色付きのアイコンと共に、各 工程の実行時間が表示されます。

応用：実行时间とコストを監視するコールバック

實際的な運用では、コスト管理も重要です。HolySheep AIの料金表を見ると、GPT-4.1は$8/MTok、Gemini 2.5 Flashは$2.50/MTokと比較的新しいモデル安価です。そこで、トークン消费量と推定コストをリアルタイムで監視するコールバックを作成しましょう：

import os
import time
from datetime import datetime
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult

load_dotenv()

価格表（$/MTok）- HolySheep AI 2026年料金
MODEL_PRICES = {
    "gpt-4.1": 8.00,
    "gpt-4o": 5.00,
    "gpt-4o-mini": 0.60,
    "claude-sonnet-4.5": 15.00,
    "claude-3-5-sonnet": 3.00,
    "gemini-2.5-flash": 2.50,
    "deepseek-v3.2": 0.42,
    "deepseek-chat": 0.27,
}

class CostTrackingCallback(BaseCallbackHandler):
    """コスト追跡用のコールバックハンドラー"""
    
    def __init__(self):
        self.total_tokens = 0
        self.prompt_tokens = 0
        self.completion_tokens = 0
        self.total_cost = 0.0
        self.start_time = None
        self.call_count = 0
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.start_time = time.time()
        self.call_count += 1
        model = serialized.get("name", "unknown")
        print(f"\n{'='*50}")
        print(f"📊 API呼び出し #{self.call_count}")
        print(f"   モデル: {model}")
        print(f"   開始時刻: {datetime.now().strftime('%H:%M:%S.%f')[:-3]}")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        elapsed_ms = (time.time() - self.start_time) * 1000
        
        if hasattr(response, 'llm_output') and response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            self.prompt_tokens += token_usage.get('prompt_tokens', 0)
            self.completion_tokens += token_usage.get('completion_tokens', 0)
            self.total_tokens += token_usage.get('total_tokens', 0)
            
            # コスト計算（入力+出力）/ 1,000,000 * 価格
            model = response.llm_output.get('model_name', 'unknown')
            price = MODEL_PRICES.get(model, 5.00)  # デフォルトは$5
            call_cost = (self.total_tokens / 1_000_000) * price
            self.total_cost = call_cost
        
        print(f"⏱️  実行時間: {elapsed_ms:.2f}ms")
        print(f"   レイテンシ: {'<50ms' if elapsed_ms < 50 else f'{elapsed_ms:.0f}ms'}")
        print(f"   累積トークン: {self.total_tokens:,}")
        print(f"   累積コスト: ${self.total_cost:.6f}")
        print(f"{'='*50}\n")
    
    def get_summary(self):
        return {
            "total_calls": self.call_count,
            "total_tokens": self.total_tokens,
            "prompt_tokens": self.prompt_tokens,
            "completion_tokens": self.completion_tokens,
            "total_cost_usd": self.total_cost,
            "total_cost_jpy": self.total_cost * 155,  # 概算レート
        }

コールバックインスタンスを作成
cost_tracker = CostTrackingCallback()

llm = ChatOpenAI(
    model="gemini-2.5-flash",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    callbacks=[cost_tracker]
)

複数回呼び出してコストを集計
print("コスト監視テスト開始...\n")

questions = [
    "LangChainとは何ですか？",
    "コールバック机制の利点は？",
    "AI应用の監視何故重要？"
]

for q in questions:
    result = llm.invoke(q)

最終サマリー
summary = cost_tracker.get_summary()
print("📈 最終サマリー")
print(f"   総呼び出し回数: {summary['total_calls']}")
print(f"   総トークン使用量: {summary['total_tokens']:,}")
print(f"   推定コスト: ${summary['total_cost_usd']:.6f} (約¥{summary['total_cost_jpy']:.2f})")

このコードを実行すると、各API呼び出しの詳細なコスト情報が 실시간で確認できます。HolyShehe AIの¥1=$1というレートを使えば、コスト管理が非常にシンプルになります。

実践的なポイント：私の場合、このコスト追跡コールバックを الإنتاج 환경に導入してからは、月次のAIコスト報告を自動生成できるようになりました。手作业での計算から解放されたのは非常に助かっています。

チェーン全体の監視：Agentとの組み合わせ

より実践的な例として、LangChainのAgentと組み合わせた包括的な監視システムを構築しましょう。Toolを使うAgentでは、LLMだけでなくToolの呼び出しも追跡できるため、システム全体の动作が可视化管理できます：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType, initialize_agent, Tool
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish

load_dotenv()

class ChainMonitoringCallback(BaseCallbackHandler):
    """チェーン全体の監視用コールバック"""
    
    def __init__(self):
        self.steps = []
        self.errors = []
    
    def on_chain_start(self, serialized, inputs, **kwargs):
        chain_name = serialized.get("name", ["unknown"])[0]
        self.steps.append({
            "event": "chain_start",
            "chain": chain_name,
            "timestamp": datetime.now().isoformat()
        })
        print(f"🔷 チェーン開始: {chain_name}")
    
    def on_chain_end(self, outputs, **kwargs):
        self.steps.append({
            "event": "chain_end",
            "timestamp": datetime.now().isoformat()
        })
        print(f"🔶 チェーン完了")
    
    def on_agent_action(self, action: AgentAction, **kwargs):
        self.steps.append({
            "event": "agent_action",
            "tool": action.tool,
            "input": str(action.tool_input)[:100],
            "timestamp": datetime.now().isoformat()
        })
        print(f"🔸 エージェント action: {action.tool}")
        print(f"   入力: {str(action.tool_input)[:80]}...")
    
    def on_agent_finish(self, finish: AgentFinish, **kwargs):
        self.steps.append({
            "event": "agent_finish",
            "output": str(finish.return_values)[:100],
            "timestamp": datetime.now().isoformat()
        })
        print(f"🔹 エージェント完了")
        print(f"   最終出力: {str(finish.return_values)[:80]}...")
    
    def get_execution_trace(self):
        return self.steps

カスタムToolの定義
def search_knowledge_base(query: str) -> str:
    """知識ベースを検索するTool"""
    knowledge = {
        "langchain": "LangChainはLLM应用開発用のフレームワークです",
        "callback": "コールバックはイベント駆動の監視机制です",
        "api": "APIはApplication Programming Interfaceの略です"
    }
    for key, value in knowledge.items():
        if key in query.lower():
            return value
    return "該当する情報が見つかりませんでした"

def calculate_expression(expr: str) -> str:
    """数式を計算するTool"""
    try:
        result = eval(expr)
        return f"計算結果: {result}"
    except:
        return "計算エラー"

Toolリストの作成
tools = [
    Tool(
        name="知識ベース検索",
        func=search_knowledge_base,
        description="LangChainやAI関連の用語を検索する際に使用"
    ),
    Tool(
        name="計算機",
        func=calculate_expression,
        description="数式を計算する際に使用（例: 2+2, 10*5）"
    ),
]

監視コールバックを設定
monitor = ChainMonitoringCallback()

llm = ChatOpenAI(
    model="deepseek-v3.2",  # 安価で高性能なモデル
    base_url="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY")
)

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    callbacks=[monitor]
)

Agentの実行
print("🤖 Agent監視テスト開始\n")
response = agent.run("LangChainのコールバックについて教えてください")

実行トレースの表示
print("\n📋 実行トレース:")
for i, step in enumerate(monitor.get_execution_trace(), 1):
    print(f"  {i}. {step}")

スクリーンショットヒント：実行結果では、AgentがどのToolを呼び出したか、いつ開始・終了したかが時系列で表示されます。異常が発生した場合에도、特定のステップで止まっていることが一目瞭然です。

WebSocketを使ったリアルタイム監視

より高度な監視が必要な場合、WebSocketベースのリアルタイムダッシュボードを構築することもできます。以下の例では、FastAPIとWebSocketを使ってAPI呼び出しをリアルタイムでストリーミングする方法を紹介します：

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import asyncio
import json
from datetime import datetime
from typing import List

app = FastAPI()

接続されたWebSocketクライアント
connected_clients: List[WebSocket] = []

class WebSocketCallback:
    """WebSocket経由でリアルタイム通知するコールバック"""
    
    @staticmethod
    async def broadcast(event_type: str, data: dict):
        """全クライアントにイベントをブロードキャスト"""
        message = json.dumps({
            "type": event_type,
            "data": data,
            "timestamp": datetime.now().isoformat()
        })
        for client in connected_clients:
            try:
                await client.send_text(message)
            except:
                pass

@app.websocket("/ws/monitor")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket監視エンドポイント"""
    await websocket.accept()
    connected_clients.append(websocket)
    try:
        while True:
            # クライアントからのping応答
            data = await websocket.receive_text()
            if data == "ping":
                await websocket.send_text(json.dumps({"type": "pong"}))
    except:
        pass
    finally:
        connected_clients.remove(websocket)

@app.get("/")
async def get_dashboard():
    """シンプルな監視ダッシュボード"""
    return HTMLResponse("""
    
    
    
        
        
    
    
        
🔍 API監視ダッシュボード
        
接続中...
        

        
    
    
    """)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

このダッシュボードを起動すると、ブラウザでリアルタイムにAPI呼び出し eventsを確認できるようになります。

よくあるエラーと対処法

実際にコールバックを実装する際に遭遇しやすいエラーと、その解决方案をまとめます。

エラー1：APIキーが認識されない

# ❌ 間違い：api_keyにリテラル文字列をそのまま記述
llm = ChatOpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY"  # これは動きません
)

✅ 正しい：環境変数から読み込む
import os
llm = ChatOpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ.get("HOLYSHEEP_API_KEY")  # 環境変数から取得
)

原因：APIキーは secretsな情报のため、コードに直接記述しても環境変数として展開されません。解決：.envファイルを作成し、python-dotenvでload_dotenv()を呼び出してから、os.environ.get()またはos.getenv()で参照してください。

エラー2：base_urlのエンドポイントが違う

# ❌ 間違い：OpenAI公式エンドポイントを指定
llm = ChatOpenAI(
    base_url="https://api.openai.com/v1",  # これはHolySheheでは動きません
    api_key=os.getenv("HOLYSHEHEP_API_KEY")
)

✅ 正しい：HolySheepのエンドポイントを指定
llm = ChatOpenAI(
    base_url="https://api.holysheep.ai/v1",  # HolySheepの正しいエンドポイント
    api_key=os.getenv("HOLYSHEEP_API_KEY")
)

原因：HolySheep AIはOpenAI互換のAPIですが、エンドポイントのURLは異なります。解決：必ずhttps://api.holysheep.ai/v1を使用してください。

エラー3：コールバック内で非同期処理が競合する

# ❌ 間違い：同期処理の中で非同期コールバックを使用
class SyncCallback(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        # ここで非同期処理を呼び出すとエラー
        asyncio.create_task(some_async_function())  # 動かない

✅ 正しい：同期用の同期版を用意、またはasyncコールバックを使用
class SyncCallback(BaseCallbackHandler):
    def __init__(self):
        self.results = []  # 結果を保存して後で処理
    
    def on_llm_end(self, response, **kwargs):
        # 結果をリストに蓄積
        self.results.append(response.generations[0][0].text)

または、AsyncIteratorCallbackHandlerを使用
from langchain.callbacks.base import AsyncCallbackHandler

class AsyncMonitoringCallback(AsyncCallbackHandler):
    async def on_llm_end(self, response, **kwargs):
        # 非同期処理が可能
        await self.send_to_slack("LLM呼び出し完了")

原因：LangChainのBaseCallbackHandlerは同期処理용、非同期処理を渡すとイベントループが競合します。解決：結果を保持するリスト用于保存するか、AsyncCallbackHandlerを継承した非同期用のハンドラーを作成してください。

エラー4：トークン使用量を取得できない

# ❌ 間違い：response.llm_outputが存在しない場合がある
def on_llm_end(self, response, **kwargs):
    tokens = response.llm_output['token_usage']['total_tokens']  # KeyError発生

✅ 正しい：存在確認后才アクセス
def on_llm_end(self, response, **kwargs):
    if hasattr(response, 'llm_output') and response.llm_output:
        token_usage = response.llm_output.get('token_usage', {})
        total = token_usage.get('total_tokens', 0)
    else:
        total = 0  # フォールバック値
    print(f"トークン数: {total}")

原因：全てのモデル・プロパイダーがtoken_usageを返すわけではありません。解決：get()メソッドで安全にアクセスし、存在しない場合のフォールバック值を設定してください。

まとめ：監視体制を確立して安定したAI应用運用を

本稿では、LangChainのコールバック机制を活用したAPI監視システムをゼロから構築する方法を学びました。ポイントをおさらいしましょう：

• カスタムコールバック：BaseCallbackHandlerを継承して好きな監視ロジックを実装
• コスト追跡：トークン使用量を記録してリアルタイムにコストを計算
• チェーン監視：AgentとToolの組み合わせた 전체的な実行追跡
• リアルタイムダッシュボード：WebSocketを使った可視化

HolySheep AIを組み合わせれば、¥1=$1の圧倒的にお得しいレートでAPIを呼び出せるため、コスト監視がより重要意义を持ちます。<50msの低レイテンシも相まって、リアルタイム監視システムとの亲和性非常高です。

コールバック机制をマスターすれば、AI应用の動作が完全に可视化され、問題発生时的原因特定が劇的に速くなりますぜひ、まずは 基本のコードから试して、少しずつ自分の用途に合わせた拡張してみてください。

👉 HolySheep AI に登録して無料クレジットを獲得