0
# Human-in-the-Loop Integration
1
2
TypedDict schemas for Agent Inbox integration, enabling human intervention and approval workflows within agent execution for interactive agent experiences.
3
4
## Capabilities
5
6
### HumanInterrupt Schema
7
8
Represents an interrupt triggered by the graph that requires human intervention. Used to pause execution and request human input through the Agent Inbox interface.
9
10
```python { .api }
11
class HumanInterrupt(TypedDict):
12
action_request: ActionRequest
13
config: HumanInterruptConfig
14
description: Optional[str]
15
```
16
17
**Fields:**
18
- `action_request`: The specific action being requested from the human
19
- `config`: Configuration defining what actions are allowed
20
- `description`: Optional detailed description of what input is needed
21
22
### ActionRequest Schema
23
24
Represents a request for human action within the graph execution, containing the action type and associated arguments.
25
26
```python { .api }
27
class ActionRequest(TypedDict):
28
action: str
29
args: dict
30
```
31
32
**Fields:**
33
- `action`: Type or name of action being requested (e.g., "Approve email send")
34
- `args`: Key-value pairs of arguments needed for the action
35
36
### HumanInterruptConfig Schema
37
38
Configuration that defines what actions are allowed for a human interrupt, controlling available interaction options when the graph is paused.
39
40
```python { .api }
41
class HumanInterruptConfig(TypedDict):
42
allow_ignore: bool
43
allow_respond: bool
44
allow_edit: bool
45
allow_accept: bool
46
```
47
48
**Fields:**
49
- `allow_ignore`: Whether human can choose to ignore/skip the current step
50
- `allow_respond`: Whether human can provide text response/feedback
51
- `allow_edit`: Whether human can edit the provided content/state
52
- `allow_accept`: Whether human can accept/approve the current state
53
54
### HumanResponse Schema
55
56
The response provided by a human to an interrupt, returned when graph execution resumes after human interaction.
57
58
```python { .api }
59
class HumanResponse(TypedDict):
60
type: Literal["accept", "ignore", "response", "edit"]
61
args: Union[None, str, ActionRequest]
62
```
63
64
**Fields:**
65
- `type`: The type of response ("accept", "ignore", "response", or "edit")
66
- `args`: The response payload (None for ignore/accept, str for responses, ActionRequest for edits)
67
68
## Usage Examples
69
70
### Basic Interrupt for Tool Approval
71
72
```python
73
from langgraph.types import interrupt
74
from langgraph.prebuilt.interrupt import HumanInterrupt, HumanResponse, ActionRequest, HumanInterruptConfig
75
76
def approval_required_tool(state):
77
"""Tool that requires human approval before execution."""
78
# Extract tool call from messages
79
tool_call = state["messages"][-1].tool_calls[0]
80
81
# Create interrupt request
82
request: HumanInterrupt = {
83
"action_request": {
84
"action": tool_call["name"],
85
"args": tool_call["args"]
86
},
87
"config": {
88
"allow_ignore": True, # Allow skipping
89
"allow_respond": True, # Allow feedback
90
"allow_edit": False, # Don't allow editing
91
"allow_accept": True # Allow approval
92
},
93
"description": f"Please review and approve the {tool_call['name']} action"
94
}
95
96
# Send interrupt and get response
97
response = interrupt([request])[0]
98
99
if response["type"] == "accept":
100
# Execute the tool
101
return execute_tool(tool_call)
102
elif response["type"] == "ignore":
103
# Skip the tool execution
104
return {"messages": [ToolMessage(content="Action skipped by user", tool_call_id=tool_call["id"])]}
105
elif response["type"] == "response":
106
# Handle user feedback
107
feedback = response["args"]
108
return {"messages": [ToolMessage(content=f"User feedback: {feedback}", tool_call_id=tool_call["id"])]}
109
```
110
111
### Email Approval Workflow
112
113
```python
114
def email_approval_node(state):
115
"""Node that requests approval before sending emails."""
116
# Extract email details from state
117
email_data = state.get("pending_email", {})
118
119
if not email_data:
120
return state
121
122
# Create interrupt for email approval
123
request: HumanInterrupt = {
124
"action_request": {
125
"action": "send_email",
126
"args": {
127
"to": email_data["to"],
128
"subject": email_data["subject"],
129
"body": email_data["body"]
130
}
131
},
132
"config": {
133
"allow_ignore": True,
134
"allow_respond": True,
135
"allow_edit": True,
136
"allow_accept": True
137
},
138
"description": f"Review email to {email_data['to']} with subject: {email_data['subject']}"
139
}
140
141
# Get human response
142
response = interrupt([request])[0]
143
144
if response["type"] == "accept":
145
# Send the email as approved
146
send_email(email_data)
147
return {"email_status": "sent", "pending_email": None}
148
149
elif response["type"] == "edit":
150
# Update email with human edits
151
edited_request = response["args"]
152
updated_email = {
153
"to": edited_request["args"]["to"],
154
"subject": edited_request["args"]["subject"],
155
"body": edited_request["args"]["body"]
156
}
157
send_email(updated_email)
158
return {"email_status": "sent_with_edits", "pending_email": None}
159
160
elif response["type"] == "response":
161
# Handle feedback without sending
162
feedback = response["args"]
163
return {
164
"email_status": "rejected",
165
"rejection_reason": feedback,
166
"pending_email": None
167
}
168
169
elif response["type"] == "ignore":
170
# Skip email sending
171
return {"email_status": "skipped", "pending_email": None}
172
```
173
174
### Content Moderation Workflow
175
176
```python
177
def content_moderation_node(state):
178
"""Node for human review of generated content."""
179
generated_content = state.get("generated_content", "")
180
181
if not generated_content:
182
return state
183
184
# Create moderation request
185
request: HumanInterrupt = {
186
"action_request": {
187
"action": "review_content",
188
"args": {"content": generated_content}
189
},
190
"config": {
191
"allow_ignore": False, # Must review
192
"allow_respond": True, # Can provide feedback
193
"allow_edit": True, # Can edit content
194
"allow_accept": True # Can approve
195
},
196
"description": "Please review the generated content for appropriateness and accuracy"
197
}
198
199
response = interrupt([request])[0]
200
201
if response["type"] == "accept":
202
return {"content_status": "approved", "final_content": generated_content}
203
204
elif response["type"] == "edit":
205
edited_content = response["args"]["args"]["content"]
206
return {"content_status": "edited", "final_content": edited_content}
207
208
elif response["type"] == "response":
209
feedback = response["args"]
210
# Could regenerate content based on feedback
211
return {
212
"content_status": "needs_revision",
213
"feedback": feedback,
214
"final_content": None
215
}
216
```
217
218
### Multi-Step Approval Process
219
220
```python
221
def multi_step_approval_workflow(state):
222
"""Workflow with multiple approval steps."""
223
steps = [
224
{
225
"action": "data_collection",
226
"description": "Approve data collection from external APIs",
227
"config": {"allow_ignore": True, "allow_respond": True, "allow_edit": False, "allow_accept": True}
228
},
229
{
230
"action": "data_processing",
231
"description": "Review data processing parameters",
232
"config": {"allow_ignore": False, "allow_respond": True, "allow_edit": True, "allow_accept": True}
233
},
234
{
235
"action": "result_publication",
236
"description": "Approve publication of results",
237
"config": {"allow_ignore": True, "allow_respond": True, "allow_edit": False, "allow_accept": True}
238
}
239
]
240
241
results = []
242
243
for step in steps:
244
request: HumanInterrupt = {
245
"action_request": {
246
"action": step["action"],
247
"args": state.get(f"{step['action']}_data", {})
248
},
249
"config": step["config"],
250
"description": step["description"]
251
}
252
253
response = interrupt([request])[0]
254
results.append({
255
"step": step["action"],
256
"response_type": response["type"],
257
"response_data": response["args"]
258
})
259
260
# Stop if any critical step is rejected
261
if response["type"] == "ignore" and step["action"] in ["data_processing"]:
262
return {"workflow_status": "aborted", "approval_results": results}
263
264
return {"workflow_status": "completed", "approval_results": results}
265
```
266
267
## Advanced Patterns
268
269
### Conditional Interrupts
270
271
```python
272
def conditional_interrupt_node(state):
273
"""Only interrupt for certain conditions."""
274
action = state.get("pending_action", {})
275
user_role = state.get("user_role", "user")
276
277
# Only require approval for sensitive actions or non-admin users
278
requires_approval = (
279
action.get("type") in ["delete", "modify_permissions"] or
280
user_role != "admin"
281
)
282
283
if not requires_approval:
284
# Execute without approval
285
return execute_action(action)
286
287
# Request approval
288
request: HumanInterrupt = {
289
"action_request": {
290
"action": action["type"],
291
"args": action["details"]
292
},
293
"config": {
294
"allow_ignore": user_role == "admin",
295
"allow_respond": True,
296
"allow_edit": user_role == "admin",
297
"allow_accept": True
298
},
299
"description": f"Approval required for {action['type']} action by {user_role}"
300
}
301
302
response = interrupt([request])[0]
303
return handle_approval_response(response, action)
304
```
305
306
### Timeout and Escalation
307
308
```python
309
import time
310
from datetime import datetime, timedelta
311
312
def interrupt_with_escalation(state):
313
"""Interrupt with escalation if no response within timeout."""
314
request: HumanInterrupt = {
315
"action_request": {
316
"action": "urgent_approval",
317
"args": state["urgent_data"]
318
},
319
"config": {
320
"allow_ignore": False,
321
"allow_respond": True,
322
"allow_edit": False,
323
"allow_accept": True
324
},
325
"description": "URGENT: Immediate approval required for critical action"
326
}
327
328
# Set timeout for response
329
start_time = datetime.now()
330
timeout_minutes = 5
331
332
# First interrupt attempt
333
response = interrupt([request])[0]
334
335
# Check if this is a timeout scenario (implementation-dependent)
336
if datetime.now() - start_time > timedelta(minutes=timeout_minutes):
337
# Escalate to manager
338
escalation_request: HumanInterrupt = {
339
"action_request": request["action_request"],
340
"config": {
341
"allow_ignore": False,
342
"allow_respond": True,
343
"allow_edit": True,
344
"allow_accept": True
345
},
346
"description": f"ESCALATED: No response received within {timeout_minutes} minutes. Manager approval required."
347
}
348
349
response = interrupt([escalation_request])[0]
350
351
return handle_escalation_response(response)
352
```
353
354
## Integration with LangGraph
355
356
### State Schema Integration
357
358
```python
359
from typing_extensions import TypedDict
360
from typing import List, Optional
361
362
class AgentStateWithApproval(TypedDict):
363
messages: List[BaseMessage]
364
pending_approvals: List[HumanInterrupt]
365
approval_responses: List[HumanResponse]
366
workflow_status: str
367
368
def approval_aware_agent():
369
"""Agent that manages approval workflows."""
370
graph = StateGraph(AgentStateWithApproval)
371
372
graph.add_node("agent", agent_node)
373
graph.add_node("request_approval", approval_request_node)
374
graph.add_node("process_approval", process_approval_response)
375
graph.add_node("tools", tool_execution_node)
376
377
# Route to approval if needed
378
def should_request_approval(state):
379
if requires_human_approval(state):
380
return "request_approval"
381
return "tools"
382
383
graph.add_conditional_edges("agent", should_request_approval)
384
graph.add_edge("request_approval", "process_approval")
385
graph.add_conditional_edges("process_approval", route_after_approval)
386
387
return graph.compile()
388
```
389
390
### Error Handling
391
392
```python
393
def robust_interrupt_handler(state):
394
"""Handle interrupts with error recovery."""
395
try:
396
request: HumanInterrupt = create_interrupt_request(state)
397
response = interrupt([request])[0]
398
return process_human_response(response, state)
399
400
except Exception as e:
401
# Fallback if interrupt system fails
402
logging.error(f"Interrupt system failed: {e}")
403
404
# Default to safe action or escalate
405
return {
406
"interrupt_error": str(e),
407
"fallback_action": "escalate_to_admin",
408
"original_request": state.get("pending_action")
409
}
410
```
411
412
## Best Practices
413
414
### Request Design
415
416
```python
417
# Good: Clear, specific action descriptions
418
request: HumanInterrupt = {
419
"action_request": {
420
"action": "send_customer_email", # Specific action name
421
"args": {
422
"recipient": "customer@example.com",
423
"template": "order_confirmation",
424
"order_id": "12345"
425
}
426
},
427
"config": {
428
"allow_ignore": False, # Critical action
429
"allow_respond": True, # Allow feedback
430
"allow_edit": True, # Allow template edits
431
"allow_accept": True
432
},
433
"description": "Send order confirmation email to customer for order #12345. Review template content and recipient before sending."
434
}
435
```
436
437
### Response Handling
438
439
```python
440
def comprehensive_response_handler(response: HumanResponse, context: dict):
441
"""Handle all possible response types comprehensively."""
442
response_type = response["type"]
443
args = response["args"]
444
445
if response_type == "accept":
446
return execute_approved_action(context)
447
448
elif response_type == "ignore":
449
return log_skipped_action(context, "User chose to skip")
450
451
elif response_type == "response":
452
feedback = args
453
return process_feedback(feedback, context)
454
455
elif response_type == "edit":
456
edited_request = args
457
return execute_edited_action(edited_request, context)
458
459
else:
460
raise ValueError(f"Unknown response type: {response_type}")
461
```
462
463
### Configuration Guidelines
464
465
```python
466
# Sensitive actions - require explicit approval
467
sensitive_config = HumanInterruptConfig(
468
allow_ignore=False, # Must make a decision
469
allow_respond=True, # Can explain reasoning
470
allow_edit=False, # No modifications allowed
471
allow_accept=True
472
)
473
474
# Content review - allow editing
475
content_review_config = HumanInterruptConfig(
476
allow_ignore=False,
477
allow_respond=True,
478
allow_edit=True, # Can modify content
479
allow_accept=True
480
)
481
482
# Optional approval - can be skipped
483
optional_approval_config = HumanInterruptConfig(
484
allow_ignore=True, # Can skip if needed
485
allow_respond=True,
486
allow_edit=True,
487
allow_accept=True
488
)
489
```