Introduction
AI agent workflow automation moves beyond single LLM calls to multi-step processes where agents call tools, wait for human approval, retry on failure, and coordinate across services. The key technical challenge is not the AI model itself but the orchestration layer — ensuring the workflow survives failures, respects timeouts, and maintains state across long-running operations.
This guide covers two orchestration approaches: LangGraph for graph-based agent workflows with conditional branching and loops, and Temporal for durable execution of long-running agent processes that must survive service restarts. Both include Python code examples with human-in-the-loop approval gates, tool integration, and error handling.
LangGraph: Stateful Agent Workflow
LangGraph models agent workflows as directed graphs with typed state. Nodes are agent steps or tool calls. Edges define transitions, including conditional routing based on agent output.
Customer Support Escalation Agent
flowchart LR
A[Customer Query] --> B{Classify Intent}
B -->|Billing| C[Billing Agent]
B -->|Technical| D[Tech Support Agent]
B -->|General| E[General Agent]
C --> F{Can Resolve?}
D --> F
E --> F
F -->|Yes| G[Respond to Customer]
F -->|No| H[Human Escalation]
H --> I[Wait for Human Approval]
I --> G
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
import json
class SupportState(TypedDict):
query: str
intent: str
resolution: str
needs_escalation: bool
human_approved: bool
ticket_id: str
def classify_intent(state: SupportState) -> SupportState:
"""Determine the customer's intent."""
prompt = f"Classify this support query as 'billing', 'technical', or 'general':\n{state['query']}"
state["intent"] = llm.invoke(prompt).strip().lower()
return state
def handle_billing(state: SupportState) -> SupportState:
"""Attempt to resolve billing issues automatically."""
response = llm.invoke(
f"Resolve this billing issue. If you cannot resolve, say ESCALATE:\n{state['query']}"
)
if "ESCALATE" in response:
state["needs_escalation"] = True
else:
state["resolution"] = response
return state
def handle_technical(state: SupportState) -> SupportState:
response = llm.invoke(
f"Resolve this technical issue. If you cannot resolve, say ESCALATE:\n{state['query']}"
)
if "ESCALATE" in response:
state["needs_escalation"] = True
else:
state["resolution"] = response
return state
def handle_general(state: SupportState) -> SupportState:
state["resolution"] = llm.invoke(f"Answer this question:\n{state['query']}")
return state
def route_by_intent(state: SupportState) -> Literal["billing", "technical", "general"]:
return state["intent"]
def should_escalate(state: SupportState) -> Literal["resolve", "escalate"]:
if state.get("needs_escalation"):
return "escalate"
return "resolve"
def human_escalation(state: SupportState) -> SupportState:
"""Create a ticket and wait for human approval.
In production, this would send a notification to a human agent
and pause execution until the human responds.
"""
state["ticket_id"] = create_ticket(state["query"], state["intent"])
# The workflow pauses here until human approval is received
state["human_approved"] = wait_for_human_approval(state["ticket_id"])
if state["human_approved"]:
state["resolution"] = llm.invoke(
f"The human supervisor approved escalation. Draft a response for: {state['query']}"
)
else:
state["resolution"] = "We are still reviewing your issue. A team member will follow up."
return state
# Build the graph
graph = StateGraph(SupportState)
graph.add_node("classify", classify_intent)
graph.add_node("billing", handle_billing)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)
graph.add_node("escalate", human_escalation)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_by_intent, {
"billing": "billing", "technical": "technical", "general": "general"
})
graph.add_conditional_edges("billing", should_escalate, {
"resolve": END, "escalate": "escalate"
})
graph.add_conditional_edges("technical", should_escalate, {
"resolve": END, "escalate": "escalate"
})
graph.add_edge("general", END)
graph.add_edge("escalate", END)
agent = graph.compile()
# Run the workflow
result = agent.invoke({
"query": "I was charged twice for my subscription this month.",
"intent": "",
"resolution": "",
"needs_escalation": False,
"human_approved": False,
"ticket_id": ""
})
print(f"Resolution: {result['resolution']}")
Temporal: Durable Long-Running Agents
For workflows that must survive server restarts — agents that wait hours for human approval, process large batches, or poll external APIs — Temporal provides durable execution. The workflow code runs to completion even if the worker process crashes mid-way.
from temporalio import workflow
from temporalio.activity import Activity
from datetime import timedelta
@workflow.defn
class DocumentProcessingAgent:
"""Long-running agent that processes documents with human review steps.
If the worker crashes while waiting for human approval, Temporal
replays the workflow from the last completed step when the worker restarts.
"""
@workflow.run
async def run(self, document_id: str) -> dict:
# Step 1: Extract text from document (durable activity)
text = await workflow.execute_activity(
extract_text, document_id,
start_to_close_timeout=timedelta(minutes=5)
)
# Step 2: Classify document (retry on failure)
classification = await workflow.execute_activity(
classify_document, text,
start_to_close_timeout=timedelta(minutes=2),
retry_policy={"maximum_attempts": 3}
)
# Step 3: Human review (can take days — Temporal preserves state)
if classification["confidence"] < 0.9:
approval = await workflow.execute_activity(
request_human_review,
{"document_id": document_id, "classification": classification},
start_to_close_timeout=timedelta(days=7)
)
if not approval["approved"]:
return {"status": "rejected", "reason": approval["reason"]}
# Step 4: Generate final output
result = await workflow.execute_activity(
generate_report, {"text": text, "classification": classification},
start_to_close_timeout=timedelta(hours=1)
)
return {"status": "completed", "result": result}
Human-in-the-Loop Patterns
Approval Gate (LangGraph)
def wait_for_human_approval(ticket_id: str, timeout_hours: int = 24) -> bool:
"""Poll for human approval decision.
In production, this would listen on a webhook or message queue.
The function blocks until a decision is received or the timeout expires.
"""
deadline = datetime.now() + timedelta(hours=timeout_hours)
while datetime.now() < deadline:
decision = check_approval_status(ticket_id)
if decision is not None:
return decision
time.sleep(30) # Poll every 30 seconds
return False # Timeout — default to denying the request
Approval Notification (Slack Integration)
def notify_human_for_approval(ticket_id: str, query: str, intent: str):
"""Send an approval request to a Slack channel."""
import requests
requests.post(SLACK_WEBHOOK_URL, json={
"channel": "#support-escalations",
"blocks": [
{
"type": "section",
"text": {"text": f"*Escalation Request*\nTicket: {ticket_id}\nQuery: {query}\nIntent: {intent}"}
},
{
"type": "actions",
"elements": [
{"type": "button", "text": "Approve", "value": f"approve:{ticket_id}", "style": "primary"},
{"type": "button", "text": "Deny", "value": f"deny:{ticket_id}", "style": "danger"}
]
}
]
})
Orchestration Approach Decision
| Factor | LangGraph | Temporal |
|---|---|---|
| Execution model | In-process graph traversal | Durable (survives crashes) |
| State persistence | Manual (checkpointing) | Automatic (event history) |
| Long-running tasks | Limited (in-memory) | Native (hours/days) |
| Human-in-loop | Polling + webhooks | Durable signals + queries |
| Best for | Short-lived agent interactions | Mission-critical, days-long workflows |
Resources
- LangGraph Documentation — State graphs, checkpointing, human-in-loop
- Temporal Python SDK — Durable execution for workflows
- LangGraph Human-in-the-Loop — Approval patterns
- Temporal Long-Running Activities — Async completion pattern
Comments