mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-13 05:07:51 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7c09532a51 | ||
|
|
98201fa98d | ||
|
|
0334c81eb2 | ||
|
|
0302621c24 | ||
|
|
921a8edf66 | ||
|
|
dfee69f4b2 | ||
|
|
fc83746a29 | ||
|
|
20befcb86a | ||
|
|
d97d18ad35 | ||
|
|
2f096a3c2e | ||
|
|
d7084a4c83 | ||
|
|
4b2fc401ff | ||
|
|
a3ebd02e71 | ||
|
|
c76431e57c | ||
|
|
5b9e65bbbd | ||
|
|
3326ac2e0d | ||
|
|
630944b7dd | ||
|
|
2fc293a2f4 | ||
|
|
f181695856 | ||
|
|
a7ab82fdd2 | ||
|
|
5632cc003a | ||
|
|
4d098c4cb5 | ||
|
|
2d097ee9d1 | ||
|
|
2f14631c84 | ||
|
|
9b67fcfc87 | ||
|
|
cde4b9092f | ||
|
|
620adb3bcf | ||
|
|
c8dd449d1a | ||
|
|
6eef693d6e | ||
|
|
193b4c34ec | ||
|
|
b242b5499c | ||
|
|
93bdd275f0 | ||
|
|
80c100273c | ||
|
|
4096463586 | ||
|
|
dc446846e8 | ||
|
|
b3cbd71ed5 | ||
|
|
3d2bc9069e | ||
|
|
89e1933ab0 | ||
|
|
5197b6261b | ||
|
|
bf2e321fb3 | ||
|
|
767d34872f | ||
|
|
acdc1f7f6b | ||
|
|
6c3f84bc4d | ||
|
|
223150870c | ||
|
|
50eb12d14f | ||
|
|
d8ce3d71cf | ||
|
|
1adef5b6f1 | ||
|
|
944947e889 | ||
|
|
a38727133b | ||
|
|
5d929c30b4 | ||
|
|
fb091de4f2 | ||
|
|
9eacc9ff67 | ||
|
|
4e57940152 | ||
|
|
c227b0dea1 | ||
|
|
b6e8816e7d | ||
|
|
779d454fcc | ||
|
|
9423a99515 | ||
|
|
4c04e023c8 | ||
|
|
4566b2aed4 | ||
|
|
55f6f052b6 | ||
|
|
c22a86f11e | ||
|
|
6fee6811a0 | ||
|
|
85180f5054 | ||
|
|
6e30e457d4 | ||
|
|
f6512d5357 | ||
|
|
3408e525f8 | ||
|
|
777ab1123f | ||
|
|
a5a0286568 | ||
|
|
f86427fe37 | ||
|
|
69f4ab0a79 | ||
|
|
9284412aac | ||
|
|
220b3b1d33 | ||
|
|
d0be6843ba | ||
|
|
7fb9f7c870 | ||
|
|
cc54df296e | ||
|
|
a5b61361af | ||
|
|
2f96c5c021 | ||
|
|
0915784a87 | ||
|
|
6952d3a1e0 | ||
|
|
d411013b6f | ||
|
|
ce9473293f | ||
|
|
7fd8666608 | ||
|
|
adf42e7ca1 | ||
|
|
b0b6e3d91f | ||
|
|
30e00b8c7b |
@@ -0,0 +1,47 @@
|
||||
#!/usr/bin/env python
|
||||
"""Debug why InjectedState deprecation warning is not being emitted."""
|
||||
|
||||
from typing import Annotated
|
||||
from langgraph.prebuilt import ToolNode, InjectedState, InjectedStore
|
||||
from langgraph.prebuilt.tool_node import _get_state_args, _get_store_arg, _get_reserved_keyword_args
|
||||
from langgraph.store.base import BaseStore
|
||||
from langchain_core.tools.base import create_schema_from_function
|
||||
from langchain_core.tools import StructuredTool
|
||||
|
||||
|
||||
def tool_with_injected_state(x: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
"""Tool using deprecated InjectedState annotation."""
|
||||
return f"state: {state.get('foo', 'none')}"
|
||||
|
||||
def tool_with_injected_store(x: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
"""Tool using deprecated InjectedStore annotation."""
|
||||
return "has store"
|
||||
|
||||
# Convert to tools
|
||||
tool1 = create_tool(tool_with_injected_state)
|
||||
tool2 = create_tool(tool_with_injected_store)
|
||||
|
||||
print("Debugging annotation detection:")
|
||||
print(f"\nTool 1 (InjectedState):")
|
||||
print(f" _get_state_args: {_get_state_args(tool1)}")
|
||||
print(f" _get_reserved_keyword_args: {_get_reserved_keyword_args(tool1)}")
|
||||
|
||||
print(f"\nTool 2 (InjectedStore):")
|
||||
print(f" _get_store_arg: {_get_store_arg(tool2)}")
|
||||
print(f" _get_reserved_keyword_args: {_get_reserved_keyword_args(tool2)}")
|
||||
|
||||
# Check the condition for warnings
|
||||
state_args = _get_state_args(tool1)
|
||||
reserved_args = _get_reserved_keyword_args(tool1)
|
||||
print(f"\nTool 1 warning condition:")
|
||||
print(f" state_args: {state_args}")
|
||||
print(f" reserved_args.get('state'): {reserved_args.get('state')}")
|
||||
print(f" Should warn: {bool(state_args and not reserved_args.get('state'))}")
|
||||
|
||||
store_arg = _get_store_arg(tool2)
|
||||
reserved_args2 = _get_reserved_keyword_args(tool2)
|
||||
print(f"\nTool 2 warning condition:")
|
||||
print(f" store_arg: {store_arg}")
|
||||
print(f" reserved_args.get('runtime'): {reserved_args2.get('runtime')}")
|
||||
print(f" Should warn: {bool(store_arg and not reserved_args2.get('runtime'))}")
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
#!/usr/bin/env python
|
||||
"""Debug why InjectedState warning is not being emitted."""
|
||||
|
||||
from typing import Annotated, get_args
|
||||
from langgraph.prebuilt import InjectedState, InjectedStore
|
||||
from langgraph.prebuilt.tool_node import get_all_basemodel_annotations, _is_injection
|
||||
from langchain_core.tools import StructuredTool
|
||||
from langgraph.store.base import BaseStore
|
||||
|
||||
|
||||
def tool_with_injected_state(x: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
"""Tool using deprecated InjectedState annotation."""
|
||||
return f"state: {state.get('foo', 'none')}"
|
||||
|
||||
def tool_with_injected_store(x: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
"""Tool using deprecated InjectedStore annotation."""
|
||||
return "has store"
|
||||
|
||||
# Convert to tools
|
||||
tool1 = StructuredTool.from_function(tool_with_injected_state)
|
||||
tool2 = StructuredTool.from_function(tool_with_injected_store)
|
||||
|
||||
print("Debugging annotation detection:")
|
||||
|
||||
print(f"\nTool 1 (InjectedState):")
|
||||
schema1 = tool1.get_input_schema()
|
||||
print(f" Schema fields: {list(schema1.__fields__.keys()) if hasattr(schema1, '__fields__') else list(schema1.model_fields.keys())}")
|
||||
|
||||
for name, type_ in get_all_basemodel_annotations(schema1).items():
|
||||
print(f" Field '{name}': type={type_}")
|
||||
args = get_args(type_)
|
||||
print(f" Type args: {args}")
|
||||
for arg in args:
|
||||
print(f" Is InjectedState? {_is_injection(arg, InjectedState)}")
|
||||
print(f" Type of arg: {type(arg)}")
|
||||
|
||||
print(f"\nTool 2 (InjectedStore):")
|
||||
schema2 = tool2.get_input_schema()
|
||||
print(f" Schema fields: {list(schema2.__fields__.keys()) if hasattr(schema2, '__fields__') else list(schema2.model_fields.keys())}")
|
||||
|
||||
for name, type_ in get_all_basemodel_annotations(schema2).items():
|
||||
print(f" Field '{name}': type={type_}")
|
||||
args = get_args(type_)
|
||||
print(f" Type args: {args}")
|
||||
for arg in args:
|
||||
print(f" Is InjectedStore? {_is_injection(arg, InjectedStore)}")
|
||||
print(f" Type of arg: {type(arg)}")
|
||||
@@ -0,0 +1,49 @@
|
||||
#!/usr/bin/env python
|
||||
"""Debug reserved keyword detection for annotated tools."""
|
||||
|
||||
from typing import Annotated
|
||||
from langgraph.prebuilt import InjectedState, InjectedStore
|
||||
from langgraph.prebuilt.tool_node import _get_reserved_keyword_args
|
||||
from langchain_core.tools import StructuredTool
|
||||
from langgraph.store.base import BaseStore
|
||||
|
||||
|
||||
def tool_with_injected_state(x: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
"""Tool using deprecated InjectedState annotation."""
|
||||
return f"state: {state.get('foo', 'none')}"
|
||||
|
||||
def tool_with_injected_store(x: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
"""Tool using deprecated InjectedStore annotation."""
|
||||
return "has store"
|
||||
|
||||
def tool_with_reserved_state(x: int, state) -> str:
|
||||
"""Tool using reserved keyword 'state'."""
|
||||
return f"state: {state.get('foo', 'none')}"
|
||||
|
||||
# Convert to tools
|
||||
tool1 = StructuredTool.from_function(tool_with_injected_state)
|
||||
tool2 = StructuredTool.from_function(tool_with_injected_store)
|
||||
tool3 = StructuredTool.from_function(tool_with_reserved_state)
|
||||
|
||||
print("Debugging reserved keyword detection:")
|
||||
|
||||
print(f"\nTool 1 (Annotated[dict, InjectedState]):")
|
||||
reserved1 = _get_reserved_keyword_args(tool1)
|
||||
print(f" Reserved args: {reserved1}")
|
||||
print(f" Has 'state' as reserved? {'state' in reserved1}")
|
||||
|
||||
print(f"\nTool 2 (Annotated[BaseStore, InjectedStore()]):")
|
||||
reserved2 = _get_reserved_keyword_args(tool2)
|
||||
print(f" Reserved args: {reserved2}")
|
||||
print(f" Has 'runtime' as reserved? {'runtime' in reserved2}")
|
||||
|
||||
print(f"\nTool 3 (plain 'state' parameter):")
|
||||
reserved3 = _get_reserved_keyword_args(tool3)
|
||||
print(f" Reserved args: {reserved3}")
|
||||
print(f" Has 'state' as reserved? {'state' in reserved3}")
|
||||
|
||||
# The issue might be that 'state' is being detected as a reserved keyword
|
||||
# even when it has an annotation
|
||||
print("\nConclusion:")
|
||||
print("If tool1 shows 'state' as reserved, that's the bug - it shouldn't be")
|
||||
print("considered reserved when it has an InjectedState annotation.")
|
||||
@@ -384,7 +384,7 @@ Putting this together, here is how you can implement a simple multi-agent system
|
||||
```python
|
||||
from typing import Annotated
|
||||
from langchain_core.tools import tool, InjectedToolCallId
|
||||
from langgraph.prebuilt import create_react_agent, InjectedState
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
from langgraph.graph import StateGraph, START, MessagesState
|
||||
from langgraph.types import Command
|
||||
|
||||
@@ -395,7 +395,7 @@ def create_handoff_tool(*, agent_name: str, description: str | None = None):
|
||||
@tool(name, description=description)
|
||||
def handoff_tool(
|
||||
# highlight-next-line
|
||||
state: Annotated[MessagesState, InjectedState], # (1)!
|
||||
state, # (1)! Reserved keyword - automatically injected
|
||||
# highlight-next-line
|
||||
tool_call_id: Annotated[str, InjectedToolCallId],
|
||||
) -> Command:
|
||||
@@ -636,3 +636,5 @@ Check out LangGraph [supervisor](https://github.com/langchain-ai/langgraph-super
|
||||
:::js
|
||||
Check out LangGraph [supervisor](https://github.com/langchain-ai/langgraphjs/tree/main/libs/langgraph-supervisor#customizing-handoff-tools) and [swarm](https://github.com/langchain-ai/langgraphjs/tree/main/libs/langgraph-swarm#customizing-handoff-tools) documentation to learn how to customize handoffs.
|
||||
:::
|
||||
|
||||
|
||||
|
||||
@@ -498,15 +498,14 @@ In this variant of the [supervisor](#supervisor) architecture, we define a super
|
||||
:::python
|
||||
|
||||
```python
|
||||
from typing import Annotated
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langgraph.prebuilt import InjectedState, create_react_agent
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
|
||||
model = ChatOpenAI()
|
||||
|
||||
# this is the agent function that will be called as tool
|
||||
# notice that you can pass the state to the tool via InjectedState annotation
|
||||
def agent_1(state: Annotated[dict, InjectedState]):
|
||||
# notice that you can pass the state to the tool via the reserved keyword 'state'
|
||||
def agent_1(state): # 'state' is a reserved keyword - automatically injected
|
||||
# you can pass relevant parts of the state to the LLM (e.g., state["messages"])
|
||||
# and add any additional logic (different models, custom prompts, structured output, etc.)
|
||||
response = model.invoke(...)
|
||||
@@ -515,7 +514,7 @@ def agent_1(state: Annotated[dict, InjectedState]):
|
||||
# by the prebuilt create_react_agent (supervisor)
|
||||
return response.content
|
||||
|
||||
def agent_2(state: Annotated[dict, InjectedState]):
|
||||
def agent_2(state): # 'state' is a reserved keyword - automatically injected
|
||||
response = model.invoke(...)
|
||||
return response.content
|
||||
|
||||
@@ -899,3 +898,4 @@ An agent might need to have a different state schema from the rest of the agents
|
||||
|
||||
- Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, it's important to [add input / output transformations](../how-tos/subgraph.ipynb#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
|
||||
- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ To implement handoffs, you can return `Command` objects from your agent nodes or
|
||||
```python
|
||||
from typing import Annotated
|
||||
from langchain_core.tools import tool, InjectedToolCallId
|
||||
from langgraph.prebuilt import create_react_agent, InjectedState
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
from langgraph.graph import StateGraph, START, MessagesState
|
||||
from langgraph.types import Command
|
||||
|
||||
@@ -37,7 +37,7 @@ def create_handoff_tool(*, agent_name: str, description: str | None = None):
|
||||
@tool(name, description=description)
|
||||
def handoff_tool(
|
||||
# highlight-next-line
|
||||
state: Annotated[MessagesState, InjectedState], # (1)!
|
||||
state, # (1)! Reserved keyword - automatically injected
|
||||
# highlight-next-line
|
||||
tool_call_id: Annotated[str, InjectedToolCallId],
|
||||
) -> Command:
|
||||
@@ -58,7 +58,7 @@ def create_handoff_tool(*, agent_name: str, description: str | None = None):
|
||||
return handoff_tool
|
||||
```
|
||||
|
||||
1. Access the [state](../concepts/low_level.md#state) of the agent that is calling the handoff tool using the @[InjectedState] annotation.
|
||||
1. Access the [state](../concepts/low_level.md#state) of the agent using the reserved keyword `state`. No annotation needed - LangGraph automatically injects the state when it sees this parameter name.
|
||||
2. The `Command` primitive allows specifying a state update and a node transition as a single operation, making it useful for implementing handoffs.
|
||||
3. Name of the agent or node to hand off to.
|
||||
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
|
||||
@@ -183,7 +183,6 @@ You can use the @[`Send()`][Send] primitive to directly send data to the worker
|
||||
|
||||
from typing import Annotated
|
||||
from langchain_core.tools import tool, InjectedToolCallId
|
||||
from langgraph.prebuilt import InjectedState
|
||||
from langgraph.graph import StateGraph, START, MessagesState
|
||||
# highlight-next-line
|
||||
from langgraph.types import Command, Send
|
||||
@@ -202,7 +201,7 @@ def create_task_description_handoff_tool(
|
||||
"Description of what the next agent should do, including all of the relevant context.",
|
||||
],
|
||||
# these parameters are ignored by the LLM
|
||||
state: Annotated[MessagesState, InjectedState],
|
||||
state, # Reserved keyword - automatically injected
|
||||
) -> Command:
|
||||
task_description_message = {"role": "user", "content": task_description}
|
||||
agent_input = {**state, "messages": [task_description_message]}
|
||||
@@ -382,7 +381,7 @@ const multiAgentGraph = new StateGraph(MessagesZodState)
|
||||
from typing import Annotated
|
||||
from langchain_core.messages import convert_to_messages
|
||||
from langchain_core.tools import tool, InjectedToolCallId
|
||||
from langgraph.prebuilt import create_react_agent, InjectedState
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
from langgraph.graph import StateGraph, START, MessagesState
|
||||
from langgraph.types import Command
|
||||
|
||||
@@ -435,7 +434,7 @@ const multiAgentGraph = new StateGraph(MessagesZodState)
|
||||
@tool(name, description=description)
|
||||
def handoff_tool(
|
||||
# highlight-next-line
|
||||
state: Annotated[MessagesState, InjectedState], # (1)!
|
||||
state, # (1)! Reserved keyword - automatically injected
|
||||
# highlight-next-line
|
||||
tool_call_id: Annotated[str, InjectedToolCallId],
|
||||
) -> Command:
|
||||
@@ -789,7 +788,7 @@ function agent(state: MessagesState): Command {
|
||||
```python
|
||||
from langchain_anthropic import ChatAnthropic
|
||||
from langgraph.graph import MessagesState, StateGraph, START
|
||||
from langgraph.prebuilt import create_react_agent, InjectedState
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
from langgraph.types import Command, interrupt
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
@@ -1250,4 +1249,9 @@ LangGraph comes with prebuilt implementations of two of the most popular multi-a
|
||||
:::js
|
||||
- [supervisor](../agents/multi-agent.md#supervisor) — individual agents are coordinated by a central supervisor agent. The supervisor controls all communication flow and task delegation, making decisions about which agent to invoke based on the current context and task requirements. You can use [`langgraph-supervisor`](https://github.com/langchain-ai/langgraph-supervisor-js) library to create a supervisor multi-agent systems.
|
||||
- [swarm](../agents/multi-agent.md#supervisor) — agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent. You can use [`langgraph-swarm`](https://github.com/langchain-ai/langgraph-swarm-js) library to create a swarm multi-agent systems.
|
||||
:::
|
||||
:::
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1141,12 +1141,12 @@ await agent.invoke(
|
||||
Short-term memory maintains **dynamic** state that changes during a single execution.
|
||||
|
||||
:::python
|
||||
To **access** (read) the graph state inside the tools, you can use a special parameter **annotation** — @[`InjectedState`][InjectedState]:
|
||||
To **access** (read) the graph state inside the tools, you can use the reserved keyword parameter `state`:
|
||||
|
||||
```python
|
||||
from typing import Annotated, NotRequired
|
||||
from typing import NotRequired
|
||||
from langchain_core.tools import tool
|
||||
from langgraph.prebuilt import InjectedState, create_react_agent
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
from langgraph.prebuilt.chat_agent_executor import AgentState
|
||||
|
||||
class CustomState(AgentState):
|
||||
@@ -1156,7 +1156,7 @@ class CustomState(AgentState):
|
||||
@tool
|
||||
def get_user_name(
|
||||
# highlight-next-line
|
||||
state: Annotated[CustomState, InjectedState]
|
||||
state # Reserved keyword - automatically injected
|
||||
) -> str:
|
||||
"""Retrieve the current user-name from state."""
|
||||
# Return stored name or a default if not set
|
||||
@@ -1173,6 +1173,9 @@ agent = create_react_agent(
|
||||
agent.invoke({"messages": "what's my name?"})
|
||||
```
|
||||
|
||||
!!! note "Migration from Annotations"
|
||||
The old annotation-based approach using `Annotated[CustomState, InjectedState]` is deprecated but still supported for backward compatibility. Simply use the reserved keyword `state` without any type annotation for cleaner code.
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
@@ -2387,3 +2390,4 @@ Some commonly used tool categories include:
|
||||
|
||||
These integrations can be configured and added to your agents using the same `tools` parameter shown in the examples above.
|
||||
:::
|
||||
|
||||
|
||||
@@ -383,7 +383,6 @@ We will implement handoffs via **handoff tools** and give these tools to the sup
|
||||
```python
|
||||
from typing import Annotated
|
||||
from langchain_core.tools import tool, InjectedToolCallId
|
||||
from langgraph.prebuilt import InjectedState
|
||||
from langgraph.graph import StateGraph, START, MessagesState
|
||||
from langgraph.types import Command
|
||||
|
||||
@@ -394,7 +393,7 @@ def create_handoff_tool(*, agent_name: str, description: str | None = None):
|
||||
|
||||
@tool(name, description=description)
|
||||
def handoff_tool(
|
||||
state: Annotated[MessagesState, InjectedState],
|
||||
state, # Reserved keyword - automatically injected
|
||||
tool_call_id: Annotated[str, InjectedToolCallId],
|
||||
) -> Command:
|
||||
tool_message = {
|
||||
@@ -644,7 +643,7 @@ def create_task_description_handoff_tool(
|
||||
"Description of what the next agent should do, including all of the relevant context.",
|
||||
],
|
||||
# these parameters are ignored by the LLM
|
||||
state: Annotated[MessagesState, InjectedState],
|
||||
state, # Reserved keyword - automatically injected
|
||||
) -> Command:
|
||||
task_description_message = {"role": "user", "content": task_description}
|
||||
agent_input = {**state, "messages": [task_description_message]}
|
||||
@@ -766,3 +765,5 @@ Update from subgraph research_agent:
|
||||
|
||||
{"query": "2024 United States GDP value from a reputable source", "follow_up_questions": null, "answer": null, "images": [], "results": [{"url": "https://www.focus-economics.com/countries/united-states/", "title": "United States Economy Overview - Focus Economics", "content": "The United States' Macroeconomic Analysis:\n------------------------------------------\n\n**Nominal GDP of USD 29,185 billion in 2024.**\n\n**Nominal GDP of USD 29,179 billion in 2024.**\n\n**GDP per capita of USD 86,635 compared to the global average of USD 10,589.**\n\n**GDP per capita of USD 86,652 compared to the global average of USD 10,589.**\n\n**Average real GDP growth of 2.5% over the last decade.**\n\n**Average real GDP growth of ```
|
||||
```
|
||||
|
||||
|
||||
|
||||
@@ -34,6 +34,7 @@ Typical Usage:
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
import warnings
|
||||
from copy import copy, deepcopy
|
||||
from dataclasses import replace
|
||||
from typing import (
|
||||
@@ -74,6 +75,7 @@ from typing_extensions import Annotated, get_args, get_origin
|
||||
from langgraph._internal._runnable import RunnableCallable
|
||||
from langgraph.errors import GraphBubbleUp
|
||||
from langgraph.graph.message import REMOVE_ALL_MESSAGES
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.store.base import BaseStore
|
||||
from langgraph.types import Command, Send
|
||||
|
||||
@@ -340,14 +342,71 @@ class ToolNode(RunnableCallable):
|
||||
self.tools_by_name: dict[str, BaseTool] = {}
|
||||
self.tool_to_state_args: dict[str, dict[str, Optional[str]]] = {}
|
||||
self.tool_to_store_arg: dict[str, Optional[str]] = {}
|
||||
self.tool_to_runtime_arg: dict[str, Optional[str]] = {}
|
||||
self.handle_tool_errors = handle_tool_errors
|
||||
self.messages_key = messages_key
|
||||
for tool_ in tools:
|
||||
if not isinstance(tool_, BaseTool):
|
||||
tool_ = create_tool(tool_)
|
||||
|
||||
# Check for deprecated annotation usage and emit warnings
|
||||
# We need to check for annotations directly, not just the presence of state/store args
|
||||
# because _get_state_args returns both reserved keywords and annotations
|
||||
reserved_args = _get_reserved_keyword_args(tool_)
|
||||
|
||||
# Check for InjectedState and InjectedStore annotations
|
||||
full_schema = tool_.get_input_schema()
|
||||
has_injected_state = False
|
||||
has_injected_store = False
|
||||
|
||||
for name, type_ in get_all_basemodel_annotations(full_schema).items():
|
||||
type_args = get_args(type_)
|
||||
|
||||
# Check for InjectedState (can be class or instance)
|
||||
for type_arg in type_args:
|
||||
if _is_injection(type_arg, InjectedState):
|
||||
if "state" not in reserved_args:
|
||||
has_injected_state = True
|
||||
break
|
||||
|
||||
# Check for InjectedStore (can be class or instance)
|
||||
for type_arg in type_args:
|
||||
if _is_injection(type_arg, InjectedStore):
|
||||
if "runtime" not in reserved_args:
|
||||
has_injected_store = True
|
||||
break
|
||||
|
||||
# Emit deprecation warnings
|
||||
if has_injected_state:
|
||||
warnings.warn(
|
||||
f"Tool '{tool_.name}' uses deprecated InjectedState annotation. "
|
||||
f"Please update to use reserved keyword 'state' instead. "
|
||||
f"Example: def {tool_.name}(..., state) instead of "
|
||||
f"def {tool_.name}(..., state: Annotated[dict, InjectedState]). "
|
||||
f"The annotation-based approach will be removed in a future version.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
if has_injected_store:
|
||||
warnings.warn(
|
||||
f"Tool '{tool_.name}' uses deprecated InjectedStore annotation. "
|
||||
f"Please update to use reserved keyword 'runtime' instead. "
|
||||
f"Example: def {tool_.name}(..., runtime) and access store via runtime.store. "
|
||||
f"The annotation-based approach will be removed in a future version.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
# Check for reserved keywords and wrap the tool if needed
|
||||
# Only wrap tools with reserved keywords for now to avoid breaking existing functionality
|
||||
if reserved_args:
|
||||
tool_ = _wrap_tool_with_reserved_keywords(tool_, reserved_args)
|
||||
|
||||
self.tools_by_name[tool_.name] = tool_
|
||||
self.tool_to_state_args[tool_.name] = _get_state_args(tool_)
|
||||
self.tool_to_store_arg[tool_.name] = _get_store_arg(tool_)
|
||||
self.tool_to_runtime_arg[tool_.name] = _get_runtime_arg(tool_)
|
||||
|
||||
def _func(
|
||||
self,
|
||||
@@ -360,7 +419,7 @@ class ToolNode(RunnableCallable):
|
||||
*,
|
||||
store: Optional[BaseStore],
|
||||
) -> Any:
|
||||
tool_calls, input_type = self._parse_input(input, store)
|
||||
tool_calls, input_type = self._parse_input(input, store, config)
|
||||
config_list = get_config_list(config, len(tool_calls))
|
||||
input_types = [input_type] * len(tool_calls)
|
||||
with get_executor_for_config(config) as executor:
|
||||
@@ -381,7 +440,7 @@ class ToolNode(RunnableCallable):
|
||||
*,
|
||||
store: Optional[BaseStore],
|
||||
) -> Any:
|
||||
tool_calls, input_type = self._parse_input(input, store)
|
||||
tool_calls, input_type = self._parse_input(input, store, config)
|
||||
outputs = await asyncio.gather(
|
||||
*(self._arun_one(call, input_type, config) for call in tool_calls)
|
||||
)
|
||||
@@ -554,6 +613,7 @@ class ToolNode(RunnableCallable):
|
||||
BaseModel,
|
||||
],
|
||||
store: Optional[BaseStore],
|
||||
config: Optional[RunnableConfig] = None,
|
||||
) -> Tuple[list[ToolCall], Literal["list", "dict", "tool_calls"]]:
|
||||
input_type: Literal["list", "dict", "tool_calls"]
|
||||
if isinstance(input, list):
|
||||
@@ -580,7 +640,7 @@ class ToolNode(RunnableCallable):
|
||||
raise ValueError("No AIMessage found in input")
|
||||
|
||||
tool_calls = [
|
||||
self.inject_tool_args(call, input, store)
|
||||
self.inject_tool_args(call, input, store, config)
|
||||
for call in latest_ai_message.tool_calls
|
||||
]
|
||||
return tool_calls, input_type
|
||||
@@ -661,6 +721,44 @@ class ToolNode(RunnableCallable):
|
||||
}
|
||||
return tool_call
|
||||
|
||||
def _inject_runtime(
|
||||
self,
|
||||
tool_call: ToolCall,
|
||||
store: Optional[BaseStore],
|
||||
config: RunnableConfig,
|
||||
) -> ToolCall:
|
||||
"""Inject runtime object into tool call arguments.
|
||||
|
||||
This method creates and injects a Runtime object containing store and
|
||||
context into tools that have a 'runtime' reserved keyword parameter.
|
||||
|
||||
Args:
|
||||
tool_call: The tool call dictionary to augment with runtime.
|
||||
store: The persistent store instance to include in runtime.
|
||||
config: The runnable configuration containing context.
|
||||
|
||||
Returns:
|
||||
The tool call with runtime injected if needed.
|
||||
"""
|
||||
runtime_arg = self.tool_to_runtime_arg[tool_call["name"]]
|
||||
if not runtime_arg:
|
||||
return tool_call
|
||||
|
||||
# Create a Runtime object with store and context from config
|
||||
# The context will be available from the config if set
|
||||
runtime = Runtime(
|
||||
context=config.get("configurable", {}).get("context"),
|
||||
store=store,
|
||||
stream_writer=lambda _: None, # Default no-op stream writer
|
||||
previous=None,
|
||||
)
|
||||
|
||||
tool_call["args"] = {
|
||||
**tool_call["args"],
|
||||
runtime_arg: runtime,
|
||||
}
|
||||
return tool_call
|
||||
|
||||
def inject_tool_args(
|
||||
self,
|
||||
tool_call: ToolCall,
|
||||
@@ -670,13 +768,15 @@ class ToolNode(RunnableCallable):
|
||||
BaseModel,
|
||||
],
|
||||
store: Optional[BaseStore],
|
||||
config: Optional[RunnableConfig] = None,
|
||||
) -> ToolCall:
|
||||
"""Inject graph state and store into tool call arguments.
|
||||
"""Inject graph state, store, and runtime into tool call arguments.
|
||||
|
||||
This method enables tools to access graph context that should not be controlled
|
||||
by the model. Tools can declare dependencies on graph state or persistent storage
|
||||
using InjectedState and InjectedStore annotations. This method automatically
|
||||
identifies these dependencies and injects the appropriate values.
|
||||
by the model. Tools can declare dependencies using either reserved keywords
|
||||
('state' and 'runtime') or annotations (InjectedState and InjectedStore for
|
||||
backward compatibility). This method automatically identifies these dependencies
|
||||
and injects the appropriate values.
|
||||
|
||||
The injection process preserves the original tool call structure while adding
|
||||
the necessary context arguments. This allows tools to be both model-callable
|
||||
@@ -689,10 +789,11 @@ class ToolNode(RunnableCallable):
|
||||
Can be a message list, state dictionary, or BaseModel instance.
|
||||
store: The persistent store instance to inject into tools requiring storage.
|
||||
Will be None if no store is configured for the graph.
|
||||
config: The runnable configuration containing context for runtime injection.
|
||||
|
||||
Returns:
|
||||
A new ToolCall dictionary with the same structure as the input but with
|
||||
additional arguments injected based on the tool's annotation requirements.
|
||||
additional arguments injected based on the tool's requirements.
|
||||
|
||||
Raises:
|
||||
ValueError: If a tool requires store injection but no store is provided,
|
||||
@@ -710,6 +811,11 @@ class ToolNode(RunnableCallable):
|
||||
tool_call_copy: ToolCall = copy(tool_call)
|
||||
tool_call_with_state = self._inject_state(tool_call_copy, input)
|
||||
tool_call_with_store = self._inject_store(tool_call_with_state, store)
|
||||
if config:
|
||||
tool_call_with_runtime = self._inject_runtime(
|
||||
tool_call_with_store, store, config
|
||||
)
|
||||
return tool_call_with_runtime
|
||||
return tool_call_with_store
|
||||
|
||||
def _validate_tool_command(
|
||||
@@ -854,6 +960,16 @@ def tools_condition(
|
||||
class InjectedState(InjectedToolArg):
|
||||
"""Annotation for injecting graph state into tool arguments.
|
||||
|
||||
.. deprecated:: 0.2.0
|
||||
Use reserved keyword 'state' instead of InjectedState annotation.
|
||||
The annotation-based approach will be removed in a future version.
|
||||
|
||||
Instead of:
|
||||
def tool(x: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
|
||||
Use:
|
||||
def tool(x: int, state) -> str:
|
||||
|
||||
This annotation enables tools to access graph state without exposing state
|
||||
management details to the language model. Tools annotated with InjectedState
|
||||
receive state data automatically during execution while remaining invisible
|
||||
@@ -928,6 +1044,17 @@ class InjectedState(InjectedToolArg):
|
||||
class InjectedStore(InjectedToolArg):
|
||||
"""Annotation for injecting persistent store into tool arguments.
|
||||
|
||||
.. deprecated:: 0.2.0
|
||||
Use reserved keyword 'runtime' instead of InjectedStore annotation.
|
||||
The annotation-based approach will be removed in a future version.
|
||||
|
||||
Instead of:
|
||||
def tool(x: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
|
||||
Use:
|
||||
def tool(x: int, runtime) -> str:
|
||||
# Access store via runtime.store
|
||||
|
||||
This annotation enables tools to access LangGraph's persistent storage system
|
||||
without exposing storage details to the language model. Tools annotated with
|
||||
InjectedStore receive the store instance automatically during execution while
|
||||
@@ -1027,12 +1154,159 @@ def _is_injection(
|
||||
return False
|
||||
|
||||
|
||||
def _get_state_args(tool: BaseTool) -> dict[str, Optional[str]]:
|
||||
"""Extract state injection mappings from tool annotations.
|
||||
def _get_reserved_keyword_args(tool: BaseTool) -> dict[str, str]:
|
||||
"""Extract reserved keyword arguments from tool function signature.
|
||||
|
||||
This function analyzes a tool's input schema to identify arguments that should
|
||||
be injected with graph state. It processes InjectedState annotations to build
|
||||
a mapping of tool argument names to state field names.
|
||||
This function inspects the tool's underlying function signature to identify
|
||||
parameters with reserved names ('state' and 'runtime') that should be injected
|
||||
automatically without requiring annotations.
|
||||
|
||||
Args:
|
||||
tool: The tool to analyze for reserved keyword parameters.
|
||||
|
||||
Returns:
|
||||
A dictionary mapping reserved parameter names to their injection type.
|
||||
Keys are parameter names, values are either 'state' or 'runtime'.
|
||||
"""
|
||||
reserved_args: dict[str, str] = {}
|
||||
|
||||
# Get the underlying function from the tool
|
||||
if hasattr(tool, "func"):
|
||||
func = tool.func
|
||||
elif hasattr(tool, "_run"):
|
||||
func = tool._run
|
||||
else:
|
||||
return reserved_args
|
||||
|
||||
# Inspect the function signature
|
||||
try:
|
||||
sig = inspect.signature(func)
|
||||
for param_name, param in sig.parameters.items():
|
||||
# Check for reserved keywords only if they don't have injection annotations
|
||||
# Parameters with InjectedState or InjectedStore annotations should not be
|
||||
# considered reserved keywords (they use the old annotation-based approach)
|
||||
if param_name == "state" and param.annotation == inspect.Parameter.empty:
|
||||
# Only consider 'state' as reserved if it has no annotation
|
||||
reserved_args["state"] = "state"
|
||||
elif (
|
||||
param_name == "runtime" and param.annotation == inspect.Parameter.empty
|
||||
):
|
||||
# Only consider 'runtime' as reserved if it has no annotation
|
||||
reserved_args["runtime"] = "runtime"
|
||||
except (ValueError, TypeError):
|
||||
# If we can't inspect the signature, return empty
|
||||
pass
|
||||
|
||||
return reserved_args
|
||||
|
||||
|
||||
def _wrap_tool_with_reserved_keywords(
|
||||
tool: BaseTool, reserved_args: dict[str, str]
|
||||
) -> BaseTool:
|
||||
"""Wrap a tool to exclude reserved keyword parameters from its schema.
|
||||
|
||||
This function creates a wrapper around tools that use reserved keywords
|
||||
('state' and 'runtime') to ensure these parameters are excluded from the
|
||||
schema presented to LLMs, similar to how InjectedToolArg annotations work.
|
||||
|
||||
Args:
|
||||
tool: The original tool to wrap.
|
||||
reserved_args: Dictionary of reserved keyword parameters to exclude.
|
||||
|
||||
Returns:
|
||||
A wrapped tool with reserved keywords excluded from its schema.
|
||||
"""
|
||||
if not reserved_args:
|
||||
return tool
|
||||
|
||||
# Create a wrapper tool that filters the schema
|
||||
# Use type: ignore to suppress mypy error for dynamic class creation
|
||||
class FilteredTool(tool.__class__): # type: ignore[name-defined]
|
||||
"""Tool wrapper that excludes reserved keywords from schema."""
|
||||
|
||||
def get_input_schema(
|
||||
self, config: Optional[RunnableConfig] = None
|
||||
) -> Type[BaseModel]:
|
||||
"""Return the filtered schema without reserved keywords."""
|
||||
original_schema = super().get_input_schema(config)
|
||||
|
||||
# If no reserved args to filter, return original
|
||||
if not reserved_args:
|
||||
return original_schema
|
||||
|
||||
# Create a new schema class dynamically
|
||||
from pydantic import create_model
|
||||
|
||||
# Get fields to keep (exclude reserved keywords)
|
||||
if hasattr(original_schema, "model_fields"):
|
||||
# Pydantic v2
|
||||
fields_to_keep: dict[str, Any] = {}
|
||||
for name, field in original_schema.model_fields.items():
|
||||
if name not in reserved_args:
|
||||
# For create_model, we need to properly extract field information
|
||||
field_type = (
|
||||
field.annotation if hasattr(field, "annotation") else Any
|
||||
)
|
||||
|
||||
# Handle field defaults and constraints
|
||||
if hasattr(field, "default") and field.default is not ...:
|
||||
# Field has a default value
|
||||
fields_to_keep[name] = (field_type, field.default)
|
||||
else:
|
||||
# Field has no default (required field)
|
||||
fields_to_keep[name] = (field_type, ...)
|
||||
else:
|
||||
# Pydantic v1 fallback
|
||||
fields_to_keep = {}
|
||||
|
||||
# Create filtered schema
|
||||
try:
|
||||
filtered_schema = create_model(
|
||||
f"{original_schema.__name__}Filtered",
|
||||
**fields_to_keep,
|
||||
)
|
||||
return filtered_schema
|
||||
except Exception:
|
||||
# If schema creation fails, return original
|
||||
return original_schema
|
||||
|
||||
# Create the filtered tool instance
|
||||
filtered_tool = FilteredTool(
|
||||
name=tool.name,
|
||||
description=tool.description,
|
||||
func=getattr(tool, "func", None),
|
||||
args_schema=tool.get_input_schema(), # Use original schema for initialization
|
||||
)
|
||||
|
||||
# Copy over other attributes
|
||||
for attr in [
|
||||
"return_direct",
|
||||
"verbose",
|
||||
"callbacks",
|
||||
"tags",
|
||||
"metadata",
|
||||
"handle_tool_error",
|
||||
"handle_validation_error",
|
||||
"response_format",
|
||||
]:
|
||||
if hasattr(tool, attr):
|
||||
setattr(filtered_tool, attr, getattr(tool, attr))
|
||||
|
||||
# Copy run methods
|
||||
if hasattr(tool, "_run"):
|
||||
filtered_tool._run = tool._run
|
||||
if hasattr(tool, "_arun"):
|
||||
filtered_tool._arun = tool._arun
|
||||
|
||||
return filtered_tool
|
||||
|
||||
|
||||
def _get_state_args(tool: BaseTool) -> dict[str, Optional[str]]:
|
||||
"""Extract state injection mappings from tool annotations or reserved keywords.
|
||||
|
||||
This function analyzes a tool to identify arguments that should be injected
|
||||
with graph state. It first checks for the reserved keyword 'state', then
|
||||
falls back to processing InjectedState annotations for backward compatibility.
|
||||
|
||||
Args:
|
||||
tool: The tool to analyze for state injection requirements.
|
||||
@@ -1041,10 +1315,20 @@ def _get_state_args(tool: BaseTool) -> dict[str, Optional[str]]:
|
||||
A dictionary mapping tool argument names to state field names. If a field
|
||||
name is None, the entire state should be injected for that argument.
|
||||
"""
|
||||
full_schema = tool.get_input_schema()
|
||||
tool_args_to_state_fields: dict = {}
|
||||
|
||||
# First check for reserved keywords
|
||||
reserved_args = _get_reserved_keyword_args(tool)
|
||||
if "state" in reserved_args:
|
||||
tool_args_to_state_fields["state"] = None
|
||||
|
||||
# Then check for annotation-based injection (backward compatibility)
|
||||
full_schema = tool.get_input_schema()
|
||||
for name, type_ in get_all_basemodel_annotations(full_schema).items():
|
||||
# Skip if already handled by reserved keyword
|
||||
if name in tool_args_to_state_fields:
|
||||
continue
|
||||
|
||||
injections = [
|
||||
type_arg
|
||||
for type_arg in get_args(type_)
|
||||
@@ -1073,6 +1357,9 @@ def _get_store_arg(tool: BaseTool) -> Optional[str]:
|
||||
should be injected with the graph store. Only one store argument is supported
|
||||
per tool.
|
||||
|
||||
Note: With the new reserved keyword approach, store is accessed via the
|
||||
'runtime' parameter which provides both store and context access.
|
||||
|
||||
Args:
|
||||
tool: The tool to analyze for store injection requirements.
|
||||
|
||||
@@ -1083,6 +1370,7 @@ def _get_store_arg(tool: BaseTool) -> Optional[str]:
|
||||
Raises:
|
||||
ValueError: If a tool argument has multiple InjectedStore annotations.
|
||||
"""
|
||||
# Check for annotation-based injection (backward compatibility)
|
||||
full_schema = tool.get_input_schema()
|
||||
for name, type_ in get_all_basemodel_annotations(full_schema).items():
|
||||
injections = [
|
||||
@@ -1101,3 +1389,19 @@ def _get_store_arg(tool: BaseTool) -> Optional[str]:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _get_runtime_arg(tool: BaseTool) -> Optional[str]:
|
||||
"""Extract runtime injection argument from tool signature.
|
||||
|
||||
This function checks if a tool has a 'runtime' reserved keyword parameter
|
||||
that should be injected with a Runtime object containing store and context.
|
||||
|
||||
Args:
|
||||
tool: The tool to analyze for runtime injection requirements.
|
||||
|
||||
Returns:
|
||||
The string 'runtime' if the tool has a runtime parameter, or None otherwise.
|
||||
"""
|
||||
reserved_args = _get_reserved_keyword_args(tool)
|
||||
return "runtime" if "runtime" in reserved_args else None
|
||||
|
||||
@@ -910,6 +910,142 @@ def test_tool_node_inject_store() -> None:
|
||||
failing_graph.invoke({"messages": [msg], "bar": "baz"})
|
||||
|
||||
|
||||
def test_tool_node_inject_state_reserved_keyword() -> None:
|
||||
"""Test that tools can use 'state' as a reserved keyword parameter."""
|
||||
|
||||
def tool1(some_val: int, state) -> str:
|
||||
"""Tool 1 with reserved keyword 'state'."""
|
||||
if isinstance(state, dict):
|
||||
return state["foo"]
|
||||
else:
|
||||
return getattr(state, "foo")
|
||||
|
||||
def tool2(some_val: int, state) -> str:
|
||||
"""Tool 2 with reserved keyword 'state'."""
|
||||
if isinstance(state, dict):
|
||||
return f"val: {some_val}, foo: {state['foo']}"
|
||||
else:
|
||||
return f"val: {some_val}, foo: {getattr(state, 'foo')}"
|
||||
|
||||
def tool3(some_val: int, y: str, state) -> str:
|
||||
"""Tool 3 with reserved keyword 'state' and other params."""
|
||||
if isinstance(state, dict):
|
||||
return f"{y}: {state['foo']}"
|
||||
else:
|
||||
return f"{y}: {getattr(state, 'foo')}"
|
||||
|
||||
# Test with dict state
|
||||
node = ToolNode([tool1, tool2, tool3])
|
||||
|
||||
# Verify that 'state' is excluded from tool schemas
|
||||
for tool in [tool1, tool2, tool3]:
|
||||
schema = node.tools_by_name[tool.__name__].get_input_schema()
|
||||
if hasattr(schema, "model_fields"):
|
||||
assert "state" not in schema.model_fields, (
|
||||
f"'state' should be excluded from {tool.__name__} schema"
|
||||
)
|
||||
else:
|
||||
assert "state" not in schema.__fields__, (
|
||||
f"'state' should be excluded from {tool.__name__} schema"
|
||||
)
|
||||
|
||||
for tool_name in ("tool1", "tool2"):
|
||||
tool_call = {
|
||||
"name": tool_name,
|
||||
"args": {"some_val": 1},
|
||||
"id": "some 0",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg], "foo": "bar"})
|
||||
tool_message = result["messages"][-1]
|
||||
if tool_name == "tool1":
|
||||
assert tool_message.content == "bar", f"Failed for tool={tool_name}"
|
||||
else:
|
||||
assert tool_message.content == "val: 1, foo: bar", (
|
||||
f"Failed for tool={tool_name}"
|
||||
)
|
||||
|
||||
# Test tool3 with additional parameter
|
||||
tool_call = {
|
||||
"name": "tool3",
|
||||
"args": {"some_val": 1, "y": "test"},
|
||||
"id": "some 0",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg], "foo": "bar"})
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "test: bar"
|
||||
|
||||
# Test with Pydantic state
|
||||
class State(MessagesState):
|
||||
foo: str
|
||||
|
||||
node_pydantic = ToolNode([tool1, tool2, tool3])
|
||||
for tool_name in ("tool1", "tool2"):
|
||||
tool_call = {
|
||||
"name": tool_name,
|
||||
"args": {"some_val": 2},
|
||||
"id": "some 1",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node_pydantic.invoke(State(messages=[msg], foo="baz"))
|
||||
tool_message = result["messages"][-1]
|
||||
if tool_name == "tool1":
|
||||
assert tool_message.content == "baz", (
|
||||
f"Failed for tool={tool_name} with Pydantic state"
|
||||
)
|
||||
else:
|
||||
assert tool_message.content == "val: 2, foo: baz", (
|
||||
f"Failed for tool={tool_name} with Pydantic state"
|
||||
)
|
||||
|
||||
|
||||
def test_tool_node_inject_runtime_reserved_keyword() -> None:
|
||||
"""Test that tools can use 'runtime' as a reserved keyword parameter."""
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
def tool1(some_val: int, runtime) -> str:
|
||||
"""Tool 1 with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime)
|
||||
if runtime.store:
|
||||
store_val = runtime.store.get(("test",), "test_key")
|
||||
if store_val:
|
||||
return f"val: {some_val}, store: {store_val.value['foo']}"
|
||||
return f"val: {some_val}, no store"
|
||||
|
||||
store = InMemoryStore()
|
||||
store.put(("test",), "test_key", {"foo": "bar"})
|
||||
|
||||
node = ToolNode([tool1])
|
||||
|
||||
# Verify that 'runtime' is excluded from tool schemas
|
||||
schema = node.tools_by_name[tool1.__name__].get_input_schema()
|
||||
if hasattr(schema, "model_fields"):
|
||||
assert "runtime" not in schema.model_fields
|
||||
else:
|
||||
assert "runtime" not in schema.__fields__
|
||||
|
||||
# Test with store
|
||||
tool_call = {
|
||||
"name": "tool1",
|
||||
"args": {"some_val": 1},
|
||||
"id": "some 0",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, store=store)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "val: 1, store: bar"
|
||||
|
||||
|
||||
def test_tool_node_mixed_injection_styles() -> None:
|
||||
"""Test that tools can mix reserved keywords and annotations."""
|
||||
pass # TODO: Implementation to be added
|
||||
|
||||
|
||||
def test_tool_node_ensure_utf8() -> None:
|
||||
@dec_tool
|
||||
def get_day_list(days: list[str]) -> list[str]:
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
"""Test reserved keywords for tool injection."""
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
from langgraph.prebuilt import InjectedState, InjectedStore, ToolNode
|
||||
from langgraph.store.base import BaseStore
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
|
||||
|
||||
def test_tool_node_inject_runtime_reserved_keyword() -> None:
|
||||
"""Test that tools can use 'runtime' as a reserved keyword parameter."""
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
def tool1(some_val: int, runtime) -> str:
|
||||
"""Tool 1 with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime), "runtime should be a Runtime instance"
|
||||
# Access store from runtime
|
||||
if runtime.store:
|
||||
store_val = runtime.store.get(("test",), "test_key")
|
||||
if store_val:
|
||||
return f"val: {some_val}, store: {store_val.value['foo']}"
|
||||
return f"val: {some_val}, no store"
|
||||
|
||||
def tool2(some_val: int, runtime) -> str:
|
||||
"""Tool 2 with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime), "runtime should be a Runtime instance"
|
||||
# Access context from runtime
|
||||
if runtime.context:
|
||||
return (
|
||||
f"val: {some_val}, context: {runtime.context.get('user_id', 'unknown')}"
|
||||
)
|
||||
return f"val: {some_val}, no context"
|
||||
|
||||
def tool3(x: int, y: str, runtime) -> str:
|
||||
"""Tool 3 with reserved keyword 'runtime' and other params."""
|
||||
assert isinstance(runtime, Runtime), "runtime should be a Runtime instance"
|
||||
has_store = "yes" if runtime.store else "no"
|
||||
has_context = "yes" if runtime.context else "no"
|
||||
return f"x: {x}, y: {y}, store: {has_store}, context: {has_context}"
|
||||
|
||||
store = InMemoryStore()
|
||||
store.put(("test",), "test_key", {"foo": "bar"})
|
||||
|
||||
node = ToolNode([tool1, tool2, tool3])
|
||||
|
||||
# Verify that 'runtime' is excluded from tool schemas
|
||||
for tool in [tool1, tool2, tool3]:
|
||||
schema = node.tools_by_name[tool.__name__].get_input_schema()
|
||||
if hasattr(schema, "model_fields"):
|
||||
assert "runtime" not in schema.model_fields, (
|
||||
f"'runtime' should be excluded from {tool.__name__} schema"
|
||||
)
|
||||
else:
|
||||
assert "runtime" not in schema.__fields__, (
|
||||
f"'runtime' should be excluded from {tool.__name__} schema"
|
||||
)
|
||||
|
||||
# Test with store
|
||||
tool_call = {
|
||||
"name": "tool1",
|
||||
"args": {"some_val": 1},
|
||||
"id": "some 0",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, store=store)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "val: 1, store: bar"
|
||||
|
||||
# Test with context
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
|
||||
config = RunnableConfig(configurable={"context": {"user_id": "test_user"}})
|
||||
|
||||
tool_call = {
|
||||
"name": "tool2",
|
||||
"args": {"some_val": 2},
|
||||
"id": "some 1",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, config=config)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "val: 2, context: test_user"
|
||||
|
||||
# Test with both store and context
|
||||
tool_call = {
|
||||
"name": "tool3",
|
||||
"args": {"x": 3, "y": "test"},
|
||||
"id": "some 2",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, store=store, config=config)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "x: 3, y: test, store: yes, context: yes"
|
||||
|
||||
|
||||
def test_tool_node_mixed_injection_styles() -> None:
|
||||
"""Test that tools can mix reserved keywords and annotations."""
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
def tool1(some_val: int, state) -> str:
|
||||
"""Tool with reserved keyword 'state'."""
|
||||
if isinstance(state, dict):
|
||||
return f"reserved state: {state['foo']}"
|
||||
else:
|
||||
return f"reserved state: {getattr(state, 'foo')}"
|
||||
|
||||
def tool2(some_val: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
"""Tool with annotation-based state injection."""
|
||||
return f"annotated state: {state['foo']}"
|
||||
|
||||
def tool3(some_val: int, runtime) -> str:
|
||||
"""Tool with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime)
|
||||
return f"reserved runtime: {runtime.context.get('user_id', 'none') if runtime.context else 'none'}"
|
||||
|
||||
def tool4(some_val: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
"""Tool with annotation-based store injection."""
|
||||
store_val = store.get(("test",), "test_key")
|
||||
return f"annotated store: {store_val.value['foo'] if store_val else 'none'}"
|
||||
|
||||
def tool5(x: int, state, runtime) -> str:
|
||||
"""Tool with both reserved keywords."""
|
||||
assert isinstance(runtime, Runtime)
|
||||
if isinstance(state, dict):
|
||||
return f"both: state={state['foo']}, runtime={runtime.context.get('user_id', 'none') if runtime.context else 'none'}"
|
||||
else:
|
||||
return f"both: state={getattr(state, 'foo')}, runtime={runtime.context.get('user_id', 'none') if runtime.context else 'none'}"
|
||||
|
||||
store = InMemoryStore()
|
||||
store.put(("test",), "test_key", {"foo": "bar"})
|
||||
|
||||
node = ToolNode([tool1, tool2, tool3, tool4, tool5])
|
||||
|
||||
# Verify schemas exclude injected parameters
|
||||
for tool_name, expected_excluded in [
|
||||
("tool1", ["state"]),
|
||||
("tool2", ["state"]),
|
||||
("tool3", ["runtime"]),
|
||||
("tool4", ["store"]),
|
||||
("tool5", ["state", "runtime"]),
|
||||
]:
|
||||
schema = node.tools_by_name[tool_name].get_input_schema()
|
||||
if hasattr(schema, "model_fields"):
|
||||
fields = schema.model_fields
|
||||
else:
|
||||
fields = schema.__fields__
|
||||
for param in expected_excluded:
|
||||
assert param not in fields, (
|
||||
f"'{param}' should be excluded from {tool_name} schema"
|
||||
)
|
||||
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
|
||||
config = RunnableConfig(configurable={"context": {"user_id": "test_user"}})
|
||||
|
||||
# Test each tool
|
||||
test_cases = [
|
||||
("tool1", {"some_val": 1}, "reserved state: baz"),
|
||||
("tool2", {"some_val": 2}, "annotated state: baz"),
|
||||
("tool3", {"some_val": 3}, "reserved runtime: test_user"),
|
||||
("tool4", {"some_val": 4}, "annotated store: bar"),
|
||||
("tool5", {"x": 5}, "both: state=baz, runtime=test_user"),
|
||||
]
|
||||
|
||||
for tool_name, args, expected in test_cases:
|
||||
tool_call = {
|
||||
"name": tool_name,
|
||||
"args": args,
|
||||
"id": f"id_{tool_name}",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("test", tool_calls=[tool_call])
|
||||
result = node.invoke(
|
||||
{"messages": [msg], "foo": "baz"}, store=store, config=config
|
||||
)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == expected, (
|
||||
f"Failed for {tool_name}: got {tool_message.content}, expected {expected}"
|
||||
)
|
||||
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python
|
||||
"""Test that deprecation warnings are properly emitted for InjectedState and InjectedStore."""
|
||||
|
||||
import warnings
|
||||
from typing import Annotated
|
||||
from langchain_core.messages import AIMessage
|
||||
from langgraph.prebuilt import ToolNode, InjectedState, InjectedStore
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
from langgraph.store.base import BaseStore
|
||||
|
||||
|
||||
def test_deprecation_warnings():
|
||||
"""Test that deprecation warnings are emitted for annotation-based injection."""
|
||||
|
||||
# Define tools using deprecated annotations
|
||||
def tool_with_injected_state(x: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
"""Tool using deprecated InjectedState annotation."""
|
||||
return f"state: {state.get('foo', 'none')}"
|
||||
|
||||
def tool_with_injected_store(x: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
"""Tool using deprecated InjectedStore annotation."""
|
||||
return "has store"
|
||||
|
||||
# Define tools using new reserved keywords (should not trigger warnings)
|
||||
def tool_with_reserved_state(x: int, state) -> str:
|
||||
"""Tool using reserved keyword 'state'."""
|
||||
return f"state: {state.get('foo', 'none')}"
|
||||
|
||||
def tool_with_reserved_runtime(x: int, runtime) -> str:
|
||||
"""Tool using reserved keyword 'runtime'."""
|
||||
return "has runtime"
|
||||
|
||||
print("Testing deprecation warnings...")
|
||||
|
||||
# Capture warnings
|
||||
with warnings.catch_warnings(record=True) as w:
|
||||
warnings.simplefilter("always")
|
||||
|
||||
# Create ToolNode with deprecated annotation tools
|
||||
print("\n1. Creating ToolNode with deprecated annotation tools...")
|
||||
node1 = ToolNode([tool_with_injected_state, tool_with_injected_store])
|
||||
|
||||
# Check that warnings were emitted
|
||||
print(f" Got {len(w)} warnings:")
|
||||
for warning in w:
|
||||
print(f" - {warning.message}")
|
||||
assert len(w) >= 1, f"Expected at least 1 warning, got {len(w)}"
|
||||
|
||||
# Check warning messages
|
||||
warning_messages = [str(warning.message) for warning in w]
|
||||
assert any("InjectedState" in msg for msg in warning_messages), "Missing InjectedState warning"
|
||||
assert any("InjectedStore" in msg for msg in warning_messages), "Missing InjectedStore warning"
|
||||
|
||||
print(f" ✓ Emitted {len(w)} deprecation warnings for annotation-based tools")
|
||||
for warning in w:
|
||||
print(f" - {warning.message}")
|
||||
|
||||
# Test that reserved keywords don't trigger warnings
|
||||
with warnings.catch_warnings(record=True) as w:
|
||||
warnings.simplefilter("always")
|
||||
|
||||
print("\n2. Creating ToolNode with reserved keyword tools...")
|
||||
node2 = ToolNode([tool_with_reserved_state, tool_with_reserved_runtime])
|
||||
|
||||
# Check that no warnings were emitted
|
||||
assert len(w) == 0, f"Expected 0 warnings for reserved keywords, got {len(w)}"
|
||||
print(f" ✓ No warnings emitted for reserved keyword tools")
|
||||
|
||||
# Test mixed usage
|
||||
with warnings.catch_warnings(record=True) as w:
|
||||
warnings.simplefilter("always")
|
||||
|
||||
print("\n3. Creating ToolNode with mixed tools...")
|
||||
node3 = ToolNode([
|
||||
tool_with_injected_state, # Should warn
|
||||
tool_with_reserved_state, # Should not warn
|
||||
tool_with_injected_store, # Should warn
|
||||
tool_with_reserved_runtime # Should not warn
|
||||
])
|
||||
|
||||
# Check that only 2 warnings were emitted (for the annotation-based tools)
|
||||
assert len(w) == 2, f"Expected 2 warnings for mixed tools, got {len(w)}"
|
||||
print(f" ✓ Emitted {len(w)} warnings for annotation-based tools only")
|
||||
|
||||
print("\n✅ All deprecation warning tests passed!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_deprecation_warnings()
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test script to verify reserved keyword injection works correctly."""
|
||||
|
||||
from typing import Any
|
||||
from langchain_core.tools import tool
|
||||
from langgraph.prebuilt import ToolNode
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
from langchain_core.messages import AIMessage, ToolCall
|
||||
|
||||
|
||||
# Test tool with reserved keyword 'state'
|
||||
@tool
|
||||
def tool_with_state(x: int, state) -> str:
|
||||
"""Tool that uses reserved keyword 'state'."""
|
||||
return f"x={x}, state_keys={list(state.keys()) if isinstance(state, dict) else 'not_dict'}"
|
||||
|
||||
|
||||
# Test tool with reserved keyword 'runtime'
|
||||
@tool
|
||||
def tool_with_runtime(x: int, runtime) -> str:
|
||||
"""Tool that uses reserved keyword 'runtime'."""
|
||||
has_store = runtime.store is not None if hasattr(runtime, 'store') else False
|
||||
return f"x={x}, has_store={has_store}"
|
||||
|
||||
|
||||
# Test tool with both reserved keywords
|
||||
@tool
|
||||
def tool_with_both(x: int, state, runtime) -> str:
|
||||
"""Tool that uses both reserved keywords."""
|
||||
has_store = runtime.store is not None if hasattr(runtime, 'store') else False
|
||||
return f"x={x}, state_keys={list(state.keys()) if isinstance(state, dict) else 'not_dict'}, has_store={has_store}"
|
||||
|
||||
|
||||
# Test regular tool without injection
|
||||
@tool
|
||||
def regular_tool(x: int, y: str) -> str:
|
||||
"""Regular tool without injection."""
|
||||
return f"x={x}, y={y}"
|
||||
|
||||
|
||||
def test_reserved_keywords():
|
||||
"""Test that reserved keywords work correctly."""
|
||||
|
||||
# Create ToolNode with all test tools
|
||||
tools = [tool_with_state, tool_with_runtime, tool_with_both, regular_tool]
|
||||
node = ToolNode(tools)
|
||||
|
||||
# Check that reserved keywords are detected
|
||||
print("Tool to state args:", node.tool_to_state_args)
|
||||
print("Tool to runtime args:", node.tool_to_runtime_arg)
|
||||
|
||||
# Check tool schemas - reserved keywords should be excluded
|
||||
for tool_name, tool_obj in node.tools_by_name.items():
|
||||
schema = tool_obj.get_input_schema()
|
||||
print(f"\n{tool_name} schema fields:", list(schema.__fields__.keys()))
|
||||
|
||||
# Verify reserved keywords are not in the schema
|
||||
if tool_name == "tool_with_state":
|
||||
assert "state" not in schema.__fields__, f"'state' should be excluded from {tool_name} schema"
|
||||
elif tool_name == "tool_with_runtime":
|
||||
assert "runtime" not in schema.__fields__, f"'runtime' should be excluded from {tool_name} schema"
|
||||
elif tool_name == "tool_with_both":
|
||||
assert "state" not in schema.__fields__, f"'state' should be excluded from {tool_name} schema"
|
||||
assert "runtime" not in schema.__fields__, f"'runtime' should be excluded from {tool_name} schema"
|
||||
|
||||
print("\nAll schema checks passed!")
|
||||
|
||||
# Test actual injection
|
||||
store = InMemoryStore()
|
||||
state = {"messages": [], "foo": "bar"}
|
||||
|
||||
# Create tool calls
|
||||
tool_call1: ToolCall = {
|
||||
"name": "tool_with_state",
|
||||
"args": {"x": 1},
|
||||
"id": "1",
|
||||
"type": "tool_call"
|
||||
}
|
||||
|
||||
tool_call2: ToolCall = {
|
||||
"name": "tool_with_runtime",
|
||||
"args": {"x": 2},
|
||||
"id": "2",
|
||||
"type": "tool_call"
|
||||
}
|
||||
|
||||
tool_call3: ToolCall = {
|
||||
"name": "regular_tool",
|
||||
"args": {"x": 3, "y": "test"},
|
||||
"id": "3",
|
||||
"type": "tool_call"
|
||||
}
|
||||
|
||||
# Test injection
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
config = RunnableConfig(configurable={"context": {"user_id": "test_user"}})
|
||||
|
||||
injected1 = node.inject_tool_args(tool_call1, state, store, config)
|
||||
print(f"\nInjected args for tool_with_state: {injected1['args']}")
|
||||
assert "state" in injected1["args"], "State should be injected"
|
||||
|
||||
injected2 = node.inject_tool_args(tool_call2, state, store, config)
|
||||
print(f"Injected args for tool_with_runtime: {injected2['args']}")
|
||||
assert "runtime" in injected2["args"], "Runtime should be injected"
|
||||
|
||||
injected3 = node.inject_tool_args(tool_call3, state, store, config)
|
||||
print(f"Injected args for regular_tool: {injected3['args']}")
|
||||
assert "state" not in injected3["args"], "State should not be injected"
|
||||
assert "runtime" not in injected3["args"], "Runtime should not be injected"
|
||||
|
||||
print("\nAll injection tests passed!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_reserved_keywords()
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
"""Test reserved keywords for tool injection."""
|
||||
from typing import Annotated, List
|
||||
from langchain_core.messages import AIMessage, AnyMessage
|
||||
from langgraph.prebuilt import ToolNode, InjectedState, InjectedStore
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
from langgraph.store.base import BaseStore
|
||||
from langgraph.graph import MessagesState
|
||||
|
||||
|
||||
def test_tool_node_inject_runtime_reserved_keyword() -> None:
|
||||
"""Test that tools can use 'runtime' as a reserved keyword parameter."""
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
def tool1(some_val: int, runtime) -> str:
|
||||
"""Tool 1 with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime), "runtime should be a Runtime instance"
|
||||
# Access store from runtime
|
||||
if runtime.store:
|
||||
store_val = runtime.store.get(("test",), "test_key")
|
||||
if store_val:
|
||||
return f"val: {some_val}, store: {store_val.value['foo']}"
|
||||
return f"val: {some_val}, no store"
|
||||
|
||||
def tool2(some_val: int, runtime) -> str:
|
||||
"""Tool 2 with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime), "runtime should be a Runtime instance"
|
||||
# Access context from runtime
|
||||
if runtime.context:
|
||||
return f"val: {some_val}, context: {runtime.context.get('user_id', 'unknown')}"
|
||||
return f"val: {some_val}, no context"
|
||||
|
||||
def tool3(x: int, y: str, runtime) -> str:
|
||||
"""Tool 3 with reserved keyword 'runtime' and other params."""
|
||||
assert isinstance(runtime, Runtime), "runtime should be a Runtime instance"
|
||||
has_store = "yes" if runtime.store else "no"
|
||||
has_context = "yes" if runtime.context else "no"
|
||||
return f"x: {x}, y: {y}, store: {has_store}, context: {has_context}"
|
||||
|
||||
store = InMemoryStore()
|
||||
store.put(("test",), "test_key", {"foo": "bar"})
|
||||
|
||||
node = ToolNode([tool1, tool2, tool3])
|
||||
|
||||
# Verify that 'runtime' is excluded from tool schemas
|
||||
for tool in [tool1, tool2, tool3]:
|
||||
schema = node.tools_by_name[tool.__name__].get_input_schema()
|
||||
if hasattr(schema, 'model_fields'):
|
||||
assert "runtime" not in schema.model_fields, f"'runtime' should be excluded from {tool.__name__} schema"
|
||||
else:
|
||||
assert "runtime" not in schema.__fields__, f"'runtime' should be excluded from {tool.__name__} schema"
|
||||
|
||||
# Test with store
|
||||
tool_call = {
|
||||
"name": "tool1",
|
||||
"args": {"some_val": 1},
|
||||
"id": "some 0",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, store=store)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "val: 1, store: bar"
|
||||
|
||||
# Test with context
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
config = RunnableConfig(configurable={"context": {"user_id": "test_user"}})
|
||||
|
||||
tool_call = {
|
||||
"name": "tool2",
|
||||
"args": {"some_val": 2},
|
||||
"id": "some 1",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, config=config)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "val: 2, context: test_user"
|
||||
|
||||
# Test with both store and context
|
||||
tool_call = {
|
||||
"name": "tool3",
|
||||
"args": {"x": 3, "y": "test"},
|
||||
"id": "some 2",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("hi?", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg]}, store=store, config=config)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == "x: 3, y: test, store: yes, context: yes"
|
||||
|
||||
|
||||
def test_tool_node_mixed_injection_styles() -> None:
|
||||
"""Test that tools can mix reserved keywords and annotations."""
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
def tool1(some_val: int, state) -> str:
|
||||
"""Tool with reserved keyword 'state'."""
|
||||
if isinstance(state, dict):
|
||||
return f"reserved state: {state['foo']}"
|
||||
else:
|
||||
return f"reserved state: {getattr(state, 'foo')}"
|
||||
|
||||
def tool2(some_val: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
"""Tool with annotation-based state injection."""
|
||||
return f"annotated state: {state['foo']}"
|
||||
|
||||
def tool3(some_val: int, runtime) -> str:
|
||||
"""Tool with reserved keyword 'runtime'."""
|
||||
assert isinstance(runtime, Runtime)
|
||||
return f"reserved runtime: {runtime.context.get('user_id', 'none') if runtime.context else 'none'}"
|
||||
|
||||
def tool4(some_val: int, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
"""Tool with annotation-based store injection."""
|
||||
store_val = store.get(("test",), "test_key")
|
||||
return f"annotated store: {store_val.value['foo'] if store_val else 'none'}"
|
||||
|
||||
def tool5(x: int, state, runtime) -> str:
|
||||
"""Tool with both reserved keywords."""
|
||||
assert isinstance(runtime, Runtime)
|
||||
if isinstance(state, dict):
|
||||
return f"both: state={state['foo']}, runtime={runtime.context.get('user_id', 'none') if runtime.context else 'none'}"
|
||||
else:
|
||||
return f"both: state={getattr(state, 'foo')}, runtime={runtime.context.get('user_id', 'none') if runtime.context else 'none'}"
|
||||
|
||||
store = InMemoryStore()
|
||||
store.put(("test",), "test_key", {"foo": "bar"})
|
||||
|
||||
node = ToolNode([tool1, tool2, tool3, tool4, tool5])
|
||||
|
||||
# Verify schemas exclude injected parameters
|
||||
for tool_name, expected_excluded in [
|
||||
("tool1", ["state"]),
|
||||
("tool2", ["state"]),
|
||||
("tool3", ["runtime"]),
|
||||
("tool4", ["store"]),
|
||||
("tool5", ["state", "runtime"]),
|
||||
]:
|
||||
schema = node.tools_by_name[tool_name].get_input_schema()
|
||||
if hasattr(schema, 'model_fields'):
|
||||
fields = schema.model_fields
|
||||
else:
|
||||
fields = schema.__fields__
|
||||
for param in expected_excluded:
|
||||
assert param not in fields, f"'{param}' should be excluded from {tool_name} schema"
|
||||
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
config = RunnableConfig(configurable={"context": {"user_id": "test_user"}})
|
||||
|
||||
# Test each tool
|
||||
test_cases = [
|
||||
("tool1", {"some_val": 1}, "reserved state: baz"),
|
||||
("tool2", {"some_val": 2}, "annotated state: baz"),
|
||||
("tool3", {"some_val": 3}, "reserved runtime: test_user"),
|
||||
("tool4", {"some_val": 4}, "annotated store: bar"),
|
||||
("tool5", {"x": 5}, "both: state=baz, runtime=test_user"),
|
||||
]
|
||||
|
||||
for tool_name, args, expected in test_cases:
|
||||
tool_call = {
|
||||
"name": tool_name,
|
||||
"args": args,
|
||||
"id": f"id_{tool_name}",
|
||||
"type": "tool_call",
|
||||
}
|
||||
msg = AIMessage("test", tool_calls=[tool_call])
|
||||
result = node.invoke({"messages": [msg], "foo": "baz"}, store=store, config=config)
|
||||
tool_message = result["messages"][-1]
|
||||
assert tool_message.content == expected, f"Failed for {tool_name}: got {tool_message.content}, expected {expected}"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Testing runtime reserved keyword...")
|
||||
test_tool_node_inject_runtime_reserved_keyword()
|
||||
print("✓ Runtime reserved keyword test passed!")
|
||||
|
||||
print("\nTesting mixed injection styles...")
|
||||
test_tool_node_mixed_injection_styles()
|
||||
print("✓ Mixed injection styles test passed!")
|
||||
|
||||
print("\nAll tests passed!")
|
||||
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
#!/usr/bin/env python
|
||||
"""Verify that the reserved keyword tests work correctly."""
|
||||
|
||||
import sys
|
||||
sys.path.insert(0, 'libs/prebuilt')
|
||||
|
||||
from tests.test_react_agent import test_tool_node_inject_state_reserved_keyword
|
||||
|
||||
print("Running test_tool_node_inject_state_reserved_keyword...")
|
||||
try:
|
||||
test_tool_node_inject_state_reserved_keyword()
|
||||
print("✓ State reserved keyword test passed!")
|
||||
except Exception as e:
|
||||
print(f"✗ State reserved keyword test failed: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
print("\nAll existing tests passed!")
|
||||
Reference in New Issue
Block a user