Day9:从零开始写一个Agent项目(1)
让我们开始吧。。。
嘿呦,这里是惬鹤频道!
继上次的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中指向提示词配置文件的变量读取成文本,
也就是我们人类看到的提示词。
结尾
这就是本次项目中重要的五个工具文件的展示,后续的各个部分都会使用到这些工具。
好了,过两天我会继续更新的,感谢大家的支持!下期再见了!
更多推荐
所有评论(0)