LangChain Tool-Aufruf-Abfangen -- @wrap_tool_call
@wrap_tool_call ermöglicht es Ihnen, Steuerungsfähigkeiten ähnlich wie @wrap_model_call auf der Ebene der Tool-Ausführung zu implementieren—Wiederholung, Caching, Parameterumschreibung und Ergebnisnachbearbeitung.
Grundstruktur
Die Struktur von @wrap_tool_call ähnelt der von @wrap_model_call und empfängt zwei Parameter: request und handler:
Beispiel
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: Enthält den Toolnamen und die Parameter
# request.tool: Das Tool-Objekt selbst
# request.state: Der aktuelle Zustand des Agenten
# request.runtime: Laufzeitkontext
# Das Tool wird erst dann tatsächlich ausgeführt, wenn handler(request) aufgerufen wird
result = handler(request)
# result ist ein ToolMessage oder Command
return result
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call: Enthält den Toolnamen und die Parameter
# request.tool: Das Tool-Objekt selbst
# request.state: Der aktuelle Zustand des Agenten
# request.runtime: Laufzeitkontext
# Das Tool wird erst dann tatsächlich ausgeführt, wenn handler(request) aufgerufen wird
result = handler(request)
# result ist ein ToolMessage oder Command
return result
Szenario 1: Wiederholung des Tool-Aufrufs
Die Tool-Ausführung kann aufgrund instabiler externer Dienste fehlschlagen; automatische Wiederholungen können die Zuverlässigkeit verbessern:
Beispiel
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):
"""Automatisch wiederholen, wenn ein Tool-Aufruf fehlschlägt"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Prüfen, ob es sich um ein Fehlerergebnis handelt
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Wiederholung] Tool hat einen Fehler zurückgegeben, erneuter Versuch #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Wiederholung erfolgreich] Versuch #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Wiederholung] Ausnahme {e}, erneuter Versuch #{attempt + 1}...")
else:
raise
return last_result
# Simuliere ein Tool, das fehlschlagen kann
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Kursdaten von EXAMPLE abrufen.
Argumente:
course_id: Kurs-ID
"""
global call_count
call_count += 1
# Simuliere einen Fehlschlag für die ersten beiden Versuche, Erfolg beim dritten
if call_count < 3:
raise Exception(f"Netzwerkfehler: Verbindung zum Kursdienst nicht möglich (Versuch #{call_count})")
return f"Kurs {course_id}: Python3-Basistutorial, 30 Kapitel, kostenlos"
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="Du bist der Kursassistent für EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Hilf mir, die Informationen für den Kurs python-001 nachzuschlagen")]
})
print(f"\nEndgültige Antwort: {result['messages'][-1].content}")
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):
"""Automatisch wiederholen, wenn ein Tool-Aufruf fehlschlägt"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Prüfen, ob es sich um ein Fehlerergebnis handelt
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Wiederholung] Tool hat einen Fehler zurückgegeben, erneuter Versuch #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Wiederholung erfolgreich] Versuch #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Wiederholung] Ausnahme {e}, erneuter Versuch #{attempt + 1}...")
else:
raise
return last_result
# Simuliere ein Tool, das fehlschlagen kann
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Kursdaten von EXAMPLE abrufen.
Argumente:
course_id: Kurs-ID
"""
global call_count
call_count += 1
# Simuliere einen Fehlschlag für die ersten beiden Versuche, Erfolg beim dritten
if call_count < 3:
raise Exception(f"Netzwerkfehler: Verbindung zum Kursdienst nicht möglich (Versuch #{call_count})")
return f"Kurs {course_id}: Python3-Basistutorial, 30 Kapitel, kostenlos"
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="Du bist der Kursassistent für EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Hilf mir, die Informationen für den Kurs python-001 nachzuschlagen")]
})
print(f"\nEndgültige Antwort: {result['messages'][-1].content}")
Ausgabe:
[重试] 异常 网络错误：无法连接到课程服务（第 1 次尝试），1 次重试... [重试] 异常 网络错误：无法连接到课程服务（第 2 次尝试），2 次重试... [重试成功] 第 3 次尝试 Final reply: Course python-001 is the Python3 basics tutorial, 30 chapters, free of charge.
Szenario 2: Ändern von Tool-Parametern
Dynamisches Ändern von Parametern vor der Tool-Ausführung ermöglicht die Parametertransformation, ohne den Tool-Code zu ändern:
Beispiel
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def normalize_city_name(request, handler):
"""Städtenamen automatisch normalisieren (Vollbreite in Halbbreite umwandeln, überflüssige Leerzeichen entfernen usw.)"""
tool_call = request.tool_call
# Nur Tool-Aufrufe verarbeiten, die den city-Parameter enthalten
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Städtenamen normalisieren: Leerzeichen entfernen, Groß-/Kleinschreibung vereinheitlichen
normalized = city.strip().replace(" ", "") # Vollbreite Leerzeichen entfernen
# Den Parameter ersetzen
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)
@wrap_tool_call
def normalize_city_name(request, handler):
"""Städtenamen automatisch normalisieren (Vollbreite in Halbbreite umwandeln, überflüssige Leerzeichen entfernen usw.)"""
tool_call = request.tool_call
# Nur Tool-Aufrufe verarbeiten, die den city-Parameter enthalten
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Städtenamen normalisieren: Leerzeichen entfernen, Groß-/Kleinschreibung vereinheitlichen
normalized = city.strip().replace(" ", "") # Vollbreite Leerzeichen entfernen
# Den Parameter ersetzen
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)
Szenario 3: Zwischenspeichern von Tool-Ergebnissen
Für wiederholte Tool-Aufrufe (gleiches Tool + gleiche Parameter) können Ergebnisse zwischengespeichert werden:
Beispiel
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-Ausführungsergebnisse zwischenspeichern"""
# Cache-Schlüssel generieren: Tool-Name + Parameter
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Cache prüfen
if cache_key in tool_cache:
print(f"[Tool-Cache-Treffer] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Das Tool ausführen
result = handler(request)
# Im Cache speichern
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Tool-Cache-Schreibvorgang] {tool_name}, derzeit {len(tool_cache)} Einträge")
return result
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Tool-Ausführungsergebnisse zwischenspeichern"""
# Cache-Schlüssel generieren: Tool-Name + Parameter
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Cache prüfen
if cache_key in tool_cache:
print(f"[Tool-Cache-Treffer] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Das Tool ausführen
result = handler(request)
# Im Cache speichern
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Tool-Cache-Schreibvorgang] {tool_name}, derzeit {len(tool_cache)} Einträge")
return result
Szenario 4: Protokollierung und Überwachung von Tool-Aufrufen
Detaillierte Informationen für alle Tool-Aufrufe aufzeichnen:
Beispiel
import time
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Überwacht Leistungskennzahlen für Tool-Aufrufe"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Startzeit erfassen
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Erfolgreichen Aufruf aufzeichnen
print(f"[Monitor] {tool_name}({tool_args}) erfolgreich ausgeführt, dauerte {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Fehlgeschlagenen Aufruf aufzeichnen
print(f"[Monitor] {tool_name}({tool_args}) fehlgeschlagen, dauerte {elapsed:.2f}s, Fehler: {e}")
raise
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Überwacht Leistungskennzahlen für Tool-Aufrufe"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Startzeit erfassen
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Erfolgreichen Aufruf aufzeichnen
print(f"[Monitor] {tool_name}({tool_args}) erfolgreich ausgeführt, dauerte {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Fehlgeschlagenen Aufruf aufzeichnen
print(f"[Monitor] {tool_name}({tool_args}) fehlgeschlagen, dauerte {elapsed:.2f}s, Fehler: {e}")
raise
Szenario 5: Nachfolgenden Ablauf anhand der Ergebnisse festlegen
Sie können anhand des Tool-Ausführungsergebnisses entscheiden, ob die Agent-Schleife fortgesetzt wird:
Beispiel
from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Wenn das Tool ein leeres Ergebnis zurückgibt, beende den Agenten direkt, ohne Modellaufrufe zu verschwenden"""
result = handler(request)
# Prüfen, ob ein leeres Ergebnis zurückgegeben wurde
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Direkt einen Command zurückgeben, um den Status zu aktualisieren
# Eine KI-Nachricht hinzufügen, um die Situation zu erklären
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Entschuldigung, es wurden keine relevanten Informationen gefunden. Bitte versuchen Sie es mit einem anderen Schlüsselwort.")
]
})
return result
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Wenn das Tool ein leeres Ergebnis zurückgibt, beende den Agenten direkt, ohne Modellaufrufe zu verschwenden"""
result = handler(request)
# Prüfen, ob ein leeres Ergebnis zurückgegeben wurde
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Direkt einen Command zurückgeben, um den Status zu aktualisieren
# Eine KI-Nachricht hinzufügen, um die Situation zu erklären
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Entschuldigung, es wurden keine relevanten Informationen gefunden. Bitte versuchen Sie es mit einem anderen Schlüsselwort.")
]
})
return result
Wenn wrap_tool_call ein Command zurückgibt, können Sie den Agent-Status über den Parameter update ändern. Die Verwendung eines Command ermöglicht es Ihnen, direkt eine KI-Nachricht an die Nachrichtenliste anzuhängen, und dann endet die Agent-Schleife auf natürliche Weise.
@wrap_model_call vs. @wrap_tool_call
| Dimension | @wrap_model_call | @wrap_tool_call |
|---|---|---|
| Abfangziel | Modellaufruf | Tool-Ausführung |
| Anfrageinhalt | model, messages, tools, system_prompt | tool_call, tool, state, runtime |
| Rückgabetyp | ModelResponse oder AIMessage | ToolMessage oder Command |
| Anwendbare Szenarien | Modell-Wiederholung, Fallback, Caching, Prompt-Änderung | Tool-Wiederholung, Caching, Parameter-Neuschreibung, Ergebnisverarbeitung |