HITL Configuration Guide

This guide covers all HITL configuration options and patterns for building robust approval workflows.

HITLConfig Overview

The HITLConfig class is your main interface for configuring HITL behavior:

from langcrew.hitl import HITLConfig

config = HITLConfig(
    # Tool-level interrupts
    interrupt_before_tools=["tool1", "tool2"],  # Tools requiring pre-approval
    interrupt_after_tools=["tool3"],            # Tools requiring post-review

    # Task-level interrupts (Task Mode only)
    interrupt_before_tasks=["task1"],
    interrupt_after_tasks=["task2"],

    # Agent-level interrupts (Agent Mode only)
    interrupt_before_agents=["agent1"],
    interrupt_after_agents=["agent2"],

    # Node-level interrupts (LangGraph native)
    interrupt_before_nodes=["node1"],
    interrupt_after_nodes=["node2"]
)

Configuration Patterns

Specified Tools

Target specific tools for interrupts:

Before Execution

# Approve parameters before execution
hitl_config = HITLConfig(
    interrupt_before_tools=[
        "web_search",      # Review search queries
        "file_write",      # Approve file operations
        "send_email",      # Confirm email sending
        "database_query",  # Validate database access
        "external_api"     # Review API calls
    ]
)

After Execution

# Review results after execution
hitl_config = HITLConfig(
    interrupt_after_tools=[
        "data_analysis",    # Review analysis results
        "content_generator", # Check generated content
        "report_writer",    # Validate reports
        "image_processor"   # Review processed images
    ]
)

# ⚠️ IMPORTANT: After interrupts only work within single execution session
# They will NOT trigger after workflow restarts or checkpointed state recovery

Session-Level Limitations

Understanding interrupt_after_tools Behavior

interrupt_after_tools has an important limitation that affects production deployments:

# This configuration works ONLY within a single execution session
hitl_config = HITLConfig(
    interrupt_after_tools=["data_analysis", "report_generator"]
)

# Scenario 1: Single session execution ✅
# 1. Tool executes → 2. User reviews result → 3. Workflow continues

# Scenario 2: Workflow restart ❌
# 1. Tool executes → 2. interrupt_after triggered → 3. System restarts → 4. Tool executes again (duplicate)

Why this happens:

• interrupt_after_tools occurs during tool node execution, before the next checkpoint is saved
• When system restarts, it cannot resume from the interrupt point and must restart from the last checkpoint
• The tool node will re-execute from the beginning, causing the tool to run again
• This can lead to duplicate tool executions and potential side effects

Solutions for Cross-Restart Review

Use Before Interrupts

# More reliable for critical operations
hitl_config = HITLConfig(
    interrupt_before_tools=["critical_operation"],  # Always triggers
    # interrupt_after_tools=["critical_operation"]  # May be skipped after restart
)

Re-entrant Tools

from langchain_core.tools import tool

@tool
def reentrant_analysis_tool(data: str, review_count: int = 0) -> str:
    """Analysis tool that supports multiple review cycles"""
    if review_count == 0:
        # First execution - do analysis
        result = perform_analysis(data)
        return f"Analysis complete (review #{review_count + 1}): {result}"
    else:
        # Re-execution after restart - allow re-review
        return f"Re-reviewing analysis (review #{review_count + 1}): {cached_result}"

# Configure for before-interrupt to ensure review happens
hitl_config = HITLConfig(
    interrupt_before_tools=["reentrant_analysis_tool"]
)

Node-Level Control

# Use workflow-level interrupts instead of tool-level
hitl_config = HITLConfig(
    interrupt_after_nodes=["analysis_node"],  # Works across restarts
    # interrupt_after_tools=["analysis_tool"]  # Session-limited
)

Agent-Level Configuration

Individual Agent Setup

from langcrew import Agent
from langcrew.hitl import HITLConfig

# High-privilege agent with minimal interrupts
admin_agent = Agent(
    role="System Administrator",
    tools=[SystemTool(), FileManagerTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["system_shutdown"]  # Only critical operations
    )
)

# Regular agent with standard approvals
worker_agent = Agent(
    role="Data Worker",
    tools=[DataProcessorTool(), FileWriteTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["file_write", "data_export"]
    )
)

# Public-facing agent with maximum oversight
public_agent = Agent(
    role="Public Assistant",
    tools=[WebSearchTool(), CalculatorTool(), FileWriteTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["web_search", "file_write"]  # Approve all risky operations
    )
)

Context-Aware Configuration

def create_context_aware_hitl(user_role, data_sensitivity):
    """Create HITL config based on context"""

    if user_role == "admin":
        return HITLConfig(
            interrupt_before_tools=["system_operation"]
        )
    elif data_sensitivity == "high":
        return HITLConfig(
            interrupt_before_tools=["file_write", "database_write", "external_api"]
        )
    else:
        return HITLConfig(
            interrupt_before_tools=["file_write", "external_api"]
        )

# Use in agent creation
agent = Agent(
    role="Data Processor",
    hitl=create_context_aware_hitl(
        user_role=current_user.role,
        data_sensitivity=data_classification
    )
)

Crew-Level Configuration

Unified Policy

Apply HITL policy across multiple agents:

from langcrew import Crew

# Individual agents without HITL config
researcher = Agent(role="Researcher", tools=[WebSearchTool()])
writer = Agent(role="Writer", tools=[FileWriteTool()])
reviewer = Agent(role="Reviewer", tools=[EmailTool()])

# Crew-level unified policy
crew = Crew(
    agents=[researcher, writer, reviewer],
    hitl=HITLConfig(
        interrupt_before_tools=["web_search", "file_write", "send_email"]
    )
)

Mixed Configuration

Combine agent-level and crew-level HITL:

# Agent with specific HITL config
critical_agent = Agent(
    role="Critical Operations",
    tools=[DatabaseTool()],
    hitl=HITLConfig(
        interrupt_before_tools=["database_write", "database_delete"]  # Agent-specific policy
    )
)

# Agent without HITL config (will inherit from crew)
regular_agent = Agent(
    role="Regular Worker",
    tools=[FileWriteTool()]
)

# Crew with fallback policy
crew = Crew(
    agents=[critical_agent, regular_agent],
    hitl=HITLConfig(
        interrupt_before_tools=["file_write"]  # Applies to regular_agent only
    )
)

Node-Level Interrupts

Integrate with LangGraph’s native interrupt system:

from langcrew.hitl import HITLConfig

# Combine tool and node interrupts
hitl_config = HITLConfig(
    # Tool-level interrupts (LangCrew HITL system)
    interrupt_before_tools=["critical_operation"],
    interrupt_after_tools=["data_export"],

    # Node-level interrupts (LangGraph native)
    interrupt_before_nodes=["decision_node"],
    interrupt_after_nodes=["validation_node"]
)

agent = Agent(
    role="Comprehensive Agent",
    tools=[CriticalTool(), DataExportTool()],
    hitl=hitl_config
)

Response Processing

Simple Responses

HITL accepts natural language responses:

# These all mean "approve":
responses = [
    "yes", "approve", "ok", "confirm", "accept", "agreed",  # English
    "批准", "同意", "确认", "通过", "好的", "可以",              # Chinese
    True,                                                    # Boolean
    {"approved": True}                                       # Structured
]

# These all mean "deny":
responses = [
    "no", "deny", "reject", "refuse", "cancel", "disagree", # English
    "拒绝", "不同意", "不通过", "取消", "否", "不要",            # Chinese
    False,                                                   # Boolean
    {"approved": False}                                      # Structured
]

Advanced Responses

Parameter Modification (Before Interrupt)

# User can modify tool parameters
user_response = {
    "approved": True,
    "modified_args": {
        "query": "enhanced search query with more context",
        "max_results": 10,
        "language": "en",
        "safe_search": True
    }
}

Result Enhancement (After Interrupt)

# User can modify tool results
user_response = {
    "approved": True,
    "modified_result": """
    Original analysis enhanced with:
    - Additional context from domain expertise
    - Corrected data points based on recent updates
    - User insights and recommendations
    """
}

Denial with Reason

# Provide detailed rejection reason
user_response = {
    "approved": False,
    "reason": "Query parameters are too broad and may return sensitive data"
}

Environment-Based Configuration

Development vs Production

import os

def get_hitl_config():
    """Get environment-appropriate HITL configuration"""

    env = os.getenv("ENVIRONMENT", "development")

    if env == "development":
        # Minimal interrupts for faster development
        return None  # No HITL for development

    elif env == "staging":
        # Critical operations only
        return HITLConfig(
            interrupt_before_tools=["database_write", "external_api", "file_delete"]
        )

    elif env == "production":
        # Comprehensive oversight - specify all sensitive tools
        return HITLConfig(
            interrupt_before_tools=[
                "database_write", "database_delete",
                "file_write", "file_delete",
                "external_api", "send_email"
            ]
        )

    else:
        # Default to safe configuration
        return HITLConfig(
            interrupt_before_tools=["file_write", "database_query", "external_api"]
        )

# Use in agent creation
agent = Agent(
    role="Environment-Aware Agent",
    hitl=get_hitl_config()
)

Best Practices

1. Security-First Approach

# Start by specifying sensitive operations
initial_config = HITLConfig(
    interrupt_before_tools=[
        "file_write", "file_delete",
        "database_write", "database_delete",
        "external_api", "send_email"
    ]
)

# Expand as needed based on your application's requirements
production_config = HITLConfig(
    interrupt_before_tools=[
        "file_write", "file_delete",
        "database_write", "database_delete",
        "external_api", "send_email",
        "system_command", "user_data_access"  # Add more as identified
    ]
)

2. Role-Based Configuration

def get_role_based_hitl(user_role):
    """HITL configuration based on user role"""

    role_configs = {
        "admin": HITLConfig(
            interrupt_before_tools=["system_shutdown", "user_delete"]
        ),
        "manager": HITLConfig(
            interrupt_before_tools=["data_export", "report_send", "file_delete"]
        ),
        "analyst": HITLConfig(
            interrupt_before_tools=["database_write", "external_api"]
        ),
        "viewer": HITLConfig(
            interrupt_before_tools=["file_write", "database_write", "external_api"]
        )
    }

    return role_configs.get(user_role, role_configs["viewer"])

3. Tool Classification

# Classify tools by risk level
SAFE_TOOLS = ["calculator", "date_time", "text_formatter", "unit_converter"]
MODERATE_TOOLS = ["web_search", "file_read", "data_query"]
DANGEROUS_TOOLS = ["file_write", "file_delete", "database_write", "send_email"]

# Low risk: only dangerous tools need approval
low_risk_config = HITLConfig(
    interrupt_before_tools=DANGEROUS_TOOLS
)

# High risk: dangerous and moderate tools need approval
high_risk_config = HITLConfig(
    interrupt_before_tools=DANGEROUS_TOOLS + MODERATE_TOOLS
)

# Maximum security: explicitly list all tools that need approval
maximum_security_config = HITLConfig(
    interrupt_before_tools=DANGEROUS_TOOLS + MODERATE_TOOLS + ["advanced_operation"]
)

Troubleshooting

Common Issues

1. Interrupts Not Triggering

# ❌ Problem: Tool name mismatch
hitl_config = HITLConfig(interrupt_before_tools=["file_writer"])
agent = Agent(tools=[FileWriteTool()])  # Actual tool name: "file_write"

# ✅ Solution: Use exact tool names
hitl_config = HITLConfig(interrupt_before_tools=["file_write"])

# Debug: Print tool names
for tool in agent.tools:
    print(f"Tool name: {tool.name}")

2. After-Interrupts Not Working

# ❌ Problem: Using after-interrupts across sessions
result1 = crew.kickoff(config={"configurable": {"thread_id": "session1"}})  # Tool executes, triggers after-interrupt
# System restarts...
result2 = crew.kickoff(config={"configurable": {"thread_id": "session1"}})  # After-interrupt won't trigger again

# ✅ Solution: Use before-interrupts for persistent approvals
hitl_config = HITLConfig(
    interrupt_before_tools=["critical_tool"]  # Always triggers
)

3. Response Parsing Errors

# ❌ Problem: Malformed response
user_response = "approve with modifications: query=new_query"

# ✅ Solution: Use structured response
user_response = {
    "approved": True,
    "modified_args": {"query": "new_query"}
}

Debug Mode

Enable detailed logging for troubleshooting:

import logging

# Enable HITL debug logging
logging.getLogger("langcrew.hitl").setLevel(logging.DEBUG)

# Create agent and run
agent = Agent(hitl=hitl_config, verbose=True)
result = crew.kickoff(inputs=data)