让我们开始吧。。。

嘿呦,这里是惬鹤频道!
继上次的RAG小项目落幕后,这次我又给大家带来了一个新的项目,隆重欢迎:

与RAG相结合的,带有历史会话记忆,用户界面的专业Agent项目!

这会是我这几天要忙个不停的事情。我依旧会按照学习进度,分几次给大家分享,希望大家喜欢。
(喜欢的话可以点点赞,谢谢!)
先附上这次项目的文件结构图(从教程中截图的),让大家看看这次分享的是哪一部分:

每次都会有的。。。总体结构图

在这里插入图片描述
吓哭了,反正我看到这个结构图的第一反应就是这样的。哈哈
好了说回正题。
这次要给大家分享的是五个文件,都是项目需要使用的工具(结构图右侧“utils”的五个工具)
下面我们开始吧!

代码文件:config_handler.py

"""
文件:config_handler.py
描述:之前一直使用一个普通的py文件进行配置,这里准备使用yaml。
"""
import yaml
from utils.path_tool import get_abs_path

def load_rag_config(config_path: str = get_abs_path("config/rag.yml"), encoding="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

def load_chroma_config(config_path: str = get_abs_path("config/chroma.yml"), encoding="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

def load_prompts_config(config_path: str = get_abs_path("config/prompts.yml"), encoding="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

def load_agent_config(config_path: str = get_abs_path("config/agent.yml"), encoding="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

# 等到文件传入,就可以使用这四个变量了。
rag_conf = load_rag_config()
chroma_conf = load_chroma_config()
prompts_conf = load_prompts_config()
agent_conf = load_agent_config()

if __name__ == '__main__':
    print(rag_conf["chat_model_name"])

这个文件用于加载各种模块的配置文件
之前我一直用一个普通的py文件进行配置,但是那样不够专业,这次在教程的帮助下我学着换成了yaml,
这样各种配置信息可以放在不同的yaml文件中,并统一加载,这样看起来结构分明,很舒服。
配置文件还没有全部写完,这里先把名字放出来:
agent.yml
chroma.yml
prompts.yml
rag.yml
嗯哼,光看名字你也知道这些是谁的配置文件,不是吗?

代码文件:file_handler.py

"""
文件名:file_handler.py
描述:这个文件用于一件重要的工作:把文件转换为MD5格式并存储,用于后续的去重。
"""
# 导入依赖
import os
import hashlib
from utils.logger_handler import logger
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader

# 方法:读取,转换文件为MD5格式
def get_file_md5_hex(file_path: str):

    # 错误一:路径读取失败
    if not os.path.exists(file_path):
        logger.error(f"[读取文件]读取路径:{file_path}时失败!")
        return

    # 错误二:对象不是可处理类型
    if not os.path.isfile(file_path):
        logger.error(f"[读取文件]路径:{file_path}对应的对象不是文件!")
        return

    # 通过审核,开始转换。
    chunk_size = 4096
    md5_obj = hashlib.md5()
    try:
        # 读取文件
        with open(file_path, "rb") as f:
            while chunk := f.read(chunk_size):
                md5_obj.update(chunk)
        # 将md5对象转换为16进制
        md5_hex = md5_obj.hexdigest()
        return md5_hex

    except Exception as e:
        logger.error(f"[转换文件]路径:{file_path}对应的文件转换失败,错误信息:{str(e)}")

    return None

# 方法:判断文件夹内的文件有哪些允许被处理
def listdir_with_allowed_type(file_path: str, allowed_types: tuple[str]):
    file = []

    # 判断是不是文件夹
    if not os.path.isdir(file_path):
        logger.error(f"[文件类型]文件路径:{file_path}指向的不是文件夹。")
        return allowed_types
    # 是文件夹,开始访问内部文件的后缀
    for f in os.listdir(file_path):
        if f.endswith(allowed_types):
            # 符合输入的类型条件的,写入file
            file.append(os.path.join(file_path, f))
    # 返回元组类型,防止外部修改
    return tuple(file)

# 方法:PDF格式加载器,返回文档的list类型
def pdf_loader(file_path: str, password: str) -> list[Document]:
    return PyPDFLoader(file_path,password=password).load()

# 方法:txt格式加载器,返回文档的list类型
def txt_loader(file_path: str) -> list[Document]:
    return TextLoader(file_path).load()

这个工具和之前RAG项目的一个文件knowledge_base.py作用有点像,都可以将文本转换成md5格式,但是目前还没有去重功能。
由于是工具文件,它同时也包含了两种后续要用到的加载器(txt的,和pdf的)以及一个用于指定“允许进入的文件类型”的方法。
关于代码中的logger从哪来,接下来这个工具展示了它的来源。

代码文件:logger_handler.py

"""
文件:logger_handler
描述:用于创建日志
"""
# 导入依赖
import logging
import os
from utils.path_tool import get_abs_path
from datetime import datetime

# 获取日志的根文件路径
LOG_ROOT = get_abs_path("logs")

# 配置日志文件(没有则创建)
os.makedirs(LOG_ROOT, exist_ok=True)

# 配置日志格式
DEFAULT_LOG_FORMAT = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s"
)

# 方法:创建日志
def get_logger(
        name: str = "logger",
        console_level: int = logging.INFO,
        file_level: int = logging.DEBUG,
        log_file = None
) -> logging.Logger:

    # 定义日志格式
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 判断是否有重复handler
    if logger.handlers:
        return logger

    # 创建控制台日志
    console_handler = logging.StreamHandler()
    console_handler.setLevel(console_level)
    console_handler.setFormatter(DEFAULT_LOG_FORMAT)

    # 添加日志
    logger.addHandler(console_handler)

    if not log_file:
        log_file = os.path.join(LOG_ROOT, f"{name}_{datetime.now().strftime("%Y%m%d")}.log")

    # 创建文件日志
    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    file_handler.setLevel(file_level)
    file_handler.setFormatter(DEFAULT_LOG_FORMAT)

    logger.addHandler(file_handler)

    return logger

# 快捷获取日志
logger = get_logger()

if __name__ == '__main__':
    logger.info("信息日志"),
    logger.error("错误日志")
    logger.warning("警告日志")
    logger.debug("调试日志")

这个文件涉及日志相关的知识,基本结构就是一个用于创建日志的方法get_logger,内部指定多种参数:
name:日志器名称(默认 “logger”)
console_level:控制台输出的最低日志级别(默认 logging.INFO)
file_level:文件输出的最低日志级别(默认 logging.DEBUG)
log_file:可自定义日志文件路径;若不提供,则自动生成在 LOG_ROOT 下,文件名为 {name}_{YYYYMMDD}.log

说白了就是一个日志工具模块,能帮我们快速创建一个既能打印到控制台、又能自动保存到文件的日志记录器。
它的日志所在文件夹在项目根目录下的 logs/ 中。文件名包含当天日期。

代码文件:path_tool.py

"""
文件:path_tool.py
描述:定义一些简单的方法
"""

# 导入依赖
import os

# 方法:获取工程所在的根目录
def get_project_root() -> str:
    # 获取当前代码文件的路径
    current_file = os.path.abspath(__file__)
    # 获取当前文件所在的文件夹路径
    current_dir = os.path.dirname(current_file)
    # 再往上跳一级,到达项目所在文件夹
    project_root = os.path.dirname(current_dir)
    return project_root

# 方法:传入相对路径,返回绝对路径
def get_abs_path(relative_path: str) -> str:
    # 获取绝对路径
    project_root = get_project_root()
    # 和相对路径整合后返回
    return os.path.join(project_root, relative_path)

# 测试
if __name__ == '__main__':
    print(get_abs_path("config/config.txt"))

这个文件中的方法很简单,但是用途非常广泛。
它定义了两个方法:获取当前代码文件的路径,和传入相对路径后返回绝对路径,主要避免了因传入路径不完整可能导致的混乱。
代码末尾有测试样例,可以试试看。

代码文件:prompt_loader.py

"""
文件名:prompt_loader.py
描述:用于读取各种提示词
"""
# 导入依赖
from utils.logger_handler import logger
from utils.config_handler import prompts_conf
from utils.path_tool import get_abs_path

# 加载系统提示词
def load_system_prompts():
    # 尝试从yml文件中读取提示词
    try:
        system_prompts_path = get_abs_path(prompts_conf["main_prompt_path"])
    except KeyError as e:
        logger.error(f"[读取提示词]在prompts.yml中没有main_prompt_path这一项")
        raise e

    # 可以找到对应的提示词,开始读取
    try:
        return open(system_prompts_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[读取提示词]读取系统提示词失败!{str(e)}")
        raise e

# 加载RAG提示词
def load_rag_prompts():
    # 尝试从yml文件中读取提示词
    try:
        rag_prompts_path = get_abs_path(prompts_conf["rag_summarize_prompt_path"])
    except KeyError as e:
        logger.error(f"[读取提示词]在prompts.yml中没有rag_summarize_prompt_path这一项")
        raise e

    # 可以找到对应的提示词,开始读取
    try:
        return open(rag_prompts_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[读取提示词]读取RAG提示词失败!{str(e)}")
        raise e

# 加载报告提示词
def load_report_prompts():
    # 尝试从yml文件中读取提示词
    try:
        report_prompts_path = get_abs_path(prompts_conf["report_prompt_path"])
    except KeyError as e:
        logger.error(f"[读取提示词]在prompts.yml中没有report_prompt_path这一项")
        raise e

    # 可以找到对应的提示词,开始读取
    try:
        return open(report_prompts_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[读取提示词]读取报告提示词失败!{str(e)}")
        raise e

# 测试
if __name__ == '__main__':
    print(load_system_prompts())
    print(load_rag_prompts())
    print(load_report_prompts())

最后一个,这个文件与最开始的config_handler.py结合,可以真正的把config_handler.py中指向提示词配置文件的变量读取成文本,
也就是我们人类看到的提示词。

结尾

这就是本次项目中重要的五个工具文件的展示,后续的各个部分都会使用到这些工具。
好了,过两天我会继续更新的,感谢大家的支持!下期再见了!

更多推荐