想了解更多AIGC的内容，请访问：
AI.x社区
最近，我在浏览Max.com网站时想找一部电影看。通常，这个过程包括浏览系统呈现给我的各种列表，阅读一些相关描述，然后挑选一些看起来有趣的电影。如果我知道我想看的电影的片名或我喜欢的演员的名字，我通常只会点击搜索功能。否则，搜索就没有多大用处了。
现在，我突然想到了一个新的想法：为什么我不能用自然语言来查找一部电影，更多地基于电影的氛围或实质，而不仅仅是标题或演员呢?例如，为什么我不能启动Max、Netflix或Hulu等流媒体播放平台，并在搜索栏中键入类似于以下查询之一呢：
这种方法的美妙之处超出了更自然的电影搜索方式，还保护了用户的隐私。该系统根本不会使用用户数据，不是挖掘用户的行为、喜欢和不喜欢来提供给推荐系统。唯一需要的就是一个查询。
为此，我开发了本文中要展示给大家的一个电影搜索程序。这是一个基于RAG(检索增强生成)的系统，它可以接受用户的查询，嵌入查询，并进行相似性搜索，以找到相似的电影。不过，这个程序超越了普通的RAG系统。这个系统使用了所谓的自查询检索器。该技术允许在进行相似性搜索之前，根据电影的元数据对其进行过滤。因此，如果用户有一个类似“推荐1980年后拍摄的以大量爆炸为特征的恐怖电影”的查询，搜索算法将首先过滤掉所有不是“1980年后制作的恐怖片”的电影，然后再对“以大量爆炸为主”的电影进行相似性搜索。
在本文中，我将提供一个关于我如何创建此系统的总体概述。如果您想深入了解这个程序，完整的源代码将在文后的链接参考处提供。
接下来，让我们继续作深入介绍。
检索数据
首先，该项目的数据来自电影数据库(TMDB：，并得到了所有者的许可。他们的API使用简单，维护良好，并且没有严格的费率限制。我从他们的API中提取了以下电影属性：
以下是如何使用TMDB API和Python的响应库提取数据的片段：
def get_data(API_key, Movie_ID, max_retries=5):"""函数以JSON格式提取感兴趣的电影的详细信息。parameters:API_key (str): Your API key for TMBDMovie_ID (str): TMDB id for film of interestreturns:dict: JSON格式的字典，包含您的电影的所有细节兴趣"""query = 'https://api.themoviedb.org/3/movie/' + Movie_ID + \'?api_key='+API_key + '&append_to_response=keywords,' + \'watch/providers,credits'for i in range(max_retries):response = requests.get(query)if response.status_code == 429:# If the response was a 429, wait and then try againprint(f"Request limit reached. Waiting and retrying ({i+1}/{max_retries})")time.sleep(2 ** i)# Exponential backoffelse:dict = response.json()return dict请注意，该查询需要电影ID(也是使用TMDB获得的)以及append_to_response，这允许我提取几种类型的数据，例如关键字、影片提供商、演员(导演和演员)以及有关电影的一些基本信息。还有一些基本的框架类代码，以防我达到速率限制，尽管我注意到从未发生这种情况。
然后，我们必须解析JSON响应。以下的代码片段展示了如何解析电影中的演员和导演：
credits = dict['credits']actor_list, director_list = [], []# 分析演员表cast = credits['cast']NUM_ACTORS = 5for member in cast[:NUM_ACTORS]:actor_list.append(member["name"])# 分析剧组crew = credits['crew']for member in crew:if member['job'] == 'Director':director_list.append(member["name"])actor_str = ', '.join(list(set(actor_list)))director_str = ', '.join(list(set(director_list)))请注意，我将演员数量限制在一部电影的前五名。我还必须说明，我只对导演感兴趣，因为系统的响应还包括其他类型的剧组成员，如编辑、服装设计师等。
所有这些数据随后被编译成CSV文件。上面列出的每个属性都被转换成了一列，现在每一行都代表一部特定的电影。以下是通过程序创建的2008_movie_collection_data.csv文件中的短片。在这个项目中，我获得了大约100部1920年至2023年的顶级电影。
用于演示目的的电影数据片段(作者本人提供)
信不信由你，我还没看过《功夫熊猫》。也许我必须完成这个项目。
将文档上载到pinecone网站
接下来，我必须将csv数据上传到网站([译者注]。Pinecone是一个非开源型的向量数据库。Pinecone支持在大规模向量集上进行快速且实时的搜索，具有亚秒级的查询响应时间，适用于需要高性能和实时性的大型应用，特别适合于构建实时推荐系统、电商搜索引擎和社交媒体内容过滤等)。通常，分块在RAG系统中很重要，但这里每个“文档”(CSV文件的行)都很短，所以分块不是一个问题。我首先必须将每个CSV文件转换为LangChain文档，然后指定哪些字段应该是主要内容，哪些字段应该作为元数据。
以下是用于构建这些文档的代码片段：
# 从所有csv文件加载数据loader = DirectoryLoader(path="./data",glob="*.csv",loader_cls=CSVLoader,show_progress=True)docs = loader.load()metadata_field_info = [AttributeInfo(name="Title", description="The title of the movie", type="string"),AttributeInfo(name="Runtime (minutes)",description="The runtime of the movie in minutes", type="integer"),AttributeInfo(name="Language",description="The language of the movie", type="string"),...]for doc in docs:#将page_content字符串解析到字典中page_content_dict = dict(line.split(": ", 1)for line in doc.page_content.split("\n") if ": " in line)doc.page_content = 'Overview: ' + page_content_dict.get('Overview') + '. Keywords: ' + page_content_dict.get('Keywords')doc.metadata = {field.name: page_content_dict.get(field.name) for field in metadata_field_info}#将字段从字符串转换为字符串列表for field in fields_to_convert_list:convert_to_list(doc, field)# 将字段从字符串转换为整数for field in fields_to_convert_int:convert_to_int(doc, field)LangChain的DirectoryLoader负责将所有csv文件加载到文档中。然后，我需要指定什么应该是page_content，什么应该是metadata;这是一个重要的决定。page_content将在检索阶段嵌入并用于相似性搜索。在进行相似性搜索之前，metadata将仅用于过滤目的。我决定采用overview和keywords属性并嵌入它们，其余的属性将是元数据。应该做进一步的调整，看看title是否也应该包括在page_content中，但我发现这种配置对大多数用户查询都很有效。
接下来，文件必须上传到pinecone网站。这是一个相当简单的过程：
# 如果尚未创建索引，则取消注释pc.create_index(name=PINECONE_INDEX_NAME,dimension=1536,metric="cosine",spec=PodSpec(environment="gcp-starter"))# 目标索引和检查状态pc_index = pc.Index(PINECONE_INDEX_NAME)print(pc_index.describe_index_stats())embeddings = OpenAIEmbeddings(model='text-embedding-ada-002')vectorstore = PineconeVectorStore(pc_index, embeddings)# 创建记录管理器namespace = f"pinecone/{PINECONE_INDEX_NAME}"record_manager = SQLRecordManager(namespace, db_url="sqlite:///record_manager_cache.sql")record_manager.create_schema()# 将文档上载到松果网站index(docs, record_manager, vectorstore,cleanup="full", source_id_key="Website")我只想在这里强调几个事情：
创建自查询检索器
自查询检索器将允许我们通过我们之前定义的元数据来过滤RAG期间检索到的电影。这将大大提高我们电影推荐人的实用性。
在选择矢量存储时，一个重要的考虑因素是确保它支持按元数据过滤，因为并非所有数据库都支持这种技术。链接处提供了LangChain支持自查询检索的数据库列表。另一个重要的考虑因素是对于每个矢量存储允许什么类型的比较器。比较器是我们通过元数据进行过滤的方法。例如，我们可以使用eq比较器来确保我们的电影属于科幻类型：eq('Genre', 'Science Fiction')。并非所有矢量存储都允许所有比较器。举个例子，有兴趣的读者可以观察一下开源的嵌入式数据库Chroma中支持的比较器(，以及它们与Pinecone网站中支持的比较器(有何不同。我们需要告诉模型允许使用哪些比较器，以防止它意外地写入禁止的查询。
除了告诉模型存在哪些比较器之外，我们还可以提供用户查询和相应过滤器的模型示例。这被称为小样本学习(Few-shot Learning)，这对指导您的模型是非常宝贵的。
要具体地了解这一技巧有何帮助，您可以尝试查看以下两个用户查询：
我的元数据过滤模型很容易为这些示例中的每一个编写相同的过滤查询，尽管我希望对它们进行不同的处理。第一部应该只推荐兰蒂莫斯执导的电影，而第二部应该推荐与兰蒂莫斯电影有相似氛围的电影。为了确保这种行为，我一点点细致地提供了我想要的行为的模型示例。语言模型的美妙之处在于，它们可以利用自己的“推理”能力和世界知识，将这些小样本学习示例推广到其他用户查询中。
document_content_description = "Brief overview of a movie, along with keywords"# 定义允许的比较器列表allowed_comparators = ["$eq",# Equal to (number, string, boolean)"$ne",# Not equal to (number, string, boolean)"$gt",# Greater than (number)"$gte",# Greater than or equal to (number)"$lt",# Less than (number)"$lte",# Less than or equal to (number)"$in",# In array (string or number)"$nin",# Not in array (string or number)"$exists", # Has the specified metadata field (boolean)]examples = [("Recommend some films by Yorgos Lanthimos.",{"query": "Yorgos Lanthimos","filter": 'in("Directors", ["Yorgos Lanthimos]")',},),("Films similar to Yorgos Lanthmios movies.",{"query": "Dark comedy, absurd, Greek Weird Wave","filter": 'NO_FILTER',},),...]metadata_field_info = [AttributeInfo(name="Title", description="The title of the movie", type="string"),AttributeInfo(name="Runtime (minutes)",description="The runtime of the movie in minutes", type="integer"),AttributeInfo(name="Language",description="The language of the movie", type="string"),...]constructor_prompt = get_query_constructor_prompt(document_content_description,metadata_field_info,allowed_comparators=allowed_comparators,examples=examples,)output_parser = StructuredQueryOutputParser.from_components()query_constructor = constructor_prompt | query_model | output_parserretriever = SelfQueryRetriever(query_constructor=query_constructor,vectorstore=vectorstore,structured_query_translator=PineconeTranslator(),search_kwargs={'k': 10})除了示例之外，模型还必须知道每个元数据字段的描述。这有助于它了解什么是元数据过滤。
最后，我们来构建我们的链。这里的query_model是使用OpenAI API的GPT-4 Turbo的一个实例。我建议使用GPT-4而不是3.5来编写这些元数据过滤器查询，因为这是一个关键步骤，理由是3.5会更频繁地出错。search_kwargs={'k'：10｝告诉检索器根据用户查询找出十部最相似的电影。
创建聊天模型
最后，在构建了自查询检索器之后，我们可以在此基础上构建标准的RAG模型。我们首先定义我们的聊天模型。这就是我所说的摘要模型，因为它采用上下文(检索到的电影+系统消息)，并以每个推荐的摘要作为响应。如果你想降低成本，这个模型可以是GPT-3.5 Turbo;当然，如果你想获得绝对最佳的结果，这个模型也可以是GPT-4 Turbo。
在系统消息中，我告诉机器人它的目标是什么，并提供了一系列建议和限制，其中最重要的是不要推荐自我查询检索器没有提供给它的电影。在测试中，当用户查询没有从数据库中得到电影时，我遇到了问题。例如，查询“推荐一些由韦斯·安德森执导的马特·达蒙主演的1980年之前拍摄的恐怖电影”会导致自我查询检索器无法检索到任何电影(因为尽管听起来很棒，但这部电影并不存在)。在没有电影数据的情况下，该模型会使用自己的(错误的)内存来尝试推荐一些电影。这是不好的行为。我不希望Netflix的推荐人讨论数据库中没有的电影。下面的系统消息成功阻止了此行为。我确实注意到GPT-4比GPT-3.5更善于遵循指令，这是在意料之中的事情。
chat_model = ChatOpenAI(model=SUMMARY_MODEL_NAME,temperature=0,streaming=True,)prompt = ChatPromptTemplate.from_messages([('system',"""Your goal is to recommend films to users based on theirquery and the retrieved context. If a retrieved film doesn't seemrelevant, omit it from your response. If your context is emptyor none of the retrieved films are relevant, do not recommend films, but instead tell the user you couldn't find any filmsthat match their query. Aim for three to five film recommendations,as long as the films are relevant. You cannot recommend more thanfive films. Your recommendation should be relevant, original, andat least two to three sentences long.YOU CANNOT RECOMMEND A FILM IF IT DOES NOT APPEAR IN YOURCONTEXT.# TEMPLATE FOR OUTPUT- **Title of Film**:- Runtime:- Release Year:- Streaming:- (Your reasoning for recommending this film)Question: {question}Context: {context}"""),])def format_docs(docs):return "\n\n".join(f"{doc.page_content}\n\nMetadata: {doc.metadata}" for doc in docs)# Create a chatbot Question & Answer chain from the retrieverrag_chain_from_docs = (RunnablePassthrough.assign(context=(lambda x: format_docs(x["context"])))| prompt| chat_model| StrOutputParser())rag_chain_with_source = RunnableParallel({"context": retriever, "question": RunnablePassthrough()}).assign(answer=rag_chain_from_docs)上述代码中，formatdocs用于格式化提供给模型的信息，使其易于理解和解析。我们向模型提供page_content(概述和关键字)以及元数据(所有其他电影属性);任何它可能需要用来更好地向用户推荐电影的信息。
rag_chain_from_docs是一个链，它获取检索到的文档，并使用format_docs对其进行格式化，然后将格式化的文档馈送到模型用来回答问题的上下文中。最后，我们创建了rag_chain_with_source，这是一个RunnableParallel，顾名思义，它并行运行两个操作：自查询检索器启动以检索类似的文档，而查询只是通过RunnablePassthrough()函数传递给模型。然后将来自这两个并行组件的结果进行组合，并使用rag_chain_from_docs生成答案。这里的source指的是检索器，它可以访问所有的“source”文档。
因为我希望答案是流式的(例如，像ChatGPT这样一块一块地呈现给用户)，所以我们使用了以下代码：
for chunk in rag_chain_with_source.stream(query):for key in chunk:if key == 'answer':yield chunk[key]程序展示
现在进入有趣的部分：与模型一起玩。Streamlit软件是一个用于创建前端和托管应用程序的优秀工具。当然，我不会在本文中讨论所开发软件的用户界面相关的代码;有关此用户界面实现的详细信息，请参阅文后所附的原始代码。当然，这些代码也相当简单，Streamlit网站(上还有很多其他的例子可供参考。
电影搜索实例程序的用户界面(作者本人提供图片)
您可以使用软件中提供的好几个方面的建议，但首先让我们尝试使用自己的查询：
示例查询和模型响应情况(作者本人提供图片)
在底层的代码实现中，这个自我查询的检索器确保过滤掉任何不是法语的电影。然后，它对“成长故事”进行了相似性搜索，得出了十部在此背景下的电影。最后，机器人选择了五部电影进行推荐。请注意建议的电影范围：有些电影的上映日期最早在1959年，最晚在2012年。为了方便起见，我确保机器人提供的信息中包含电影的运行时间、上映年份、流媒体提供商以及机器人手工制作的简短推荐。
(旁注：如果你还没有看过《拳》( The Blows：，请停止你正在做的任何事情，立即去看一看吧。)
值得注意的是，以前在大型语言模型中通常被视为负面的性质，例如其响应的不确定性，现在被系统认为是正面的性质。向模型提出同样的问题两次，你可能会得到略微不同的建议。
重要的是，要注意当前实施的一些局限性：
最后，您可能对本文项目作出的改进之一是，在检索后对文档(进行重新排序。另外，提供一个聊天模型也可能很有趣，因为你可以在多回合的对话中与之交谈，而不仅仅是一个QA机器人。此外，你还可以创建一个推荐器代理(，以便在查询不清楚的情况下向用户提示一个清晰的问题。
最后，祝您的电影搜索玩得开心!
链接参考
译者介绍
朱先忠，社区编辑，专家博客、讲师，潍坊一所高校计算机教师，自由编程界老兵一枚。
原文标题：How to Build a RAG System with a Self-Querying Retriever in LangChain，作者：Ed Izaguirre
链接：
想了解更多AIGC的内容，请访问：
AI.x社区