傳統RAG vs Agentic RAG：全面對比分析，為什么說智能代理是下一代AI應用的核心？ 原創 精華
用傳統 RAG 構建 AI 應用，總感覺像在跟一個"健忘癥患者"對話——它能找到相關文檔，卻無法真正理解問題的本質。問它"如何優化 RAG 系統的準確率"，它只會機械地檢索包含"RAG"和"準確率"的文檔，然后拼湊出一個看似合理的答案。但如果問題稍微復雜一點，比如"比較不同 RAG 實現方案的優缺點，并給出在資源受限場景下的最佳實踐"，傳統 RAG 就露怯了。
這就是 Agentic RAG 要解決的核心問題——讓 AI 不僅會"找"，更要會"想"。
從被動到主動
傳統 RAG 的天花板
傳統 RAG 的邏輯很簡單：
用戶提問 → 向量化 → 檢索相似文檔 → 生成答案這個流程有個致命缺陷：它是純被動的。系統不會思考"這個問題真正在問什么"，不會判斷"檢索到的內容是否真的相關"，更不會意識到"可能需要多步推理才能得到答案"。
打個比方，傳統 RAG 像圖書館的檢索系統——你輸入關鍵詞，它返回相關書籍。而 Agentic RAG 更像一個資深的圖書管理員——他會理解你的真實需求，知道該查哪幾本書，還會幫你交叉驗證信息的準確性。
Agentic RAG 的核心理念
Agentic RAG 是動態的。AI 代理主動管理信息獲取方式，將 RAG 整合到推理過程中。它不只是檢索，而是精煉查詢、將 RAG 作為精密工具使用，并隨時間管理信息。這種智能方法讓 AI 代理能更好地適應不斷變化的情況。
核心區別在于引入了自主推理和決策能力：
- 感知（Perceiving）：理解環境和情境
- 推理（Reasoning）：分解復雜問題，制定解決策略
- 規劃（Planning）：創建分步行動計劃
- 行動（Acting）：執行任務，使用各種工具
架構設計：從線性到循環
核心組件解析
Agentic RAG 系統的架構包含幾個關鍵組件，每個組件都扮演著特定的角色，共同構成一個智能的信息處理系統。
1. 推理引擎（Reasoning Engine）
推理引擎是整個系統的"大腦"，它不僅處理邏輯推理，還負責協調其他組件的工作。想象一下，這就像一個經驗豐富的偵探，能夠從蛛絲馬跡中推理出完整的真相。
推理引擎的核心職責包括：
- 意圖理解：深入分析用戶查詢背后的真實需求，而不僅僅是表面的關鍵詞
- 策略制定：根據問題的復雜度和類型，選擇最合適的解決路徑
- 結果評估：判斷當前獲得的信息是否足夠回答問題
- 決策制定：決定是繼續檢索、調整策略還是生成最終答案
class ReasoningEngine:
"""Agentic RAG 的推理引擎
這個類實現了 Agentic RAG 的核心推理邏輯。它維護著對話的記憶，
能夠制定和調整計劃，并通過自我反思不斷改進答案質量。
"""
def __init__(self, llm):
self.llm = llm # 大語言模型，用于執行推理
self.memory = ConversationMemory() # 對話記憶，保存上下文
self.planner = TaskPlanner() # 任務規劃器，負責分解復雜任務
asyncdef reason(self, query: str, context: Dict):
"""推理流程的主函數
這個方法實現了完整的推理循環：
1. 首先理解用戶的真實意圖（不只是字面意思）
2. 基于意圖制定執行計劃（可能包含多個步驟）
3. 逐步執行計劃，并在每一步后驗證結果
4. 如果結果不滿意，會重新調整計劃
5. 最終綜合所有信息生成答案
Args:
query: 用戶的原始查詢
context: 包含歷史對話、用戶偏好等上下文信息
Returns:
綜合推理后的最終答案
"""
# 步驟1：意圖理解 - 分析用戶真正想要什么
# 例如："最新的 AI 論文"可能意味著用戶想了解前沿研究動態
intent = await self.understand_intent(query, context)
# 步驟2：制定計劃 - 將復雜問題分解為可執行的子任務
# 例如：先檢索論文列表，然后篩選最新的，最后總結要點
plan = await self.planner.create_plan(intent)
# 步驟3：執行并驗證 - 這是 Agentic RAG 的核心循環
results = []
for step in plan:
# 執行當前步驟
result = await self.execute_step(step)
# 關鍵：自我反思和驗證
# 這是 Agentic RAG 區別于傳統 RAG 的重要特征
ifnot self.validate_result(result):
# 如果結果不滿意，動態調整計劃
# 可能會添加新的檢索步驟或改變策略
plan = await self.planner.revise_plan(step, result)
results.append(result)
# 步驟4：綜合答案 - 將所有收集的信息整合成連貫的回答
return self.synthesize_answer(results)2. 記憶系統（Memory System）
記憶系統是 Agentic RAG 能夠進行連貫對話和持續學習的基礎。與傳統 RAG 每次都"從零開始"不同，Agentic RAG 能夠記住之前的交互，并從中學習。
記憶系統采用分層設計，模仿人類大腦的記憶機制：
class AgentMemory:
"""分層記憶系統
這個記憶系統模仿人類的記憶結構，包含三種不同類型的記憶：
1. 短期記憶：類似于人類的工作記憶，保存當前對話的即時信息
2. 長期記憶：類似于人類的語義記憶，存儲重要的知識和概念
3. 情景記憶：類似于人類的經驗記憶，記錄成功的問題解決策略
"""
def __init__(self):
# 短期記憶：使用雙端隊列實現，自動丟棄舊的記憶
# maxlen=10 意味著只保留最近的10條交互記錄
# 這避免了上下文過長導致的性能問題
self.short_term = deque(maxlen=10)
# 長期記憶：使用向量數據庫實現，可以永久保存
# 支持語義搜索，能找到概念上相關的記憶
self.long_term = VectorStore()
# 情景記憶：記錄成功的經驗和策略
# 當遇到類似問題時，可以直接復用之前的成功策略
self.episodic = ExperienceBuffer()
def remember(self, interaction):
"""智能記憶存儲
不是所有信息都值得記住。這個方法會：
1. 將所有交互先存入短期記憶
2. 評估信息的重要性
3. 重要信息會被提升到長期記憶
4. 成功的策略會被記錄為經驗
Args:
interaction: 包含查詢、答案、策略等信息的交互記錄
"""
# 所有交互都先進入短期記憶
self.short_term.append(interaction)
# 評估重要性：基于信息量、新穎性、用戶反饋等因素
if self.is_important(interaction):
# 重要信息提升到長期記憶
# 會被向量化以支持語義搜索
self.long_term.add(interaction)
# 記錄成功的策略供未來參考
# 這使得系統能夠從經驗中學習
if interaction.successful:
self.episodic.add_experience(interaction)
def recall(self, query: str, k: int = 5):
"""智能記憶檢索
根據查詢檢索相關記憶，優先級：
1. 短期記憶（最相關的當前上下文）
2. 情景記憶（類似問題的成功經驗）
3. 長期記憶（相關的背景知識）
"""
relevant_memories = []
# 從短期記憶中查找
for memory in self.short_term:
if self.is_relevant(memory, query):
relevant_memories.append(memory)
# 從情景記憶中查找類似經驗
similar_experiences = self.episodic.find_similar(query, k=3)
relevant_memories.extend(similar_experiences)
# 從長期記憶中語義搜索
semantic_memories = self.long_term.search(query, k=k)
relevant_memories.extend(semantic_memories)
return relevant_memories3. 工具系統（Tool System）
工具系統讓 Agentic RAG 能夠執行具體的任務，而不僅僅是生成文本。每個工具都是一個獨立的功能模塊，可以被推理引擎調用來完成特定任務。
工具的設計遵循統一的接口規范，這使得系統可以輕松擴展新的能力：
from typing import List, Callable
from pydantic import BaseModel
class Tool(BaseModel):
"""工具基類
所有工具都必須繼承這個基類，確保接口的一致性。
這種設計讓推理引擎可以用統一的方式調用不同的工具。
"""
name: str # 工具的唯一標識符
description: str # 工具的功能描述，幫助 LLM 理解何時使用
parameters: Dict # 工具需要的參數定義
asyncdef execute(self, **kwargs):
"""執行工具的抽象方法"""
raise NotImplementedError
class SearchTool(Tool):
"""搜索工具
這個工具展示了如何實現一個具體的功能。
它可以搜索互聯網獲取最新信息，這是傳統 RAG 做不到的。
"""
name: str = "web_search"
description: str = "搜索互聯網獲取最新信息，適用于需要實時數據的查詢"
asyncdef execute(self, query: str):
"""執行搜索
這個方法會：
1. 調用搜索 API（如 Google、Bing）
2. 解析返回的結果
3. 提取關鍵信息
4. 返回結構化的數據
"""
# 調用實際的搜索 API
results = await search_api(query)
# 解析和清洗結果
# 這一步很重要，因為原始搜索結果可能包含很多噪音
cleaned_results = self.parse_results(results)
# 返回結構化數據，方便后續處理
return {
"tool": self.name,
"query": query,
"results": cleaned_results,
"timestamp": datetime.now()
}
class DatabaseTool(Tool):
"""數據庫查詢工具
這個工具可以查詢內部數據庫，獲取結構化數據。
這對于回答需要精確數據的問題非常重要。
"""
name: str = "db_query"
description: str = "查詢內部數據庫獲取精確的業務數據"
asyncdef execute(self, sql: str):
"""安全地執行 SQL 查詢
注意：這里需要特別注意 SQL 注入等安全問題
實際使用時應該：
1. 使用參數化查詢
2. 限制查詢權限
3. 添加查詢超時
"""
# 驗證 SQL 安全性
ifnot self.is_safe_query(sql):
raise ValueError("Unsafe SQL query detected")
# 執行查詢
results = await db.execute(sql)
# 格式化結果
return self.format_results(results)
class CalculatorTool(Tool):
"""計算工具
用于執行數學計算，這在處理數據分析類問題時很有用。
"""
name: str = "calculator"
description: str = "執行數學計算，支持基本運算和科學計算"
asyncdef execute(self, expression: str):
"""安全地計算數學表達式
使用 AST（抽象語法樹）來安全地計算表達式，
避免執行任意代碼的安全風險。
"""
import ast
import operator
# 定義允許的操作符
allowed_ops = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow
}
# 安全地解析和計算表達式
try:
tree = ast.parse(expression, mode='eval')
result = self.eval_expr(tree.body, allowed_ops)
return {"result": result, "expression": expression}
except Exception as e:
return {"error": str(e), "expression": expression}ReAct 框架：推理與行動的統一
ReAct（Reasoning and Acting）框架是 Agentic RAG 最流行的實現模式。它將推理和行動交織在一起，形成一個動態的問題解決循環。
理解 ReAct 的關鍵是認識到它的三個核心階段：
- Thought（思考）：分析當前狀態，決定下一步做什么
- Action（行動）：執行具體的操作，如檢索、計算等
- Observation（觀察）：觀察行動的結果，為下一輪思考提供輸入
class ReActAgent:
"""基于 ReAct 框架的代理
ReAct 框架的核心思想是將推理和行動交織進行：
- 不是先想好所有步驟再執行（傳統規劃方式）
- 而是想一步、做一步、看結果、再想下一步
這種方式更加靈活，能夠根據中間結果動態調整策略。
"""
def __init__(self, llm, tools: List[Tool]):
self.llm = llm
# 將工具列表轉換為字典，方便通過名稱查找
self.tools = {tool.name: tool for tool in tools}
# 設置最大迭代次數，防止無限循環
self.max_iterations = 5
asyncdef run(self, query: str):
"""ReAct 主循環
這個方法實現了 ReAct 的核心循環：
1. Thought：基于當前信息思考下一步
2. Action：執行選定的行動
3. Observation：觀察結果
4. 重復直到得到答案或達到最大迭代次數
這種循環模式的優勢：
- 可以根據中間結果調整策略
- 能夠處理需要多步推理的復雜問題
- 可以在過程中發現并糾正錯誤
"""
thoughts = [] # 記錄所有的思考過程
observations = [] # 記錄所有的觀察結果
for i in range(self.max_iterations):
# Thought: 思考下一步
# 這里 LLM 會分析當前情況，決定：
# - 是否已經有足夠信息回答問題
# - 如果沒有，需要使用什么工具獲取什么信息
thought = await self.think(query, thoughts, observations)
thoughts.append(thought)
# 檢查是否已經得到最終答案
# 如果 LLM 認為已經有足夠信息，就結束循環
if self.is_final_answer(thought):
return thought.answer
# Action: 執行行動
# 從思考結果中提取要執行的動作
action = self.parse_action(thought)
# Observation: 觀察結果
# 調用相應的工具并獲取結果
observation = await self.tools[action.tool].execute(**action.params)
observations.append(observation)
# 驗證進展
# 這是一個重要的安全機制，防止陷入循環
ifnot self.is_making_progress(thoughts, observations):
# 如果沒有進展，嘗試改變策略
await self.revise_strategy()
# 如果達到最大迭代次數，基于現有信息生成答案
return self.synthesize_from_observations(observations)
asyncdef think(self, query, thoughts, observations):
"""生成思考
這個方法構造提示詞，讓 LLM 基于歷史信息生成下一步的思考。
關鍵是提供足夠的上下文，讓 LLM 能做出明智的決策。
"""
prompt = f"""
Question: {query}
Previous thoughts: {thoughts}
Previous observations: {observations}
Based on the above information, think step by step:
1. What do we know so far?
2. What information is still missing?
3. What should be the next action?
If you have enough information to answer the question,
provide the final answer. Otherwise, specify what tool
to use next and with what parameters.
"""
# LLM 會返回結構化的思考結果
response = await self.llm.generate(prompt)
# 解析 LLM 的響應，提取思考內容和可能的動作
return self.parse_thought(response)
def is_making_progress(self, thoughts, observations):
"""檢查是否在取得進展
這個方法防止代理陷入循環或重復相同的操作。
通過分析歷史記錄，判斷是否：
1. 在獲取新信息
2. 在接近答案
3. 陷入了循環
"""
if len(observations) < 2:
returnTrue# 剛開始，認為在進展
# 檢查最近的觀察是否帶來新信息
last_obs = observations[-1]
prev_obs = observations[-2]
# 如果連續兩次得到相似的結果，可能陷入循環
if self.are_similar(last_obs, prev_obs):
returnFalse
# 檢查是否在重復相同的思考
if len(thoughts) > 2and thoughts[-1] == thoughts[-3]:
returnFalse
returnTrue實戰實現：用 LangGraph 構建 Agentic RAG
現在讓我們通過一個完整的實例，展示如何使用 LangGraph 構建生產級的 Agentic RAG 系統。LangGraph 是 LangChain 推出的用于構建有狀態的、多步驟的 AI 應用的框架。
基礎架構搭建
首先，我們需要定義系統的狀態和基本組件。在 Agentic RAG 中，狀態管理特別重要，因為系統需要在多個步驟之間傳遞和更新信息。
from langgraph.graph import StateGraph, MessagesState
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from typing import List, Dict, Any
import asyncio
# 定義狀態
class AgenticRAGState(MessagesState):
"""Agentic RAG 狀態
這個狀態類定義了在整個處理流程中需要傳遞的所有信息。
繼承自 MessagesState，自動包含了消息歷史記錄的管理。
每個字段的作用：
- query: 用戶的原始查詢，保持不變，用于最終答案的生成
- plan: 任務分解后的執行計劃，可能會根據執行情況動態調整
- retrieved_docs: 檢索到的文檔，累積存儲所有相關信息
- reflections: 系統的自我反思記錄，用于改進策略
- final_answer: 最終生成的答案
- iteration: 當前迭代次數，用于控制循環
- max_iterations: 最大迭代次數，防止無限循環
"""
query: str
plan: List[str] = []
retrieved_docs: List[Dict] = []
reflections: List[str] = []
final_answer: str = ""
iteration: int = 0
max_iterations: int = 5
# 初始化 LLM
# 使用 GPT-4，temperature=0 確保輸出的一致性和可預測性
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 創建檢索工具
def create_retriever_tool(vectorstore):
"""創建檢索工具
這個函數將向量數據庫封裝成一個工具，使其可以被 Agentic RAG 調用。
關鍵特性：
1. 異步執行，提高性能
2. 返回結構化數據，包含內容和元數據
3. 支持調整檢索數量（k 參數）
"""
asyncdef retrieve(query: str, k: int = 5) -> List[Dict]:
"""檢索相關文檔
Args:
query: 檢索查詢
k: 返回的文檔數量
Returns:
包含文檔內容和元數據的列表
"""
# 使用向量數據庫進行語義搜索
docs = await vectorstore.similarity_search(query, k=k)
# 將文檔對象轉換為字典，方便后續處理
# 保留元數據很重要，可能包含來源、時間戳等信息
return [
{
"content": doc.page_content,
"metadata": doc.metadata,
"relevance_score": doc.score if hasattr(doc, 'score') elseNone
}
for doc in docs
]
return Tool(
name="retrieve",
func=retrieve,
description="檢索相關文檔，支持語義搜索"
)節點函數實現
接下來，我們實現 Agentic RAG 的核心節點。每個節點負責特定的任務，通過組合這些節點，我們可以構建復雜的推理流程。
async def plan_node(state: AgenticRAGState):
"""規劃節點：分解任務
這是 Agentic RAG 的第一步，也是最關鍵的一步。
它將用戶的復雜查詢分解為可執行的子任務。
為什么需要任務分解？
1. 復雜問題往往需要多個步驟才能解答
2. 分解后的子任務更容易檢索到相關信息
3. 可以并行處理多個子任務，提高效率
例如，查詢"比較 A 和 B 的優缺點"會被分解為：
- 了解 A 的特點
- 了解 B 的特點
- 分析 A 的優勢
- 分析 B 的優勢
- 對比兩者的差異
"""
prompt = f"""
用戶問題：{state.query}
請將這個問題分解為幾個子任務，每個子任務應該：
1. 具體且可執行 - 能夠通過檢索或推理完成
2. 相對獨立 - 每個任務聚焦一個方面
3. 邏輯有序 - 任務之間有合理的執行順序
4. 完整覆蓋 - 所有任務完成后能回答原始問題
返回 JSON 格式的任務列表，例如：
["任務1", "任務2", "任務3"]
"""
# 調用 LLM 生成執行計劃
response = await llm.ainvoke(prompt)
# 解析 LLM 返回的 JSON
# 這里需要健壯的錯誤處理，因為 LLM 可能返回格式不正確的內容
try:
plan = parse_json(response.content)
except Exception as e:
# 如果解析失敗，使用備用策略
plan = [state.query] # 降級為直接檢索原始查詢
return {"plan": plan}
asyncdef retrieve_node(state: AgenticRAGState):
"""檢索節點：智能檢索
這個節點負責執行實際的信息檢索。
與傳統 RAG 的區別在于：
1. 為每個子任務生成優化的查詢
2. 評估檢索結果的相關性
3. 可以根據需要進行多輪檢索
檢索策略：
- 初次檢索：廣泛搜索，獲取更多可能相關的信息
- 精確檢索：基于初次結果，進行更targeted的搜索
- 驗證檢索：交叉驗證信息的準確性
"""
retrieved_docs = []
for task in state.plan:
# 步驟1：查詢優化
# 不直接使用任務描述，而是生成更適合檢索的查詢
optimized_query = await optimize_query(task, state.query)
# 步驟2：執行檢索
# 可能使用多種檢索策略：
# - 語義檢索（向量相似度）
# - 關鍵詞檢索（BM25）
# - 混合檢索（結合多種方法）
docs = await retriever_tool.func(optimized_query)
# 步驟3：相關性評估
# 不是所有檢索到的文檔都真正相關
# 使用 LLM 或其他方法評估相關性
relevant_docs = await evaluate_relevance(docs, task)
# 步驟4：去重和合并
# 避免重復信息，保持結果的簡潔性
for doc in relevant_docs:
ifnot is_duplicate(doc, retrieved_docs):
retrieved_docs.append(doc)
return {"retrieved_docs": retrieved_docs}
asyncdef optimize_query(task: str, original_query: str) -> str:
"""優化檢索查詢
這個函數將任務描述轉換為更適合檢索的查詢。
優化策略包括：
1. 添加相關關鍵詞
2. 擴展同義詞
3. 考慮上下文
4. 使用領域特定的術語
"""
prompt = f"""
原始問題：{original_query}
當前子任務：{task}
請生成一個優化的檢索查詢，要求：
1. 保留核心語義
2. 添加相關關鍵詞
3. 使用更專業的術語
4. 考慮可能的同義詞
直接返回優化后的查詢，不要解釋。
"""
response = await llm.ainvoke(prompt)
return response.content.strip()
asyncdef evaluate_relevance(docs: List[Dict], task: str) -> List[Dict]:
"""評估文檔相關性
這個函數篩選出真正相關的文檔。
評估標準：
1. 內容相關性：文檔是否包含任務所需的信息
2. 信息質量：信息是否準確、完整、最新
3. 來源可靠性：文檔來源是否可信
"""
relevant_docs = []
for doc in docs:
prompt = f"""
任務：{task}
文檔內容：{doc['content'][:500]} # 只看前500字符
這個文檔對完成任務的相關性如何？
評分標準（1-10分）：
- 直接相關（8-10分）：包含任務所需的核心信息
- 部分相關（5-7分）：包含有用但不完整的信息
- 邊緣相關（3-4分）：只有少量相關信息
- 不相關（1-2分）：與任務無關
只返回數字分數。
"""
response = await llm.ainvoke(prompt)
try:
score = float(response.content.strip())
if score >= 5: # 只保留部分相關以上的文檔
doc['relevance_score'] = score
relevant_docs.append(doc)
except:
pass# 忽略無法解析的響應
# 按相關性分數排序
relevant_docs.sort(key=lambda x: x.get('relevance_score', 0), reverse=True)
return relevant_docs[:5] # 只返回最相關的5個文檔
asyncdef reflect_node(state: AgenticRAGState):
"""反思節點：自我評估
這是 Agentic RAG 的獨特特性，系統會：
1. 評估當前收集的信息是否足夠
2. 識別信息中的矛盾和空白
3. 決定是否需要調整策略
反思的重要性：
- 避免盲目執行，確保方向正確
- 及時發現問題，避免錯誤累積
- 持續優化策略，提高效率
"""
prompt = f"""
原始問題：{state.query}
執行計劃：{state.plan}
檢索到的文檔數量：{len(state.retrieved_docs)}
請進行深度分析和反思：
1. 信息完整性評估
- 檢索到的信息是否覆蓋了所有子任務？
- 是否有關鍵信息缺失？
- 信息的深度是否足夠？
2. 信息質量評估
- 信息是否可靠和準確？
- 是否存在矛盾的信息？
- 信息是否過時？
3. 策略有效性評估
- 當前的檢索策略是否有效？
- 是否需要調整搜索方向？
- 是否需要使用其他工具或方法？
4. 下一步建議
- 如果信息充足，可以生成答案
- 如果信息不足，建議具體的改進措施
請給出你的分析和建議。
"""
# 為了讓 LLM 更好地分析，我們提供文檔摘要
doc_summary = summarize_documents(state.retrieved_docs)
prompt = prompt.replace("檢索到的文檔數量", f"檢索到的文檔摘要：\n{doc_summary}\n\n文檔數量")
reflection = await llm.ainvoke(prompt)
# 更新狀態
return {
"reflections": state.reflections + [reflection.content],
"iteration": state.iteration + 1
}
asyncdef answer_node(state: AgenticRAGState):
"""回答節點：綜合生成答案
這是最后一步，將所有收集的信息綜合成一個完整的答案。
答案生成策略：
1. 結構化：按照邏輯順序組織信息
2. 完整性：確保回答了用戶的所有問題
3. 準確性：基于檢索到的事實，避免臆測
4. 可讀性：語言清晰，易于理解
"""
# 準備答案生成的上下文
context = prepare_answer_context(state)
prompt = f"""
基于以下信息，請為用戶問題提供一個全面、準確的答案。
用戶問題：{state.query}
執行的任務分解：
{format_list(state.plan)}
檢索到的關鍵信息：
{context['key_findings']}
系統反思和分析：
{format_list(state.reflections)}
要求：
1. 直接回答用戶的問題，不要贅述過程
2. 如果信息不完整，誠實地指出局限性
3. 使用檢索到的事實支撐你的觀點
4. 保持客觀中立，不要加入個人偏見
5. 如果有多個觀點，公平地呈現各方觀點
請生成最終答案：
"""
answer = await llm.ainvoke(prompt)
# 可以添加后處理步驟
# 例如：事實核查、格式優化、添加引用等
final_answer = post_process_answer(answer.content, state)
return {"final_answer": final_answer}
def should_continue(state: AgenticRAGState):
"""決定是否繼續迭代
這個函數是 Agentic RAG 的"大腦"，決定系統的下一步行動。
它需要平衡多個因素：
1. 信息是否充足
2. 是否達到迭代上限
3. 是否在取得進展
4. 成本考慮
這種動態決策機制是 Agentic RAG 的核心優勢。
"""
# 檢查是否達到最大迭代次數
if state.iteration >= state.max_iterations:
return"answer"# 強制結束，生成答案
# 分析最新的反思結果
if state.reflections:
latest_reflection = state.reflections[-1]
# 使用簡單的關鍵詞匹配判斷
# 實際應用中可以用更復雜的方法
if"信息充足"in latest_reflection or"可以回答"in latest_reflection:
return"answer"
if"需要更多信息"in latest_reflection or"信息不足"in latest_reflection:
return"retrieve"
# 基于文檔數量的啟發式判斷
if len(state.retrieved_docs) < 3:
return"retrieve"# 文檔太少，繼續檢索
if len(state.retrieved_docs) > 20:
return"answer"# 文檔足夠多，可以回答
# 默認繼續檢索
return"retrieve"構建和運行 Agentic RAG 圖
現在我們將所有組件組合成一個完整的系統：
def build_agentic_rag_graph():
"""構建 Agentic RAG 圖
這個函數使用 LangGraph 將所有節點和邊組合成一個有向圖。
圖的結構定義了信息流和決策邏輯。
關鍵設計決策：
1. 單一入口點：從 plan 節點開始
2. 循環結構：retrieve -> reflect -> retrieve（如需要）
3. 明確的退出點：answer 節點后結束
"""
graph = StateGraph(AgenticRAGState)
# 添加節點
# 每個節點都是一個異步函數，處理特定任務
graph.add_node("plan", plan_node)
graph.add_node("retrieve", retrieve_node)
graph.add_node("reflect", reflect_node)
graph.add_node("answer", answer_node)
# 添加邊 - 定義節點之間的連接
# 設置入口點 - 所有處理從規劃開始
graph.set_entry_point("plan")
# plan 完成后總是進入 retrieve
graph.add_edge("plan", "retrieve")
# retrieve 完成后總是進入 reflect
graph.add_edge("retrieve", "reflect")
# 條件邊 - 根據條件決定下一步
# 這是實現動態行為的關鍵
graph.add_conditional_edges(
"reflect", # 從 reflect 節點
should_continue, # 使用這個函數決定
{
"retrieve": "retrieve", # 繼續檢索
"answer": "answer" # 生成答案
}
)
# answer 節點后結束
graph.add_edge("answer", END)
# 編譯圖 - 優化執行
return graph.compile()
# 使用示例
asyncdef main():
"""主函數 - 演示如何使用 Agentic RAG 系統"""
# 創建 Agentic RAG 系統
agentic_rag = build_agentic_rag_graph()
# 復雜查詢示例
# 這種查詢需要多方面的信息和深度分析
query = """
比較 LangChain 和 LlamaIndex 在構建 RAG 系統時的優缺點，
并推薦適合初創公司的方案，考慮因素包括：
1. 開發效率
2. 性能表現
3. 成本控制
4. 可維護性
"""
# 執行查詢
print("開始處理查詢...")
print(f"查詢內容：{query}\n")
result = await agentic_rag.ainvoke({
"query": query,
"messages": [], # 初始化空消息列表
"plan": [],
"retrieved_docs": [],
"reflections": [],
"iteration": 0
})
# 輸出結果
print("=" * 60)
print("最終答案：")
print("=" * 60)
print(result['final_answer'])
print("\n" + "=" * 60)
print(f"執行統計：")
print(f"- 迭代次數：{result['iteration']}")
print(f"- 檢索文檔數：{len(result['retrieved_docs'])}")
print(f"- 任務分解數：{len(result['plan'])}")
# 顯示執行過程（用于調試和優化）
print("\n" + "=" * 60)
print("執行計劃：")
for i, task in enumerate(result['plan'], 1):
print(f"{i}. {task}")
print("\n" + "=" * 60)
print("反思記錄：")
for i, reflection in enumerate(result['reflections'], 1):
print(f"\n第 {i} 次反思：")
print(reflection[:200] + "..."if len(reflection) > 200else reflection)
# 運行主函數
if __name__ == "__main__":
asyncio.run(main())多代理協作系統
在復雜場景中，單個代理往往力不從心。多代理系統通過專業化分工和協作，可以處理更復雜的任務。這就像一個專家團隊，每個成員都有自己的專長。
from typing import Optional
from enum import Enum
class AgentRole(Enum):
"""代理角色定義
每個角色代表一種專業能力。
這種設計模式類似于微服務架構，每個代理負責特定領域。
"""
MASTER = "master" # 主控代理 - 協調者
RETRIEVER = "retriever" # 檢索專家 - 信息獲取
ANALYZER = "analyzer" # 分析專家 - 數據分析
VALIDATOR = "validator" # 驗證專家 - 質量控制
SYNTHESIZER = "synthesizer"# 綜合專家 - 信息整合
class SpecializedAgent:
"""專門化代理基類
這個基類定義了所有專業代理的共同接口。
每個代理都有：
1. 特定的角色和職責
2. 專門的工具集
3. 領域知識
"""
def __init__(self, role: AgentRole, llm):
self.role = role
self.llm = llm
self.tools = self._initialize_tools()
self.expertise = self._load_expertise()
def _initialize_tools(self):
"""初始化角色特定工具
不同角色需要不同的工具：
- 檢索專家需要各種搜索工具
- 分析專家需要統計和數據處理工具
- 驗證專家需要事實核查工具
"""
if self.role == AgentRole.RETRIEVER:
return [
SearchTool(), # 網絡搜索
DatabaseTool(), # 數據庫查詢
APITool(), # API 調用
DocumentLoader() # 文檔加載
]
elif self.role == AgentRole.ANALYZER:
return [
StatisticalTool(), # 統計分析
NERTool(), # 命名實體識別
SentimentTool(), # 情感分析
TrendAnalyzer() # 趨勢分析
]
elif self.role == AgentRole.VALIDATOR:
return [
FactCheckTool(), # 事實核查
SourceVerificationTool(), # 來源驗證
ConsistencyChecker(), # 一致性檢查
BiasDetector() # 偏見檢測
]
elif self.role == AgentRole.SYNTHESIZER:
return [
Summarizer(), # 摘要生成
ReportGenerator(), # 報告生成
VisualizationTool() # 數據可視化
]
return []
asyncdef process(self, task: Dict, context: Dict):
"""處理任務的抽象方法
每個專業代理需要實現自己的處理邏輯
"""
raise NotImplementedError
class MasterAgent(SpecializedAgent):
"""主控代理：協調其他代理
主控代理是整個系統的指揮官，負責：
1. 分析用戶需求
2. 制定執行策略
3. 分配任務給合適的代理
4. 協調代理之間的協作
5. 整合最終結果
這種設計模式類似于管理者模式（Manager Pattern）。
"""
def __init__(self, llm, sub_agents: Dict[AgentRole, SpecializedAgent]):
super().__init__(AgentRole.MASTER, llm)
self.sub_agents = sub_agents
self.execution_history = []
asyncdef process(self, query: str):
"""協調處理流程
這個方法實現了復雜的多代理協作邏輯。
關鍵步驟：
1. 理解問題的復雜性和需求
2. 設計合適的執行策略
3. 智能地分配任務
4. 處理代理間的依賴關系
5. 確保結果的質量
"""
# 步驟1：分析查詢，理解需求的復雜性
analysis = await self.analyze_query(query)
# 步驟2：制定執行策略
# 根據查詢的特點決定：
# - 需要哪些代理參與
# - 任務的執行順序
# - 是否可以并行處理
strategy = await self.create_strategy(analysis)
# 步驟3：分解任務
# 將大任務分解為各個代理可以處理的小任務
tasks = await self.decompose_tasks(strategy, query)
# 步驟4：執行任務
# 這里展示了并行和串行執行的混合策略
results = []
# 并行執行獨立任務
parallel_tasks = [t for t in tasks ifnot t.get('depends_on')]
if parallel_tasks:
parallel_results = await asyncio.gather(*[
self.delegate_task(task) for task in parallel_tasks
])
results.extend(parallel_results)
# 串行執行依賴任務
sequential_tasks = [t for t in tasks if t.get('depends_on')]
for task in sequential_tasks:
# 等待依賴任務完成
await self.wait_for_dependencies(task, results)
result = await self.delegate_task(task)
results.append(result)
# 步驟5：整合結果
final_result = await self.integrate_results(results, query)
# 記錄執行歷史，用于學習和優化
self.execution_history.append({
'query': query,
'strategy': strategy,
'results': final_result,
'performance': self.measure_performance()
})
return final_result
asyncdef delegate_task(self, task: Dict):
"""委派任務給合適的代理
這個方法實現了智能的任務分配：
1. 選擇最合適的代理
2. 準備任務上下文
3. 監控執行過程
4. 處理失敗情況
"""
# 選擇合適的代理
agent_role = self.select_agent(task)
agent = self.sub_agents[agent_role]
# 準備上下文
context = self.prepare_context(task)
try:
# 執行任務
result = await agent.process(task, context)
# 驗證結果質量
if self.needs_validation(result):
validator = self.sub_agents[AgentRole.VALIDATOR]
validation_result = await validator.process(result, task)
# 如果驗證失敗，可能需要重試
ifnot validation_result['is_valid']:
returnawait self.handle_validation_failure(
task, result, validation_result
)
return result
except Exception as e:
# 錯誤處理和恢復
returnawait self.handle_agent_failure(task, agent_role, e)
def select_agent(self, task: Dict) -> AgentRole:
"""選擇最合適的代理
基于任務特征選擇代理：
- 信息獲取任務 → 檢索專家
- 數據分析任務 → 分析專家
- 質量檢查任務 → 驗證專家
- 報告生成任務 → 綜合專家
"""
task_type = task.get('type', '')
if task_type in ['search', 'retrieve', 'fetch']:
return AgentRole.RETRIEVER
elif task_type in ['analyze', 'compute', 'extract']:
return AgentRole.ANALYZER
elif task_type in ['verify', 'validate', 'check']:
return AgentRole.VALIDATOR
elif task_type in ['summarize', 'synthesize', 'report']:
return AgentRole.SYNTHESIZER
else:
# 默認使用檢索專家
return AgentRole.RETRIEVER
class RetrieverAgent(SpecializedAgent):
"""檢索專家：負責信息獲取
檢索專家是信息獲取的專家，掌握各種檢索技術：
1. 多源檢索：從不同來源獲取信息
2. 查詢優化：改寫查詢以獲得更好的結果
3. 結果融合：合并和去重多源結果
4. 相關性排序：確保最相關的信息排在前面
"""
asyncdef process(self, task: Dict, context: Dict):
"""智能檢索流程
這個方法實現了高級的檢索策略，
不只是簡單的關鍵詞匹配，而是真正理解信息需求。
"""
query = task["query"]
# 步驟1：查詢理解和改寫
# 一個查詢可能需要多種表述才能找到所有相關信息
rewritten_queries = await self.rewrite_query(query, context)
# 步驟2：多源并行檢索
all_results = []
# 為每個改寫的查詢執行多種檢索
for rq in rewritten_queries:
# 并行執行不同類型的檢索
search_tasks = [
self.vector_search(rq), # 語義檢索
self.keyword_search(rq), # 關鍵詞檢索
self.graph_search(rq), # 知識圖譜檢索
self.structured_search(rq) # 結構化數據檢索
]
# 等待所有檢索完成
results = await asyncio.gather(*search_tasks)
# 合并結果
for result_set in results:
all_results.extend(result_set)
# 步驟3：結果處理
# 去重
unique_results = self.deduplicate(all_results)
# 相關性評分
scored_results = await self.score_relevance(unique_results, query)
# 重排序
reranked_results = await self.rerank(scored_results, query, context)
# 步驟4：結果增強
# 添加元數據、摘要等
enhanced_results = await self.enhance_results(reranked_results)
return {
"query": query,
"results": enhanced_results[:10], # 返回前10個結果
"total_found": len(unique_results),
"search_strategies": ["vector", "keyword", "graph", "structured"]
}
asyncdef rewrite_query(self, query: str, context: Dict) -> List[str]:
"""查詢改寫
生成多個查詢變體以提高召回率：
1. 同義詞替換
2. 概念擴展
3. 上下文補充
4. 領域特定術語
"""
rewrites = [query] # 保留原始查詢
# 使用 LLM 生成查詢變體
prompt = f"""
原始查詢：{query}
上下文：{context.get('history', '')}
請生成3-5個查詢變體，包括：
1. 使用同義詞的版本
2. 更具體的版本
3. 更寬泛的版本
4. 使用專業術語的版本
每行一個查詢。
"""
response = await self.llm.ainvoke(prompt)
variants = response.content.strip().split('\n')
rewrites.extend(variants)
return rewrites[:5] # 限制最多5個查詢
class ValidatorAgent(SpecializedAgent):
"""驗證專家：確保信息質量
驗證專家是質量把關者，負責：
1. 事實核查
2. 來源驗證
3. 一致性檢查
4. 偏見檢測
這對于生成可信賴的答案至關重要。
"""
asyncdef process(self, result: Dict, task: Dict):
"""驗證流程
多維度驗證確保信息的可靠性。
每個維度都有具體的檢查標準和方法。
"""
validations = []
# 維度1：事實準確性
# 檢查信息是否與已知事實一致
fact_check = await self.fact_check(result["content"])
validations.append({
"type": "fact_check",
"score": fact_check["accuracy"],
"issues": fact_check.get("issues", [])
})
# 維度2：來源可靠性
# 評估信息來源的權威性和可信度
source_check = await self.verify_sources(result.get("sources", []))
validations.append({
"type": "source_verification",
"score": source_check["reliability"],
"trusted_sources": source_check["trusted"],
"questionable_sources": source_check["questionable"]
})
# 維度3：內部一致性
# 檢查信息是否自相矛盾
consistency_check = await self.check_consistency(
result["content"],
task.get("context", {})
)
validations.append({
"type": "consistency",
"score": consistency_check["consistency_score"],
"contradictions": consistency_check.get("contradictions", [])
})
# 維度4：偏見檢測
# 識別可能的偏見和片面性
bias_check = await self.detect_bias(result["content"])
validations.append({
"type": "bias_detection",
"score": 1 - bias_check["bias_level"], # 轉換為質量分數
"detected_biases": bias_check.get("biases", [])
})
# 綜合評分
# 使用加權平均計算總體可信度
weights = {
"fact_check": 0.4,
"source_verification": 0.3,
"consistency": 0.2,
"bias_detection": 0.1
}
confidence_score = sum(
v["score"] * weights[v["type"]]
for v in validations
)
# 生成驗證報告
return {
**result, # 保留原始結果
"validations": validations,
"confidence": confidence_score,
"is_valid": confidence_score > 0.7, # 閾值判斷
"recommendations": self.generate_recommendations(validations)
}
def generate_recommendations(self, validations):
"""基于驗證結果生成改進建議
幫助系統理解如何改進信息質量。
"""
recommendations = []
for validation in validations:
if validation["score"] < 0.6:
if validation["type"] == "fact_check":
recommendations.append("需要更多事實核查")
elif validation["type"] == "source_verification":
recommendations.append("尋找更可靠的信息來源")
elif validation["type"] == "consistency":
recommendations.append("解決信息矛盾")
elif validation["type"] == "bias_detection":
recommendations.append("平衡不同觀點")
return recommendations創建和使用多代理系統
現在讓我們看看如何將這些代理組合成一個完整的系統：
async def create_multi_agent_system():
"""創建多代理 RAG 系統
這個函數組裝整個多代理系統。
關鍵設計考慮：
1. 代理的初始化順序
2. 代理之間的依賴關系
3. 通信機制
4. 錯誤處理
"""
# 初始化共享的 LLM
# 所有代理使用相同的 LLM，但可以有不同的配置
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 創建專門化代理
# 每個代理都有自己的專業領域和工具集
retriever = RetrieverAgent(AgentRole.RETRIEVER, llm)
analyzer = AnalyzerAgent(AgentRole.ANALYZER, llm)
validator = ValidatorAgent(AgentRole.VALIDATOR, llm)
synthesizer = SynthesizerAgent(AgentRole.SYNTHESIZER, llm)
# 配置代理的特定參數
retriever.max_results = 20
analyzer.confidence_threshold = 0.8
validator.strict_mode = True
synthesizer.output_format = "detailed"
# 創建主控代理
# 主控代理需要知道所有子代理
master = MasterAgent(llm, {
AgentRole.RETRIEVER: retriever,
AgentRole.ANALYZER: analyzer,
AgentRole.VALIDATOR: validator,
AgentRole.SYNTHESIZER: synthesizer
})
# 設置代理間通信
# 可以添加消息隊列、事件總線等機制
setup_inter_agent_communication(master, [retriever, analyzer, validator, synthesizer])
return master
# 使用多代理系統
asyncdef run_multi_agent_query():
"""運行多代理查詢示例
這個例子展示了多代理系統如何處理復雜查詢。
"""
# 創建系統
system = await create_multi_agent_system()
# 復雜查詢示例
# 這種查詢需要多個代理協作：
# - 檢索：獲取行業數據和新聞
# - 分析：趨勢分析和模式識別
# - 驗證：事實核查和可靠性評估
# - 綜合：生成報告和預測
query = """
分析過去三個月科技行業的并購趨勢，
預測未來六個月可能的大型并購案，
并評估對相關公司股價的潛在影響。
要求：
1. 基于真實數據
2. 考慮多個信息源
3. 提供置信度評估
4. 給出具體的投資建議
"""
print("開始多代理協作處理...")
print(f"查詢：{query}\n")
# 執行查詢
start_time = time.time()
result = await system.process(query)
end_time = time.time()
# 輸出結果
print("=" * 60)
print("分析結果：")
print("=" * 60)
print(f"\n{result['analysis']}")
print("\n" + "=" * 60)
print("預測：")
print("=" * 60)
for i, prediction in enumerate(result['predictions'], 1):
print(f"\n{i}. {prediction['company']}")
print(f" 可能性：{prediction['probability']:.1%}")
print(f" 預期影響：{prediction['impact']}")
print("\n" + "=" * 60)
print("投資建議：")
print("=" * 60)
print(result['recommendations'])
print("\n" + "=" * 60)
print(f"置信度評估：{result['confidence']:.1%}")
print(f"處理時間：{end_time - start_time:.2f} 秒")
print(f"參與代理：{result['agents_involved']}")
print(f"信息源數量：{result['sources_count']}")
# 運行示例
if __name__ == "__main__":
asyncio.run(run_multi_agent_query())高級技術：提升 Agentic RAG 性能
1. 自適應查詢改寫
查詢改寫是提高檢索質量的關鍵技術。好的查詢改寫可以顯著提高召回率和準確率。
class AdaptiveQueryRewriter:
"""自適應查詢改寫器
這個類實現了多種查詢改寫策略，并能根據效果動態調整。
核心思想：
1. 不同類型的查詢需要不同的改寫策略
2. 從歷史成功案例中學習
3. 根據反饋持續優化
"""
def __init__(self, llm):
self.llm = llm
self.rewrite_history = [] # 記錄改寫歷史
self.strategy_performance = {} # 策略性能統計
asyncdef rewrite(self, query: str, context: Dict) -> List[str]:
"""多策略查詢改寫
這個方法綜合使用多種改寫策略，
每種策略針對不同的問題：
- 查詢太復雜 → 分解
- 缺少上下文 → 補充
- 術語不準確 → 同義詞擴展
- 沒有歷史 → 學習優化
"""
rewrites = []
# 策略1：分解復雜查詢
# 復雜查詢往往包含多個子問題
if self.is_complex(query):
sub_queries = await self.decompose_query(query)
rewrites.extend(sub_queries)
# 記錄策略使用
self.record_strategy_use("decomposition", query, sub_queries)
# 策略2：上下文增強
# 添加對話歷史中的相關信息
contextualized = await self.add_context(query, context)
rewrites.append(contextualized)
self.record_strategy_use("contextualization", query, [contextualized])
# 策略3：同義詞和概念擴展
# 使用不同的表達方式可能找到不同的信息
expanded = await self.expand_synonyms(query)
rewrites.append(expanded)
self.record_strategy_use("expansion", query, [expanded])
# 策略4：基于歷史學習的優化
# 從成功的歷史改寫中學習模式
if self.rewrite_history:
optimized = await self.learn_from_history(query)
rewrites.append(optimized)
self.record_strategy_use("learning", query, [optimized])
# 去重并返回
return self.deduplicate(rewrites)
def is_complex(self, query: str) -> bool:
"""判斷查詢是否復雜
復雜查詢的特征：
- 包含多個子句
- 有多個并列的問題
- 包含條件判斷
- 字數較多
"""
# 簡單的啟發式判斷
indicators = [
len(query.split()) > 20, # 詞數多
'并且'in query or'and'in query.lower(), # 并列關系
'如果'in query or'if'in query.lower(), # 條件關系
query.count('，') > 2or query.count(',') > 2, # 多個子句
'比較'in query or'compare'in query.lower() # 比較關系
]
return sum(indicators) >= 2
asyncdef decompose_query(self, query: str) -> List[str]:
"""分解復雜查詢
將一個復雜查詢分解為多個簡單查詢。
每個子查詢應該：
1. 聚焦單一方面
2. 保持完整的語義
3. 可以獨立檢索
"""
prompt = f"""
請將以下復雜問題分解為2-4個簡單的子問題：
{query}
要求：
1. 每個子問題聚焦一個方面
2. 子問題應該是完整的句子
3. 所有子問題合起來能回答原問題
4. 避免重復
直接列出子問題，每行一個。
"""
response = await self.llm.ainvoke(prompt)
sub_queries = response.content.strip().split('\n')
# 過濾和清理
return [q.strip() for q in sub_queries if len(q.strip()) > 10]
asyncdef add_context(self, query: str, context: Dict) -> str:
"""添加上下文信息
很多查詢缺少必要的上下文，
比如"他"指誰，"這個"指什么。
這個方法補充這些信息。
"""
ifnot context ornot context.get('history'):
return query
prompt = f"""
原始查詢：{query}
對話歷史：{context['history'][-3:]} # 最近3輪對話
請補充必要的上下文信息，使查詢更明確。
例如：
- 將代詞替換為具體名稱
- 補充時間、地點等信息
- 明確模糊的指代
直接返回改寫后的查詢。
"""
response = await self.llm.ainvoke(prompt)
return response.content.strip()
asyncdef learn_from_history(self, query: str) -> str:
"""從歷史中學習最佳改寫模式
這個方法分析歷史上成功的改寫案例，
找出模式并應用到當前查詢。
這是一種簡單的在線學習機制。
"""
# 找出最成功的改寫案例
successful_rewrites = [
h for h in self.rewrite_history
if h.get("success_score", 0) > 0.8
]
ifnot successful_rewrites:
return query
# 找出最相似的歷史查詢
similar_cases = self.find_similar_queries(query, successful_rewrites)
ifnot similar_cases:
return query
prompt = f"""
當前查詢：{query}
類似的成功改寫案例：
{self.format_cases(similar_cases[:3])}
基于這些成功案例的模式，改寫當前查詢。
直接返回改寫結果。
"""
response = await self.llm.ainvoke(prompt)
return response.content.strip()
def record_strategy_use(self, strategy: str, original: str, rewrites: List[str]):
"""記錄策略使用情況
用于后續分析和優化。
"""
record = {
"strategy": strategy,
"original": original,
"rewrites": rewrites,
"timestamp": datetime.now()
}
self.rewrite_history.append(record)
def update_performance(self, strategy: str, success: bool):
"""更新策略性能統計
跟蹤每種策略的成功率，
可以用于動態調整策略選擇。
"""
if strategy notin self.strategy_performance:
self.strategy_performance[strategy] = {"success": 0, "total": 0}
self.strategy_performance[strategy]["total"] += 1
if success:
self.strategy_performance[strategy]["success"] += 12. 動態工具選擇
不同的查詢需要不同的工具。動態選擇工具可以提高效率和準確性。
class DynamicToolSelector:
"""動態工具選擇器
這個類實現了智能的工具選擇機制：
1. 分析查詢特征
2. 評估工具相關性
3. 考慮歷史性能
4. 動態調整選擇策略
目標是為每個查詢選擇最合適的工具組合。
"""
def __init__(self, tools: List[Tool], llm):
self.tools = tools
self.llm = llm
# 使用 defaultdict 自動初始化統計數據
self.tool_usage_stats = defaultdict(lambda: {"success": 0, "total": 0})
self.query_patterns = {} # 查詢模式與工具的映射
asyncdef select_tools(self, query: str, context: Dict) -> List[Tool]:
"""智能選擇工具
這個方法綜合考慮多個因素來選擇工具：
1. 查詢的語義特征
2. 工具的功能描述
3. 歷史使用效果
4. 當前系統負載
返回按優先級排序的工具列表。
"""
# 步驟1：分析查詢特征
# 提取查詢的關鍵特征，如意圖、實體、時態等
query_features = await self.analyze_query_features(query)
# 步驟2：計算每個工具的相關性分數
tool_scores = []
for tool in self.tools:
# 基于語義相關性的分數（0-1）
relevance_score = await self.calculate_relevance(
tool,
query_features
)
# 基于歷史性能的分數（0-1）
performance_score = self.get_performance_score(tool)
# 基于當前負載的分數（0-1）
# 避免過度使用某些工具
load_score = self.get_load_balance_score(tool)
# 綜合評分
# 可以調整權重以改變選擇策略
final_score = (
0.5 * relevance_score + # 相關性最重要
0.3 * performance_score + # 歷史性能次之
0.2 * load_score # 負載均衡最后
)
tool_scores.append((tool, final_score))
# 步驟3：選擇得分最高的工具
# 按分數排序
tool_scores.sort(key=lambda x: x[1], reverse=True)
# 選擇策略：
# - 如果最高分很高（>0.8），只選擇最好的1-2個
# - 如果分數都不高，選擇前3個以提高覆蓋率
# - 如果是復雜查詢，可能需要更多工具
if tool_scores[0][1] > 0.8:
# 高置信度，選擇最好的
selected = [tool for tool, score in tool_scores[:2] if score > 0.6]
elif self.is_complex_query(query_features):
# 復雜查詢，需要更多工具
selected = [tool for tool, score in tool_scores[:4] if score > 0.4]
else:
# 一般情況，選擇前3個
selected = [tool for tool, score in tool_scores[:3] if score > 0.5]
# 記錄選擇決策，用于學習
self.record_selection(query, query_features, selected)
return selected
asyncdef analyze_query_features(self, query: str) -> Dict:
"""分析查詢特征
提取有助于工具選擇的特征：
- 查詢類型（事實、分析、創造等）
- 時間特征（歷史、當前、預測）
- 數據類型（文本、數字、圖表等）
- 復雜度
"""
prompt = f"""
分析以下查詢的特征：
{query}
請識別：
1. 查詢類型：[事實查詢|分析查詢|比較查詢|預測查詢|創造性查詢]
2. 時間范圍：[歷史|當前|未來]
3. 需要的數據類型：[文本|數值|圖表|代碼|混合]
4. 復雜度：[簡單|中等|復雜]
5. 關鍵實體：列出主要實體
返回 JSON 格式。
"""
response = await self.llm.ainvoke(prompt)
try:
features = json.loads(response.content)
except:
# 解析失敗時的默認特征
features = {
"type": "fact",
"time": "current",
"data_type": "text",
"complexity": "simple",
"entities": []
}
return features
asyncdef calculate_relevance(self, tool: Tool, features: Dict) -> float:
"""計算工具相關性
基于工具描述和查詢特征計算相關性。
這里可以使用更復雜的方法，如：
- 語義相似度
- 規則匹配
- 機器學習模型
"""
# 使用 LLM 評估相關性
prompt = f"""
工具信息：
名稱：{tool.name}
描述：{tool.description}
查詢特征：
{json.dumps(features, indent=2)}
評估這個工具對處理該查詢的相關性（0-10分）。
只返回數字。
"""
response = await self.llm.ainvoke(prompt)
try:
score = float(response.content.strip())
return score / 10# 歸一化到 0-1
except:
return0.5# 默認中等相關性
def get_performance_score(self, tool: Tool) -> float:
"""獲取工具的歷史性能分數
基于成功率計算性能分數。
新工具給予探索獎勵。
"""
stats = self.tool_usage_stats[tool.name]
if stats["total"] == 0:
# 新工具，給予探索獎勵
return0.7
# 計算成功率
success_rate = stats["success"] / stats["total"]
# 考慮使用次數，避免因少量失敗就放棄工具
# 使用貝葉斯平均
confidence = min(stats["total"] / 10, 1.0)
# 加權平均，新工具有更多機會
return success_rate * confidence + 0.5 * (1 - confidence)
def get_load_balance_score(self, tool: Tool) -> float:
"""計算負載均衡分數
避免過度使用某些工具，
促進工具使用的多樣性。
"""
total_uses = sum(s["total"] for s in self.tool_usage_stats.values())
if total_uses == 0:
return1.0
tool_uses = self.tool_usage_stats[tool.name]["total"]
# 使用率越高，分數越低
usage_rate = tool_uses / total_uses
# 使用反sigmoid函數，使分數變化更平滑
return1 / (1 + math.exp(10 * (usage_rate - 0.3)))
def update_stats(self, tool: Tool, success: bool):
"""更新工具使用統計
記錄工具的使用結果，用于持續優化選擇策略。
"""
self.tool_usage_stats[tool.name]["total"] += 1
if success:
self.tool_usage_stats[tool.name]["success"] += 1
# 定期打印統計信息（用于監控）
if sum(s["total"] for s in self.tool_usage_stats.values()) % 100 == 0:
self.print_statistics()
def print_statistics(self):
"""打印工具使用統計
用于監控和調試。
"""
print("\n工具使用統計：")
print("-" * 40)
for tool_name, stats in self.tool_usage_stats.items():
if stats["total"] > 0:
success_rate = stats["success"] / stats["total"]
print(f"{tool_name}:")
print(f" 使用次數：{stats['total']}")
print(f" 成功率：{success_rate:.2%}")3. 結果驗證與自我糾正
Agentic RAG 的一個關鍵優勢是能夠驗證和糾正自己的輸出。
class SelfCorrectingAgent:
"""自我糾正代理
這個類實現了自我驗證和糾正機制。
核心理念：
1. 不盲目相信第一次的輸出
2. 通過多維度驗證發現問題
3. 基于問題進行針對性糾正
4. 迭代改進直到滿意
"""
def __init__(self, llm):
self.llm = llm
self.correction_attempts = 0
self.max_corrections = 3
self.validation_history = []
asyncdef generate_with_verification(self, query: str, context: Dict):
"""生成并驗證答案
這個方法實現了完整的生成-驗證-糾正循環。
關鍵特點：
1. 多次嘗試機會
2. 每次糾正都基于具體問題
3. 保留所有嘗試的歷史
4. 最終選擇最佳答案
"""
attempts = [] # 記錄所有嘗試
while self.correction_attempts < self.max_corrections:
# 生成答案
# 如果是糾正嘗試，會包含之前的錯誤信息
answer = await self.generate_answer(query, context)
# 驗證答案
validation_result = await self.validate_answer(answer, query)
# 記錄嘗試
attempts.append({
"answer": answer,
"validation": validation_result,
"attempt": self.correction_attempts
})
# 判斷是否滿意
if validation_result.is_valid:
return answer
# 準備糾正
# 關鍵：基于驗證結果準備針對性的糾正上下文
context = await self.prepare_correction_context(
answer,
validation_result,
context
)
self.correction_attempts += 1
print(f"第 {self.correction_attempts} 次糾正...")
# 達到最大嘗試次數，選擇最佳答案
return self.select_best_answer(attempts)
asyncdef validate_answer(self, answer: str, query: str):
"""全面驗證答案質量
多維度驗證確保答案質量：
1. 相關性：是否回答了問題
2. 完整性：是否覆蓋所有方面
3. 一致性：是否自相矛盾
4. 準確性：事實是否正確
每個維度都有具體的檢查方法。
"""
validations = []
# 維度1：相關性檢查
relevance = await self.check_relevance(answer, query)
validations.append(("relevance", relevance))
# 維度2：完整性檢查
completeness = await self.check_completeness(answer, query)
validations.append(("completeness", completeness))
# 維度3：一致性檢查
consistency = await self.check_consistency(answer)
validations.append(("consistency", consistency))
# 維度4：事實準確性檢查
accuracy = await self.check_factual_accuracy(answer)
validations.append(("accuracy", accuracy))
# 綜合評估
# 所有維度都需要達到閾值
is_valid = all(score > 0.7for _, score in validations)
# 識別主要問題
problems = [
name for name, score in validations
if score < 0.7
]
# 生成改進建議
suggestions = self.generate_suggestions(validations)
return ValidationResult(
is_valid=is_valid,
validations=dict(validations),
problems=problems,
suggestions=suggestions
)
asyncdef check_relevance(self, answer: str, query: str) -> float:
"""檢查答案相關性
評估答案是否真正回答了用戶的問題。
"""
prompt = f"""
問題：{query}
答案：{answer}
評估答案對問題的相關性（0-10分）：
- 10分：完全回答了問題
- 7-9分：基本回答了問題，有小的偏差
- 4-6分：部分回答了問題
- 1-3分：答案偏題
- 0分：完全不相關
只返回數字分數。
"""
response = await self.llm.ainvoke(prompt)
try:
return float(response.content.strip()) / 10
except:
return0.5
asyncdef check_completeness(self, answer: str, query: str) -> float:
"""檢查答案完整性
評估答案是否覆蓋了問題的所有方面。
"""
prompt = f"""
問題：{query}
答案：{answer}
分析問題包含的所有方面，評估答案的完整性（0-10分）：
- 識別問題的所有子問題
- 檢查每個子問題是否被回答
- 評估答案的深度是否足夠
只返回數字分數。
"""
response = await self.llm.ainvoke(prompt)
try:
return float(response.content.strip()) / 10
except:
return0.5
asyncdef check_consistency(self, answer: str) -> float:
"""檢查內部一致性
檢測答案中是否有自相矛盾的地方。
"""
prompt = f"""
檢查以下答案的內部一致性：
{answer}
尋找：
1. 邏輯矛盾
2. 事實沖突
3. 數字不一致
4. 時間線混亂
如果完全一致，返回10分；
每發現一個問題，減2分；
最低0分。
只返回數字分數。
"""
response = await self.llm.ainvoke(prompt)
try:
return float(response.content.strip()) / 10
except:
return0.8# 默認認為基本一致
asyncdef prepare_correction_context(
self,
answer: str,
validation: ValidationResult,
original_context: Dict
) -> Dict:
"""準備糾正上下文
這是自我糾正的關鍵：
基于驗證結果，準備針對性的糾正指導。
"""
correction_context = original_context.copy()
# 添加之前的答案和問題
correction_context["previous_answer"] = answer
correction_context["validation_results"] = validation.validations
correction_context["problems"] = validation.problems
# 生成具體的改進指導
improvement_guide = []
if"relevance"in validation.problems:
improvement_guide.append("重新聚焦用戶的核心問題")
if"completeness"in validation.problems:
improvement_guide.append("補充遺漏的方面")
# 可能需要額外檢索
missing_info = await self.identify_missing_information(
answer,
original_context["query"]
)
if missing_info:
correction_context["missing_information"] = missing_info
if"consistency"in validation.problems:
improvement_guide.append("解決答案中的矛盾")
contradictions = await self.identify_contradictions(answer)
correction_context["contradictions"] = contradictions
if"accuracy"in validation.problems:
improvement_guide.append("糾正事實錯誤")
# 觸發事實核查
fact_checks = await self.fact_check_details(answer)
correction_context["fact_corrections"] = fact_checks
correction_context["improvement_guide"] = improvement_guide
return correction_context
def select_best_answer(self, attempts: List[Dict]) -> str:
"""選擇最佳答案
當所有嘗試都不完美時，
選擇相對最好的答案。
"""
ifnot attempts:
return"無法生成滿意的答案"
# 計算每個答案的綜合分數
scored_attempts = []
for attempt in attempts:
validation = attempt["validation"]
# 加權平均
score = (
validation.validations.get("relevance", 0) * 0.4 +
validation.validations.get("completeness", 0) * 0.3 +
validation.validations.get("consistency", 0) * 0.2 +
validation.validations.get("accuracy", 0) * 0.1
)
scored_attempts.append((score, attempt["answer"]))
# 返回得分最高的答案
scored_attempts.sort(reverse=True)
return scored_attempts[0][1]評估與優化：讓系統持續進化
評估框架
評估 Agentic RAG 系統比評估傳統 RAG 更復雜，因為需要考慮推理質量、工具使用效率等多個維度。
class AgenticRAGEvaluator:
"""Agentic RAG 評估器
全面評估系統的各個方面：
1. 檢索質量
2. 推理能力
3. 答案質量
4. 系統效率
5. 成本控制
評估結果用于系統優化和調參。
"""
def __init__(self):
self.metrics = {
"retrieval": RetrievalMetrics(),
"reasoning": ReasoningMetrics(),
"generation": GenerationMetrics(),
"efficiency": EfficiencyMetrics(),
"cost": CostMetrics()
}
self.evaluation_history = []
asyncdef evaluate(self, system, test_set):
"""全面評估系統
使用測試集評估系統的各個維度。
測試集應該包含：
- 查詢
- 預期答案
- 相關文檔
- 難度等級
"""
results = {
"retrieval_precision": [],
"reasoning_accuracy": [],
"answer_quality": [],
"latency": [],
"cost": [],
"tool_efficiency": []
}
for test_case in test_set:
print(f"評估測試用例：{test_case['id']}")
# 記錄執行軌跡
# 這是評估的基礎數據
trace = await self.trace_execution(system, test_case)
# 評估各個維度
# 1. 檢索質量
retrieval_score = self.evaluate_retrieval(trace, test_case)
results["retrieval_precision"].append(retrieval_score)
# 2. 推理質量
reasoning_score = self.evaluate_reasoning(trace, test_case)
results["reasoning_accuracy"].append(reasoning_score)
# 3. 答案質量
answer_score = self.evaluate_answer(trace, test_case)
results["answer_quality"].append(answer_score)
# 4. 系統效率
results["latency"].append(trace.total_time)
results["tool_efficiency"].append(self.evaluate_tool_usage(trace))
# 5. 成本
cost = self.calculate_cost(trace)
results["cost"].append(cost)
# 聚合結果
aggregated = self.aggregate_results(results)
# 生成評估報告
report = self.generate_report(aggregated, test_set)
# 保存評估歷史
self.evaluation_history.append({
"timestamp": datetime.now(),
"results": aggregated,
"report": report
})
return report
def evaluate_retrieval(self, trace, test_case):
"""評估檢索質量
衡量標準：
- 準確率：檢索到的相關文檔比例
- 召回率：相關文檔被檢索到的比例
- MRR：平均倒數排名
- NDCG：歸一化折損累積增益
"""
retrieved_docs = trace.retrieved_docs
ground_truth = test_case.get("relevant_docs", [])
ifnot ground_truth:
# 沒有標注數據，使用啟發式評估
return self.heuristic_retrieval_evaluation(retrieved_docs, test_case)
# 計算標準指標
retrieved_ids = [doc["id"] for doc in retrieved_docs if"id"in doc]
truth_ids = [doc["id"] for doc in ground_truth]
# 準確率
if retrieved_ids:
precision = len(set(retrieved_ids) & set(truth_ids)) / len(retrieved_ids)
else:
precision = 0
# 召回率
if truth_ids:
recall = len(set(retrieved_ids) & set(truth_ids)) / len(truth_ids)
else:
recall = 1# 沒有相關文檔，認為召回完整
# F1分數
if precision + recall > 0:
f1 = 2 * precision * recall / (precision + recall)
else:
f1 = 0
# MRR（平均倒數排名）
mrr = self.calculate_mrr(retrieved_ids, truth_ids)
return {
"precision": precision,
"recall": recall,
"f1": f1,
"mrr": mrr
}
def evaluate_reasoning(self, trace, test_case):
"""評估推理質量
這是 Agentic RAG 特有的評估維度。
衡量標準：
- 推理步驟的邏輯性
- 計劃的合理性
- 工具使用的恰當性
- 自我糾正的有效性
"""
reasoning_steps = trace.reasoning_steps
scores = {}
# 1. 邏輯一致性
# 檢查推理步驟之間是否有邏輯關系
scores["logical_consistency"] = self.check_logical_consistency(reasoning_steps)
# 2. 計劃質量
# 評估任務分解是否合理
if trace.plan:
scores["plan_quality"] = self.evaluate_plan_quality(
trace.plan,
test_case["query"]
)
# 3. 工具使用恰當性
# 評估是否選擇了正確的工具
scores["tool_appropriateness"] = self.evaluate_tool_selection(
trace.tool_calls,
test_case["query"]
)
# 4. 自我糾正效果
# 如果有糾正，評估糾正的效果
if trace.corrections:
scores["correction_effectiveness"] = self.evaluate_corrections(
trace.corrections
)
# 5. 推理深度
# 評估推理的深度和完整性
expected_steps = test_case.get("expected_reasoning_steps", 3)
actual_steps = len(reasoning_steps)
scores["reasoning_depth"] = min(actual_steps / expected_steps, 1.0)
# 綜合評分
weights = {
"logical_consistency": 0.3,
"plan_quality": 0.2,
"tool_appropriateness": 0.2,
"correction_effectiveness": 0.1,
"reasoning_depth": 0.2
}
overall_score = sum(
scores.get(key, 0.5) * weight
for key, weight in weights.items()
)
return {
"overall": overall_score,
"details": scores
}
def evaluate_answer(self, trace, test_case):
"""評估答案質量
多維度評估答案：
- 正確性
- 完整性
- 清晰度
- 有用性
"""
generated_answer = trace.final_answer
scores = {}
# 如果有參考答案，計算相似度
if"expected_answer"in test_case:
scores["correctness"] = self.calculate_answer_similarity(
generated_answer,
test_case["expected_answer"]
)
# 完整性：是否回答了所有子問題
scores["completeness"] = self.check_answer_completeness(
generated_answer,
test_case["query"]
)
# 清晰度：答案是否組織良好
scores["clarity"] = self.evaluate_answer_clarity(generated_answer)
# 有用性：答案是否提供了實用信息
scores["usefulness"] = self.evaluate_answer_usefulness(
generated_answer,
test_case["query"]
)
# 綜合評分
overall = sum(scores.values()) / len(scores)
return {
"overall": overall,
"details": scores
}
def evaluate_tool_usage(self, trace):
"""評估工具使用效率
衡量工具使用的效率和效果。
"""
tool_calls = trace.tool_calls
ifnot tool_calls:
return1.0# 沒有使用工具，認為效率最高
# 計算各種效率指標
# 1. 成功率
successful_calls = [tc for tc in tool_calls if tc.get("success")]
success_rate = len(successful_calls) / len(tool_calls)
# 2. 重復率（越低越好）
unique_calls = len(set(
(tc["tool"], tc.get("query", ""))
for tc in tool_calls
))
duplication_rate = 1 - (unique_calls / len(tool_calls))
# 3. 相關性（工具選擇是否恰當）
relevance_scores = [
tc.get("relevance_score", 0.5)
for tc in tool_calls
]
average_relevance = sum(relevance_scores) / len(relevance_scores)
# 綜合效率分數
efficiency_score = (
0.4 * success_rate +
0.3 * (1 - duplication_rate) +
0.3 * average_relevance
)
return efficiency_score
def generate_report(self, results, test_set):
"""生成評估報告
生成詳細的評估報告，包括：
- 總體性能
- 各維度分析
- 問題診斷
- 優化建議
"""
report = []
report.append("=" * 60)
report.append("Agentic RAG 系統評估報告")
report.append("=" * 60)
# 總體性能
report.append("\n## 總體性能")
report.append(f"測試用例數：{len(test_set)}")
report.append(f"平均檢索精度：{results['retrieval_precision']:.2%}")
report.append(f"平均推理準確率：{results['reasoning_accuracy']:.2%}")
report.append(f"平均答案質量：{results['answer_quality']:.2%}")
report.append(f"平均響應時間：{results['avg_latency']:.2f} 秒")
report.append(f"平均成本：${results['avg_cost']:.4f}")
# 性能分布
report.append("\n## 性能分布")
report.append(f"最快響應：{results['min_latency']:.2f} 秒")
report.append(f"最慢響應：{results['max_latency']:.2f} 秒")
report.append(f"P95 延遲：{results['p95_latency']:.2f} 秒")
# 問題診斷
report.append("\n## 問題診斷")
if results['reasoning_accuracy'] < 0.7:
report.append("?? 推理準確率偏低，建議：")
report.append(" - 優化任務分解策略")
report.append(" - 加強推理鏈的邏輯驗證")
if results['avg_latency'] > 10:
report.append("?? 響應時間過長，建議：")
report.append(" - 優化工具調用策略")
report.append(" - 實現并行處理")
report.append(" - 添加緩存機制")
if results['avg_cost'] > 0.1:
report.append("?? 成本偏高，建議：")
report.append(" - 減少不必要的 LLM 調用")
report.append(" - 使用更經濟的模型")
report.append(" - 優化 prompt 長度")
# 優化建議
report.append("\n## 優化建議")
recommendations = self.generate_recommendations(results)
for rec in recommendations:
report.append(f"? {rec}")
return"\n".join(report)總結
Agentic RAG 不只是 RAG 的升級版——它代表了一種全新的思維方式。這種轉變的意義是深遠的。在傳統 RAG 中，我們在構建"更好的搜索引擎"；而在 Agentic RAG 中，我們在構建"會思考的合作伙伴"。
當然，這項技術還在快速演進中。成本、延遲、可靠性等問題仍需解決。但方向是明確的——未來的 AI 系統將更加智能、自主、可靠。
記住：不是每個釘子都需要錘子，但當你真的需要錘子時，請選擇最好的那把。
最近建了?
?langchain&langgraph??的交流群，想加這個群的朋友可以點贊關注之后加我備注 langgraph 即可，不以學習為目的的就不要進了
本文轉載自??AI 博物院?? 作者：longyunfeigu
?著作權歸作者所有，如需轉載，請注明出處，否則將追究法律責任
已于2025-9-5 06:40:49修改
贊
收藏
回復
分享
微博
QQ
微信
舉報
回復
相關推薦
