记忆点:让 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%。"
