Control Flow Patterns

Execution control structures for agent systems: sequential, parallel, DAG, dynamic routing.

Control Flow Taxonomy

Control Flow
├── Sequential              ← Linear, step-by-step
├── Parallel                ← Independent, simultaneous
├── DAG (Directed Acyclic)  ← Dependencies, partial order
├── Dynamic Routing         ← LLM decides next at runtime
├── Loops/Recursion         ← Repeat until condition
├── Human-in-Loop           ← Approval gates
├── Streaming               ← Token-by-token, progressive
└── Event-Driven            ← Reactive, async triggers

Sequential Flow

Structure

Step 1 → Step 2 → Step 3 → ... → Step N

When to Use

• Strict dependencies (Step N needs Step N-1 output)
• Simple, deterministic workflows
• Audit trail required

Implementation

def sequential(steps, initial_state):
    state = initial_state
    for step in steps:
        state = step.execute(state)
        if step.should_terminate(state):
            break
    return state

Error Handling

try:
    result = step.execute(state)
except RetryableError:
    result = retry_with_backoff(step, state)
except FatalError:
    return handle_failure(state, step)

Parallel Flow

Structure

        ┌── Step A ──┐
Start ──┤            ├── Join → Next
        └── Step B ──┘

When to Use

• Independent steps (no data dependencies)
• I/O bound operations (multiple API calls)
• Reduce latency

Implementation

import asyncio

async def parallel(steps, state):
    tasks = [step.execute(state) for step in steps]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return merge_results(results)

# With concurrency limit
semaphore = asyncio.Semaphore(5)

async def bounded_step(step, state):
    async with semaphore:
        return await step.execute(state)

Join Strategies

DAG (Directed Acyclic Graph)

Structure

    A
   / \
  B   C
   \ /
    D
   / \
  E   F

Representation

@dataclass
class DAGNode:
    id: str
    step: Step
    dependencies: list[str]  # Node IDs that must complete first

class DAGExecutor:
    def __init__(self, nodes: list[DAGNode]):
        self.nodes = {n.id: n for n in nodes}
        self.completed = set()
        self.results = {}

    async def execute(self, initial_state):
        ready = [n for n in self.nodes.values() if not n.dependencies]

        while ready:
            # Execute all ready in parallel
            tasks = [self._execute_node(n, initial_state) for n in ready]
            await asyncio.gather(*tasks)

            # Update ready set
            self.completed.update(n.id for n in ready)
            ready = [
                n for n in self.nodes.values()
                if n.id not in self.completed
                and all(d in self.completed for d in n.dependencies)
            ]
        return self.results

Topological Sort (for validation)

def validate_dag(nodes):
    # Kahn's algorithm
    in_degree = {n.id: len(n.dependencies) for n in nodes}
    queue = [n.id for n in nodes if in_degree[n.id] == 0]
    sorted_order = []

    while queue:
        node_id = queue.pop(0)
        sorted_order.append(node_id)
        for n in nodes:
            if node_id in n.dependencies:
                in_degree[n.id] -= 1
                if in_degree[n.id] == 0:
                    queue.append(n.id)

    if len(sorted_order) != len(nodes):
        raise CycleDetected()
    return sorted_order

Dynamic DAG (LLM-Generated)

def llm_generate_dag(goal, tools):
    prompt = f"""
    Goal: {goal}
    Available tools: {tools}
    Generate a DAG plan as JSON:
    {{
      "nodes": [
        {{"id": "1", "tool": "search", "args": "...", "deps": []}},
        {{"id": "2", "tool": "analyze", "args": "...", "deps": ["1"]}}
      ]
    }}
    """
    return parse_dag(llm.generate(prompt))

Dynamic Routing (LLM-Controlled)

Structure

Current State → LLM Router → Next Step / Tool / Agent
                    ↓
            (can loop, branch, terminate)

Router Prompt Pattern

ROUTER_PROMPT = """
You are a router. Given the current state, decide the next action.

State: {state}
Available actions: {actions}
History: {history}

Respond with JSON:
{{
  "action": "tool_name" | "agent_name" | "finish",
  "args": {{...}},
  "reasoning": "why this action"
}}
"""

Routing Patterns

Implementation

class DynamicRouter:
    def __init__(self, llm, tools, agents):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.agents = {a.name: a for a in agents}

    async def route(self, state, history):
        decision = await self.llm.generate(ROUTER_PROMPT.format(
            state=state, actions=self._list_actions(), history=history
        ))
        action = parse_json(decision)

        if action["action"] == "finish":
            return Finish(action["args"])
        elif action["action"] in self.tools:
            return await self.tools[action["action"]].execute(action["args"])
        elif action["action"] in self.agents:
            return await self.agents[action["action"]].run(action["args"])

Loops & Recursion

While Loop

def while_loop(condition, body, max_iterations=100):
    for i in range(max_iterations):
        if not condition(state):
            break
        state = body(state)
    else:
        raise MaxIterationsExceeded()
    return state

Recursive Decomposition

def recursive_solve(problem, depth=0, max_depth=10):
    if depth >= max_depth or is_atomic(problem):
        return solve_directly(problem)

    subproblems = decompose(problem)
    results = [recursive_solve(p, depth+1) for p in subproblems]
    return compose(results)

Loop Invariants (for correctness)

class LoopInvariant:
    def __init__(self, invariant_fn, variant_fn):
        self.invariant = invariant_fn  # Must hold before/after each iteration
        self.variant = variant_fn      # Must decrease (ensures termination)

    def check(self, state):
        assert self.invariant(state), "Invariant violated"
        # Variant decreases → progress toward termination

Human-in-the-Loop (HITL)

Approval Gates

class HITLGate:
    def __init__(self, timeout=300, default_action="reject"):
        self.timeout = timeout
        self.default = default_action

    async def request_approval(self, context, options):
        # Send to human (Slack, email, dashboard, etc.)
        request = ApprovalRequest(
            context=context,
            options=options,
            expires=datetime.now() + timedelta(seconds=self.timeout)
        )
        response = await self.notify_human(request)

        if response is None:
            return self.default
        return response.choice

HITL Patterns

Streaming / Progressive

Token Streaming

async def stream_response(prompt):
    async for token in llm.stream(prompt):
        yield token  # Send to client immediately

Progressive Disclosure (UI)

async def progressive_execution(plan):
    for step in plan:
        # Send step start
        yield Event("step_start", step=step)

        # Stream tool execution
        async for event in stream_tool(step):
            yield event

        # Send step complete
        yield Event("step_complete", step=step, result=result)

Event-Driven / Reactive

Event Loop

class EventDrivenAgent:
    def __init__(self):
        self.handlers = {}  # event_type → handler
        self.queue = asyncio.Queue()

    async def run(self):
        while True:
            event = await self.queue.get()
            handler = self.handlers.get(event.type)
            if handler:
                await handler(event)

    def on(self, event_type):
        def decorator(fn):
            self.handlers[event_type] = fn
            return fn
        return decorator

Event Types

Control Flow in Frameworks

Choosing Control Flow

Related Notes

• Single Agent Patterns — Patterns using these flows
• Cognitive Architectures — Symbolic planners as control flow
• LangGraph — Graph-based control flow implementation
• AutoGen — Conversation-based flow
• OpenAI Agents SDK — Handoffs as control flow
• Checkpointing — Persisting flow state
• Coordination Patterns — Multi-agent control flow
• Observability — Tracing control flow

Sources: LangGraph docs, AutoGen docs, CrewAI docs, OpenAI Agents SDK, PydanticAI docs, workflow engine literature (Airflow, Prefect, Dagster), distributed systems patterns, Petri nets, state machine theory.