intermediate⏱35 minutesLesson 6 of 10
Approval Workflows
Implement structured approval workflows with conditional interrupts, before/after node interrupts, and dynamic human-in-the-loop routing.
Approval Workflows
Approval workflows combine interrupts with conditional routing to create structured human-in-the-loop processes. This lesson covers before-node interrupts, after-node interrupts, and dynamic approval patterns.
Approval Pattern Overview
Agent proposes action → Human reviews → Approve? → Yes → Execute
↓ No
Reject → Provide feedback → Revise or Stop
The agent presents a proposal, the human decides, and execution follows accordingly.
Before-Node Interrupt
Interrupt before a node executes — the human decides whether the node should run:
python
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict
class ApprovalState(TypedDict):
query: str
search_plan: str
approved: bool
feedback: str
result: str
def plan_node(state: ApprovalState) -> dict:
plan = f"Search for: {state['query']}"
return {"search_plan": plan}
def approve_search(state: ApprovalState) -> dict:
# BEFORE interrupt — human decides if search should happen
response = interrupt({
"type": "approval",
"agent_plan": state["search_plan"],
"prompt": "Approve this search operation?"
})
return {"approved": response.get("approved", False),
"feedback": response.get("feedback")}
def execute_search(state: ApprovalState) -> dict:
if not state["approved"]:
return {"result": f"Rejected. Feedback: {state.get('feedback')}"}
return {"result": f"Executed: {state['search_plan']}"}
def after_search_review(state: ApprovalState) -> dict:
# AFTER interrupt — human reviews results
response = interrupt({
"type": "review",
"result": state["result"],
"prompt": "Are these results acceptable?"
})
return {"approved": response.get("approved", False)}
builder = StateGraph(ApprovalState)
builder.add_node("plan", plan_node)
builder.add_node("approve_search", approve_search)
builder.add_node("execute", execute_search)
builder.add_edge(START, "plan")
builder.add_edge("plan", "approve_search")
builder.add_edge("approve_search", "execute")
builder.add_edge("execute", END)
app = builder.compile(checkpointer=MemorySaver())Note
A before-node interrupt prevents an action from happening until approved. This is the most common approval pattern — it prevents unauthorized or risky actions.
After-Node Interrupt
Interrupt after a node executes — the human reviews the result:
python
def draft_email(state: EmailState) -> dict:
email = llm.invoke(f"Draft email about: {state['topic']}")
return {"draft": email.content}
def review_draft(state: EmailState) -> dict:
# AFTER interrupt — human reviews the draft
response = interrupt({
"type": "review",
"draft": state["draft"],
"prompt": "Edit or approve this email draft?"
})
return {
"approved": response.get("approved", False),
"revision": response.get("revision", "")
}
def finalize_or_revise(state: EmailState) -> dict:
if state["approved"]:
return {"final": state["draft"]}
return {"draft": state["revision"], "approved": False}Tip
After-node interrupts are useful for content review, result validation, and any workflow where the human should verify output before it's used.
Conditional Approval Routing
Route based on whether the human approved or rejected:
python
def approval_router(state: ApprovalState) -> str:
if state["approved"]:
return "approved"
return "rejected"
builder.add_conditional_edges(
"human_review",
approval_router,
{
"approved": "execute",
"rejected": "revise_plan" # Send back for revision
}
)Revision Loop Pattern
python
class RevisionState(TypedDict):
draft: str
approved: bool
revision_count: int
def generate_draft(state: RevisionState) -> dict:
draft = llm.invoke(f"Write content about: {state['topic']}")
return {"draft": draft.content}
def human_review(state: RevisionState) -> dict:
response = interrupt({
"draft": state["draft"],
"revision_count": state["revision_count"],
"prompt": "Approve, or provide revision instructions."
})
return {
"approved": response.get("approved", False),
"feedback": response.get("feedback", ""),
"revision_count": state["revision_count"] + 1
}
def revise_draft(state: RevisionState) -> dict:
revised = llm.invoke(
f"Revise this draft based on feedback:\n"
f"Draft: {state['draft']}\nFeedback: {state['feedback']}"
)
return {"draft": revised.content}
def revision_router(state: RevisionState) -> str:
if state["approved"]:
return "finalize"
if state["revision_count"] >= 5:
return "max_revisions"
return "revise"
builder.add_conditional_edges(
"human_review",
revision_router,
{
"finalize": "finalize",
"revise": "revise_draft",
"max_revisions": "error_handler"
}
)
builder.add_edge("revise_draft", "human_review") # Loop back for another reviewWarning
Always cap revision loops. Without a maximum, an unsatisfied reviewer could loop forever.
Escalation Workflow
Route to a senior reviewer if junior rejects or if the request is high-risk:
python
class EscalationState(TypedDict):
request: str
risk_level: str
junior_approved: bool
senior_approved: bool
def junior_review(state: EscalationState) -> dict:
response = interrupt({"request": state["request"], "risk": state["risk_level"]})
return {"junior_approved": response.get("approved", False)}
def escalate(state: EscalationState) -> dict:
response = interrupt({"request": state["request"],
"risk": state["risk_level"],
"note": "Escalated from junior review"})
return {"senior_approved": response.get("approved", False)}
def escalation_router(state: EscalationState) -> str:
if state["junior_approved"]:
return "approved"
if state["risk_level"] == "high":
return "escalate"
return "rejected"
builder.add_conditional_edges(
"junior_review",
escalation_router,
{
"approved": "execute",
"escalate": "senior_review",
"rejected": "reject_node"
}
)Approval with Dynamic Data
Include relevant data in the approval request:
python
def approval_with_context(state: State) -> dict:
# Gather context for the human reviewer
context = {
"request_id": state["request_id"],
"requester": state["user_name"],
"action": state["proposed_action"],
"cost_estimate": state.get("estimated_cost", "Unknown"),
"similar_approved_count": state.get("similar_count", 0),
"policy_reference": "Policy #42: External API calls require approval",
"timestamp": datetime.now().isoformat()
}
response = interrupt(context)
return {"approval_response": response}Timeout for Approvals
Handle the case where the human never responds:
python
def approval_with_timeout(state: State) -> dict:
# Set a timeout via configuration
config = state.get("config", {})
timeout = config.get("approval_timeout_minutes", 60)
response = interrupt({
"prompt": "Approve within {timeout} minutes or it will be auto-rejected.",
"request": state["request"]
})
# If human doesn't respond within timeout,
# calling invoke again with a timeout command handles it
return {"human_response": response}Note
LangGraph does not have built-in timeout for interrupts. Implement timeout logic at the application layer by checking how long a thread has been paused.
Complete Approval Workflow Example
python
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict
from datetime import datetime
class WorkflowState(TypedDict):
request: str
risk_score: float
draft_response: str
approved: bool
feedback: str
result: str
step: str
def analyze_request(state: WorkflowState) -> dict:
risk = len(state["request"]) * 0.01 # Simplified risk calc
return {"risk_score": min(risk, 1.0), "step": "analyzed"}
def generate_draft(state: WorkflowState) -> dict:
draft = f"Response to: {state['request']}"
return {"draft_response": draft, "step": "drafted"}
def human_approval(state: WorkflowState) -> dict:
response = interrupt({
"request": state["request"],
"draft": state["draft_response"],
"risk": state["risk_score"],
"timestamp": datetime.now().isoformat()
})
return {"approved": response.get("approved", False),
"feedback": response.get("feedback", ""),
"step": "reviewed"}
def execute(state: WorkflowState) -> dict:
if state["approved"]:
return {"result": state["draft_response"], "step": "completed"}
return {"result": f"Rejected: {state.get('feedback')}", "step": "rejected"}
builder = StateGraph(WorkflowState)
builder.add_node("analyze", analyze_request)
builder.add_node("draft", generate_draft)
builder.add_node("approve", human_approval)
builder.add_node("execute", execute)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "draft")
builder.add_edge("draft", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)
app = builder.compile(checkpointer=MemorySaver())Practice Questions
Practice Question
What is a before-node interrupt?
Practice Question
What is an after-node interrupt useful for?
Practice Question
How do you create a revision loop in an approval workflow?
Practice Question
Why should revision loops have a maximum count?
Practice Question
What is an escalation workflow?
Practice Question
How do you include context for the human reviewer in an interrupt?
Practice Question
What determines which path the graph takes after an approval interrupt?
Practice Question
What happens when a human rejects a proposal in an approval workflow?
Practice Question
Can an approval workflow have multiple sequential approval steps?
Practice Question
How does the approval workflow pattern enhance LangGraph agents?
Success
Key Takeaways
- Before-node interrupts prevent actions until human approval
- After-node interrupts let humans review results before proceeding
- Conditional edges route based on approval/rejection decisions
- Revision loops cycle until approval or max revisions reached
- Escalation workflows route to senior reviewers when needed
- Pass rich context in interrupt() to help humans make informed decisions
- Always cap revision and escalation loops with maximum counts
- Multiple sequential approvals can be chained in a single graph
Progress60%