LlamaIndex Agents

LlamaIndex's agent framework: query engines, workflow agents, and multi-agent orchestration built on LlamaIndex's retrieval and indexing.

Core Concepts

Agent Types

from llama_index.agent import (
    ReActAgent,
    OpenAIAgent,
    FunctionCallingAgent,
    AgentRunner
)
from llama_index.core.tools import FunctionTool, QueryEngineTool
from llama_index.core import VectorStoreIndex

# ReAct Agent (reasoning + acting)
agent = ReActAgent.from_tools(
    tools=[search_tool, calc_tool],
    llm=llm,
    verbose=True
)

# OpenAI Agent (function calling native)
agent = OpenAIAgent.from_tools(
    tools=[search_tool, calc_tool],
    llm=llm
)

# Function Calling Agent (generic)
agent = FunctionCallingAgent.from_tools(
    tools=[search_tool, calc_tool],
    llm=llm
)

Query Engine as Tool

from llama_index.core import VectorStoreIndex
from llama_index.core.tools import QueryEngineTool

# Build index
index = VectorStoreIndex.from_documents(documents)

# Create query engine
query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="compact"
)

# Wrap as tool
rag_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="document_search",
    description="Search internal documents for information"
)

# Use in agent
agent = ReActAgent.from_tools([rag_tool, web_search_tool], llm=llm)

Workflow Agents (Structured Processes)

Workflow Definition

from llama_index.core.workflow import (
    Workflow, StartEvent, StopEvent, Event, step, Context
)
from typing import AsyncGenerator

class ResearchWorkflow(Workflow):
    @step
    async def plan(self, ctx: Context, ev: StartEvent) -> PlanEvent:
        topic = ev.topic
        plan = await llm.acomplete(f"Create research plan for: {topic}")
        return PlanEvent(plan=plan.text)

    @step
    async def research(self, ctx: Context, ev: PlanEvent) -> ResearchEvent:
        findings = []
        for query in ev.plan.queries:
            result = await search_tool.acall(query)
            findings.append(result)
        return ResearchEvent(findings=findings)

    @step
    async def synthesize(self, ctx: Context, ev: ResearchEvent) -> StopEvent:
        report = await llm.acomplete(f"Synthesize: {ev.findings}")
        return StopEvent(result=report.text)

# Run
workflow = ResearchWorkflow()
result = await workflow.run(topic="AI Agents in 2024")

Event-Driven Architecture

from llama_index.core.workflow import Event

class CustomEvent(Event):
    data: str
    metadata: dict

# Type-safe event passing
@step
async def step1(ctx: Context, ev: StartEvent) -> CustomEvent:
    return CustomEvent(data="result", metadata={"source": "step1"})

@step
async def step2(ctx: Context, ev: CustomEvent) -> StopEvent:
    return StopEvent(result=process(ev.data))

Concurrency & Parallelism

@step
async def parallel_research(ctx: Context, ev: PlanEvent) -> ResearchEvent:
    # Run multiple searches in parallel
    tasks = [search_tool.acall(q) for q in ev.plan.queries]
    results = await asyncio.gather(*tasks)
    return ResearchEvent(findings=results)

# Or use built-in concurrency
@step(num_workers=4)
async def parallel_step(ctx: Context, ev: Event) -> Event:
    return process(ev)

Multi-Agent Orchestration

AgentTeam

from llama_index.agent import AgentTeam

team = AgentTeam(
    agents=[
        researcher_agent,
        writer_agent,
        editor_agent
    ],
    orchestrator=orchestrator_agent  # Optional
)

result = await team.run("Research and write about AI agents")

Custom Orchestration

class CustomOrchestrator:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents

    async def run(self, task: str) -> str:
        # Route to appropriate agent
        agent_name = await self.route(task)
        agent = self.agents[agent_name]
        return await agent.run(task)

    async def route(self, task: str) -> str:
        return await router_agent.run(task)

Tools

FunctionTool

from llama_index.core.tools import FunctionTool
from pydantic import BaseModel

class SearchInput(BaseModel):
    query: str
    max_results: int = 10

def search_web(input: SearchInput) -> list[SearchResult]:
    return search_client.search(input.query, input.max_results)

tool = FunctionTool.from_defaults(
    fn=search_web,
    name="web_search",
    description="Search the web for current information"
)

QueryEngineTool (RAG)

from llama_index.core.tools import QueryEngineTool

tool = QueryEngineTool.from_defaults(
    query_engine=index.as_query_engine(),
    name="doc_search",
    description="Search internal knowledge base"
)

Tool Specs (Pre-built Tool Sets)

from llama_index.agent import OpenAIAgent
from llama_index.tools import (
    GmailToolSpec,
    GoogleCalendarToolSpec,
    SlackToolSpec,
    NotionToolSpec,
    JiraToolSpec,
    SalesforceToolSpec
)

# Gmail tools
gmail_spec = GmailToolSpec()
gmail_tools = gmail_spec.to_tool_list()

agent = OpenAIAgent.from_tools(gmail_tools, llm=llm)

Memory & Context

Chat Memory

from llama_index.core.memory import ChatMemoryBuffer

memory = ChatMemoryBuffer.from_defaults(
    token_limit=3000,
    tokenizer_fn=llm.tokenizer
)

agent = ReActAgent.from_tools(
    tools=tools,
    llm=llm,
    memory=memory
)

Long-term Memory (Vector Store)

from llama_index.core.memory import VectorStoreMemory
from llama_index.vector_stores import PineconeVectorStore

vector_store = PineconeVectorStore(index_name="agent-memory")
memory = VectorStoreMemory.from_defaults(
    vector_store=vector_store,
    embed_model=embed_model,
    retriever_kwargs={"similarity_top_k": 5}
)

agent = ReActAgent.from_tools(
    tools=tools,
    llm=llm,
    memory=memory
)

Evaluation

Agent Evaluation

from llama_index.core.evaluation import (
    CorrectnessEvaluator,
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    PairwiseComparisonEvaluator
)

evaluator = CorrectnessEvaluator(llm=llm)
result = evaluator.evaluate(
    query="What is AI?",
    response="AI is artificial intelligence...",
    reference="Artificial Intelligence refers to..."
)
print(result.score)  # 0-1
print(result.feedback)

Batch Evaluation

from llama_index.core.evaluation import BatchEvalRunner

evaluators = {
    "correctness": CorrectnessEvaluator(llm),
    "faithfulness": FaithfulnessEvaluator(llm),
    "relevancy": RelevancyEvaluator(llm)
}

runner = BatchEvalRunner(evaluators, workers=4)
results = await runner.aevaluate_queries(
    queries=test_queries,
    agents=[agent1, agent2]
)

LlamaIndex Integration

Full Pipeline

from llama_index.core import (
    SimpleDirectoryReader,
    VectorStoreIndex,
    Settings
)
from llama_index.llms import OpenAI
from llama_index.embeddings import OpenAIEmbedding

# Configure
Settings.llm = OpenAI(model="gpt-4o")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512
Settings.chunk_overlap = 50

# Ingest
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)

# Query engine
query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="tree_summarize"
)

# Agent with RAG
rag_tool = QueryEngineTool.from_defaults(query_engine=query_engine)
agent = ReActAgent.from_tools([rag_tool, web_search], llm=Settings.llm)

Advanced Retrieval

from llama_index.core.retrievers import (
    VectorIndexRetriever,
    KeywordTableRetriever,
    AutoMergingRetriever
)
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    KeywordNodePostprocessor,
    LongContextReorder
)

# Hybrid retrieval
retriever = VectorIndexRetriever(index=index, similarity_top_k=10)
retriever = AutoMergingRetriever(retriever, ...)

query_engine = RetrieverQueryEngine.from_args(
    retriever,
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7),
        KeywordNodePostprocessor(required_keywords=["AI"]),
        LongContextReorder()
    ]
)

LlamaCloud (Managed)

from llama_index import LlamaCloud

# Managed indexing, retrieval, agents
cloud = LlamaCloud(api_key="...")

# Managed index
index = cloud.index("my-index")

# Managed agent
agent = cloud.agent("my-agent")
result = await agent.run("Query")

Deployment

FastAPI

from fastapi import FastAPI
from llama_index.core.workflow import Workflow

app = FastAPI()
workflow = MyWorkflow()

@app.post("/run")
async def run_workflow(request: Request):
    data = await request.json()
    handler = workflow.run(**data)
    async for event in handler.stream_events():
        if isinstance(event, StopEvent):
            return {"result": event.result}

Streaming

@app.post("/stream")
async def stream_workflow(request: Request):
    data = await request.json()
    handler = workflow.run(**data)

    async def stream():
        async for event in handler.stream_events():
            yield f"data: {event.json()}\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

Related Notes

• Framework Comparison — vs LangGraph, AutoGen
• RAG Fundamentals — RAG foundations
• Advanced RAG — Advanced retrieval
• LangGraph — Alternative for workflows
• AutoGen — Alternative for multi-agent
• Memory Systems — Memory patterns
• Agent Benchmarks — LlamaIndex evaluators
• Observability — Observability

Sources: LlamaIndex docs (docs.llamaindex.ai), LlamaIndex GitHub, LlamaIndex workflow docs, LlamaCloud docs, agent examples, evaluation docs.