基于Prefect Weave和RAGAS的RAG运行开发实战

几个月前，我颁布了一款电影搜查运行程序“Film Search”，这是一个检索增强生成（RAG）运行程序，旨在依据用户查问成功电影介绍。例如，用户或者会收回上方的英文提问：

中文意思：“给我找一部长度不到2小时、以狗为主角的英文剧情片。”

之后，用户会收到相似上方这样的英文介绍：

Title of Film: Hachi: A Dog’s TaleRuntime: 93 minutesRelease Year: 2009Streaming: Not available for streamingThis film tells the poignant true story of Hachiko, an Akita dog known for his remarkable loyalty to his owner. The emotional depth and the themes of friendship and loyalty resonate strongly, making it a touching drama that showcases the profound bond between humans and dogs. It’s perfect for anyone looking for a heartfelt story that highlights the importance of companionship.

对应的中文回答意思是：

这个软件不只仅是一个繁难的RAG运行程序。该程序中经常使用了所谓的自查问检索。这象征着，机器人接受用户的查问，并经过增加元数据过滤器对其启动转换。这样就确保了拉入聊天模型高低文的任何文档都遵守用户查问设置的解放。无关更多信息，我倡导检查我文后提供的我早些时刻宣布的文章的链接。

但遗憾的是，该运行程序尚存在如下一些疑问：

在本文中，我将简明引见对以前开发的那款电影搜查运行程序“Film Search”所做的一些改良，详细的改良内容将包括：

在咱们正式开局之前，还有一个细节须要说明：我发现“Film Search”这个名字有点抽象，所以我把这个运行程序从新命名为“Rosebud”，如下图所示。不用多作解释，我想任何一位真正的电影迷都会明确这个意思的（【译者注】影片Citizen Kane（公民凯恩）的故事是由报业巨子凯恩临死前说的一个字“玫瑰花蕾”（Rosebud）引出的）。

程序名字“Film Search”更改为“Rosebud”，此图来自Unsplash网站

线下评价

能够判别对LLM运行程序所做的更改是提高还是降落了程序性能，这一点是十分关键的。可怜的是，LLM运行程序的评价是一个艰巨而陈腐的畛域。关于什么是好的评价，基本没有达成太多的共识。

在新的程序Rosebud中，我选择处置所谓的“RAG Triad（RAG三元组）：”疑问。这种方法由TruLens推出，TruLens是一个评价和跟踪LLM运行程序的平台。

概括来看，三元组涵盖了RAG运行程序的三个方面：

目前，曾经存在几种方法可以尝试评价RAG运行程序的这三特性能。一种方法是借助人类专家评价员。可怜的是，这种方法十分低廉，而且无法裁减。在新的程序Rosebud中，我选择经常使用大型数据模型启动评价。这象征着，经常使用聊天模型来检查上述三个规范中的每一个，并为每个规范调配0到1的分数值。这种方法具备老本低、可裁减性好的好处。为了成功这一点，我经常使用了RAGAS（），这是一个盛行的框架，可以协助你评价RAG运行程序。RAGAS框架包括上述三个目的，可以很容易地经常使用它们来评价你的运行程序。上方是一个代码片段，演示了我是如何经常使用开源的RAGAS框架启动离线评价的：

from ragas import evaluatefrom ragas.metrics import AnswerRelevancy, ContextRelevancy, Faithfulnessimport weave@weave.op()def evaluate_with_ragas(query, model_output):#将数据放入一个数据集对象中data = {"question": [query],"contexts": [[model_output['context']]],"answer": [model_output['answer']]}dataset =>

在上述代码中，有几点留意事项：

从通常上讲，如今咱们可以调整一个超参数（如温度），从新运转评价，看看调整能否有踊跃或消极的影响了。但遗憾的是，在通常中，我发现大型言语评判者的评判很挑剔，而且我也不是惟逐一个发现这一点的人（

大型言语模型评价仿佛不太长于经常使用浮点数来评价这些目的。相反，它们仿佛在分类方面做得更好些，例如回答“赞同/不赞同”这样的疑问。，RAGAS尚不支持经常使用LLM评判者启动分类。间接手写无关代码仿佛也并不难，兴许在未来的更新中，我或者会自己尝试一下。

在线评价

离线评价有助于了解调整超参数如何影响性能，在我看来，在线评价要有用得多。在新的程序Rosebud中，我如今曾经经常使用“赞同/不赞同”的打算——经常使用每个照应底部的两个相应按钮来提供反应。

当用户点击上图中底部任一按钮时，就会原告知他们的反应已被记载。以下给出在Streamlit运行程序界面中如何成功这一点的代码片段：

def start_log_feedback(feedback):print("Logging feedback.")st.session_state.feedback_given = Truest.session_state.sentiment = feedbackthread = threading.Thread(target=log_feedback, args=(st.session_state.sentiment,st.session_state.query,st.session_state.query_constructor,st.session_state.context,st.session_state.response))thread.start()def log_feedback(sentiment, query, query_constructor, context, response):ct = datetime.datetime.now()wandb.init(project="film-search",name=f"query: {ct}")table = wandb.Table(columns=["sentiment", "query", "query_constructor", "context", "response"])table.add_data(sentiment,query,query_constructor,context,response)wandb.log({"Query Log": table})wandb.finish()

请留意，向W&B发送反应的环节是在独自的线程上运转的，而不是在主线程上运转。这是为了防止用户在期待日志记载成功时被卡住几秒钟。

咱们经常使用了一个W&B表格用于存储反应。表中记载了五个数值：

{"query": "drama English dogs","filter": {"operator": "and","arguments": [{"comparator": "eq", "attribute": "Genre", "value": "Drama"},{"comparator": "eq", "attribute": "Language", "value": "English"},{"comparator": "lt", "attribute": "Runtime (minutes)", "value": 120}]},}

一切这些都可以繁难地记载在与前面显示的Weave评价相反的名目中。如今，当查问“不赞同”状况时，只要按下拇指向下的图标按钮即可检追究竟出现了什么。这将有助于使介绍运行程序Rosebud的迭代和改良加极速度。

模型照应可观测性展现（请留意左侧的W&B和Weave之间的无缝过渡）

借助Prefect智能提取数据

为了使介绍程序Rosebud坚持准确性，将数据提取和上行到Pinecone向量数据库的环节智能化十分关键。关于这个义务，我选用经常使用Prefect（）。Prefect是一个盛行的上班流编排工具。我不时在寻觅一些轻量级、易于学习和Python格调的程序。最后，我在Prefect中找到了这一切。

Prefect提供的用于提取和更新Pinecone向量存储的智能流程

Prefect支持提供多种方式来布局你的上班流程。我选择经常使用带有智能基础设备性能的推送上班池方式。我发现这种设置在繁难性和可性能性之间取得了平衡。它准许用户委托Prefect智能性能在所选云提供商中运转流所需的一切基础设备。经过几番掂量后，我选用在Azure上部署，但是在GCP或AWS上部署的话只要要更改几行代码即可。无关更多详细信息，请参阅pinecone_flow.py文件。上方代码只是提供了一个简化的流程：

@taskdef start():"""启动：审核一切上班或失败的速度快！"""#打印出一些调试信息print("Starting flow!")# 确保用户曾经设置了适当的环境变量assert os.environ['LANGCHAIN_API_KEY']assert os.environ['OPENAI_API_KEY']...@task(retries=3, retry_delay_seconds=[1, 10, 100])def pull_data_to_csv(config):TMBD_API_KEY = os.getenv('TMBD_API_KEY')YEARS = range(config["years"][0], config["years"][-1] + 1)CSV_HEADER = ['Title', 'Runtime (minutes)', 'Language', 'Overview', ...]for year in YEARS:# 失掉一切在{Year}中制造的电影的id列表movie_list = list(set(get_id_list(TMBD_API_KEY, year)))FILE_NAME = f'./data/{year}_movie_collection_data.csv'#生成文件with open(FILE_NAME, 'w') as f:writer = csv.writer(f)writer.writerow(CSV_HEADER)...print("Successfully pulled,description="The title of the movie", type="string"),AttributeInfo(name="Runtime (minutes)",description="The runtime of the movie in minutes", type="integer"),...]def convert_to_list(doc, field):if field in doc.metadata and doc.metadata[field] is not None:doc.metadata[field] = [item.strip()for item in doc.metadata[field].split(',')]...fields_to_convert_list = ['Genre', 'Actors', 'Directors','Production Companies', 'Stream', 'Buy', 'Rent']...# 将'overview' 和'keywords' 设置为'page_content'，其余字段设置为'metadata'for doc in docs:#将page_counte字符串解析为字典page_content_dict = dict(line.split(": ", 1)for line in doc.page_content.split("\n") if ": " in line)doc.page_content = ('Title: ' + page_content_dict.get('Title') +'. Overview: ' + page_content_dict.get('Overview') +...)...print("Successfully took csv files and created docs")return docs@taskdef upload_docs_to_pinecone(docs, config):# 创立空索引PINECONE_KEY, PINECONE_INDEX_NAME = os.getenv('PINECONE_API_KEY'), os.getenv('PINECONE_INDEX_NAME')pc = Pinecone(api_key=PINECONE_KEY)# 目的索引和审核形态pc_index = pc.Index(PINECONE_INDEX_NAME)print(pc_index.describe_index_stats())embeddings = OpenAIEmbeddings(model=config['EMBEDDING_MODEL_NAME'])namespace = "film_search_prod"PineconeVectorStore.from_documents(docs,...)print("Successfully uploaded docs to Pinecone vector store")@taskdef publish_dataset_to_weave(docs):#初始化Weaveweave.init('film-search')rows = []for doc in docs:row = {'Title': doc.metadata.get('Title'),'Runtime (minutes)': doc.metadata.get('Runtime (minutes)'),...}rows.append(row)dataset =, rows=rows)weave.publish(dataset)print("Successfully published,work_pool_name="my-aci-pool",cron="0 0 * * 0",image=DeploymentImage(name="prefect-flows:latest",platform="linux/amd64",))

请留意，将Python函数转换为Prefect流是十分繁难的事件。你只要要在主函数上经常使用@task装璜器和@flow装璜器来设计一些子函数。还要留意，在将文档上行到Pinecone向量数据库后，咱们流程的最后一步是将数据集颁布到Weave。这关于再现性很关键。为了学习Prefect的基础常识，我倡导你阅读一下他们官方上的教程（）。

在上方脚本的最后，咱们看到部署是如何在Prefect中成功的。

要经常使用命令行方式在Azure上部署此流，请运转以下命令：

prefect work-pool create --type azure-container-instance:push --provision-infra my-aci-poolprefect deployment run 'get_repo_info/my-deployment'

这些命令将智能在Azure上提供一切必要的基础设备。这包括一个Azure容器注册表（ACR），它将保留一个Docker映像，其中蕴含目录中的一切文件以及requirements.txt中列出的任何必要的依赖库。它还将包括一个Azure容器实例（ACI）标识，该标识将具备部署具备上述Docker映像的容器所需的权限。最后，经常使用deployment run命令布置每周运转的代码。你可以经过Prefect控制面板来检查你的流能否运转：

Prefect中的流正在成功运转的情景

经过每周更新我的Pinecone向量库，我可以确保来自程序Rosebud的介绍结果准确。

总结

在本文中，我引见了我的改良后的Rosebud运行程序的一些改良打算。这包括整合离线和在线评价的环节，以及智能更新我的Pinecone向量库等。

本文还有未提及的其余一些改良，包括：

正如文章一开局提到的，该运行程序如今可以100%无偿经常使用！在可预感的未来，我将为查问买单（因此选用gpt-4o-mini而不是更低廉的gpt-4o）。我真的很想取得在消费环境中运转运行程序的阅历，并让我的读者测试Rosebud，这是一个很好的方法。万一运行程序真的“火爆”了，我将不得不想出其余的融资形式。但这会是一个很大的疑问。

上方，请纵情享用经常使用Rosebud程序搜查精彩电影的乐趣吧！

译者引见

朱先忠，社区编辑，专家博客、讲师，潍坊一所高校计算机老师，自在编程界老兵一枚。