一、什么是 Agentic RAG？

1.1 传统 RAG vs Agentic RAG

1.2 Agentic RAG 的核心特征

1.3 Agentic RAG 架构设计

二、Agentic RAG 实战实现

2.1 使用 LlamaIndex 构建 Agentic RAG

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.llms.openai import OpenAI

# 1. 创建多个知识库
documents_product = SimpleDirectoryReader("./data/products").load_data()
documents_policy = SimpleDirectoryReader("./data/policies").load_data()

# 2. 构建索引
index_product = VectorStoreIndex.from_documents(documents_product)
index_policy = VectorStoreIndex.from_documents(documents_policy)

# 3. 创建查询引擎
query_engine_product = index_product.as_query_engine(similarity_top_k=3)
query_engine_policy = index_policy.as_query_engine(similarity_top_k=3)

# 4. 定义工具
tools = [
    QueryEngineTool(
        query_engine=query_engine_product,
        metadata=ToolMetadata(
            name="product_knowledge",
            description="包含产品功能、技术规格、使用说明等信息"
        )
    ),
    QueryEngineTool(
        query_engine=query_engine_policy,
        metadata=ToolMetadata(
            name="policy_knowledge",
            description="包含退换货政策、保修条款、隐私政策等信息"
        )
    )
]

# 5. 创建 Agentic RAG Agent
llm = OpenAI(model="gpt-4")
agent = ReActAgent.from_tools(
    tools=tools,
    llm=llm,
    verbose=True,
    max_iterations=5
)

# 6. 执行查询
response = agent.chat(
    "这个产品支持退货吗？退货需要什么条件？"
)
print(response)

执行流程说明：

2.2 高级检索策略

🎯 多策略检索

from llama_index.core.retrievers import (
    VectorIndexRetriever,
    KeywordTableRetriever,
    BM25Retriever
)
from llama_index.core.query_engine import RetrieverQueryEngine

class MultiStrategyRetrieval:
    def __init__(self, index):
        # 1. 向量检索
        self.vector_retriever = VectorIndexRetriever(
            index=index,
            similarity_top_k=5
        )
        
        # 2. 关键词检索
        self.keyword_retriever = KeywordTableRetriever(
            index=index
        )
        
        # 3. BM25 检索
        self.bm25_retriever = BM25Retriever.from_defaults(
            index=index,
            similarity_top_k=5
        )
    
    async def hybrid_retrieve(self, query: str):
        """混合检索策略"""
        # 并行执行多种检索
        vector_results = await self.vector_retriever.aretrieve(query)
        keyword_results = await self.keyword_retriever.aretrieve(query)
        bm25_results = await self.bm25_retriever.aretrieve(query)
        
        # 结果融合（使用 Reciprocal Rank Fusion）
        merged_results = self.reciprocal_rank_fusion([
            vector_results,
            keyword_results,
            bm25_results
        ])
        
        return merged_results
    
    def reciprocal_rank_fusion(self, result_lists, k=60):
        """倒数排名融合算法"""
        scores = {}
        for results in result_lists:
            for rank, result in enumerate(results):
                doc_id = result.node_id
                if doc_id not in scores:
                    scores[doc_id] = 0
                scores[doc_id] += 1 / (rank + k)
        
        # 按分数排序
        sorted_results = sorted(
            scores.items(),
            key=lambda x: x[1],
            reverse=True
        )
        return sorted_results

🔄 查询重写与分解

from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool

class QueryDecomposition:
    def __init__(self, llm, query_engines):
        self.llm = llm
        self.query_engines = query_engines
    
    def create_sub_question_engine(self):
        """创建子问题查询引擎"""
        # 将查询引擎包装为工具
        tools = [
            QueryEngineTool(
                query_engine=engine,
                metadata=ToolMetadata(
                    name=f"engine_{i}",
                    description=desc
                )
            )
            for i, (engine, desc) in enumerate(self.query_engines)
        ]
        
        # 创建子问题引擎
        query_engine = SubQuestionQueryEngine.from_defaults(
            query_engine_tools=tools,
            llm=self.llm,
            verbose=True
        )
        
        return query_engine

# 使用示例
query_engines = [
    (query_engine_product, "产品技术信息"),
    (query_engine_policy, "政策和条款"),
    (query_engine_faq, "常见问题")
]

decomposer = QueryDecomposition(llm, query_engines)
engine = decomposer.create_sub_question_engine()

# 复杂查询会被自动分解
response = engine.query(
    "对比产品A和产品B的性能差异，并说明各自的保修政策"
)

查询分解示意：

三、多 AI Agent 系统（Multi-Agent System）

3.1 为什么需要多 Agent？

单一 Agent 的局限性：

• ❌ 难以处理极复杂任务
• ❌ 专业领域知识不足
• ❌ 单点故障风险
• ❌ 扩展性受限

多 Agent 系统的优势：

• ✅ 任务并行处理
• ✅ 专业分工明确
• ✅ 容错能力强
• ✅ 易于扩展

3.2 多 Agent 架构模式

3.3 使用 AutoGen 构建多 Agent 系统

示例 1：协作式多 Agent

import autogen

# 配置 LLM
config_list = [{
    "model": "gpt-4",
    "api_key": "YOUR_API_KEY"
}]

llm_config = {
    "config_list": config_list,
    "temperature": 0.7
}

# 1. 创建用户代理
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "coding"}
)

# 2. 创建研究员 Agent
researcher = autogen.AssistantAgent(
    name="Researcher",
    llm_config=llm_config,
    system_message="""
    你是一个专业的研究员。你的职责是：
    1. 收集相关信息和数据
    2. 进行初步分析
    3. 提供研究报告
    """
)

# 3. 创建数据分析师 Agent
analyst = autogen.AssistantAgent(
    name="Analyst",
    llm_config=llm_config,
    system_message="""
    你是一个数据分析师。你的职责是：
    1. 分析研究员提供的数据
    2. 发现趋势和模式
    3. 提供数据洞察
    """
)

# 4. 创建报告撰写者 Agent
writer = autogen.AssistantAgent(
    name="Writer",
    llm_config=llm_config,
    system_message="""
    你是一个技术写作专家。你的职责是：
    1. 整合研究和分析结果
    2. 撰写清晰、专业的报告
    3. 确保内容逻辑连贯
    """
)

# 5. 创建协调者
coordinator = autogen.AssistantAgent(
    name="Coordinator",
    llm_config=llm_config,
    system_message="""
    你是项目协调者。你的职责是：
    1. 分配任务给各个专家
    2. 协调工作流程
    3. 整合最终结果
    """
)

# 6. 创建群聊
group_chat = autogen.GroupChat(
    agents=[user_proxy, researcher, analyst, writer, coordinator],
    messages=[],
    max_round=10
)

manager = autogen.GroupChatManager(
    groupchat=group_chat,
    llm_config=llm_config
)

# 7. 启动任务
user_proxy.initiate_chat(
    manager,
    message="分析2024年AI Agent市场趋势，并撰写详细报告"
)

执行流程：

示例 2：竞争式多 Agent

# 创建多个方案生成 Agent
planner_a = autogen.AssistantAgent(
    name="PlannerA",
    llm_config=llm_config,
    system_message="你倾向于激进的创新方案"
)

planner_b = autogen.AssistantAgent(
    name="PlannerB",
    llm_config=llm_config,
    system_message="你倾向于稳健的保守方案"
)

planner_c = autogen.AssistantAgent(
    name="PlannerC",
    llm_config=llm_config,
    system_message="你倾向于平衡的折中方案"
)

# 创建评审 Agent
judge = autogen.AssistantAgent(
    name="Judge",
    llm_config=llm_config,
    system_message="""
    你是方案评审专家。评估标准：
    1. 可行性（40%）
    2. 创新性（30%）
    3. 成本效益（30%）
    
    选出最优方案并说明理由。
    """
)

# 竞争流程
group_chat = autogen.GroupChat(
    agents=[user_proxy, planner_a, planner_b, planner_c, judge],
    messages=[],
    max_round=8
)

3.4 多 Agent 通信协议

四、AI Agent 的自我改进机制

4.1 Agent 记忆系统

4.2 实现自我改进的 Agent

from datetime import datetime
import json

class SelfImprovingAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.memory = {
            "successful_patterns": [],
            "failed_patterns": [],
            "user_feedback": []
        }
        self.performance_metrics = {
            "success_rate": 0.0,
            "avg_response_time": 0.0,
            "user_satisfaction": 0.0
        }
    
    async def run_with_learning(self, user_input: str):
        """执行任务并学习"""
        start_time = datetime.now()
        
        # 1. 检索相似历史案例
        similar_cases = self.retrieve_similar_cases(user_input)
        
        # 2. 应用历史经验
        context = self.build_context_from_history(similar_cases)
        
        # 3. 执行任务
        try:
            result = await self.agent.arun(
                user_input,
                context=context
            )
            success = True
        except Exception as e:
            result = str(e)
            success = False
        
        # 4. 记录执行结果
        execution_time = (datetime.now() - start_time).total_seconds()
        self.record_execution(
            user_input=user_input,
            result=result,
            success=success,
            execution_time=execution_time
        )
        
        # 5. 更新性能指标
        self.update_metrics()
        
        # 6. 自我反思与改进
        if not success:
            await self.reflect_and_improve(user_input, result)
        
        return result
    
    def record_execution(
        self,
        user_input: str,
        result: str,
        success: bool,
        execution_time: float
    ):
        """记录执行历史"""
        record = {
            "timestamp": datetime.now().isoformat(),
            "input": user_input,
            "result": result,
            "success": success,
            "execution_time": execution_time
        }
        
        if success:
            self.memory["successful_patterns"].append(record)
        else:
            self.memory["failed_patterns"].append(record)
        
        # 保存到向量数据库
        self.save_to_vector_db(record)
    
    async def reflect_and_improve(
        self,
        failed_input: str,
        error_message: str
    ):
        """反思失败并改进"""
        reflection_prompt = f"""
        任务失败分析：
        
        输入: {failed_input}
        错误: {error_message}
        
        请分析：
        1. 失败的根本原因是什么？
        2. 应该如何改进策略？
        3. 需要添加什么新工具或知识？
        
        基于历史成功案例:
        {json.dumps(self.memory["successful_patterns"][-5:], indent=2)}
        """
        
        reflection = await self.llm.agenerate(reflection_prompt)
        
        # 提取改进建议并应用
        self.apply_improvements(reflection)
    
    def retrieve_similar_cases(self, query: str, top_k: int = 3):
        """检索相似历史案例"""
        # 使用向量搜索找到相似案例
        similar = []
        # ... 向量搜索实现
        return similar
    
    def build_context_from_history(self, similar_cases):
        """从历史案例构建上下文"""
        context = "相关历史经验:\n"
        for case in similar_cases:
            context += f"- 输入: {case['input']}\n"
            context += f"  策略: {case.get('strategy', 'N/A')}\n"
            context += f"  结果: {'成功' if case['success'] else '失败'}\n\n"
        return context
    
    def update_metrics(self):
        """更新性能指标"""
        total = len(self.memory["successful_patterns"]) + \
                len(self.memory["failed_patterns"])
        
        if total > 0:
            self.performance_metrics["success_rate"] = \
                len(self.memory["successful_patterns"]) / total

4.3 反思（Reflexion）模式

Reflexion 实现示例：

class ReflexionAgent:
    def __init__(self, llm, max_iterations=3):
        self.llm = llm
        self.max_iterations = max_iterations
        self.reflections = []
    
    async def run_with_reflexion(self, task: str):
        """使用反思模式执行任务"""
        for iteration in range(self.max_iterations):
            print(f"\n=== 尝试 {iteration + 1} ===")
            
            # 1. 执行任务
            result = await self.execute_task(task)
            
            # 2. 评估结果
            is_successful, feedback = await self.evaluate_result(
                task, result
            )
            
            if is_successful:
                print("✅ 任务成功完成!")
                return result
            
            # 3. 生成反思
            reflection = await self.generate_reflection(
                task=task,
                attempt=result,
                feedback=feedback,
                previous_reflections=self.reflections
            )
            
            self.reflections.append(reflection)
            print(f"💭 反思: {reflection}")
        
        print("❌ 达到最大尝试次数")
        return result
    
    async def generate_reflection(
        self,
        task: str,
        attempt: str,
        feedback: str,
        previous_reflections: list
    ) -> str:
        """生成反思"""
        prompt = f"""
        任务: {task}
        
        当前尝试结果:
        {attempt}
        
        评估反馈:
        {feedback}
        
        之前的反思:
        {chr(10).join(previous_reflections)}
        
        请深入反思:
        1. 这次尝试哪里做得不好？
        2. 根本问题是什么？
        3. 下次应该如何改进？
        
        提供具体、可执行的改进建议。
        """
        
        reflection = await self.llm.agenerate(prompt)
        return reflection
    
    async def execute_task(self, task: str):
        """执行任务（包含历史反思）"""
        context = "\n".join([
            "历史经验和反思:",
            *self.reflections
        ])
        
        full_prompt = f"{context}\n\n当前任务: {task}"
        result = await self.agent.arun(full_prompt)
        return result

五、高级技术实战案例

5.1 构建智能客服 Agent（综合案例）

from llama_index.core import VectorStoreIndex
from llama_index.core.agent import ReActAgent
import autogen

class IntelligentCustomerService:
    """智能客服系统 - 综合 Agentic RAG + Multi-Agent"""
    
    def __init__(self):
        # 1. 知识库（Agentic RAG）
        self.knowledge_bases = self.setup_knowledge_bases()
        
        # 2. 专业 Agents（Multi-Agent）
        self.agents = self.setup_agents()
        
        # 3. 记忆系统
        self.memory = SelfImprovingAgent()
    
    def setup_knowledge_bases(self):
        """设置多个知识库"""
        return {
            "product": VectorStoreIndex.from_documents(
                load_documents("./data/products")
            ),
            "policy": VectorStoreIndex.from_documents(
                load_documents("./data/policies")
            ),
            "faq": VectorStoreIndex.from_documents(
                load_documents("./data/faq")
            )
        }
    
    def setup_agents(self):
        """设置专业 Agents"""
        return {
            "triage": autogen.AssistantAgent(
                name="TriageAgent",
                system_message="你负责分析用户问题类型并分发"
            ),
            "product_expert": autogen.AssistantAgent(
                name="ProductExpert",
                system_message="你是产品专家，回答产品相关问题"
            ),
            "policy_expert": autogen.AssistantAgent(
                name="PolicyExpert",
                system_message="你是政策专家，处理退换货等政策问题"
            ),
            "technical_support": autogen.AssistantAgent(
                name="TechnicalSupport",
                system_message="你是技术支持，解决技术问题"
            )
        }
    
    async def handle_customer_query(self, query: str):
        """处理客户查询"""
        # 1. 问题分类
        category = await self.classify_query(query)
        
        # 2. 检索相关知识
        knowledge = await self.retrieve_knowledge(query, category)
        
        # 3. 分配给专业 Agent
        expert = self.agents[category]
        response = await expert.arun(
            query,
            context=knowledge
        )
        
        # 4. 记录并学习
        await self.memory.record_and_learn(query, response)
        
        return response

5.2 性能对比

六、总结与最佳实践

✅ 技术选型建议