企业官网建设流程全解析

llmware 语义检索实战：用 Query 类构建从文档库到语义查询的完整 RAG 检索链路

【免费下载链接】llmwareUnified framework for building enterprise RAG pipelines with small, specialized models项目地址: https://gitcode.com/GitHub_Trending/ll/llmware

llmware 的检索（Retrieval）能力是其企业级 RAG 管线的核心环节。本文基于官方示例文档 docs/examples/retrieval.md，以一个自包含的可运行示例为主线，完整讲清楚三件事：如何创建并加载一个财务文档样本库（FinDocs）、如何为其构建嵌入索引（embeddings）、如何用Query类的semantic_query方法执行语义查询并通过embedding_distance_threshold控制召回质量。读完后你可以直接复制示例代码在本地跑通"建库 → 嵌入 → 查询"全流程，并理解每个参数在 llmware/retrieval.py 源码中的真实作用。

一、示例的整体流程

官方示例的核心逻辑在 llmware/retrieval.py 的模块文档中被概括为：Query类提供对Library集合的高层查询接口，支持三类检索策略：

• 文本检索（text retrieval）：直接作用于文本集合数据库，不依赖向量库；
• 语义检索（semantic retrieval）：依赖向量数据库，且要求已为该 Library 预先构建 embeddings；
• 混合策略（hybrid）：结合文本与语义查询的便捷方法，如dual_pass_query。

示例代码的完整可执行版本位于 solutions/sources/semantic_retrieval.py，与文档中的代码完全一致。整体流程分三步：

LLMWareConfig().set_active_db("sqlite") # 选择文本集合存储 ↓ Library().create_new_library("lib_semantic_query_1") + Setup().load_sample_files() # 下载 FinDocs 财务文档 + library.add_files(...) # 解析入库 + library.install_new_embedding(...) # 构建嵌入索引 ↓ Query(library).semantic_query("ESG initiatives", result_count=20)

二、前提：选择文本集合的底层数据库

示例入口处执行：

LLMWareConfig().set_active_db("sqlite")

这一行决定了 Library 的文本集合（解析后的 block 数据）存储在哪种数据库中，与语义检索所用的向量库（chromadb/Milvus 等）是两个独立的选择。从 llmware/configs.py 的源码看，set_active_db会将新值写入配置项collection_db，并校验该值必须在支持列表cls._supported["collection_db"]内，否则抛出LLMWareException：

@classmethod def set_active_db(cls, new_db): """ Sets the default database for Library text collections """ if new_db in cls._supported["collection_db"]: cls._conf["collection_db"] = new_db else: raise LLMWareException(message=f"LLMWareConfig - set_active_db - selected " f"db is not supported - {new_db}")

使用sqlite意味着零外部依赖，适合首次在本机体验完整流程。

三、步骤一：创建并加载 FinDocs 样本库

import os from llmware.library import Library from llmware.setup import Setup def create_fin_docs_sample_library(library_name): print(f"update: creating library - {library_name}") library = Library().create_new_library(library_name) sample_files_path = Setup().load_sample_files(over_write=False) ingestion_folder_path = os.path.join(sample_files_path, "FinDocs") parsing_output = library.add_files(ingestion_folder_path) return library

三个关键调用逐一说明：

1. Library().create_new_library(library_name)：显式构造器，创建新库；从 llmware/library.py 的 docstring 可见，"如果同名库已存在，则加载已有库"（If a library with the same name already exists, it will load the existing library），并且库名会被做安全性检查与改写。因此该步骤是幂等的，重复运行示例不会因库名冲突而失败。
2. Setup().load_sample_files(over_write=False)：从 llmware 维护的公开 AWS S3 桶下载样本文件到<llmware_path>/sample_files。从 llmware/setup.py 的源码看：若sample_files目录已存在且over_write=False，则直接返回本地路径而不重新下载（这也是示例注释"may take a few minutes the first time"的原因）；传over_write=True会拉取最新版本。样本文件覆盖八个领域，与本示例相关的FinDocs 约为 15 份财务年报、财报与 10-K 文件——这正好解释了为何示例查询词选择了 "ESG initiatives"、"stock performance" 这类财务语料主题。
3. library.add_files(ingestion_folder_path)：批量解析目录中的文档，产出结构化的 text/table/image block 并写入文本集合，返回解析统计结果（示例中赋值给parsing_output，可用于确认解析块数与文档数）。

四、步骤二：构建嵌入索引

library.install_new_embedding( embedding_model_name="industry-bert-sec", vector_db="chromadb", batch_size=200 )

该调用为整个库构建语义检索的向量索引，有三个参数需要把握（install_new_embedding定义见 llmware/library.py）：

文档中给出的两条实践建议值得保留：

• 内存受限的笔记本：(1) 调小batch_size；(2) 换用更小的嵌入模型"mini-lm-sbert"；
• 已有向量库环境：可将vector_db替换为 Milvus 等，Query类会自动从库的嵌入记录（embedding record）中读取对应的库名与模型名，无需在查询侧额外配置。

五、步骤三：用 Query 类执行语义查询

5.1 实例化与返回键控制

from llmware.retrieval import Query q = Query(library) # 可选：只返回需要的键，默认返回完整键集 q.query_result_return_keys = ["distance", "file_source", "page_num", "text"]

Query的初始化逻辑（llmware/retrieval.py）值得细看，它决定了语义检索能否"开箱即用"：

• 构造器会读取该库的embedding 状态记录（self.library.get_embedding_status()）。若库上存在状态为"yes"的嵌入记录，则自动绑定对应的embedding_db与embedding_model_name，并将search_mode置为"semantic"，随后加载嵌入模型；若找不到有效嵌入记录，则回落到"text"模式。
• 若库上存在多组嵌入（多个嵌入模型或多个向量库），可通过构造参数embedding_model_name、vector_db显式指定查询哪一组。
• 返回键有三档默认集，源码 llmware/retrieval.py 定义如下：

# 完整键集（默认值） self.query_result_standard_keys = ["_id", "text", "doc_ID", "block_ID", "page_num", "content_type", "author_or_speaker", "special_field1", "file_source", "added_to_collection", "table", "coords_x", "coords_y", "coords_cx", "coords_cy", "external_files", "score", "similarity", "distance", "matches"] # 精简集 self.query_result_short_keys = ["text", "file_source", "page_num", "score", "distance", "matches"] # 最小必需集（set_output_keys 时会自动补齐并合并进结果） self.query_result_min_required_keys = ["text", "file_source", "page_num"]

因此默认每条结果都会携带约 20 个字段（含坐标、相似度、命中位置matches等）；示例中把query_result_return_keys手动收缩为 4 个键，只为打印和下游消费保留distance、file_source、page_num、text。如需程序化设置，也可以调用q.set_output_keys([...])，它会对键做合法性校验并自动补回text/file_source/page_num这三个最小必需键（见 llmware/retrieval.py）。

5.2 三个递进的查询

# 查询 1：基本语义查询 my_query = "ESG initiatives" query_results1 = q.semantic_query(my_query, result_count=20) for i, result in enumerate(query_results1): print("results - ", i, result) # 查询 2：换主题、换召回数量 my_query2 = "stock performance" query_results2 = q.semantic_query(my_query2, result_count=10) # 查询 3：加大召回并设置距离阈值 my_query3 = "cloud computing" # 注意：embedding_distance_threshold 会截断 distance >= 1.0 的结果 query_results3 = q.semantic_query(my_query3, result_count=50, embedding_distance_threshold=1.0)

semantic_query的完整签名（llmware/retrieval.py）为：

def semantic_query(self, query, result_count=20, embedding_distance_threshold=None, custom_filter=None, results_only=True):

各参数的语义与默认行为：

• query：查询文本，会被嵌入模型编码为查询向量；
• result_count（默认 20）：请求返回的块数上限；
• embedding_distance_threshold：距离阈值。不传时使用实例属性self.semantic_distance_threshold，其默认值在__init__中设为1000（源码注释：# basic shut off at such a high level，即默认实际不做截断）。传入1.0后，只保留嵌入空间距离小于 1.0 的块——这正好对应查询 3 中"cloud computing"与财务语料语义距离较远、需要阈值过滤无关召回的场景。
• custom_filter：可选的{键: 值}精确过滤字典，在语义结果返回后应用（见下节源码）；
• results_only（默认 True）：True 时返回结果列表；False 时返回包含query/results/doc_ID/file_source的完整字典。

从实现看（llmware/retrieval.py），semantic_query的执行链路为：

1. self.load_embedding_model()确保嵌入模型就绪，否则抛出ModelNotFoundException；
2. self.embedding_model.embedding(query)生成查询向量；
3. 调用self.embeddings.search_index(...)（EmbeddingHandler方法，定义于 llmware/embeddings.py）在向量库中检索，返回的每个元素是[block数据, 距离]的二元组；
4. 逐条过滤：if blocks[1] < embedding_distance_threshold才保留，并写入distance、semantic: "semantic"、score: 0.0；
5. 若提供custom_filter，调用apply_custom_filter做键值全匹配的二次筛选；
6. 交给内部方法_cursor_to_qr打包：定位命中位置生成matches、补默认score/similarity/distance为零值、按query_result_return_keys抽取输出键、附加account_name/library_name，并把本次查询登记进query_history（由save_history控制，默认开启）。

另外注意一个健壮性设计：在通用的query()入口中，若请求query_type="semantic"但嵌入模型不可用，会静默回退到文本查询（llmware/retrieval.py），而直接调用semantic_query则会抛错——示例直接调用semantic_query，正是因为库上已确认构建了有效嵌入。

六、结果如何解读

每条结果是一个 dict。示例输出中你至少应关注：

• text：召回的文本块内容（库的默认分块目标大小约为 400 字符，见 llmware/library.py 中block_size_target_characters = 400）；
• distance：查询向量与该块嵌入向量的距离，数值越小语义越接近，查询 3 的embedding_distance_threshold=1.0即以此为截断线；
• file_source/page_num：来源文件与页码（内部字段名为master_index，打包时统一映射为page_num，见 llmware/retrieval.py），是 RAG 答案溯源与引用标注的基础；
• matches：查询词在块内文本中的命中位置列表（locate_query_match生成）。

七、由该示例延伸的 Query 能力

文档主线是语义查询，但同一个Query实例还支持若干实用变体，可在 llmware/retrieval.py 中逐一查证：

• text_query(query, exact_mode=False, ...)：基于倒排/文本匹配的检索，支持精确匹配预处理；
• text_query_with_document_filter/semantic_query_with_document_filter：在结果集上叠加doc_ID或file_source文档级过滤；
• text_query_with_custom_filter(query, filter_dict, ...)：按任意合法键（如content_type、page_num）做字典过滤，filter_dict中每个键值对等价于 AND 条件；
• similar_blocks_embedding(block, embedding_distance_threshold=10, ...)：以某个已有块为"锚点"查找语义近邻块；
• dual_pass_query(query, result_count=20, primary="text", ...)：同时执行文本与语义两路查询，按_id交叉比对，把两路都命中的块标记为matched并优先排在合并结果头部（match_status标记matched/primary_only/secondary_only）。源码中有一个显式的性能安全阀：当result_count > 100时会告警并自动钳制为 100（n² 比对不擅长超长列表），可通过safety_check=False关闭（llmware/retrieval.py）。

对于 RAG 场景，"文本 + 语义"双路召回再重排的dual_pass_query通常是比单路语义查询更稳健的生产选择，可以作为示例之后的进阶练习。

八、完整可运行示例

以下代码整合了原文档的全部要素，可直接保存为脚本运行（首次运行需要网络下载样本文件与嵌入模型）：

import os from llmware.library import Library from llmware.retrieval import Query from llmware.setup import Setup from llmware.configs import LLMWareConfig def create_fin_docs_sample_library(library_name): print(f"update: creating library - {library_name}") library = Library().create_new_library(library_name) sample_files_path = Setup().load_sample_files(over_write=False) ingestion_folder_path = os.path.join(sample_files_path, "FinDocs") parsing_output = library.add_files(ingestion_folder_path) print("update: building embeddings - may take a few minutes the first time") # 如已安装 Milvus 或其他向量库，可替换 vector_db # 内存受限时：调小 batch_size，或换用 "mini-lm-sbert" 嵌入模型 library.install_new_embedding(embedding_model_name="industry-bert-sec", vector_db="chromadb", batch_size=200) return library def basic_semantic_retrieval_example(library): q = Query(library) q.query_result_return_keys = ["distance", "file_source", "page_num", "text"] query_results1 = q.semantic_query("ESG initiatives", result_count=20) print("\nQuery 1 - ESG initiatives") for i, result in enumerate(query_results1): print("results - ", i, result) query_results2 = q.semantic_query("stock performance", result_count=10) print("\nQuery 2 - stock performance") for i, result in enumerate(query_results2): print("results - ", i, result) # embedding_distance_threshold=1.0 会截断 distance >= 1.0 的结果 query_results3 = q.semantic_query("cloud computing", result_count=50, embedding_distance_threshold=1.0) print("\nQuery 3 - cloud computing") for i, result in enumerate(query_results3): print("result - ", i, result) return [query_results1, query_results2, query_results3] if __name__ == "__main__": print("Example - Running a Basic Semantic Query") LLMWareConfig().set_active_db("sqlite") lib = create_fin_docs_sample_library("lib_semantic_query_1") my_results = basic_semantic_retrieval_example(lib)

九、关键参考文件

适用前提提示：示例默认使用chromadb向量库与sqlite文本库，适合单机快速验证；生产环境建议将vector_db替换为常驻的 Milvus 等向量数据库，并依据内存情况调整batch_size。

【免费下载链接】llmwareUnified framework for building enterprise RAG pipelines with small, specialized models项目地址: https://gitcode.com/GitHub_Trending/ll/llmware