几个月前，我发布了一款电影搜索应用程序“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程序搜索精彩电影的乐趣吧！

译者介绍

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