Event-Driven Orchestration: Coordinating Agents over Message Buses

Synchronous REST or RPC calls between agents create tight coupling. If Agent A calls Agent B synchronously and Agent B takes 20 seconds to process a document, Agent A’s connection thread remains blocked. Furthermore, if Agent B crashes, the entire workflow fails instantly without backpressure or persistent queues.

In enterprise architectures, Event-Driven Agent Orchestration decouples agent communication. Agents communicate asynchronously by publishing and subscribing to typed events on a message bus (Kafka, RabbitMQ, or NATS).

Advantages of Event-Driven Agent Architectures

1. Decoupled Scalability: Add new specialized worker agents (e.g., fraud detectors, compliance checkers) simply by subscribing them to existing event topics without modifying existing producer code.
2. Durable Queuing & Backpressure: If an influx of 10,000 document processing requests arrives, message queues buffer events safely while worker agents consume them at maximum sustainable throughput.
3. Replayable Event Sourcing: Persisted event streams allow developers to replay historical agent event sequences to debug failures or train updated model prompts.

[Trigger Event] --> (Message Bus / Topic: "order.created")
                         |
      +------------------+------------------+
      |                                     |
[Inventory Agent]                     [Fraud Detection Agent]
 (Emits: "inventory.reserved")         (Emits: "fraud.cleared")
      |                                     |
      +------------------+------------------+
                         v
             (Topic: "order.processed")
                         v
                  [Shipping Agent]

Event-Driven Workflows with LlamaIndex

LlamaIndex Workflows provide a event-driven framework where agent steps are triggered by specific Event classes.

import asyncio
from llamaindex.core.workflow import Workflow, Event, StartEvent, StopEvent, step

# 1. Define Strongly Typed Domain Events
class DocumentIngestedEvent(Event):
    doc_id: str
    text_content: str

class EntityExtractedEvent(Event):
    doc_id: str
    entities: list[str]

class CompliancePassedEvent(Event):
    doc_id: str
    status: str

# 2. Construct Event-Driven Workflow Class
class DocumentProcessingWorkflow(Workflow):

    @step
    async def ingest_document(self, ev: StartEvent) -> DocumentIngestedEvent:
        doc_id = ev.get("doc_id")
        raw_text = ev.get("raw_text")
        print(f"[Step 1] Ingested Doc {doc_id}")
        return DocumentIngestedEvent(doc_id=doc_id, text_content=raw_text)

    @step
    async def extract_entities(self, ev: DocumentIngestedEvent) -> EntityExtractedEvent:
        # Agent extracts named entities from text asynchronously
        print(f"[Step 2] Extracting entities for {ev.doc_id}")
        entities = ["ACME Corp", "$50,000 USD", "Contract-2026"]
        return EntityExtractedEvent(doc_id=ev.doc_id, entities=entities)

    @step
    async def verify_compliance(self, ev: EntityExtractedEvent) -> StopEvent:
        # Agent checks extracted entities against policy
        print(f"[Step 3] Compliance check for {ev.doc_id}")
        result = f"Doc {ev.doc_id} APPROVED. Found entities: {', '.join(ev.entities)}"
        return StopEvent(result=result)

Running the Event Workflow

async def main():
    wf = DocumentProcessingWorkflow(timeout=30.0, verbose=True)
    
    # Trigger workflow by sending StartEvent
    result = await wf.run(
        doc_id="DOC-9921",
        raw_text="Enterprise license agreement with ACME Corp for $50,000 USD."
    )
    print("Workflow Final Result:", result)

if __name__ == "__main__":
    asyncio.run(main())

Integrating Message Buses in Production

In distributed cluster setups, wrap each workflow step handler inside a Kafka consumer loop:

• Consume: Read incoming message from Kafka topic documents.raw.
• Process: Trigger LlamaIndex Workflow step.
• Publish: Emit output DocumentIngestedEvent to Kafka topic documents.ingested.

For complete documentation on event handlers, state streaming, and workflow visualization, explore the official LlamaIndex Documentation.