State Management, Workflows & Orchestration

"State is the silent partner in every agent conversation. Lose track of it and everything unravels."

Census, Statefully Orchestrated AI Agent

1. State Machines for Agent Workflows

Complex agent workflows require explicit state management. A customer onboarding workflow progresses through states: application received, identity verified, credit checked, compliance approved, account created, welcome email sent. At each state, different agents are responsible, different tools are available, and different failure modes apply. Modeling this as a state machine makes the workflow predictable, debuggable, and resumable.

LangGraph provides the most popular state machine implementation for agent workflows. State is represented as a TypedDict that flows between nodes (functions). Edges define transitions, and conditional edges enable branching based on state values. The state is checkpointed after each node execution, enabling the workflow to be paused and resumed at any point. This is critical for workflows that span hours or days, where the system must survive restarts and infrastructure changes.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Optional, Literal

class OnboardingState(TypedDict):
 customer_id: str
 application: dict
 identity_verified: Optional[bool]
 credit_score: Optional[int]
 compliance_status: Optional[str]
 account_id: Optional[str]
 status: str

def verify_identity(state: OnboardingState) -> dict:
 result = identity_agent.verify(state["application"])
 return {"identity_verified": result.verified, "status": "identity_checked"}

def check_credit(state: OnboardingState) -> dict:
 score = credit_agent.check(state["customer_id"])
 return {"credit_score": score, "status": "credit_checked"}

def compliance_review(state: OnboardingState) -> dict:
 result = compliance_agent.review(state)
 return {"compliance_status": result.status, "status": "compliance_reviewed"}

def route_after_compliance(state: OnboardingState) -> str:
 if state["compliance_status"] == "approved":
 return "create_account"
 elif state["compliance_status"] == "needs_review":
 return "human_review"
 return "reject"

# Build workflow with checkpointing
checkpointer = SqliteSaver.from_conn_string("onboarding.db")
graph = StateGraph(OnboardingState)
graph.add_node("verify_identity", verify_identity)
graph.add_node("check_credit", check_credit)
graph.add_node("compliance", compliance_review)
graph.add_node("create_account", create_account)
graph.add_node("human_review", request_human_review)
graph.add_node("reject", send_rejection)

graph.set_entry_point("verify_identity")
graph.add_edge("verify_identity", "check_credit")
graph.add_edge("check_credit", "compliance")
graph.add_conditional_edges("compliance", route_after_compliance)
graph.add_edge("create_account", END)
graph.add_edge("human_review", "compliance") # Re-check after human review
graph.add_edge("reject", END)

workflow = graph.compile(checkpointer=checkpointer)

2. Durable Execution with Temporal

For workflows that span extended time periods, involve external service calls that may take minutes to complete, or need guaranteed execution despite infrastructure failures, durable execution frameworks like Temporal provide stronger reliability guarantees than in-process state machines. Temporal persists the complete execution history, automatically retries failed activities, and can resume workflows after server restarts without losing progress.

The integration pattern is straightforward: each agent step becomes a Temporal activity, the workflow definition describes the orchestration logic, and Temporal handles the infrastructure concerns (persistence, retries, timeouts, visibility). This separates the agent logic (what to do) from the infrastructure logic (how to do it reliably), making the system easier to reason about and maintain.

3. Parallel Execution and Fan-Out

Many agent workflows contain steps that can execute in parallel. When a supervisor delegates independent subtasks to multiple specialists, those subtasks can run concurrently rather than sequentially. LangGraph supports this through the Send API, which fans out execution to multiple node instances and collects results. This can reduce wall-clock time dramatically for tasks with parallelizable subtasks.

The challenge with parallel execution is error handling. If three of five parallel subtasks succeed but two fail, what should the system do? Continue with partial results? Retry the failed tasks? Fail the entire workflow? The answer depends on the task: a research workflow can proceed with partial results, while a compliance workflow must have all checks pass. Design your fan-out patterns with explicit handling for partial failure scenarios.

Exercises