Skip to content
🔗 分享本题
查看我的学习进度 →

Agentic RAG 动漫知识图:Agent 规划子问题、选择检索工具、检查证据并循环改写,系统控制工具权限、预算和停止条件

记忆点:让 Agent 决定怎么找,但系统决定能找什么、何时停。

💡 答案要点

Agentic RAG = Agent主动控制检索策略,而非被动执行单次检索

普通RAG vs Agentic RAG

普通RAG(被动):
用户问 → 检索1次 → 固定Top-K → LLM生成
(检索策略固定,不管结果好不好)

Agentic RAG(主动):
用户问 → Agent分析 → 决定是否检索/检索什么/怎么检索
         → 评估结果 → 决定是否再检索
         → 循环直到信息充分 → 生成

核心能力

展开 Python 代码示例(100 行)
python
class AgenticRAG:
    """Agent主动控制检索的RAG系统"""

    def __init__(self, llm, vectordb, search_engine):
        self.llm = llm
        self.vectordb = vectordb
        self.search_engine = search_engine
        self.max_retrieval_rounds = 3

    def run(self, user_query: str) -> str:
        context = []
        retrieval_count = 0

        while retrieval_count < self.max_retrieval_rounds:
            # Step 1: Agent分析当前上下文,决定下一步
            decision = self.agent_decide(user_query, context)

            if decision["action"] == "answer":
                # 信息充分,直接回答
                break

            elif decision["action"] == "retrieve_vector":
                # 语义检索
                query = decision["query"]
                docs = self.vectordb.search(query, k=5)
                context.extend(docs)

            elif decision["action"] == "retrieve_web":
                # 实时网络检索
                query = decision["query"]
                docs = self.search_engine.search(query)
                context.extend(docs)

            elif decision["action"] == "decompose":
                # 分解子问题,分别检索
                sub_questions = decision["sub_questions"]
                for sq in sub_questions:
                    docs = self.vectordb.search(sq, k=3)
                    context.extend(docs)

            retrieval_count += 1

        # 最终生成
        return self.generate(user_query, context)

    def agent_decide(self, query: str, context: list) -> dict:
        """Agent决策:下一步做什么"""

        context_text = "\n".join(context[-5:]) if context else "无"

        prompt = f"""
        你是RAG系统的检索策略Agent。

        用户问题:{query}
        当前已检索到的信息:{context_text}

        请判断下一步操作(输出JSON):

        选项:
        1. {{"action": "answer"}} - 信息已充分,可以回答
        2. {{"action": "retrieve_vector", "query": "检索词"}} - 需要语义检索
        3. {{"action": "retrieve_web", "query": "搜索词"}} - 需要最新网络信息
        4. {{"action": "decompose", "sub_questions": ["子问题1", "子问题2"]}} - 需要分解问题

        判断依据:
        - 当前信息能否回答问题?
        - 是否需要最新数据?
        - 问题是否包含多个子问题?

        输出:
        """

        result = self.llm.generate(prompt, temperature=0)
        return json.loads(result)

    def generate(self, query: str, context: list) -> str:
        """基于收集到的上下文生成最终答案"""
        ctx_text = "\n---\n".join(context)

        prompt = f"""
        基于以下信息回答问题:

        {ctx_text}

        问题:{query}

        请综合所有信息给出完整回答:
        """
        return self.llm.generate(prompt)

# 使用场景示例:复杂多步问题
rag = AgenticRAG(llm, vectordb, search_engine)

result = rag.run("比较GPT-4和Claude 3的价格,哪个更适合做长文档摘要?")

# Agent执行过程:
# 轮次1:retrieve_web("GPT-4最新价格 2024")  → 获取GPT-4价格
# 轮次2:retrieve_web("Claude 3 Opus价格")   → 获取Claude价格
# 轮次3:retrieve_vector("长文档摘要模型对比") → 获取性能对比
# → 信息充分,生成综合对比回答

多跳推理实现

展开 Python 代码示例(41 行)
python
class MultiHopRAG:
    """多跳推理:每次检索结果作为下一次检索的输入"""

    def multi_hop_retrieve(self, query: str, max_hops=3) -> str:
        reasoning_trace = []
        current_query = query
        accumulated_context = []

        for hop in range(max_hops):
            # 检索当前子问题
            docs = self.vectordb.search(current_query, k=3)
            accumulated_context.extend(docs)

            # 评估是否已有足够信息
            eval_prompt = f"""
            原始问题:{query}
            已收集信息:{" ".join(accumulated_context)}

            判断(JSON):
            1. 能直接回答原始问题吗?{{"can_answer": true}}
            2. 还需要追问什么?{{"can_answer": false, "next_question": "下一个子问题"}}
            """

            eval_result = json.loads(self.llm.generate(eval_prompt, temperature=0))

            if eval_result["can_answer"]:
                break  # 信息足够,停止检索

            # 继续追问
            current_query = eval_result["next_question"]
            reasoning_trace.append(f"Hop {hop+1}: {current_query}")

        print("推理链:", " → ".join(reasoning_trace))
        return self.generate(query, accumulated_context)

# 示例:
# 问题:"参与过阿里投资的公司中,谁的CEO毕业于清华?"
# Hop 1: 检索"阿里巴巴投资的公司" → 找到饿了么、优酷等
# Hop 2: 检索"饿了么CEO教育背景" → 张旭豪复旦大学
# Hop 3: 检索"优酷CEO教育背景" → ...
# 逐步推理找到答案

面试话术:

示例表达(仅在能用本人经历或可复现实验佐证时使用): "Agentic RAG让Agent主动决策检索策略,而不是固定执行一次检索。我实现了3种决策:1)信息充分直接回答;2)需要最新信息走实时搜索;3)复杂问题先分解再分别检索。核心是Agent每轮评估'信息是否足够回答问题',不够就继续检索,最多3轮避免无限循环。实测复杂多跳问题准确率从普通RAG的52%提升到Agentic RAG的78%。"