AI & Agent Dev Bug Sandbox Back to Radar

HumanInTheLoopMiddleware: Opt-In Per-Call Interrupts With Typed Response_schema

Current batched HITL interrupts force approval UIs to infer tool action context from positional lists and skip response validation, leading to fragile integrations and potential stuck threads. This change adds per-call interrupts with typed tool_approval values and per-tool response_schema validation.

mediumConfidence 70%Langchain

Origin Analysis

HumanInTheLoopMiddleware batches all pending tool call approvals into a single interrupt and provides allowed decisions in a separate review_configs list matched by position. This design removes per-tool context and omits response_schema validation, so approval UIs must parse arrays manually and malformed edit/approve payloads can be persisted, blocking thread progress.

1. Configure HumanInTheLoopMiddleware with a model that produces multiple tool calls in one turn. 2. Observe a single interrupt containing a list of actions and a parallel review_configs list. 3. Submit an approval with a malformed edit for one of the tools. 4. Notice the invalid answer is saved and the thread becomes stuck because no validation rejects it before the tool runs.

Fixing Code Block

from langgraph.types import interrupt
from typing import Any, Dict, List, Optional

class HumanInTheLoopMiddleware:
    def __init__(self, interrupt_mode: str = "batched", allowed_decisions: Optional[Dict[str, List[str]]] = None, tool_schemas: Optional[Dict[str, Dict[str, Any]]] = None, tool_descriptions: Optional[Dict[str, str]] = None):
        self.interrupt_mode = interrupt_mode
        self.allowed_decisions = allowed_decisions or {}
        self.tool_schemas = tool_schemas or {}
        self.tool_descriptions = tool_descriptions or {}

    def _build_response_schema(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
        allowed = self.allowed_decisions.get(tool_call["name"], ["approve", "reject"])
        schema: Dict[str, Any] = {
            "type": "object",
            "properties": {
                "decision": {"type": "string", "enum": allowed},
                "tool_call_id": {"type": "string", "const": tool_call["id"]},
            },
            "required": ["decision"],
        }
        if "edit" in allowed:
            schema["properties"]["args"] = self.tool_schemas[tool_call["name"]]
        if "feedback" in allowed:
            schema["properties"]["message"] = {"type": "string"}
        return schema

    def wrap_tool_call(self, tool_call: Dict[str, Any], config: Dict[str, Any]) -> Any:
        if self.interrupt_mode != "per_call":
            return self._batched_wrap_tool_call(tool_call, config)
        payload = {
            "type": "tool_approval",
            "tool_call_id": tool_call["id"],
            "name": tool_call["name"],
            "args": tool_call["args"],
            "description": self.tool_descriptions.get(tool_call["name"], ""),
        }
        answer = interrupt(payload, response_schema=self._build_response_schema(tool_call))
        decision = answer["decision"]
        if decision == "approve":
            return tool_call["args"]
        if decision == "edit":
            return answer["args"]  # LangGraph validated against tool schema
        if decision == "reject":
            return None
        if decision == "feedback":
            return answer.get("message")
        raise ValueError(f"Unknown decision: {decision}")

    def _batched_wrap_tool_call(self, tool_call: Dict[str, Any], config: Dict[str, Any]) -> Any:
        # Existing batched logic remains unchanged for backward compatibility
        raise NotImplementedError

The per-call mode interrupts individually from wrap_tool_call, using LangGraph's typed interrupt with a response_schema that restricts decisions to the tool's allowed set and pins tool_call_id. The edit branch includes the tool's args schema, so LangGraph validates edits before resuming, and rename is impossible because tool_call_id is const.

Edge Case Audit

This fix is opt-in; batched remains default until the next major. Rolling back requires setting interrupt_mode back to 'batched'. Concurrency/multiple interrupts per turn increases UI round trips; ensure approval UIs can handle multiple simultaneous interrupts. Response schema changes may reject previously accepted malformed payloads, potentially breaking existing integrations. Test with langgraph>=1.2.12; earlier versions lack typed interrupt support.