LlamaIndex 2026：从 RAG 框架到 Agent 平台

┌──────────────────────────────────────────────────┐
│              Agent Layer                          │
│  Workflows │ Data Agents │ Multi-Agent │ Tools   │
├──────────────────────────────────────────────────┤
│              Orchestration Layer                  │
│  Query Engine │ Router │ Planner │ Evaluator     │
├──────────────────────────────────────────────────┤
│              Data Layer                           │
│  ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│  │ Ingestion│ │ Indexing │ │ Retrieval        │ │
│  │ Pipeline │ │ (多索引) │ │ (混合检索)       │ │
│  └──────────┘ └──────────┘ └──────────────────┘ │
├──────────────────────────────────────────────────┤
│              Integration Layer                    │
│  50+ LLM │ 30+ Vector DB │ 100+ Data Source     │
└──────────────────────────────────────────────────┘

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

class ResearchEvent(Event):
    query: str
    sources: list

class AnalysisEvent(Event):
    data: dict
    confidence: float

class ReportEvent(Event):
    report: str
    metadata: dict

class ResearchWorkflow(Workflow):
    @step
    async def research(self, ctx: Context, ev: StartEvent) -> ResearchEvent:
        query = ev.query
        # 并行从多个数据源获取数据
        results = await asyncio.gather(
            web_search(query),
            academic_search(query),
            knowledge_base_query(query)
        )
        return ResearchEvent(
            query=query,
            sources=list(itertools.chain(*results))
        )
    
    @step
    async def analyze(self, ctx: Context, ev: ResearchEvent) -> AnalysisEvent:
        # 使用 LLM 分析收集到的数据
        analysis = await self.llm.analyze(
            sources=ev.sources,
            instructions="提取关键发现和数据点"
        )
        
        # 评估分析质量
        confidence = await self.evaluator.evaluate(analysis, ev.sources)
        
        if confidence < 0.7:
            # 置信度低，重新搜索
            ctx.send_event(ResearchEvent(
                query=f"{ev.query} 补充数据",
                sources=ev.sources
            ))
        
        return AnalysisEvent(data=analysis, confidence=confidence)
    
    @step
    async def report(self, ctx: Context, ev: AnalysisEvent) -> ReportEvent:
        report = await self.writer.generate(
            data=ev.data,
            template="research_report",
            format="markdown"
        )
        return ReportEvent(report=report, metadata={"confidence": ev.confidence})
    
    @step
    async def finalize(self, ctx: Context, ev: ReportEvent) -> StopEvent:
        return StopEvent(result=ev.report)

# 运行 Workflow
workflow = ResearchWorkflow(timeout=300)
result = await workflow.run(query="2026 年 AGI 发展趋势分析")

from llama_index.agent import DataAgent
from llama_index.core.tools import ToolSpec

# 创建数据管理 Agent
agent = DataAgent(
    tools=[
        ToolSpec(name="sql_query", fn=execute_sql, 
                 description="执行 SQL 查询"),
        ToolSpec(name="csv_export", fn=export_csv,
                 description="导出数据为 CSV"),
        ToolSpec(name="chart_gen", fn=generate_chart,
                 description="生成数据图表"),
        ToolSpec(name="data_merge", fn=merge_datasets,
                 description="合并多个数据集"),
        ToolSpec(name="stat_analysis", fn=statistical_analysis,
                 description="统计分析"),
    ],
    llm="gpt-4o",
    system_prompt="""
    你是一个数据分析 Agent。你可以查询数据库、生成图表、
    执行统计分析。始终先理解用户的分析意图，再选择合适的工具。
    """,
    max_iterations=15,
    verbose=True
)

# Agent 自主决定调用哪些工具
result = await agent.achat(
    "分析过去 3 个月的销售数据，找出趋势并生成可视化报告"
)

# Agent 执行流程：
# 1. sql_query: 查询 3 个月销售数据
# 2. stat_analysis: 计算趋势和环比
# 3. chart_gen: 生成趋势图
# 4. csv_export: 导出原始数据

from llama_index.core import VectorStoreIndex, KnowledgeGraphIndex
from llama_index.core.schema import NodeWithScore
from llama_index.core.retrievers import (
    VectorIndexRetriever,
    KGTableRetriever,
    HybridFusionRetriever
)
from llama_index.core.postprocessor import (
    SentenceTransformerRerank,
    LongContextReorder,
    PIIRedactor
)

# 多索引混合检索
vector_index = VectorStoreIndex.from_documents(
    documents, 
    embed_model="text-embedding-3-large",
    transformations=[
        SemanticSplitterNodeParser(
            buffer_size=1, 
            breakpoint_percentile_threshold=95
        ),
        # 元数据提取
        MetadataExtractor(
            extractors=[
                TitleExtractor(nodes=5),
                QuestionsAnsweredExtractor(questions=3),
                SummaryExtractor(summaries=["prev", "self"])
            ]
        )
    ]
)

kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    max_triplets_per_chunk=10,
    include_embeddings=True,
    kg_extract_template="entity_relation_v2"
)

# 混合检索
hybrid_retriever = HybridFusionRetriever(
    retrievers=[
        VectorIndexRetriever(index=vector_index, similarity_top_k=20),
        KGTableRetriever(index=kg_index, retriever_mode="hybrid", top_k=15)
    ],
    fusion_mode="reciprocal_rerank",
    num_queries=3  # 查询扩展
)

# 后处理管道
query_engine = RetrieverQueryEngine.from_args(
    hybrid_retriever,
    node_postprocessors=[
        SentenceTransformerRerank(
            model="BAAI/bge-reranker-v2-m3",
            top_n=5
        ),
        LongContextReorder(),  # 重新排序避免中间遗忘
        PIIRedactor(),         # PII 脱敏
    ],
    response_synthesizer_mode="tree_summarize",
    streaming=True
)

# 多轮对话 RAG
chat_engine = vector_index.as_chat_engine(
    chat_mode="context",
    memory=ChatMemoryBuffer.from_defaults(token_limit=8000),
    system_prompt="你是一个知识助手，基于提供的上下文回答问题...",
    verbose=True
)

from llama_index.cloud import LlamaCloud

cloud = LlamaCloud(api_key="llama-xxx")

# 创建托管知识库
pipeline = cloud.create_pipeline(
    name="company-knowledge",
    data_sources=[
        {"type": "web", "urls": ["https://company.com/docs"]},
        {"type": "s3", "bucket": "company-docs", "prefix": "policies/"},
        {"type": "confluence", "space": "ENG"},
        {"type": "sharepoint", "site": "company.sharepoint.com"}
    ],
    transformations={
        "chunking": "semantic",
        "embed_model": "text-embedding-3-large",
        "metadata_extraction": True
    },
    retrieval={
        "method": "hybrid",
        "reranking": "bge-reranker-v2-m3",
        "top_k": 10
    }
)

# 自动索引更新
pipeline.schedule_sync(
    frequency="hourly",
    incremental=True
)

# 通过 API 检索
results = pipeline.retrieve(
    query="公司的远程办公政策是什么？",
    top_k=5,
    filters={"department": "HR", "year": "2026"}
)