从0开始学RAG之Query Construction
输入“/”快速插入内容
从0开始学RAG之Query Construction
从0开始学RAG之Query Construction
2024年8月23日修改
作者：lucas大叔
人们对将自然语言与各种类型的数据（结构化、非结构化和 半结构化 ）无缝连接具有非常浓厚的兴趣。但是，对每种数据类型而言，这种新兴的“LUI”（ language user interface ）方式都有特定的挑战和考虑：
什么是 Query Construction
典型的RAG方法将用户query转换为向量表示，然后将该向量与源文档的向量表示进行比较，找到具有最相似的向量表示的源文档。这种方式对非结构化数据相当有效，但对结构化数据呢？
许多用户query的最佳答案不仅是通过在embedding空间中找到相似的文档或数据，而且还利用数据中固有的以及在用户query中表达的结构。
例如，对于“what are movies about aliens in the year 1980 ”这个查询，一部分（aliens ）可能要从语义上查找，但也有一部分（"year == 1980"）我们想以精确的方式查找。
Text-to-metadata-filter
代码块
from langchain_community.document_loaders import YoutubeLoader
docs = YoutubeLoader.from_youtube_url(
"https://www.youtube.com/watch?v=pbAd8O1Lvm4", add_video_info=True
).load()
docs[0].metadata
查看video的元数据
代码块
{'source': 'pbAd8O1Lvm4',
'title': 'Self-reflective RAG with LangGraph: Self-RAG and CRAG',
'description': 'Unknown',
'view_count': 9006,
'thumbnail_url': 'https://i.ytimg.com/vi/pbAd8O1Lvm4/hq720.jpg',
'publish_date': '2024-02-07 00:00:00',
'length': 1058,
'author': 'LangChain'}
为了生成结构化的query，首先需要定义 query schema 。可以看到每个文档都有标题、浏览次数、发布日期和长度（以秒为单位）。假设已经构建了一个索引，该索引允许我们对每个文档的内容和标题执行非结构化搜索，并对浏览次数、发布日期和长度等维度使用区间过滤。
首先，创建 schema定义浏览次数、发布日期和视频长度的最小和最大属性，以便后续可以基于这些属性过滤。此外，还将添加单独的属性用于针对内容和视频标题的检索。
代码块
import datetime
from typing import Literal, Optional, Tuple
class TutorialSearch(BaseModel):
"""Search over a database of tutorial videos about a software library."""
content_search: str = Field(
...,
description="Similarity search query applied to video transcripts.",
)
title_search: str = Field(
...,
description=(
"Alternate version of the content search query to apply to video titles. "
"Should be succinct and only include key words that could be in a video "
"title."
),
)
min_view_count: Optional[int] = Field(
None,
description="Minimum view count filter, inclusive. Only use if explicitly specified.",
)
max_view_count: Optional[int] = Field(
None,
description="Maximum view count filter, exclusive. Only use if explicitly specified.",
)
earliest_publish_date: Optional[datetime.date] = Field(
None,
description="Earliest publish date filter, inclusive. Only use if explicitly specified.",
)
latest_publish_date: Optional[datetime.date] = Field(
None,
description="Latest publish date filter, exclusive. Only use if explicitly specified.",
)
min_length_sec: Optional[int] = Field(
None,
description="Minimum video length in seconds, inclusive. Only use if explicitly specified.",
)
max_length_sec: Optional[int] = Field(
None,
description="Maximum video length in seconds, exclusive. Only use if explicitly specified.",
)
def pretty_print(self) -> None:
for field in self.__fields__:
if getattr(self, field) is not None and getattr(self, field) != getattr(
):
print(f"{field}: {getattr(self, field)}")