# 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
```python
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
```python
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
```python
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
| Strategy | Description |
|---|---|
| **Wait All** | All must complete (default) |
| **Wait Any** | First success continues |
| **Wait N** | Quorum (e.g., 2 of 3) |
| **Timeout** | Proceed with partial after deadline |

---

## DAG (Directed Acyclic Graph)

### Structure
```
    A
   / \
  B   C
   \ /
    D
   / \
  E   F
```

### Representation
```python
@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)
```python
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)
```python
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
```python
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
| Pattern | Description |
|---|---|
| **Tool Router** | LLM picks next tool |
| **Agent Router** | LLM delegates to sub-agent |
| **Conditional** | If X then A else B |
| **Loop** | Repeat until condition |
| **Fallback** | Try A, if fails try B |

### Implementation
```python
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
```python
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
```python
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)
```python
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
```python
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
| Pattern | Description | Use Case |
|---|---|---|
| **Pre-action** | Approve before execution | Destructive actions |
| **Post-action** | Review result, approve/rollback | High-stakes output |
| **Plan approval** | Approve full plan before execution | Complex, costly workflows |
| **Escalation** | Route to human on failure/uncertainty | Edge cases |
| **Continuous** | Human monitors, can intervene anytime | Long-running autonomous |

---

## Streaming / Progressive

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

### Progressive Disclosure (UI)
```python
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
```python
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
| Event | Source | Handler |
|---|---|---|
| `user_message` | Chat interface | Conversation handler |
| `tool_result` | Tool execution | Continue/route |
| `timer` | Scheduler | Periodic tasks |
| `webhook` | External system | Integration handler |
| `alert` | Monitoring | Escalation handler |
| `human_input` | HITL gate | Resume execution |

---

## Control Flow in Frameworks

| Framework | Primary Model | DAG | Dynamic | HITL | Streaming |
|---|---|---|---|---|---|
| **LangGraph** | StateGraph (DAG) | ✅ Native | ✅ Conditional edges | ✅ Interrupt | ✅ Native |
| **AutoGen** | Conversation | ❌ | ✅ GroupChat | ✅ UserProxy | ❌ |
| **CrewAI** | Process (sequential/hierarchical) | ⚠️ Sequential | ⚠️ Limited | ✅ Human tools | ❌ |
| **OpenAI Agents SDK** | Handoffs | ⚠️ Via handoffs | ✅ Function tools | ✅ Guardrails | ✅ Streaming |
| **PydanticAI** | Structured agents | ⚠️ Via deps | ✅ Tools | ✅ Validators | ✅ Streaming |
| **Semantic Kernel** | Planners | ⚠️ Plan steps | ✅ Stepwise | ✅ Filters | ⚠️ |

---

## Choosing Control Flow

| Requirement | Recommended |
|---|---|
| Simple, linear | Sequential |
| Independent I/O | Parallel |
| Complex dependencies | DAG |
| Unpredictable paths | Dynamic Routing |
| Repetitive until condition | Loop |
| Human oversight needed | HITL gates |
| Real-time UX | Streaming |
| External triggers | Event-Driven |

---

## Related Notes

- [[Architectures/Single-Agent-Patterns.md]] — Patterns using these flows
- [[Architectures/Cognitive-Architectures.md]] — Symbolic planners as control flow
- [[Frameworks/LangGraph.md]] — Graph-based control flow implementation
- [[Frameworks/AutoGen.md]] — Conversation-based flow
- [[Frameworks/OpenAI-Agents-SDK.md]] — Handoffs as control flow
- [[StateManagement/Checkpointing.md]] — Persisting flow state
- [[MultiAgent/Coordination-Patterns.md]] — Multi-agent control flow
- [[Production/Observability.md]] — 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.*