Compare commits

...
Author SHA1 Message Date
open-swe[bot] 7c09532a51 Apply patch [skip ci] 2025-08-22 15:24:37 +00:00
open-swe[bot] 98201fa98d Apply patch [skip ci] 2025-08-22 15:23:02 +00:00
open-swe[bot] 0334c81eb2 Apply patch [skip ci] 2025-08-22 15:22:51 +00:00
open-swe[bot] 0302621c24 Apply patch [skip ci] 2025-08-22 15:22:40 +00:00
open-swe[bot] 921a8edf66 Apply patch [skip ci] 2025-08-22 15:22:11 +00:00
open-swe[bot] dfee69f4b2 Apply patch [skip ci] 2025-08-22 15:22:00 +00:00
open-swe[bot] fc83746a29 Apply patch [skip ci] 2025-08-22 15:21:48 +00:00
open-swe[bot] 20befcb86a Apply patch [skip ci] 2025-08-22 15:20:42 +00:00
open-swe[bot] d97d18ad35 Apply patch [skip ci] 2025-08-22 15:20:31 +00:00
open-swe[bot] 2f096a3c2e Apply patch [skip ci] 2025-08-22 15:20:20 +00:00
open-swe[bot] d7084a4c83 Apply patch [skip ci] 2025-08-22 15:19:53 +00:00
open-swe[bot] 4b2fc401ff Apply patch [skip ci] 2025-08-22 15:19:41 +00:00
open-swe[bot] a3ebd02e71 Apply patch [skip ci] 2025-08-22 15:19:30 +00:00
open-swe[bot] c76431e57c Apply patch [skip ci] 2025-08-22 15:18:26 +00:00
open-swe[bot] 5b9e65bbbd Apply patch [skip ci] 2025-08-22 15:17:40 +00:00
open-swe[bot] 3326ac2e0d Apply patch [skip ci] 2025-08-22 15:17:30 +00:00
open-swe[bot] 630944b7dd Apply patch [skip ci] 2025-08-22 15:17:19 +00:00
open-swe[bot] 2fc293a2f4 Apply patch [skip ci] 2025-08-22 15:16:54 +00:00
open-swe[bot] f181695856 Apply patch [skip ci] 2025-08-22 15:16:28 +00:00
open-swe[bot] a7ab82fdd2 Apply patch [skip ci] 2025-08-22 15:16:18 +00:00
open-swe[bot] 5632cc003a Apply patch [skip ci] 2025-08-22 15:03:28 +00:00
open-swe[bot] 4d098c4cb5 Apply patch [skip ci] 2025-08-22 15:03:10 +00:00
open-swe[bot] 2d097ee9d1 Apply patch [skip ci] 2025-08-22 15:01:44 +00:00
open-swe[bot] 2f14631c84 Apply patch [skip ci] 2025-08-22 14:58:39 +00:00
open-swe[bot] 9b67fcfc87 Apply patch [skip ci] 2025-08-22 14:58:13 +00:00
open-swe[bot] cde4b9092f Apply patch [skip ci] 2025-08-22 14:57:10 +00:00
open-swe[bot] 620adb3bcf Apply patch [skip ci] 2025-08-22 14:56:50 +00:00
open-swe[bot] c8dd449d1a Apply patch [skip ci] 2025-08-22 14:56:02 +00:00
open-swe[bot] 6eef693d6e Apply patch [skip ci] 2025-08-22 14:55:46 +00:00
open-swe[bot] 193b4c34ec Apply patch [skip ci] 2025-08-22 14:54:53 +00:00
open-swe[bot] b242b5499c Apply patch [skip ci] 2025-08-22 14:54:29 +00:00
open-swe[bot] 93bdd275f0 Apply patch [skip ci] 2025-08-22 13:31:54 +00:00
open-swe[bot] 80c100273c Apply patch [skip ci] 2025-08-22 13:31:34 +00:00
open-swe[bot] 4096463586 Apply patch [skip ci] 2025-08-22 13:30:13 +00:00
open-swe[bot] dc446846e8 Apply patch [skip ci] 2025-08-22 13:29:36 +00:00
open-swe[bot] b3cbd71ed5 Apply patch [skip ci] 2025-08-22 13:28:58 +00:00
open-swe[bot] 3d2bc9069e Apply patch [skip ci] 2025-08-22 13:28:09 +00:00
open-swe[bot] 89e1933ab0 Apply patch [skip ci] 2025-08-22 13:27:53 +00:00
open-swe[bot] 5197b6261b Apply patch [skip ci] 2025-08-22 13:27:38 +00:00
open-swe[bot] bf2e321fb3 Apply patch [skip ci] 2025-08-22 13:27:22 +00:00
open-swe[bot] 767d34872f Apply patch [skip ci] 2025-08-22 13:27:08 +00:00
open-swe[bot] acdc1f7f6b Apply patch [skip ci] 2025-08-22 13:26:40 +00:00
open-swe[bot] 6c3f84bc4d Apply patch [skip ci] 2025-08-22 13:25:54 +00:00
open-swe[bot] 223150870c Apply patch [skip ci] 2025-08-22 13:23:45 +00:00
open-swe[bot] 50eb12d14f Apply patch [skip ci] 2025-08-22 13:22:13 +00:00
open-swe[bot] d8ce3d71cf Apply patch [skip ci] 2025-08-22 13:21:28 +00:00
open-swe[bot] 1adef5b6f1 Apply patch [skip ci] 2025-08-22 13:20:35 +00:00
open-swe[bot] 944947e889 Apply patch [skip ci] 2025-08-22 13:20:00 +00:00
open-swe[bot] a38727133b Apply patch [skip ci] 2025-08-22 13:19:24 +00:00
open-swe[bot] 5d929c30b4 Apply patch [skip ci] 2025-08-22 13:18:32 +00:00
open-swe[bot] fb091de4f2 Apply patch [skip ci] 2025-08-22 13:17:09 +00:00
open-swe[bot] 9eacc9ff67 Apply patch [skip ci] 2025-08-22 13:16:32 +00:00
open-swe[bot] 4e57940152 Apply patch [skip ci] 2025-08-22 13:15:41 +00:00
open-swe[bot] c227b0dea1 Apply patch [skip ci] 2025-08-22 13:14:05 +00:00
open-swe[bot] b6e8816e7d Apply patch [skip ci] 2025-08-22 13:13:15 +00:00
open-swe[bot] 779d454fcc Apply patch [skip ci] 2025-08-22 13:12:28 +00:00
open-swe[bot] 9423a99515 Apply patch [skip ci] 2025-08-22 13:11:55 +00:00
open-swe[bot] 4c04e023c8 Apply patch [skip ci] 2025-08-22 13:11:14 +00:00
open-swe[bot] 4566b2aed4 Apply patch [skip ci] 2025-08-22 13:10:45 +00:00
open-swe[bot] 55f6f052b6 Apply patch [skip ci] 2025-08-22 13:08:00 +00:00
open-swe[bot] c22a86f11e Apply patch [skip ci] 2025-08-22 13:06:23 +00:00
open-swe[bot] 6fee6811a0 Apply patch [skip ci] 2025-08-22 13:05:28 +00:00
open-swe[bot] 85180f5054 Apply patch [skip ci] 2025-08-22 13:02:01 +00:00
open-swe[bot] 6e30e457d4 Apply patch [skip ci] 2025-08-22 12:58:40 +00:00
open-swe[bot] f6512d5357 Apply patch [skip ci] 2025-08-22 12:58:19 +00:00
open-swe[bot] 3408e525f8 Apply patch [skip ci] 2025-08-22 12:57:52 +00:00
open-swe[bot] 777ab1123f Apply patch [skip ci] 2025-08-22 12:53:36 +00:00
open-swe[bot] a5a0286568 Apply patch [skip ci] 2025-08-22 12:52:47 +00:00
open-swe[bot] f86427fe37 Apply patch [skip ci] 2025-08-22 12:49:25 +00:00
open-swe[bot] 69f4ab0a79 Apply patch [skip ci] 2025-08-22 12:48:29 +00:00
open-swe[bot] 9284412aac Apply patch [skip ci] 2025-08-22 12:47:23 +00:00
open-swe[bot] 220b3b1d33 Apply patch [skip ci] 2025-08-22 12:47:00 +00:00
open-swe[bot] d0be6843ba Apply patch [skip ci] 2025-08-22 12:45:51 +00:00
open-swe[bot] 7fb9f7c870 Apply patch [skip ci] 2025-08-22 12:44:56 +00:00
open-swe[bot] cc54df296e Apply patch [skip ci] 2025-08-22 12:42:37 +00:00
open-swe[bot] a5b61361af Apply patch [skip ci] 2025-08-22 12:41:04 +00:00
open-swe[bot] 2f96c5c021 Apply patch [skip ci] 2025-08-22 12:40:44 +00:00
open-swe[bot] 0915784a87 Apply patch [skip ci] 2025-08-22 12:40:15 +00:00
open-swe[bot] 6952d3a1e0 Apply patch [skip ci] 2025-08-22 12:39:17 +00:00
open-swe[bot] d411013b6f Apply patch [skip ci] 2025-08-22 12:37:24 +00:00
open-swe[bot] ce9473293f Apply patch [skip ci] 2025-08-22 12:36:40 +00:00
open-swe[bot] 7fd8666608 Apply patch [skip ci] 2025-08-22 12:35:55 +00:00
open-swe[bot] adf42e7ca1 Apply patch [skip ci] 2025-08-22 12:35:09 +00:00
open-swe[bot] b0b6e3d91f Apply patch [skip ci] 2025-08-22 12:34:26 +00:00
open-swe[bot] 30e00b8c7b Apply patch [skip ci] 2025-08-22 12:33:34 +00:00
15 changed files with 1221 additions and 37 deletions
+47
View File
@@ -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'))}")
+47
View File
@@ -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)}")
+49
View File
@@ -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.")
+4 -2
View File
@@ -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.
:::
+5 -5
View File
@@ -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.
+13 -9
View File
@@ -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.
:::
:::
+8 -4
View File
@@ -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 ```
```
+318 -14
View File
@@ -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
+136
View File
@@ -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}"
)
+90
View File
@@ -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()
+117
View File
@@ -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()
+182
View File
@@ -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!")
+17
View File
@@ -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!")