前言

上一章【项目实战】基于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. 1. 如果意图识别错了，那么后续的处理就失效了，答案一定会错；

2. 2. 我们固然可以给if...else...增加修正策略，比如：rag检索不到那就去sql里再查一下....但是如果我们需求很复杂呢，这个策略编写和维护成本也是指数级上升；

3. 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_prompt

3、创建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。