github/langchain
1
0
Fork 0
mirror of https://github.com/hwchase17/langchain.git synced 2026-07-12 19:31:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(langchain): format and lint touchups for HITL conditional interrupts

Browse Source
This commit is contained in:
Nick Hollon
2026-05-12 15:49:18 -07:00
parent 700d420755
commit 902a181268
1 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_human_in_the_loop.py
View File

@@ -12,6 +12,7 @@ from langchain.agents.middleware.human_in_the_loop import (
HumanInTheLoopMiddleware,
)
from langchain.agents.middleware.types import AgentState
from langchain.tools import ToolRuntime
def test_human_in_the_loop_middleware_initialization() -> None:
@@ -887,8 +888,6 @@ def test_human_in_the_loop_middleware_preserves_order_with_rejections() -> None:
def test_interrupt_on_config_accepts_interrupt_when() -> None:
"""`InterruptOnConfig` accepts an optional `interrupt_when` predicate."""
from langchain.agents.middleware.human_in_the_loop import _InterruptWhen # noqa: F401
config: InterruptOnConfig = {
"allowed_decisions": ["approve", "reject"],
"interrupt_when": lambda _tc, _rt: True,
@@ -913,7 +912,8 @@ def test_interrupt_when_false_auto_approves() -> None:
state = AgentState[Any](messages=[HumanMessage(content="Hi"), ai_message])
def fail_if_called(_: Any) -> dict[str, Any]:
raise AssertionError("interrupt() should not be called when predicate returns False")
msg = "interrupt() should not be called when predicate returns False"
raise AssertionError(msg)
with patch(
"langchain.agents.middleware.human_in_the_loop.interrupt",
@@ -1045,8 +1045,6 @@ def test_interrupt_when_mixed_configured_tools() -> None:
def test_interrupt_when_tool_runtime_fields() -> None:
"""Predicate receives a `ToolRuntime` with correct fields and documented deviations."""
from langchain.tools import ToolRuntime
captured: dict[str, Any] = {}
def capture_predicate(tc: ToolCall, rt: ToolRuntime[Any, Any]) -> bool:
@@ -1084,11 +1082,12 @@ def test_interrupt_when_tool_runtime_fields() -> None:
def test_interrupt_when_exceptions_propagate() -> None:
"""A predicate that raises propagates the exception (no silent auto-approval)."""
class PredicateBug(RuntimeError):
class PredicateBugError(RuntimeError):
pass
def bad_predicate(_tc: Any, _rt: Any) -> bool:
raise PredicateBug("kaboom")
msg = "kaboom"
raise PredicateBugError(msg)
middleware = HumanInTheLoopMiddleware(
interrupt_on={
@@ -1104,5 +1103,5 @@ def test_interrupt_when_exceptions_propagate() -> None:
)
state = AgentState[Any](messages=[HumanMessage(content="Hi"), ai_message])
with pytest.raises(PredicateBug, match="kaboom"):
with pytest.raises(PredicateBugError, match="kaboom"):
middleware.after_model(state, Runtime())