mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-31 20:29:46 +02:00
Remove prebuilt
This commit is contained in:
Nuno Campos
@@ -1,21 +0,0 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2024 LangChain, Inc.
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -1,78 +0,0 @@
|
||||
.PHONY: all format lint test test_watch integration_tests spell_check spell_fix benchmark profile
|
||||
|
||||
# Default target executed when no arguments are given to make.
|
||||
all: help
|
||||
|
||||
######################
|
||||
# TESTING AND COVERAGE
|
||||
######################
|
||||
|
||||
start-postgres:
|
||||
docker compose -f tests/compose-postgres.yml up -V --force-recreate --wait --remove-orphans
|
||||
|
||||
stop-postgres:
|
||||
docker compose -f tests/compose-postgres.yml down -v
|
||||
|
||||
TEST ?= .
|
||||
|
||||
test:
|
||||
make start-postgres && poetry run pytest $(TEST); \
|
||||
EXIT_CODE=$$?; \
|
||||
make stop-postgres; \
|
||||
exit $$EXIT_CODE
|
||||
|
||||
test_watch:
|
||||
make start-postgres && poetry run ptw $(TEST); \
|
||||
EXIT_CODE=$$?; \
|
||||
make stop-postgres; \
|
||||
exit $$EXIT_CODE
|
||||
|
||||
######################
|
||||
# LINTING AND FORMATTING
|
||||
######################
|
||||
|
||||
# Define a variable for Python and notebook files.
|
||||
PYTHON_FILES=.
|
||||
MYPY_CACHE=.mypy_cache
|
||||
lint format: PYTHON_FILES=.
|
||||
lint_diff format_diff: PYTHON_FILES=$(shell git diff --name-only --relative --diff-filter=d main . | grep -E '\.py$$|\.ipynb$$')
|
||||
lint_package: PYTHON_FILES=langgraph
|
||||
lint_tests: PYTHON_FILES=tests
|
||||
lint_tests: MYPY_CACHE=.mypy_cache_test
|
||||
|
||||
lint lint_diff lint_package lint_tests:
|
||||
poetry run ruff check .
|
||||
[ "$(PYTHON_FILES)" = "" ] || poetry run ruff format $(PYTHON_FILES) --diff
|
||||
[ "$(PYTHON_FILES)" = "" ] || poetry run ruff check --select I $(PYTHON_FILES)
|
||||
[ "$(PYTHON_FILES)" = "" ] || mkdir -p $(MYPY_CACHE)
|
||||
[ "$(PYTHON_FILES)" = "" ] || poetry run mypy langgraph --cache-dir $(MYPY_CACHE)
|
||||
|
||||
format format_diff:
|
||||
poetry run ruff format $(PYTHON_FILES)
|
||||
poetry run ruff check --select I --fix $(PYTHON_FILES)
|
||||
|
||||
spell_check:
|
||||
poetry run codespell --toml pyproject.toml
|
||||
|
||||
spell_fix:
|
||||
poetry run codespell --toml pyproject.toml -w
|
||||
|
||||
|
||||
######################
|
||||
# HELP
|
||||
######################
|
||||
|
||||
help:
|
||||
@echo '===================='
|
||||
@echo '-- DOCUMENTATION --'
|
||||
|
||||
@echo '-- LINTING --'
|
||||
@echo 'format - run code formatters'
|
||||
@echo 'lint - run linters'
|
||||
@echo 'spell_check - run codespell on the project'
|
||||
@echo 'spell_fix - run codespell on the project and fix the errors'
|
||||
@echo '-- TESTS --'
|
||||
@echo 'coverage - run unit tests and generate coverage report'
|
||||
@echo 'test - run unit tests'
|
||||
@echo 'test TEST_FILE=<test_file> - run all tests in file'
|
||||
@echo 'test_watch - run unit tests in watch mode'
|
||||
@@ -1,117 +0,0 @@
|
||||
# LangGraph Prebuilt
|
||||
|
||||
This library defines high-level APIs for creating and executing LangGraph agents and tools.
|
||||
|
||||
> [!IMPORTANT]
|
||||
> This library is meant to be bundled with `langgraph`, don't install it directly
|
||||
|
||||
## Agents
|
||||
|
||||
`langgraph-prebuilt` provides an [implementation](https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.chat_agent_executor.create_react_agent) of a tool-calling [ReAct-style](https://langchain-ai.github.io/langgraph/concepts/agentic_concepts/#react-implementation) agent - `create_react_agent`:
|
||||
|
||||
```bash
|
||||
pip install langchain-anthropic
|
||||
```
|
||||
|
||||
```python
|
||||
from langchain_anthropic import ChatAnthropic
|
||||
from langgraph.prebuilt import create_react_agent
|
||||
|
||||
# Define the tools for the agent to use
|
||||
def search(query: str):
|
||||
"""Call to surf the web."""
|
||||
# This is a placeholder, but don't tell the LLM that...
|
||||
if "sf" in query.lower() or "san francisco" in query.lower():
|
||||
return "It's 60 degrees and foggy."
|
||||
return "It's 90 degrees and sunny."
|
||||
|
||||
tools = [search]
|
||||
model = ChatAnthropic(model="claude-3-7-sonnet-latest")
|
||||
|
||||
app = create_react_agent(model, tools)
|
||||
# run the agent
|
||||
app.invoke(
|
||||
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
|
||||
)
|
||||
```
|
||||
|
||||
## Tools
|
||||
|
||||
### ToolNode
|
||||
|
||||
`langgraph-prebuilt` provides an [implementation](https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.tool_node.ToolNode) of a node that executes tool calls - `ToolNode`:
|
||||
|
||||
```python
|
||||
from langgraph.prebuilt import ToolNode
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
def search(query: str):
|
||||
"""Call to surf the web."""
|
||||
# This is a placeholder, but don't tell the LLM that...
|
||||
if "sf" in query.lower() or "san francisco" in query.lower():
|
||||
return "It's 60 degrees and foggy."
|
||||
return "It's 90 degrees and sunny."
|
||||
|
||||
tool_node = ToolNode([search])
|
||||
tool_calls = [{"name": "search", "args": {"query": "what is the weather in sf"}, "id": "1"}]
|
||||
ai_message = AIMessage(content="", tool_calls=tool_calls)
|
||||
# execute tool call
|
||||
tool_node.invoke({"messages": [ai_message]})
|
||||
```
|
||||
|
||||
### ValidationNode
|
||||
|
||||
`langgraph-prebuilt` provides an [implementation](https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.tool_validator.ValidationNode) of a node that validates tool calls against a pydantic schema - `ValidationNode`:
|
||||
|
||||
```python
|
||||
from pydantic import BaseModel, field_validator
|
||||
from langgraph.prebuilt import ValidationNode
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
|
||||
class SelectNumber(BaseModel):
|
||||
a: int
|
||||
|
||||
@field_validator("a")
|
||||
def a_must_be_meaningful(cls, v):
|
||||
if v != 37:
|
||||
raise ValueError("Only 37 is allowed")
|
||||
return v
|
||||
|
||||
validation_node = ValidationNode([SelectNumber])
|
||||
validation_node.invoke({
|
||||
"messages": [AIMessage("", tool_calls=[{"name": "SelectNumber", "args": {"a": 42}, "id": "1"}])]
|
||||
})
|
||||
```
|
||||
|
||||
## Agent Inbox
|
||||
|
||||
The library contains schemas for using the [Agent Inbox](https://github.com/langchain-ai/agent-inbox) with LangGraph agents. Learn more about how to use Agent Inbox [here](https://github.com/langchain-ai/agent-inbox#interrupts).
|
||||
|
||||
```python
|
||||
from langgraph.types import interrupt
|
||||
from langgraph.prebuilt.interrupt import HumanInterrupt, HumanResponse
|
||||
|
||||
def my_graph_function():
|
||||
# Extract the last tool call from the `messages` field in the state
|
||||
tool_call = state["messages"][-1].tool_calls[0]
|
||||
# Create an interrupt
|
||||
request: HumanInterrupt = {
|
||||
"action_request": {
|
||||
"action": tool_call['name'],
|
||||
"args": tool_call['args']
|
||||
},
|
||||
"config": {
|
||||
"allow_ignore": True,
|
||||
"allow_respond": True,
|
||||
"allow_edit": False,
|
||||
"allow_accept": False
|
||||
},
|
||||
"description": _generate_email_markdown(state) # Generate a detailed markdown description.
|
||||
}
|
||||
# Send the interrupt request inside a list, and extract the first response
|
||||
response = interrupt([request])[0]
|
||||
if response['type'] == "response":
|
||||
# Do something with the response
|
||||
...
|
||||
```
|
||||
@@ -1,19 +0,0 @@
|
||||
"""langgraph.prebuilt exposes a higher-level API for creating and executing agents and tools."""
|
||||
|
||||
from langgraph.prebuilt.chat_agent_executor import create_react_agent
|
||||
from langgraph.prebuilt.tool_node import (
|
||||
InjectedState,
|
||||
InjectedStore,
|
||||
ToolNode,
|
||||
tools_condition,
|
||||
)
|
||||
from langgraph.prebuilt.tool_validator import ValidationNode
|
||||
|
||||
__all__ = [
|
||||
"create_react_agent",
|
||||
"ToolNode",
|
||||
"tools_condition",
|
||||
"ValidationNode",
|
||||
"InjectedState",
|
||||
"InjectedStore",
|
||||
]
|
||||
@@ -1,839 +0,0 @@
|
||||
import functools
|
||||
import inspect
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Literal,
|
||||
Optional,
|
||||
Sequence,
|
||||
Type,
|
||||
TypeVar,
|
||||
Union,
|
||||
cast,
|
||||
)
|
||||
|
||||
from langchain_core.language_models import (
|
||||
BaseChatModel,
|
||||
LanguageModelInput,
|
||||
LanguageModelLike,
|
||||
)
|
||||
from langchain_core.messages import AIMessage, BaseMessage, SystemMessage, ToolMessage
|
||||
from langchain_core.runnables import (
|
||||
Runnable,
|
||||
RunnableBinding,
|
||||
RunnableConfig,
|
||||
)
|
||||
from langchain_core.tools import BaseTool
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import Annotated, TypedDict
|
||||
|
||||
from langgraph.errors import ErrorCode, create_error_message
|
||||
from langgraph.graph import END, StateGraph
|
||||
from langgraph.graph.graph import CompiledGraph
|
||||
from langgraph.graph.message import add_messages
|
||||
from langgraph.managed import IsLastStep, RemainingSteps
|
||||
from langgraph.prebuilt.tool_node import ToolNode
|
||||
from langgraph.store.base import BaseStore
|
||||
from langgraph.types import Checkpointer, Send
|
||||
from langgraph.utils.runnable import RunnableCallable
|
||||
|
||||
StructuredResponse = Union[dict, BaseModel]
|
||||
StructuredResponseSchema = Union[dict, type[BaseModel]]
|
||||
F = TypeVar("F", bound=Callable[..., Any])
|
||||
|
||||
|
||||
# We create the AgentState that we will pass around
|
||||
# This simply involves a list of messages
|
||||
# We want steps to return messages to append to the list
|
||||
# So we annotate the messages attribute with `add_messages` reducer
|
||||
class AgentState(TypedDict):
|
||||
"""The state of the agent."""
|
||||
|
||||
messages: Annotated[Sequence[BaseMessage], add_messages]
|
||||
|
||||
is_last_step: IsLastStep
|
||||
|
||||
remaining_steps: RemainingSteps
|
||||
|
||||
|
||||
class AgentStateWithStructuredResponse(AgentState):
|
||||
"""The state of the agent with a structured response."""
|
||||
|
||||
structured_response: StructuredResponse
|
||||
|
||||
|
||||
StateSchema = TypeVar("StateSchema", bound=AgentState)
|
||||
StateSchemaType = Type[StateSchema]
|
||||
|
||||
PROMPT_RUNNABLE_NAME = "Prompt"
|
||||
|
||||
Prompt = Union[
|
||||
SystemMessage,
|
||||
str,
|
||||
Callable[[StateSchema], LanguageModelInput],
|
||||
Runnable[StateSchema, LanguageModelInput],
|
||||
]
|
||||
|
||||
|
||||
def _get_prompt_runnable(prompt: Optional[Prompt]) -> Runnable:
|
||||
prompt_runnable: Runnable
|
||||
if prompt is None:
|
||||
prompt_runnable = RunnableCallable(
|
||||
lambda state: state["messages"], name=PROMPT_RUNNABLE_NAME
|
||||
)
|
||||
elif isinstance(prompt, str):
|
||||
_system_message: BaseMessage = SystemMessage(content=prompt)
|
||||
prompt_runnable = RunnableCallable(
|
||||
lambda state: [_system_message] + state["messages"],
|
||||
name=PROMPT_RUNNABLE_NAME,
|
||||
)
|
||||
elif isinstance(prompt, SystemMessage):
|
||||
prompt_runnable = RunnableCallable(
|
||||
lambda state: [prompt] + state["messages"],
|
||||
name=PROMPT_RUNNABLE_NAME,
|
||||
)
|
||||
elif inspect.iscoroutinefunction(prompt):
|
||||
prompt_runnable = RunnableCallable(
|
||||
None,
|
||||
prompt,
|
||||
name=PROMPT_RUNNABLE_NAME,
|
||||
)
|
||||
elif callable(prompt):
|
||||
prompt_runnable = RunnableCallable(
|
||||
prompt,
|
||||
name=PROMPT_RUNNABLE_NAME,
|
||||
)
|
||||
elif isinstance(prompt, Runnable):
|
||||
prompt_runnable = prompt
|
||||
else:
|
||||
raise ValueError(f"Got unexpected type for `prompt`: {type(prompt)}")
|
||||
|
||||
return prompt_runnable
|
||||
|
||||
|
||||
def _convert_modifier_to_prompt(func: F) -> F:
|
||||
"""Decorator that converts state_modifier kwarg to prompt kwarg."""
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
prompt = kwargs.get("prompt")
|
||||
state_modifier = kwargs.pop("state_modifier", None)
|
||||
if sum(p is not None for p in (prompt, state_modifier)) > 1:
|
||||
raise ValueError(
|
||||
"Expected only one of (prompt, state_modifier), got multiple values"
|
||||
)
|
||||
|
||||
if state_modifier is not None:
|
||||
prompt = state_modifier
|
||||
|
||||
kwargs["prompt"] = prompt
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return cast(F, wrapper)
|
||||
|
||||
|
||||
def _should_bind_tools(model: LanguageModelLike, tools: Sequence[BaseTool]) -> bool:
|
||||
if not isinstance(model, RunnableBinding):
|
||||
return True
|
||||
|
||||
if "tools" not in model.kwargs:
|
||||
return True
|
||||
|
||||
bound_tools = model.kwargs["tools"]
|
||||
if len(tools) != len(bound_tools):
|
||||
raise ValueError(
|
||||
"Number of tools in the model.bind_tools() and tools passed to create_react_agent must match"
|
||||
)
|
||||
|
||||
tool_names = set(tool.name for tool in tools)
|
||||
bound_tool_names = set()
|
||||
for bound_tool in bound_tools:
|
||||
# OpenAI-style tool
|
||||
if bound_tool.get("type") == "function":
|
||||
bound_tool_name = bound_tool["function"]["name"]
|
||||
# Anthropic-style tool
|
||||
elif bound_tool.get("name"):
|
||||
bound_tool_name = bound_tool["name"]
|
||||
else:
|
||||
# unknown tool type so we'll ignore it
|
||||
continue
|
||||
|
||||
bound_tool_names.add(bound_tool_name)
|
||||
|
||||
if missing_tools := tool_names - bound_tool_names:
|
||||
raise ValueError(f"Missing tools '{missing_tools}' in the model.bind_tools()")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def _get_model(model: LanguageModelLike) -> BaseChatModel:
|
||||
"""Get the underlying model from a RunnableBinding or return the model itself."""
|
||||
if isinstance(model, RunnableBinding):
|
||||
model = model.bound
|
||||
|
||||
if not isinstance(model, BaseChatModel):
|
||||
raise TypeError(
|
||||
f"Expected `model` to be a ChatModel or RunnableBinding (e.g. model.bind_tools(...)), got {type(model)}"
|
||||
)
|
||||
|
||||
return model
|
||||
|
||||
|
||||
def _validate_chat_history(
|
||||
messages: Sequence[BaseMessage],
|
||||
) -> None:
|
||||
"""Validate that all tool calls in AIMessages have a corresponding ToolMessage."""
|
||||
all_tool_calls = [
|
||||
tool_call
|
||||
for message in messages
|
||||
if isinstance(message, AIMessage)
|
||||
for tool_call in message.tool_calls
|
||||
]
|
||||
tool_call_ids_with_results = {
|
||||
message.tool_call_id for message in messages if isinstance(message, ToolMessage)
|
||||
}
|
||||
tool_calls_without_results = [
|
||||
tool_call
|
||||
for tool_call in all_tool_calls
|
||||
if tool_call["id"] not in tool_call_ids_with_results
|
||||
]
|
||||
if not tool_calls_without_results:
|
||||
return
|
||||
|
||||
error_message = create_error_message(
|
||||
message="Found AIMessages with tool_calls that do not have a corresponding ToolMessage. "
|
||||
f"Here are the first few of those tool calls: {tool_calls_without_results[:3]}.\n\n"
|
||||
"Every tool call (LLM requesting to call a tool) in the message history MUST have a corresponding ToolMessage "
|
||||
"(result of a tool invocation to return to the LLM) - this is required by most LLM providers.",
|
||||
error_code=ErrorCode.INVALID_CHAT_HISTORY,
|
||||
)
|
||||
raise ValueError(error_message)
|
||||
|
||||
|
||||
@_convert_modifier_to_prompt
|
||||
def create_react_agent(
|
||||
model: Union[str, LanguageModelLike],
|
||||
tools: Union[Sequence[BaseTool], ToolNode],
|
||||
*,
|
||||
prompt: Optional[Prompt] = None,
|
||||
response_format: Optional[
|
||||
Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]
|
||||
] = None,
|
||||
state_schema: Optional[StateSchemaType] = None,
|
||||
config_schema: Optional[Type[Any]] = None,
|
||||
checkpointer: Optional[Checkpointer] = None,
|
||||
store: Optional[BaseStore] = None,
|
||||
interrupt_before: Optional[list[str]] = None,
|
||||
interrupt_after: Optional[list[str]] = None,
|
||||
debug: bool = False,
|
||||
version: Literal["v1", "v2"] = "v1",
|
||||
name: Optional[str] = None,
|
||||
) -> CompiledGraph:
|
||||
"""Creates a graph that works with a chat model that utilizes tool calling.
|
||||
|
||||
Args:
|
||||
model: The `LangChain` chat model that supports tool calling.
|
||||
tools: A list of tools or a ToolNode instance.
|
||||
If an empty list is provided, the agent will consist of a single LLM node without tool calling.
|
||||
prompt: An optional prompt for the LLM. Can take a few different forms:
|
||||
|
||||
- str: This is converted to a SystemMessage and added to the beginning of the list of messages in state["messages"].
|
||||
- SystemMessage: this is added to the beginning of the list of messages in state["messages"].
|
||||
- Callable: This function should take in full graph state and the output is then passed to the language model.
|
||||
- Runnable: This runnable should take in full graph state and the output is then passed to the language model.
|
||||
|
||||
response_format: An optional schema for the final agent output.
|
||||
|
||||
If provided, output will be formatted to match the given schema and returned in the 'structured_response' state key.
|
||||
If not provided, `structured_response` will not be present in the output state.
|
||||
Can be passed in as:
|
||||
|
||||
- an OpenAI function/tool schema,
|
||||
- a JSON Schema,
|
||||
- a TypedDict class,
|
||||
- or a Pydantic class.
|
||||
- a tuple (prompt, schema), where schema is one of the above.
|
||||
The prompt will be used together with the model that is being used to generate the structured response.
|
||||
|
||||
!!! Important
|
||||
`response_format` requires the model to support `.with_structured_output`
|
||||
|
||||
!!! Note
|
||||
The graph will make a separate call to the LLM to generate the structured response after the agent loop is finished.
|
||||
This is not the only strategy to get structured responses, see more options in [this guide](https://langchain-ai.github.io/langgraph/how-tos/react-agent-structured-output/).
|
||||
state_schema: An optional state schema that defines graph state.
|
||||
Must have `messages` and `is_last_step` keys.
|
||||
Defaults to `AgentState` that defines those two keys.
|
||||
config_schema: An optional schema for configuration.
|
||||
Use this to expose configurable parameters via agent.config_specs.
|
||||
checkpointer: An optional checkpoint saver object. This is used for persisting
|
||||
the state of the graph (e.g., as chat memory) for a single thread (e.g., a single conversation).
|
||||
store: An optional store object. This is used for persisting data
|
||||
across multiple threads (e.g., multiple conversations / users).
|
||||
interrupt_before: An optional list of node names to interrupt before.
|
||||
Should be one of the following: "agent", "tools".
|
||||
This is useful if you want to add a user confirmation or other interrupt before taking an action.
|
||||
interrupt_after: An optional list of node names to interrupt after.
|
||||
Should be one of the following: "agent", "tools".
|
||||
This is useful if you want to return directly or run additional processing on an output.
|
||||
debug: A flag indicating whether to enable debug mode.
|
||||
version: Determines the version of the graph to create.
|
||||
Can be one of:
|
||||
|
||||
- `"v1"`: The tool node processes a single message. All tool
|
||||
calls in the message are executed in parallel within the tool node.
|
||||
- `"v2"`: The tool node processes a tool call.
|
||||
Tool calls are distributed across multiple instances of the tool
|
||||
node using the [Send](https://langchain-ai.github.io/langgraph/concepts/low_level/#send)
|
||||
API.
|
||||
name: An optional name for the CompiledStateGraph.
|
||||
This name will be automatically used when adding ReAct agent graph to another graph as a subgraph node -
|
||||
particularly useful for building multi-agent systems.
|
||||
|
||||
Returns:
|
||||
A compiled LangChain runnable that can be used for chat interactions.
|
||||
|
||||
The resulting graph looks like this:
|
||||
|
||||
``` mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> Start
|
||||
Start --> Agent
|
||||
Agent --> Tools : continue
|
||||
Tools --> Agent
|
||||
Agent --> End : end
|
||||
End --> [*]
|
||||
|
||||
classDef startClass fill:#ffdfba;
|
||||
classDef endClass fill:#baffc9;
|
||||
classDef otherClass fill:#fad7de;
|
||||
|
||||
class Start startClass
|
||||
class End endClass
|
||||
class Agent,Tools otherClass
|
||||
```
|
||||
|
||||
The "agent" node calls the language model with the messages list (after applying the messages modifier).
|
||||
If the resulting AIMessage contains `tool_calls`, the graph will then call the ["tools"][langgraph.prebuilt.tool_node.ToolNode].
|
||||
The "tools" node executes the tools (1 tool per `tool_call`) and adds the responses to the messages list
|
||||
as `ToolMessage` objects. The agent node then calls the language model again.
|
||||
The process repeats until no more `tool_calls` are present in the response.
|
||||
The agent then returns the full list of messages as a dictionary containing the key "messages".
|
||||
|
||||
``` mermaid
|
||||
sequenceDiagram
|
||||
participant U as User
|
||||
participant A as Agent (LLM)
|
||||
participant T as Tools
|
||||
U->>A: Initial input
|
||||
Note over A: Messages modifier + LLM
|
||||
loop while tool_calls present
|
||||
A->>T: Execute tools
|
||||
T-->>A: ToolMessage for each tool_calls
|
||||
end
|
||||
A->>U: Return final state
|
||||
```
|
||||
|
||||
Examples:
|
||||
Use with a simple tool:
|
||||
|
||||
```pycon
|
||||
>>> from datetime import datetime
|
||||
>>> from langchain_openai import ChatOpenAI
|
||||
>>> from langgraph.prebuilt import create_react_agent
|
||||
|
||||
|
||||
... def check_weather(location: str, at_time: datetime | None = None) -> str:
|
||||
... '''Return the weather forecast for the specified location.'''
|
||||
... return f"It's always sunny in {location}"
|
||||
>>>
|
||||
>>> tools = [check_weather]
|
||||
>>> model = ChatOpenAI(model="gpt-4o")
|
||||
>>> graph = create_react_agent(model, tools=tools)
|
||||
>>> inputs = {"messages": [("user", "what is the weather in sf")]}
|
||||
>>> for s in graph.stream(inputs, stream_mode="values"):
|
||||
... message = s["messages"][-1]
|
||||
... if isinstance(message, tuple):
|
||||
... print(message)
|
||||
... else:
|
||||
... message.pretty_print()
|
||||
('user', 'what is the weather in sf')
|
||||
================================== Ai Message ==================================
|
||||
Tool Calls:
|
||||
check_weather (call_LUzFvKJRuaWQPeXvBOzwhQOu)
|
||||
Call ID: call_LUzFvKJRuaWQPeXvBOzwhQOu
|
||||
Args:
|
||||
location: San Francisco
|
||||
================================= Tool Message =================================
|
||||
Name: check_weather
|
||||
It's always sunny in San Francisco
|
||||
================================== Ai Message ==================================
|
||||
The weather in San Francisco is sunny.
|
||||
```
|
||||
Add a system prompt for the LLM:
|
||||
|
||||
```pycon
|
||||
>>> system_prompt = "You are a helpful bot named Fred."
|
||||
>>> graph = create_react_agent(model, tools, prompt=system_prompt)
|
||||
>>> inputs = {"messages": [("user", "What's your name? And what's the weather in SF?")]}
|
||||
>>> for s in graph.stream(inputs, stream_mode="values"):
|
||||
... message = s["messages"][-1]
|
||||
... if isinstance(message, tuple):
|
||||
... print(message)
|
||||
... else:
|
||||
... message.pretty_print()
|
||||
('user', "What's your name? And what's the weather in SF?")
|
||||
================================== Ai Message ==================================
|
||||
Hi, my name is Fred. Let me check the weather in San Francisco for you.
|
||||
Tool Calls:
|
||||
check_weather (call_lqhj4O0hXYkW9eknB4S41EXk)
|
||||
Call ID: call_lqhj4O0hXYkW9eknB4S41EXk
|
||||
Args:
|
||||
location: San Francisco
|
||||
================================= Tool Message =================================
|
||||
Name: check_weather
|
||||
It's always sunny in San Francisco
|
||||
================================== Ai Message ==================================
|
||||
The weather in San Francisco is currently sunny. If you need any more details or have other questions, feel free to ask!
|
||||
```
|
||||
|
||||
Add a more complex prompt for the LLM:
|
||||
|
||||
```pycon
|
||||
>>> from langchain_core.prompts import ChatPromptTemplate
|
||||
>>> prompt = ChatPromptTemplate.from_messages([
|
||||
... ("system", "You are a helpful bot named Fred."),
|
||||
... ("placeholder", "{messages}"),
|
||||
... ("user", "Remember, always be polite!"),
|
||||
... ])
|
||||
>>>
|
||||
>>> graph = create_react_agent(model, tools, prompt=prompt)
|
||||
>>> inputs = {"messages": [("user", "What's your name? And what's the weather in SF?")]}
|
||||
>>> for s in graph.stream(inputs, stream_mode="values"):
|
||||
... message = s["messages"][-1]
|
||||
... if isinstance(message, tuple):
|
||||
... print(message)
|
||||
... else:
|
||||
... message.pretty_print()
|
||||
```
|
||||
|
||||
Add complex prompt with custom graph state:
|
||||
|
||||
```pycon
|
||||
>>> from typing_extensions import TypedDict
|
||||
>>>
|
||||
>>> from langgraph.managed import IsLastStep
|
||||
>>> prompt = ChatPromptTemplate.from_messages(
|
||||
... [
|
||||
... ("system", "Today is {today}"),
|
||||
... ("placeholder", "{messages}"),
|
||||
... ]
|
||||
... )
|
||||
>>>
|
||||
>>> class CustomState(TypedDict):
|
||||
... today: str
|
||||
... messages: Annotated[list[BaseMessage], add_messages]
|
||||
... is_last_step: IsLastStep
|
||||
>>>
|
||||
>>> graph = create_react_agent(model, tools, state_schema=CustomState, prompt=prompt)
|
||||
>>> inputs = {"messages": [("user", "What's today's date? And what's the weather in SF?")], "today": "July 16, 2004"}
|
||||
>>> for s in graph.stream(inputs, stream_mode="values"):
|
||||
... message = s["messages"][-1]
|
||||
... if isinstance(message, tuple):
|
||||
... print(message)
|
||||
... else:
|
||||
... message.pretty_print()
|
||||
```
|
||||
|
||||
Add thread-level "chat memory" to the graph:
|
||||
|
||||
```pycon
|
||||
>>> from langgraph.checkpoint.memory import MemorySaver
|
||||
>>> graph = create_react_agent(model, tools, checkpointer=MemorySaver())
|
||||
>>> config = {"configurable": {"thread_id": "thread-1"}}
|
||||
>>> def print_stream(graph, inputs, config):
|
||||
... for s in graph.stream(inputs, config, stream_mode="values"):
|
||||
... message = s["messages"][-1]
|
||||
... if isinstance(message, tuple):
|
||||
... print(message)
|
||||
... else:
|
||||
... message.pretty_print()
|
||||
>>> inputs = {"messages": [("user", "What's the weather in SF?")]}
|
||||
>>> print_stream(graph, inputs, config)
|
||||
>>> inputs2 = {"messages": [("user", "Cool, so then should i go biking today?")]}
|
||||
>>> print_stream(graph, inputs2, config)
|
||||
('user', "What's the weather in SF?")
|
||||
================================== Ai Message ==================================
|
||||
Tool Calls:
|
||||
check_weather (call_ChndaktJxpr6EMPEB5JfOFYc)
|
||||
Call ID: call_ChndaktJxpr6EMPEB5JfOFYc
|
||||
Args:
|
||||
location: San Francisco
|
||||
================================= Tool Message =================================
|
||||
Name: check_weather
|
||||
It's always sunny in San Francisco
|
||||
================================== Ai Message ==================================
|
||||
The weather in San Francisco is sunny. Enjoy your day!
|
||||
================================ Human Message =================================
|
||||
Cool, so then should i go biking today?
|
||||
================================== Ai Message ==================================
|
||||
Since the weather in San Francisco is sunny, it sounds like a great day for biking! Enjoy your ride!
|
||||
```
|
||||
|
||||
Add an interrupt to let the user confirm before taking an action:
|
||||
|
||||
```pycon
|
||||
>>> graph = create_react_agent(
|
||||
... model, tools, interrupt_before=["tools"], checkpointer=MemorySaver()
|
||||
>>> )
|
||||
>>> config = {"configurable": {"thread_id": "thread-1"}}
|
||||
|
||||
>>> inputs = {"messages": [("user", "What's the weather in SF?")]}
|
||||
>>> print_stream(graph, inputs, config)
|
||||
>>> snapshot = graph.get_state(config)
|
||||
>>> print("Next step: ", snapshot.next)
|
||||
>>> print_stream(graph, None, config)
|
||||
```
|
||||
|
||||
Add cross-thread memory to the graph:
|
||||
|
||||
```pycon
|
||||
>>> from langgraph.prebuilt import InjectedStore
|
||||
>>> from langgraph.store.base import BaseStore
|
||||
|
||||
>>> def save_memory(memory: str, *, config: RunnableConfig, store: Annotated[BaseStore, InjectedStore()]) -> str:
|
||||
... '''Save the given memory for the current user.'''
|
||||
... # This is a **tool** the model can use to save memories to storage
|
||||
... user_id = config.get("configurable", {}).get("user_id")
|
||||
... namespace = ("memories", user_id)
|
||||
... store.put(namespace, f"memory_{len(store.search(namespace))}", {"data": memory})
|
||||
... return f"Saved memory: {memory}"
|
||||
|
||||
>>> def prepare_model_inputs(state: AgentState, config: RunnableConfig, store: BaseStore):
|
||||
... # Retrieve user memories and add them to the system message
|
||||
... # This function is called **every time** the model is prompted. It converts the state to a prompt
|
||||
... user_id = config.get("configurable", {}).get("user_id")
|
||||
... namespace = ("memories", user_id)
|
||||
... memories = [m.value["data"] for m in store.search(namespace)]
|
||||
... system_msg = f"User memories: {', '.join(memories)}"
|
||||
... return [{"role": "system", "content": system_msg)] + state["messages"]
|
||||
|
||||
>>> from langgraph.checkpoint.memory import MemorySaver
|
||||
>>> from langgraph.store.memory import InMemoryStore
|
||||
>>> store = InMemoryStore()
|
||||
>>> graph = create_react_agent(model, [save_memory], prompt=prepare_model_inputs, store=store, checkpointer=MemorySaver())
|
||||
>>> config = {"configurable": {"thread_id": "thread-1", "user_id": "1"}}
|
||||
|
||||
>>> inputs = {"messages": [("user", "Hey I'm Will, how's it going?")]}
|
||||
>>> print_stream(graph, inputs, config)
|
||||
('user', "Hey I'm Will, how's it going?")
|
||||
================================== Ai Message ==================================
|
||||
Hello Will! It's nice to meet you. I'm doing well, thank you for asking. How are you doing today?
|
||||
|
||||
>>> inputs2 = {"messages": [("user", "I like to bike")]}
|
||||
>>> print_stream(graph, inputs2, config)
|
||||
================================ Human Message =================================
|
||||
I like to bike
|
||||
================================== Ai Message ==================================
|
||||
That's great to hear, Will! Biking is an excellent hobby and form of exercise. It's a fun way to stay active and explore your surroundings. Do you have any favorite biking routes or trails you enjoy? Or perhaps you're into a specific type of biking, like mountain biking or road cycling?
|
||||
|
||||
>>> config = {"configurable": {"thread_id": "thread-2", "user_id": "1"}}
|
||||
>>> inputs3 = {"messages": [("user", "Hi there! Remember me?")]}
|
||||
>>> print_stream(graph, inputs3, config)
|
||||
================================ Human Message =================================
|
||||
Hi there! Remember me?
|
||||
================================== Ai Message ==================================
|
||||
User memories:
|
||||
Hello! Of course, I remember you, Will! You mentioned earlier that you like to bike. It's great to hear from you again. How have you been? Have you been on any interesting bike rides lately?
|
||||
```
|
||||
|
||||
Add a timeout for a given step:
|
||||
|
||||
```pycon
|
||||
>>> import time
|
||||
... def check_weather(location: str, at_time: datetime | None = None) -> float:
|
||||
... '''Return the weather forecast for the specified location.'''
|
||||
... time.sleep(2)
|
||||
... return f"It's always sunny in {location}"
|
||||
>>>
|
||||
>>> tools = [check_weather]
|
||||
>>> graph = create_react_agent(model, tools)
|
||||
>>> graph.step_timeout = 1 # Seconds
|
||||
>>> for s in graph.stream({"messages": [("user", "what is the weather in sf")]}):
|
||||
... print(s)
|
||||
TimeoutError: Timed out at step 2
|
||||
```
|
||||
"""
|
||||
if version not in ("v1", "v2"):
|
||||
raise ValueError(
|
||||
f"Invalid version {version}. Supported versions are 'v1' and 'v2'."
|
||||
)
|
||||
|
||||
if state_schema is not None:
|
||||
required_keys = {"messages", "remaining_steps"}
|
||||
if response_format is not None:
|
||||
required_keys.add("structured_response")
|
||||
|
||||
if missing_keys := required_keys - set(state_schema.__annotations__):
|
||||
raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")
|
||||
|
||||
if state_schema is None:
|
||||
state_schema = (
|
||||
AgentStateWithStructuredResponse
|
||||
if response_format is not None
|
||||
else AgentState
|
||||
)
|
||||
|
||||
if isinstance(tools, ToolNode):
|
||||
tool_classes = list(tools.tools_by_name.values())
|
||||
tool_node = tools
|
||||
else:
|
||||
tool_node = ToolNode(tools)
|
||||
# get the tool functions wrapped in a tool class from the ToolNode
|
||||
tool_classes = list(tool_node.tools_by_name.values())
|
||||
|
||||
if isinstance(model, str):
|
||||
try:
|
||||
from langchain.chat_models import ( # type: ignore[import-not-found]
|
||||
init_chat_model,
|
||||
)
|
||||
except ImportError:
|
||||
raise ImportError(
|
||||
"Please install langchain (`pip install langchain`) to use '<provider>:<model>' string syntax for `model` parameter."
|
||||
)
|
||||
|
||||
model = cast(BaseChatModel, init_chat_model(model))
|
||||
|
||||
tool_calling_enabled = len(tool_classes) > 0
|
||||
|
||||
if _should_bind_tools(model, tool_classes) and tool_calling_enabled:
|
||||
model = cast(BaseChatModel, model).bind_tools(tool_classes)
|
||||
|
||||
model_runnable = _get_prompt_runnable(prompt) | model
|
||||
|
||||
# If any of the tools are configured to return_directly after running,
|
||||
# our graph needs to check if these were called
|
||||
should_return_direct = {t.name for t in tool_classes if t.return_direct}
|
||||
|
||||
# Define the function that calls the model
|
||||
def call_model(state: AgentState, config: RunnableConfig) -> AgentState:
|
||||
_validate_chat_history(state["messages"])
|
||||
response = cast(AIMessage, model_runnable.invoke(state, config))
|
||||
# add agent name to the AIMessage
|
||||
response.name = name
|
||||
has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
|
||||
all_tools_return_direct = (
|
||||
all(call["name"] in should_return_direct for call in response.tool_calls)
|
||||
if isinstance(response, AIMessage)
|
||||
else False
|
||||
)
|
||||
if (
|
||||
(
|
||||
"remaining_steps" not in state
|
||||
and state.get("is_last_step", False)
|
||||
and has_tool_calls
|
||||
)
|
||||
or (
|
||||
"remaining_steps" in state
|
||||
and state["remaining_steps"] < 1
|
||||
and all_tools_return_direct
|
||||
)
|
||||
or (
|
||||
"remaining_steps" in state
|
||||
and state["remaining_steps"] < 2
|
||||
and has_tool_calls
|
||||
)
|
||||
):
|
||||
return {
|
||||
"messages": [
|
||||
AIMessage(
|
||||
id=response.id,
|
||||
content="Sorry, need more steps to process this request.",
|
||||
)
|
||||
]
|
||||
}
|
||||
# We return a list, because this will get added to the existing list
|
||||
return {"messages": [response]}
|
||||
|
||||
async def acall_model(state: AgentState, config: RunnableConfig) -> AgentState:
|
||||
_validate_chat_history(state["messages"])
|
||||
response = cast(AIMessage, await model_runnable.ainvoke(state, config))
|
||||
# add agent name to the AIMessage
|
||||
response.name = name
|
||||
has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
|
||||
all_tools_return_direct = (
|
||||
all(call["name"] in should_return_direct for call in response.tool_calls)
|
||||
if isinstance(response, AIMessage)
|
||||
else False
|
||||
)
|
||||
if (
|
||||
(
|
||||
"remaining_steps" not in state
|
||||
and state.get("is_last_step", False)
|
||||
and has_tool_calls
|
||||
)
|
||||
or (
|
||||
"remaining_steps" in state
|
||||
and state["remaining_steps"] < 1
|
||||
and all_tools_return_direct
|
||||
)
|
||||
or (
|
||||
"remaining_steps" in state
|
||||
and state["remaining_steps"] < 2
|
||||
and has_tool_calls
|
||||
)
|
||||
):
|
||||
return {
|
||||
"messages": [
|
||||
AIMessage(
|
||||
id=response.id,
|
||||
content="Sorry, need more steps to process this request.",
|
||||
)
|
||||
]
|
||||
}
|
||||
# We return a list, because this will get added to the existing list
|
||||
return {"messages": [response]}
|
||||
|
||||
def generate_structured_response(
|
||||
state: AgentState, config: RunnableConfig
|
||||
) -> AgentState:
|
||||
# NOTE: we exclude the last message because there is enough information
|
||||
# for the LLM to generate the structured response
|
||||
messages = state["messages"][:-1]
|
||||
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(
|
||||
cast(StructuredResponseSchema, structured_response_schema)
|
||||
)
|
||||
response = model_with_structured_output.invoke(messages, config)
|
||||
return {"structured_response": response}
|
||||
|
||||
async def agenerate_structured_response(
|
||||
state: AgentState, config: RunnableConfig
|
||||
) -> AgentState:
|
||||
# NOTE: we exclude the last message because there is enough information
|
||||
# for the LLM to generate the structured response
|
||||
messages = state["messages"][:-1]
|
||||
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(
|
||||
cast(StructuredResponseSchema, structured_response_schema)
|
||||
)
|
||||
response = await model_with_structured_output.ainvoke(messages, config)
|
||||
return {"structured_response": response}
|
||||
|
||||
if not tool_calling_enabled:
|
||||
# Define a new graph
|
||||
workflow = StateGraph(state_schema, config_schema=config_schema)
|
||||
workflow.add_node("agent", RunnableCallable(call_model, acall_model))
|
||||
workflow.set_entry_point("agent")
|
||||
if response_format is not None:
|
||||
workflow.add_node(
|
||||
"generate_structured_response",
|
||||
RunnableCallable(
|
||||
generate_structured_response, agenerate_structured_response
|
||||
),
|
||||
)
|
||||
workflow.add_edge("agent", "generate_structured_response")
|
||||
|
||||
return workflow.compile(
|
||||
checkpointer=checkpointer,
|
||||
store=store,
|
||||
interrupt_before=interrupt_before,
|
||||
interrupt_after=interrupt_after,
|
||||
debug=debug,
|
||||
name=name,
|
||||
)
|
||||
|
||||
# Define the function that determines whether to continue or not
|
||||
def should_continue(state: AgentState) -> Union[str, list]:
|
||||
messages = state["messages"]
|
||||
last_message = messages[-1]
|
||||
# If there is no function call, then we finish
|
||||
if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
|
||||
return END if response_format is None else "generate_structured_response"
|
||||
# Otherwise if there is, we continue
|
||||
else:
|
||||
if version == "v1":
|
||||
return "tools"
|
||||
elif version == "v2":
|
||||
tool_calls = [
|
||||
tool_node.inject_tool_args(call, state, store) # type: ignore[arg-type]
|
||||
for call in last_message.tool_calls
|
||||
]
|
||||
return [Send("tools", [tool_call]) for tool_call in tool_calls]
|
||||
|
||||
# Define a new graph
|
||||
workflow = StateGraph(state_schema or AgentState, config_schema=config_schema)
|
||||
|
||||
# Define the two nodes we will cycle between
|
||||
workflow.add_node("agent", RunnableCallable(call_model, acall_model))
|
||||
workflow.add_node("tools", tool_node)
|
||||
|
||||
# Set the entrypoint as `agent`
|
||||
# This means that this node is the first one called
|
||||
workflow.set_entry_point("agent")
|
||||
|
||||
# Add a structured output node if response_format is provided
|
||||
if response_format is not None:
|
||||
workflow.add_node(
|
||||
"generate_structured_response",
|
||||
RunnableCallable(
|
||||
generate_structured_response, agenerate_structured_response
|
||||
),
|
||||
)
|
||||
workflow.add_edge("generate_structured_response", END)
|
||||
should_continue_destinations = ["tools", "generate_structured_response"]
|
||||
else:
|
||||
should_continue_destinations = ["tools", END]
|
||||
|
||||
# We now add a conditional edge
|
||||
workflow.add_conditional_edges(
|
||||
# First, we define the start node. We use `agent`.
|
||||
# This means these are the edges taken after the `agent` node is called.
|
||||
"agent",
|
||||
# Next, we pass in the function that will determine which node is called next.
|
||||
should_continue,
|
||||
path_map=should_continue_destinations,
|
||||
)
|
||||
|
||||
def route_tool_responses(state: AgentState) -> Literal["agent", "__end__"]:
|
||||
for m in reversed(state["messages"]):
|
||||
if not isinstance(m, ToolMessage):
|
||||
break
|
||||
if m.name in should_return_direct:
|
||||
return END
|
||||
return "agent"
|
||||
|
||||
if should_return_direct:
|
||||
workflow.add_conditional_edges("tools", route_tool_responses)
|
||||
else:
|
||||
workflow.add_edge("tools", "agent")
|
||||
|
||||
# Finally, we compile it!
|
||||
# This compiles it into a LangChain Runnable,
|
||||
# meaning you can use it as you would any other runnable
|
||||
return workflow.compile(
|
||||
checkpointer=checkpointer,
|
||||
store=store,
|
||||
interrupt_before=interrupt_before,
|
||||
interrupt_after=interrupt_after,
|
||||
debug=debug,
|
||||
name=name,
|
||||
)
|
||||
|
||||
|
||||
# Keep for backwards compatibility
|
||||
create_tool_calling_executor = create_react_agent
|
||||
|
||||
__all__ = [
|
||||
"create_react_agent",
|
||||
"create_tool_calling_executor",
|
||||
"AgentState",
|
||||
]
|
||||
@@ -1,94 +0,0 @@
|
||||
from typing import (
|
||||
Literal,
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
|
||||
class HumanInterruptConfig(TypedDict):
|
||||
"""Configuration that defines what actions are allowed for a human interrupt.
|
||||
|
||||
This controls the available interaction options when the graph is paused for human input.
|
||||
|
||||
Attributes:
|
||||
allow_ignore: Whether the human can choose to ignore/skip the current step
|
||||
allow_respond: Whether the human can provide a text response/feedback
|
||||
allow_edit: Whether the human can edit the provided content/state
|
||||
allow_accept: Whether the human can accept/approve the current state
|
||||
"""
|
||||
|
||||
allow_ignore: bool
|
||||
allow_respond: bool
|
||||
allow_edit: bool
|
||||
allow_accept: bool
|
||||
|
||||
|
||||
class ActionRequest(TypedDict):
|
||||
"""Represents a request for human action within the graph execution.
|
||||
|
||||
Contains the action type and any associated arguments needed for the action.
|
||||
|
||||
Attributes:
|
||||
action: The type or name of action being requested (e.g., "Approve XYZ action")
|
||||
args: Key-value pairs of arguments needed for the action
|
||||
"""
|
||||
|
||||
action: str
|
||||
args: dict
|
||||
|
||||
|
||||
class HumanInterrupt(TypedDict):
|
||||
"""Represents an interrupt triggered by the graph that requires human intervention.
|
||||
|
||||
This is passed to the `interrupt` function when execution is paused for human input.
|
||||
|
||||
Attributes:
|
||||
action_request: The specific action being requested from the human
|
||||
config: Configuration defining what actions are allowed
|
||||
description: Optional detailed description of what input is needed
|
||||
|
||||
Example:
|
||||
```python
|
||||
# Extract a tool call from the state and create an interrupt request
|
||||
request = HumanInterrupt(
|
||||
action_request=ActionRequest(
|
||||
action="run_command", # The action being requested
|
||||
args={"command": "ls", "args": ["-l"]} # Arguments for the action
|
||||
),
|
||||
config=HumanInterruptConfig(
|
||||
allow_ignore=True, # Allow skipping this step
|
||||
allow_respond=True, # Allow text feedback
|
||||
allow_edit=False, # Don't allow editing
|
||||
allow_accept=True # Allow direct acceptance
|
||||
),
|
||||
description="Please review the command before execution"
|
||||
)
|
||||
# Send the interrupt request and get the response
|
||||
response = interrupt([request])[0]
|
||||
```
|
||||
"""
|
||||
|
||||
action_request: ActionRequest
|
||||
config: HumanInterruptConfig
|
||||
description: Optional[str]
|
||||
|
||||
|
||||
class HumanResponse(TypedDict):
|
||||
"""The response provided by a human to an interrupt, which is returned when graph execution resumes.
|
||||
|
||||
Attributes:
|
||||
type: The type of response:
|
||||
- "accept": Approves the current state without changes
|
||||
- "ignore": Skips/ignores the current step
|
||||
- "response": Provides text feedback or instructions
|
||||
- "edit": Modifies the current state/content
|
||||
arg: The response payload:
|
||||
- None: For ignore/accept actions
|
||||
- str: For text responses
|
||||
- ActionRequest: For edit actions with updated content
|
||||
"""
|
||||
|
||||
type: Literal["accept", "ignore", "response", "edit"]
|
||||
args: Union[None, str, ActionRequest]
|
||||
@@ -1,848 +0,0 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
from copy import copy, deepcopy
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Literal,
|
||||
Optional,
|
||||
Sequence,
|
||||
Tuple,
|
||||
Type,
|
||||
Union,
|
||||
cast,
|
||||
get_type_hints,
|
||||
)
|
||||
|
||||
from langchain_core.messages import (
|
||||
AIMessage,
|
||||
AnyMessage,
|
||||
ToolCall,
|
||||
ToolMessage,
|
||||
convert_to_messages,
|
||||
)
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
from langchain_core.runnables.config import (
|
||||
get_config_list,
|
||||
get_executor_for_config,
|
||||
)
|
||||
from langchain_core.tools import BaseTool, InjectedToolArg
|
||||
from langchain_core.tools import tool as create_tool
|
||||
from langchain_core.tools.base import get_all_basemodel_annotations
|
||||
from pydantic import BaseModel
|
||||
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
|
||||
from langgraph.utils.runnable import RunnableCallable
|
||||
|
||||
INVALID_TOOL_NAME_ERROR_TEMPLATE = (
|
||||
"Error: {requested_tool} is not a valid tool, try one of [{available_tools}]."
|
||||
)
|
||||
TOOL_CALL_ERROR_TEMPLATE = "Error: {error}\n Please fix your mistakes."
|
||||
|
||||
|
||||
def msg_content_output(output: Any) -> Union[str, list[dict]]:
|
||||
recognized_content_block_types = ("image", "image_url", "text", "json")
|
||||
if isinstance(output, str):
|
||||
return output
|
||||
elif isinstance(output, list) and all(
|
||||
[
|
||||
isinstance(x, dict) and x.get("type") in recognized_content_block_types
|
||||
for x in output
|
||||
]
|
||||
):
|
||||
return output
|
||||
# Technically a list of strings is also valid message content but it's not currently
|
||||
# well tested that all chat models support this. And for backwards compatibility
|
||||
# we want to make sure we don't break any existing ToolNode usage.
|
||||
else:
|
||||
try:
|
||||
return json.dumps(output, ensure_ascii=False)
|
||||
except Exception:
|
||||
return str(output)
|
||||
|
||||
|
||||
def _handle_tool_error(
|
||||
e: Exception,
|
||||
*,
|
||||
flag: Union[
|
||||
bool,
|
||||
str,
|
||||
Callable[..., str],
|
||||
tuple[type[Exception], ...],
|
||||
],
|
||||
) -> str:
|
||||
if isinstance(flag, (bool, tuple)):
|
||||
content = TOOL_CALL_ERROR_TEMPLATE.format(error=repr(e))
|
||||
elif isinstance(flag, str):
|
||||
content = flag
|
||||
elif callable(flag):
|
||||
content = flag(e)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Got unexpected type of `handle_tool_error`. Expected bool, str "
|
||||
f"or callable. Received: {flag}"
|
||||
)
|
||||
return content
|
||||
|
||||
|
||||
def _infer_handled_types(handler: Callable[..., str]) -> tuple[type[Exception], ...]:
|
||||
sig = inspect.signature(handler)
|
||||
params = list(sig.parameters.values())
|
||||
if params:
|
||||
# If it's a method, the first argument is typically 'self' or 'cls'
|
||||
if params[0].name in ["self", "cls"] and len(params) == 2:
|
||||
first_param = params[1]
|
||||
else:
|
||||
first_param = params[0]
|
||||
|
||||
type_hints = get_type_hints(handler)
|
||||
if first_param.name in type_hints:
|
||||
origin = get_origin(first_param.annotation)
|
||||
if origin is Union:
|
||||
args = get_args(first_param.annotation)
|
||||
if all(issubclass(arg, Exception) for arg in args):
|
||||
return tuple(args)
|
||||
else:
|
||||
raise ValueError(
|
||||
"All types in the error handler error annotation must be Exception types. "
|
||||
"For example, `def custom_handler(e: Union[ValueError, TypeError])`. "
|
||||
f"Got '{first_param.annotation}' instead."
|
||||
)
|
||||
|
||||
exception_type = type_hints[first_param.name]
|
||||
if Exception in exception_type.__mro__:
|
||||
return (exception_type,)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Arbitrary types are not supported in the error handler signature. "
|
||||
"Please annotate the error with either a specific Exception type or a union of Exception types. "
|
||||
"For example, `def custom_handler(e: ValueError)` or `def custom_handler(e: Union[ValueError, TypeError])`. "
|
||||
f"Got '{exception_type}' instead."
|
||||
)
|
||||
|
||||
# If no type information is available, return (Exception,) for backwards compatibility.
|
||||
return (Exception,)
|
||||
|
||||
|
||||
class ToolNode(RunnableCallable):
|
||||
"""A node that runs the tools called in the last AIMessage.
|
||||
|
||||
It can be used either in StateGraph with a "messages" state key (or a custom key passed via ToolNode's 'messages_key').
|
||||
If multiple tool calls are requested, they will be run in parallel. The output will be
|
||||
a list of ToolMessages, one for each tool call.
|
||||
|
||||
Tool calls can also be passed directly as a list of `ToolCall` dicts.
|
||||
|
||||
Args:
|
||||
tools: A sequence of tools that can be invoked by the ToolNode.
|
||||
name: The name of the ToolNode in the graph. Defaults to "tools".
|
||||
tags: Optional tags to associate with the node. Defaults to None.
|
||||
handle_tool_errors: How to handle tool errors raised by tools inside the node. Defaults to True.
|
||||
Must be one of the following:
|
||||
|
||||
- True: all errors will be caught and
|
||||
a ToolMessage with a default error message (TOOL_CALL_ERROR_TEMPLATE) will be returned.
|
||||
- str: all errors will be caught and
|
||||
a ToolMessage with the string value of 'handle_tool_errors' will be returned.
|
||||
- tuple[type[Exception], ...]: exceptions in the tuple will be caught and
|
||||
a ToolMessage with a default error message (TOOL_CALL_ERROR_TEMPLATE) will be returned.
|
||||
- Callable[..., str]: exceptions from the signature of the callable will be caught and
|
||||
a ToolMessage with the string value of the result of the 'handle_tool_errors' callable will be returned.
|
||||
- False: none of the errors raised by the tools will be caught
|
||||
messages_key: The state key in the input that contains the list of messages.
|
||||
The same key will be used for the output from the ToolNode.
|
||||
Defaults to "messages".
|
||||
|
||||
The `ToolNode` is roughly analogous to:
|
||||
|
||||
```python
|
||||
tools_by_name = {tool.name: tool for tool in tools}
|
||||
def tool_node(state: dict):
|
||||
result = []
|
||||
for tool_call in state["messages"][-1].tool_calls:
|
||||
tool = tools_by_name[tool_call["name"]]
|
||||
observation = tool.invoke(tool_call["args"])
|
||||
result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
|
||||
return {"messages": result}
|
||||
```
|
||||
|
||||
Tool calls can also be passed directly to a ToolNode. This can be useful when using
|
||||
the Send API, e.g., in a conditional edge:
|
||||
|
||||
```python
|
||||
def example_conditional_edge(state: dict) -> List[Send]:
|
||||
tool_calls = state["messages"][-1].tool_calls
|
||||
# If tools rely on state or store variables (whose values are not generated
|
||||
# directly by a model), you can inject them into the tool calls.
|
||||
tool_calls = [
|
||||
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]
|
||||
```
|
||||
|
||||
Important:
|
||||
- The input state can be one of the following:
|
||||
- A dict with a messages key containing a list of messages.
|
||||
- A list of messages.
|
||||
- A list of tool calls.
|
||||
- If operating on a message list, the last message must be an `AIMessage` with
|
||||
`tool_calls` populated.
|
||||
"""
|
||||
|
||||
name: str = "ToolNode"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tools: Sequence[Union[BaseTool, Callable]],
|
||||
*,
|
||||
name: str = "tools",
|
||||
tags: Optional[list[str]] = None,
|
||||
handle_tool_errors: Union[
|
||||
bool, str, Callable[..., str], tuple[type[Exception], ...]
|
||||
] = True,
|
||||
messages_key: str = "messages",
|
||||
) -> None:
|
||||
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_)
|
||||
|
||||
def _func(
|
||||
self,
|
||||
input: Union[
|
||||
list[AnyMessage],
|
||||
dict[str, Any],
|
||||
BaseModel,
|
||||
],
|
||||
config: RunnableConfig,
|
||||
*,
|
||||
store: Optional[BaseStore],
|
||||
) -> Any:
|
||||
tool_calls, input_type = self._parse_input(input, 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)
|
||||
]
|
||||
|
||||
# preserve existing behavior for non-command tool outputs for backwards
|
||||
# compatibility
|
||||
if not any(isinstance(output, Command) for output in outputs):
|
||||
# TypedDict, pydantic, dataclass, etc. should all be able to load from dict
|
||||
return outputs if input_type == "list" else {self.messages_key: outputs}
|
||||
|
||||
# LangGraph will automatically handle list of Command and non-command node
|
||||
# updates
|
||||
combined_outputs: list[
|
||||
Command | list[ToolMessage] | dict[str, list[ToolMessage]]
|
||||
] = []
|
||||
for output in outputs:
|
||||
if isinstance(output, Command):
|
||||
combined_outputs.append(output)
|
||||
else:
|
||||
combined_outputs.append(
|
||||
[output] if input_type == "list" else {self.messages_key: [output]}
|
||||
)
|
||||
return combined_outputs
|
||||
|
||||
async def _afunc(
|
||||
self,
|
||||
input: Union[
|
||||
list[AnyMessage],
|
||||
dict[str, Any],
|
||||
BaseModel,
|
||||
],
|
||||
config: RunnableConfig,
|
||||
*,
|
||||
store: Optional[BaseStore],
|
||||
) -> Any:
|
||||
tool_calls, input_type = self._parse_input(input, store)
|
||||
outputs = await asyncio.gather(
|
||||
*(self._arun_one(call, input_type, config) for call in tool_calls)
|
||||
)
|
||||
|
||||
# preserve existing behavior for non-command tool outputs for backwards compatibility
|
||||
if not any(isinstance(output, Command) for output in outputs):
|
||||
# TypedDict, pydantic, dataclass, etc. should all be able to load from dict
|
||||
return outputs if input_type == "list" else {self.messages_key: outputs}
|
||||
|
||||
# LangGraph will automatically handle list of Command and non-command node updates
|
||||
combined_outputs: list[
|
||||
Command | list[ToolMessage] | dict[str, list[ToolMessage]]
|
||||
] = []
|
||||
for output in outputs:
|
||||
if isinstance(output, Command):
|
||||
combined_outputs.append(output)
|
||||
else:
|
||||
combined_outputs.append(
|
||||
[output] if input_type == "list" else {self.messages_key: [output]}
|
||||
)
|
||||
return combined_outputs
|
||||
|
||||
def _run_one(
|
||||
self,
|
||||
call: ToolCall,
|
||||
input_type: Literal["list", "dict", "tool_calls"],
|
||||
config: RunnableConfig,
|
||||
) -> ToolMessage:
|
||||
if invalid_tool_message := self._validate_tool_call(call):
|
||||
return invalid_tool_message
|
||||
|
||||
try:
|
||||
input = {**call, **{"type": "tool_call"}}
|
||||
response = self.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)
|
||||
except GraphBubbleUp as e:
|
||||
raise e
|
||||
except Exception as e:
|
||||
if isinstance(self.handle_tool_errors, tuple):
|
||||
handled_types: tuple = self.handle_tool_errors
|
||||
elif callable(self.handle_tool_errors):
|
||||
handled_types = _infer_handled_types(self.handle_tool_errors)
|
||||
else:
|
||||
# default behavior is catching all exceptions
|
||||
handled_types = (Exception,)
|
||||
|
||||
# Unhandled
|
||||
if not self.handle_tool_errors or not isinstance(e, handled_types):
|
||||
raise e
|
||||
# Handled
|
||||
else:
|
||||
content = _handle_tool_error(e, flag=self.handle_tool_errors)
|
||||
return ToolMessage(
|
||||
content=content,
|
||||
name=call["name"],
|
||||
tool_call_id=call["id"],
|
||||
status="error",
|
||||
)
|
||||
|
||||
if isinstance(response, Command):
|
||||
return self._validate_tool_command(response, call, input_type)
|
||||
elif isinstance(response, ToolMessage):
|
||||
response.content = cast(
|
||||
Union[str, list], msg_content_output(response.content)
|
||||
)
|
||||
return response
|
||||
else:
|
||||
raise TypeError(
|
||||
f"Tool {call['name']} returned unexpected type: {type(response)}"
|
||||
)
|
||||
|
||||
async def _arun_one(
|
||||
self,
|
||||
call: ToolCall,
|
||||
input_type: Literal["list", "dict", "tool_calls"],
|
||||
config: RunnableConfig,
|
||||
) -> ToolMessage:
|
||||
if invalid_tool_message := self._validate_tool_call(call):
|
||||
return invalid_tool_message
|
||||
|
||||
try:
|
||||
input = {**call, **{"type": "tool_call"}}
|
||||
response = await self.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)
|
||||
except GraphBubbleUp as e:
|
||||
raise e
|
||||
except Exception as e:
|
||||
if isinstance(self.handle_tool_errors, tuple):
|
||||
handled_types: tuple = self.handle_tool_errors
|
||||
elif callable(self.handle_tool_errors):
|
||||
handled_types = _infer_handled_types(self.handle_tool_errors)
|
||||
else:
|
||||
# default behavior is catching all exceptions
|
||||
handled_types = (Exception,)
|
||||
|
||||
# Unhandled
|
||||
if not self.handle_tool_errors or not isinstance(e, handled_types):
|
||||
raise e
|
||||
# Handled
|
||||
else:
|
||||
content = _handle_tool_error(e, flag=self.handle_tool_errors)
|
||||
|
||||
return ToolMessage(
|
||||
content=content,
|
||||
name=call["name"],
|
||||
tool_call_id=call["id"],
|
||||
status="error",
|
||||
)
|
||||
|
||||
if isinstance(response, Command):
|
||||
return self._validate_tool_command(response, call, input_type)
|
||||
elif isinstance(response, ToolMessage):
|
||||
response.content = cast(
|
||||
Union[str, list], msg_content_output(response.content)
|
||||
)
|
||||
return response
|
||||
else:
|
||||
raise TypeError(
|
||||
f"Tool {call['name']} returned unexpected type: {type(response)}"
|
||||
)
|
||||
|
||||
def _parse_input(
|
||||
self,
|
||||
input: Union[
|
||||
list[AnyMessage],
|
||||
dict[str, Any],
|
||||
BaseModel,
|
||||
],
|
||||
store: Optional[BaseStore],
|
||||
) -> Tuple[list[ToolCall], 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"
|
||||
tool_calls = input
|
||||
return tool_calls, input_type
|
||||
else:
|
||||
input_type = "list"
|
||||
message: AnyMessage = input[-1]
|
||||
elif isinstance(input, dict) and (messages := input.get(self.messages_key, [])):
|
||||
input_type = "dict"
|
||||
message = messages[-1]
|
||||
elif messages := getattr(input, self.messages_key, None):
|
||||
# Assume dataclass-like state that can coerce from dict
|
||||
input_type = "dict"
|
||||
message = messages[-1]
|
||||
else:
|
||||
raise ValueError("No message found in input")
|
||||
|
||||
if not isinstance(message, AIMessage):
|
||||
raise ValueError("Last message is not an AIMessage")
|
||||
|
||||
tool_calls = [
|
||||
self.inject_tool_args(call, input, store) for call in 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:
|
||||
content = INVALID_TOOL_NAME_ERROR_TEMPLATE.format(
|
||||
requested_tool=requested_tool,
|
||||
available_tools=", ".join(self.tools_by_name.keys()),
|
||||
)
|
||||
return ToolMessage(
|
||||
content, name=requested_tool, tool_call_id=call["id"], status="error"
|
||||
)
|
||||
else:
|
||||
return None
|
||||
|
||||
def _inject_state(
|
||||
self,
|
||||
tool_call: ToolCall,
|
||||
input: Union[
|
||||
list[AnyMessage],
|
||||
dict[str, Any],
|
||||
BaseModel,
|
||||
],
|
||||
) -> ToolCall:
|
||||
state_args = self.tool_to_state_args[tool_call["name"]]
|
||||
if state_args and isinstance(input, list):
|
||||
required_fields = list(state_args.values())
|
||||
if (
|
||||
len(required_fields) == 1
|
||||
and required_fields[0] == self.messages_key
|
||||
or required_fields[0] is None
|
||||
):
|
||||
input = {self.messages_key: input}
|
||||
else:
|
||||
err_msg = (
|
||||
f"Invalid input to ToolNode. Tool {tool_call['name']} requires "
|
||||
f"graph state dict as input."
|
||||
)
|
||||
if any(state_field for state_field in state_args.values()):
|
||||
required_fields_str = ", ".join(f for f in required_fields if f)
|
||||
err_msg += f" State should contain fields {required_fields_str}."
|
||||
raise ValueError(err_msg)
|
||||
if isinstance(input, dict):
|
||||
tool_state_args = {
|
||||
tool_arg: input[state_field] if state_field else input
|
||||
for tool_arg, state_field in state_args.items()
|
||||
}
|
||||
|
||||
else:
|
||||
tool_state_args = {
|
||||
tool_arg: getattr(input, state_field) if state_field else input
|
||||
for tool_arg, state_field in state_args.items()
|
||||
}
|
||||
|
||||
tool_call["args"] = {
|
||||
**tool_call["args"],
|
||||
**tool_state_args,
|
||||
}
|
||||
return tool_call
|
||||
|
||||
def _inject_store(
|
||||
self, tool_call: ToolCall, store: Optional[BaseStore]
|
||||
) -> ToolCall:
|
||||
store_arg = self.tool_to_store_arg[tool_call["name"]]
|
||||
if not store_arg:
|
||||
return tool_call
|
||||
|
||||
if store is None:
|
||||
raise ValueError(
|
||||
"Cannot inject store into tools with InjectedStore annotations - "
|
||||
"please compile your graph with a store."
|
||||
)
|
||||
|
||||
tool_call["args"] = {
|
||||
**tool_call["args"],
|
||||
store_arg: store,
|
||||
}
|
||||
return tool_call
|
||||
|
||||
def inject_tool_args(
|
||||
self,
|
||||
tool_call: ToolCall,
|
||||
input: Union[
|
||||
list[AnyMessage],
|
||||
dict[str, Any],
|
||||
BaseModel,
|
||||
],
|
||||
store: Optional[BaseStore],
|
||||
) -> ToolCall:
|
||||
"""Injects the state and store into the tool call.
|
||||
|
||||
Tool arguments with types annotated as `InjectedState` and `InjectedStore` are
|
||||
ignored in tool schemas for generation purposes. This method injects them into
|
||||
tool calls for tool invocation.
|
||||
|
||||
Args:
|
||||
tool_call (ToolCall): The tool call to inject state and store into.
|
||||
input (Union[list[AnyMessage], dict[str, Any], BaseModel]): The input state
|
||||
to inject.
|
||||
store (Optional[BaseStore]): The store to inject.
|
||||
|
||||
Returns:
|
||||
ToolCall: The tool call with injected state and store.
|
||||
"""
|
||||
if tool_call["name"] not in self.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)
|
||||
return tool_call_with_store
|
||||
|
||||
def _validate_tool_command(
|
||||
self,
|
||||
command: Command,
|
||||
call: ToolCall,
|
||||
input_type: Literal["list", "dict", "tool_calls"],
|
||||
) -> Command:
|
||||
if isinstance(command.update, dict):
|
||||
# input type is dict when ToolNode is invoked with a dict input (e.g. {"messages": [AIMessage(..., tool_calls=[...])]})
|
||||
if input_type not in ("dict", "tool_calls"):
|
||||
raise ValueError(
|
||||
f"Tools can provide a dict in Command.update only when using dict with '{self.messages_key}' key as ToolNode input, "
|
||||
f"got: {command.update} for tool '{call['name']}'"
|
||||
)
|
||||
|
||||
updated_command = deepcopy(command)
|
||||
state_update = cast(dict[str, Any], updated_command.update) or {}
|
||||
messages_update = state_update.get(self.messages_key, [])
|
||||
elif isinstance(command.update, list):
|
||||
# input type is list when ToolNode is invoked with a list input (e.g. [AIMessage(..., tool_calls=[...])])
|
||||
if input_type != "list":
|
||||
raise ValueError(
|
||||
f"Tools can provide a list of messages in Command.update only when using list of messages as ToolNode input, "
|
||||
f"got: {command.update} for tool '{call['name']}'"
|
||||
)
|
||||
|
||||
updated_command = deepcopy(command)
|
||||
messages_update = updated_command.update
|
||||
else:
|
||||
return command
|
||||
|
||||
# convert to message objects if updates are in a dict format
|
||||
messages_update = convert_to_messages(messages_update)
|
||||
has_matching_tool_message = False
|
||||
for message in messages_update:
|
||||
if not isinstance(message, ToolMessage):
|
||||
continue
|
||||
|
||||
if message.tool_call_id == call["id"]:
|
||||
message.name = call["name"]
|
||||
has_matching_tool_message = True
|
||||
|
||||
# validate that we always have a ToolMessage matching the tool call in
|
||||
# Command.update if command is sent to the CURRENT graph
|
||||
if updated_command.graph is None and not has_matching_tool_message:
|
||||
example_update = (
|
||||
'`Command(update={"messages": [ToolMessage("Success", tool_call_id=tool_call_id), ...]}, ...)`'
|
||||
if input_type == "dict"
|
||||
else '`Command(update=[ToolMessage("Success", tool_call_id=tool_call_id), ...], ...)`'
|
||||
)
|
||||
raise ValueError(
|
||||
f"Expected to have a matching ToolMessage in Command.update for tool '{call['name']}', got: {messages_update}. "
|
||||
"Every tool call (LLM requesting to call a tool) in the message history MUST have a corresponding ToolMessage. "
|
||||
f"You can fix it by modifying the tool to return {example_update}."
|
||||
)
|
||||
return updated_command
|
||||
|
||||
|
||||
def tools_condition(
|
||||
state: Union[list[AnyMessage], dict[str, Any], BaseModel],
|
||||
messages_key: str = "messages",
|
||||
) -> Literal["tools", "__end__"]:
|
||||
"""Use in the conditional_edge to route to the ToolNode if the last message
|
||||
|
||||
has tool calls. Otherwise, route to the end.
|
||||
|
||||
Args:
|
||||
state (Union[list[AnyMessage], dict[str, Any], BaseModel]): The state to check for
|
||||
tool calls. Must have a list of messages (MessageGraph) or have the
|
||||
"messages" key (StateGraph).
|
||||
|
||||
Returns:
|
||||
The next node to route to.
|
||||
|
||||
|
||||
Examples:
|
||||
Create a custom ReAct-style agent with tools.
|
||||
|
||||
```pycon
|
||||
>>> from langchain_anthropic import ChatAnthropic
|
||||
>>> from langchain_core.tools import tool
|
||||
...
|
||||
>>> from langgraph.graph import StateGraph
|
||||
>>> from langgraph.prebuilt import ToolNode, tools_condition
|
||||
>>> from langgraph.graph.message import add_messages
|
||||
...
|
||||
>>> from typing import Annotated
|
||||
>>> from typing_extensions import TypedDict
|
||||
...
|
||||
>>> @tool
|
||||
>>> def divide(a: float, b: float) -> int:
|
||||
... \"\"\"Return a / b.\"\"\"
|
||||
... return a / b
|
||||
...
|
||||
>>> llm = ChatAnthropic(model="claude-3-haiku-20240307")
|
||||
>>> tools = [divide]
|
||||
...
|
||||
>>> class State(TypedDict):
|
||||
... messages: Annotated[list, add_messages]
|
||||
>>>
|
||||
>>> graph_builder = StateGraph(State)
|
||||
>>> graph_builder.add_node("tools", ToolNode(tools))
|
||||
>>> graph_builder.add_node("chatbot", lambda state: {"messages":llm.bind_tools(tools).invoke(state['messages'])})
|
||||
>>> graph_builder.add_edge("tools", "chatbot")
|
||||
>>> graph_builder.add_conditional_edges(
|
||||
... "chatbot", tools_condition
|
||||
... )
|
||||
>>> graph_builder.set_entry_point("chatbot")
|
||||
>>> graph = graph_builder.compile()
|
||||
>>> graph.invoke({"messages": {"role": "user", "content": "What's 329993 divided by 13662?"}})
|
||||
```
|
||||
"""
|
||||
if isinstance(state, list):
|
||||
ai_message = state[-1]
|
||||
elif isinstance(state, dict) and (messages := state.get(messages_key, [])):
|
||||
ai_message = messages[-1]
|
||||
elif messages := getattr(state, messages_key, []):
|
||||
ai_message = messages[-1]
|
||||
else:
|
||||
raise ValueError(f"No messages found in input state to tool_edge: {state}")
|
||||
if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
|
||||
return "tools"
|
||||
return "__end__"
|
||||
|
||||
|
||||
class InjectedState(InjectedToolArg):
|
||||
"""Annotation for a Tool arg that is meant to be populated with the graph state.
|
||||
|
||||
Any Tool argument annotated with InjectedState will be hidden from a tool-calling
|
||||
model, so that the model doesn't attempt to generate the argument. If using
|
||||
ToolNode, the appropriate graph state field will be automatically injected into
|
||||
the model-generated tool args.
|
||||
|
||||
Args:
|
||||
field: The key from state to insert. If None, the entire state is expected to
|
||||
be passed in.
|
||||
|
||||
Example:
|
||||
```python
|
||||
from typing import List
|
||||
from typing_extensions import Annotated, TypedDict
|
||||
|
||||
from langchain_core.messages import BaseMessage, AIMessage
|
||||
from langchain_core.tools import tool
|
||||
|
||||
from langgraph.prebuilt import InjectedState, ToolNode
|
||||
|
||||
|
||||
class AgentState(TypedDict):
|
||||
messages: List[BaseMessage]
|
||||
foo: str
|
||||
|
||||
@tool
|
||||
def state_tool(x: int, state: Annotated[dict, InjectedState]) -> str:
|
||||
'''Do something with state.'''
|
||||
if len(state["messages"]) > 2:
|
||||
return state["foo"] + str(x)
|
||||
else:
|
||||
return "not enough messages"
|
||||
|
||||
@tool
|
||||
def foo_tool(x: int, foo: Annotated[str, InjectedState("foo")]) -> str:
|
||||
'''Do something else with state.'''
|
||||
return foo + str(x + 1)
|
||||
|
||||
node = ToolNode([state_tool, foo_tool])
|
||||
|
||||
tool_call1 = {"name": "state_tool", "args": {"x": 1}, "id": "1", "type": "tool_call"}
|
||||
tool_call2 = {"name": "foo_tool", "args": {"x": 1}, "id": "2", "type": "tool_call"}
|
||||
state = {
|
||||
"messages": [AIMessage("", tool_calls=[tool_call1, tool_call2])],
|
||||
"foo": "bar",
|
||||
}
|
||||
node.invoke(state)
|
||||
```
|
||||
|
||||
```pycon
|
||||
[
|
||||
ToolMessage(content='not enough messages', name='state_tool', tool_call_id='1'),
|
||||
ToolMessage(content='bar2', name='foo_tool', tool_call_id='2')
|
||||
]
|
||||
```
|
||||
""" # noqa: E501
|
||||
|
||||
def __init__(self, field: Optional[str] = None) -> None:
|
||||
self.field = field
|
||||
|
||||
|
||||
class InjectedStore(InjectedToolArg):
|
||||
"""Annotation for a Tool arg that is meant to be populated with LangGraph store.
|
||||
|
||||
Any Tool argument annotated with InjectedStore will be hidden from a tool-calling
|
||||
model, so that the model doesn't attempt to generate the argument. If using
|
||||
ToolNode, the appropriate store field will be automatically injected into
|
||||
the model-generated tool args. Note: if a graph is compiled with a store object,
|
||||
the store will be automatically propagated to the tools with InjectedStore args
|
||||
when using ToolNode.
|
||||
|
||||
!!! Warning
|
||||
`InjectedStore` annotation requires `langchain-core >= 0.3.8`
|
||||
|
||||
Example:
|
||||
```python
|
||||
from typing import Any
|
||||
from typing_extensions import Annotated
|
||||
|
||||
from langchain_core.messages import AIMessage
|
||||
from langchain_core.tools import tool
|
||||
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
from langgraph.prebuilt import InjectedStore, ToolNode
|
||||
|
||||
store = InMemoryStore()
|
||||
store.put(("values",), "foo", {"bar": 2})
|
||||
|
||||
@tool
|
||||
def store_tool(x: int, my_store: Annotated[Any, InjectedStore()]) -> str:
|
||||
'''Do something with store.'''
|
||||
stored_value = my_store.get(("values",), "foo").value["bar"]
|
||||
return stored_value + x
|
||||
|
||||
node = ToolNode([store_tool])
|
||||
|
||||
tool_call = {"name": "store_tool", "args": {"x": 1}, "id": "1", "type": "tool_call"}
|
||||
state = {
|
||||
"messages": [AIMessage("", tool_calls=[tool_call])],
|
||||
}
|
||||
|
||||
node.invoke(state, store=store)
|
||||
```
|
||||
|
||||
```pycon
|
||||
{
|
||||
"messages": [
|
||||
ToolMessage(content='3', name='store_tool', tool_call_id='1'),
|
||||
]
|
||||
}
|
||||
```
|
||||
""" # noqa: E501
|
||||
|
||||
|
||||
def _is_injection(
|
||||
type_arg: Any, injection_type: Union[Type[InjectedState], Type[InjectedStore]]
|
||||
) -> bool:
|
||||
if isinstance(type_arg, injection_type) or (
|
||||
isinstance(type_arg, type) and issubclass(type_arg, injection_type)
|
||||
):
|
||||
return True
|
||||
origin_ = get_origin(type_arg)
|
||||
if origin_ is Union or origin_ is Annotated:
|
||||
return any(_is_injection(ta, injection_type) for ta in get_args(type_arg))
|
||||
return False
|
||||
|
||||
|
||||
def _get_state_args(tool: BaseTool) -> dict[str, Optional[str]]:
|
||||
full_schema = tool.get_input_schema()
|
||||
tool_args_to_state_fields: dict = {}
|
||||
|
||||
for name, type_ in get_all_basemodel_annotations(full_schema).items():
|
||||
injections = [
|
||||
type_arg
|
||||
for type_arg in get_args(type_)
|
||||
if _is_injection(type_arg, InjectedState)
|
||||
]
|
||||
if len(injections) > 1:
|
||||
raise ValueError(
|
||||
"A tool argument should not be annotated with InjectedState more than "
|
||||
f"once. Received arg {name} with annotations {injections}."
|
||||
)
|
||||
elif len(injections) == 1:
|
||||
injection = injections[0]
|
||||
if isinstance(injection, InjectedState) and injection.field:
|
||||
tool_args_to_state_fields[name] = injection.field
|
||||
else:
|
||||
tool_args_to_state_fields[name] = None
|
||||
else:
|
||||
pass
|
||||
return tool_args_to_state_fields
|
||||
|
||||
|
||||
def _get_store_arg(tool: BaseTool) -> Optional[str]:
|
||||
full_schema = tool.get_input_schema()
|
||||
for name, type_ in get_all_basemodel_annotations(full_schema).items():
|
||||
injections = [
|
||||
type_arg
|
||||
for type_arg in get_args(type_)
|
||||
if _is_injection(type_arg, InjectedStore)
|
||||
]
|
||||
if len(injections) > 1:
|
||||
ValueError(
|
||||
"A tool argument should not be annotated with InjectedStore more than "
|
||||
f"once. Received arg {name} with annotations {injections}."
|
||||
)
|
||||
elif len(injections) == 1:
|
||||
return name
|
||||
else:
|
||||
pass
|
||||
|
||||
return None
|
||||
@@ -1,251 +0,0 @@
|
||||
"""This module provides a ValidationNode class that can be used to validate tool calls
|
||||
in a langchain graph. It applies a pydantic schema to tool_calls in the models' outputs,
|
||||
and returns a ToolMessage with the validated content. If the schema is not valid, it
|
||||
returns a ToolMessage with the error message. The ValidationNode can be used in a
|
||||
StateGraph with a "messages" key or in a MessageGraph. If multiple tool calls are
|
||||
requested, they will be run in parallel.
|
||||
"""
|
||||
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Dict,
|
||||
Optional,
|
||||
Sequence,
|
||||
Tuple,
|
||||
Type,
|
||||
Union,
|
||||
cast,
|
||||
)
|
||||
|
||||
from langchain_core.messages import (
|
||||
AIMessage,
|
||||
AnyMessage,
|
||||
ToolCall,
|
||||
ToolMessage,
|
||||
)
|
||||
from langchain_core.runnables import (
|
||||
RunnableConfig,
|
||||
)
|
||||
from langchain_core.runnables.config import get_executor_for_config
|
||||
from langchain_core.tools import BaseTool, create_schema_from_function
|
||||
from langchain_core.utils.pydantic import is_basemodel_subclass
|
||||
from pydantic import BaseModel, ValidationError
|
||||
from pydantic.v1 import BaseModel as BaseModelV1
|
||||
from pydantic.v1 import ValidationError as ValidationErrorV1
|
||||
|
||||
from langgraph.utils.runnable import RunnableCallable
|
||||
|
||||
|
||||
def _default_format_error(
|
||||
error: BaseException,
|
||||
call: ToolCall,
|
||||
schema: Union[Type[BaseModel], Type[BaseModelV1]],
|
||||
) -> str:
|
||||
"""Default error formatting function."""
|
||||
return f"{repr(error)}\n\nRespond after fixing all validation errors."
|
||||
|
||||
|
||||
class ValidationNode(RunnableCallable):
|
||||
"""A node that validates all tools requests from the last AIMessage.
|
||||
|
||||
It can be used either in StateGraph with a "messages" key or in MessageGraph.
|
||||
|
||||
!!! note
|
||||
|
||||
This node does not actually **run** the tools, it only validates the tool calls,
|
||||
which is useful for extraction and other use cases where you need to generate
|
||||
structured output that conforms to a complex schema without losing the original
|
||||
messages and tool IDs (for use in multi-turn conversations).
|
||||
|
||||
Args:
|
||||
schemas: A list of schemas to validate the tool calls with. These can be
|
||||
any of the following:
|
||||
- A pydantic BaseModel class
|
||||
- A BaseTool instance (the args_schema will be used)
|
||||
- A function (a schema will be created from the function signature)
|
||||
format_error: A function that takes an exception, a ToolCall, and a schema
|
||||
and returns a formatted error string. By default, it returns the
|
||||
exception repr and a message to respond after fixing validation errors.
|
||||
name: The name of the node.
|
||||
tags: A list of tags to add to the node.
|
||||
|
||||
Returns:
|
||||
(Union[Dict[str, List[ToolMessage]], Sequence[ToolMessage]]): A list of ToolMessages with the validated content or error messages.
|
||||
|
||||
Examples:
|
||||
Example usage for re-prompting the model to generate a valid response:
|
||||
>>> from typing import Literal, Annotated
|
||||
>>> from typing_extensions import TypedDict
|
||||
...
|
||||
>>> from langchain_anthropic import ChatAnthropic
|
||||
>>> from pydantic import BaseModel, field_validator
|
||||
...
|
||||
>>> from langgraph.graph import END, START, StateGraph
|
||||
>>> from langgraph.prebuilt import ValidationNode
|
||||
>>> from langgraph.graph.message import add_messages
|
||||
...
|
||||
...
|
||||
>>> class SelectNumber(BaseModel):
|
||||
... a: int
|
||||
...
|
||||
... @field_validator("a")
|
||||
... def a_must_be_meaningful(cls, v):
|
||||
... if v != 37:
|
||||
... raise ValueError("Only 37 is allowed")
|
||||
... return v
|
||||
...
|
||||
...
|
||||
>>> builder = StateGraph(Annotated[list, add_messages])
|
||||
>>> llm = ChatAnthropic(model="claude-3-5-haiku-latest").bind_tools([SelectNumber])
|
||||
>>> builder.add_node("model", llm)
|
||||
>>> builder.add_node("validation", ValidationNode([SelectNumber]))
|
||||
>>> builder.add_edge(START, "model")
|
||||
...
|
||||
...
|
||||
>>> def should_validate(state: list) -> Literal["validation", "__end__"]:
|
||||
... if state[-1].tool_calls:
|
||||
... return "validation"
|
||||
... return END
|
||||
...
|
||||
...
|
||||
>>> builder.add_conditional_edges("model", should_validate)
|
||||
...
|
||||
...
|
||||
>>> def should_reprompt(state: list) -> Literal["model", "__end__"]:
|
||||
... for msg in state[::-1]:
|
||||
... # None of the tool calls were errors
|
||||
... if msg.type == "ai":
|
||||
... return END
|
||||
... if msg.additional_kwargs.get("is_error"):
|
||||
... return "model"
|
||||
... return END
|
||||
...
|
||||
...
|
||||
>>> builder.add_conditional_edges("validation", should_reprompt)
|
||||
...
|
||||
...
|
||||
>>> graph = builder.compile()
|
||||
>>> res = graph.invoke(("user", "Select a number, any number"))
|
||||
>>> # Show the retry logic
|
||||
>>> for msg in res:
|
||||
... msg.pretty_print()
|
||||
================================ Human Message =================================
|
||||
Select a number, any number
|
||||
================================== Ai Message ==================================
|
||||
[{'id': 'toolu_01JSjT9Pq8hGmTgmMPc6KnvM', 'input': {'a': 42}, 'name': 'SelectNumber', 'type': 'tool_use'}]
|
||||
Tool Calls:
|
||||
SelectNumber (toolu_01JSjT9Pq8hGmTgmMPc6KnvM)
|
||||
Call ID: toolu_01JSjT9Pq8hGmTgmMPc6KnvM
|
||||
Args:
|
||||
a: 42
|
||||
================================= Tool Message =================================
|
||||
Name: SelectNumber
|
||||
ValidationError(model='SelectNumber', errors=[{'loc': ('a',), 'msg': 'Only 37 is allowed', 'type': 'value_error'}])
|
||||
Respond after fixing all validation errors.
|
||||
================================== Ai Message ==================================
|
||||
[{'id': 'toolu_01PkxSVxNxc5wqwCPW1FiSmV', 'input': {'a': 37}, 'name': 'SelectNumber', 'type': 'tool_use'}]
|
||||
Tool Calls:
|
||||
SelectNumber (toolu_01PkxSVxNxc5wqwCPW1FiSmV)
|
||||
Call ID: toolu_01PkxSVxNxc5wqwCPW1FiSmV
|
||||
Args:
|
||||
a: 37
|
||||
================================= Tool Message =================================
|
||||
Name: SelectNumber
|
||||
{"a": 37}
|
||||
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
schemas: Sequence[Union[BaseTool, Type[BaseModel], Callable]],
|
||||
*,
|
||||
format_error: Optional[
|
||||
Callable[[BaseException, ToolCall, Type[BaseModel]], str]
|
||||
] = None,
|
||||
name: str = "validation",
|
||||
tags: Optional[list[str]] = None,
|
||||
) -> None:
|
||||
super().__init__(self._func, None, name=name, tags=tags, trace=False)
|
||||
self._format_error = format_error or _default_format_error
|
||||
self.schemas_by_name: Dict[str, Type[BaseModel]] = {}
|
||||
for schema in schemas:
|
||||
if isinstance(schema, BaseTool):
|
||||
if schema.args_schema is None:
|
||||
raise ValueError(
|
||||
f"Tool {schema.name} does not have an args_schema defined."
|
||||
)
|
||||
elif not isinstance(
|
||||
schema.args_schema, type
|
||||
) or not is_basemodel_subclass(schema.args_schema):
|
||||
raise ValueError(
|
||||
"Validation node only works with tools that have a pydantic BaseModel args_schema. "
|
||||
f"Got {schema.name} with args_schema: {schema.args_schema}."
|
||||
)
|
||||
self.schemas_by_name[schema.name] = schema.args_schema
|
||||
elif isinstance(schema, type) and issubclass(
|
||||
schema, (BaseModel, BaseModelV1)
|
||||
):
|
||||
self.schemas_by_name[schema.__name__] = cast(Type[BaseModel], schema)
|
||||
elif callable(schema):
|
||||
base_model = create_schema_from_function("Validation", schema)
|
||||
self.schemas_by_name[schema.__name__] = base_model
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unsupported input to ValidationNode. Expected BaseModel, tool or function. Got: {type(schema)}."
|
||||
)
|
||||
|
||||
def _get_message(
|
||||
self, input: Union[list[AnyMessage], dict[str, Any]]
|
||||
) -> Tuple[str, AIMessage]:
|
||||
"""Extract the last AIMessage from the input."""
|
||||
if isinstance(input, list):
|
||||
output_type = "list"
|
||||
messages: list = input
|
||||
elif messages := input.get("messages", []):
|
||||
output_type = "dict"
|
||||
else:
|
||||
raise ValueError("No message found in input")
|
||||
message: AnyMessage = messages[-1]
|
||||
if not isinstance(message, AIMessage):
|
||||
raise ValueError("Last message is not an AIMessage")
|
||||
return output_type, message
|
||||
|
||||
def _func(
|
||||
self, input: Union[list[AnyMessage], dict[str, Any]], config: RunnableConfig
|
||||
) -> Any:
|
||||
"""Validate and run tool calls synchronously."""
|
||||
output_type, message = self._get_message(input)
|
||||
|
||||
def run_one(call: ToolCall) -> ToolMessage:
|
||||
schema = self.schemas_by_name[call["name"]]
|
||||
try:
|
||||
if issubclass(schema, BaseModel):
|
||||
output = schema.model_validate(call["args"])
|
||||
content = output.model_dump_json()
|
||||
elif issubclass(schema, BaseModelV1):
|
||||
output = schema.validate(call["args"])
|
||||
content = output.json()
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unsupported schema type: {type(schema)}. Expected BaseModel or BaseModelV1."
|
||||
)
|
||||
return ToolMessage(
|
||||
content=content,
|
||||
name=call["name"],
|
||||
tool_call_id=cast(str, call["id"]),
|
||||
)
|
||||
except (ValidationError, ValidationErrorV1) as e:
|
||||
return ToolMessage(
|
||||
content=self._format_error(e, call, schema),
|
||||
name=call["name"],
|
||||
tool_call_id=cast(str, call["id"]),
|
||||
additional_kwargs={"is_error": True},
|
||||
)
|
||||
|
||||
with get_executor_for_config(config) as executor:
|
||||
outputs = [*executor.map(run_one, message.tool_calls)]
|
||||
if output_type == "list":
|
||||
return outputs
|
||||
else:
|
||||
return {"messages": outputs}
|
||||
Generated
-1476
File diff suppressed because it is too large
Load Diff
@@ -1,62 +0,0 @@
|
||||
[tool.poetry]
|
||||
name = "langgraph-prebuilt"
|
||||
version = "0.1.1"
|
||||
description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
|
||||
authors = []
|
||||
license = "MIT"
|
||||
readme = "README.md"
|
||||
repository = "https://www.github.com/langchain-ai/langgraph"
|
||||
packages = [{ include = "langgraph" }]
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python = "^3.9.0,<4.0"
|
||||
langgraph-checkpoint = "^2.0.10"
|
||||
langchain-core = ">=0.2.43,<0.4.0,!=0.3.0,!=0.3.1,!=0.3.2,!=0.3.3,!=0.3.4,!=0.3.5,!=0.3.6,!=0.3.7,!=0.3.8,!=0.3.9,!=0.3.10,!=0.3.11,!=0.3.12,!=0.3.13,!=0.3.14,!=0.3.15,!=0.3.16,!=0.3.17,!=0.3.18,!=0.3.19,!=0.3.20,!=0.3.21,!=0.3.22"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
ruff = "^0.6.2"
|
||||
codespell = "^2.2.0"
|
||||
pytest = "^7.2.1"
|
||||
pytest-asyncio = "^0.21.1"
|
||||
pytest-mock = "^3.11.1"
|
||||
pytest-watcher = "^0.4.1"
|
||||
mypy = "^1.10.0"
|
||||
langgraph = {path = "../langgraph", develop = true}
|
||||
langgraph-checkpoint = {path = "../checkpoint", develop = true}
|
||||
langgraph-checkpoint-sqlite = {path = "../checkpoint-sqlite", develop = true}
|
||||
langgraph-checkpoint-postgres = {path = "../checkpoint-postgres", develop = true}
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
# --strict-markers will raise errors on unknown marks.
|
||||
# https://docs.pytest.org/en/7.1.x/how-to/mark.html#raising-errors-on-unknown-marks
|
||||
#
|
||||
# https://docs.pytest.org/en/7.1.x/reference/reference.html
|
||||
# --strict-config any warnings encountered while parsing the `pytest`
|
||||
# section of the configuration file raise errors.
|
||||
addopts = "--strict-markers --strict-config --durations=5 -vv"
|
||||
asyncio_mode = "auto"
|
||||
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core"]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
|
||||
[tool.ruff]
|
||||
lint.select = [ "E", "F", "I", "TID251" ]
|
||||
lint.ignore = [ "E501" ]
|
||||
|
||||
[tool.pytest-watcher]
|
||||
now = true
|
||||
delay = 0.1
|
||||
runner_args = ["--ff", "-v", "--tb", "short"]
|
||||
patterns = ["*.py"]
|
||||
|
||||
[tool.mypy]
|
||||
# https://mypy.readthedocs.io/en/stable/config_file.html
|
||||
disallow_untyped_defs = "True"
|
||||
explicit_package_bases = "True"
|
||||
warn_no_return = "False"
|
||||
warn_unused_ignores = "True"
|
||||
warn_redundant_casts = "True"
|
||||
allow_redefinition = "True"
|
||||
disable_error_code = "typeddict-item, return-value"
|
||||
@@ -1,86 +0,0 @@
|
||||
import re
|
||||
from typing import Any, Sequence, Union
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
|
||||
class FloatBetween(float):
|
||||
def __new__(cls, min_value: float, max_value: float) -> Self:
|
||||
return super().__new__(cls, min_value)
|
||||
|
||||
def __init__(self, min_value: float, max_value: float) -> None:
|
||||
super().__init__()
|
||||
self.min_value = min_value
|
||||
self.max_value = max_value
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
return (
|
||||
isinstance(other, float)
|
||||
and other >= self.min_value
|
||||
and other <= self.max_value
|
||||
)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((float(self), self.min_value, self.max_value))
|
||||
|
||||
|
||||
class AnyStr(str):
|
||||
def __init__(self, prefix: Union[str, re.Pattern] = "") -> None:
|
||||
super().__init__()
|
||||
self.prefix = prefix
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
return isinstance(other, str) and (
|
||||
other.startswith(self.prefix)
|
||||
if isinstance(self.prefix, str)
|
||||
else self.prefix.match(other)
|
||||
)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((str(self), self.prefix))
|
||||
|
||||
|
||||
class AnyDict(dict):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if not isinstance(other, dict) or len(self) != len(other):
|
||||
return False
|
||||
for k, v in self.items():
|
||||
if kk := next((kk for kk in other if kk == k), None):
|
||||
if v == other[kk]:
|
||||
continue
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
|
||||
class AnyVersion:
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
return isinstance(other, (str, int, float))
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(str(self))
|
||||
|
||||
|
||||
class UnsortedSequence:
|
||||
def __init__(self, *values: Any) -> None:
|
||||
self.seq = values
|
||||
|
||||
def __eq__(self, value: object) -> bool:
|
||||
return (
|
||||
isinstance(value, Sequence)
|
||||
and len(self.seq) == len(value)
|
||||
and all(a in value for a in self.seq)
|
||||
)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(frozenset(self.seq))
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return repr(self.seq)
|
||||
@@ -1,17 +0,0 @@
|
||||
name: langgraph-tests
|
||||
services:
|
||||
postgres-test:
|
||||
image: postgres:16
|
||||
ports:
|
||||
- "5442:5432"
|
||||
environment:
|
||||
POSTGRES_DB: postgres
|
||||
POSTGRES_USER: postgres
|
||||
POSTGRES_PASSWORD: postgres
|
||||
healthcheck:
|
||||
test: pg_isready -U postgres
|
||||
start_period: 10s
|
||||
timeout: 1s
|
||||
retries: 5
|
||||
interval: 60s
|
||||
start_interval: 1s
|
||||
@@ -1,448 +0,0 @@
|
||||
import sys
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import AsyncIterator, Optional
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
import pytest
|
||||
from langchain_core import __version__ as core_version
|
||||
from packaging import version
|
||||
from psycopg import AsyncConnection, Connection
|
||||
from psycopg_pool import AsyncConnectionPool, ConnectionPool
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from langgraph.checkpoint.base import BaseCheckpointSaver
|
||||
from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver
|
||||
from langgraph.checkpoint.postgres.aio import (
|
||||
AsyncPostgresSaver,
|
||||
AsyncShallowPostgresSaver,
|
||||
)
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||||
from langgraph.store.base import BaseStore
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
from langgraph.store.postgres import AsyncPostgresStore, PostgresStore
|
||||
|
||||
pytest.register_assert_rewrite("tests.memory_assert")
|
||||
|
||||
DEFAULT_POSTGRES_URI = "postgres://postgres:postgres@localhost:5442/"
|
||||
# TODO: fix this once core is released
|
||||
IS_LANGCHAIN_CORE_030_OR_GREATER = version.parse(core_version) >= version.parse(
|
||||
"0.3.0.dev0"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend():
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def deterministic_uuids(mocker: MockerFixture) -> MockerFixture:
|
||||
side_effect = (
|
||||
UUID(f"00000000-0000-4000-8000-{i:012}", version=4) for i in range(10000)
|
||||
)
|
||||
return mocker.patch("uuid.uuid4", side_effect=side_effect)
|
||||
|
||||
|
||||
# checkpointer fixtures
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def checkpointer_memory():
|
||||
from tests.memory_assert import MemorySaverAssertImmutable
|
||||
|
||||
yield MemorySaverAssertImmutable()
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def checkpointer_sqlite():
|
||||
with SqliteSaver.from_conn_string(":memory:") as checkpointer:
|
||||
yield checkpointer
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _checkpointer_sqlite_aio():
|
||||
async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
|
||||
yield checkpointer
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def checkpointer_postgres():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
with PostgresSaver.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as checkpointer:
|
||||
checkpointer.setup()
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def checkpointer_postgres_shallow():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
with ShallowPostgresSaver.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as checkpointer:
|
||||
checkpointer.setup()
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def checkpointer_postgres_pipe():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
with PostgresSaver.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as checkpointer:
|
||||
checkpointer.setup()
|
||||
# setup can't run inside pipeline because of implicit transaction
|
||||
with checkpointer.conn.pipeline() as pipe:
|
||||
checkpointer.pipe = pipe
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def checkpointer_postgres_pool():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
with ConnectionPool(
|
||||
DEFAULT_POSTGRES_URI + database, max_size=10, kwargs={"autocommit": True}
|
||||
) as pool:
|
||||
checkpointer = PostgresSaver(pool)
|
||||
checkpointer.setup()
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _checkpointer_postgres_aio():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
async with AsyncPostgresSaver.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as checkpointer:
|
||||
await checkpointer.setup()
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _checkpointer_postgres_aio_shallow():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
async with AsyncShallowPostgresSaver.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as checkpointer:
|
||||
await checkpointer.setup()
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _checkpointer_postgres_aio_pipe():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
async with AsyncPostgresSaver.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as checkpointer:
|
||||
await checkpointer.setup()
|
||||
# setup can't run inside pipeline because of implicit transaction
|
||||
async with checkpointer.conn.pipeline() as pipe:
|
||||
checkpointer.pipe = pipe
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _checkpointer_postgres_aio_pool():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield checkpointer
|
||||
async with AsyncConnectionPool(
|
||||
DEFAULT_POSTGRES_URI + database, max_size=10, kwargs={"autocommit": True}
|
||||
) as pool:
|
||||
checkpointer = AsyncPostgresSaver(pool)
|
||||
await checkpointer.setup()
|
||||
yield checkpointer
|
||||
finally:
|
||||
# drop unique db
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def awith_checkpointer(
|
||||
checkpointer_name: Optional[str],
|
||||
) -> AsyncIterator[BaseCheckpointSaver]:
|
||||
if checkpointer_name is None:
|
||||
yield None
|
||||
elif checkpointer_name == "memory":
|
||||
from tests.memory_assert import MemorySaverAssertImmutable
|
||||
|
||||
yield MemorySaverAssertImmutable()
|
||||
elif checkpointer_name == "sqlite_aio":
|
||||
async with _checkpointer_sqlite_aio() as checkpointer:
|
||||
yield checkpointer
|
||||
elif checkpointer_name == "postgres_aio":
|
||||
async with _checkpointer_postgres_aio() as checkpointer:
|
||||
yield checkpointer
|
||||
elif checkpointer_name == "postgres_aio_shallow":
|
||||
async with _checkpointer_postgres_aio_shallow() as checkpointer:
|
||||
yield checkpointer
|
||||
elif checkpointer_name == "postgres_aio_pipe":
|
||||
async with _checkpointer_postgres_aio_pipe() as checkpointer:
|
||||
yield checkpointer
|
||||
elif checkpointer_name == "postgres_aio_pool":
|
||||
async with _checkpointer_postgres_aio_pool() as checkpointer:
|
||||
yield checkpointer
|
||||
else:
|
||||
raise NotImplementedError(f"Unknown checkpointer: {checkpointer_name}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _store_postgres_aio():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
async with AsyncPostgresStore.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as store:
|
||||
await store.setup()
|
||||
yield store
|
||||
finally:
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _store_postgres_aio_pipe():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
async with AsyncPostgresStore.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database
|
||||
) as store:
|
||||
await store.setup() # Run in its own transaction
|
||||
async with AsyncPostgresStore.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database, pipeline=True
|
||||
) as store:
|
||||
yield store
|
||||
finally:
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _store_postgres_aio_pool():
|
||||
if sys.version_info < (3, 10):
|
||||
pytest.skip("Async Postgres tests require Python 3.10+")
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
async with AsyncPostgresStore.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database,
|
||||
pool_config={"max_size": 10},
|
||||
) as store:
|
||||
await store.setup()
|
||||
yield store
|
||||
finally:
|
||||
async with await AsyncConnection.connect(
|
||||
DEFAULT_POSTGRES_URI, autocommit=True
|
||||
) as conn:
|
||||
await conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def store_postgres():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield store
|
||||
with PostgresStore.from_conn_string(DEFAULT_POSTGRES_URI + database) as store:
|
||||
store.setup()
|
||||
yield store
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def store_postgres_pipe():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield store
|
||||
with PostgresStore.from_conn_string(DEFAULT_POSTGRES_URI + database) as store:
|
||||
store.setup() # Run in its own transaction
|
||||
with PostgresStore.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database, pipeline=True
|
||||
) as store:
|
||||
yield store
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def store_postgres_pool():
|
||||
database = f"test_{uuid4().hex[:16]}"
|
||||
# create unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"CREATE DATABASE {database}")
|
||||
try:
|
||||
# yield store
|
||||
with PostgresStore.from_conn_string(
|
||||
DEFAULT_POSTGRES_URI + database, pool_config={"max_size": 10}
|
||||
) as store:
|
||||
store.setup()
|
||||
yield store
|
||||
finally:
|
||||
# drop unique db
|
||||
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
|
||||
conn.execute(f"DROP DATABASE {database}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def store_in_memory():
|
||||
yield InMemoryStore()
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def awith_store(store_name: Optional[str]) -> AsyncIterator[BaseStore]:
|
||||
if store_name is None:
|
||||
yield None
|
||||
elif store_name == "in_memory":
|
||||
yield InMemoryStore()
|
||||
elif store_name == "postgres_aio":
|
||||
async with _store_postgres_aio() as store:
|
||||
yield store
|
||||
elif store_name == "postgres_aio_pipe":
|
||||
async with _store_postgres_aio_pipe() as store:
|
||||
yield store
|
||||
elif store_name == "postgres_aio_pool":
|
||||
async with _store_postgres_aio_pool() as store:
|
||||
yield store
|
||||
else:
|
||||
raise NotImplementedError(f"Unknown store {store_name}")
|
||||
|
||||
|
||||
ALL_CHECKPOINTERS_SYNC = [
|
||||
"memory",
|
||||
"sqlite",
|
||||
"postgres",
|
||||
"postgres_pipe",
|
||||
"postgres_pool",
|
||||
"postgres_shallow",
|
||||
]
|
||||
|
||||
ALL_CHECKPOINTERS_ASYNC = [
|
||||
"memory",
|
||||
"sqlite_aio",
|
||||
"postgres_aio",
|
||||
"postgres_aio_pipe",
|
||||
"postgres_aio_pool",
|
||||
"postgres_aio_shallow",
|
||||
]
|
||||
@@ -1,134 +0,0 @@
|
||||
import asyncio
|
||||
import os
|
||||
import tempfile
|
||||
from collections import defaultdict
|
||||
from functools import partial
|
||||
from typing import Any, Optional
|
||||
|
||||
from langchain_core.runnables import RunnableConfig
|
||||
|
||||
from langgraph.checkpoint.base import (
|
||||
ChannelVersions,
|
||||
Checkpoint,
|
||||
CheckpointMetadata,
|
||||
CheckpointTuple,
|
||||
SerializerProtocol,
|
||||
copy_checkpoint,
|
||||
)
|
||||
from langgraph.checkpoint.memory import InMemorySaver, PersistentDict
|
||||
|
||||
|
||||
class NoopSerializer(SerializerProtocol):
|
||||
def loads_typed(self, data: tuple[str, bytes]) -> Any:
|
||||
return data[1]
|
||||
|
||||
def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
|
||||
return "type", obj
|
||||
|
||||
|
||||
class MemorySaverAssertImmutable(InMemorySaver):
|
||||
storage_for_copies: defaultdict[str, dict[str, dict[str, Checkpoint]]]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
serde: Optional[SerializerProtocol] = None,
|
||||
put_sleep: Optional[float] = None,
|
||||
) -> None:
|
||||
_, filename = tempfile.mkstemp()
|
||||
super().__init__(
|
||||
serde=serde, factory=partial(PersistentDict, filename=filename)
|
||||
)
|
||||
self.storage_for_copies = defaultdict(lambda: defaultdict(dict))
|
||||
self.put_sleep = put_sleep
|
||||
self.stack.callback(os.remove, filename)
|
||||
|
||||
def put(
|
||||
self,
|
||||
config: dict,
|
||||
checkpoint: Checkpoint,
|
||||
metadata: CheckpointMetadata,
|
||||
new_versions: ChannelVersions,
|
||||
) -> None:
|
||||
if self.put_sleep:
|
||||
import time
|
||||
|
||||
time.sleep(self.put_sleep)
|
||||
# assert checkpoint hasn't been modified since last written
|
||||
thread_id = config["configurable"]["thread_id"]
|
||||
checkpoint_ns = config["configurable"]["checkpoint_ns"]
|
||||
if saved := super().get(config):
|
||||
assert (
|
||||
self.serde.loads_typed(
|
||||
self.storage_for_copies[thread_id][checkpoint_ns][saved["id"]]
|
||||
)
|
||||
== saved
|
||||
)
|
||||
self.storage_for_copies[thread_id][checkpoint_ns][checkpoint["id"]] = (
|
||||
self.serde.dumps_typed(copy_checkpoint(checkpoint))
|
||||
)
|
||||
# call super to write checkpoint
|
||||
return super().put(config, checkpoint, metadata, new_versions)
|
||||
|
||||
|
||||
class MemorySaverAssertCheckpointMetadata(InMemorySaver):
|
||||
"""This custom checkpointer is for verifying that a run's configurable
|
||||
fields are merged with the previous checkpoint config for each step in
|
||||
the run. This is the desired behavior. Because the checkpointer's (a)put()
|
||||
method is called for each step, the implementation of this checkpointer
|
||||
should produce a side effect that can be asserted.
|
||||
"""
|
||||
|
||||
def put(
|
||||
self,
|
||||
config: RunnableConfig,
|
||||
checkpoint: Checkpoint,
|
||||
metadata: CheckpointMetadata,
|
||||
new_versions: ChannelVersions,
|
||||
) -> None:
|
||||
"""The implementation of put() merges config["configurable"] (a run's
|
||||
configurable fields) with the metadata field. The state of the
|
||||
checkpoint metadata can be asserted to confirm that the run's
|
||||
configurable fields were merged with the previous checkpoint config.
|
||||
"""
|
||||
configurable = config["configurable"].copy()
|
||||
|
||||
# remove checkpoint_id to make testing simpler
|
||||
checkpoint_id = configurable.pop("checkpoint_id", None)
|
||||
thread_id = config["configurable"]["thread_id"]
|
||||
checkpoint_ns = config["configurable"]["checkpoint_ns"]
|
||||
self.storage[thread_id][checkpoint_ns].update(
|
||||
{
|
||||
checkpoint["id"]: (
|
||||
self.serde.dumps_typed(checkpoint),
|
||||
# merge configurable fields and metadata
|
||||
self.serde.dumps_typed({**configurable, **metadata}),
|
||||
checkpoint_id,
|
||||
)
|
||||
}
|
||||
)
|
||||
return {
|
||||
"configurable": {
|
||||
"thread_id": config["configurable"]["thread_id"],
|
||||
"checkpoint_id": checkpoint["id"],
|
||||
}
|
||||
}
|
||||
|
||||
async def aput(
|
||||
self,
|
||||
config: RunnableConfig,
|
||||
checkpoint: Checkpoint,
|
||||
metadata: CheckpointMetadata,
|
||||
new_versions: ChannelVersions,
|
||||
) -> RunnableConfig:
|
||||
return await asyncio.get_running_loop().run_in_executor(
|
||||
None, self.put, config, checkpoint, metadata, new_versions
|
||||
)
|
||||
|
||||
|
||||
class MemorySaverNoPending(InMemorySaver):
|
||||
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
|
||||
result = super().get_tuple(config)
|
||||
if result:
|
||||
return CheckpointTuple(result.config, result.checkpoint, result.metadata)
|
||||
return result
|
||||
@@ -1,50 +0,0 @@
|
||||
"""Redefined messages as a work-around for pydantic issue with AnyStr.
|
||||
|
||||
The code below creates version of pydantic models
|
||||
that will work in unit tests with AnyStr as id field
|
||||
Please note that the `id` field is assigned AFTER the model is created
|
||||
to workaround an issue with pydantic ignoring the __eq__ method on
|
||||
subclassed strings.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.documents import Document
|
||||
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, ToolMessage
|
||||
|
||||
from tests.any_str import AnyStr
|
||||
|
||||
|
||||
def _AnyIdDocument(**kwargs: Any) -> Document:
|
||||
"""Create a document with an id field."""
|
||||
message = Document(**kwargs)
|
||||
message.id = AnyStr()
|
||||
return message
|
||||
|
||||
|
||||
def _AnyIdAIMessage(**kwargs: Any) -> AIMessage:
|
||||
"""Create ai message with an any id field."""
|
||||
message = AIMessage(**kwargs)
|
||||
message.id = AnyStr()
|
||||
return message
|
||||
|
||||
|
||||
def _AnyIdAIMessageChunk(**kwargs: Any) -> AIMessageChunk:
|
||||
"""Create ai message with an any id field."""
|
||||
message = AIMessageChunk(**kwargs)
|
||||
message.id = AnyStr()
|
||||
return message
|
||||
|
||||
|
||||
def _AnyIdHumanMessage(**kwargs: Any) -> HumanMessage:
|
||||
"""Create a human message with an any id field."""
|
||||
message = HumanMessage(**kwargs)
|
||||
message.id = AnyStr()
|
||||
return message
|
||||
|
||||
|
||||
def _AnyIdToolMessage(**kwargs: Any) -> ToolMessage:
|
||||
"""Create a tool message with an any id field."""
|
||||
message = ToolMessage(**kwargs)
|
||||
message.id = AnyStr()
|
||||
return message
|
||||
@@ -1,98 +0,0 @@
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Dict,
|
||||
List,
|
||||
Literal,
|
||||
Optional,
|
||||
Sequence,
|
||||
Type,
|
||||
Union,
|
||||
)
|
||||
|
||||
from langchain_core.callbacks import CallbackManagerForLLMRun
|
||||
from langchain_core.language_models import BaseChatModel, LanguageModelInput
|
||||
from langchain_core.messages import (
|
||||
AIMessage,
|
||||
BaseMessage,
|
||||
ToolCall,
|
||||
)
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
from langchain_core.runnables import Runnable, RunnableLambda
|
||||
from langchain_core.tools import BaseTool
|
||||
from pydantic import BaseModel
|
||||
|
||||
from langgraph.prebuilt.chat_agent_executor import StructuredResponse
|
||||
|
||||
|
||||
class FakeToolCallingModel(BaseChatModel):
|
||||
tool_calls: Optional[list[list[ToolCall]]] = None
|
||||
structured_response: Optional[StructuredResponse] = None
|
||||
index: int = 0
|
||||
tool_style: Literal["openai", "anthropic"] = "openai"
|
||||
|
||||
def _generate(
|
||||
self,
|
||||
messages: List[BaseMessage],
|
||||
stop: Optional[List[str]] = None,
|
||||
run_manager: Optional[CallbackManagerForLLMRun] = None,
|
||||
**kwargs: Any,
|
||||
) -> ChatResult:
|
||||
"""Top Level call"""
|
||||
messages_string = "-".join([m.content for m in messages])
|
||||
tool_calls = (
|
||||
self.tool_calls[self.index % len(self.tool_calls)]
|
||||
if self.tool_calls
|
||||
else []
|
||||
)
|
||||
message = AIMessage(
|
||||
content=messages_string, id=str(self.index), tool_calls=tool_calls.copy()
|
||||
)
|
||||
self.index += 1
|
||||
return ChatResult(generations=[ChatGeneration(message=message)])
|
||||
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
return "fake-tool-call-model"
|
||||
|
||||
def with_structured_output(
|
||||
self, schema: Type[BaseModel]
|
||||
) -> Runnable[LanguageModelInput, StructuredResponse]:
|
||||
if self.structured_response is None:
|
||||
raise ValueError("Structured response is not set")
|
||||
|
||||
return RunnableLambda(lambda x: self.structured_response)
|
||||
|
||||
def bind_tools(
|
||||
self,
|
||||
tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
|
||||
**kwargs: Any,
|
||||
) -> Runnable[LanguageModelInput, BaseMessage]:
|
||||
if len(tools) == 0:
|
||||
raise ValueError("Must provide at least one tool")
|
||||
|
||||
tool_dicts = []
|
||||
for tool in tools:
|
||||
if not isinstance(tool, BaseTool):
|
||||
raise TypeError(
|
||||
"Only BaseTool is supported by FakeToolCallingModel.bind_tools"
|
||||
)
|
||||
|
||||
# NOTE: this is a simplified tool spec for testing purposes only
|
||||
if self.tool_style == "openai":
|
||||
tool_dicts.append(
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": tool.name,
|
||||
},
|
||||
}
|
||||
)
|
||||
elif self.tool_style == "anthropic":
|
||||
tool_dicts.append(
|
||||
{
|
||||
"name": tool.name,
|
||||
}
|
||||
)
|
||||
|
||||
return self.bind(tools=tool_dicts)
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -1,81 +0,0 @@
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage
|
||||
from langchain_core.tools import tool as dec_tool
|
||||
from pydantic import BaseModel
|
||||
from pydantic.v1 import BaseModel as BaseModelV1
|
||||
|
||||
from langgraph.prebuilt import ValidationNode
|
||||
|
||||
pytestmark = pytest.mark.anyio
|
||||
|
||||
|
||||
def my_function(some_val: int, some_other_val: str) -> str:
|
||||
return f"{some_val} - {some_other_val}"
|
||||
|
||||
|
||||
class MyModel(BaseModel):
|
||||
some_val: int
|
||||
some_other_val: str
|
||||
|
||||
|
||||
class MyModelV1(BaseModelV1):
|
||||
some_val: int
|
||||
some_other_val: str
|
||||
|
||||
|
||||
@dec_tool
|
||||
def my_tool(some_val: int, some_other_val: str) -> str:
|
||||
"""Cool."""
|
||||
return f"{some_val} - {some_other_val}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tool_schema",
|
||||
[
|
||||
my_function,
|
||||
MyModel,
|
||||
MyModelV1,
|
||||
my_tool,
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("use_message_key", [True, False])
|
||||
async def test_validation_node(tool_schema: Any, use_message_key: bool):
|
||||
validation_node = ValidationNode([tool_schema])
|
||||
tool_name = getattr(tool_schema, "name", getattr(tool_schema, "__name__", None))
|
||||
inputs = [
|
||||
AIMessage(
|
||||
"hi?",
|
||||
tool_calls=[
|
||||
{
|
||||
"name": tool_name,
|
||||
"args": {"some_val": 1, "some_other_val": "foo"},
|
||||
"id": "some 0",
|
||||
},
|
||||
{
|
||||
"name": tool_name,
|
||||
# Wrong type for some_val
|
||||
"args": {"some_val": "bar", "some_other_val": "foo"},
|
||||
"id": "some 1",
|
||||
},
|
||||
],
|
||||
),
|
||||
]
|
||||
if use_message_key:
|
||||
inputs = {"messages": inputs}
|
||||
result = await validation_node.ainvoke(inputs)
|
||||
if use_message_key:
|
||||
result = result["messages"]
|
||||
|
||||
def check_results(messages: list):
|
||||
assert len(messages) == 2
|
||||
assert all(m.type == "tool" for m in messages)
|
||||
assert not messages[0].additional_kwargs.get("is_error")
|
||||
assert messages[1].additional_kwargs.get("is_error")
|
||||
|
||||
check_results(result)
|
||||
result_sync = validation_node.invoke(inputs)
|
||||
if use_message_key:
|
||||
result_sync = result_sync["messages"]
|
||||
check_results(result_sync)
|
||||
Reference in New Issue
Block a user