LlamaIndex 2026：从RAG框架到Agent平台的转型

传统RAG：用户提问 → 检索 → 生成回答
Agent化：用户提问 → 理解意图 → 规划任务 → 多步检索/行动 → 综合回答

from llama_index.core.agent.workflow import (
    AgentWorkflow,
    FunctionAgent,
    ReActAgent,
    RouterAgent
)
from llama_index.core.tools import QueryEngineTool, ToolMetadata

# 创建RAG工具
stock_engine = create_stock_query_engine()    # 股票数据引擎
news_engine = create_news_query_engine()      # 新闻数据引擎
report_engine = create_report_query_engine()  # 财报数据引擎

# 定义工具
tools = [
    QueryEngineTool(
        query_engine=stock_engine,
        metadata=ToolMetadata(
            name="stock_data",
            description="查询股票实时行情和历史数据"
        )
    ),
    QueryEngineTool(
        query_engine=news_engine,
        metadata=ToolMetadata(
            name="news_data",
            description="搜索财经新闻和公告"
        )
    ),
    QueryEngineTool(
        query_engine=report_engine,
        metadata=ToolMetadata(
            name="report_data",
            description="查询上市公司财报数据"
        )
    )
]

# 创建专业Agent
analyst_agent = ReActAgent(
    name="分析Agent",
    description="负责数据分析和技术指标计算",
    tools=[stock_engine, report_engine],
    llm=OpenAI(model="gpt-4o"),
    system_prompt="你是一位专业的金融分析师..."
)

research_agent = ReActAgent(
    name="研究Agent",
    description="负责新闻搜集和信息检索",
    tools=[news_engine],
    llm=OpenAI(model="gpt-4o"),
    system_prompt="你是一位财经研究员..."
)

writer_agent = FunctionAgent(
    name="撰写Agent",
    description="负责综合分析报告撰写",
    llm=OpenAI(model="gpt-4o"),
    system_prompt="你是一位专业的金融报告撰写人..."
)

# 创建工作流
workflow = AgentWorkflow(
    agents=[analyst_agent, research_agent, writer_agent],
    root_agent=analyst_agent,  # 主入口Agent
    max_steps=20,              # 最大执行步数
    max_time=180,              # 最大执行时间（秒）
)

# 执行
response = await workflow.run(
    "分析贵州茅台2025年年报，结合近期新闻，给出投资建议"
)

from llama_index.agent.data import DataAgent

# 创建数据代理
data_agent = DataAgent(
    data_sources=[
        "postgresql://user:pass@db/mydb",       # 数据库
        "s3://my-bucket/data/",                 # S3存储
        "https://api.example.com/data",          # API
        "./documents/",                          # 本地文档
    ],
    
    # 数据处理能力
    capabilities=[
        "query",        # 查询数据
        "analyze",      # 分析数据
        "visualize",    # 可视化
        "export",       # 导出数据
        "transform",    # 数据转换
    ],
    
    # 自动schema发现
    auto_discover_schema=True,
    
    # 查询优化
    query_optimizer=True,  # 自动优化SQL查询
    
    llm=OpenAI(model="gpt-4o")
)

# 自然语言数据查询
result = await data_agent.run(
    "对比2024和2025年Q4各产品线的销售额，生成柱状图"
)

┌──────────────────────────────────────────┐
│           Data Agent 能力层               │
├──────────────────────────────────────────┤
│                                          │
│  查询层：NL2SQL / NL2API / NL2File      │
│  ↓                                        │
│  分析层：统计 / 预测 / 异常检测          │
│  ↓                                        │
│  可视化层：图表 / 仪表盘 / 报告          │
│  ↓                                        │
│  导出层：CSV / Excel / PDF / API         │
│                                          │
└──────────────────────────────────────────┘

from llama_index.core import MultiModalIndex, MultiModalRetriever

# 创建多模态索引
index = MultiModalIndex.from_documents(
    documents=[
        TextDocument(text="2025年度财务报告..."),
        ImageDocument(image_path="charts/revenue.png"),
        TableDocument(data=quarterly_data),
        VideoDocument(video_path="product_demo.mp4"),
    ],
    embed_models={
        "text": OpenAIEmbedding(model="text-embedding-3-small"),
        "image": ClipEmbedding(model="clip-vit-large"),
        "table": TableEmbedding(model="tapa-large"),
        "video": VideoEmbedding(model="video-clip-v2"),
    }
)

# 多模态检索
retriever = MultiModalRetriever.from_index(index)
results = retriever.retrieve("收入增长趋势")

# results 包含：相关文本段落 + 图表图片 + 数据表格 + 视频片段

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

class AnalyzeEvent(Event):
    data: str

class ReportEvent(Event):
    analysis: str

class MyWorkflow(Workflow):
    @step
    async def analyze(self, ctx: Context, ev: AnalyzeEvent) -> ReportEvent:
        # 分析逻辑
        result = await llm.acomplete(f"分析: {ev.data}")
        return ReportEvent(analysis=result.text)
    
    @step
    async def report(self, ctx: Context, ev: ReportEvent) -> str:
        # 生成报告
        report = await llm.acomplete(f"基于分析生成报告: {ev.analysis}")
        return report.text

# 执行
workflow = MyWorkflow(timeout=120)
result = await workflow.run(event=AnalyzeEvent(data="..."))

用户请求 → CrewAI（分配任务）→ LangGraph（控制流程）→ LlamaIndex（数据检索）