Build a research agent with LlamaIndex Workflows

Install

pip install llama-index-core llama-index-llms-anthropic httpx

Already have LlamaIndex Workflows wired up? TheLlamaIndex Workflows integration pagehas the connection config on its own — this page assumes you have it and gets on with building.

Step by step

1. 1

   Define the events

   Events are the wiring. Each is a typed payload, and a step is selected by the event type it accepts — so the flow is legible from the type signatures alone.

   from llama_index.core.workflow import Event
   
   
   class SearchEvent(Event):
       query: str
   
   
   class FindingsEvent(Event):
       query: str
       hits: list[dict]

2. 2

   Write the steps

   A step takes an event and returns the next one. StartEvent begins the run and StopEvent ends it; everything between is yours.

   from llama_index.core.workflow import (
       Workflow, StartEvent, StopEvent, step, Context
   )
   
   
   class ResearchWorkflow(Workflow):
       @step
       async def plan(self, ctx: Context, ev: StartEvent) -> SearchEvent:
           await ctx.set("question", ev.question)
           return SearchEvent(query=ev.question)
   
       @step
       async def search(self, ctx: Context, ev: SearchEvent) -> FindingsEvent:
           r = await self.client.get(
               "https://api.unlob.com/search",
               headers={"x-api-key": self.api_key},
               params={"q": ev.query, "min_independent_sources": 2, "limit": 8},
           )
           return FindingsEvent(query=ev.query, hits=r.json()["results"])

3. 3

   Fan out across sub-questions

   Here is where the event model earns its keep. Emit one SearchEvent per sub-question and the runtime runs the search step concurrently for each; collect_events waits for all of them. Returning a list does not fan out — emitting several events does.

   from llama_index.core.workflow import Context, step
   
   
   class ResearchWorkflow(Workflow):
       @step
       async def plan(self, ctx: Context, ev: StartEvent) -> SearchEvent | None:
           await ctx.set("question", ev.question)
           sub_questions = [
               ev.question,
               f"{ev.question} — official sources",
               f"{ev.question} — criticism and dissent",
           ]
           await ctx.set("expected", len(sub_questions))
           # Emit rather than return: each becomes its own concurrent run of
           # the search step.
           for q in sub_questions:
               ctx.send_event(SearchEvent(query=q))
           return None
   
       @step
       async def synthesise(self, ctx: Context, ev: FindingsEvent) -> StopEvent | None:
           expected = await ctx.get("expected")
           results = ctx.collect_events(ev, [FindingsEvent] * expected)
           if results is None:
               return None          # not all branches have landed yet
           hits = [h for r in results for h in r.hits]
           # ... synthesise over the merged, deduplicated hits
           return StopEvent(result=hits)

What happens when it runs

What will go wrong

The failures that cost an afternoon rather than a minute, because they produce something that looks like it is working.

• !

  A step whose emitted event type nothing consumes makes the workflow hang until the timeout rather than failing fast. If a run stalls, check that every event type has a consumer.

• !

  Set `timeout` on the Workflow. The default is short enough that a real research run with several searches will be cut off mid-flight.

• !

  Context is shared mutable state across steps and is easy to overuse. Pass data on the event where you can — it keeps the flow readable and the steps independently testable.

• !

  To fan out, emit several events from one step and use `ctx.collect_events` to gather them. Returning a list does not fan out; it just returns a list.