披上Agent盔甲的RAG,從此不再只是召回生成! 精華
嘿,大家好!這里是一個專注于AI智能體的頻道!
今天我要和大家聊聊一些實戰相關的內容。大模型在實際的工業場景下最常見的2個場景分別為應用助手(copilot), 文檔/知識庫問答(RAG)。事實上后者也逐漸在往更復雜的Agentic方向發展了,今天我們來看以下如何搭建一個可控的RAG Agent。
RAG Agent核心是它的“大腦”,一個復雜的確定性圖,它能讓AI進行復雜的推理。而且,這個系統一般還能防止“幻覺”。確保所有答案都基于提供的數據,而不是憑空想象。
那我們要完成的這個系統需要具備哪些方面的能力呢?
- 可控自主,能夠回答自定義數據集中的重要問題。
- 充當系統的“大腦”agent ,能實現復雜的推理。
- 幻覺低,確保答案僅基于提供的數據,避免人工智能幻覺。
- 多步驟推理,將復雜的用戶查詢分解為可管理的子任務。
- 適應性規劃,根據新信息不斷更新其計劃。
大體的框架圖如下:
它是怎么工作的呢?簡單來說,就是先把PDF文檔加載進來,然后進行文本預處理,生成每個章節的摘要,再把這些內容編碼到向量庫中。當有人問問題時,AI會先對問題脫敏,生成一個規劃,然后再根據這個規劃進行細化出執行任務,最后生成最終答案。詳細步驟如下:
S1:構建adavanced RAG
數據準備,召回的retriever,可以理解為給文檔建立出索引,用戶后續的召回。當然在S1的搭建高級RAG中,我們只需要分段的即可。
接下來,可以構建標準的adavanced RAG的流程,召回塊 -> 保留與query相關的塊 -> 根據是否相關來決定是否需要改寫 -> 答案如果可用則結束了 (這里的很多函數都可以在langgraph的官方examples中找到)
S2:從adavanced RAG 到 Agent
對于更復雜的任務,僅通過基于語義相似性檢索信息無法回答問題,需要更復雜的pipeline。為了實現這一目標,我們先忘記adavanced rag的流程。我們需要定義出Agent的工具,一般RAG的tool就是S1中的retriever(召回)。(為了更復雜一些,我們在S1數據準備中,準備了3個retriever(文檔塊、摘要、引用)單獨作為不同的tool)
可以得到3個子圖:
有了工具,那接下來就是Agent的核心,planning部分了。
計劃制定
首先需要制定計劃->計劃細化到工具上。
制定計劃
計劃細化
示例:
question = {"question": "主人公是如何打敗反派的?"}
my_plan = planner.invoke(question) # Generate a plan to answer the question
print(my_plan)
refined_plan = break_down_plan_chain.invoke(my_plan.steps) # Refine the plan
print(refined_plan)
#### output
steps1 = [
'識別故事中的主人公和反派。',
'找到主人公和反派之間的高潮或最終對決。',
'分析主人公在這次對決中采取的行動。',
'確定導致反派失敗的具體行動或策略。',
'總結發現,回答主人公是如何打敗反派的。'
]
steps2 = [
'通過從書籍塊的向量存儲、章節摘要或書籍引用中檢索相關信息來識別故事中的主人公和反派。',
'通過從書籍塊的向量存儲、章節摘要或書籍引用中檢索相關信息來定位主人公和反派之間的高潮或最終對決。',
'通過從書籍塊的向量存儲、章節摘要或書籍引用中檢索相關信息來分析主人公在這次對決中采取的行動。',
'通過從書籍塊的向量存儲、章節摘要或書籍引用中檢索相關信息來確定導致反派失敗的具體行動或策略。',
'通過根據給定上下文回答問題來總結發現,回答主人公是如何打敗反派的。'
]
計劃更新
給定原始問題、當前計劃、過去的步驟以及迄今為止匯總的信息,更新計劃 (這個類似于一個迭代用到的,一次計劃無法完成任務,通過多次收集信息迭代)
任務處理
定義任務處理程序 - 決定是使用哪個工具來處理計劃中的每個任務
問題脫敏
脫敏
為了生成一個總體計劃,不帶任何基于任何先驗知識的偏見LLM,我們首先對輸入問題進行匿名化,并將名稱實體映射到變量中
還原
上面的串聯起來:
# 用戶問題
state1 = {'question': "how did the harry beat quirrell? \n"}
print(f'question: {state1["question"]}')
# 脫敏
anonymized_question_output = anonymize_question_chain.invoke(state1)
## 脫敏后的問題和脫敏字段
anonymized_question = anonymized_question_output["anonymized_question"]
mapping = anonymized_question_output["mapping"]
print(f'anonimized_querry: {anonymized_question} \n')
print(f'mapping: {mapping} \n')
# 制定計劃
plan = planner.invoke({"question": anonymized_question})
print(text_wrap(f'plan: {plan.steps}'))
print("")
# 計劃的脫敏信息還原
deanonimzed_plan = de_anonymize_plan_chain.invoke({"plan": plan.steps, "mapping": mapping})
## 還原后的計劃
print(text_wrap(f'deanonimized_plan: {deanonimzed_plan.plan}'))
# output
question:harry是如何打敗Quirrell的?
anonimized_querry:X是如何打敗Y的?
mapping:{'X': 'harry', 'Y': 'Quirrell'}
plan:[
'確定查詢的上下文或領域(例如,體育、競賽、游戲等)。',
'收集X和Y參加的事件或競賽的信息。',
'找到X與Y競爭的特定實例或比賽。',
'查找那個特定實例或比賽的結果。',
'分析比賽的細節,以了解X是如何設法打敗Y的。',
'總結解釋X如何打敗Y的關鍵點。'
]
deanonimized_plan:[
'確定查詢的上下文或領域(例如,體育、競賽、游戲等)。',
'收集harry和Quirrell參加的事件或競賽的信息。',
'找到harry與Quirrell競爭的特定實例或比賽。',
'查找那個特定實例或比賽的結果。',
'分析比賽的細節,以了解harry是如何設法打敗Quirrell的。',
'總結解釋harry如何打敗Quirrell的關鍵點。'
]
最后在加一個判斷,確定是否能根據信息推出答案。
整體的流程圖如下:
本文轉載自 ??探索AGI??,作者: 獼猴桃
