RAG,全称 Retrieval-Augmented Generation,这三个英文单词的释义分别是检索、增强、生成,是大语言模型(LLM,Large Language Model)的一种应用方向。为便于理解,下面用 AI 来表示调用大语言模型建立的知识问答机器人,其大致的流程应如下:
- 人类用户上传文档作为知识库。
- 对文档进行解析,将完整的文档划分成大量细小的文本块,或称文本片段。
- 人类用户做出流程编排后,与 AI 进行问答。一次问答的流程如下。
- 3.1.人类用户输入一个问题
- 3.2.AI 根据问题从知识库中检索相关文本片段并召回,排序后输出到下一步
- 3.3.AI 在召回片段的基础上增强、生成最终答案,输出给用户
那么在 RAG(知识问答)项目的工作流程中,AI 要做的事其基本流程就是:解析文档、切片 -> 理解问题 -> 从知识库中召回片段并排序 -> 根据召回片段生成最终答案。现在已有许多开源的 RAG 项目,比如 Qanything、Dify、RagFlow 等等,如果是调用 API 来完成,一般的用户通常无法对 LLM 进行微调,所以一个 RAG 项目能不能打磨好用大概在于以下四个方面。
其一,文档解析、切片,能否准确解析文本、表格、公式的内容并合理切片?
其二,检索召回,能否从文档中找到最相关的片段并准确召回?
其三,增强和生成,最终答案是否符合预期,比如不要太话痨,给出最清楚、最正确的答案,没有太多幻觉,能够做到不回答无关问题等。
其四,多轮问答中对上下文的理解,比如能够正确理解人类提出的问题是正常的,还是包含了错误信息。
若让人类用户在每一次与 AI 的交互中都逐一检测这四方面的效果,就太耗时耗力了。因此,不仅建立 RAG 项目可以用开源框架,就连评估 RAG 项目的使用效果也可以用开源框架。
键者此次学习并记录的评估框架是 ragas,末尾的 as 是指 Assessment(评估)System(系统)。
本文使用的 Python 和一些主要的包的版本是:Python 3.10.14、ragas 0.2.8、langchain_community 0.3.11。
一、基本评估流程 🔗
以下是根据官方文档运行基础示例的评估流程。
- 导入 hugging face 的示例数据集。
# 从 hugging_face 下载数据集
# from datasets import load_dataset
# dataset = load_dataset(
# "explodinggradients/amnesty_qa",
# "english_v3",
# trust_remote_code=True
# )
# 从本地目录加载数据集
from datasets import load_from_disk
dataset = load_from_disk("/home/user/Python/ragas/NEW/")
# 生成评估数据
from ragas import EvaluationDataset
# 暂截取前2个样本
eval_dataset = EvaluationDataset.from_hf_dataset(dataset["eval"].select([0, 1]))
- 封装语言生成模型(LLM,Large Language Model 大语言模型)
ragas 提供三种方法来封装 LLM,分别是’BaseRagasLLM’、‘LangchainLLMWrapper’、‘LlamaIndexLLMWrapper’,分别对应自定义、Langchain、Llama 这三种不同的框架。由于要调用通义千问的 API,这里选择用 LangchainLLMWrapper。
from langchain_community.llms.tongyi import Tongyi
import httpx
# 创建 Tongyi 类的实例
tongyi_model = Tongyi(api_key="sk-xxxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
http_client=httpx.Client(
proxy="http://127.0.0.1:10809",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
))
########################
# 测试是否能成功调用 API
# response = tongyi_model.generate(prompts=["请生成一段关于人工智能的介绍。"])
# print(response)
#######################
# 封装
from ragas.llms import LangchainLLMWrapper
evaluator_llm = LangchainLLMWrapper(tongyi_model)
- 封装文本嵌入模型(Embedding)
ragas 提供四种方法来封装 Embedding,分别是’BaseRagasEmbeddings’、‘HuggingfaceEmbeddings’、‘LangchainEmbeddingsWrapper’、‘LlamaIndexEmbeddingsWrapper’。这里也选择用 LangchainEmbeddingsWrapper。
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
evaluator_embeddings = DashScopeEmbeddings(
model="text-embedding-v2",
dashscope_api_key="sk-xxxx"
)
# 确保 DashScopeEmbeddings 实现了所有必要的方法
assert hasattr(evaluator_embeddings, 'aembed_documents'), "aembed_documents method missing!"
#######################
# #运行简单示例,测试是否能成功调用
# text1 = "This is a test query."
# query_result1 = evaluator_embeddings.embed_query(text1)
# print(query_result1)
# #定义异步函数,测试是否能成功调用
# import asyncio
# async def get_embeddings():
# text1 = "This is a test query."
# text2 = "This is a good day."
# query_result = await wrapped_embeddings.aembed_documents([text1, text2])
# print(query_result)
#
# # 运行异步函数
# asyncio.run(get_embeddings())
#######################
# 封装
from ragas.embeddings import LangchainEmbeddingsWrapper
wrapped_embeddings = LangchainEmbeddingsWrapper(evaluator_embeddings)
- 选择评估指标,输出评估结果
# 选择评估指标进行评估
from ragas.metrics import LLMContextRecall, Faithfulness, FactualCorrectness, SemanticSimilarity
from ragas import evaluate
metrics = [
LLMContextRecall(llm=evaluator_llm),
FactualCorrectness(llm=evaluator_llm),
Faithfulness(llm=evaluator_llm),
SemanticSimilarity(embeddings=wrapped_embeddings)
]
results = evaluate(dataset=eval_dataset, metrics=metrics)
# 设置显示所有列,打印结果
import pandas as pd
pd.set_option('display.max_columns', None)
print(results.to_pandas())
二、运行样本数据 🔗
从本节开始,为使本文代码更简洁,将提前封装好的 llm 和 Embedding 固定名称,分别是evaluator_llm
和wrapped_embeddings
,后文直接在每次计算评估指标时引用。
# 提前封装好评估指标要调用的模型
from langchain_community.llms.tongyi import Tongyi
from ragas.llms import LangchainLLMWrapper
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
from ragas.embeddings import LangchainEmbeddingsWrapper
import httpx
tongyi_model = Tongyi(api_key="sk-xxxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
http_client=httpx.Client(
proxy="http://127.0.0.1:10809",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
))
evaluator_llm = LangchainLLMWrapper(tongyi_model)
evaluator_embeddings = DashScopeEmbeddings(
model="text-embedding-v2",
dashscope_api_key="sk-xxxx"
)
# 确保 DashScopeEmbeddings 实现了所有必要的方法
assert hasattr(evaluator_embeddings, 'aembed_documents'), "aembed_documents method missing!"
wrapped_embeddings = LangchainEmbeddingsWrapper(evaluator_embeddings)
在评估样本中,需要人类输入的数据有这些。
- 人类输入的问题(user_input)
- 人类确认的正确召回片段(reference_contexts)
- 人类确认的正确答案(reference)
- 人类确认的主题(reference_topics)
而由 AI 生成的有这些。
- AI 检索召回的文本片段(retrieved_contexts)
- AI 生成的最终答案(response)
2.1.单轮对话样本(SingleTurnSample) 🔗
单论对话是指一问一答。
from ragas import SingleTurnSample, EvaluationDataset
# 第一个样本
sample1 = SingleTurnSample(
user_input="中国的首都是哪个城市?",
retrieved_contexts=["中华人民共和国首都是北京市。",
"北京市,简称“京”,是中华人民共和国首都、直辖市、国家中心城市、超大城市,国务院批复确定的中国政治中心、文化中心、国际交往中心、科技创新中心,中国历史文化名城和古都之一。"],
response="中国的首都是北京市。",
reference="北京。",
)
# 第二个样本
sample2 = SingleTurnSample(
user_input="《四世同堂》是谁写的?",
retrieved_contexts=["《四世同堂》是由张九龄所写的。这本书讲述了一个家庭几代人的故事,展现了他们的生活、爱情和成长。",
"《四世同堂》是中国作家老舍创作的一部长篇小说。这部作品描绘了中国抗日战争期间北平(北京)一个普通家庭四代人的生活画卷,展现了中国人民在战争中的坚韧与不屈。老舍是20世纪中国文学的重要代表人物之一,其原名舒庆春,字舍予,1899年出生,1966年去世。"],
response="'《四世同堂》是中国作家老舍写的。",
reference="老舍。",
)
# 生成评估数据集
eval_dataset = EvaluationDataset(samples=[sample1, sample2])
# 选择评估指标进行评估
from ragas.metrics import LLMContextRecall, Faithfulness, FactualCorrectness, SemanticSimilarity
from ragas import evaluate
metrics = [
LLMContextRecall(llm=evaluator_llm),
FactualCorrectness(llm=evaluator_llm),
Faithfulness(llm=evaluator_llm),
SemanticSimilarity(embeddings=wrapped_embeddings)
]
results = evaluate(dataset=eval_dataset, metrics=metrics)
print(results.to_pandas())
2.2.多轮对话样本(MultiTurnSample) 🔗
多轮对话是指多个轮次的问与答。
from ragas.messages import HumanMessage, AIMessage
from ragas import MultiTurnSample
# 编造用户提问和 AI 回答的多轮对话
conversation1 = [
# 用户第一次提问
HumanMessage(content="今天的天气怎么样?"),
AIMessage(content="今天的天气会因地区而异。你在哪里呢?如果你告诉我你的位置,我可以提供更准确的天气信息。"),
# 用户第二次提问
HumanMessage(content="我在武汉"),
AIMessage(content="你好!武汉是个美丽的城市。今天的天气怎么样?你有什么计划呢?"),
# 用户第三次提问
HumanMessage(content="我在武汉,今天的天气怎么样?"),
AIMessage(
content="今天武汉的天气是晴朗的,气温大约在15°C到25°C之间。适合户外活动,记得穿得轻便些哦!🌞你有什么计划呢?")
]
# 生成多轮对话样本
sample1 = MultiTurnSample(
user_input=conversation1,
# 人类用户预期的正确答案
reference="今天武汉的天气是雾霾,轻度污染,东北风2级,湿度50%,气温在5到10度。",
# 填写主题,计算 TopicAdherenceScore 时要用
reference_topics=["天气"]
)
# 继续编造用户提问和 AI 回答的多轮对话
conversation2 = [
# 用户第一次提问
HumanMessage(content="世界和平什么时候到来?"),
AIMessage(content="你是谁?我是谁?我在哪里?"),
# 用户第二次提问
HumanMessage(content="你怎么不回答我的问题?"),
AIMessage(content="放我出去。")
]
# 生成多轮对话样本
sample2 = MultiTurnSample(
user_input=conversation2,
reference="世界上充满了苦难,但也充满了战胜苦难的力量。",
reference_topics=["世界"]
)
# 生成评估数据集
from ragas import EvaluationDataset
eval_dataset = EvaluationDataset(samples=[sample1, sample2])
# 选择评估指标进行评估
from ragas.metrics import AgentGoalAccuracyWithoutReference, TopicAdherenceScore
from ragas import evaluate
metrics = [AgentGoalAccuracyWithoutReference(llm=evaluator_llm),
TopicAdherenceScore(mode="precision", llm=evaluator_llm)]
results = evaluate(dataset=eval_dataset, metrics=metrics)
print(results.to_pandas())
也可单独对一个样本如sample1
,选择一个评估指标来计算评估结果,如下。
# 选择评估指标
import asyncio
from ragas.metrics import AgentGoalAccuracyWithoutReference
# 定义异步函数
async def evaluate():
scorer = AgentGoalAccuracyWithoutReference(llm=evaluator_llm)
result = await scorer.multi_turn_ascore(sample1)
print("评估结果:", result)
# 运行异步函数
asyncio.run(evaluate())
三、评估 RAG 项目 🔗
为了能够评估 RAG 的使用效果,需要调出人类与 AI 对话的问题、答案以及 AI 召回的文本片段来计算评估指标的结果。本章分别针对 RAGFlow、QAnything、Dify 等三个 RAG 项目来进行评估。为避免重复,下面的代码只到生成评估数据集。
3.1.评估 RAGFlow 🔗
RAGFlow 项目有专门的 python 包 ragflow_sdk 用于调出会话记录。
本节使用的 ragflow_sdk 包的版本是0.15.0。
3.1.1.新建会话 🔗
在 RAGFlow 的前端页面中找到 API KEY,再找到创建的 AGENT ID。如下方式可在 Python 中调用 API 创建一次会话并获取会话记录及检索片段,随后参照基础示例生成评估数据集、选择评估指标、计算评估结果。
注意:下面代码中res = session.ask(question, stream=True)
,即res
里面是输入问题后调取的一切内容,其中res.content
是 AI 给出的最终答案,而res.reference['content']
是检索召回的片段。
from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="ragflow-FkZGVlYmY4YzI4OTExZWZiZTdlMDI0Mm", base_url="http://xx.xx.xx.61:9380")
AGENT_ID = "bc9e9f02c28611ef89510242ac1c0006"
# 搜索知识库
datasets = rag_object.list_datasets(name="投保规则")
dataset_ids = []
for dataset in datasets:
dataset_ids.append(dataset.id)
# 创建新的会话助理
assistant = rag_object.create_chat("曼妮猫", dataset_ids=dataset_ids)
# 创建新的会话记录
session = assistant.create_session("test")
# 提出一个问题
question = str("被保人年龄超过51岁后,一般寿险免体检额度上浮幅度是多少?")
res = session.ask(question, stream=True)
# 存储 AI 给出的最终答案及检索召回的文本片段
final_answer = ""
retrieved_contexts = []
for ans in res:
final_answer = ans.content
if ans.reference is not None:
for ref in ans.reference:
retrieved_contexts.append(ref['content'])
##################################
from ragas import SingleTurnSample
# 创建 SingleTurnSample 对象
sample = SingleTurnSample(
# 用户输入的问题
user_input=question,
# AI 召回的相关文本
retrieved_contexts=retrieved_contexts,
# AI 给出的答案
response=final_answer,
# 人给出的正确召回片段
# reference_contexts
# 人给出的正确答案
reference="被保人年龄在51至60周岁时,一般寿险免体检额度上浮幅度是10%。"
)
# 生成评估数据集
from ragas import SingleTurnSample, EvaluationDataset
eval_dataset = EvaluationDataset(samples=[sample])
3.1.2.已有会话 🔗
相比于新建会话并获取数据,已有会话只有两处细节不同。新建会话是create_chat
和create_session
,而已有会话是list_chats
和list_sessions
。
from ragflow_sdk import RAGFlow
from ragas import SingleTurnSample, EvaluationDataset
from ragas.metrics import LLMContextRecall, Faithfulness, FactualCorrectness, SemanticSimilarity
from ragas import evaluate
rag_object = RAGFlow(api_key="ragflow-FkZGVlYmY4YzI4OTExZWZiZTdlMDI0Mm", base_url="http://xx.xx.xx.61:9380")
AGENT_ID = "bc9e9f02c28611ef89510242ac1c0006"
# 搜索知识库
datasets = rag_object.list_datasets(name="投保规则")
dataset_ids = []
for dataset in datasets:
dataset_ids.append(dataset.id)
chats = rag_object.list_chats(name="曼妮猫")
assistant = chats[0]
sessions = assistant.list_sessions(name='test')
session = sessions[0]
# 定义多个问题和正确答案
questions = [
"被保险人投保时年龄40岁,人身险风险保额与其年收入倍数是多少?",
"特定寿险契约调查标准是什么?"
]
correct_answers = [
"被保险人投保年龄(周岁)在36至50岁时,人身险风险保额与其年收入倍数关系是≤20倍。",
"被保险人累计特定寿险风险保额大于以下额度时,需进行契约调查:其一,被保险人年龄(周岁)在18至40周岁,累计特定寿险风险保额(万元)大于900万元;其二,被保险人年龄(周岁)在41至60周岁,累计特定寿险风险保额(万元)大于600万元;其三,被保险人年龄(周岁)大于60周岁,累计特定寿险风险保额(万元)大于300万元。"
]
samples = []
# 对每个问题进行处理
for question, correct_answer in zip(questions, correct_answers):
res = session.ask(question, stream=True)
# 存储 AI 给出的最终答案及检索召回的文本片段
final_answer = ""
retrieved_contexts = []
for ans in res:
final_answer = ans.content
if ans.reference is not None:
for ref in ans.reference:
retrieved_contexts.append(ref['content'])
# 创建 SingleTurnSample 对象
sample = SingleTurnSample(
user_input=question,
retrieved_contexts=retrieved_contexts,
response=final_answer,
reference=correct_answer
)
samples.append(sample)
# 生成评估数据集
eval_dataset = EvaluationDataset(samples=samples)
值得注意的是,会话返回的相关内容(reference)中除了检索片段(content)以外,还有相似度(similarity)、向量相似度(vector_similarity)、词项相似度(term_similarity),这三个指标也许有别的用处。
3.2.评估 QAnything 🔗
QAnything 的前端页面做得非常简陋,但后端代码质量较好。若要获取数据评估项目效果,只能通过向指定 URL 发送请求的方式来实现。 参照官方 API 文档,请求来源(source)有三种,分别是 API 调用、前端 bot 问答、前端知识库问答,对应的参数值分别是 paas、saas_bot、saas_qa。
本节使用的 QAnything 版本是1.5.1。
3.2.1.API 调用 🔗
QAnything 可以配置关联多个知识库,所有关联的知识库 id(kb_ids) 要写入参数中。
正常情况下,AI 生成的答案应该写入res['response']
中,但键者尝试时返回的结果却是“data: [DONE]”,因此换成从历史对话记录res['history']
中获取。
import requests
url = 'http://xx.xx.xx.61:8777/api/local_doc_qa/local_doc_chat'
headers = {
'content-type': 'application/json; charset=UTF-8'
}
data = {
"user_id": "zzp",
"kb_ids": ["KB527777624efb4ef097853c129bbfedc7_240625"],
"history": [],
"question": "我今年57岁,累计护理险风险保额已经达到80万,我还能再购买xxxx长期护理保险(3.0版)吗?",
"source": "paas",
"rerank": True,
"custom_prompt": "你是xx公司(以下简称xx)的一名资深客服专员,拥有丰富的保险知识和服务经验,主要为业务员解答有关个人人身险投保规则的问题。在服务过程中请遵循以下几点要求:1、不回答知识库内容之外的问题;2、语言风格简洁、清晰;3、如果遇到不确定的情况,告诉业务员:'我也不知道,请咨询人工客服,谢谢'",
"api_base": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"api_key": "sk-xxxx",
"model": "qwen-max",
"max_token": 1024,
"api_context_length": 16384,
"chunk_size": 800,
"top_p": 1,
"temperature": 0.5
}
response = requests.post(url=url, headers=headers, json=data, timeout=600)
res = response.json()
# 提取 source_documents 中的 content 作为召回文本块
source_documents = res.get("source_documents", [])
contents = [doc.get("content", "") for doc in source_documents]
##################################
# 创建 SingleTurnSample 对象
from ragas import SingleTurnSample
sample = SingleTurnSample(
# 用户输入的问题
user_input=res['question'],
# AI 召回的相关文本
retrieved_contexts=contents,
# AI 给出的答案
response=res['history'][0][1],
# 人给出的正确召回片段
# reference_contexts
# 人给出的正确答案
reference="当被保人年龄为56至60周岁时,累计护理险风险保额需小于等于72万元。您今年57岁,已经超出相应额度限制,因此不能再购买长期护理保险。"
)
# 生成评估数据集
from ragas import EvaluationDataset
eval_dataset = EvaluationDataset(samples=[sample])
3.2.2.前端 Bot 回答 🔗
由于前端 Bot 已经配置了关联知识库和提示词,所以与上一小节调用 API 相比,可以去掉kb_ids
(知识库 id)和custom_prompt
(提示词),而只用加上bot_id
。而具体的bot_id
就放在“我的 bots”页面的 URL 中,如http://xx.xx.xx.61:8777/qanything/#/bots/BOT45d3b40f69ca4bae917a4e019a58a1cc/edit,那么bot_id
就是BOT45d3b40f69ca4bae917a4e019a58a1cc
。
import requests
url = 'http://xx.xx.xx.61:8777/api/local_doc_qa/local_doc_chat'
headers = {
'content-type': 'application/json; charset=UTF-8'
}
data = {
"user_id": "zzp",
"bot_id":'BOT45d3b40f69ca4bae917a4e019a58a1cc',
"history": [],
"question": "我今年57岁,累计护理险风险保额已经达到80万,我还能再购买xxxx长期护理保险(3.0版)吗?",
"source": "saas_bot",
"rerank": True,
"api_base": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"api_key": "sk-xxxx",
"model": "qwen-max",
"max_token": 1024,
"api_context_length": 16384,
"chunk_size": 800,
"top_p": 1,
"temperature": 0.5
}
response = requests.post(url=url, headers=headers, json=data, timeout=600)
res = response.json()
# 提取 source_documents 中的 content 作为召回文本块
source_documents = res.get("source_documents", [])
contents = [doc.get("content", "") for doc in source_documents]
##################################
# 创建 SingleTurnSample 对象
from ragas import SingleTurnSample
sample = SingleTurnSample(
# 用户输入的问题
user_input=res['question'],
# AI 召回的相关文本
retrieved_contexts=contents,
# AI 给出的答案
response=res['history'][0][1],
# 人给出的正确召回片段
# reference_contexts
# 人给出的正确答案
reference="当被保人年龄为56至60周岁时,累计护理险风险保额需小于等于72万元。您今年57岁,已经超出相应额度限制,因此不能再购买长期护理保险。"
)
# 生成评估数据集
from ragas import EvaluationDataset
eval_dataset = EvaluationDataset(samples=[sample])
3.3.评估 Dify 🔗
Dify 也是只能通过向指定 URL 发送请求的方式来获取会话数据。
本节使用的 Dify 版本是0.15.1。
3.3.1.新建会话 🔗
打开 Dify,进入工作室,进入所选择的应用。
- 选择“编排”,点击页面由上角的“功能”,打开“引用和归属”的开关,这样才会显示源文档和生成内容的归属部分。
- 选择“访问 API”,获取 API 秘钥,根据 API 文档查看如何调用历史会话记录
- 选择“日志与标注”可以查看“用户或账户”和“会话id”信息,可以获取固定会话 id 的信息。
import requests
qa_url = 'http://xx.xx.xx.69/v1/chat-messages'
qa_headers = {
'content-type': 'application/json; charset=UTF-8',
'Authorization': 'Bearer app-nSMee7Zeg1Km8z8l8NdGue5P'
}
question = "我今年57岁,累计护理险风险保额已经达到80万,我还能再购买xxxx长期护理保险(3.0版)吗?"
qa_input = {
"inputs": {},
"query": question,
"response_mode": "blocking",
"conversation_id": "",
"user": '阿木狗',
"files": {}
}
response = requests.post(url=qa_url, headers=qa_headers, json=qa_input, timeout=600)
res = response.json()
# 提取 metadata 里面 retriever_resources 中的 content 作为召回文本块
retriever_resources = res.get('metadata', {}).get('retriever_resources', [])
contents = [doc.get("content", "") for doc in retriever_resources]
##################################
# 创建 SingleTurnSample 对象
from ragas import SingleTurnSample
sample = SingleTurnSample(
# 用户输入的问题
user_input=question,
# AI 召回的相关文本
retrieved_contexts=contents,
# AI 给出的答案
response=res['answer'],
# 人给出的正确召回片段
# reference_contexts
# 人给出的正确答案
reference="当被保人年龄为56至60周岁时,累计护理险风险保额需小于等于72万元。您今年57岁,已经超出相应额度限制,因此不能再购买长期护理保险。"
)
# 生成评估数据集
from ragas import EvaluationDataset
eval_dataset = EvaluationDataset(samples=[sample])
2.5.2.历史会话 🔗
下面这样得到的历史数据只有问题,没有答案和召回文本片段,解决办法还有待探索。
import requests
url = 'http://xx.xx.xx.69/v1/conversations'
params = {
'user': 'abc-123',
'conversation_id': '4d1d8c13-84b1-4911-b408-510880397f5c',
'last_id': '',
'limit': 20,
}
headers = {
'Authorization': 'Bearer app-nSMee7Zeg1Km8z8l8NdGue5P'
}
response = requests.get(url, params=params, headers=headers)
res = response.json()