LlamaIndex — Agents, Tools & Workflows¶

from llama_index.core.tools import FunctionTool, QueryEngineTool


def order_status(order_id: str, include_history: bool = False) -> str:
    """Return the delivery status of an order.

    Args:
        order_id: Order id such as "o-1042".
        include_history: Also return past status changes.
    """
    return "shipped"


status_tool = FunctionTool.from_defaults(order_status)            # name, description, schema from the function
kb_tool = QueryEngineTool.from_defaults(
    index.as_query_engine(llm=llm, similarity_top_k=3),
    name="kb",
    description="Search the support knowledge base. Input: a full question.",
)

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI

agent = FunctionAgent(
    name="support",
    description="Answers support questions",
    system_prompt="You are a support agent. Use the kb tool for policies and order_status for orders.",
    tools=[kb_tool, status_tool],
    llm=OpenAI(model="gpt-4o-mini", temperature=0),
)

response = await agent.run("Where is order o-1042 and what is the refund window?", max_iterations=10)
print(response)                          # AgentOutput; str() is the final text
response.tool_calls                      # list[ToolCallResult] made during the run

from llama_index.core.agent.workflow import AgentOutput, AgentStream, ToolCall, ToolCallResult

handler = agent.run("Where is order o-1042?")
async for event in handler.stream_events():
    if isinstance(event, AgentStream):
        print(event.delta, end="", flush=True)                  # text tokens
    elif isinstance(event, ToolCall):
        print(f"\n-> {event.tool_name}({event.tool_kwargs})")
    elif isinstance(event, ToolCallResult):
        print(f"<- {event.tool_output.content}")
result = await handler

from llama_index.core.memory import Memory
from llama_index.core.workflow import Context


async def remember_name(ctx: Context, name: str) -> str:
    """Remember the customer's name."""
    async with ctx.store.edit_state() as state:          # atomic update
        state["state"]["name"] = name
    return f"saved {name}"


agent = FunctionAgent(tools=[remember_name, status_tool], llm=llm, initial_state={"name": None})
ctx, memory = Context(agent), Memory.from_defaults(session_id="user-42")
await agent.run("I am Ann", ctx=ctx, memory=memory)
await agent.run("Where is o-1042?", ctx=ctx, memory=memory)   # same ctx: state continues; same memory: history continues
await ctx.store.get("state")                                  # {'name': 'Ann'}

from llama_index.core.workflow import Context, HumanResponseEvent, InputRequiredEvent


async def refund(ctx: Context, order_id: str) -> str:
    """Refund an order. A human must approve first."""
    answer = await ctx.wait_for_event(
        HumanResponseEvent,
        waiter_id=f"refund-{order_id}",
        waiter_event=InputRequiredEvent(prefix=f"Approve refund for {order_id}? (yes/no)"),
    )
    return "refunded" if answer.response.strip().lower() == "yes" else "refused by approver"


agent = FunctionAgent(tools=[refund], llm=llm)
handler = agent.run("Refund order o-1042")
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        print(event.prefix)
        handler.ctx.send_event(HumanResponseEvent(response="yes"))      # from a UI, queue, or Slack callback
print(await handler)

from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent, ToolCall

triage = FunctionAgent(
    name="triage", description="Routes requests to the right specialist",
    system_prompt="Hand off order questions to the orders agent.", tools=[], llm=llm, can_handoff_to=["orders"],
)
orders = FunctionAgent(
    name="orders", description="Looks up order status",
    system_prompt="Answer order questions with order_status.", tools=[status_tool], llm=llm, can_handoff_to=["triage"],
)
workflow = AgentWorkflow(agents=[triage, orders], root_agent="triage", initial_state={"seen": []}, timeout=120)

handler = workflow.run("Where is order o-1042?")
async for event in handler.stream_events():
    if isinstance(event, ToolCall) and event.tool_name == "handoff":
        print("handoff:", event.tool_kwargs)            # {'to_agent': 'orders', 'reason': '...'}
print(await handler)
await handler.ctx.store.get("current_agent_name")       # 'orders': the next run starts here

from llama_index.core.response_synthesizers import get_response_synthesizer
from llama_index.core.schema import NodeWithScore
from llama_index.core.workflow import Context, Event, StartEvent, StopEvent, Workflow, step


class RetrievedEvent(Event):
    query: str
    nodes: list[NodeWithScore]


class RAGWorkflow(Workflow):
    def __init__(self, index, llm, **kwargs):
        super().__init__(**kwargs)
        self.retriever = index.as_retriever(similarity_top_k=3)
        self.llm = llm

    @step
    async def retrieve(self, ev: StartEvent) -> RetrievedEvent | StopEvent:
        query = ev.get("query")
        if not query:
            return StopEvent(result="Please ask a question.")          # branch: stop early
        return RetrievedEvent(query=query, nodes=await self.retriever.aretrieve(query))

    @step
    async def synthesize(self, ev: RetrievedEvent) -> StopEvent:
        synth = get_response_synthesizer(llm=self.llm, response_mode="compact")
        return StopEvent(result=await synth.asynthesize(ev.query, nodes=ev.nodes))


workflow = RAGWorkflow(index, llm, timeout=60)
response = await workflow.run(query="express delivery time")           # StartEvent fields as keyword arguments
print(response, [n.node_id for n in response.source_nodes])

import asyncio

from workflows.retry_policy import ConstantDelayRetryPolicy


class Item(Event):
    n: int


class Done(Event):
    n: int


class FanOut(Workflow):
    @step
    async def start(self, ev: StartEvent) -> list[Item]:
        return [Item(n=i) for i in range(3)]                    # a list fans out: one event per item

    @step(num_workers=3, retry_policy=ConstantDelayRetryPolicy(delay=0.5, maximum_attempts=3))
    async def work(self, ev: Item) -> Done:                    # up to 3 copies run concurrently
        await asyncio.sleep(0.2)
        return Done(n=ev.n * 10)

    @step
    async def join(self, ev: list[Done]) -> StopEvent:         # a list parameter waits for all results
        return StopEvent(result=sorted(e.n for e in ev))


await FanOut(timeout=30).run()                                   # [0, 10, 20]

from llama_index.tools.mcp import BasicMCPClient, McpToolSpec

client = BasicMCPClient("python", args=["orders_mcp_server.py"])          # stdio server
# client = BasicMCPClient("https://mcp.example.com/mcp", headers={"Authorization": "Bearer <TOKEN>"})   # HTTP
spec = McpToolSpec(client=client, allowed_tools=["order_status"])         # allowlist; default is every tool
tools = await spec.to_tool_list_async()

agent = FunctionAgent(tools=tools, llm=llm, system_prompt="Support agent")
await agent.run("Status of o-1?")

from llama_index.core.memory import Memory, StaticMemoryBlock

memory = Memory.from_defaults(
    session_id="user-42",                                     # key in the chat store
    token_limit=30000,                                        # default
    async_database_uri="sqlite+aiosqlite:///memory.db",       # default is an in-memory SQLite database
    memory_blocks=[StaticMemoryBlock(name="persona", static_content="Always answer in English.")],
)
await agent.run("Hi, I am Ann", memory=memory)
await agent.run("What is my name?", memory=memory)           # history comes from the store
await memory.aget()                                          # list[ChatMessage] as sent to the LLM

from pydantic import BaseModel


class Ticket(BaseModel):
    order_id: str
    action: str


agent = FunctionAgent(tools=[status_tool], llm=llm, output_cls=Ticket)
result = await agent.run("Status of o-1042, then propose an action")
result.structured_response            # {'order_id': 'o-1042', 'action': 'none'}
result.get_pydantic_model(Ticket)     # Ticket(order_id='o-1042', action='none')