LangChain モデル呼び出しインターセプト -- @wrap_model_call
@wrap_model_call はミドルウェア（Middleware）の中でも最も強力なフックです。
@wrap_model_call は before/after のように単に観察するだけでなく、次のことができます。モデルの実行プロセスを完全に制御する——再試行、ダウングレード、キャッシュ、さらにはモデルをスキップして事前設定の返答を直接使用することもできます。
handler コールバックを理解する
@wrap_model_call の核心は次のものです。handler コールバック関数。handler(request) を呼び出して初めてモデルが実際に実行されます；呼び出さない場合はモデルをスキップします。
実例
# wrap_model_call の基本構造
# request: モデル、メッセージ、ツールなどすべての情報を含む
# handler: 呼び出し可能なオブジェクト。これを実行して初めてモデルが実際に呼び出される
@wrap_model_call
def my_middleware(request, handler):
# モデル呼び出しの前に何でもできる
print("モデルが呼び出されようとしています...")
# handler(request) を呼び出して初めてモデルが実行される
response = handler(request)
# モデル呼び出しの後に何でもできる
print(モデル呼び出しが完了しました)
return response
# request: モデル、メッセージ、ツールなどすべての情報を含む
# handler: 呼び出し可能なオブジェクト。これを実行して初めてモデルが実際に呼び出される
@wrap_model_call
def my_middleware(request, handler):
# モデル呼び出しの前に何でもできる
print("モデルが呼び出されようとしています...")
# handler(request) を呼び出して初めてモデルが実行される
response = handler(request)
# モデル呼び出しの後に何でもできる
print(モデル呼び出しが完了しました)
return response
シナリオ1：リトライメカニズム
これは最も一般的なシナリオです。モデル呼び出しがネットワーク問題で失敗する可能性があるため、自動リトライで信頼性を高められます。
実例
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""モデル呼び出しが失敗した場合に自動リトライ、最大3回"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [リトライ成功] {attempt + 1} 回目の試行")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # 待機時間を増加：2s, 4s
print(f" [リトライ] {attempt + 1} 回目の失敗、{wait_time}秒後に再試行...")
import time
time.sleep(wait_time)
# すべてのリトライが失敗した
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="あなたはサンプル のアシスタントです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="サンプルについて紹介してください")]
})
print(f"\n返信: {result['messages'][-1].content[:100]}...")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""モデル呼び出しが失敗した場合に自動リトライ、最大3回"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [リトライ成功] {attempt + 1} 回目の試行")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # 待機時間を増加：2s, 4s
print(f" [リトライ] {attempt + 1} 回目の失敗、{wait_time}秒後に再試行...")
import time
time.sleep(wait_time)
# すべてのリトライが失敗した
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="あなたはサンプル のアシスタントです。",
)
result = agent.invoke({
"messages": [HumanMessage(content="サンプルについて紹介してください")]
})
print(f"\n返信: {result['messages'][-1].content[:100]}...")
シナリオ2：モデルダウングレード/フェイルオーバー
プライマリモデルが利用できない場合、自動的にバックアップモデルに切り替えます：
実例
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# メインモデルとバックアップモデル
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # モデルダウングレード時にトークンを制限
@wrap_model_call
def fallback_on_error(request, handler):
"""メインモデルが失敗したら自動的にバックアップモデルへ切り替える"""
try:
# メインモデルを試す
return handler(request)
except Exception as e:
print(f"[ダウングレード] メインモデル失敗: {e}、バックアップモデルに切り替え...")
# request 内の model を上書きし、バックアップモデルに切り替える
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[ダウングレード] バックアップモデルも失敗: {e2}")
# 両方のモデルが失敗したら、親切なメッセージを返す
return AIMessage(
content=申し訳ございません、サービスは一時的に利用できません。しばらくしてから再試行してください。
)
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# メインモデルとバックアップモデル
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # モデルダウングレード時にトークンを制限
@wrap_model_call
def fallback_on_error(request, handler):
"""メインモデルが失敗したら自動的にバックアップモデルへ切り替える"""
try:
# メインモデルを試す
return handler(request)
except Exception as e:
print(f"[ダウングレード] メインモデル失敗: {e}、バックアップモデルに切り替え...")
# request 内の model を上書きし、バックアップモデルに切り替える
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[ダウングレード] バックアップモデルも失敗: {e2}")
# 両方のモデルが失敗したら、親切なメッセージを返す
return AIMessage(
content=申し訳ございません、サービスは一時的に利用できません。しばらくしてから再試行してください。
)
request.override() は不変メソッドです。新しい request のコピーを返し、元のオブジェクトは変更しません。これにより、各呼び出しが独立して安全であることが保証されます。
シナリオ3：モデルレスポンスのキャッシュ
繰り返しのクエリに対して、モデルレスポンスをキャッシュして API 呼び出しコストを削減できます：
実例
from langchain.agents.middleware import wrap_model_call
from langchain.messages import AIMessage
# シンプルなメモリキャッシュ
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""モデルレスポンスをキャッシュし、同じ質問を繰り返し呼び出さない"""
# 最後のユーザーメッセージの内容をキャッシュキーとして使用
messages = request.messages
if not messages:
return handler(request)
# キャッシュキーを生成
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # 長すぎる内容は切り詰める
# キャッシュをチェック
if cache_key in cache:
print(f"[キャッシュヒット] キャッシュ結果を直接返す")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(キャッシュから)*")
# キャッシュミス、モデルを呼び出す
result = handler(request)
AIMessage の content がリストに混在する可能性があるため、最初の要素を直接取得
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[キャッシュミス] キャッシュに保存済み、現在 {len(cache)} 件")
elif hasattr(result, 'model_response'):
# ExtendedModelResponse の場合
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="あなたはサンプル のアシスタントです。回答は簡潔に。",
)
# 1回目のクエリ（キャッシュミス）
result = agent.invoke({
"messages": [{"role": "user", "content": "Python とは何ですか？"}]
})
print(f"1回目: {result['messages'][-1].content[:80]}...\n")
# 2回目に同じ質問（キャッシュヒット）
result = agent.invoke({
"messages": [{"role": "user", "content": "Python とは何ですか？"}]
})
print(f"2回目: {result['messages'][-1].content[:80]}...")
from langchain.messages import AIMessage
# シンプルなメモリキャッシュ
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""モデルレスポンスをキャッシュし、同じ質問を繰り返し呼び出さない"""
# 最後のユーザーメッセージの内容をキャッシュキーとして使用
messages = request.messages
if not messages:
return handler(request)
# キャッシュキーを生成
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # 長すぎる内容は切り詰める
# キャッシュをチェック
if cache_key in cache:
print(f"[キャッシュヒット] キャッシュ結果を直接返す")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(キャッシュから)*")
# キャッシュミス、モデルを呼び出す
result = handler(request)
AIMessage の content がリストに混在する可能性があるため、最初の要素を直接取得
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[キャッシュミス] キャッシュに保存済み、現在 {len(cache)} 件")
elif hasattr(result, 'model_response'):
# ExtendedModelResponse の場合
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="あなたはサンプル のアシスタントです。回答は簡潔に。",
)
# 1回目のクエリ（キャッシュミス）
result = agent.invoke({
"messages": [{"role": "user", "content": "Python とは何ですか？"}]
})
print(f"1回目: {result['messages'][-1].content[:80]}...\n")
# 2回目に同じ質問（キャッシュヒット）
result = agent.invoke({
"messages": [{"role": "user", "content": "Python とは何ですか？"}]
})
print(f"2回目: {result['messages'][-1].content[:80]}...")
実行結果：
[缓存未命中] 已存入缓存，当前 1 条 第一次: Python 是一种高级编程语言，以简洁易读的语法著称... [缓存命中] 直接返回缓存结果 第二次: Python 是一种高级编程语言，以简洁易读的语法著称... *(来自缓存)*
シナリオ4：リクエストの変更——システムメッセージの動的注入
実例
from datetime import datetime
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
毎回のモデル呼び出し前に現在時刻情報を注入する
now = datetime.now()
time_context = (
f"現在時刻：{now.strftime('%Y年%m月%d日 %H:%M')}。"
f"今日は{['月曜日','火曜日','水曜日','木曜日','金曜日','土曜日','日曜日'][now.weekday()]}。"
)
# 既存の system_message に時刻情報を追加
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# override を使用して新しい request を作成
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
毎回のモデル呼び出し前に現在時刻情報を注入する
now = datetime.now()
time_context = (
f"現在時刻：{now.strftime('%Y年%m月%d日 %H:%M')}。"
f"今日は{['月曜日','火曜日','水曜日','木曜日','金曜日','土曜日','日曜日'][now.weekday()]}。"
)
# 既存の system_message に時刻情報を追加
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# override を使用して新しい request を作成
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
シナリオ5：複数の wrap_model_call の組み合わせ
複数の wrap_model_call ミドルウェアは自動的に順番に組み合わされ、最初のものが最も外側になります：
実例
from langchain.agents.middleware import wrap_model_call
@wrap_model_call
def outer_middleware(request, handler):
"""最外層のミドルウェア"""
print("[外側] 開始")
result = handler(request) # ここで inner_middleware に入る
print([外側] 終了)
return result
@wrap_model_call
def inner_middleware(request, handler):
"""内側のミドルウェア"""
print(" [内側] 開始")
result = handler(request) # ここで実際にモデルを呼び出す
print([内側] 終了)
return result
# 実行順序：
# [外側] 開始
# [内側] 開始
# → 実際にモデルを呼び出す
# [内側] 終了
# [外側] 終了
@wrap_model_call
def outer_middleware(request, handler):
"""最外層のミドルウェア"""
print("[外側] 開始")
result = handler(request) # ここで inner_middleware に入る
print([外側] 終了)
return result
@wrap_model_call
def inner_middleware(request, handler):
"""内側のミドルウェア"""
print(" [内側] 開始")
result = handler(request) # ここで実際にモデルを呼び出す
print([内側] 終了)
return result
# 実行順序：
# [外側] 開始
# [内側] 開始
# → 実際にモデルを呼び出す
# [内側] 終了
# [外側] 終了
その他の拡張複数の wrap_model_call はタマネギのように何層にも重なります。最も外側が最初に実行され、最後に返ります。これにより、キャッシュチェックを外側、リトライを内側にするなど、独立した複数の機能を互いに干渉させずに組み合わせることができます。