LangChain-Tool-Zugriff -- InjectedState und InjectedStore
Manchmal benötigen Tools Zugriff auf mehr kontextbezogene Informationen, wie den aktuellen Konversationszustand, die persistenten Daten des Benutzers usw.
LangChain verwendet einen Dependency-Injection-Mechanismus, damit Tool-Funktionen diese Informationen automatisch erhalten.
InjectedState — Zugriff auf den Agentenstatus in Tools
Standardmäßig können Tools nur vom Modell bereitgestellte Daten über Parameter empfangen. Manchmal müssen Tools jedoch den Kontext der aktuellen Konversation kennen — wie den bisherigen Gesprächsverlauf, vom Benutzer bereits bestätigte Informationen usw.
InjectedStateErmöglicht Tools, den vollständigen Zustand des Agenten direkt zu lesen.
Beispiel
from typing import Annotated, Any
from dotenv import load_dotenv
load_dotenv()
from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def remember_preference(
preference: str,
state: Annotated[dict[str, Any], InjectedState],
) -> str:
"""Merke die Präferenzeinstellungen des Benutzers.
Argumente:
preference: der Präferenzinhalt des Benutzers
state: der aktuelle Agent-Status, der automatisch vom System injiziert wird
"""
# Den vorherigen Nachrichtenverlauf aus dem state abrufen
messages = state.get("messages", [])
message_count = len(messages)
# Kann jedes Feld im state lesen
previous_prefs = state.get("user_preferences", "none")
return (
f"Präferenz gemerkt: {preference}."
f"(Die aktuelle Konversation hat {message_count} Nachrichten,"
f"Vorherige Präferenz: {previous_prefs})"
)
# InjectedState wird automatisch injiziert, der Agent muss diesen Parameter nicht übergeben
result = remember_preference.invoke({
"preference": "dark theme",
# state muss nicht übergeben werden, er wird automatisch vom Framework injiziert
})
print(result)
from dotenv import load_dotenv
load_dotenv()
from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def remember_preference(
preference: str,
state: Annotated[dict[str, Any], InjectedState],
) -> str:
"""Merke die Präferenzeinstellungen des Benutzers.
Argumente:
preference: der Präferenzinhalt des Benutzers
state: der aktuelle Agent-Status, der automatisch vom System injiziert wird
"""
# Den vorherigen Nachrichtenverlauf aus dem state abrufen
messages = state.get("messages", [])
message_count = len(messages)
# Kann jedes Feld im state lesen
previous_prefs = state.get("user_preferences", "none")
return (
f"Präferenz gemerkt: {preference}."
f"(Die aktuelle Konversation hat {message_count} Nachrichten,"
f"Vorherige Präferenz: {previous_prefs})"
)
# InjectedState wird automatisch injiziert, der Agent muss diesen Parameter nicht übergeben
result = remember_preference.invoke({
"preference": "dark theme",
# state muss nicht übergeben werden, er wird automatisch vom Framework injiziert
})
print(result)
Run result:
已记住偏好: 暗色主题。(当前对话共 0 条消息，之前偏好: 无)
Verwendung von InjectedState in einem Agenten
Beispiel
from typing import Annotated, Any
from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def conversation_stats(
state: Annotated[dict[str, Any], InjectedState],
) -> str:
"""Statistische Informationen zur aktuellen Unterhaltung abrufen, wie z. B. Nachrichtenanzahl, Unterhaltungslänge usw.
Es sind keine Parameter erforderlich; die Statistiken werden automatisch aus dem aktuellen Zustand gelesen.
"""
messages = state.get("messages", [])
human_msgs = [m for m in messages if m.type == "human"]
ai_msgs = [m for m in messages if m.type == "ai"]
tool_msgs = [m for m in messages if m.type == "tool"]
return (
f"Unterhaltungsstatistiken: {len(messages)} Nachrichten insgesamt | "
f"Menschliche Nachrichten: {len(human_msgs)} | "
f"KI-Antworten: {len(ai_msgs)} | "
f"Tool-Aufrufe: {len(tool_msgs)}"
)
from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def conversation_stats(
state: Annotated[dict[str, Any], InjectedState],
) -> str:
"""Statistische Informationen zur aktuellen Unterhaltung abrufen, wie z. B. Nachrichtenanzahl, Unterhaltungslänge usw.
Es sind keine Parameter erforderlich; die Statistiken werden automatisch aus dem aktuellen Zustand gelesen.
"""
messages = state.get("messages", [])
human_msgs = [m for m in messages if m.type == "human"]
ai_msgs = [m for m in messages if m.type == "ai"]
tool_msgs = [m for m in messages if m.type == "tool"]
return (
f"Unterhaltungsstatistiken: {len(messages)} Nachrichten insgesamt | "
f"Menschliche Nachrichten: {len(human_msgs)} | "
f"KI-Antworten: {len(ai_msgs)} | "
f"Tool-Aufrufe: {len(tool_msgs)}"
)
InjectedState ermöglicht den Zugriff auf alle Felder in AgentState. Wenn du state_schema erweiterst (benutzerdefinierte Felder hinzufügst), können diese Felder auch von InjectedState gelesen werden.
InjectedStore — Zugriff auf persistenten Speicher in Tools
Der Agent-Status ist auf Gesprächsebene; er verschwindet, wenn das Gespräch endet. WohingegenStorees ein sitzungsübergreifender persistenter Speicher ist, der verwendet werden kann, um langfristige Informationen wie Benutzerpräferenzen, Lernfortschritt usw. zu speichern.
InjectedStoreErmöglicht es Tools, den Store direkt zu lesen und zu schreiben.
Beispiel
from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Einen persistenten Store erstellen (in echten Projekten kann eine Datenbank verwendet werden)
store = InMemoryStore()
# Einige Daten voreinstellen
store.put(("users", "user_001"), "profile", {
"data": {
"name": "Xiao Ming",
"level": "Anfänger",
"completed_courses": ["HTML-Grundlagen-Tutorial"]
}
})
@tool
def get_user_profile(
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Ruft die Lernprofilinformationen des aktuellen Benutzers ab.
Liest Benutzerdaten aus dem persistenten Speicher.
"""
# Daten aus dem Store lesen
# Store verwendet Namespace (namespace, key), um Daten zu organisieren
item = store.get(("users", "user_001"), "profile")
if item is None:
return "Benutzerprofil nicht gefunden"
profile = item.value["data"]
return (
f"Benutzerprofil: Name={profile['name']},"
f"Stufe={profile['level']},"
f"abgeschlossene Kurse={', '.join(profile['completed_courses'])}"
)
@tool
def save_course_progress(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Speichere den Lernfortschritt des Benutzers im persistenten Speicher.
Argumente:
course_name: der Name des abgeschlossenen Kurses
store: der vom System automatisch injizierte persistente Speicher
"""
# Vorhandene Daten lesen
item = store.get(("users", "user_001"), "profile")
profile = item.value["data"] if item else {"name": "Xiao Ming", "level": "Anfänger", "completed_courses": []}
# Kursliste aktualisieren
if course_name not in profile["completed_courses"]:
profile[completed_courses"].append(course_name)
# Zurück in den Store schreiben
store.put(("users", "user_001"), "profile", {"data": profile})
return (
f"Lernfortschritt aktualisiert! {len(profile['completed_courses'])} Kurse abgeschlossen:"
f"{', '.join(profile['completed_courses'])}"
)
# Das Tool testen
print(get_user_profile.invoke({}))
print(save_course_progress.invoke({"course_name": "Python3 Basic Tutorial"}))
print(get_user_profile.invoke({}))
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Einen persistenten Store erstellen (in echten Projekten kann eine Datenbank verwendet werden)
store = InMemoryStore()
# Einige Daten voreinstellen
store.put(("users", "user_001"), "profile", {
"data": {
"name": "Xiao Ming",
"level": "Anfänger",
"completed_courses": ["HTML-Grundlagen-Tutorial"]
}
})
@tool
def get_user_profile(
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Ruft die Lernprofilinformationen des aktuellen Benutzers ab.
Liest Benutzerdaten aus dem persistenten Speicher.
"""
# Daten aus dem Store lesen
# Store verwendet Namespace (namespace, key), um Daten zu organisieren
item = store.get(("users", "user_001"), "profile")
if item is None:
return "Benutzerprofil nicht gefunden"
profile = item.value["data"]
return (
f"Benutzerprofil: Name={profile['name']},"
f"Stufe={profile['level']},"
f"abgeschlossene Kurse={', '.join(profile['completed_courses'])}"
)
@tool
def save_course_progress(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Speichere den Lernfortschritt des Benutzers im persistenten Speicher.
Argumente:
course_name: der Name des abgeschlossenen Kurses
store: der vom System automatisch injizierte persistente Speicher
"""
# Vorhandene Daten lesen
item = store.get(("users", "user_001"), "profile")
profile = item.value["data"] if item else {"name": "Xiao Ming", "level": "Anfänger", "completed_courses": []}
# Kursliste aktualisieren
if course_name not in profile["completed_courses"]:
profile[completed_courses"].append(course_name)
# Zurück in den Store schreiben
store.put(("users", "user_001"), "profile", {"data": profile})
return (
f"Lernfortschritt aktualisiert! {len(profile['completed_courses'])} Kurse abgeschlossen:"
f"{', '.join(profile['completed_courses'])}"
)
# Das Tool testen
print(get_user_profile.invoke({}))
print(save_course_progress.invoke({"course_name": "Python3 Basic Tutorial"}))
print(get_user_profile.invoke({}))
Run result:
用户档案：姓名=小明，水平=入门，已完成课程=HTML 基础教程 学习进度已更新！已完成 2 门课程：HTML 基础教程, Python3 基础教程 用户档案：姓名=小明，水平=入门，已完成课程=HTML 基础教程, Python3 基础教程
Store in einem Agenten verwenden
Übergeben Sie den Store an create_agent(), und alle Tools im Agenten können über InjectedStore darauf zugreifen:
Beispiel
from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Store erstellen und Daten voreinstellen
store = InMemoryStore()
store.put(("example", "courses"), "catalog", {
"data": {
"Python3-Grundlagen-Tutorial": {"price": "Kostenlos", "duration": "20 Stunden"},
"Python-Datenanalyse": {"price": "Mitglied", "duration": "30 Stunden"},
"HTML-Grundlagen-Tutorial": {"price": "Kostenlos", "duration": "15 Stunden"},
}
})
@tool
def query_course_price(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Fragt die Preisinformationen eines angegebenen Kurses auf EXAMPLE ab.
Argumente:
course_name: der Kursname
"""
item = store.get(("example", "courses"), "catalog")
catalog = item.value["data"] if item else {}
if course_name in catalog:
info = catalog[course_name]
return f"{course_name} - Preis: {info['price']}, Lerndauer: {info['duration']}"
return f"Kurs nicht gefunden: {course_name}"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[query_course_price],
store=store, # Den Store an den Agenten übergeben
system_prompt="Sie sind ein Kursberater bei EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Wie viel kosten Python3 Basic Tutorial bzw. Python Data Analysis jeweils?")]
})
print(result["messages"][-1].content)
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Store erstellen und Daten voreinstellen
store = InMemoryStore()
store.put(("example", "courses"), "catalog", {
"data": {
"Python3-Grundlagen-Tutorial": {"price": "Kostenlos", "duration": "20 Stunden"},
"Python-Datenanalyse": {"price": "Mitglied", "duration": "30 Stunden"},
"HTML-Grundlagen-Tutorial": {"price": "Kostenlos", "duration": "15 Stunden"},
}
})
@tool
def query_course_price(
course_name: str,
store: Annotated[BaseStore, InjectedStore()],
) -> str:
"""Fragt die Preisinformationen eines angegebenen Kurses auf EXAMPLE ab.
Argumente:
course_name: der Kursname
"""
item = store.get(("example", "courses"), "catalog")
catalog = item.value["data"] if item else {}
if course_name in catalog:
info = catalog[course_name]
return f"{course_name} - Preis: {info['price']}, Lerndauer: {info['duration']}"
return f"Kurs nicht gefunden: {course_name}"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[query_course_price],
store=store, # Den Store an den Agenten übergeben
system_prompt="Sie sind ein Kursberater bei EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Wie viel kosten Python3 Basic Tutorial bzw. Python Data Analysis jeweils?")]
})
print(result["messages"][-1].content)
Run result:
根据查询结果： - 《Python3 基础教程》是免费的，学习时长约 20 小时 - 《Python 数据分析》是会员课程，学习时长约 30 小时 建议先从免费的 Python3 基础教程开始学习。
InjectedState vs InjectedStore Vergleich
| Dimension | InjectedState | InjectedStore |
|---|---|---|
| Geltungsbereich | Aktuelle Konversation (einzelner Agent-Lauf) | Sitzungsübergreifend (über mehrere Agent-Läufe hinweg geteilt) |
| Lebenszyklus | Verschwindet, wenn die Konversation endet | Dauerhafte Speicherung |
| Typische Verwendungen | Lesen des Nachrichtenverlaufs, Zwischenergebnisse der aktuellen Konversation | Benutzereinstellungen, Lernfortschritt, Konfigurationsinformationen |
| Injektionsmethode | InjectedState (automatisch injiziert) | InjectedStore() (Klammern erforderlich) |
| Datenorganisation | Flaches Wörterbuch | Namensraum + Schlüsselhierarchie |
InjectedToolArg — Markierung generischer injizierter Parameter
Zusätzlich zu den dedizierten InjectedState und InjectedStore kannst du auchInjectedToolArgverwenden, um jeden Parameter zu markieren, der vom Framework injiziert werden muss:
Beispiel
from typing import Annotated
from langchain.tools import tool, InjectedToolArg
@tool
def my_tool(
normal_param: str,
injected_param: Annotated[str, InjectedToolArg],
) -> str:
"""Ein Beispiel-Tool, das einen injizierten Parameter enthält.
Argumente:
normal_param: Dieser Parameter wird vom Modell bereitgestellt
injected_param: Dieser Parameter wird vom Framework injiziert (der Agent muss ihn nicht bereitstellen)
"""
return f"normal={normal_param}, injected={injected_param}"
from langchain.tools import tool, InjectedToolArg
@tool
def my_tool(
normal_param: str,
injected_param: Annotated[str, InjectedToolArg],
) -> str:
"""Ein Beispiel-Tool, das einen injizierten Parameter enthält.
Argumente:
normal_param: Dieser Parameter wird vom Modell bereitgestellt
injected_param: Dieser Parameter wird vom Framework injiziert (der Agent muss ihn nicht bereitstellen)
"""
return f"normal={normal_param}, injected={injected_param}"
Weitere ErweiterungenInjectedToolArg ist ein generischer Injektionsmarker; InjectedState, InjectedStore und InjectedToolCallId sind alle darauf basierend implementiert. In den meisten Fällen reicht die Verwendung der dedizierten Injektionsmarker aus; InjectedToolArg wird verwendet, um benutzerdefinierte Injektionslogik zu erweitern.