HumanInTheLoopMiddleware Class — langchain Architecture

Architecture documentation for the HumanInTheLoopMiddleware class in human_in_the_loop.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  b706912e_28f0_afbf_eb10_723aa3e74c52["HumanInTheLoopMiddleware"]
  de5a7878_b3fe_95d7_2575_7f534546dc1e["AIMessage"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|extends| de5a7878_b3fe_95d7_2575_7f534546dc1e
  23036d87_b616_3e6b_013c_17cd312d2a62["human_in_the_loop.py"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|defined in| 23036d87_b616_3e6b_013c_17cd312d2a62
  4590d195_e88c_0828_70fa_70775b1c6825["__init__()"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|method| 4590d195_e88c_0828_70fa_70775b1c6825
  ba9e7a24_05d8_9c3c_8018_6e7b08711eb2["_create_action_and_config()"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|method| ba9e7a24_05d8_9c3c_8018_6e7b08711eb2
  786034cd_4c3c_2a16_2580_fdaf6714c856["_process_decision()"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|method| 786034cd_4c3c_2a16_2580_fdaf6714c856
  c27d424a_40fd_dcec_b4d0_8dbf62b73ff2["after_model()"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|method| c27d424a_40fd_dcec_b4d0_8dbf62b73ff2
  7648034b_ec9a_ba69_ba13_196efa046263["aafter_model()"]
  b706912e_28f0_afbf_eb10_723aa3e74c52 -->|method| 7648034b_ec9a_ba69_ba13_196efa046263

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/human_in_the_loop.py lines 167–387

class HumanInTheLoopMiddleware(AgentMiddleware[StateT, ContextT, ResponseT]):
    """Human in the loop middleware."""

    def __init__(
        self,
        interrupt_on: dict[str, bool | InterruptOnConfig],
        *,
        description_prefix: str = "Tool execution requires approval",
    ) -> None:
        """Initialize the human in the loop middleware.

        Args:
            interrupt_on: Mapping of tool name to allowed actions.

                If a tool doesn't have an entry, it's auto-approved by default.

                * `True` indicates all decisions are allowed: approve, edit, and reject.
                * `False` indicates that the tool is auto-approved.
                * `InterruptOnConfig` indicates the specific decisions allowed for this
                    tool.

                    The `InterruptOnConfig` can include a `description` field (`str` or
                    `Callable`) for custom formatting of the interrupt description.
            description_prefix: The prefix to use when constructing action requests.

                This is used to provide context about the tool call and the action being
                requested.

                Not used if a tool has a `description` in its `InterruptOnConfig`.
        """
        super().__init__()
        resolved_configs: dict[str, InterruptOnConfig] = {}
        for tool_name, tool_config in interrupt_on.items():
            if isinstance(tool_config, bool):
                if tool_config is True:
                    resolved_configs[tool_name] = InterruptOnConfig(
                        allowed_decisions=["approve", "edit", "reject"]
                    )
            elif tool_config.get("allowed_decisions"):
                resolved_configs[tool_name] = tool_config
        self.interrupt_on = resolved_configs
        self.description_prefix = description_prefix

    def _create_action_and_config(
        self,
        tool_call: ToolCall,
        config: InterruptOnConfig,
        state: AgentState[Any],
        runtime: Runtime[ContextT],
    ) -> tuple[ActionRequest, ReviewConfig]:
        """Create an ActionRequest and ReviewConfig for a tool call."""
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        # Generate description using the description field (str or callable)
        description_value = config.get("description")
        if callable(description_value):
            description = description_value(tool_call, state, runtime)
        elif description_value is not None:
            description = description_value
        else:
            description = f"{self.description_prefix}\n\nTool: {tool_name}\nArgs: {tool_args}"

        # Create ActionRequest with description
        action_request = ActionRequest(
            name=tool_name,
            args=tool_args,
            description=description,
        )

        # Create ReviewConfig
        # eventually can get tool information and populate args_schema from there
        review_config = ReviewConfig(
            action_name=tool_name,
            allowed_decisions=config["allowed_decisions"],
        )

        return action_request, review_config

    @staticmethod
    def _process_decision(