当前位置： 首页 > article >正文

搭建基于Agent的金融问答系统

article 2025/3/4 22:25:30

需求痛点

用户输入的问题，可能涉及到招股说明书的内容（PDF文件，非结构化的RAG数据），也可能是基金股票的持仓信息（结构化的SQL数据）等，前者是需要去多个PDF文件里查询，后者需要进行SQL语句查询，即前者是文本类数据，后者是数据表。

如何让金融问答系统能够自动判断，应该选择哪个进行查询回答呢？查SQL还是RAG检索呢？

需求分析

模型是不能调用外部工具（函数），但它能判断是否需要调用函数，如需调用函数，获取相关结果，则需使用代理模式 Agent。

上一篇 RAG项目实战：金融问答系统 完成了RAG的基本流程，本篇将介绍 Agent 的运行框架，以及如何将 RAG 数据和 SQL 查询数据进行整合，实现金融问题的自主回答。

这里需要探讨的一个核心问题，即：用户输入一个问题后，如何实现用户问题的自动查询（是SQL查询还是RAG检索）？

解决方案

方案一：SFT微调

给大模型提前做SFT微调（few-short少样本训练也可达到同样效果），让大模型知道什么样的问题属于SQL查询问题，什么样的问题属于RAG检索问题。例如：

# 准备 few-shot 样例，用于模型训练或推理时提供示范输入输出对。
examples = [
    # 示例 1：询问基金年度报告中前10大重仓股的正收益股票数量
    {
        "input": "我想知道东方阿尔法优势产业混合C基金，在2021年年度报告中，前10大重仓股中，有多少只股票在报告期内取得正收益。",
        "output": "rag_question***我想知道东方阿尔法优势产业混合C基金，在2021年年度报告中，前10大重仓股中，有多少只股票在报告期内取得正收益。",
        # 注释：此问题涉及基金年报数据查询，归类为 rag_question 类型。
    },
    # 示例 2：询问公司产品的生产材料
    {
        "input": "森赫电梯股份有限公司产品生产材料是什么？",
        "output": "rag_question***森赫电梯股份有限公司产品生产材料是什么？",
        # 注释：此问题需要从文档或知识库中检索相关信息，归类为 rag_question 类型。
    },
    # 示例 3：询问特定日期某一行业股票的成交金额合计
    {
        "input": "20210930日，一级行业为机械的股票的成交金额合计是多少？取整。",
        "output": "agent_question***20210930日，一级行业为机械的股票的成交金额合计是多少？取整。",
        # 注释：此问题涉及计算或数据汇总，需通过代理任务处理，归类为 agent_question 类型。
    },
    # 可继续添加更多样例...
]

意图识别结束后，在代码中通过if else 对识别结果进行不同类别的后续处理

# 根据问题的意图类型（intent）执行不同的处理逻辑
if intent == "rag_question":
    # 如果问题是 RAG 类型的问题（需要从文档或知识库中检索答案）
    result = self.rag.get_result(question=question)  # 调用 RAG 模块获取答案
    return result  # 返回检索到的结果

elif intent == "agent_question":
    # 如果问题是 Agent 类型的问题（需要通过代理任务或计算处理）
    result, result_list = self.agent.get_result(input=question)  # 调用 Agent 模块获取结果
    return result  # 返回处理后的结果

else:
    # 处理其他类型的普通问题（如闲聊或未分类问题）
    result = self.chat.invoke(input=question).content  # 调用聊天模块生成回复
    return result  # 返回生成的回复内容
该方案的弊病：

1. 如果意图识别错误，那后续的处理就失效了，答案一定会错。

2. 虽然可以使用 if else来修正策略，比如RAG检索不到就去SQL再查一下，但如果需求很复杂，该策略的编写和维护成本也会指数级上升。

3. 不智能！还属于传统开发的思维，开发者控制程序的每一步，大模型只在其中扮演了一个问题分类的辅助工具角色。

方案二：使用Agent对用户问题进行自主判断和解决

Agent的技术本质：

人工智能的本质 y = F(x)，agent  做的事都不应该是大模型做的事。agent 是个工程调度问题，将很多的外部能力封装成 tool ，交给大模型参考。大模型站在大脑的视角，灵活调度这些外部工具，判断什么时候该调用什么工具，以及参数是什么。

具体的调用则是通过外部代理来实现。大模型不负责具体的调用，动作代理负责配合大模型来调用。动作代理接收大模型的调用指示，完成调用动作，并把结果返给大模型。

ReAct：Reason + Action ；  推理 + 执行 ：先思考，再执行；反复推理 + 执行，直至完成。

Agent流程：

1. 准备外部工具（函数）列表 tools ，做能力拓展

2. 准备大模型 model ，做决策大脑

3. 创建一个agent，create_react_agent

4. 调用agent。

具体实现

1.准备工具

我们的需求是查询结构化的SQL数据库和非结构化的RAG数据，所以我们需要给Agent准备2个工具。

（1）创建Agent的管理类

代码文件及目录：app/fanance_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 import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from rag.rag import RagManager
import settings

class FinanceBotEx:
    """
    金融机器人扩展类，用于处理与金融相关的复杂任务。
    """

    def __init__(self, llm=settings.llm, chat=settings.chat, embed=settings.embed):
        """
        初始化 FinanceBotEx 类。

        参数:
            llm (object): 语言模型实例，默认从 settings 中获取。
            chat (object): 对话模型实例，默认从 settings 中获取。
            embed (object): 嵌入模型实例，默认从 settings 中获取。
        """
        # 初始化核心组件
        self.llm = llm  # 语言模型
        self.chat = chat  # 对话模型
        self.embed = embed  # 嵌入模型

        # 工具列表，用于存储各种工具
        self.tools = []

        # 初始化 RAG 管理器，用于检索增强生成（Retrieval-Augmented Generation）
        self.rag = RagManager(llm=llm, embed=embed)

    # 可在此处添加其他方法，例如：
    # - 添加工具到工具列表
    # - 处理特定任务的逻辑
    # - 集成数据库工具或检索工具
（2）创建数据库SQL工具

为FinanceBotEx类添加

def init_sql_tool(self, path):
    """
    初始化 SQL 工具。

    参数:
        path (str): 数据库文件的路径。

    返回:
        list: 包含 SQL 工具的列表。
    """
    # 连接到 SQLite 数据库
    db = SQLDatabase.from_uri(f"sqlite:///{path}")

    # 创建 SQL 数据库工具包，结合语言模型使用
    toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)

    # 获取工具包中的所有工具
    sql_tools = toolkit.get_tools()

    # 返回工具列表
    return sql_tools
（3）创建RAG检索工具

为FinanceBotEx类添加

def init_rag_tools(self):
    """
    初始化 RAG（检索增强生成）工具。

    返回:
        object: RAG 检索工具。
    """
    # 获取 RAG 管理器中的检索器
    retriever = self.rag.get_retriever()

    # 创建 RAG 检索工具
    retriever_tool = create_retriever_tool(
        retriever=retriever,  # 使用 RAG 管理器提供的检索器
        name="rag_search",  # 工具名称
        description=(  # 工具描述
            "按照用户的问题搜索相关的资料，"
            "对于招股书类的问题，you must use this tool!"
        ),
    )

    # 返回创建的 RAG 检索工具
    return retriever_tool
2. 告诉Agent的工作逻辑：Prompt

为FinanceBotEx类添加

def create_prompt():
    """
    创建系统提示（Prompt），用于指导金融助手的行为和工具使用规则。

    返回:
        str: 系统提示字符串。
    """
    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}"
        "行动输入：行动的输入"
        "观察：行动的结果... （这个思考/行动/行动输入/观察的过程可以重复多次）"
        "思考：我现在知道最终答案了"
        "最终答案：原始输入问题的最终答案"

        "Begin!"
    )
    return system_prompt
3. 创建agent

为FinanceBotEx类添加

def init_agent(self):
    """
    初始化 Agent，集成 RAG 工具和 SQL 工具，并设置系统 Prompt。

    返回:
        object: 配置完成的 Agent 执行器。
    """
    # 初始化 RAG 工具，用于检索增强生成（Retrieval-Augmented Generation）
    retriever_tool = self.init_rag_tools()

    # 初始化 SQL 工具，连接到数据库并准备查询功能
    sql_tools = self.init_sql_tool(settings.SQLDATABASE_URI)

    # 创建系统 Prompt 提示语，定义 Agent 的行为规则和工具使用指南
    system_prompt = self.create_prompt()

    # 创建 REACT Agent，结合对话模型、工具列表和系统 Prompt
    agent_executor = create_react_agent(
        llm=self.chat,  # 使用配置的对话模型
        tools=[retriever_tool] + sql_tools,  # 将 RAG 工具与 SQL 工具合并为工具列表
        state_modifier=system_prompt  # 设置系统 Prompt 以指导 Agent 行为
    )

    # 返回配置完成的 Agent 执行器
    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类添加

接收用户的查询请求，通过 Agent 执行器以流式方式逐步生成回答，并将每一步的结果打印出来，最后返回完整的回答内容。

def handle_query(self, example_query):
    """
    处理用户查询，通过流式事件返回结果。

    参数:
        example_query (str): 用户输入的查询内容。

    返回:
        str: 最终生成的回答内容。
    """
    # 使用 Agent 执行器以流式方式处理用户查询
    events = self.agent_executor.stream(
        {"messages": [("user", example_query)]},  # 用户输入的消息
        stream_mode="values",  # 设置流式模式为 "values"
    )

    # 遍历流式事件，并打印每条消息的内容
    for event in events:
        event["messages"][-1].pretty_print()  # 打印当前事件中的最新消息

    # 返回最终生成的回答内容
    return event["messages"][-1].content  # 提取并返回最后一条消息的内容
测试效果

在test_framework.py中增加测试函数如下：

def test_financebot_ex():
    """
    测试 FinanceBotEx 类的主流程。
    
    步骤：
    1. 导入 FinanceBotEx 类。
    2. 初始化语言模型、对话模型和嵌入模型。
    3. 创建 FinanceBotEx 实例。
    4. 使用示例查询测试其处理能力。
    """
    # 导入 FinanceBotEx 类
    from finance_bot_ex import FinanceBotEx

    # 获取通义千问的语言模型、对话模型和嵌入模型
    llm, chat, embed = get_qwen_models()

    # 创建 FinanceBotEx 实例，传入模型参数
    financebot = FinanceBotEx(llm=llm, chat=chat, embed=embed)

    # 定义示例查询问题
    example_query = (
        "20210304日，一级行业为非银金融的股票的成交量合计是多少？取整。"
    )

    # 调用 handle_query 方法处理查询，并打印结果
    financebot.handle_query(example_query)


if __name__ == "__main__":
    """
    程序入口点。
    
    执行以下测试函数之一：
    - test_rag(): 测试 RAG 功能（可选）。
    - test_import(): 测试导入功能（可选）。
    - test_financebot_ex(): 测试 FinanceBotEx 主流程。
    """
    # 注释掉其他测试函数，当前运行主流程测试
    # test_rag()
    # test_import()
    test_financebot_ex()
总结：

1. Agent 的实现思想与传统开发有着极大的不同。传统开发对于程序的每个步骤都要清晰的控制，而Agent的思想是通过Prompt告诉大模型运行规则，然后由大模型自主思考和行动。

2. 使用Agent时的几个步骤：

  • 创建相应的工具
  • 创建对应的prompt
  • 创建agent，并赋予相应的工具和prompt
  • 通过agent的stream来处理输入的问题

http://www.kler.cn/a/568627.html

相关文章：

  • 安当防火墙登录安全解决方案：零信任认证+国密证书+动态口令构建全方位身份安全屏障
  • iOS 实现UIButton自动化点击埋点
  • 从人口焦虑到科技破局：新生人口减少不再是难题，未来社会已悄然蜕变
  • Mysql的索引失效
  • 数据库拓展操作
  • Vim 常用快捷键大全：跳转、编辑、查找替换全解析
  • 委托者模式（掌握设计模式的核心之一）
  • 华为手机自助维修的方法
  • Memcached监控本机内存（比redis速度更快）
  • C++编程指南21 - 线程detach后其注意变量的生命周期
  • leetcode第77题组合
  • next.js-学习4
  • 蓝桥杯 6.数学
  • 基于springboot+vue的线上考试系统的设计与实现
  • 在 Ubuntu 下通过 Docker 部署 Caddy 和 PHP-FPM 服务器
  • Java—锁—等待唤醒机制
  • 随机树算法 自动驾驶汽车的路径规划 静态障碍物（Matlab）
  • thinkphp6-使用psubscribe进行redis的注意callback中使用redis
  • 《Python实战进阶》No 11：微服务架构设计与 Python 实现
  • 字符串的最大公因子＜枚举＞