LangChain Callbacks: Agent Actionsのロギングとデバッグ完全ガイド

AIエージェントを本番環境に導入する際、最大の問題の一つが「ブラックボックス化」です。LangChainで構築したエージェントがどんな思考過程を経て結論に至ったのか、何件のツールを叩いて、各呼び出しのレイテンシはどの程度だったのか——これらが可視化できなければ、SLA要件の厳しいビジネス現場では信頼性のある運用ができません。

本稿では、HolySheep AIを活用したLangChain Callbacksの実装パターンを、東京のRPAベンダーを事例に紹介します。移行前後の実測値と、具体的な移行手順を交えて解説します。

LangChain Callbacksとは

LangChain Callbacksは、エージェントの実行ライフサイクル全程にフックを差し込める機構です。主なイベントには以下のものがあります：

• on_chat_model_start: LLM呼び出し開始時
• on_chat_model_end: LLM応答完了時
• on_tool_start: ツール実行開始時
• on_tool_end: ツール応答完了時
• on_chain_start/end: チェーン全体の開始・終了
• on_agent_action: エージェントの思考・判断時

これらのイベントを捕捉することで、エージェントの動作を完全にトレースできます。

事例：東京RPAベンチャーの課題

東京千代田区に本社を置く従業員数40名のRPAベンダー「ProcesStream株式会社（以下、PSP社）」は月額¥680,000のOpenAI APIコストで運用しており、月間コール数200万回超の大規模なLangChainベースエージェントを顧客企业提供していました。

PSP社が抱えていた問題は3つ 있었습니다：

• デバッグ困難: Production環境でのagent実行ログがCloudWatchに散在し、特定ユーザーのリクエストを追踪するまでに平均18分かかっていた
• コスト可視性の欠如: モデル別の使用量がわからず、コスト最適化の打率が低かった
• レイテンシ課題: 朝のピークタイム（9:00-10:00）に平均620msまで遅延が跳ね上がり、顧客からの苦情が月平均12件寄せられていた

HolySheep AIを選んだ理由

PSP社がHolySheep AIへの移行を決めた要因は3点です：

• ¥1=$1の為替レート: 公式レートの¥7.3=$1相比85%のコスト削減が見込める（例：GPT-4o出力$15/MTok → HolySheep ¥15/MTok）
• <50msの実測レイテンシ: アジア太平洋リージョン最適化により、日本のデータセンター経由の通信が最適化
• 統合ロギング: API呼び出しログがダッシュボードで一元管理でき、LLM名・トークン数・レイテンシが即時確認可能

実装：Custom Callback Handler

PSP社のLangChainエージェントに実装したCustom Callback Handlerのコードを示します。

import os
from datetime import datetime
from typing import Any, Dict, List, Optional
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult

HolySheep AI設定
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

class HolySheepDebugCallback(BaseCallbackHandler):
    """LangChainエージェントのアクションを詳細ログ出力するCallback Handler"""
    
    def __init__(self):
        self.agent_actions = []
        self.tool_calls = []
        self.llm_calls = []
        self.start_time = None
    
    def on_chain_start(
        self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs
    ) -> None:
        self.start_time = datetime.now()
        print(f"[Chain Start] {serialized.get('name', 'Unknown')}")
        print(f"  Input keys: {list(inputs.keys())}")
    
    def on_chain_end(self, outputs: Dict[str, Any], **kwargs) -> None:
        elapsed = (datetime.now() - self.start_time).total_seconds() * 1000
        print(f"[Chain End] Total elapsed: {elapsed:.2f}ms")
    
    def on_chat_model_start(
        self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs
    ) -> None:
        call_id = id(messages)
        model = serialized.get("name", "unknown")
        total_tokens = sum(len(msg.content) for batch in messages for msg in batch)
        print(f"[LLM Start] Model: {model}, Batches: {len(messages)}, Est Tokens: {total_tokens}")
        self.llm_calls.append({
            "call_id": call_id,
            "model": model,
            "start": datetime.now()
        })
    
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        # ストリーミング中のトークン表示（デバッグ用）
        print(f"  [Token] {token}", end="", flush=True)
    
    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        if self.llm_calls:
            last_call = self.llm_calls[-1]
            elapsed = (datetime.now() - last_call["start"]).total_seconds() * 1000
            # LLMResultからトークン数を抽出
            generation_info = response.generations[-1][-1].generation_info if response.generations else {}
            tokens = generation_info.get("token_usage", {}).get("total_tokens", 0)
            print(f"\n[LLM End] Elapsed: {elapsed:.2f}ms, Tokens: {tokens}")
    
    def on_agent_action(self, action: Any, **kwargs) -> None:
        print(f"[Agent Action] Tool: {action.tool}, Input: {action.tool_input}")
        self.agent_actions.append({
            "tool": action.tool,
            "tool_input": action.tool_input,
            "timestamp": datetime.now()
        })
    
    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) -> None:
        tool_name = serialized.get("name", "unknown")
        print(f"[Tool Start] {tool_name} <- {input_str[:100]}")
        self.tool_calls.append({
            "tool": tool_name,
            "input": input_str,
            "start": datetime.now()
        })
    
    def on_tool_end(self, output: str, **kwargs) -> None:
        if self.tool_calls:
            last_tool = self.tool_calls[-1]
            elapsed = (datetime.now() - last_tool["start"]).total_seconds() * 1000
            print(f"[Tool End] {last_tool['tool']} -> {elapsed:.2f}ms")
            print(f"  Output: {output[:200]}...")
    
    def get_summary(self) -> Dict[str, Any]:
        """実行サマリーを返す"""
        return {
            "total_agent_actions": len(self.agent_actions),
            "total_tool_calls": len(self.tool_calls),
            "total_llm_calls": len(self.llm_calls),
            "agent_actions": self.agent_actions,
            "tool_calls": self.tool_calls,
            "llm_calls": self.llm_calls
        }

Agent実装への組み込み

上記のCallback Handlerを実際のAgentに組み込む方法を示します。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

HolySheep API認証設定
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

カスタムCallback Handlerのインポート
from holysheep_callback import HolySheepDebugCallback

LLM初期化（HolySheep AI endpoints 사용）
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_API_BASE"],
    streaming=True
)

プロンプトテンプレート
prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは顧客サポートエージェントです。質問に対して正確かつ丁寧に回答してください。"),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

ツール定義
from langchain_core.tools import tool

@tool
def search_knowledge_base(query: str) -> str:
    """ナレッジベースを検索して関連情報を取得"""
    # 実際の実装ではDBやベクトルDB問い合わせ
    return f"ナレッジベース検索結果: {query}に関する回答を返します"

@tool
def escalate_to_human(ticket_id: str, reason: str) -> str:
    """人間のエージェントにエスカレーション"""
    return f"チケット{ticket_id}を人間のエージェントにエスカレーションしました。理由: {reason}"

tools = [search_knowledge_base, escalate_to_human]

Agent作成
agent = create_openai_functions_agent(llm, tools, prompt)

AgentExecutorにCallback Handlerを渡す
callback_handler = HolySheepDebugCallback()

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    callbacks=[callback_handler],  # ← ここでCallback登録
    max_iterations=5
)

実行例
if __name__ == "__main__":
    print("=== エージェント実行テスト ===\n")
    result = agent_executor.invoke(
        {"input": "注文番号12345の配送状況を知りたい"},
        config={"callbacks": [callback_handler]}
    )
    
    print("\n=== 実行サマリー ===")
    summary = callback_handler.get_summary()
    print(f"LLM呼び出し回数: {summary['total_llm_calls']}")
    print(f"ツール呼び出し回数: {summary['total_tool_calls']}")
    print(f"エージェントアクション回数: {summary['total_agent_actions']}")
    print(f"最終回答: {result['output']}")

カナリアデプロイ手順

PSP社では以下の手順でリスクを最小化しながらHolySheep AIへの移行を実施しました：

• Step 1: トラフィックの5%をHolySheep APIに向ける（feature flagで制御）
• Step 2: 24時間監視し、エラー率・レイテンシが閾値内か確認
• Step 3: トラフィックを25%に拡大、Cost DashboardでGPT-4o使用量を精査
• Step 4: 50% → 75% → 100%と段階的に拡大

各ステップでHolySheep AIダッシュボードのリアルタイムログを確認し、以下のKPIを監視しました：

# 監視スクリプト例（Prometheus + Grafana連携）
import requests
from datetime import datetime

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"

def check_api_health():
    """HolySheep APIの可用性をチェック"""
    # ダミーのlightweight pingリクエストでレイテンシ測定
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": "ping"}],
            "max_tokens": 5
        }
    )
    return {
        "status_code": response.status_code,
        "latency_ms": response.elapsed.total_seconds() * 1000,
        "timestamp": datetime.now().isoformat()
    }

if __name__ == "__main__":
    health = check_api_health()
    print(f"Status: {health['status_code']}, Latency: {health['latency_ms']:.2f}ms")
    
    # HolySheepの目標値: <50ms
    if health['latency_ms'] > 50:
        print("⚠️ レイテンシ閾値超過")}
    else:
        print("✅ 正常範囲内")

移行後30日の実測値

PSP社の移行後30日間のメトリクスを以下に示します：

指標	移行前（OpenAI直接）	移行後（HolySheep AI）	改善率
平均レイテンシ	620ms	142ms	▲77%
P95レイテンシ	1,240ms	198ms	▲84%
API月額コスト	$4,200 (¥680,000)	$680 (¥45,000)	▲84%
デバッグ工数/月	72時間	8時間	▲89%
顧客苦情/月	12件	1件	▲92%

特に注目すべきはコスト面です。HolySheep AIの¥1=$1レートにより、PSP社では月¥635,000の削減を達成。DeepSeek V3.2を¥0.42/MTokで活用し、構造化データ抽出タスクを低コスト化する экспери먼트も開始しています（公式価格比95%OFF）。

高度なデバッグ：Distributed Tracing

大規模システムでは、複数のLangChainチェーンをまたいだトレーサビリティが必要です。以下はOpenTelemetry互換のCallback実装です：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.trace import Status, StatusCode

OpenTelemetry初期化
provider = TracerProvider()
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

class OTelCallback(BaseCallbackHandler):
    """OpenTelemetry統合Callback for LangChain"""
    
    def on_chain_start(self, serialized: Dict, inputs: Dict, **kwargs):
        span = tracer.start_span(f"langchain.{serialized.get('name', 'chain')}")
        span.set_attribute("input.keys", list(inputs.keys()))
        return span
    
    def on_chain_end(self, outputs: Dict, span: Any = None, **kwargs):
        if span:
            span.set_attribute("output.keys", list(outputs.keys()))
            span.set_status(Status(StatusCode.OK))
            span.end()
    
    def on_llm_end(self, response: LLMResult, parent_span: Any = None, **kwargs):
        if parent_span:
            # HolySheepダッシュボードでも確認可能な共通フィールド
            parent_span.set_attribute("llm.model", response.llm_output.get("model", "unknown"))
            if response.generations:
                gen_info = response.generations[-1][-1].generation_info or {}
                parent_span.set_attribute("llm.tokens", gen_info.get("token_usage", {}).get("total_tokens", 0))
                parent_span.set_attribute("llm.finish_reason", gen_info.get("finish_reason", "unknown"))

利用例
otlp_callback = OTelCallback()
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[otlp_callback, HolySheepDebugCallback()]  # 複数Callback登録可能
)

よくあるエラーと対処法

エラー1: Callback Handlerが呼ばれない

# ❌ 間違い： callbacks引数をagentには渡してもAgentExecutorに渡していない
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools
)
ここでcallback登録してもAgent内部的のLLM呼び出しには反映されない

✅ 正しい方法： AgentExecutor側に登録
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[HolySheepDebugCallback()]  # AgentExecutorに登録
)

またはinvoke時に登録
result = agent_executor.invoke(
    {"input": "query"},
    config={"callbacks": [HolySheepDebugCallback()]}
)

エラー2: StreamingとCallbackの競合

# ❌ streaming=TrueのままCallback Handlerを使用すると出力が乱れる
llm = ChatOpenAI(
    model="gpt-4o",
    streaming=True  # ストリーミング有効
)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[HolySheepDebugCallback()]
)
on_llm_new_tokenと標準出力のタイミングが競合

✅ 解決策： streaming=Falseで初期化し、Callback内で制御
llm = ChatOpenAI(
    model="gpt-4o",
    streaming=False  # Callback Handler側で処理
)

またはCallback Handlerにstreamingフラグ追加
class HolySheepDebugCallback(BaseCallbackHandler):
    def __init__(self, handle_streaming=True):
        self.handle_streaming = handle_streaming
        self._buffer = []
    
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        if self.handle_streaming:
            self._buffer.append(token)
            # バッチで出力（100トークンごと）
            if len(self._buffer) % 100 == 0:
                print(f"Streaming: {''.join(self._buffer[-100:])}", end="\r")

エラー3: API Key認証エラー（401 Unauthorized）

# ❌ 環境変数名のよくあるミス
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

→ LangChain内部で別の環境変数名を参照しているケース
解決：明示的にClientに伝える

✅ 正しい方法
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 直接指定
    base_url="https://api.holysheep.ai/v1"  # 明示的base_url
)

環境変数を使う場合（LangChain 0.1.x系）
os.environには必ず "OPENAI_API_KEY"と"OPENAI_API_BASE"を設定
※ "ANTHROPIC_API_KEY"ではない点に注意（HolySheepはOpenAI互換）

エラー4: Context Window超過エラー

# ❌ 長い会話履歴をそのままAgentに渡す
messages = load_full_conversation_history(user_id)  # 100件超の可能性
agent_executor.invoke({"input": "最新の発注状況教えて", "chat_history": messages})

→ HolySheep APIのコンテキスト制限超え

✅ 解決策： ConversationSummaryMemoryで要約
from langchain.memory import ConversationSummaryMemory
from langchain.agents import initialize_agent

memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

直近5件のみ保持し、要約をコンテキストに注入
memory.chat_memory.add_user_message("...")
memory.chat_memory.add_ai_message("...")

要約をプロンプトに挿入
prompt = ChatPromptTemplate.from_messages([
    ("system", " conversation_summary: {chat_history}"),
    ("user", "{input}")
])

代わりにLast N messagesのみ取得
recent_messages = memory.chat_memory.messages[-10:]  # 直近10件のみ
agent_executor.invoke({
    "input": "query",
    "chat_history": recent_messages
})

HolySheep AIダッシュボード活用

HolySheep AIのダッシュボードでは以下の情報がリアルタイムで確認できます：

• Request Log: 各API呼び出しのモデル名、トークン数、レイテンシ
• Cost Analytics: モデル別・日時別のコスト内訳
• Error Rate: 400/500系エラーの発生状況
• Usage by Model: GPT-4.1 $8、Claude Sonnet 4.5 $15、Gemini 2.5 Flash $2.50、DeepSeek V3.2 $0.42など

WeChat Pay・Alipayにも対応しており、中国に開発チームを持つ企業でも簡単に決済できます。

まとめ

LangChain Callbacksは、エージェントの可観測性を大きく向上させる強力な仕組みです。HolySheep AIの¥1=$1レートと<50msレイテンシを組み合わせることで、コスト削減とパフォーマンス向上が同時に実現できます。

PSP社の事例で示したように、Custom Callback Handlerの実装と段階的なカナリアデプロイを組み合わせれば、本番環境への影響なく、安全かつ効果的な移行が可能です。

DeepSeek V3.2の¥0.42/MTokという破格の料金で非同期タスクをオフロードする экспериメントも効果的です。まずは今すぐ登録して、提供される無料クレジットでお試しください。

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