Chapter 15 · Agentic Workflows & State Machines

Part of Part IV · Production

"Agents that run in a straight line are scripts. Agents that branch, loop, and recover are systems. This chapter is about the difference."

---

Your agent from Chapter 5 hits an error: the web search API is down. It tries once, fails, and gives up. "I'm sorry, I couldn't complete the task."

A robust agent would detect the error, try an alternative search tool, and if that fails too, tell the user: "Search is unavailable right now, but here's what I know from my training data." The difference is the workflow — the agent's ability to branch, retry, and recover.

The agent loop (observe → think → act) is the foundation. But real agents need more: conditional branching, parallel execution, human approval gates, error recovery, and state persistence. This chapter is about building agentic workflows — state machines that make your agents production-ready.

---

Section 1: Beyond the Simple Loop

The simple loop has served you well since Chapter 4:

while not done:
    thought = llm.think(messages)
    action = parse(thought)
    result = execute(action)
    messages.append(result)

But here's what the simple loop can't do:

• Branch based on complex conditions. "If the search returned fewer than 3 results, broaden the query. If it returned more than 50, narrow it."
• Wait for external events. Human approval. A webhook. A scheduled time.
• Execute multiple paths in parallel. Search three sources simultaneously, then synthesize.
• Recover from errors by trying alternatives. "Search API A failed → try Search API B → try cached results → use training data."
• Persist state across sessions. User closes the browser. Comes back tomorrow. The agent remembers where it left off.

The solution is explicit state machines. Instead of a simple loop, you model your agent as a graph of states and transitions.

Simple loop:     ○ → ○ → ○ → ○ → done

State machine:        ┌─────────┐
                      │ planning │
                      └────┬─────┘
                           │
                    ┌──────▼──────┐
              ┌─────│  searching  │─────┐
              │     └──────┬──────┘     │
              │            │            │
              │     ┌──────▼──────┐     │
              │     │  analyzing  │     │
              │     └──────┬──────┘     │
              │            │            │
              │     ┌──────▼──────┐     │
              └─────│   writing   │◄────┘
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │  reviewing  │
                    └──────┬──────┘
                           │
                          done

---

Section 2: State Machines for Agents

A state machine has four components:

• States: The agent's current situation (planning, searching, analyzing, writing, done).
• Transitions: How it moves between states (found enough info → start writing, need more → search again).
• Guards: Conditions on transitions (only transition to "writing" if confidence > 0.7).
• Actions: What happens on transition (on entering "searching", execute the search query).

Here's a simple state machine for a research agent, implemented without any framework:

from enum import Enum
from dataclasses import dataclass, field

class ResearchState(Enum):
    PLANNING = "planning"
    SEARCHING = "searching"
    ANALYZING = "analyzing"
    WRITING = "writing"
    REVIEWING = "reviewing"
    DONE = "done"

@dataclass
class ResearchAgent:
    state: ResearchState = ResearchState.PLANNING
    query: str = ""
    plan: list[str] = field(default_factory=list)
    search_results: list[str] = field(default_factory=list)
    analysis: str = ""
    draft: str = ""
    final_report: str = ""

    def run(self, query: str) -> str:
        self.query = query
        while self.state != ResearchState.DONE:
            self._step()
        return self.final_report

    def _step(self):
        if self.state == ResearchState.PLANNING:
            self._plan()
        elif self.state == ResearchState.SEARCHING:
            self._search()
        elif self.state == ResearchState.ANALYZING:
            self._analyze()
        elif self.state == ResearchState.WRITING:
            self._write()
        elif self.state == ResearchState.REVIEWING:
            self._review()

    def _plan(self):
        self.plan = llm.generate_plan(self.query)
        self.state = ResearchState.SEARCHING

    def _search(self):
        for step in self.plan:
            results = search_web(step)
            self.search_results.extend(results)

        if len(self.search_results) < 3:
            # Not enough info — broaden search
            self.plan = [f"broader: {s}" for s in self.plan]
            self.state = ResearchState.SEARCHING  # Stay in search
        else:
            self.state = ResearchState.ANALYZING

    def _analyze(self):
        self.analysis = llm.analyze(self.search_results)
        self.state = ResearchState.WRITING

    def _write(self):
        self.draft = llm.write_report(self.query, self.analysis)
        self.state = ResearchState.REVIEWING

    def _review(self):
        review = llm.review(self.draft)
        if review["needs_revision"]:
            self.draft = llm.revise(self.draft, review["feedback"])
            self.state = ResearchState.WRITING  # Go back
        else:
            self.final_report = self.draft
            self.state = ResearchState.DONE

The state machine advantage: Explicit, testable, debuggable. You can look at the state machine and understand the agent's behavior. You can test each state independently. You can add new states without touching existing ones.

---

Section 3: LangGraph — State Machines as Code

LangGraph (introduced in Chapter 10) is the best tool for building agent state machines. It gives you:

• Typed state with reducer functions
• Conditional edges based on state
• Checkpointing — automatic state persistence after each node
• Threads — multiple concurrent conversations, each with its own state
• Streaming — real-time output as the graph executes

Here's the same research agent in LangGraph:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from operator import add

class ResearchState(TypedDict):
    query: str
    plan: list[str]
    search_results: Annotated[list[str], add]  # Reducer: append
    analysis: str
    draft: str
    final_report: str
    revision_count: int

def plan_node(state: ResearchState) -> ResearchState:
    plan = llm.generate_plan(state["query"])
    return {"plan": plan}

def search_node(state: ResearchState) -> ResearchState:
    results = []
    for step in state["plan"]:
        results.extend(search_web(step))
    return {"search_results": results}

def analyze_node(state: ResearchState) -> ResearchState:
    analysis = llm.analyze(state["search_results"])
    return {"analysis": analysis}

def write_node(state: ResearchState) -> ResearchState:
    draft = llm.write_report(state["query"], state["analysis"])
    return {"draft": draft}

def review_node(state: ResearchState) -> ResearchState:
    review = llm.review(state["draft"])
    if review["needs_revision"] and state.get("revision_count", 0) < 3:
        revised = llm.revise(state["draft"], review["feedback"])
        return {"draft": revised, "revision_count": state.get("revision_count", 0) + 1}
    return {"final_report": state["draft"]}

# Routing functions
def should_continue_search(state: ResearchState) -> str:
    if len(state["search_results"]) < 3:
        return "search"  # Need more results
    return "analyze"

def should_continue_review(state: ResearchState) -> str:
    if state.get("final_report"):
        return "done"
    return "write"  # Revise and try again

# Build the graph
graph = StateGraph(ResearchState)

graph.add_node("plan", plan_node)
graph.add_node("search", search_node)
graph.add_node("analyze", analyze_node)
graph.add_node("write", write_node)
graph.add_node("review", review_node)

graph.set_entry_point("plan")
graph.add_edge("plan", "search")
graph.add_conditional_edges("search", should_continue_search, {
    "search": "search",
    "analyze": "analyze",
})
graph.add_edge("analyze", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", should_continue_review, {
    "write": "write",
    "done": END,
})

app = graph.compile()

The graph visualization that LangGraph can generate:

     ┌────────┐
     │  plan  │
     └───┬────┘
         │
     ┌───▼────┐
     │ search │◄────┐
     └───┬────┘     │ (not enough results)
         │          │
     ┌───▼────┐     │
     │analyze │     │
     └───┬────┘     │
         │          │
     ┌───▼────┐     │
     │ write  │◄────┤
     └───┬────┘     │
         │          │
     ┌───▼────┐     │
     │ review ├─────┘ (needs revision)
     └───┬────┘
         │ (approved)
        END

---

Section 4: Human-in-the-Loop

The most important workflow pattern for production agents: the agent pauses and waits for human input before proceeding.

When to use it:

• Before destructive actions (delete, send to all users, modify production data)
• When confidence is low (agent is unsure about a decision)
• For compliance (certain actions require human approval by policy)
• During development (inspect agent state before continuing)

LangGraph's interrupt mechanism makes this straightforward:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

def send_email_node(state: AgentState) -> AgentState:
    """Draft an email, but require human approval before sending."""
    draft = llm.draft_email(
        to=state["recipient"],
        subject=state["subject"],
        context=state["context"],
    )

    # Pause here — human must approve
    approval = interrupt({
        "message": "Agent wants to send this email. Approve?",
        "draft": draft,
        "recipient": state["recipient"],
    })

    if approval.get("approved"):
        send_email(draft)
        return {"email_sent": True, "email_draft": draft}
    else:
        return {"email_sent": False, "email_draft": draft}

# Compile with checkpointing
app = graph.compile(checkpointer=MemorySaver())

# Run until interrupt
config = {"configurable": {"thread_id": "user-123"}}
for event in app.stream(initial_state, config):
    print(event)

# Human reviews and approves
app.update_state(config, {"approved": True})

# Resume execution
for event in app.stream(None, config):
    print(event)

The approval doesn't need to be a CLI prompt. It can be a Slack message, an email, a dashboard button — anything that can set approved: True in the state.

---

Section 5: Error Recovery Patterns

Agents fail. Workflows handle failure gracefully. Here are the patterns:

Retry with Backoff. On transient errors (API timeout, rate limit), retry with increasing delays.

import asyncio

async def search_with_retry(query: str, max_retries: int = 3) -> list[str]:
    for attempt in range(max_retries):
        try:
            return await search_web(query)
        except RateLimitError:
            wait = 2 ** attempt  # 1s, 2s, 4s
            print(f"Rate limited. Retrying in {wait}s...")
            await asyncio.sleep(wait)
        except TimeoutError:
            wait = 2 ** attempt
            print(f"Timeout. Retrying in {wait}s...")
            await asyncio.sleep(wait)
    raise SearchFailedError(f"Search failed after {max_retries} attempts")

Fallback. On persistent errors, try an alternative approach.

async def search_with_fallback(query: str) -> list[str]:
    try:
        return await search_brave(query)
    except SearchError:
        try:
            return await search_tavily(query)
        except SearchError:
            try:
                return await get_cached_results(query)
            except CacheError:
                return []  # Graceful degradation

Circuit Breaker. If a tool fails repeatedly, stop trying for a period.

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = None
        self.open = False

    def call(self, func, *args, **kwargs):
        if self.open:
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.reset_timeout):
                self.open = False  # Try again
                self.failures = 0
            else:
                raise CircuitBreakerOpenError("Circuit breaker is open")

        try:
            result = func(*args, **kwargs)
            self.failures = 0  # Reset on success
            return result
        except Exception:
            self.failures += 1
            self.last_failure_time = datetime.now()
            if self.failures >= self.failure_threshold:
                self.open = True
            raise

Graceful Degradation. If a non-critical step fails, continue with reduced functionality.

def analyze_document(document: str) -> dict:
    result = {"text_analysis": analyze_text(document)}

    try:
        result["image_analysis"] = analyze_images(document)
    except ImageAnalysisError:
        result["image_analysis"] = "Image analysis unavailable"

    try:
        result["table_analysis"] = extract_tables(document)
    except TableExtractionError:
        result["table_analysis"] = "Table extraction unavailable"

    return result  # Always returns something useful

---

Section 6: Parallel Execution

Some tasks can run in parallel. LangGraph's Send API handles this:

from langgraph.types import Send

def continue_to_searches(state: ResearchState):
    """Fan out: one search per query."""
    return [
        Send("search", {"query": q})
        for q in state["search_queries"]
    ]

def search_node(state: dict):
    results = search_web(state["query"])
    return {"search_results": [{"query": state["query"], "results": results}]}

graph.add_conditional_edges("plan", continue_to_searches, ["search"])
graph.add_edge("search", "synthesize")

The map-reduce pattern for agents:

                    ┌──────────┐
                    │   plan   │
                    └────┬─────┘
                         │
              ┌──────────┼──────────┐
              │          │          │
         ┌────▼───┐ ┌───▼────┐ ┌───▼────┐
         │search 1│ │search 2│ │search 3│   ← parallel
         └────┬───┘ └───┬────┘ └───┬────┘
              │          │          │
              └──────────┼──────────┘
                         │
                    ┌────▼─────┐
                    │synthesize│              ← reduce
                    └──────────┘

The parallelism tradeoff: Faster but more expensive. Three concurrent searches complete in the time of one, but cost 3× as much. Use when latency matters more than cost.

---

Section 7: The Complete Production Workflow

Here's a complete production agent workflow combining everything from this chapter:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Send
from typing import TypedDict, Annotated
from operator import add

class ProductionResearchState(TypedDict):
    query: str
    plan: list[str]
    search_queries: list[str]
    search_results: Annotated[list[dict], add]
    analysis: str
    draft: str
    review_feedback: str
    final_report: str
    errors: Annotated[list[str], add]

def plan_node(state):
    plan = llm.generate_plan(state["query"])
    # Human approves the plan before we spend money on searches
    approval = interrupt({"stage": "plan_approval", "plan": plan})
    if not approval.get("approved"):
        return {"final_report": "Plan was not approved."}
    search_queries = [step["search_query"] for step in plan]
    return {"plan": plan, "search_queries": search_queries}

def search_node(state):
    try:
        results = search_with_fallback(state["query"])
        return {"search_results": [{"query": state["query"], "results": results}]}
    except SearchFailedError as e:
        return {"errors": [str(e)], "search_results": []}

def synthesize_node(state):
    all_results = [r for sr in state["search_results"] for r in sr["results"]]
    analysis = llm.analyze(all_results)
    return {"analysis": analysis}

def write_node(state):
    draft = llm.write_report(state["query"], state["analysis"])
    return {"draft": draft}

def review_node(state):
    review = llm.review(state["draft"])
    if review["score"] < 7:
        return {"review_feedback": review["feedback"]}
    # Final human approval
    approval = interrupt({"stage": "final_approval", "draft": state["draft"]})
    if approval.get("approved"):
        return {"final_report": state["draft"]}
    return {"review_feedback": approval.get("feedback", "Not approved.")}

def revise_node(state):
    revised = llm.revise(state["draft"], state["review_feedback"])
    return {"draft": revised}

# Routing
def route_after_search(state):
    if not state["search_results"]:
        return END  # All searches failed
    return "synthesize"

def route_after_review(state):
    if state.get("final_report"):
        return END
    if state.get("review_feedback"):
        return "revise"
    return END

# Build graph
graph = StateGraph(ProductionResearchState)
graph.add_node("plan", plan_node)
graph.add_node("search", search_node)
graph.add_node("synthesize", synthesize_node)
graph.add_node("write", write_node)
graph.add_node("review", review_node)
graph.add_node("revise", revise_node)

graph.set_entry_point("plan")
graph.add_conditional_edges("plan", lambda s: [Send("search", {"query": q}) for q in s["search_queries"]], ["search"])
graph.add_conditional_edges("search", route_after_search, {"synthesize": "synthesize", END: END})
graph.add_edge("synthesize", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", route_after_review, {"revise": "revise", END: END})
graph.add_edge("revise", "review")

app = graph.compile(checkpointer=MemorySaver())

This workflow includes: planning with human approval, parallel search with fallbacks, synthesis, writing, automated review with revision loops, and final human approval. It handles errors gracefully and never executes expensive operations without confirmation.

---

The Turn

You now understand that agentic workflows are what separate demos from products. State machines, human-in-the-loop, error recovery, parallel execution — these patterns make agents robust enough for the real world.

The simple loop from Chapter 4 was the foundation. Everything since has been adding layers: tools, memory, reasoning, security, evaluation. The workflow ties them all together into a system that can handle the messiness of reality.

---

In the next chapter: Your agents are now robust, recoverable, and production-ready. But they still operate within the boundaries you set. What if an agent could write its own tools? What if it could generate code, test it, and execute it — all while keeping your system safe? You'll build agents that write and run code.