基于人工智能的自然语言查询用于知识发现
应用人工智能的自然语言查询技术进行知识发现
在这篇文章中，我想分享一个我一直在进行的名为UE5_documentalist的概念验证项目。这是一个令人兴奋的项目，它使用自然语言处理（NLP）来潜在地增强您在大量文档中的体验。
虽然我在这个项目中使用了Unreal Engine 5的文档，但它可以应用于任何类型的用例，比如您公司的内部文档。
什么是UE5_documentalist？
UE5_documentalist是一个智能助手，旨在简化您在Unreal Engine 5.1或任何其他文档中的导航。通过利用NLP技术，该项目允许您进行自然语言查询，轻松找到超过1700个网页中最相关的部分。
例如，您可以查询“我可以使用什么系统来避免代理之间的碰撞？”并被重定向到最适合您需求的文档。
您可以在此存储库中查看我的代码。
演示
为了让您更好地了解UE5_documentalist的功能，这里有一个快速演示展示它的能力：
演示（图片由作者提供）
它是如何工作的？
步骤1 – 网页抓取
我首先抓取了Unreal Engine 5.1的网页文档，将HTML输出转换为Markdown，并将文本保存到一个字典中。
抓取函数如下所示：
def main(limit, urls_registry, subsections_path): urls_registry = "./src/utils/urls.txt" with open(urls_registry, 'r') as f: urls = f.read() urls = urls.split('\n') # 初始化用于存储子部分的字典 subsections = {} for idx, url in enumerate(urls): # 达到限制时停止 if limit is not None: if idx > limit: break if idx % 100 == 0: print(f"处理网页 {idx}") # 发送请求 try: with urllib.request.urlopen(url) as f: content = f.read() except HTTPError as e: print(f"网页 {url} 出错") print('服务器无法完成请求。') print('错误代码：', e.code) continue except URLError as e: print(f"网页 {url} 出错") print('无法连接服务器。') print('原因：', e.reason) continue # 解析内容 md_content = md(content.decode('utf-8')) preproc_content = split_text_into_components(md_content) # 提取URL名称信息 subsection_title = extract_info_from_url(url) # 添加到字典中 subsections[url] = { "title": subsection_title, "content": preproc_content } # 保存字典 with open(subsections_path, 'w') as f: json.dump(subsections, f)该函数调用了另一个名为split_text_into_components的函数，它是一系列预处理步骤，删除链接、图像、加粗和标题等特殊字符。
最终，该函数返回一个如下所示的字典：
{url : {'title' : '一些上下文标题', 'content' : '页面内容' }...}步骤2 – 嵌入向量
基于此，我使用了Instructor-XL模型生成嵌入向量。一开始，我使用了OpenAI的text-embedding-ada-002模型。它非常便宜（超过1.7亿字符不到2美元），速度快，全球适用于我的用例。然而，我切换到了Instructor-XL模型，它运行稍慢，但可以在本地使用，而无需与在线API通信。如果您关心隐私或者处理敏感数据，这个解决方案更好。
嵌入函数如下：
def embed(subsection_dict_path, embedder, security): """嵌入目录中的文件。 参数： subsection_dict_path (dict): 包含子节的字典路径。 security (str): 安全设置。可选值为“activated”或“deactivated”。 如果不是“deactivated”，则阻止函数运行 并避免意外费用。 返回： embeddings (dict): 包含嵌入内容的字典。 """ # 如果嵌入内容已经存在，则加载 if os.path.exists(os.path.join("./embeddings", f'{embedder}_embeddings.json')): print("嵌入内容已存在。正在加载。") with open(os.path.join("./embeddings", f'{embedder}_embeddings.json'), 'r') as f: embeddings = json.load(f) else: # 初始化用于存储嵌入内容的字典 embeddings = {} # 检查安全设置，如果embedder是openai（避免错误花费$$$） if security != "deactivated": if embedder == 'openai': raise Exception("安全设置未停用。") # 加载子节 with open(subsection_dict_path, 'r') as f: subsection_dict = json.load(f) # 仅供调试目的 # 计算要嵌入的平均文本长度 dict_len = len(subsection_dict) total_text_len = 0 for url, subsection in subsection_dict.items(): total_text_len += len(subsection['content']) avg_text_len = total_text_len / dict_len # 如果embedder是'openai'，则初始化openai api if embedder == "openai": openai_model = "text-embedding-ada-002" # 从环境变量中获取API密钥，如果不存在则提示用户输入 api_key = os.getenv('API_KEY') if api_key is None: api_key = input("请输入您的OpenAI API密钥：") openai.api_key = api_key # 如果embedder是'instructor'，则初始化instructor模型 elif embedder == "instructor": instructor_model = INSTRUCTOR('hkunlp/instructor-xl') # 如果GPU可用，则设置设备为GPU if (torch.backends.mps.is_available()) and (torch.backends.mps.is_built()): device = torch.device("mps") elif torch.cuda.is_available(): device = torch.device("cuda") else: device = torch.device("cpu") else: raise ValueError(f"Embedder必须是'openai'或'instructor'。而不是{embedder}") # 遍历子节 for url, subsection in tqdm(subsection_dict.items()): subsection_name = subsection['title'] text_to_embed = subsection['content'] # 如果已经嵌入过了，则跳过 if url in embeddings.keys(): continue # 发送嵌入请求 # 情况1：openai if embedder == 'openai': try: response = openai.Embedding.create( input=text_to_embed, model=openai_model ) embedding = response['data'][0]['embedding'] except InvalidRequestError as e: print(f"url {url} 处发生错误") print('服务器无法满足请求。') print('错误代码：', e.code) print(f'尝试嵌入 {len(text_to_embed)} 个字符，而平均值为 {avg_text_len}') continue # 情况2：instructor elif embedder == 'instructor': instruction = "代表用于检索的UnrealEngine文档：" embedding = instructor_model.encode([[instruction, text_to_embed]], device=device) embedding = [float(x) for x in embedding.squeeze().tolist()] else: raise ValueError(f"Embedder必须是'openai'或'instructor'。而不是{embedder}") # 将嵌入内容添加到字典中 embeddings[url] = { "title": subsection_name, "embedding": embedding } # 每100次迭代保存一次字典 if len(embeddings) % 100 == 0: print(f"已完成 {len(embeddings)} 次迭代后保存嵌入内容。") # 将嵌入内容保存到pickle文件 with open(os.path.join("./embeddings", f'{embedder}_embeddings.pkl'), 'wb') as f: pickle.dump(embeddings, f) # 将嵌入内容保存到json文件 with open(os.path.join("./embeddings", f'{embedder}_embeddings.json'), 'w') as f: json.dump(embeddings, f) return embeddings该函数返回一个类似以下的字典：
{url : {'title' : '一些上下文标题', 'embedding' : 内容的向量表示 }...}步骤3 — 向量索引数据库
然后将这些嵌入上传到在Docker容器上运行的Qdrant向量索引数据库中。
使用以下Docker命令启动数据库：
docker pull qdrant/qdrantdocker run -d -p 6333:6333 qdrant/qdrant
并使用以下函数填充数据库（您将需要qdrant_client软件包：pip install qdrant-client）
import qdrant_client as qcimport qdrant_client.http.models as qmodelsimport uuidimport jsonimport argparsefrom tqdm import tqdmclient = qc.QdrantClient(url="localhost")METRIC = qmodels.Distance.DOTCOLLECTION_NAME = "ue5_docs"def create_index(): client.recreate_collection( collection_name=COLLECTION_NAME, vectors_config = qmodels.VectorParams( size=DIMENSION, distance=METRIC, ) )def create_subsection_vector( subsection_content, section_anchor, page_url ): id = str(uuid.uuid1().int)[:32] payload = { "text": subsection_content, "url": page_url, "section_anchor": section_anchor, "block_type": 'text' } return id, payloaddef add_doc_to_index(embeddings, content_dict): ids = [] vectors = [] payloads = [] for url, content in tqdm(embeddings.items()): section_anchor = content['title'] section_vector = content['embedding'] section_content = content_dict[url]['content'] id, payload = create_subsection_vector( section_content, section_anchor, url ) ids.append(id) vectors.append(section_vector) payloads.append(payload) # Add vectors to collection client.upsert( collection_name=COLLECTION_NAME, points=qmodels.Batch( ids = [id], vectors=[section_vector], payloads=[payload] ), )
此函数将网页URL和相关内容嵌入后，将其上传到Qdrant数据库中。
您必须上传ID、矢量和有效载荷作为矢量数据库。在这里，我们的ID是程序生成的，我们的矢量是嵌入（稍后将与查询匹配），有效载荷是额外的信息。
在我们的案例中，在有效载荷中包含的主要信息是Web URL和Web页面的内容。这样，当我们将查询与Qdrant数据库中最相关的条目匹配时，我们可以打印文档并打开网页。
最后一步 – 查询
现在我们可以对数据库进行查询。
查询必须首先与文档中使用的相同的embedder进行嵌入。我们可以使用以下函数来进行嵌入：
def embed_query(query, embedder): if embedder == "openai": # 从环境变量中获取API密钥或提示用户输入 api_key = os.getenv('API_KEY') if api_key is None: api_key = input("Please enter your OpenAI API key: ") openai_model = "text-embedding-ada-002" openai.api_key = api_key response = openai.Embedding.create( input=query, model=openai_model ) embedding = response['data'][0]['embedding'] elif embedder == "instructor": instructor_model = INSTRUCTOR('hkunlp/instructor-xl') # 如果可用，将设备设置为GPU if (torch.backends.mps.is_available()) and (torch.backends.mps.is_built()): device = torch.device("mps") elif torch.cuda.is_available(): device = torch.device("cuda") else: device = torch.device("cpu") instruction = "Represent the UnrealEngine query for retrieving supporting documents:" embedding = instructor_model.encode([[instruction, query]], device=device) embedding = [float(x) for x in embedding.squeeze().tolist()] else: raise ValueError("Embedder must be 'openai' or 'instructor'") return embedding
然后使用嵌入的查询来查询Qdrant数据库：
def query_index(query, embedder, top_k=10, block_types=None): """ 查询与给定查询匹配的Qdrant向量索引数据库中的文档。 Args: query（str）：要搜索的查询。 embedder（str）：要使用的embedder。必须是“openai”或“instructor”。 top_k（int，可选）：要返回的文档的最大数量。默认为10。 block_types（str或str列表，可选）：要搜索的文档块的类型。默认为“text”。 Returns: 一个列表，表示匹配的文档，按相关性排序。每个字典包含以下键： - “id”：文档的ID。 - “score”：文档的相关性得分。 - “text”：文档的文本内容。 - “block_type”：与查询匹配的文档块的类型。 """ collection_name = get_collection_name() if not collection_exists(collection_name): raise Exception(f"Collection {collection_name} does not exist. Exisiting collections are: {list_collections()}") vector = embed_query(query, embedder) _search_params = models.SearchParams( hnsw_ef=128, exact=False ) block_types = parse_block_types(block_types) _filter = models.Filter( must=[ models.Filter( should= [ models.FieldCondition( key="block_type", match=models.MatchValue(value=bt), ) for bt in block_types ] ) ] ) results = CLIENT.search( collection_name=collection_name, query_vector=vector, query_filter=_filter, limit=top_k, with_payload=True, search_params=_search_params, ) results = [ ( f"{res.payload['url']}#{res.payload['section_anchor']}", res.payload["text"], res.score ) for res in results ] return resultsdef ue5_docs_search( query, embedder=None, top_k=10, block_types=None, score=False, open_url=True): """ 搜索与给定查询相关的文档的Qdrant向量索引数据库，并打印顶部结果。 Args: query (str): 要搜索的查询。 embedder (str): 要使用的embedder。必须是“openai”或“instructor”。 top_k (int, optional): 要返回的文档的最大数量。默认为10。 block_types (str or list of str, optional): 要搜索的文档块的类型。默认为“text”。 score (bool, optional): 是否在输出中包含相关性得分。默认为False。 open_url (bool, optional): 是否在网络浏览器中打开顶部URL。默认为True。 Returns: None """ # 检查embedder是否为'openai'或'instructor'。如果不是则引发错误 assert embedder in ['openai', 'instructor'], f"Embedder must be 'openai' or 'instructor'. Not {embedder}" results = query_index( query, embedder=embedder, top_k=top_k, block_types=block_types ) print_results(query, results, score=score) if open_url: top_url = results[0][0] webbrowser.open(top_url)
这些函数调用我仓库中的其他内部处理函数。
就是这样！查询被嵌入，匹配到最接近的文档嵌入（您可以选择评估度量），结果将在终端中打印，并自动打开相应的网页！
想自己试试吗？
如果您有兴趣探索UE5_documentalist，请参阅我在您本地的设置步骤的详细的逐步指南。嵌入已经完成，所以您可以直接开始填充Qdrant数据库，并且您将在我的GitHub仓库中找到所有必要的资源和详细的说明。
目前，它使用Python命令运行。
为什么试一下呢？
UE5_documentalist旨在简化搜索冗长的文档，可能为您节省宝贵的开发时间。通过用简单的英语提问，您将被引导到回答您问题的特定部分。这个工具可能会改善您的体验，并允许您更多地关注使用Unreal Engine 5.1构建令人惊叹的项目或文档。
截至目前，我对UE5_documentalist的进展感到自豪，并且很希望听到您的反馈和改进建议。
太长了，没看：简要介绍：
介绍UE5_documentalist – 一款由NLP驱动的智能文档助手。它使开发人员可以使用自然语言查询轻松导航Unreal Engine 5.1的文档。查看演示，并在GitHub仓库中找到设置指南，也许，如果效果好，您可能会告别繁琐的文档搜索，并潜在地提高您的知识发现体验！
链接
鸣谢
我从一个Reddit用户那里得到了简化UE5文档的想法，他用LLMs开始了一个项目（我忘记了他的名字，但这是他的存储库的链接）。
我基于这篇TDS文章（作者是Jacob Marks）。他的代码对于大部分解析和索引步骤非常有用，没有他详细的文章，我无法成功。不要犹豫，去看看他的工作！
文章最初发布于这里，获得了转载许可。
