【AI大模型应用开发】【项目实战】31.Agent智扫引擎项目-(五)模型评估与日志打印

【AI大模型应用开发】【项目实战】31.Agent智扫引擎项目-(五)模型评估与日志打印 一.模型评估模型评估模块具体位置如下:/evals.py1.评估模块介绍整体内容如下:eval_data.csv里面的数据如下:query,reference_docs,reference_answer 这个扫地机器人能扫哪些地面?,['瓷砖、木地板、短毛地毯等,长毛地毯需确认机器越障能力。'],瓷砖、木地板、短毛地毯等,长毛地毯需确认机器越障能力。 dToF导航技术是什么?,['直接飞行时间测距(direct Time-of-Flight),比传统 LDS 测距更精准,探测距离可达 10 米。'],直接飞行时间测距(direct Time-of-Flight),比传统 LDS 测距更精准,探测距离可达 10 米。 app里面清洁记录作用是什么,['查看历史清洁面积、时长,评估清洁效率和耗材寿命。'],查看历史清洁面积、时长,评估清洁效率和耗材寿命。 它能使用哪些语音助手,['常见的有小爱同学、天猫精灵、Google Assistant 等。'],常见的有小爱同学、天猫精灵、Google Assistant 等。 HEPA 滤网要换吗?,['建议每 3-6 个月更换,水洗会降低过滤效果。'],建议每 3-6 个月更换,水洗会降低过滤效果。 异常报警提示咋整?,['查看错误代码,常见问题 APP 会有解决方案指引。'],查看错误代码,常见问题 APP 会有解决方案指引。 地毯多的家庭怎么使用,['确认地毯识别和自动增压功能,吸力≥2500Pa。'],确认地毯识别和自动增压功能,吸力≥2500Pa。 要买高端机型吗?,['适合追求自动化程度和免维护体验的用户。'],适合追求自动化程度和免维护体验的用户。 什么是自动集尘原理?,['基站内产生强大吸力,将尘盒垃圾吸入密封集尘袋。'],基站内产生强大吸力,将尘盒垃圾吸入密封集尘袋。 太阳能充电扫地机器人能使用吗?,['目前效率太低,仅限概念产品。'],目前效率太低,仅限概念产品。 拖地时怎么避免水渍?,['调节合适出水量,使用专用拖地模式。'],调节合适出水量,使用专用拖地模式。 主刷要换吗,['胶刷 1-2 年,毛刷 6-12 个月检查磨损情况。'],胶刷 1-2 年,毛刷 6-12 个月检查磨损情况。 怎么避免泄露家庭地图隐私?,['选择本地存储地图的机型,关闭云同步功能。'],选择本地存储地图的机型,关闭云同步功能。 适合商业场所吗?,['适合小面积场所,大面积需专业商用机型。'],适合小面积场所,大面积需专业商用机型。 跟洗地机相比效果怎样?,['洗地机清洁力更强但需人工操作,机器人自动化程度高'],洗地机清洁力更强但需人工操作,机器人自动化程度高 耗电吗,['很小,一次清扫约耗电 0.1-0.3 度。'],很小,一次清扫约耗电 0.1-0.3 度。 中国市场普及了吗,['2015 年后随着小米生态链产品兴起快速普及。'],2015 年后随着小米生态链产品兴起快速普及。 机器人会自杀吗?,['传感器故障时可能从高处跌落,但非常罕见。'],传感器故障时可能从高处跌落,但非常罕见。本项目的评估主要分为三类:检索内容的评估:这里使用precision(精确率),recall(召回率),mrr(平均倒数排名)三个数据进行评估对生成内容的评估(LLM的生成能力):使用的评估指标是ROUGE-1、ROUGE-L、BELU、BERTScore、Faithfulness(忠实度)总的性能评估:包括平均延时以及qps2.代码实现(1).导入依赖包# 1 导入依赖包 # 导入了用于文本评估的库(ROUGE、BERTScore)、机器学习评估库(Scikit-learn)、NLP 基础库(Jieba、Transformers) # 以及业务所需的 Agent 和工具类,同时加载了停用词表 import json import time import numpy as np import re from rouge_score import rouge_scorer from transformers import AutoModel, AutoTokenizer import torch # 如果没有安装bert_score, 则先安装: pip install bert-score from bert_score import BERTScorer from bert_score.utils import lang2model, model2layers from sklearn.metrics import precision_score, recall_score # pip install scikit-learn import pandas as pd import jieba import sys, os sys.path.append('../..') sys.path.append('.') sys.path.append('./') from Agent.ReAct import ReActAgent from Models.Factory import ChatModelFactory from Tools.Tools import * from Tools.RagQATool import rag_qa from langchain_community.chat_message_histories.in_memory import ChatMessageHistory from tqdm import tqdm import random import ast(2).准备评测数据# 多维度评估: # RAG 系统的评估不能仅靠单一指标 # ROUGE 看字面重合 # BERTScore 看语义相似度 # Faithfulness 看防幻觉能力 # 停用词过滤: # 加载 stopwords.txt 是为了在计算忠实度(Faithfulness)时,过滤掉“的、了、在”等无意义词汇,只关注核心实体,使评估更准确 device = "cuda" if torch.cuda.is_available() else "cpu" stopwords = [] with open(os.path.join(os.getcwd(), "stopwords.txt"), 'r', encoding='utf-8') as f: for line in f: stopwords.append(line.strip()) # 2. 准备评测数据 # 读取 CSV 评测集 eval_data = pd.read_csv(os.path.join(os.getcwd(), 'Eval/eval_data.csv'))(3).获取agent# 3.获取agent # 初始化大模型、工具集和 ReAct Agent llm = ChatModelFactory().get_model() # 获取工具 tools = [ rag_qa_tool, report_generation_tool, ] # Agent主逻辑: ReAct agent = ReActAgent( llm=llm, tools=tools, main_prompt_file_path="prompts/main.txt" ) # 封装 launch_agent 函数,每次调用时创建一个全新的空对话历史 def launch_agent(query, agent, uuid=1001): """ 启动智能代理执行查询任务 参数: query (str): 用户查询字符串 agent: 智能代理实例 uuid (int, optional): 用户唯一标识符,默认为1001 返回值: str: 代理执行后的回复内容 """ chat_history = ChatMessageHistory() # 智能体Agent回复 reply = agent.execute(query, uuid, chat_history=chat_history, verbose=True) return reply(4).评测指标计算函数(4.1).检索评估指标# 4. 评测指标计算函数 # 4.1 检索评估指标 def evaluate_retrieval(true_docs, retrieved_docs): """ 计算检索模块评估指标 (1)Precision@5:1.0 含义:系统针对每个查询返回的前5个检索结果(文档片段)中,全部都是真正相关的。 解读:检索的精确度极高,没有返回无关的噪声信息,保证了后续生成步骤输入源的高质量。 (2)Recall@5:1.0 含义:所有真正相关的文档片段,全部被包含在返回的前5个结果中。 解读:检索的召回能力极强,没有遗漏关键信息。结合Precision@5=1.0,表明检索模块在Top-5的设定下达到了完美表现。 (3)MRR:1.0 含义:平均倒数排名。对于每个查询,系统将第一个相关结果排在第1位。 解读:这不仅说明检索结果相关,而且表明系统能精准地将最相关的结果排序在最前面,极大地优化了信息获取效率。 参数: true_docs (list): 真实文档列表 retrieved_docs (list): 检索到的文档列表 返回值: dict: 包含Precision@5、Recall@5和MRR指标的字典 """ # 1. 防御性处理:如果检索结果为空,直接返回 0 if not retrieved_docs: return {"Precision@5": 0.0, "Recall@5": 0.0, "MRR": 0.0} # 2. 兼容字典和列表:如果 true_docs 是字典,提取它的值或键作为真实集合 if isinstance(true_docs, dict): # 根据你的实际数据结构,提取字典的值(假设值是文档ID)或键 true_set = set(true_docs.values()) else: true_set = set(true_docs) # 真实值 y_true = [1 if doc in true_set else 0 for doc in retrieved_docs] # True # 预测值 y_pred = [1] * len(retrieved_docs) # all # 衡量“找得准不准”: # 检索是 RAG 的灵魂, Precision 衡量有没有混入垃圾文档(抗噪能力), # Recall 衡量有没有漏掉关键文档(覆盖能力), # MRR 衡量最相关的文档是不是排在第一位(排序质量) # 利用 Scikit-learn 计算精确率和召回率 precision = precision_score(y_true, y_pred, zero_division=0) recall = recall_score(y_true, y_pred) # 手动计算MRR(平均倒数排名) # mrr = 1 / (true_docs.index(retrieved_docs[0]) + 1) if retrieved_docs[0] in true_docs else 0 # 4. 计算 MRR (平均倒数排名) mrr = 0.0 for i, doc in enumerate(retrieved_docs): if doc in true_set: mrr = 1.0 / (i + 1) break # MRR 只关心第一个命中的结果 return { # F1(调和平均数) = P + R "Precision@5": precision, # 精确率, 前5个都是预测对的, 命中率 "Recall@5": recall, # 召回率, 严谨精密 "MRR": mrr # 平均倒数排名 }(4.2).BERT评分器# 4.2 BERT评分器 def load_bert_scorer(): """ 加载并初始化中文BERT评分器 返回值: BERTScorer: 配置好的BERT评分器实例 """ # 加载自定义模型和分词器 model_path = 'bert-base-chinese' tokenizer = AutoTokenizer.from_pretrained(model_path) # 手动加载模型配置 # .get()用法? # 模型的层数 num_layers = model2layers.get(os.path.basename(model_path), 12) # 模型类型 model_type = lang2model.get('zh', 'bert-base-chinese') model = AutoModel.from_pretrained(model_path) return BERTScorer( model_type=model_type, num_layers=num_layers, lang='zh', use_fast_tokenizer=True, rescale_with_baseline=False, device=model.device )(4.3).文本生成质量评估指标# 4.3 文本生成质量评估指标 def evaluate_generation(precision, reference, context): """ 计算文本生成质量评估指标: 分别计算了字面相似度(ROUGE)、语义相似度(BERTScore)和基于实体交集的忠实度(Faithfulness) 为什么要这么做: 弥补 ROUGE 的缺陷:大模型喜欢意译和总结,导致 ROUGE 分数通常很低。BERTScore 通过向量空间计算,能识别出“意思一样但说法不同”的情况。 防幻觉检测:通过提取参考文本的实体,看生成文本是否包含了这些实体。这是一种轻量级的幻觉检测手段 :param precision: precision (str): 生成的文本(预测文本) :param reference: reference (str): 参考文本(真实文本) :param context: context (list): 上下文文档列表 :return: dict: 包含ROUGE-1、ROUGE-L、BERTScore和Faithfulness指标的字典 """ """ (1)ROUGE-1 / ROUGE-L:0.222 含义:基于N-gram重叠率的自动文本摘要评估指标。ROUGE-1看单词重叠,ROUGE-L看最长公共子序列。值越接近1,与参考答案的表面相似度越高。 解读:得分较低,表明生成答案在词汇选择和表面形式上与标准答案差异较大。这可能是由于模型进行了意译、总结或结构重组,而非直接复制原文片段。 """ # ROUGE scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=False) rouge = scorer.score(reference, precision) # reference表示真实结果, precision表示预测结果 """ (2)BERTScore:0.768 含义:基于BERT模型上下文嵌入的相似度计算。它评估生成答案与参考答案在语义层面的相似度。 解读:得分处于中等偏上水平。虽然表面文字匹配度低(ROUGE低),但核心语义的匹配度尚可。这说明生成答案在表达上不同,但传达了一部分关键意思。 """ # BERTScore bert_scorer = load_bert_scorer() _, _, bert_f1 = bert_scorer.score([precision], [reference]) """ (3)Faithfulness:0.833 含义:忠实度,评估生成答案的内容是否严格源自检索提供的上下文,而非模型自行“虚构”(幻觉)。 解读:得分很高,是生成质量中最关键的指标。它表明尽管生成答案的表达与参考有出入,但其83.3%的内容有据可依,幻觉控制在较低水平。结合完美的检索结果,这说明生成内容基本是对高质量检索片段的加工 """ # 忠实度(简易版:检测关键实体是否一致) random.seed(42) # 通过jieba分词进行拆分 key_entities = set([word for word in jieba.lcut(reference) if word not in stopwords]) precision_entities = set([word for word in jieba.lcut(precision) if word not in stopwords]) faithful = sum(1 for e in key_entities if e in precision_entities) / max(1, len(key_entities)) return { 'ROUGE-1': rouge['rouge1'].fmeasure, 'ROUGE-L': rouge['rougeL'].fmeasure, 'BERTScore': bert_f1.mean().item(), # 分数求平均值 'Faithfulness': faithful }(5).主评测函数def preprocess_text(text): """ 统一的文本预处理函数 参数: text (str): 待处理的原始文本 返回值: str: 预处理后的文本 """ # 转换为小写 text = text.lower() text = re.sub(r'\n', '', text) # 移除标点符号 text = re.sub(r'[- *]+', ',', text).replace(",,|:,", ',') # 移除多余空格 text = re.sub(r'\s+', ' ', text) return text # 5 主评测函数 """ 遍历数据集,记录耗时,计算各项指标,最后求平均值并输出 JSON 格式的报告 清洗模型输出:使用正则 re.sub(r"", "", ...) 剥离了思考过程,只保留最终答案,防止思考过程干扰 ROUGE 和 BERTScore 的计算。 工程化指标:除了准确率,还计算了 Latency(延迟)和 QPS(吞吐量),这对于评估系统是否具备上线条件至关重要 """ results = [] latencies = [] # 延时list, 保存每条i数据, 后续可以求平均值 # reference_docs: 参考的文本 eval_data['reference_docs'] = eval_data['reference_docs'].apply(ast.literal_eval) # for i in tqdm(range(len(eval_data[:2]))): # 前2条评估 for i in tqdm(range(len(eval_data))): # 执行Agent、清洗输出、计算检索和生成指标 item = eval_data.iloc[i] # 获取索引 start_time = time.time() query = item['query'] reference = ''.join(item['reference_docs']) reference_docs= item['reference_docs'] response = launch_agent(query, agent, 1001) if 'think' in response and '/think' in response: answer = re.sub(r"think.*?/think", "", response, flags=re.DOTALL) elif 'think' in response and '/think\n\n答:' in response: answer = re.sub(r"think.*?/think\n\n答:", "", response, flags=re.DOTALL) else: answer = response # 执行RAG流程, 获取预测RAG结果 retrieved_docs = rag_qa(item['query']) # 记录延迟 latency = time.time() - start_time latencies.append(latency) # 延迟 # 评估各模块(计算检索模块评估指标) retrieval_metrics = evaluate_retrieval(reference_docs, retrieved_docs) # 文本预处理过程 reference_processed = preprocess_text(reference) # 真实数据 precision_processed = preprocess_text(answer) # 预测数据 # 评估各模块(计算文本生成质量评估指标) generation_metrics = evaluate_generation(precision_processed, reference_processed, retrieved_docs) results.append({ **retrieval_metrics, **generation_metrics}) # 5. 汇总结果 aggregates = { "Retrieval": { "Precision@5": np.mean([r['Precision@5'] for r in results]), "Recall@5": np.mean([r['Recall@5'] for r in res