企业官网建设流程全解析

来源：黑马程序员agent智能体课程项目

一、项目介绍

智扫通Agent项目是一个面向消费者（toc）的智能客服系统，旨在为用户提供全周期的扫地机器人的相关服务。

（1）智能问答服务

·处理购买前的产品咨询（如功能、价格、对比等）。

·解决购买后的使用问题（如操作指导、故障处理、维护建议等）。

·基于RAG技术，从知识库中检索准确信息并生成自然语言回答，确保响应及时可靠。

（2）使用报告与优化建议生成

·针对已购买用户，自动分析扫地机器人的使用数据（如清洁频率、耗材状态、错误日志等）。

·生成个性化报告，总结使用情况并提供建议（如清洁计划调整、部件更换提醒等）。

·支持用户主动查询报告或系统定期推送，帮助用户最大化产品价值

二、项目框图

代码主要分为两部分：一部分是RAG的，一部分是关于智能体的。

RAG部分会有一个向量存储服务（vector_store.py），还会有一个RAG总结服务，（即以下框图中右上角两个代码块），RAG总结服务是指从向量库中拿到参考资料，将资料提供给模型，要求模型生成一个总结概括，然后把总结概括提供出去，最终会形成RAG总结的方法函数。这个函数是作为agent_tools中的其中一环，即RAG也是智能体中工具之一。

主体流程：

首先完成RAG核心流程，并且把RAG封装成一个rag_summarize的方法函数，用于Agent智能体工具，除此以外再提供一些其他工具及中间件，把它们打包构建一个智能体，最终为我们的用户在网页上提供智能体的服务。智能体就可以做到，既可以回答普通问题，又可以生成报告。

RAG知识库需要保存六个数据集，在黑马提供的这个链接下载javacaoyu/AI大模型RAG与Agent智能体开发课程_Agent项目代码 - Gitee.com

三、代码编写

代码每部分，我会先给予全部代码（注释会解释大部分知识），在每个整体代码的下面，会撰写代码着重注意的地方

现在就可以先将data文件夹中的数据集传进去了（直接拖拽就行）。

然后创建utils文件夹，用于放基础工具的开发代码。

1、utils文件夹

① path_tool.py（路径工具）

utils里第一份代码，为整个工程提供统一的绝对路径。

import os def get_project_root() -> str: """ 获取工程所在的根目录 :return: 字符串根目录 """ # 当前文件的绝对路径 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 : """ 传递相对路径，得到绝对路径 :param relative_path:相对路径 :return:绝对路径 """ project_root = get_project_root() return os.path.join(project_root,relative_path) # 通过下面来验证代码正确性 if __name__ == '__main__': print(get_abs_path('config/config.txt'))

总结：

（以当前项目的文件布局）

· os.path.abspath(__file__)：获取当前文件的绝对路径

· os.path.dirname( ) ：传入文件绝对路径，获取文件当前文件夹的绝对路径。

传入文件夹绝对路径，获取工程根目录（即得到上一级路径）

· os.path.join( 工程根目录 , 相对路径) ：传入相对路径，与工程根目录，拼在一起，可以得到绝对路径

② logger_handler.py（日志工具）

可以先大体看一下框架。

import logging from datetime import datetime from utils.path_tool import get_abs_path import os # 日志保存的根目录 存到文件里，就可以长久的保存目录 LOG_ROOT = get_abs_path("logs") # 确保日志的目录存在 os.makedirs(LOG_ROOT,exist_ok=True) # 日志的格式配置 级别：error info debug DEFAULT_LOG_FORMAT = logging.Formatter( # 时间 - 名字 - 级别 - 文件名 ： 文件哪一行 - 正文 '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s' ) def get_logger( name:str = "agent" , # 日志管理器名字，区分不同模块,默认agent console_level : int = logging.INFO, # 日志默认级别 file_lever:int = logging.DEBUG, # 文件级别 log_file = None , ) -> logging.Logger : # 创建日志管理器 logger = logging.getLogger(name) # 设置logger信息 # 设置日志级别 logger.setLevel(logging.DEBUG) # 消息先过 logger 的级别，再分别过 handler（控制台 / 文件）自己的级别 # 避免重复添加Handler if logger.handlers: return logger #保证下面代码只跑一次，避免重复输错 # 控制台Handler console_handler = logging.StreamHandler() # 得到控制台处理器 console_handler.setLevel(console_level) # 设置处理器的级别 console_handler.setFormatter(DEFAULT_LOG_FORMAT) # 输出日志格式 logger.addHandler(console_handler) # 文件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_lever) # 设置文件级别 file_handler.setFormatter(DEFAULT_LOG_FORMAT) # 格式 logger.addHandler(file_handler) return logger # 快捷获取日志管理器 logger = get_logger() # 之后用这个日志管理器，直接from当前文件import logger这个对象就可以用了 # 测试一下 if __name__ == '__main__': logger.info('信息日志') logger.error("错误日志") logger.warning("警告日志") logger.debug("调试日志") """ 级别 NOTSET(0) < DEBUG(10) < INFO(20) < WARNING(30) < ERROR(40) < CRITICAL(50) """

输出：

2026-10-24 15:12:51,841 - agent - INFO - logger_handler.py:61 - 信息日志 2026-10-24 15:12:51,841 - agent - ERROR - logger_handler.py:62 - 错误日志 2026-10-24 15:12:51,841 - agent - WARNING - logger_handler.py:63 - 警告日志

为什么没有“调试日志”的输出呢？

因为默认构建logger里关于控制台的日志级别是info，所以只有info和级别大于info的才会输出显示。

级别：

NOTSET(0) < DEBUG(10) < INFO(20) < WARNING(30) < ERROR(40) < CRITICAL(50)

并且左边会出现logs文件夹，即文件handler，文件中的默认级别为DEBUG，如下图所示，文件会显示DEBUG。

总结：

1、确定目录存在，没有会创建：os.makedirs(路径,exist_ok = True)

2、日志配置格式：logging.Formatter(
# 时间 - 名字 - 级别 - 文件名 ： 文件哪一行 - 正文
'%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s')

3、创建日志管理器对象：logging.getLogger(管理器名字)

4、设置日志/控制器handler/文件handler级别：对象.setLevel(logging.级别)

控制台

5、得到控制台handler对象：logging.StreamHandler()

6、控制台输出日志格式：对象.setFormatter(日志配置格式)

7、添加到管理器：管理器对象.addHandler（handler对象)

文件

8、获取当前时间年月日：datetime_now().strftime("%Y%M%d")

9、构建文件对象：logging.FileHandler(文件路径，encoding='utf-8')

10：文件输出日志格式：同6

11：添加到管理器：同7

③ config_handler.py（配置文件处理）

需要实现四个配置文件：rag的、向量数据库的、提示词的以及agent本身的，对应四个不同的函数。

使用配置文件，可以选用yaml格式，和字典类似形式类似。

首先，创建config文件和四个yml文件，为了一会测试代码正确性，可以先在rag.yml里添加以下代码

chat_model_name : qwen3-max embedding_model_name : text-embedding-v4

下面可以开始代码编写了：

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) # 把 yaml 文件里的文本，翻译成 Python 对象（字典 / 列表） 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'])

总结：

读取解析 yaml 文件：yaml.load( 文件对象 , Loader=加载器 )

loader	特点
yaml.BaseLoader	只解析基础类型（字符串、数字），最严格
yaml.SafeLoader	安全模式，禁止执行 Python 对象、函数，✅ 开发项目首选，对应yaml.safe_load()
yaml.FullLoader	能解析 yaml 全部原生标签，可以序列化 Python 对象，有安全风险，如果 yml 里被人写了恶意代码，会被执行
yaml.UnsafeLoader	完全无限制，风险最高

④ file_handler.py（文件处理工具）

先介绍一下海象运算符 " := "：

作用：一边赋值，一边返回这个值。

普通写法：

data = input("输入：") if data: print(data)

海象写法：

if data := input("输入：") print(data)

import hashlib import os from utils.logger_handler import logger from langchain_core.documents import Document from langchain_community.document_loaders import PyPDFLoader,TextLoader def get_file_md5_hex(filepath:str): # 获取文件的md5的十六进制字符串 if not os.path.exists(filepath): logger.error(f"[md5计算]文件{filepath}不存在") return if not os.path.isfile(filepath): # 如果是文件夹也没法处理，所以再判断一下 logger.error(f"[md5计算]路径{filepath}不是文件") return md5_obj = hashlib.md5() # 得到md5对象 chunk_size = 4096 # 4kb分片，避免文件过大爆内存 try: with open(filepath,'rb') as f : # rb：以二进制方式去读取 必须二进制读取 while chunk := f.read(chunk_size): md5_obj.update(chunk) """ chunk = f.read(chunk_size) while chunk: md5_obj.update(chunk) chunk = f.read(chunk_size) """ md5_hex = md5_obj.hexdigest() return md5_hex except Exception as e: logger.error((f"计算文件{filepath}md5失败，{str(e)}")) """ Exception：代表捕获所有常规类型错误（文件不存在、没有权限、文件损坏等等） as e：把报错信息这个对象，取名叫 e str(e)：把报错信息转成字符串，打印到日志里，告诉你为啥失败 """ return None def listdir_with_allowed_type(path:str,allowed_types:tuple[str]): # 返回文件夹内的文件列表（允许的文件后缀） files = [] if not os.path.isdir(path): logger.error(f"[listdir_with_allowed_type]{path}不是文件夹") return allowed_types for f in os.listdir(path): # 循环文件夹下所有直接子项（文件夹名和文件名） if f.endswith(allowed_types): # f只是名字 files.append(os.path.join(path,f)) return tuple(files) # 转成元组可以防止list被动态修改 def pdf_loader(filepath:str,passwd =None) -> list[Document]: return PyPDFLoader(filepath,passwd).load() # 全量加载 def txt_loader(filepath:str) -> list[Document]: return TextLoader(filepath).load()

总结：

1、判断路径是否为文件夹 :os.path.isdir(路径)

2、列出文件夹下所有直接子项（文件名，文件夹名）：os.listdir(路径)

3、判断字符串后缀：.endswith(后缀)

⑤ prompt_loader.py（提示词加载工具）

在prompts.yml里加入以下三个路径代码

main_prompt_path : prompts/main_prompt.txt rag_summarize_prompt_path : prompts/rag_summarize.txt report_prompt_path : prompts/report_prompt.txt

先创建prompts文件夹，放入三个文件，内容去链接内复制，如下javacaoyu/AI大模型RAG与Agent智能体开发课程_Agent项目代码 - Gitee.com

from utils.config_handler import prompts_conf from utils.path_tool import get_abs_path from utils.logger_handler import logger def load_system_prompts(): try: system_prompt_path = get_abs_path(prompts_conf["main_prompt_path"]) except KeyError as e: logger.error(f"[load_system_prompts]在yaml配置项中没有main_prompt_path配置项") raise e try: return open(system_prompt_path,"r",encoding="utf-8").read() except Exception as e: logger.error(f"[load_system_prompts]解析系统提示词出错，{str(e)}") raise e def load_rag_prompts(): try: rag_prompt_path = get_abs_path(prompts_conf["rag_summarize_prompt_path"]) except KeyError as e: logger.error(f"[load_system_prompts]在yaml配置项中没有rag_summarize_prompt_path配置项") raise e try: return open(rag_prompt_path, "r", encoding="utf-8").read() except Exception as e: logger.error(f"[load_system_prompts]解析rag总结提示词出错，{str(e)}") raise e def load_report_prompts(): try: report_prompt_path = get_abs_path(prompts_conf["report_prompt_path"]) except KeyError as e: logger.error(f"[load_system_prompts]在yaml配置项中没有report_prompt_path配置项") raise e try: return open(report_prompt_path, "r", encoding="utf-8").read() except Exception as e: logger.error(f"[load_system_prompts]解析报告生成提示词出错，{str(e)}") raise e # 测试 if __name__ == "__main__": print(load_report_prompts())

总结：

raise会将异常扔出去，若有上层接住，即

# 上层调用 try: load_system_prompts() except KeyError: print("上层捕获到配置缺失，我自己处理，不崩溃")

这时 raise 抛出来的异常被上层接住了，程序不会暂停，会跑 except 里面的逻辑。否则程序报错、停止运行

好啦！现在完成了五个工具代码的编写，下一节将会进行向量存储服务的开发，我们一起努力吧！