LangChain ツール呼び出しインターセプト -- @wrap_tool_call

from langchain.agents.middleware import wrap_tool_call

@wrap_tool_call
def my_tool_wrapper(request, handler):
    # request.tool_call: ツール名とパラメータを含む
    # request.tool: ツールオブジェクト自体
    # request.state: Agent の現在の状態
    # request.runtime: ランタイムコンテキスト

    # handler(request) を呼び出して初めてツールが実際に実行される
    result = handler(request)

    # result は ToolMessage または Command
    return result

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool


@wrap_tool_call
def retry_tool_on_error(request, handler):
    """ツール呼び出し失敗時に自動リトライ"""
    max_retries = 3
    last_result = None

    for attempt in range(max_retries):
        try:
            result = handler(request)
            # エラー結果かどうかを確認
            if hasattr(result, 'status') and result.status == "error":
                if attempt < max_retries - 1:
                    print(f" [リトライ] ツールがエラーを返しました。{attempt + 1} 回目のリトライ...")
                    continue
            if attempt > 0:
                print(f" [リトライ成功] {attempt + 1} 回目の試行")
            return result
        except Exception as e:
            if attempt < max_retries - 1:
                import time
                time.sleep((attempt + 1) * 2)
                print(f" [リトライ] 例外 {e}、{attempt + 1} 回目のリトライ...")
            else:
                raise

    return last_result


# 失敗する可能性のあるツールをシミュレート
call_count = 0

@tool
def fetch_course_data(course_id: str) -> str:
    """サンプルからコースデータを取得します。

    Args:
course_id: コースID
    """
    global call_count
    call_count += 1
    # 最初の2回は失敗、3回目は成功をシミュレート
    if call_count < 3:
        raise Exception(f"ネットワークエラー：コースサービスに接続できません（{call_count} 回目の試行）")
    return f"コース {course_id}: Python3 基礎チュートリアル、30章、無料"


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[fetch_course_data],
    middleware=[retry_tool_on_error],
    system_prompt="あなたは サンプルのコースアシスタントです。",
)

result = agent.invoke({
    "messages": [HumanMessage(content="コース python-001 の情報を調べてください")]
})
print(f"\n最終応答: {result['messages'][-1].content}")

from langchain.agents.middleware import wrap_tool_call


@wrap_tool_call
def normalize_city_name(request, handler):
    """都市名を自動的に正規化します（全角を半角に変換、余分な空白を削除など）"""
    tool_call = request.tool_call

    # city パラメータを含むツール呼び出しのみ処理
    if "city" in tool_call.get("args", {}):
        city = tool_call["args"]["city"]
        # 都市名を正規化：スペースを削除、大文字小文字を統一
        normalized = city.strip().replace("　", "")  # 全角スペースを削除

        # パラメータを置換
        new_args = {**tool_call["args"], "city": normalized}
        new_tool_call = {**tool_call, "args": new_args}
        request = request.override(tool_call=new_tool_call)

    return handler(request)

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

tool_cache = {}


@wrap_tool_call
def cache_tool_results(request, handler):
    """ツール実行結果をキャッシュします"""
    # キャッシュキーを生成：ツール名 + パラメータ
    tool_name = request.tool_call.get("name", "unknown")
    tool_args = str(request.tool_call.get("args", {}))

    cache_key = f"{tool_name}:{tool_args}"

    # キャッシュを確認
    if cache_key in tool_cache:
        print(f"[ツールキャッシュヒット] {tool_name}")
        cached_content = tool_cache[cache_key]
        return ToolMessage(
            content=cached_content,
            tool_call_id=request.tool_call.get("id", ""),
            name=tool_name,
        )

    # ツールを実行
    result = handler(request)

    # キャッシュに保存
    if hasattr(result, 'content'):
        tool_cache[cache_key] = result.content
        print(f[ツールキャッシュ書き込み] {tool_name}、現在 {len(tool_cache)} 件)

    return result

import time
from langchain.agents.middleware import wrap_tool_call


@wrap_tool_call
def monitor_tool_performance(request, handler):
    """ツール呼び出しのパフォーマンス指標を監視します"""
    tool_name = request.tool_call.get("name", "unknown")
    tool_args = request.tool_call.get("args", {})

    # 開始時間を記録
    start_time = time.time()

    try:
        result = handler(request)
        elapsed = time.time() - start_time

        # 成功した呼び出しを記録
        print(f"[モニタリング] {tool_name}({tool_args}) 成功、所要時間 {elapsed:.2f}s")
        return result
    except Exception as e:
        elapsed = time.time() - start_time
        # 失敗した呼び出しを記録
        print(f"[モニタリング] {tool_name}({tool_args}) 失敗、所要時間 {elapsed:.2f}s、エラー: {e}")
        raise

from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command


@wrap_tool_call
def check_empty_result(request, handler):
    """ツールが空の結果を返した場合、Agent を直接終了し、モデル呼び出しを無駄にしません"""
    result = handler(request)

    # 空の結果が返されたかどうかを確認
    if hasattr(result, 'content') and (
        "見つかりません" in str(result.content)
        or "結果なし" in str(result.content)
        or "ない" in str(result.content)
    ):
        # 直接 Command を返して状態を更新
        # 状況を説明する AI メッセージを追加
        from langchain.messages import AIMessage
        return Command(update={
            "messages": [
                AIMessage(content="申し訳ありませんが、関連情報が見つかりませんでした。別のキーワードをお試しください。")
            ]
        })

    return result