TopolearnSplitting a complex job across focused agents keeps each prompt and tool set small and reliable.
One agent with twenty tools and a giant system prompt gets confused. Splitting the job across several specialised agents, each with a few tools and a narrow role, usually works better.
`AgentWorkflow` is the built-in way to do this with handoffs:
research = FunctionAgent(name="researcher", description="Finds facts with web search",
tools=[search], can_handoff_to=["writer"], llm=llm)
writer = FunctionAgent(name="writer", description="Writes the final report", llm=llm)
workflow = AgentWorkflow(agents=[research, writer], root_agent="researcher")
response = await workflow.run(user_msg="Write a short report on heat pumps.")- Every agent needs a unique
nameand adescription; other agents read the description when deciding whom to hand off to. can_handoff_tolimits who an agent may pass control to.root_agentreceives the user's message first.initial_state={...}creates shared state that tools can read and write throughctx.store, for example research notes the writer later uses.
Only one agent is active at a time; a handoff passes the conversation along.
Other patterns:
- Orchestrator: one top-level agent calls other agents as tools (wrap each sub-agent's
runin an async function). The orchestrator keeps control and decides when to finish. - Custom workflow: write a
Workflowthat calls agents in a fixed order or in parallel. Most predictable.
Stream events to watch handoffs: AgentOutput and the current_agent_name field on events show which agent is speaking. Multi-agent runs make many LLM calls, so measure cost and use simpler designs when they suffice.
Example
import asyncio
from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent
from llama_index.core.workflow import Context
from llama_index.llms.openai import OpenAI
llm = OpenAI(model="gpt-4o-mini")
async def save_note(ctx: Context, note: str) -> str:
"""Save a research note for the writer."""
state = await ctx.store.get("state")
state["notes"].append(note)
await ctx.store.set("state", state)
return "saved"
researcher = FunctionAgent(name="researcher", description="Collects key facts as notes.",
system_prompt="Save 3 short facts with save_note, then hand off to writer.",
tools=[save_note], llm=llm, can_handoff_to=["writer"])
writer = FunctionAgent(name="writer", description="Writes the final answer from notes.",
system_prompt="Write a 3-sentence summary from the saved notes in state.",
tools=[], llm=llm)
workflow = AgentWorkflow(agents=[researcher, writer], root_agent="researcher",
initial_state={"notes": []})
async def main():
print(await workflow.run(user_msg="Summarize why heat pumps are efficient."))
asyncio.run(main())Remember
AgentWorkflow(agents, root_agent)coordinates agents through handoffs- Names, descriptions and
can_handoff_tosteer who gets control - Shared
initial_statelets agents pass data viactx.store
Watch out for
- Writing vague agent descriptions, so agents hand off to the wrong colleague or never hand off.