位置：首页 > 技术资讯 > Agent驱动的金融问答系统代码重构与优化实践

Agent驱动的金融问答系统代码重构与优化实践

时间：2026-08-18  |  作者：星河游者  |  阅读：0

在上一章【项目实战】基于Agent的金融问答系统：前后端流程打通，我们已经完成了金融问答系统的前后端搭建，形成了可用的Demo。

本章，我们将介绍代码重构的过程，并介绍一些优化点。

代码重构简介

在正式进入主题之前，先聊聊代码重构这件事。

回顾过去参与过的项目，代码重构几乎从未被认真对待过。

那些像天书一样的代码，包括自己曾经写出来的东西，回头看真是让人不忍直视。它们没有被清理，反而在这堆“遗产”上不断叠加新需求。

结果就是：迭代越来越艰难，Bug越来越多。

这种情况直到今天还在反复发生，所以确实有必要专门说一说代码重构。

什么是代码重构

代码重构是指对现有代码进行修改，以改善其结构、可读性和可维护性，而不改变其外部行为。

重构的主要目的是提高代码质量，使其更易于理解和扩展。

代码重构的目的

  • 提高可读性：使代码更易于理解，便于团队成员快速上手。
  • 增强可维护性：降低后续修改和扩展的难度，减少潜在的错误。
  • 优化性能：在不改变功能的情况下，提升代码的执行效率。
  • 消除重复代码：通过抽象和重用，减少冗余，提高代码的整洁性。

代码重构的重要性

据统计，不好的代码会占用更多开发的时间。

Agent驱动的金融问答系统代码重构与优化实践_wishdown.com

代码重构的难点

既然代码重构对提升代码质量如此重要，那为什么实际项目中很少开展呢？

说到底，原因可能有三点：

  • 第一种：没有精力重构。

    开发工程师往往被老板或产品牵着鼻子走，一个需求刚结束，下一个需求就来了。

    这种情况在技术导向的项目中还好，一旦遇到产品或市场驱动的项目，尤为严重。

  • 第二种：没有重构的思维。

    很多开发工程师并没有重构的意识，他们的目标是完成需求交付，交付即意味着工作结束。

    有一位从谷歌回来的朋友曾和我聊起：为什么国内研发人员的代码质量意识薄弱？

    他提到一个关键点：硅谷的很多从业者，是因为热爱才做编程——热爱技术、追求代码如艺术品般的完美；而国内不少人选择开发岗位，更多的是出于生计，工资相对较高，能养家糊口。

    缺少热爱，自然交差了事。这一点我深有同感。

  • 第三种：没有重构的方法论。

    即使有心去做重构，这项工作就像修理一辆越开越慢的车子。

    如果没有科学的方法，很可能拆完重装后，反而多出来几个螺丝，让老板更加头疼。

本章，我们就以这个金融问答系统为例，简单梳理一下代码重构的原则和方法。

代码重构的过程

1、搭建测试框架以及用例集

在动手重构之前，先搭建一个便于回归测试的测试框架。

边重构边回归，能快速定位问题，降低排查成本。

我们在app目录下已经创建了一个test_framework.py，继续补充测试用例集，例如：

Agent驱动的金融问答系统代码重构与优化实践_wishdown.com

在大厂中，回归测试一般会使用单元测试框架（如pytest）来执行。

由于本例的方法较为简单，所以没有引入pytest。

2、消灭代码中的坏味道

2.1、统一管理配置相关内容

在之前实现的RAG管理模块中，有很多配置是硬编码写在代码初始化里的，例如：

# 原始的rag.py
class RagManager:
    def __init__(self,
                 chroma_server_type="http",
                 host="localhost", port=8000,
                 persist_path="chroma_db",
                 llm=None, embed=None):
        self.llm = llm
        self.embed = embed

        chrom_db = ChromaDB(chroma_server_type=chroma_server_type,
                            host=host, port=port,
                            persist_path=persist_path,
                            embed=embed)
        self.store = chrom_db.get_store()

我们可以将所有的配置抽取到一个settings.py中。

然后在使用的代码中通过引用settings.py来获取配置。

# setttings.py

"""
Chroma向量数据库使用时的相关的配置
"""
# 默认的ChromaDB的服务器类别
CHROMA_SERVER_TYPE ="http"
# 默认本地数据库的持久化目录
CHROMA_PERSIST_DB_PATH ="chroma_db"

CHROMA_HOST = os.getenv("CHROMA_HOST","localhost")
CHROMA_PORT = int(os.getenv("CHROMA_PORT",8000))
CHROMA_COLLECTION_NAME ="langchain"

说明：为了与普通变量区分，配置项我们统一使用大写命名，例如CHROMA_HOST、CHROMA_PORT等。

# 重构的rag.py
import settings


class RagManager:
    def __init__(self,
                 vector_db_class=ChromaDB,  # 默认使用 ChromaDB
                 db_config=None,            # 数据库配置参数
                 llm=None, embed=None,
                 retriever_cls=SimpleRetrieverWrapper, **retriever_kwargs):
        self.llm = llm
        self.embed = embed
        logger.info(f'初始化llm大模型：{self.llm}')
        logger.info(f'初始化embed模型：{self.embed}')

        # 如果没有提供 db_config，使用默认配置
        if db_config is None:
            db_config = {
                "chroma_server_type": settings.CHROMA_SERVER_TYPE,
                "host": settings.CHROMA_HOST,
                "port": settings.CHROMA_PORT,
                "persist_path": settings.CHROMA_PERSIST_DB_PATH,
                "collection_name": settings.CHROMA_COLLECTION_NAME
            }
            logger.info(f'初始化向量数据库配置：{db_config}')

        # 创建向量数据库实例
        self.vector_db = vector_db_class(**db_config, embed=self.embed)
        self.store = self.vector_db.get_store()

说明：通过import settings，使用settings.CHROMA_SERVER_TYPE等来引用配置，清晰且易于管理。

2.2、处理参数过长的问题

原始代码中，随着需求迭代，创建RAG时需要传入多个参数，例如：

  • chroma_server_type
  • host
  • port
  • persist_path
  • collection_name

如果按照原来的写法，函数参数列表会变得非常长：

RagManager(chroma_server_type="http", host="localhost", port=8000, persist_path="chroma_db", collection_name="langchain", llm, embed)

对于这类参数问题，可以用字典来处理：

db_config = {
    "chroma_server_type": settings.CHROMA_SERVER_TYPE,
    "host": settings.CHROMA_HOST,
    "port": settings.CHROMA_PORT,
    "persist_path": settings.CHROMA_PERSIST_DB_PATH,
    "collection_name": settings.CHROMA_COLLECTION_NAME,
}

RagManager(vector_db_class=ChromaDB, db_config=db_config, llm=self.llm, embed=self.embed)

说明：db_config是一个字典，可包含多个配置参数；通过**解包传递到函数中，简洁且可扩展。

2.3、减少重复代码

在之前的章节中，我们实现了pdf_processor.py，主要工作包括：

def process_pdfs(self)      # 处理pdf文件
def process_pdfs_group(self, pdf_files_group)  # 分组处理pdf文件
def load_pdf_files(self)    # 加载pdf文件
def load_pdf_content(self, pdf_path)  # 读取pdf文件内容
def split_text(self, documents)       # 分割读取到的文本
def insert_docs_chromadb(self, docs, batch_size)  # 向向量数据库中插入数据

如果要将PDF文件导入ElasticSearch服务，大部分实现逻辑是一样的。

唯一区别只是插入对象不同——一个是向量数据库，一个是ElasticSearch。

这种情况下：

  • 不好的做法：复制上述代码到一个新函数，然后将最后的insert_docs_chromadb()改为insert_docs_elasticsearch()，导致大量重复。
  • 较好的做法：对插入过程进行重构，将具体的插入函数通过参数传入，用db_type决定插入目标。
# 重构后的pdf_processor.py

import os
import logging
import time
from tqdm import tqdm
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from rag.vector_db import VectorDB
from rag.elasticsearch_db import TraditionDB
from utils.logger_config import LoggerManager

logger = LoggerManager().logger


class PDFProcessor:
    def __init__(self, directory, db_type='vector', **kwargs):
        """
        初始化 PDF 处理器
        :param directory: PDF 文件所在目录
        :param db_type: 数据库类型 ('vector' 或 'es')
        :param kwargs: 其他参数
        """
        self.directory = directory
        self.db_type = db_type
        self.file_group_num = kwargs.get('file_group_num', 20)
        self.batch_num = kwargs.get('batch_num', 6)
        self.chunksize = kwargs.get('chunksize', 500)
        self.overlap = kwargs.get('overlap', 100)
        logger.info(f"""
                    初始化PDF文件导入器:
                    配置参数：
                    - 导入的文件路径：{self.directory}
                    - 每次处理文件数：{self.file_group_num}
                    - 每批次处理样本数：{self.batch_num}
                    - 切分文本的大小：{self.chunksize}
                    - 切分文本重叠大小：{self.overlap}
                    """)

        if db_type == 'vector':
            self.vector_db = kwargs.get('vector_db')
            self.es_client = None
            logger.info(f'导入的目标数据库为：向量数据库')
        elif db_type == 'es':
            self.vector_db = None
            self.es_client = kwargs.get('es_client')
            logger.info(f'导入的目标数据库为：ES数据库')
        else:
            raise ValueError("db_type must be either 'vector' or 'es'.")

    def load_pdf_files(self):
        # 这部分代码未做修改，具体内容省略

    def load_pdf_content(self, pdf_path):
        # 这部分代码未做修改，具体内容省略

    def split_text(self, documents):
        # 这部分代码未做修改，具体内容省略

    def process_pdfs(self):
        # 这部分代码未做修改，具体内容省略

    def insert_docs(self, docs, insert_function, batch_size=None):
        """
        将文档插入到指定的数据库，并显示进度
        :param docs: 要插入的文档列表
        :param insert_function: 插入函数
        :param batch_size: 批次大小
        """
        if batch_size is None:
            batch_size = self.batch_num

        logging.info(f"Inserting {len(docs)} documents.")
        start_time = time.time()
        total_docs_inserted = 0

        total_batches = (len(docs) + batch_size - 1) // batch_size

        with tqdm(total=total_batches, desc="Inserting batches", unit="batch") as pbar:
            for i in range(0, len(docs), batch_size):
                batch = docs[i:i + batch_size]
                insert_function(batch)

                total_docs_inserted += len(batch)

                elapsed_time = time.time() - start_time
                if elapsed_time > 0:
                    tpm = (total_docs_inserted / elapsed_time) * 60
                    pbar.set_postfix({"TPM": f"{tpm:.2f}"})

                pbar.update(1)

    def insert_to_vector_db(self, docs):
        """
        将文档插入到 VectorDB
        """
        self.vector_db.add_with_langchain(docs)

    def insert_to_elasticsearch(self, docs):
        """
        将文档插入到 Elasticsearch
        """
        self.es_client.add_documents(docs)

    def process_pdfs_group(self, pdf_files_group):
        pdf_contents = []

        for pdf_path in pdf_files_group:
            documents = self.load_pdf_content(pdf_path)
            pdf_contents.extend(documents)

        docs = self.split_text(pdf_contents)

        if self.db_type == 'vector':
            self.insert_docs(docs, self.insert_to_vector_db)
        elif self.db_type == 'es':
            self.insert_docs(docs, self.insert_to_elasticsearch)
        else:
            raise ValueError("db_type must be either 'vector' or 'es'.")

说明：通过参数db_type和传入的数据库实例，将插入逻辑与数据处理逻辑解耦，实现了代码复用。

2.4、使用静态扫描工具优化代码风格

可以利用静态扫描工具（如Pylint、Flake8）来优化代码风格，PyCharm中通常自带这些功能。

具体方法：

  • 1. 启动PyCharm
  • 2. 打开工程时，选择app目录
  • 3. 打开任意.py文件后，右上角会有静态扫描问题提示（如下图）
  • 4. 根据提示进行代码风格修正（常见问题请参考附录）
Agent驱动的金融问答系统代码重构与优化实践_wishdown.com

3、回归测试

上述每一步重构完成后，都必须使用test_framework.py进行回归测试。

这样才能确保修改没有引入新错误。

本项目重构细节很多，无法一一列举，完整代码请查看Gitee或Github仓库。

内容小结

  • 代码重构是提升代码质量的关键手段，能改善可读性、可维护性，为后续迭代奠定基础。
  • 重构的一般步骤：
    • 重构前，先准备好测试框架和测试用例。
    • 重构时，逐模块优化并同步进行单元测试。
    • 完成一个模块的测试后再进入下一个模块。
    • 整体重构完成后，进行全面测试，确保无回归问题。
  • 常见优化方向：
    • 统一配置管理，避免硬编码。
    • 减少重复代码，利用函数、类进行封装。
    • 参数过长时采用字典或元组结构。
    • 使用静态扫描工具保持代码风格一致。
    • 其他……

《夸克》非常好用的免费AI浏览器

下载APP查看

免责声明：文中图文均来自网络，如有侵权请联系删除，心愿游戏发布此文仅为传递信息，不代表心愿游戏认同其观点或证实其描述。

相关文章

更多

精选合集

更多

大家都在玩

热门话题

大家都在看

更多
手机版 | 关于我们 | 电脑版 | 返回顶部
本站所有游戏都由网友上传，如有侵犯您的版权
请发邮件wishdown2023@163.com