WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-30 19:59:40 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
7051bccc305b4b3335d85e30117cb6bc11a6daee
langgraph/libs/langgraph/langgraph/prebuilt/chat_agent_executor.py
T

695 lines
29 KiB
Python
Raw Blame History

from typing import Callable, Literal, Optional, Sequence, Type, TypeVar, Union, cast
from langchain_core.language_models import BaseChatModel, 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 typing_extensions import Annotated, TypedDict
from langgraph._api.deprecation import deprecated_parameter
from langgraph.errors import ErrorCode, create_error_message
from langgraph.graph import StateGraph
from langgraph.graph.graph import CompiledGraph
from langgraph.graph.message import add_messages
from langgraph.managed import IsLastStep, RemainingSteps
from langgraph.prebuilt.tool_executor import ToolExecutor
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.store.base import BaseStore
from langgraph.types import Checkpointer
from langgraph.utils.runnable import RunnableCallable
# 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 operator.add
class AgentState(TypedDict):
"""The state of the agent."""
messages: Annotated[Sequence[BaseMessage], add_messages]
is_last_step: IsLastStep
remaining_steps: RemainingSteps
StateSchema = TypeVar("StateSchema", bound=AgentState)
StateSchemaType = Type[StateSchema]
STATE_MODIFIER_RUNNABLE_NAME = "StateModifier"
MessagesModifier = Union[
SystemMessage,
str,
Callable[[Sequence[BaseMessage]], Sequence[BaseMessage]],
Runnable[Sequence[BaseMessage], Sequence[BaseMessage]],
]
StateModifier = Union[
SystemMessage,
str,
Callable[[StateSchema], Sequence[BaseMessage]],
Runnable[StateSchema, Sequence[BaseMessage]],
]
def _get_state_modifier_runnable(
state_modifier: Optional[StateModifier], store: Optional[BaseStore] = None
) -> Runnable:
state_modifier_runnable: Runnable
if state_modifier is None:
state_modifier_runnable = RunnableCallable(
lambda state: state["messages"], name=STATE_MODIFIER_RUNNABLE_NAME
)
elif isinstance(state_modifier, str):
_system_message: BaseMessage = SystemMessage(content=state_modifier)
state_modifier_runnable = RunnableCallable(
lambda state: [_system_message] + state["messages"],
name=STATE_MODIFIER_RUNNABLE_NAME,
)
elif isinstance(state_modifier, SystemMessage):
state_modifier_runnable = RunnableCallable(
lambda state: [state_modifier] + state["messages"],
name=STATE_MODIFIER_RUNNABLE_NAME,
)
elif callable(state_modifier):
state_modifier_runnable = RunnableCallable(
state_modifier,
name=STATE_MODIFIER_RUNNABLE_NAME,
)
elif isinstance(state_modifier, Runnable):
state_modifier_runnable = state_modifier
else:
raise ValueError(
f"Got unexpected type for `state_modifier`: {type(state_modifier)}"
)
return state_modifier_runnable
def _convert_messages_modifier_to_state_modifier(
messages_modifier: MessagesModifier,
) -> StateModifier:
state_modifier: StateModifier
if isinstance(messages_modifier, (str, SystemMessage)):
return messages_modifier
elif callable(messages_modifier):
def state_modifier(state: AgentState) -> Sequence[BaseMessage]:
return messages_modifier(state["messages"])
return state_modifier
elif isinstance(messages_modifier, Runnable):
state_modifier = (lambda state: state["messages"]) | messages_modifier
return state_modifier
raise ValueError(
f"Got unexpected type for `messages_modifier`: {type(messages_modifier)}"
)
def _get_model_preprocessing_runnable(
state_modifier: Optional[StateModifier],
messages_modifier: Optional[MessagesModifier],
store: Optional[BaseStore],
) -> Runnable:
# Add the state or message modifier, if exists
if state_modifier is not None and messages_modifier is not None:
raise ValueError(
"Expected value for either state_modifier or messages_modifier, got values for both"
)
if state_modifier is None and messages_modifier is not None:
state_modifier = _convert_messages_modifier_to_state_modifier(messages_modifier)
return _get_state_modifier_runnable(state_modifier, store)
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 _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)
@deprecated_parameter("messages_modifier", "0.1.9", "state_modifier", removal="0.3.0")
def create_react_agent(
model: LanguageModelLike,
tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode],
*,
state_schema: Optional[StateSchemaType] = None,
messages_modifier: Optional[MessagesModifier] = None,
state_modifier: Optional[StateModifier] = None,
checkpointer: Optional[Checkpointer] = None,
store: Optional[BaseStore] = None,
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
debug: bool = False,
) -> 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, a ToolExecutor, or a ToolNode instance.
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.
messages_modifier: An optional
messages modifier. This applies to messages BEFORE they are passed into the LLM.
Can take a few different forms:
- SystemMessage: this is added to the beginning of the list of messages.
- str: This is converted to a SystemMessage and added to the beginning of the list of messages.
- Callable: This function should take in a list of messages and the output is then passed to the language model.
- Runnable: This runnable should take in a list of messages and the output is then passed to the language model.
!!! Warning
`messages_modifier` parameter is deprecated as of version 0.1.9 and will be removed in 0.2.0
state_modifier: An optional
state modifier. This takes full graph state BEFORE the LLM is called and prepares the input to LLM.
Can take a few different forms:
- SystemMessage: this is added to the beginning of the list of messages in state["messages"].
- str: This is converted to a SystemMessage and 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.
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.
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, state_modifier=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!"),
... ])
>>> def format_for_model(state: AgentState):
... # You can do more complex modifications here
... return prompt.invoke({"messages": state["messages"]})
>>>
>>> graph = create_react_agent(model, tools, state_modifier=format_for_model)
>>> 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 import TypedDict
>>> prompt = ChatPromptTemplate.from_messages(
... [
... ("system", "Today is {today}"),
... ("placeholder", "{messages}"),
... ]
... )
>>>
>>> class CustomState(TypedDict):
... today: str
... messages: Annotated[list[BaseMessage], add_messages]
... is_last_step: str
>>>
>>> graph = create_react_agent(model, tools, state_schema=CustomState, state_modifier=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], state_modifier=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 state_schema is not None:
if missing_keys := {"messages", "is_last_step"} - set(
state_schema.__annotations__
):
raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")
if isinstance(tools, ToolExecutor):
tool_classes: Sequence[BaseTool] = tools.tools
tool_node = ToolNode(tool_classes)
elif 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 _should_bind_tools(model, tool_classes):
model = cast(BaseChatModel, model).bind_tools(tool_classes)
# Define the function that determines whether to continue or not
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
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__"
# Otherwise if there is, we continue
else:
return "tools"
# we're passing store here for validation
preprocessor = _get_model_preprocessing_runnable(
state_modifier, messages_modifier, store
)
model_runnable = preprocessor | model
# Define the function that calls the model
def call_model(state: AgentState, config: RunnableConfig) -> AgentState:
_validate_chat_history(state["messages"])
response = model_runnable.invoke(state, config)
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["is_last_step"]
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 = await model_runnable.ainvoke(state, config)
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["is_last_step"]
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]}
# Define a new graph
workflow = StateGraph(state_schema or AgentState)
# 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")
# 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,
)
# 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}
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,
)
# Keep for backwards compatibility
create_tool_calling_executor = create_react_agent
__all__ = [
"create_react_agent",
"create_tool_calling_executor",
"AgentState",
]
Reference in New Issue View Git Blame Copy Permalink