Event-Driven Agent Workflows with LlamaIndex

Event-driven architectures have long been the standard for building scalable backend systems. Now, with LlamaIndex Workflows, this same paradigm applies to AI agent orchestration. Instead of rigid sequential chains, event-driven workflows let you build flexible, composable agent pipelines where each step emits and responds to typed events.

This approach is particularly powerful for document-heavy, data-intensive applications where traditional linear agent chains break down. Let’s explore how to build event-driven agent workflows using LlamaIndex, with practical code examples you can adapt to your use case.

Why This Matters

Traditional agent chains follow a fixed execution path: input → step 1 → step 2 → output. This works for simple use cases but becomes limiting when:

• Steps need to run in parallel: A document processing pipeline might need to extract text, generate embeddings, and pull metadata simultaneously.
• Steps need to react to multiple events: An agent might need to respond to both user queries and system events.
• Error handling requires branching: When one step fails, the workflow needs alternative paths rather than simply stopping.

Event-driven workflows solve these problems by decoupling event producers from consumers. Each workflow step is an event handler that can emit events, listen for events, and make decisions based on the full event context. This makes workflows inherently more flexible and easier to extend.

Architecture of Event-Driven Workflows

LlamaIndex Workflows model agent execution as a graph of event handlers. The key components are:

1. Events: Typed data structures that flow between steps
2. Steps: Functions that handle specific events and emit new events
3. Workflow: The graph that connects steps and manages event routing

Here’s a minimal example showing the core concepts:

from llama_index.core.workflow import Workflow, step, Context, StartEvent, StopEvent
from llama_index.llms.openai import OpenAI

class SimpleWorkflow(Workflow):
    @step
    async def first_step(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """Handle the initial input and produce output."""
        llm = OpenAI(model="gpt-4o")
        result = await llm.acomplete(f"Process this: {ev.input}")
        return StopEvent(result=result.result)

# Run the workflow
workflow = SimpleWorkflow()
result = await workflow.run(input="Hello, world!")
print(result)

The StartEvent and StopEvent are built-in event types that mark the beginning and end of a workflow. Custom events let you build more complex patterns.

Building a Document Processing Pipeline

Let’s build a practical event-driven workflow for processing documents with multiple extraction and analysis steps:

from llama_index.core.workflow import (
    Workflow, step, Context, StartEvent, StopEvent, Event
)
from llama_index.llms.openai import OpenAI
from dataclasses import dataclass
from typing import Optional

# Define custom events
@dataclass
class DocumentLoaded(Event):
    content: str
    source: str
    doc_type: str

@dataclass
class ContentExtracted(Event):
    text: str
    metadata: dict
    chunks: list[str]

@dataclass
class AnalysisComplete(Event):
    summary: str
    key_findings: list[str]
    entities: list[str]

class DocumentProcessingWorkflow(Workflow):
    @step
    async def load_document(self, ctx: Context, ev: StartEvent) -> DocumentLoaded:
        """Step 1: Load and normalize the document."""
        # In production, handle PDF, HTML, markdown, etc.
        content = ev.get("content", "")
        source = ev.get("source", "unknown")
        
        # Normalize content
        normalized = content.strip()
        
        ctx.send_event(DocumentLoaded(
            content=normalized,
            source=source,
            doc_type=ev.get("doc_type", "text")
        ))
        # Return None to keep the workflow running
        return None
    
    @step
    async def extract_content(self, ctx: Context, ev: DocumentLoaded) -> ContentExtracted:
        """Step 2: Extract structured content from the document."""
        llm = OpenAI(model="gpt-4o")
        
        extraction_prompt = f"""Extract the following from this document:
        1. Clean text content
        2. Key metadata (title, author, date if available)
        3. Break into logical chunks (sections, paragraphs)
        
        Document:
        {ev.content[:3000]}
        
        Return as JSON with keys: text, metadata, chunks"""
        
        response = await llm.acomplete(extraction_prompt)
        
        # Parse response (simplified for example)
        import json
        try:
            data = json.loads(response.text)
        except json.JSONDecodeError:
            data = {
                "text": ev.content,
                "metadata": {"source": ev.source},
                "chunks": [ev.content]
            }
        
        return ContentExtracted(
            text=data.get("text", ev.content),
            metadata=data.get("metadata", {}),
            chunks=data.get("chunks", [ev.content])
        )
    
    @step
    async def analyze_content(self, ctx: Context, ev: ContentExtracted) -> AnalysisComplete:
        """Step 3: Analyze the extracted content."""
        llm = OpenAI(model="gpt-4o")
        
        analysis_prompt = f"""Analyze this document content and provide:
        1. A concise summary (2-3 sentences)
        2. Key findings (3-5 bullet points)
        3. Named entities mentioned
        
        Content:
        {ev.text[:4000]}
        
        Return as JSON with keys: summary, key_findings, entities"""
        
        response = await llm.acomplete(analysis_prompt)
        
        import json
        try:
            data = json.loads(response.text)
        except json.JSONDecodeError:
            data = {
                "summary": "Analysis complete",
                "key_findings": [],
                "entities": []
            }
        
        return AnalysisComplete(
            summary=data.get("summary", ""),
            key_findings=data.get("key_findings", []),
            entities=data.get("entities", [])
        )
    
    @step
    async def finalize(self, ctx: Context, ev: AnalysisComplete) -> StopEvent:
        """Step 4: Format and return the final result."""
        result = {
            "summary": ev.summary,
            "findings": ev.key_findings,
            "entities": ev.entities,
            "status": "complete"
        }
        return StopEvent(result=result)

# Run the workflow
workflow = DocumentProcessingWorkflow()
result = await workflow.run(
    content="Your document content here...",
    source="document.pdf",
    doc_type="pdf"
)

Multi-Agent Event-Driven Patterns

The real power of event-driven workflows emerges when coordinating multiple agents. Here’s a pattern for a research team where different agents handle different aspects:

from llama_index.core.workflow import (
    Workflow, step, Context, StartEvent, StopEvent, Event
)
from llama_index.llms.openai import OpenAI
from dataclasses import dataclass
import asyncio

# Events for multi-agent coordination
@dataclass
class ResearchQuery(Event):
    topic: str
    depth: str  # "surface", "deep", "comprehensive"

@dataclass
class WebSearchResults(Event):
    sources: list[dict]
    query: str

@dataclass
class AnalysisResults(Event):
    analysis: str
    confidence: float

@dataclass
class SynthesisRequest(Event):
    research_data: dict
    analysis_data: dict

class MultiAgentResearchWorkflow(Workflow):
    @step
    async def coordinate_research(self, ctx: Context, ev: StartEvent) -> ResearchQuery:
        """Orchestrate multiple research agents."""
        topic = ev.get("topic", "")
        depth = ev.get("depth", "surface")
        
        ctx.send_event(ResearchQuery(topic=topic, depth=depth))
        return None
    
    @step
    async def search_agent(self, ctx: Context, ev: ResearchQuery) -> WebSearchResults:
        """Agent that performs web searches."""
        llm = OpenAI(model="gpt-4o")
        
        search_prompt = f"""Generate search queries for: {ev.topic}
        Depth: {ev.depth}
        
        Return 3-5 specific search queries as a JSON list."""
        
        response = await llm.acomplete(search_prompt)
        
        # Simulate search results (integrate with actual search API)
        results = [
            {"query": f"search result for {ev.topic}", 
             "snippet": "Relevant content..."}
        ]
        
        return WebSearchResults(sources=results, query=ev.topic)
    
    @step
    async def analysis_agent(self, ctx: Context, ev: ResearchQuery) -> AnalysisResults:
        """Agent that analyzes the topic from available information."""
        llm = OpenAI(model="gpt-4o")
        
        analysis_prompt = f"""Provide an initial analysis of: {ev.topic}
        
        Return as JSON with keys: analysis, confidence (0-1)"""
        
        response = await llm.acomplete(analysis_prompt)
        
        import json
        try:
            data = json.loads(response.text)
        except json.JSONDecodeError:
            data = {"analysis": response.text, "confidence": 0.5}
        
        return AnalysisResults(
            analysis=data.get("analysis", ""),
            confidence=data.get("confidence", 0.5)
        )
    
    @step
    async def synthesize_results(
        self, ctx: Context, 
        ev1: WebSearchResults, 
        ev2: AnalysisResults
    ) -> StopEvent:
        """Combine results from multiple agents."""
        llm = OpenAI(model="gpt-4o")
        
        synthesis_prompt = f"""Synthesize the following research into a 
        comprehensive analysis:
        
        Web Research: {ev1.sources}
        Initial Analysis: {ev2.analysis}
        Confidence: {ev2.confidence}
        
        Provide a balanced, well-sourced analysis."""
        
        response = await llm.acomplete(synthesis_prompt)
        
        return StopEvent(result={
            "synthesis": response.text,
            "sources_used": len(ev1.sources),
            "confidence": ev2.confidence
        })

# Run the multi-agent workflow
workflow = MultiAgentResearchWorkflow()
result = await workflow.run(
    topic="Impact of AI agents on software development",
    depth="comprehensive"
)

Error Handling and Resilience

Production event-driven workflows need robust error handling. LlamaIndex Workflows support retry patterns and alternative paths:

from llama_index.core.workflow import (
    Workflow, step, Context, StartEvent, StopEvent, Event
)
from llama_index.llms.openai import OpenAI
import asyncio
from typing import Optional

class ResilientWorkflow(Workflow):
    @step
    async def primary_processing(self, ctx: Context, ev: StartEvent) -> Optional[StopEvent]:
        """Primary processing with fallback."""
        llm = OpenAI(model="gpt-4o")
        
        try:
            response = await asyncio.wait_for(
                llm.acomplete(f"Process: {ev.input}"),
                timeout=30.0
            )
            return StopEvent(result={"status": "success", "output": response.text})
        except asyncio.TimeoutError:
            # Log the failure
            print(f"Primary processing timed out for: {ev.input}")
            # Emit a fallback event or return a degraded result
            return StopEvent(result={
                "status": "timeout", 
                "output": "Processing timed out. Please try a simpler query."
            })
        except Exception as e:
            print(f"Primary processing failed: {e}")
            return StopEvent(result={
                "status": "error",
                "output": f"An error occurred: {str(e)}"
            })
    
    @step
    async def batch_processor(self, ctx: Context, ev: StartEvent) -> Optional[StopEvent]:
        """Process multiple items with partial failure handling."""
        items = ev.get("items", [])
        results = []
        
        for item in items:
            try:
                # Process each item
                result = await self._process_item(item)
                results.append({"item": item, "status": "success", "result": result})
            except Exception as e:
                results.append({"item": item, "status": "failed", "error": str(e)})
        
        return StopEvent(result={
            "total": len(items),
            "successful": sum(1 for r in results if r["status"] == "success"),
            "results": results
        })
    
    async def _process_item(self, item: str) -> str:
        llm = OpenAI(model="gpt-4o")
        response = await llm.acomplete(f"Process item: {item}")
        return response.text

Best Practices

Event Design

Keep events small and focused. Each event should carry exactly the data needed by its consumers—no more, no less. Overly large events waste memory and make workflows harder to reason about.

Step Independence

Design steps to be as independent as possible. Steps should not depend on internal workflow state beyond what’s passed through events. This makes steps reusable and workflows easier to test.

Error Boundaries

Define clear error boundaries at each step. Decide whether a failure in one step should stop the entire workflow, retry, or fall back to alternative logic. LlamaIndex Workflows make this explicit through your step implementations.

Async Everything

Use async operations throughout your workflow. LlamaIndex Workflows are designed for async execution, and blocking operations will bottleneck your entire pipeline. Use await for LLM calls, database operations, and any I/O.

Start Simple, Add Complexity

Begin with a linear workflow and add branching, parallel steps, and error handling as needed. Over-engineering event-driven workflows early makes them harder to debug and modify.

Next Steps

Event-driven workflows in LlamaIndex provide a powerful pattern for building composable, flexible agent systems. Start by identifying your workflow’s natural event boundaries, then build outward from there.

For production deployment, consider adding observability hooks at each step to trace event flow and identify bottlenecks. LlamaIndex integrates with observability tools like Langfuse and Arize Phoenix that can help you monitor workflow execution.

The event-driven paradigm scales naturally from simple linear workflows to complex multi-agent systems. Master the fundamentals, and you’ll be able to build increasingly sophisticated agent architectures without restructuring your core patterns.