前言
上一章【项目实战】基于Agent的金融问答系统：RAG检索模块初建成，我们已经完成了RAG的基本流程，本章我将梳理Agent的运行框架以及介绍如何将RAG数据和SQL查询数据进行整合，实现金融问题的自主回答。
核心问题
在进行Agent框架介绍之前，我们需要探讨一个核心问题，即：用户输入一个问题后，如何实现用户问题的自动查询(是SQL查询还是RAG检索) 。
意图识别方案
关于这个问题，我们曾有一个方案如下：
• 对于用户输入的问题，是需要查询SQL，还是需要去RAG检索，我们通过一个意图识别模块来实现；
• 意图识别模块会接受用户的问题，然后将用户的问题交给大模型去分辨；
• 大模型我们提前做了sft微调(few-shot少样本训练也可以达到同样效果)，让大模型知道什么样的问题属于SQL查询问题，什么样的问题属于RAG检索问题，例如：
# 准备few-shot样例
examples =[
{
"inn":"我想知道东方阿尔法优势产业混合C基金，在2021年年度报告中，前10大重仓股中，有多少只股票在报告期内取得正收益。",
"out":"rag_question***我想知道东方阿尔法优势产业混合C基金，在2021年年度报告中，前10大重仓股中，有多少只股票在报告期内取得正收益。"
},
{
"inn":"森赫电梯股份有限公司产品生产材料是什么？",
"out":"rag_question***森赫电梯股份有限公司产品生产材料是什么？"
},
{
"inn":"20210930日，一级行业为机械的股票的成交金额合计是多少？取整。",
"out":"agent_question***20210930日，一级行业为机械的股票的成交金额合计是多少？取整。"
},
...
]• 意图识别结果之后，在代码中通过if...else对识别结果进行不同类别的后续处理,如下：
if intent =="rag_question":
# 如果是RAG相关的问题
result = self.rag.get_result(question=question)
return result
elif intent =="agent_question":
# 如果是Agent相关的问题
result, result_list = self.agent.get_result(input=question)
return result
else:
# 其他类问题
result = self.chat.invoke(input=question).content
return result意图识别方案的问题
这一方案有个核心问题：
1. 如果意图识别错了，那么后续的处理就失效了，答案一定会错；
2. 我们固然可以给if...else...增加修正策略，比如：rag检索不到那就去sql里再查一下....但是如果我们需求很复杂呢，这个策略编写和维护成本也是指数级上升；
3. 最为重要的一点是：这种方式我觉得一点都不智能！
这种方式还是传统开发的思维：由开发者来清晰地控制着程序的每一步骤，大模型在其中只是扮演一个辅助工具角色(例如：做个问题分类)
但是，在我首次接触Agent时，Agent的自主思考和自主行动给我留下了深刻的印象，所以我决定换一种方式：使用Agent来进行用户问题的自主判断、自主解决。
Agent框架
在【课程总结】day30：大模型之Agent的初步了解中，我们曾初步了解Agent框架如下：
由上图可知，如果我们给Agent赋予相应的工具，那么Agent依托于大模型的思考力，可以自主开展相应的planning以及action。因此，我们首要解决如何给Agent赋予相应的工具。
1、赋予Agent工具
因为我们的需求是查询结构化的SQL数据库和非结构化的RAG数据，所以我们需要给Agent赋予两个工具，具体如下：
1.1、创建Agent的管理类
代码文件及目录：app/finance_bot_ex.py
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)
1.2、创建数据库SQL工具
为FinanceBotEx类添加
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_tools
说明：
• 上述函数中path用来传入SQlite数据库的路径
• 使用langchain封装的 SQLDatabase 连接数据库
• 使用langchain封装的 SQLDatabaseToolkit 得到SQL相关的工具，用以后续提供给Agent
1.3、创建RAG检索工具
为FinanceBotEx类添加
def init_rag_tools(self):
# 给大模型 RAG 检索器工具
retriever = self.rag.get_retriever()
retriever_tool = create_retriever_tool(
retriever=retriever,
name="rag_search",
description="按照用户的问题搜索相关的资料，对于招股书类的问题，you must use this tool!",
)
return retriever_tool说明：
• 该函数中的
self.rag是我们封装的RAG管理类• 通过这个管理类的get_retriever()得到RAG的检索器
• 借助langchain封装的create_retriever_tool()，将检索器创建为一个工具：
• 工具的
name为rag_search• 工具的
description用于告诉大模型，这个工具能做什么用(这一点很重要)
2、告诉Agent的工作逻辑：Prompt
为FinanceBotEx类添加
def create_prompt():
system_prompt = """你是一位金融助手，可以帮助用户查询数据库中的信息。
你要尽可能的回答用户提出的问题，为了更好的回答问题，你可以使用工具进行多轮的尝试。
# 关于retriever_tool工具的使用：
1、你需要结合对检索出来的上下文进行回答问题。
2、如果你不知道答案，就说你不知道。请使用不超过三句话的简洁回答。
# 关于sql类工具的使用：
## 工具使用规则
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_prompt3、创建agent
在准备好相应的工具以及prompt之后，创建Agent并传入提前准备好的工具和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说明：
•
settings.SQLDATABASE_URI是一个本地SQLite文件的路径，我将它单独维护在app/settings.py中，方便配置的修改。• 代码文件：
app/settings.py# 连接数据库db文件的地址根据需要需要更换
SQLDATABASE_URI = os.path.join(os.getcwd(), "app/dataset/dataset/博金杯比赛数据.db")
4、驱动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测试效果
在test_framework.py中增加测试函数如下：
# 测试 FinanceBotEx 主流程
deftest_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()运行结果：
通过日志可以看到，Agent收到问题后，先去SQL数据库中查询看有哪些表可能有这些数据，通过返回的表信息自主实现对应的SQL语句，然后查询结果后给出答案。
如果将上面的问题变成如下问题：
example_query = "根据联化科技股份有限公司招股意见书，精细化工产品的通常利润率是多少？"
运行结果：
通过日志可以看到，Agent对于这个问题认为应该直接去RAG中搜索，所以直接调用工具rag_search。
