fix(langchain): improve HITL rejection guidance (#37859)

Improves the default `reject` `ToolMessage` so models see that the human
denied the action, the tool was not executed, and the same call should
not be retried unless the user asks. Also documents that clients can
provide a custom `reject` message for domain-specific guidance.

[Docs](https://github.com/langchain-ai/docs/pull/4269)

_Opened collaboratively by Mason Daugherty and open-swe._

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <61371264+mdrxy@users.noreply.github.com>
Co-authored-by: Mason Daugherty <github@mdrxy.com>
This commit is contained in:
open-swe[bot]
2026-06-02 15:59:50 -04:00
committed by GitHub
parent 14b1a243e5
commit 17d1c274c4
4 changed files with 136 additions and 4 deletions

View File

@@ -101,7 +101,11 @@ class RejectDecision(TypedDict):
"""The type of response when a human rejects the action.""" """The type of response when a human rejects the action."""
message: NotRequired[str] message: NotRequired[str]
"""The message sent to the model explaining why the action was rejected.""" """The message sent to the model explaining why the action was rejected.
If omitted, the model is told that the tool was not executed and should not
retry the same tool call unless the user asks for it.
"""
class RespondDecision(TypedDict): class RespondDecision(TypedDict):
@@ -315,9 +319,10 @@ class HumanInTheLoopMiddleware(AgentMiddleware[StateT, ContextT, ResponseT]):
None, None,
) )
if decision["type"] == "reject" and "reject" in allowed_decisions: if decision["type"] == "reject" and "reject" in allowed_decisions:
# Create a tool message with the human's text response
content = decision.get("message") or ( content = decision.get("message") or (
f"User rejected the tool call for `{tool_call['name']}` with id {tool_call['id']}" f"User rejected the tool call for `{tool_call['name']}` with id {tool_call['id']}. "
"The tool was not executed. Do not retry this tool call unless the user "
"explicitly requests it."
) )
tool_message = ToolMessage( tool_message = ToolMessage(
content=content, content=content,

View File

@@ -0,0 +1,94 @@
"""Integration tests for `HumanInTheLoopMiddleware` with `create_agent`.
These tests exercise the reject decision against real models to confirm that the
default rejection guidance actually discourages the model from retrying a rejected
tool call. The exact message wording is asserted by the unit tests; here we verify
the end-to-end behavior that the guidance is meant to produce.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import pytest
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from langchain.agents import create_agent
from langchain.agents.middleware.human_in_the_loop import HumanInTheLoopMiddleware
if TYPE_CHECKING:
from langchain_core.runnables import RunnableConfig
from langgraph.graph.state import CompiledStateGraph
from langchain.agents.middleware.types import _InputAgentState
def _get_model(provider: str) -> Any:
"""Get chat model for the specified provider."""
if provider == "anthropic":
return pytest.importorskip("langchain_anthropic").ChatAnthropic(
model="claude-sonnet-4-5-20250929"
)
if provider == "openai":
# Matches the model reported in the originating issue (langchain-ai/langchain#33787).
return pytest.importorskip("langchain_openai").ChatOpenAI(model="gpt-5-nano")
msg = f"Unknown provider: {provider}"
raise ValueError(msg)
@tool
def get_weather(location: str) -> str:
"""Get the current weather for a location."""
return f"It is sunny in {location}."
@pytest.mark.parametrize("provider", ["anthropic", "openai"])
def test_hitl_reject_does_not_retry(provider: str) -> None:
"""A rejected tool call should not be retried after the default guidance.
Because `interrupt_on` still applies to `get_weather`, any retry would re-trigger
the interrupt. So a completed run with no new `__interrupt__` is a reliable signal
that the model honored the guidance and did not retry.
"""
agent: CompiledStateGraph[Any, Any, _InputAgentState, Any] = create_agent(
model=_get_model(provider),
tools=[get_weather],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={"get_weather": {"allowed_decisions": ["approve", "reject"]}}
)
],
checkpointer=InMemorySaver(),
)
config: RunnableConfig = {"configurable": {"thread_id": "reject-test"}}
interrupted = agent.invoke(
{"messages": [HumanMessage("What is the weather in Paris?")]}, config
)
assert "__interrupt__" in interrupted, "Expected the tool call to trigger an interrupt"
final = agent.invoke(Command(resume={"decisions": [{"type": "reject"}]}), config)
# The model must not retry: a retry would re-interrupt instead of completing.
assert "__interrupt__" not in final, "Model retried the rejected tool call"
# Exactly one rejection ToolMessage for the original (rejected) call.
reject_messages = [
msg
for msg in final["messages"]
if isinstance(msg, ToolMessage) and msg.name == "get_weather" and msg.status == "error"
]
assert len(reject_messages) == 1, "Expected exactly one rejection ToolMessage"
# The tool should have been called exactly once (the rejected call), never re-invoked.
weather_tool_calls = [
tool_call
for msg in final["messages"]
if isinstance(msg, AIMessage)
for tool_call in msg.tool_calls
if tool_call["name"] == "get_weather"
]
assert len(weather_tool_calls) == 1, "Model issued more than one get_weather call"

View File

@@ -1,4 +1,4 @@
"""Integration tests for ShellToolMiddleware with create_agent.""" """Integration tests for `ShellToolMiddleware` with `create_agent`."""
from __future__ import annotations from __future__ import annotations

View File

@@ -151,6 +151,39 @@ def test_human_in_the_loop_middleware_single_tool_response() -> None:
assert result["messages"][1].tool_call_id == "1" assert result["messages"][1].tool_call_id == "1"
def test_human_in_the_loop_middleware_default_rejection_message() -> None:
"""Test reject decision default message discourages retries."""
middleware = HumanInTheLoopMiddleware(
interrupt_on={"test_tool": {"allowed_decisions": ["approve", "edit", "reject"]}}
)
ai_message = AIMessage(
content="I'll help you",
tool_calls=[{"name": "test_tool", "args": {"input": "test"}, "id": "1"}],
)
state = AgentState[Any](messages=[HumanMessage(content="Hello"), ai_message])
def mock_response(_: Any) -> dict[str, Any]:
return {"decisions": [{"type": "reject"}]}
with patch(
"langchain.agents.middleware.human_in_the_loop.interrupt", side_effect=mock_response
):
result = middleware.after_model(state, Runtime())
assert result is not None
assert len(result["messages"]) == 2
tool_message = result["messages"][1]
assert isinstance(tool_message, ToolMessage)
assert tool_message.content == (
"User rejected the tool call for `test_tool` with id 1. "
"The tool was not executed. Do not retry this tool call unless the user "
"explicitly requests it."
)
assert tool_message.status == "error"
assert tool_message.name == "test_tool"
assert tool_message.tool_call_id == "1"
def test_human_in_the_loop_middleware_single_tool_respond() -> None: def test_human_in_the_loop_middleware_single_tool_respond() -> None:
"""Test HumanInTheLoopMiddleware with `respond` decision producing a success ToolMessage.""" """Test HumanInTheLoopMiddleware with `respond` decision producing a success ToolMessage."""
middleware = HumanInTheLoopMiddleware( middleware = HumanInTheLoopMiddleware(