tessl/pypi-langgraph-prebuilt
Library with high-level APIs for creating and executing LangGraph agents and tools.
human-in-the-loop.mddocs/
Human-in-the-Loop Integration
TypedDict schemas for Agent Inbox integration, enabling human intervention and approval workflows within agent execution for interactive agent experiences.
Capabilities
HumanInterrupt Schema
Represents an interrupt triggered by the graph that requires human intervention. Used to pause execution and request human input through the Agent Inbox interface.
class HumanInterrupt(TypedDict):
action_request: ActionRequest
config: HumanInterruptConfig
description: Optional[str]Fields:
action_request: The specific action being requested from the humanconfig: Configuration defining what actions are alloweddescription: Optional detailed description of what input is needed
ActionRequest Schema
Represents a request for human action within the graph execution, containing the action type and associated arguments.
class ActionRequest(TypedDict):
action: str
args: dictFields:
action: Type or name of action being requested (e.g., "Approve email send")args: Key-value pairs of arguments needed for the action
HumanInterruptConfig Schema
Configuration that defines what actions are allowed for a human interrupt, controlling available interaction options when the graph is paused.
class HumanInterruptConfig(TypedDict):
allow_ignore: bool
allow_respond: bool
allow_edit: bool
allow_accept: boolFields:
allow_ignore: Whether human can choose to ignore/skip the current stepallow_respond: Whether human can provide text response/feedbackallow_edit: Whether human can edit the provided content/stateallow_accept: Whether human can accept/approve the current state
HumanResponse Schema
The response provided by a human to an interrupt, returned when graph execution resumes after human interaction.
class HumanResponse(TypedDict):
type: Literal["accept", "ignore", "response", "edit"]
args: Union[None, str, ActionRequest]Fields:
type: The type of response ("accept", "ignore", "response", or "edit")args: The response payload (None for ignore/accept, str for responses, ActionRequest for edits)
Usage Examples
Basic Interrupt for Tool Approval
from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterrupt, HumanResponse, ActionRequest, HumanInterruptConfig
def approval_required_tool(state):
"""Tool that requires human approval before execution."""
# Extract tool call from messages
tool_call = state["messages"][-1].tool_calls[0]
# Create interrupt request
request: HumanInterrupt = {
"action_request": {
"action": tool_call["name"],
"args": tool_call["args"]
},
"config": {
"allow_ignore": True, # Allow skipping
"allow_respond": True, # Allow feedback
"allow_edit": False, # Don't allow editing
"allow_accept": True # Allow approval
},
"description": f"Please review and approve the {tool_call['name']} action"
}
# Send interrupt and get response
response = interrupt([request])[0]
if response["type"] == "accept":
# Execute the tool
return execute_tool(tool_call)
elif response["type"] == "ignore":
# Skip the tool execution
return {"messages": [ToolMessage(content="Action skipped by user", tool_call_id=tool_call["id"])]}
elif response["type"] == "response":
# Handle user feedback
feedback = response["args"]
return {"messages": [ToolMessage(content=f"User feedback: {feedback}", tool_call_id=tool_call["id"])]}Email Approval Workflow
def email_approval_node(state):
"""Node that requests approval before sending emails."""
# Extract email details from state
email_data = state.get("pending_email", {})
if not email_data:
return state
# Create interrupt for email approval
request: HumanInterrupt = {
"action_request": {
"action": "send_email",
"args": {
"to": email_data["to"],
"subject": email_data["subject"],
"body": email_data["body"]
}
},
"config": {
"allow_ignore": True,
"allow_respond": True,
"allow_edit": True,
"allow_accept": True
},
"description": f"Review email to {email_data['to']} with subject: {email_data['subject']}"
}
# Get human response
response = interrupt([request])[0]
if response["type"] == "accept":
# Send the email as approved
send_email(email_data)
return {"email_status": "sent", "pending_email": None}
elif response["type"] == "edit":
# Update email with human edits
edited_request = response["args"]
updated_email = {
"to": edited_request["args"]["to"],
"subject": edited_request["args"]["subject"],
"body": edited_request["args"]["body"]
}
send_email(updated_email)
return {"email_status": "sent_with_edits", "pending_email": None}
elif response["type"] == "response":
# Handle feedback without sending
feedback = response["args"]
return {
"email_status": "rejected",
"rejection_reason": feedback,
"pending_email": None
}
elif response["type"] == "ignore":
# Skip email sending
return {"email_status": "skipped", "pending_email": None}Content Moderation Workflow
def content_moderation_node(state):
"""Node for human review of generated content."""
generated_content = state.get("generated_content", "")
if not generated_content:
return state
# Create moderation request
request: HumanInterrupt = {
"action_request": {
"action": "review_content",
"args": {"content": generated_content}
},
"config": {
"allow_ignore": False, # Must review
"allow_respond": True, # Can provide feedback
"allow_edit": True, # Can edit content
"allow_accept": True # Can approve
},
"description": "Please review the generated content for appropriateness and accuracy"
}
response = interrupt([request])[0]
if response["type"] == "accept":
return {"content_status": "approved", "final_content": generated_content}
elif response["type"] == "edit":
edited_content = response["args"]["args"]["content"]
return {"content_status": "edited", "final_content": edited_content}
elif response["type"] == "response":
feedback = response["args"]
# Could regenerate content based on feedback
return {
"content_status": "needs_revision",
"feedback": feedback,
"final_content": None
}Multi-Step Approval Process
def multi_step_approval_workflow(state):
"""Workflow with multiple approval steps."""
steps = [
{
"action": "data_collection",
"description": "Approve data collection from external APIs",
"config": {"allow_ignore": True, "allow_respond": True, "allow_edit": False, "allow_accept": True}
},
{
"action": "data_processing",
"description": "Review data processing parameters",
"config": {"allow_ignore": False, "allow_respond": True, "allow_edit": True, "allow_accept": True}
},
{
"action": "result_publication",
"description": "Approve publication of results",
"config": {"allow_ignore": True, "allow_respond": True, "allow_edit": False, "allow_accept": True}
}
]
results = []
for step in steps:
request: HumanInterrupt = {
"action_request": {
"action": step["action"],
"args": state.get(f"{step['action']}_data", {})
},
"config": step["config"],
"description": step["description"]
}
response = interrupt([request])[0]
results.append({
"step": step["action"],
"response_type": response["type"],
"response_data": response["args"]
})
# Stop if any critical step is rejected
if response["type"] == "ignore" and step["action"] in ["data_processing"]:
return {"workflow_status": "aborted", "approval_results": results}
return {"workflow_status": "completed", "approval_results": results}Advanced Patterns
Conditional Interrupts
def conditional_interrupt_node(state):
"""Only interrupt for certain conditions."""
action = state.get("pending_action", {})
user_role = state.get("user_role", "user")
# Only require approval for sensitive actions or non-admin users
requires_approval = (
action.get("type") in ["delete", "modify_permissions"] or
user_role != "admin"
)
if not requires_approval:
# Execute without approval
return execute_action(action)
# Request approval
request: HumanInterrupt = {
"action_request": {
"action": action["type"],
"args": action["details"]
},
"config": {
"allow_ignore": user_role == "admin",
"allow_respond": True,
"allow_edit": user_role == "admin",
"allow_accept": True
},
"description": f"Approval required for {action['type']} action by {user_role}"
}
response = interrupt([request])[0]
return handle_approval_response(response, action)Timeout and Escalation
import time
from datetime import datetime, timedelta
def interrupt_with_escalation(state):
"""Interrupt with escalation if no response within timeout."""
request: HumanInterrupt = {
"action_request": {
"action": "urgent_approval",
"args": state["urgent_data"]
},
"config": {
"allow_ignore": False,
"allow_respond": True,
"allow_edit": False,
"allow_accept": True
},
"description": "URGENT: Immediate approval required for critical action"
}
# Set timeout for response
start_time = datetime.now()
timeout_minutes = 5
# First interrupt attempt
response = interrupt([request])[0]
# Check if this is a timeout scenario (implementation-dependent)
if datetime.now() - start_time > timedelta(minutes=timeout_minutes):
# Escalate to manager
escalation_request: HumanInterrupt = {
"action_request": request["action_request"],
"config": {
"allow_ignore": False,
"allow_respond": True,
"allow_edit": True,
"allow_accept": True
},
"description": f"ESCALATED: No response received within {timeout_minutes} minutes. Manager approval required."
}
response = interrupt([escalation_request])[0]
return handle_escalation_response(response)Integration with LangGraph
State Schema Integration
from typing_extensions import TypedDict
from typing import List, Optional
class AgentStateWithApproval(TypedDict):
messages: List[BaseMessage]
pending_approvals: List[HumanInterrupt]
approval_responses: List[HumanResponse]
workflow_status: str
def approval_aware_agent():
"""Agent that manages approval workflows."""
graph = StateGraph(AgentStateWithApproval)
graph.add_node("agent", agent_node)
graph.add_node("request_approval", approval_request_node)
graph.add_node("process_approval", process_approval_response)
graph.add_node("tools", tool_execution_node)
# Route to approval if needed
def should_request_approval(state):
if requires_human_approval(state):
return "request_approval"
return "tools"
graph.add_conditional_edges("agent", should_request_approval)
graph.add_edge("request_approval", "process_approval")
graph.add_conditional_edges("process_approval", route_after_approval)
return graph.compile()Error Handling
def robust_interrupt_handler(state):
"""Handle interrupts with error recovery."""
try:
request: HumanInterrupt = create_interrupt_request(state)
response = interrupt([request])[0]
return process_human_response(response, state)
except Exception as e:
# Fallback if interrupt system fails
logging.error(f"Interrupt system failed: {e}")
# Default to safe action or escalate
return {
"interrupt_error": str(e),
"fallback_action": "escalate_to_admin",
"original_request": state.get("pending_action")
}Best Practices
Request Design
# Good: Clear, specific action descriptions
request: HumanInterrupt = {
"action_request": {
"action": "send_customer_email", # Specific action name
"args": {
"recipient": "customer@example.com",
"template": "order_confirmation",
"order_id": "12345"
}
},
"config": {
"allow_ignore": False, # Critical action
"allow_respond": True, # Allow feedback
"allow_edit": True, # Allow template edits
"allow_accept": True
},
"description": "Send order confirmation email to customer for order #12345. Review template content and recipient before sending."
}Response Handling
def comprehensive_response_handler(response: HumanResponse, context: dict):
"""Handle all possible response types comprehensively."""
response_type = response["type"]
args = response["args"]
if response_type == "accept":
return execute_approved_action(context)
elif response_type == "ignore":
return log_skipped_action(context, "User chose to skip")
elif response_type == "response":
feedback = args
return process_feedback(feedback, context)
elif response_type == "edit":
edited_request = args
return execute_edited_action(edited_request, context)
else:
raise ValueError(f"Unknown response type: {response_type}")Configuration Guidelines
# Sensitive actions - require explicit approval
sensitive_config = HumanInterruptConfig(
allow_ignore=False, # Must make a decision
allow_respond=True, # Can explain reasoning
allow_edit=False, # No modifications allowed
allow_accept=True
)
# Content review - allow editing
content_review_config = HumanInterruptConfig(
allow_ignore=False,
allow_respond=True,
allow_edit=True, # Can modify content
allow_accept=True
)
# Optional approval - can be skipped
optional_approval_config = HumanInterruptConfig(
allow_ignore=True, # Can skip if needed
allow_respond=True,
allow_edit=True,
allow_accept=True
)Install with Tessl CLI
npx tessl i tessl/pypi-langgraph-prebuilt