diff --git a/libs/prebuilt/langgraph/prebuilt/interrupt.py b/libs/prebuilt/langgraph/prebuilt/interrupt.py index 0b5778d45..70c7713c8 100644 --- a/libs/prebuilt/langgraph/prebuilt/interrupt.py +++ b/libs/prebuilt/langgraph/prebuilt/interrupt.py @@ -1,12 +1,7 @@ -from copy import deepcopy -from typing import Any, Literal, Optional, Union, cast +from typing import Literal, Optional, Union -from langchain_core.messages import ToolCall, ToolMessage from typing_extensions import TypedDict -from langgraph.types import Command, interrupt -from langgraph.utils.runnable import RunnableCallable - class HumanInterruptConfig(TypedDict): """Configuration that defines what actions are allowed for a human interrupt. @@ -93,159 +88,3 @@ class HumanResponse(TypedDict): type: Literal["accept", "ignore", "response", "edit"] args: Union[None, str, ActionRequest] - - -class InterruptToolNode(RunnableCallable): - """Prebuilt post model hook node used to enable common patterns for tool interrupts. - - For any tools with specified policies, an interrupt will be raised when the LLM returns - a tool call for said tool. The interrupt policy will be used to determine what sort of resume logic is allowed. - Any of the following resume patterns are supported: - - * accept: the tool call is executed as planned - * edit: the args for the tool call are edited and then the tool call is executed - * response: text response/feedback is fed back into the LLM - * ignore: the current tool call is ignored / skipped - - Args: - **interrupt_policy: a mapping of tool names to [`HumanInterruptConfig`][prebuilt.interrupt.HumanInterruptConfig] dictionaries - specifying which interrupt patterns to enable for said tool. - - Example: - ```python - from langgraph.prebuilt import create_react_agent - from langgraph.checkpoint.memory import InMemorySaver - from langgraph.prebuilt.interrupt import HumanInterruptConfig, InterruptToolNode - from langgraph.types import Command - - - def book_hotel(hotel_name: str) -> str: - '''Book a room at the provided hotel.''' - # Some hotel API calls, a sensitive / expensive operation - return f"Booked a hotel at {hotel_name}." - - - agent = create_react_agent( - "openai:gpt-4.1", - tools=[book_hotel], - prompt="You are a hotel booking assistant.", - post_model_hook=InterruptToolNode( - book_hotel=HumanInterruptConfig( - allow_accept=True, - allow_edit=True, - allow_ignore=True, - allow_respond=True, - ) - ), - checkpointer=InMemorySaver(), - ) - - config = {"configurable": {"thread_id": 1}} - - response = agent.invoke( - {"messages": [{"role": "user", "content": "please book a hotel at the hilton inn in boston."}]}, - config=config, - ) - - response = agent.invoke(Command(resume={"type": "accept"}), config=config) - ``` - """ - - def __init__(self, **interrupt_policy: HumanInterruptConfig): - super().__init__(self._func, self._afunc) - self.interrupt_policy = interrupt_policy - - def _interrupt( - self, - tool_call: ToolCall, - interrupt_config: HumanInterruptConfig, - ) -> Union[ToolCall, ToolMessage]: - """Interrupt before a tool call and ask for human input.""" - call_id = tool_call["id"] - tool_name = tool_call["name"] - - request = HumanInterrupt( - action_request=ActionRequest( - action=tool_name, - args=tool_call["args"], - ), - config=interrupt_config, - description=f"Please review tool call for `{tool_name}` before execution.", - ) - response = interrupt([request]) - - # resume provided by agent inbox as a list - response = response[0] if isinstance(response, list) else response - - try: - response_type = response.get("type") - except AttributeError: - raise TypeError( - f"Unexpected resume value: {response}." - f"Expected a dict with `'type'` key." - ) - - if response_type == "accept" and interrupt_config["allow_accept"]: - return tool_call - elif response_type == "edit" and interrupt_config["allow_edit"]: - return ToolCall( - args=cast(ActionRequest, response)["args"]["args"], - name=tool_name, - id=call_id, - type="tool_call", - ) - elif response_type == "response" and interrupt_config["allow_respond"]: - return ToolMessage( - content=cast(str, response["args"]), - name=tool_name, - tool_call_id=call_id, - status="error", - ) - elif response_type == "ignore" and interrupt_config["allow_ignore"]: - return ToolMessage( - content=f"User ignored the tool call for `{tool_name}` with id {call_id}", - name=tool_name, - tool_call_id=call_id, - status="success", - ) - - allowed_types = [ - type_name - for type_name, is_allowed in { - "accept": interrupt_config["allow_accept"], - "edit": interrupt_config["allow_edit"], - "response": interrupt_config["allow_respond"], - "ignore": interrupt_config["allow_ignore"], - }.items() - if is_allowed - ] - - raise ValueError( - f"Unexpected human response: {response}. " - f"Expected one with `'type'` in {allowed_types} based on {tool_name}'s interrupt configuration." - ) - - def _func(self, input: dict[str, Any]) -> Command: - ai_msg = input["messages"][-1] - tool_calls: list[ToolCall] = deepcopy(ai_msg.tool_calls) or [] - tool_messages: list[ToolMessage] = [] - - for idx, tool_call in enumerate(tool_calls): - if interrupt_config := self.interrupt_policy.get(tool_call["name"]): - interrupt_result = self._interrupt( - tool_call=tool_call, interrupt_config=interrupt_config - ) - - if isinstance(interrupt_result, ToolMessage): - tool_messages.append(interrupt_result) - else: - tool_calls[idx] = interrupt_result - - updated_ai_msg = ai_msg.copy(update={"tool_calls": tool_calls}) - - # conditional routing logic for post_model_hook will direct to the tools node - # or agent node depending on if there are pending tool calls - return {"messages": [updated_ai_msg, *tool_messages]} - - async def _afunc(self, input: dict[str, Any]) -> Command: - return self._func(input) diff --git a/libs/prebuilt/tests/test_interrupt_tool_node.py b/libs/prebuilt/tests/test_interrupt_tool_node.py deleted file mode 100644 index e305d1201..000000000 --- a/libs/prebuilt/tests/test_interrupt_tool_node.py +++ /dev/null @@ -1,191 +0,0 @@ -import pytest -from langchain_core.messages import ToolMessage -from langchain_core.runnables import RunnableConfig - -from langgraph.checkpoint.base import BaseCheckpointSaver -from langgraph.prebuilt import create_react_agent -from langgraph.prebuilt.interrupt import HumanInterruptConfig, InterruptToolNode -from langgraph.types import Command -from tests.model import FakeToolCallingModel - - -def hello_tool(name: str) -> str: - """Return a greeting for the provided person.""" - return f"Hello, {name}!" - - -post_model_hook = InterruptToolNode( - hello_tool=HumanInterruptConfig( - allow_accept=True, - allow_edit=True, - allow_ignore=True, - allow_respond=True, - ) -) - -default_model = FakeToolCallingModel( - tool_calls=[ - [ - { - "name": "hello_tool", - "args": {"name": "lady gaga"}, - "id": "some-random-id", - } - ] - ] -) - - -def test_interrupt_surfaced( - request: pytest.FixtureRequest, - sync_checkpointer: BaseCheckpointSaver, -) -> None: - agent = create_react_agent( - default_model, - [hello_tool], - checkpointer=sync_checkpointer, - post_model_hook=post_model_hook, - ) - config: RunnableConfig = {"configurable": {"thread_id": "1"}} - result = agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config) - - interrupt_data = result["__interrupt__"] - assert interrupt_data[0].value == [ - { - "action_request": {"action": "hello_tool", "args": {"name": "lady gaga"}}, - "config": { - "allow_accept": True, - "allow_edit": True, - "allow_ignore": True, - "allow_respond": True, - }, - "description": "Please review tool call for `hello_tool` before execution.", - } - ] - - response = agent.invoke(Command(resume={"type": "accept"}), config=config) - tool_message: ToolMessage = response["messages"][-2] - assert tool_message.content == "Hello, lady gaga!" - assert tool_message.name == "hello_tool" - - -@pytest.mark.parametrize( - "resume, expected_content", - [ - ({"type": "accept"}, "Hello, lady gaga!"), - ( - {"type": "ignore"}, - "User ignored the tool call for `hello_tool` with id some-random-id", - ), - ( - { - "type": "edit", - "args": {"action": "hello_tool", "args": {"name": "bruno mars"}}, - }, - "Hello, bruno mars!", - ), - ], -) -def test_interrupt_resume_variants( - request: pytest.FixtureRequest, - sync_checkpointer: BaseCheckpointSaver, - resume: dict, - expected_content: str, -) -> None: - agent = create_react_agent( - default_model, - [hello_tool], - checkpointer=sync_checkpointer, - post_model_hook=post_model_hook, - ) - - config: RunnableConfig = {"configurable": {"thread_id": "1"}} - agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config) - - response = agent.invoke(Command(resume=resume), config=config) - tool_message: ToolMessage = response["messages"][-2] - assert tool_message.name == "hello_tool" - assert tool_message.content == expected_content - - if resume["type"] == "edit": - ai_msg = response["messages"][-1] - assert ai_msg.tool_calls == [ - { - "name": "hello_tool", - "args": {"name": "lady gaga"}, - "id": "some-random-id", - "type": "tool_call", - } - ] - - -def test_resume_with_response( - request: pytest.FixtureRequest, - sync_checkpointer: BaseCheckpointSaver, -) -> None: - model = FakeToolCallingModel( - tool_calls=[ - [ - { - "name": "hello_tool", - "args": {"name": "lady gaga"}, - "id": "some-random-id", - } - ], - [ - { - "name": "hello_tool", - "args": {"name": "bruno mars"}, - "id": "some-random-id-2", - } - ], - ] - ) - - agent = create_react_agent( - model, - [hello_tool], - checkpointer=sync_checkpointer, - post_model_hook=post_model_hook, - ) - - config: RunnableConfig = {"configurable": {"thread_id": "1"}} - agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config) - - # Provide user response - agent.invoke( - Command( - resume={ - "type": "response", - "args": "actually, please say hello to bruno mars", - } - ), - config=config, - ) - - # Accept the updated call - response = agent.invoke(Command(resume={"type": "accept"}), config=config) - - assert len(response["messages"]) == 6 - tool_message: ToolMessage = response["messages"][-2] - assert tool_message.name == "hello_tool" - assert tool_message.content == "Hello, bruno mars!" - - -def test_resume_with_type_not_allowed(sync_checkpointer: BaseCheckpointSaver) -> None: - agent = create_react_agent( - default_model, - [hello_tool], - checkpointer=sync_checkpointer, - post_model_hook=post_model_hook, - ) - config: RunnableConfig = {"configurable": {"thread_id": "1"}} - agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config) - - with pytest.raises(ValueError) as exc_info: - agent.invoke(Command(resume={"type": "not-allowed"}), config=config) - - assert ( - str(exc_info.value) - == "Unexpected human response: {'type': 'not-allowed'}. Expected one with `'type'` in ['accept', 'edit', 'response', 'ignore'] based on hello_tool's interrupt configuration." - )