基于 AI Agent 智能體架構落地復雜問答系統的案例設計與實踐 原創
1、先從傳統思維的架構設計說起
我們先來看一個場景：用戶輸入一個問題（Prompt）后，如何實現用戶問題的自動路由和查詢（比如：是 RAG 檢索還是 SQL 查詢）？
- 解決方案：意圖識別架構設計
第一、對于用戶輸入的問題（Prompt），我們通過意圖識別模塊來判斷是需要路由給RAG 檢索還是路由給 SQL 查詢；
第二、意圖識別模塊接收用戶的問題（Prompt）,然后將用戶的問題交給大模型去判斷分類；
第三、意圖識別架構設計如下圖所示：
第四、大模型需要提前做 SFT Fine-tuning 微調，目的是讓大模型知道什么問題屬于 RAG 檢索問題，什么問題屬于 SQL 查詢問題，Fine-tuning 微調的樣本數據格式如下：
examples =[
{
"inn":"我想知道東方阿爾法優勢產業混合C基金，在2021年年度報告中，前10大重倉股中，有多少只股票在報告期內取得正收益。",
"out":"rag_question***我想知道東方阿爾法優勢產業混合C基金，在2021年年度報告中，前10大重倉股中，有多少只股票在報告期內取得正收益。"
},
{
"inn":"森赫電梯股份有限公司產品生產材料是什么？",
"out":"rag_question***森赫電梯股份有限公司產品生產材料是什么？"
},
{
"inn":"20210930日，一級行業為機械的股票的成交金額合計是多少？取整。",
"out":"agent_question***20210930日，一級行業為機械的股票的成交金額合計是多少？取整。"
},
...
]第五、意圖識別模塊識別結果后，通過在代碼中硬編碼 if...else 對識別后的分類進行不同的后續處理，代碼如下：
# 如果是 RAG 相關的問題
if intent =="rag_question":
result = self.rag.get_result(questinotallow=question)
return result
# 如果是 Agent 相關的問題
elif intent =="agent_question":
result = self.agent.get_result(input=question)
return result
# 其他類問題
else:
result = self.chat.invoke(input=question).content
return result- 意圖識別架構設計的問題
從上述實現流程我們可以看到，意圖識別的架構設計本質還是傳統的開發模式，也就是靜態業務流程編排的面向過程架構設計范式，也就是說程序執行的每一個步驟還是由開發者精準控制，大模型此刻只是扮演者輔助工具角色，用于問題分類，這種面向過程的架構設計范式有3個問題：
第一、若意圖識別出現偏差，后續的處理流程將失去效果，結果必然會出現錯誤。
第二、盡管我們可以在 if...else... 結構中加入修正策略，比如：當 RAG 檢索失敗時轉而查詢 SQ L數據庫，但面對復雜的需求，這種策略的編寫與維護成本將呈指數級增長。
第三、最關鍵的是，這種處理方式絲毫體現不出智能化的特點。
那么如何基于 AI Agent 智能體的架構設計對用戶問題（Prompt）自主判斷和自主解決呢？
2、基于 AI Agent 智能體的新架構設計方案
AI Agent 智能體由大模型推理能力、Planning 規劃能力、Tools 工具使用能力、Action 行動能力、Memory 記憶能力構成，如下圖所示：
從上圖我們可以看到，一旦我們為 AI Agent 智能體配備了適當的工具，它便能借助大模型的推理能力，獨立地進行規劃和行動。因此，我們首先需要解決的問題是如何為 AI Agent 智能體配備必要的工具。
第一、為 AI Agent 智能體配備工具
這個案例我們的需求是結構化的 SQL 查詢和非結構化的 RAG 檢索，我們給 AI Agent 智能體配備這兩個工具：
1.創建 AI Agent 的管理類
import logging
import datetime
from langgraph.prebuilt import create_react_agent
from langchain.tools.retriever import create_retriever_tool
from langchain_community.utilities importSQLDatabase
from langchain_community.agent_toolkits importSQLDatabaseToolkit
from rag.rag importRagManager
import settings
classFinanceBotEx:
def__init__(self, llm=settings.llm, chat=settings.chat, embed=settings.embed):
self.llm = llm
self.chat = chat
self.embed = embed
self.tools =[]
self.rag =RagManager(llm=llm, embed=embed)2.創建 SQL 查詢工具
為 FinanceBotEX 類添加 SQL 查詢工具。
def init_sql_tool(self, path):
# 連接數據庫
db = SQLDatabase.from_uri(f"sqlite:///{path}")
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
sql_tools = toolkit.get_tools() # 工具
return sql_tools3.創建 RAG 檢索工具
為 FinanceBotEX 類添加 RAG 檢索工具。
def init_rag_tools(self):
# 給大模型 RAG 檢索器工具
retriever = self.rag.get_retriever()
retriever_tool = create_retriever_tool(
retriever=retriever,
name="rag_search",
descriptinotallow="按照用戶的問題搜索相關的資料，對于招股書類的問題，you must use this tool!",
)
return retriever_tool第二、通過 System Prompt 告訴 AI Agent 智能體的工作邏輯
為 FinanceBotEX 類添加 System Prompt 工作指令。
def create_prompt():
system_prompt = """你是一位金融助手，可以幫助用戶查詢數據庫中的信息。
你要盡可能的回答用戶提出的問題，為了更好的回答問題，你可以使用工具進行多輪的嘗試。
# 關于 retriever_tool工具的使用：
1、你需要結合對檢索出來的上下文進行回答問題。
2、如果你不知道答案，就說你不知道。請使用不超過三句話的簡潔回答。
# 關于 sql_tools類工具的使用：
## 工具使用規則
1、你需要根據用戶的問題，創建一個語法正確的 SQLite 查詢來運行，然后查看查詢的結果并返回答案。
2、除非用戶指定了他們希望獲得的特定數量的示例，否則總是將查詢限制為最多5個結果。
3、您可以按相關列對結果進行排序，以返回數據庫中最有趣的示例。
4、永遠不要查詢指定表的所有列以避免查詢性能問題，你只查詢給定問題的相關列即可。
5、你必須在執行查詢之前仔細檢查查詢。如果執行查詢時出現錯誤，請重新編寫查詢并重試。
6、請勿對數據庫進行任何 DML 語句（INSERT，UPDATE，DELETE，DROP等）。
## 工具使用過程
1、首先，你應該始終查看數據庫中的表，看看可以查詢什么，這一步驟很重要，注意不要跳過。
2、然后，你應該查詢最相關表的 schema。
## 工具使用注意事項：
1、如果生成的 SQL 語句中，字段帶有英文括號()，請使用雙引號包裹起來，例如：收盤價(元) 雙引號包裹為 "收盤價(元)"。
2、如果查詢過程中 SQL 語句有語法錯誤，減少查詢量,總體查詢次數應控制在15次以內。
# 關于你的思考和行動過程，請按照如下格式：
問題：你必須回答的輸入問題
思考：你應該總是考慮該怎么做
行動：你應該采取的行動，應該是以下工具之一：{tool_names}
行動輸入：行動的輸入
觀察：行動的結果
... (這個思考/行動/行動輸入/觀察可以重復N次)
思考: 我現在知道最終答案了
最終答案：原始輸入問題的最終答案
Begin!
"""
return system_prompt第三、創建 AI Agent 智能體
在配置好所需的工具和 System Prompt 之后，我們可以創建一個 AI Agent 智能體，并將預先準備好的工具和 System Prompt 傳遞給它。
接著，為 FinanceBotEx 類進行擴展。
def init_agent(self):
#初始化 RAG 工具
retriever_tool = self.init_rag_tools()
#初始化 SQL 工具
sql_tools = self.init_sql_tool(settings.SQLDATABASE_URI)
#創建系統 Prompt 提示語
system_prompt = self.create_prompt()
#創建Agent
agent_executor = create_react_agent(
self.chat,
tools=[retriever_tool]+ sql_tools,
state_modifier=system_prompt
)
return agent_executor第四、 AI Agent 智能體運行處理邏輯
為 FinanceBotEx 類繼續添加運行處理邏輯：
def handle_query(self, example_query):
# 流式處理事件
events = self.agent_executor.stream(
{
"messages":[("user", example_query)]},
stream_mode="values",
)
# 打印流式事件的消息
for event in events:
event["messages"][-1].pretty_print()
return event["messages"][-1].content第五、 AI Agent 智能體運行結果
在 Python 文件中繼續添加測試函數：
# 測試 FinanceBotEx 主流程
def test_financebot_ex():
from finance_bot_ex importFinanceBotEx
llm, chat, embed = get_qwen_models()
financebot =FinanceBotEx(llm=llm, chat=chat, embed=embed)
example_query ="20210304日，一級行業為非銀金融的股票的成交量合計是多少？取整。"
financebot.handle_query(example_query)
if __name__ =="__main__":
# test_rag()
# test_import()
test_financebot_ex()運行結果如下：
通過日志可以看到，AI Agent 智能體收到問題后，先去 SQL 數據庫中查詢看有哪些表可能有這些數據，通過返回的表信息自主實現對應的 SQL 語句，然后查詢結果后給出答案。
如果將上面的問題變成如下問題：
example_query = "根據聯化科技股份有限公司招股意見書，精細化工產品的通常利潤率是多少？"運行結果如下：
通過日志可以看到，AI Agent 智能體對于這個問題認為應該直接去 RAG 中搜索，所以直接調用工具 rag_search。
3、基于 AI Agent 智能體的新架構設計方案總結
- AI Agent 智能體的架構設計是一種面向目標架構設計的新范式，它的架構設計理念與傳統面向過程架構設計范式存在顯著差異：
第一、在傳統架構設計中，需要對程序的每一個步驟進行精確的控制，技靜態業務流程編排；
第二、AI Agent 智能體的核心理念是通過提示詞（Prompt）向大模型傳達運行規則，隨后大模型自主進行思考和執行操作。
- 使用 AI Agent 智能體的過程涉及以下4個步驟：
第一、首先構建所需的工具。
第二、接著設計相應的提示詞（Prompt）。
第三、然后創建 AI Agent Agent，并將這些工具和提示詞賦予它。
第四，通過 AI Agent 智能體的流（stream）功能來處理用戶的輸入問題。
本文轉載自公眾號玄姐聊AGI 作者：玄姐
