State Machines & Hybrid Control Flow
Agent systems tend toward one of two extremes. On one end, you have fully autonomous agents — a ReAct loop where the model decides every step, picks every tool, and determines when to stop. On the other end, you have rigid workflows — prompt chains and DAGs where the developer hardcodes every transition in advance. Most production systems need something in between.
A state machine gives you that middle ground. You define explicit states and transitions that the agent moves through, while the model remains free within each state to reason, call tools, and generate output however it sees fit. The structure is deterministic. The work inside each state remains flexible. This hybrid — explicit control flow wrapping flexible LLM behavior — turns out to be a better fit for most real-world agent tasks than either extreme alone.
Why Pure LLM Routing Breaks Down #
An autonomous agent decides its own control flow. At each step, the model looks at the conversation so far and picks what to do next. This works well for open-ended tasks with paths that emerge during execution — exploratory research, creative brainstorming, general-purpose assistants. Tasks with known structure benefit from stronger control.
The model skips steps. An onboarding workflow needs to collect a user's name, validate their email, check their account status, and then provision access — in that order. A fully autonomous agent might decide to skip validation because the user "seems legitimate," or jump straight to provisioning because it is eager to be helpful. The model prioritizes a helpful response while the workflow requires process compliance.
Transitions are invisible. When the model decides to move from "gathering information" to "taking action," the shift happens inside its reasoning. Application code therefore misses the transition hooks needed to run validation, create an audit log, and enforce policy. The execution trace reveals skipped steps only after the fact.
Error recovery is ad hoc. If step three of a five-step process fails, a ReAct agent might retry, skip it, or start the whole task over — depending on what the model thinks is best. A structured fallback path gives tasks that interact with external systems (payment processing, account creation, data pipelines) predictable recovery behavior.
Pure LLM routing — what can go wrong:
User: "Set up my account"
Model thinks: "I should create the account"
│
├──► Skips email validation (seemed obvious)
├──► Creates account (writes to database)
├──► Tries to send welcome email
│ └── Fails: email was invalid
└──► Now has a broken account with no valid contact
With a state machine, the transition from
"collect info" → "validate" → "provision"
is enforced, not suggested.
State Machines for Agent Control Flow #
A state machine defines a finite set of states, a set of valid transitions between them, and conditions that trigger each transition. Applied to agents, each state corresponds to a phase of work where the model operates freely, and transitions represent the boundaries where control returns to deterministic code.
from dataclasses import dataclass, field
from enum import Enum, auto
from inspect import isawaitable
from typing import Any, Callable
class TransitionResult(Enum):
CONTINUE = auto() # Stay in current state
ADVANCE = auto() # Move to next state
FAIL = auto() # Transition to error state
COMPLETE = auto() # Task is done
@dataclass
class State:
name: str
system_prompt: str
tools: list["ToolDefinition"] = field(default_factory=list)
max_steps: int = 10
on_enter: Callable | None = None
on_exit: Callable | None = None
validate_exit: Callable | None = None
@dataclass
class Transition:
from_state: str
to_state: str
condition: Callable[..., bool] | None = None
class AgentStateMachine:
def __init__(
self,
states: list[State],
transitions: list[Transition],
initial_state: str,
model: "ModelClient",
):
self.states = {s.name: s for s in states}
self.transitions = transitions
self.current_state_name = initial_state
self.model = model
self.context: dict[str, Any] = {}
self.history: list[dict] = []
@property
def current_state(self) -> State:
return self.states[self.current_state_name]
async def run(self, task: str) -> dict[str, Any]:
self.context["task"] = task
self._current_messages = [{"role": "user", "content": task}]
return await self._continue_from_current_state()
async def _continue_from_current_state(self) -> dict[str, Any]:
"""Continue execution using the current state and message history."""
while True:
state = self.current_state
if state.on_enter:
await state.on_enter(self.context)
# Run the LLM freely within this state
result = await self._run_state(state, self._current_messages)
self.history.append({
"state": state.name,
"result": result,
})
# Determine the transition
transition = self._resolve_transition(state.name, result)
if transition is TransitionResult.COMPLETE:
return {"result": result, "history": self.history}
if transition is TransitionResult.FAIL:
return {"error": result, "history": self.history}
if state.on_exit:
await state.on_exit(self.context)
self.current_state_name = transition.to_state
# Carry relevant output forward
self._current_messages.append({
"role": "assistant",
"content": result["output"],
})
async def _run_state(self, state: State, messages: list[dict]) -> dict:
"""Run the LLM loop within a single state."""
state_messages = messages.copy()
steps = 0
tool_calls = []
while steps < state.max_steps:
response = await self.model.chat(
system=state.system_prompt,
messages=state_messages,
tools=state.tools,
)
steps += 1
if response.has_tool_calls:
for call in response.tool_calls:
result = await self._execute_tool(call, state)
tool_calls.append({"tool": call.name, "arguments": call.arguments})
state_messages.append({
"role": "tool",
"content": str(result),
"tool_call_id": call.id,
})
self.context[f"{state.name}_{call.name}_result"] = result
else:
return {
"output": response.text,
"steps": steps,
"tool_calls": tool_calls,
}
return {
"output": "State step limit reached",
"steps": steps,
"tool_calls": tool_calls,
}
def _resolve_transition(
self,
state_name: str,
result: dict,
) -> Transition | TransitionResult:
state = self.states[state_name]
if state.validate_exit and not state.validate_exit(result, self.context):
return TransitionResult.FAIL
for t in self.transitions:
if t.from_state == state_name:
if t.condition is None or t.condition(result, self.context):
return t
return TransitionResult.COMPLETE
async def _execute_tool(self, call, state: State) -> str:
tool = next((t for t in state.tools if t.name == call.name), None)
if tool is None:
return f"Unknown tool: {call.name}"
result = tool.fn(**call.arguments)
return await result if isawaitable(result) else result
The critical design choice: each state gets its own system prompt and its own tool set. The model operating inside the "validate email" state sees only validation tools. Its available affordances keep execution in the current phase until a transition grants access to the provisioning tools.
A Concrete Example: Customer Onboarding #
Here is the onboarding flow from earlier, built as a state machine:
# State definitions
collect_info = State(
name="collect_info",
system_prompt="""You are collecting user information for account setup.
Ask for their full name, email address, and company name.
When you have all three, output them in a structured format.""",
tools=[],
max_steps=5,
validate_exit=lambda result, ctx: all(
field in result["output"].lower()
for field in ["name:", "email:", "company:"]
),
)
validate = State(
name="validate",
system_prompt="""You are validating user information.
Use the validate_email tool to check the email address.
Use the check_company tool to verify the company exists.
Report which validations passed and which failed.""",
tools=[validate_email_tool, check_company_tool],
max_steps=5,
validate_exit=lambda result, ctx: "validation passed" in result["output"].lower(),
)
provision = State(
name="provision",
system_prompt="""You are provisioning a new account.
Use the create_account tool to set up the account.
Use the send_welcome_email tool to notify the user.
Confirm that both steps completed successfully.""",
tools=[create_account_tool, send_welcome_email_tool],
max_steps=5,
)
# Wired together
onboarding = AgentStateMachine(
states=[collect_info, validate, provision],
transitions=[
Transition("collect_info", "validate"),
Transition("validate", "provision"),
],
initial_state="collect_info",
model=model_client,
)
result = await onboarding.run(
"Set up an account for Ada Lovelace, ada@example.com, Analytical Engines Ltd."
)
┌───────────────┐ ┌──────────────┐ ┌──────────────┐
│ collect_info │────────▶│ validate │───────▶│ provision │
│ │ │ │ │ │
│ Prompt: "Ask │ │ Prompt: "Use │ │ Prompt: "Use │
│ for name, │ │ tools to │ │ tools to │
│ email, co." │ │ check email │ │ create acct │
│ │ │ and company"│ │ and send │
│ Tools: none │ │ │ │ welcome" │
│ │ │ Tools: │ │ │
│ Exit check: │ │ - validate │ │ Tools: │
│ has all 3 │ │ - check_co │ │ - create │
│ fields? │ │ │ │ - send_mail │
│ │ │ Exit check: │ │ │
│ │ │ passed? │ │ │
└───────────────┘ └──────────────┘ └──────────────┘
Compare this to a pure ReAct agent with all six tools available from the start. The ReAct agent might follow the right sequence. The state machine guarantees that validation happens before provisioning, that the exit condition is checked programmatically, and that each state only sees the tools it needs.
Conditional and Branching Transitions #
Many workflows branch based on what the model found or what a tool returned.
# A support ticket workflow with conditional routing
classify = State(
name="classify",
system_prompt="""Classify this support ticket into exactly one category:
BILLING, TECHNICAL, or ACCOUNT.
Output only the category name.""",
tools=[],
max_steps=2,
)
billing_handler = State(
name="billing",
system_prompt="You handle billing inquiries. Use the lookup_invoice tool...",
tools=[lookup_invoice_tool, issue_refund_tool],
max_steps=8,
)
technical_handler = State(
name="technical",
system_prompt="You handle technical issues. Use the search_docs tool...",
tools=[search_docs_tool, create_ticket_tool],
max_steps=8,
)
account_handler = State(
name="account",
system_prompt="You handle account issues. Use the lookup_account tool...",
tools=[lookup_account_tool, update_account_tool],
max_steps=8,
)
support_flow = AgentStateMachine(
states=[classify, billing_handler, technical_handler, account_handler],
transitions=[
Transition("classify", "billing",
condition=lambda r, ctx: "BILLING" in r["output"].upper()),
Transition("classify", "technical",
condition=lambda r, ctx: "TECHNICAL" in r["output"].upper()),
Transition("classify", "account",
condition=lambda r, ctx: "ACCOUNT" in r["output"].upper()),
],
initial_state="classify",
model=model_client,
)
The classify state uses the model purely for classification. The transition conditions are evaluated in code. This means you get a structured routing decision that you can log, audit, and test deterministically (mock the model to return "BILLING" and verify it reaches the billing handler).
Hybrid Control Flow #
The cleanest agent architectures use state machines for the overall flow and LLM-driven reasoning within each state. But the interesting design space is the transition logic — who decides when to move between states?
There are three models, each with different trade-offs:
Developer-Controlled Transitions #
The transitions are hardcoded conditions evaluated in application code. The validate_exit function holds authority over state completion.
# The model cannot skip validation
validate = State(
name="validate",
system_prompt="Validate the email and company before proceeding.",
validate_exit=lambda result, ctx: (
ctx.get("validate_validate_email_result") is True
and ctx.get("validate_check_company_result") is True
),
)
When to use: compliance-sensitive flows. Payment processing, identity verification, regulated workflows. The trade-off is rigidity — the fixed path also governs unusual situations where an alternative sequence would be valid.
Model-Controlled Transitions #
The model itself decides when to transition, typically by producing a structured signal like a specific keyword or a tool call.
# Give the model a "transition" tool
advance_tool = ToolDefinition(
name="advance_to_next_step",
description="Call this when the current step is complete and you are ready to proceed",
parameters={
"type": "object",
"properties": {
"summary": {
"type": "string",
"description": "Summary of what was accomplished in this step",
}
},
"required": ["summary"],
},
fn=lambda summary: summary,
)
# The state includes the transition tool alongside work tools
research_state = State(
name="research",
system_prompt="""Research the given topic.
When you have gathered enough information, call advance_to_next_step.""",
tools=[search_tool, fetch_tool, advance_tool],
max_steps=15,
)
When to use: exploratory tasks where the model is better positioned than the developer to judge when enough information has been gathered. Research, analysis, open-ended problem solving. The trade-off is that the model might advance prematurely or stall indefinitely.
Hybrid Transitions #
Combine both: the model signals readiness, but application code validates the signal before allowing the transition.
def hybrid_transition(state_name: str, result: dict, ctx: dict) -> TransitionResult:
"""
Model signals completion, but code validates
before allowing the transition.
"""
# Check if model signaled readiness
model_ready = "advance_to_next_step" in [
tc["tool"] for tc in result.get("tool_calls", [])
]
if not model_ready:
return TransitionResult.CONTINUE
# Code validates the signal
if state_name == "research":
sources = ctx.get("research_sources", [])
if len(sources) < 2:
return TransitionResult.CONTINUE # Not enough sources yet
return TransitionResult.ADVANCE
if state_name == "draft":
word_count = len(result["output"].split())
if word_count < 100:
return TransitionResult.CONTINUE # Draft too short
return TransitionResult.ADVANCE
return TransitionResult.ADVANCE
This is the sweet spot for most production systems. The model does what it is good at (judging semantic completion: "I have enough information") and the code does what it is good at (checking structural requirements: "there are at least two sources" or "the output is at least 100 words").
Deterministic Sub-Flows Within Agentic Systems #
Sometimes you need a stretch of purely deterministic logic inside an otherwise agentic system. The model handles the ambiguous parts; code handles the mechanical parts.
async def process_order(order_data: dict, model: "ModelClient") -> dict:
"""
Hybrid flow: model extracts intent, code executes the mechanics.
"""
# --- Agentic: model interprets the request ---
interpretation = await model.chat(
system="Extract the order details: items, quantities, shipping address.",
messages=[{"role": "user", "content": order_data["raw_request"]}],
)
parsed = parse_structured_output(interpretation.text)
# --- Deterministic: code handles the pipeline ---
inventory = check_inventory(parsed["items"])
if not inventory["all_available"]:
return {"status": "failed", "reason": "Items out of stock",
"unavailable": inventory["missing"]}
tax = calculate_tax(parsed["address"], parsed["items"])
total = sum(item["price"] * item["qty"] for item in parsed["items"]) + tax
payment = await charge_payment(order_data["payment_method"], total)
if not payment["success"]:
return {"status": "failed", "reason": "Payment declined"}
shipping = await create_shipment(parsed["items"], parsed["address"])
# --- Agentic: model generates the confirmation ---
import json
confirmation = await model.chat(
system="Write a friendly order confirmation email.",
messages=[{"role": "user", "content": json.dumps({
"items": parsed["items"],
"total": total,
"tracking": shipping["tracking_number"],
})}],
)
return {
"status": "success",
"confirmation": confirmation.text,
"tracking": shipping["tracking_number"],
}
┌────────────────────────────────────────────────────────┐
│ Hybrid Order Processing │
│ │
│ ┌──────────────┐ │
│ │ LLM: Parse │ ◄── Agentic (ambiguous input) │
│ │ order from │ │
│ │ natural lang│ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Check stock │ ◄── Deterministic (database lookup) │
│ ├──────────────┤ │
│ │ Calculate tax│ ◄── Deterministic (tax rules) │
│ ├──────────────┤ │
│ │ Charge card │ ◄── Deterministic (payment API) │
│ ├──────────────┤ │
│ │ Create label │ ◄── Deterministic (shipping API) │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ LLM: Write │ ◄── Agentic (natural language out) │
│ │ confirmation │ │
│ └──────────────┘ │
└────────────────────────────────────────────────────────┘
The principle: use the model for language understanding or generation, and use code for correctness guarantees. Code should calculate tax; a model can parse a freeform order request.
It is the same boundary that workflow orchestration draws between developer-controlled topology and model-controlled reasoning. State machines just make that boundary explicit, enforceable, and visible in the code structure.
State Persistence and Recovery #
State machines pair naturally with durability. Each state transition is a well-defined checkpoint — you know exactly which state the agent was in, what context it had accumulated, and what transition it was attempting. This makes long-running and durable agents much simpler to implement.
@dataclass
class MachineCheckpoint:
current_state: str
context: dict[str, Any]
history: list[dict]
messages: list[dict]
timestamp: float
class DurableStateMachine(AgentStateMachine):
def __init__(self, *args, storage: "CheckpointStore", **kwargs):
super().__init__(*args, **kwargs)
self.storage = storage
async def save_checkpoint(self, task_id: str) -> None:
import time
checkpoint = MachineCheckpoint(
current_state=self.current_state_name,
context=self.context,
history=self.history,
messages=self._current_messages,
timestamp=time.time(),
)
await self.storage.save(task_id, checkpoint)
async def resume(self, task_id: str) -> dict[str, Any]:
checkpoint = await self.storage.load(task_id)
self.current_state_name = checkpoint.current_state
self.context = checkpoint.context
self.history = checkpoint.history
self._current_messages = checkpoint.messages
# Continue the run loop from where we left off
return await self._continue_from_current_state()
Every state transition triggers a checkpoint save. If the process crashes mid-state, you resume from the last completed transition. If it crashes mid-tool-call within a state, you restart that state from the beginning — which is safe because states are designed as bounded, retryable units of work.
This is easier to get right than checkpointing a free-form ReAct loop. In a ReAct loop, any step might have produced side effects, and restarting from a checkpoint risks repeating a single-execution tool call such as sending an email or charging a payment. In a state machine, the side-effect boundary is the state, and you can design each state to be idempotent or to check for already-completed work before proceeding.
State Machines vs. DAGs #
State machines and directed acyclic graphs (DAGs) both impose structure on agent execution, but they solve different problems.
A DAG defines a fixed, acyclic execution plan. Each node runs once, dependencies are resolved at compile time, and execution moves forward through a one-pass traversal. DAGs are the backbone of data pipelines and CI/CD systems where the topology is known in advance.
A state machine allows cycles. A state can transition back to a previous state (retry, re-gather information, re-validate after correction). This is essential for agent workflows where failure recovery means going backward.
DAG (no cycles): State machine (cycles allowed):
A ──► B ──► C ──► D A ──► B ──► C ──► D
│ ▲ │
└──► E ──► F │ │
└─────┘
Each node runs once. C can loop back to B
No backtracking. (e.g., validation failure
triggers re-collection).
For most agent workflows, state machines are a better fit than DAGs precisely because failure recovery often means revisiting an earlier phase. An invalid email sends the machine back from "validate" to "collect_info." Too few research sources send it back to "research" with refined queries. These cycles are natural in agent workflows, while a strict DAG supports a forward pass.
If your workflow is acyclic — every step succeeds exactly once in a forward sequence — a simple prompt chain or DAG is the right abstraction and a state machine adds unnecessary machinery.
When to Use Each Approach #
Choosing between a ReAct loop, a workflow (chain/DAG), and a state machine depends on the task structure:
| Approach | Structure | Best for | Watch out for |
|---|---|---|---|
| ReAct loop | Model controls everything | Open-ended exploration, unknown step count | Step skipping, cost blowup, hard to audit |
| Prompt chain / DAG | Developer controls everything | Fixed, forward-only pipelines | Rigid — partial failure requires external recovery |
| State machine | Developer controls flow, model controls reasoning | Multi-phase tasks with validation gates | Over-engineering simple tasks |
| Hybrid (state machine + ReAct within states) | Mixed control | Regulated workflows with exploratory sub-tasks | Complexity at the boundary |
The decision heuristic: if you can draw the states and transitions on a whiteboard before writing any code, a state machine is the right tool. When the states emerge during execution, use a ReAct loop — possibly inside a state that handles the exploratory phase.
Conclusion #
State machines are an established software engineering technique that gives agent systems structured progression through phases with flexibility within each phase.
The key trade-offs to carry forward:
- State machines enforce ordering. Code owns the transitions, preventing skipped steps, enabling auditing, and making compliance-sensitive flows tractable.
- Per-state tool sets constrain the model. During validation, visibility remains limited to validation tools, keeping the agent in the current phase.
- Hybrid transitions balance flexibility and control. Let the model signal semantic readiness; let the code verify structural requirements.
- Deterministic sub-flows belong in code. Tax calculations, payment processing, inventory checks — anything with correctness requirements should follow a deterministic code path.
- State machines enable durability. Each transition is a natural checkpoint. Recovery restarts the current state and preserves earlier completed work.
- Cycles beat DAGs for agent workflows. Failure recovery often moves backward, and state machines represent those return paths directly.
The best agent architectures combine autonomy and determinism, using code for structure and models for reasoning, with state machines as the connective tissue between them.