Multi-Step Workflows

Real-world agent applications combine multiple patterns into cohesive workflows. This lesson shows how to chain actions and manage state across complex operations.

Workflow Architecture

A well-designed workflow connects patterns into a processing pipeline:

class AgentWorkflow:
    def __init__(self):
        self.state = {}
        self.history = []

    def run(self, task):
        # 1. Plan the approach
        plan = self.plan(task)

        # 2. Execute with tools
        for step in plan.steps:
            result = self.execute_step(step)
            self.state[step.name] = result
            self.history.append({"step": step, "result": result})

            # 3. Reflect and adjust
            if self.should_reflect(result):
                improved = self.reflect(result)
                self.state[step.name] = improved

        # 4. Synthesize final output
        return self.synthesize(self.state)

State Management

Maintaining state across steps is crucial for complex workflows:

State Type	Purpose	Example
Working memory	Current step context	Variables, intermediate results
Conversation history	What happened before	Previous messages, decisions
Execution trace	Debugging & learning	Actions taken, outcomes
External state	World changes	Database updates, file changes

Common Workflow Patterns

Sequential Pipeline

Input → Step 1 → Step 2 → Step 3 → Output

Best for: Data processing, content generation

Branching Logic

Input → Decision → [Path A] or [Path B] → Output

Best for: Customer support, conditional processing

Parallel Execution

Input → [Step 1] + [Step 2] + [Step 3] → Merge → Output

Best for: Research, gathering multiple data sources

Loop with Exit

Input → Process → Check → (repeat or exit) → Output

Best for: Iterative improvement, search tasks

Practical Example: Content Research Agent

async def research_workflow(topic):
    # Step 1: Plan research approach
    research_plan = await plan_research(topic)

    # Step 2: Gather sources (parallel)
    sources = await asyncio.gather(
        search_academic(topic),
        search_news(topic),
        search_industry(topic)
    )

    # Step 3: Analyze and synthesize
    analysis = await analyze_sources(sources)

    # Step 4: Reflect on completeness
    gaps = await identify_gaps(analysis, research_plan)

    # Step 5: Fill gaps if needed
    if gaps:
        additional = await targeted_research(gaps)
        analysis = await merge_findings(analysis, additional)

    # Step 6: Generate final report
    return await generate_report(analysis)

Key Takeaways

• Combine patterns thoughtfully based on task requirements
• Manage state to enable complex multi-step reasoning
• Build in checkpoints for debugging and recovery
• Design for failure with fallbacks at each step

Now test your understanding with the module quiz! :::