Compare commits

...
Author SHA1 Message Date
Eugene Yurtsev 9d79f94742 update 2025-07-16 13:59:53 -04:00
4 changed files with 272 additions and 59 deletions
@@ -187,7 +187,7 @@ def _should_bind_tools(
return False
def _get_model(model: LanguageModelLike) -> BaseChatModel:
def _get_underlying_model(model: LanguageModelLike) -> BaseChatModel:
"""Get the underlying model from a RunnableBinding or return the model itself."""
if isinstance(model, RunnableSequence):
model = next(
@@ -241,8 +241,11 @@ def _validate_chat_history(
raise ValueError(error_message)
DynamicModel = Callable[[StateSchemaType, RunnableConfig], BaseChatModel]
def create_react_agent(
model: Union[str, LanguageModelLike],
model: Union[str, LanguageModelLike, DynamicModel],
tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
*,
prompt: Optional[Prompt] = None,
@@ -432,6 +435,8 @@ def create_react_agent(
tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
tool_classes = list(tool_node.tools_by_name.values())
tool_calling_enabled = len(tool_classes) > 0
if isinstance(model, str):
try:
from langchain.chat_models import ( # type: ignore[import-not-found]
@@ -441,18 +446,52 @@ def create_react_agent(
raise ImportError(
"Please install langchain (`pip install langchain`) to use '<provider>:<model>' string syntax for `model` parameter."
)
model_instance = cast(BaseChatModel, init_chat_model(model))
elif isinstance(model, Runnable):
model_instance = _get_underlying_model(model)
elif callable(model):
model_instance = None
else:
raise TypeError(
"Expected `model` to be a string, LanguageModelLike, "
f"or callable, got {type(model)}"
)
model = cast(BaseChatModel, init_chat_model(model))
# If we have a static model, we'll attempt to bind tools to it.
if model_instance:
# Apply tool binding for static model
if (
_should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools))
and len(tool_classes + llm_builtin_tools) > 0
):
model_instance = cast(BaseChatModel, model_instance).bind_tools(
tool_classes + llm_builtin_tools
) # type: ignore[operator]
static_model: Optional[BaseChatModel] = (
_get_prompt_runnable(prompt) | model_instance
)
else:
static_model = None
tool_calling_enabled = len(tool_classes) > 0
def _resolve_model(state: StateSchema, config: RunnableConfig) -> BaseChatModel:
"""Resolve the model, handling both static and dynamic models."""
if static_model:
return static_model
if (
_should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools))
and len(tool_classes + llm_builtin_tools) > 0
):
model = cast(BaseChatModel, model).bind_tools(tool_classes + llm_builtin_tools) # type: ignore[operator]
# If we have a dynamic model, we need to resolve it at runtime
resolved_model = model(state, config) # type: ignore[call-arg]
# Apply tools binding if needed
if (
_should_bind_tools(
resolved_model, tool_classes, num_builtin=len(llm_builtin_tools)
)
and len(tool_classes + llm_builtin_tools) > 0
):
resolved_model = cast(BaseChatModel, resolved_model).bind_tools(
tool_classes + llm_builtin_tools
)
model_runnable = _get_prompt_runnable(prompt) | model
return cast(BaseChatModel, _get_prompt_runnable(prompt) | resolved_model)
# If any of the tools are configured to return_directly after running,
# our graph needs to check if these were called
@@ -504,7 +543,8 @@ def create_react_agent(
# Define the function that calls the model
def call_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
state = _get_model_input_state(state)
response = cast(AIMessage, model_runnable.invoke(state, config))
runnable = _resolve_model(state, config)
response = runnable.invoke(state, config)
# add agent name to the AIMessage
response.name = name
@@ -522,7 +562,8 @@ def create_react_agent(
async def acall_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
state = _get_model_input_state(state)
response = cast(AIMessage, await model_runnable.ainvoke(state, config))
runnable = _resolve_model(state, config)
response = await runnable.ainvoke(state, config)
# add agent name to the AIMessage
response.name = name
if _are_more_steps_needed(state, response):
@@ -561,13 +602,17 @@ def create_react_agent(
def generate_structured_response(
state: StateSchema, config: RunnableConfig
) -> StateSchema:
"""Generate a structured response from the model."""
messages = _get_state_value(state, "messages")
structured_response_schema = response_format
if isinstance(response_format, tuple):
system_prompt, structured_response_schema = response_format
messages = [SystemMessage(content=system_prompt)] + list(messages)
model_with_structured_output = _get_model(model).with_structured_output(
# We need to re-bind structured outputs to the model
model_instance_ = _resolve_model(state, config)
underlying_model = _get_underlying_model(model_instance_)
model_with_structured_output = underlying_model.with_structured_output(
cast(StructuredResponseSchema, structured_response_schema)
)
response = model_with_structured_output.invoke(messages, config)
@@ -576,13 +621,17 @@ def create_react_agent(
async def agenerate_structured_response(
state: StateSchema, config: RunnableConfig
) -> StateSchema:
"""Generate a structured response from the model."""
messages = _get_state_value(state, "messages")
structured_response_schema = response_format
if isinstance(response_format, tuple):
system_prompt, structured_response_schema = response_format
messages = [SystemMessage(content=system_prompt)] + list(messages)
model_with_structured_output = _get_model(model).with_structured_output(
# We need to re-bind structured outputs to the model
model_instance_ = _resolve_model(state, config)
underlying_model = _get_underlying_model(model_instance_)
model_with_structured_output = underlying_model.with_structured_output(
cast(StructuredResponseSchema, structured_response_schema)
)
response = await model_with_structured_output.ainvoke(messages, config)
@@ -651,7 +700,7 @@ def create_react_agent(
if post_model_hook is not None:
return "post_model_hook"
tool_calls = [
tool_node.inject_tool_args(call, state, store) # type: ignore[arg-type]
tool_node.inject_tool_args(call, state, store)
for call in last_message.tool_calls
]
return [Send("tools", [tool_call]) for tool_call in tool_calls]
@@ -734,7 +783,7 @@ def create_react_agent(
if pending_tool_calls:
pending_tool_calls = [
tool_node.inject_tool_args(call, state, store) # type: ignore[arg-type]
tool_node.inject_tool_args(call, state, store)
for call in pending_tool_calls
]
return [Send("tools", [tool_call]) for tool_call in pending_tool_calls]
+153 -36
View File
@@ -35,7 +35,8 @@ import asyncio
import inspect
import json
from copy import copy, deepcopy
from dataclasses import replace
from dataclasses import dataclass, replace
from itertools import repeat
from typing import (
Any,
Callable,
@@ -73,6 +74,7 @@ from typing_extensions import Annotated, get_args, get_origin
from langgraph.errors import GraphBubbleUp
from langgraph.store.base import BaseStore
from langgraph.types import Command, Send
from langgraph.typing import StateLike
from langgraph.utils.runnable import RunnableCallable
INVALID_TOOL_NAME_ERROR_TEMPLATE = (
@@ -81,6 +83,43 @@ INVALID_TOOL_NAME_ERROR_TEMPLATE = (
TOOL_CALL_ERROR_TEMPLATE = "Error: {error}\n Please fix your mistakes."
@dataclass(frozen=True)
class ToolResolver:
"""Encapsulates tool metadata for thread-safe tool execution.
This class holds all the precomputed metadata needed for tool execution,
including tool instances, state injection mappings, and store injection settings.
"""
tools_by_name: dict[str, BaseTool]
tool_to_state_args: dict[str, dict[str, Optional[str]]]
tool_to_store_arg: dict[str, Optional[str]]
@staticmethod
def from_tools(tools: Sequence[Union[BaseTool, Callable]]) -> "ToolResolver":
"""Create a ToolResolver from a sequence of tools.
Args:
tools: Sequence of tools to process. Can be BaseTool instances or
callables that will be converted to tools.
Returns:
A ToolResolver containing all the metadata for the provided tools.
"""
tools_by_name = {}
tool_to_state_args = {}
tool_to_store_arg = {}
for tool in tools:
if not isinstance(tool, BaseTool):
tool = create_tool(tool)
tools_by_name[tool.name] = tool
tool_to_state_args[tool.name] = _get_state_args(tool)
tool_to_store_arg[tool.name] = _get_store_arg(tool)
return ToolResolver(tools_by_name, tool_to_state_args, tool_to_store_arg)
def msg_content_output(output: Any) -> Union[str, list[dict]]:
"""Convert tool output to valid message content format.
@@ -244,8 +283,11 @@ class ToolNode(RunnableCallable):
Tool calls can also be passed directly as a list of `ToolCall` dicts.
Args:
tools: A sequence of tools that can be invoked by this node. Tools can be
BaseTool instances or plain functions that will be converted to tools.
tools: Either a sequence of tools that can be invoked by this node, or a
callable that returns tools dynamically based on input, config, and store.
Static tools can be BaseTool instances or plain functions that will be
converted to tools. Dynamic tool providers receive (input, config, store)
and should return a sequence of tools for that specific execution.
name: The name identifier for this node in the graph. Used for debugging
and visualization. Defaults to "tools".
tags: Optional metadata tags to associate with the node for filtering
@@ -316,7 +358,13 @@ class ToolNode(RunnableCallable):
def __init__(
self,
tools: Sequence[Union[BaseTool, Callable]],
tools: Union[
Sequence[Union[BaseTool, Callable]],
Callable[
[StateLike, RunnableConfig, Optional[BaseStore]],
Sequence[Union[BaseTool, Callable]],
],
],
*,
name: str = "tools",
tags: Optional[list[str]] = None,
@@ -328,24 +376,64 @@ class ToolNode(RunnableCallable):
"""Initialize the ToolNode with the provided tools and configuration.
Args:
tools: Sequence of tools to make available for execution.
tools: Either a sequence of tools to make available for execution,
or a callable that returns tools dynamically based on input,
config, and store.
name: Node name for graph identification.
tags: Optional metadata tags.
handle_tool_errors: Error handling configuration.
messages_key: State key containing messages.
"""
super().__init__(self._func, self._afunc, name=name, tags=tags, trace=False)
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.handle_tool_errors = handle_tool_errors
self.messages_key = messages_key
for tool_ in tools:
if not isinstance(tool_, BaseTool):
tool_ = create_tool(tool_)
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_)
if callable(tools):
# Dynamic tool provider
self._tool_provider_fn = tools
self._static_resolver: Optional[ToolResolver] = None
# Likely migrate to property and raise a RunTimeError
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]] = {}
else:
# Static tools
self._tool_provider_fn = None
self._static_resolver = ToolResolver.from_tools(tools)
self.tools_by_name = self._static_resolver.tools_by_name
self.tool_to_state_args = self._static_resolver.tool_to_state_args
self.tool_to_store_arg = self._static_resolver.tool_to_store_arg
def _get_resolver(
self,
input: Union[list[AnyMessage], dict[str, Any], BaseModel],
config: Optional[RunnableConfig],
store: Optional[BaseStore],
) -> ToolResolver:
"""Get the appropriate ToolResolver for this execution.
Args:
input: The input to the tool node
config: The runnable configuration
store: The optional store instance
Returns:
A ToolResolver containing the tools and metadata for this execution.
Raises:
RuntimeError: If no tools are configured.
"""
if self._tool_provider_fn:
# Dynamic tools compute fresh for each run
# Note: config may be None during _parse_input,
# but tool provider should handle this
tools = self._tool_provider_fn(input, config, store)
return ToolResolver.from_tools(tools)
elif self._static_resolver:
# Static tools - use cached resolver
return self._static_resolver
else:
raise RuntimeError("ToolNode has no tools configured")
def _func(
self,
@@ -358,12 +446,16 @@ 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)
resolver = self._get_resolver(input, config, store)
config_list = get_config_list(config, len(tool_calls))
input_types = [input_type] * len(tool_calls)
with get_executor_for_config(config) as executor:
outputs = [
*executor.map(self._run_one, tool_calls, input_types, config_list)
*executor.map(
lambda args: self._run_one(*args),
list(zip(tool_calls, input_types, config_list, repeat(resolver))),
)
]
return self._combine_tool_outputs(outputs, input_type)
@@ -379,9 +471,10 @@ 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)
resolver = self._get_resolver(input, config, store)
outputs = await asyncio.gather(
*(self._arun_one(call, input_type, config) for call in tool_calls)
*(self._arun_one(call, input_type, config, resolver) for call in tool_calls)
)
return self._combine_tool_outputs(outputs, input_type)
@@ -435,20 +528,23 @@ class ToolNode(RunnableCallable):
call: ToolCall,
input_type: Literal["list", "dict", "tool_calls"],
config: RunnableConfig,
resolver: ToolResolver,
) -> ToolMessage:
if invalid_tool_message := self._validate_tool_call(call):
if invalid_tool_message := self._validate_tool_call(call, resolver):
return invalid_tool_message
try:
input = {**call, **{"type": "tool_call"}}
response = self.tools_by_name[call["name"]].invoke(input, config)
response = resolver.tools_by_name[call["name"]].invoke(input, config)
# GraphInterrupt is a special exception that will always be raised.
# It can be triggered in the following scenarios:
# (1) a NodeInterrupt is raised inside a tool
# (2) a NodeInterrupt is raised inside a graph node for a graph called as a tool
# (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph called as a tool
# (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture)
# (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph
# called as a tool
# ---
# (2) and (3) can happen in a "supervisor w/ tools" multi-agent architecture
except GraphBubbleUp as e:
raise e
except Exception as e:
@@ -490,20 +586,23 @@ class ToolNode(RunnableCallable):
call: ToolCall,
input_type: Literal["list", "dict", "tool_calls"],
config: RunnableConfig,
resolver: ToolResolver,
) -> ToolMessage:
if invalid_tool_message := self._validate_tool_call(call):
if invalid_tool_message := self._validate_tool_call(call, resolver):
return invalid_tool_message
try:
input = {**call, **{"type": "tool_call"}}
response = await self.tools_by_name[call["name"]].ainvoke(input, config)
response = await resolver.tools_by_name[call["name"]].ainvoke(input, config)
# GraphInterrupt is a special exception that will always be raised.
# It can be triggered in the following scenarios:
# (1) a NodeInterrupt is raised inside a tool
# (2) a NodeInterrupt is raised inside a graph node for a graph called as a tool
# (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph called as a tool
# (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture)
# (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph
# called as a tool
# ---
# (2) and (3) can happen in a "supervisor w/ tools" multi-agent architecture
except GraphBubbleUp as e:
raise e
except Exception as e:
@@ -549,7 +648,10 @@ class ToolNode(RunnableCallable):
BaseModel,
],
store: Optional[BaseStore],
config: RunnableConfig,
) -> Tuple[list[ToolCall], Literal["list", "dict", "tool_calls"]]:
"""Parse the input to extract tool calls and determine input type."""
input_type: Literal["list", "dict", "tool_calls"]
if isinstance(input, list):
if isinstance(input[-1], dict) and input[-1].get("type") == "tool_call":
input_type = "tool_calls"
@@ -573,17 +675,22 @@ class ToolNode(RunnableCallable):
except StopIteration:
raise ValueError("No AIMessage found in input")
# For _parse_input, we need to compute the resolver to inject tool args
# We pass None for config since we don't have it yet at this stage
resolver = self._get_resolver(input, config=config, store=store)
tool_calls = [
self.inject_tool_args(call, input, store)
self.inject_tool_args(call, input, store, resolver)
for call in latest_ai_message.tool_calls
]
return tool_calls, input_type
def _validate_tool_call(self, call: ToolCall) -> Optional[ToolMessage]:
if (requested_tool := call["name"]) not in self.tools_by_name:
def _validate_tool_call(
self, call: ToolCall, resolver: ToolResolver
) -> Optional[ToolMessage]:
if (requested_tool := call["name"]) not in resolver.tools_by_name:
content = INVALID_TOOL_NAME_ERROR_TEMPLATE.format(
requested_tool=requested_tool,
available_tools=", ".join(self.tools_by_name.keys()),
available_tools=", ".join(resolver.tools_by_name.keys()),
)
return ToolMessage(
content, name=requested_tool, tool_call_id=call["id"], status="error"
@@ -599,8 +706,9 @@ class ToolNode(RunnableCallable):
dict[str, Any],
BaseModel,
],
resolver: ToolResolver,
) -> ToolCall:
state_args = self.tool_to_state_args[tool_call["name"]]
state_args = resolver.tool_to_state_args[tool_call["name"]]
if state_args and isinstance(input, list):
required_fields = list(state_args.values())
if (
@@ -637,9 +745,9 @@ class ToolNode(RunnableCallable):
return tool_call
def _inject_store(
self, tool_call: ToolCall, store: Optional[BaseStore]
self, tool_call: ToolCall, store: Optional[BaseStore], resolver: ToolResolver
) -> ToolCall:
store_arg = self.tool_to_store_arg[tool_call["name"]]
store_arg = resolver.tool_to_store_arg[tool_call["name"]]
if not store_arg:
return tool_call
@@ -664,6 +772,8 @@ class ToolNode(RunnableCallable):
BaseModel,
],
store: Optional[BaseStore],
# TODO(EUGENE): Potentially breaking change?
resolver: Optional[ToolResolver] = None,
) -> ToolCall:
"""Inject graph state and store into tool call arguments.
@@ -683,6 +793,8 @@ 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.
resolver: The ToolResolver instance containing metadata about available
tools, including their state and store injection requirements.
Returns:
A new ToolCall dictionary with the same structure as the input but with
@@ -698,12 +810,17 @@ class ToolNode(RunnableCallable):
The injection is performed on a copy of the tool call to avoid mutating
the original.
"""
if tool_call["name"] not in self.tools_by_name:
resolver_ = resolver or self._static_resolver
if tool_call["name"] not in resolver_.tools_by_name:
return tool_call
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)
tool_call_with_state = self._inject_state(tool_call_copy, input, resolver_)
tool_call_with_store = self._inject_store(
tool_call_with_state, store, resolver_
)
return tool_call_with_store
def _validate_tool_command(
+8 -7
View File
@@ -5,6 +5,7 @@ from functools import partial
from typing import (
Annotated,
List,
Literal,
Optional,
Type,
TypeVar,
@@ -40,7 +41,7 @@ from langgraph.prebuilt.chat_agent_executor import (
AgentState,
AgentStatePydantic,
StateSchemaType,
_get_model,
_get_underlying_model,
_should_bind_tools,
_validate_chat_history,
)
@@ -470,7 +471,7 @@ def test__infer_handled_types() -> None:
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_react_agent_with_structured_response(version: str) -> None:
def test_react_agent_with_structured_response(version: Literal["v1", "v2"]) -> None:
class WeatherResponse(BaseModel):
temperature: float = Field(description="The temperature in fahrenheit")
@@ -1346,7 +1347,7 @@ def test_should_bind_tools(tool_style: str) -> None:
def test_get_model() -> None:
model = FakeToolCallingModel(tool_calls=[])
assert _get_model(model) == model
assert _get_underlying_model(model) == model
@dec_tool
def some_tool(some_val: int) -> str:
@@ -1354,18 +1355,18 @@ def test_get_model() -> None:
return "meow"
model_with_tools = model.bind_tools([some_tool])
assert _get_model(model_with_tools) == model
assert _get_underlying_model(model_with_tools) == model
seq = model | RunnableLambda(lambda message: message)
assert _get_model(seq) == model
assert _get_underlying_model(seq) == model
seq_with_tools = model.bind_tools([some_tool]) | RunnableLambda(
lambda message: message
)
assert _get_model(seq_with_tools) == model
assert _get_underlying_model(seq_with_tools) == model
with pytest.raises(TypeError):
_get_model(RunnableLambda(lambda message: message))
_get_underlying_model(RunnableLambda(lambda message: message))
def test_pre_model_hook() -> None:
+46
View File
@@ -1129,3 +1129,49 @@ def test_tool_node_parent_command_with_send():
graph=Command.PARENT,
)
]
async def test_tool_node_dynamic_tools_basic() -> None:
"""Test basic dynamic tool functionality."""
@dec_tool()
def answer_to_life() -> int:
"""A very nice dynamic tool."""
return 42
def tool_provider(state, config, store) -> list:
"""Provide dynamic tools based on state or config."""
return [answer_to_life]
tool_node = ToolNode(tool_provider)
result = await tool_node.ainvoke(
[
{
"name": "answer_to_life",
"args": {},
"id": "1",
"type": "tool_call",
}
]
)
assert result == {
"messages": [
ToolMessage(
content="42",
tool_call_id="1",
name="answer_to_life",
)
]
}
# Test invoking a tool that's not defined
with pytest.raises(ValueError):
await tool_node.ainvoke(
{
"name": "not_available",
"args": {},
"id": "1",
"type": "tool_call",
}
)