LlamaIndex Agent 评测：从 RAG 到 Agent 的进化

from llama_index.core.agent import FunctionAgent
from llama_index.llms.openai import OpenAI
from llama_index.core.tools import FunctionTool, QueryEngineTool

llm = OpenAI(model="gpt-4o")

# 工具1：网络搜索
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    from llama_index.tools.tavilyi import TavilyToolSpec
    return TavilyToolSpec().search(query)

web_tool = FunctionTool.from_defaults(fn=search_web)

# 工具2：RAG 查询（复用现有索引）
from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=5)

rag_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="knowledge_base",
    description="查询内部知识库，包含产品文档和技术规范"
)

# 创建 Data Agent
agent = FunctionAgent(
    tools=[web_tool, rag_tool],
    llm=llm,
    system_prompt="你是一个研究助手。先查内部知识库，找不到再搜网络。"
)

response = await agent.run("我们的API限流策略是什么？和业界最佳实践比如何？")
# Agent会: 1. 先用rag_tool查内部  2. 再用web_tool搜业界实践  3. 综合回答

AgentRunner（外层控制）
  └── AgentWorker（执行单元）

from llama_index.core.agent import ReActAgent

# ReActAgent = AgentRunner + ReActAgentWorker
agent = ReActAgent.from_tools(
    tools=[search_tool, rag_tool, calculator_tool],
    llm=llm,
    verbose=True,
    max_iterations=10,
    # 自定义 AgentWorker 行为
    react_chat_history=True
)

# 也可以分别使用
from llama_index.core.agent import AgentRunner
from llama_index.core.agent import ReActAgentWorker

worker = ReActAgentWorker.from_tools(
    tools=[search_tool, rag_tool],
    llm=llm,
    max_iterations=10
)
agent = AgentRunner(worker)

response = agent.chat("分析我们产品文档中提到的所有API，给出分类表")

from llama_index.core.workflow import (
    Workflow, step, Event, Context
)

class ResearchEvent(Event):
    query: str

class DraftEvent(Event):
    research_data: str

class ReviewEvent(Event):
    draft: str

class ResearchWorkflow(Workflow):
    @step
    async def research(self, ctx: Context, ev: ResearchEvent) -> DraftEvent:
        results = await rag_tool.run(ev.query)
        web_results = await search_tool.run(ev.query)
        combined = f"内部知识: {results}\n网络资料: {web_results}"
        return DraftEvent(research_data=combined)
    
    @step
    async def draft(self, ctx: Context, ev: DraftEvent) -> ReviewEvent:
        draft = await llm.acomplete(f"基于以下资料写报告:\n{ev.research_data}")
        return ReviewEvent(draft=draft.text)
    
    @step
    async def review(self, ctx: Context, ev: ReviewEvent) -> str:
        review = await llm.acomplete(f"审核以下报告，给出修改意见或通过:\n{ev.draft}")
        return review.text

workflow = ResearchWorkflow(timeout=120)
result = await workflow.run(ResearchEvent(query="AI Agent框架对比分析"))

# Agent 自主决定是否需要检索、检索几次、如何组合结果
agent = FunctionAgent(
    tools=[rag_tool_multi_index],  # 多个知识库索引
    llm=llm,
    system_prompt="""你是知识库研究助手。
对每个问题：
1. 先判断需要查哪些知识库
2. 分别检索，评估结果质量
3. 如果信息不足，换关键词重新检索
4. 综合所有信息回答"""
)

from llama_index.core.tools import QueryEngineTool

# 三个不同领域的索引
product_index = VectorStoreIndex.from_documents(product_docs)
api_index = VectorStoreIndex.from_documents(api_docs)
faq_index = VectorStoreIndex.from_documents(faq_docs)

tools = [
    QueryEngineTool.from_defaults(
        query_engine=product_index.as_query_engine(),
        name="product_docs",
        description="产品功能文档和用户指南"
    ),
    QueryEngineTool.from_defaults(
        query_engine=api_index.as_query_engine(),
        name="api_docs", 
        description="API参考文档，含请求/响应格式"
    ),
    QueryEngineTool.from_defaults(
        query_engine=faq_index.as_query_engine(),
        name="faq",
        description="常见问题解答"
    ),
]

agent = FunctionAgent(
    tools=tools,
    llm=llm,
    system_prompt="根据问题选择合适的知识库查询，可同时查多个。"
)

def iterative_research(query: str, max_rounds: int = 3) -> str:
    """Agent 迭代检索：每轮根据前一轮结果决定下一步"""
    agent = FunctionAgent(
        tools=[rag_tool, web_tool],
        llm=llm,
        system_prompt=f"""执行迭代研究，最多{max_rounds}轮：
每轮：1.回顾已知信息 2.找出信息缺口 3.检索补充 4.判断是否足够
信息足够时输出 FINAL: 答案"""
    )
    return agent.chat(query)

# 混合方案：LangGraph 编排 + LlamaIndex 检索
from llama_index.core import VectorStoreIndex
from langgraph.graph import StateGraph

# LlamaIndex 负责检索
index = VectorStoreIndex.from_documents(docs)
retriever = index.as_retriever()

# LangGraph 负责编排
def retrieve_node(state):
    docs = retriever.retrieve(state["query"])
    return {"context": [d.text for d in docs]}

def generate_node(state):
    response = llm.invoke(f"Context: {state['context']}\nQ: {state['query']}")
    return {"answer": response.content}