1.前言

为了避免 llm正确的废话和幻觉，知识库可以说是现在开发 agent的必备了。同时，作为 rag中的 r，知识库召回的成功率会极大的影响 llm的最终回复效果。一般，会把知识库召回的内容作为背景知识给到 llm，并在 prompt中明确告诉它优先使用知识库召回的内容回答用户问题，甚至要求必须基于知识库回答也是可以的。

环境信息如下：

python-3.13、langchain-0.3.26

2.知识库设计

这里设计一个轻量级知识库，基于 csv格式的数据，可用于小型项目，维护成本极低。样例如下：

报错信息,原因分析,修复建议
"java.sql.SQLException: EXECUTION FAILED: Task DDL error HiveException: [Error 20604] Can not create table: create table busi.dwd_cd_eksiq_024323331015027700004_20250720_de083461-142e-4768-38si-bd7046532ca9 failed","Hive相关服务负载过高","排除服务故障后可尝试重做任务"
"The ownership on the staging directory file:/tmp/hadoop/mapred/staging/KS_MACHINE1666781422/.staging is not as expected. It is owned by hctr. The directory must be owned by the submitter KS_MACHINE or KS_MACHINE@LAKEHOUSE.IK","权限信息错误或者kinit没有正确连接","优先联系管理员确认权限问题"
"org.apache.hadoop.hive.ql.metadata.HiveException: Invalid Value ""1"" for decimal(3,3) column when inserting. Column to inserting: name","字段和字段内容不匹配导致无法写入","decimal字段小数点位数不能超过保留位数"

这是一个运维场景，记录了任务报错的日志、原因分析和修复建议。日常维护可以使用 excel进行管理，然后导出成 csv即可，非常的方便快捷。

对于一线的运维来说只是简单记录日常工作，而且能反过来提高自己的工作效率，故一般都不会排斥。

3.加载知识库

首先，读取和解析 csv需要用到 langchain_community的 CSVLoader。如下：

loader = CSVLoader(file_path="task-records.csv",encoding="utf-8",metadata_columns=["原因分析", "修复建议"],content_columns=["报错信息"],
)
docs = loader.load()

这里的重点是：content_columns，内容列，知识库召回时会基于此列的内容，使用 embedding模型进行向量化的时候也是用这一列。

metadata_columns是元数据列，不会参与召回，可用于丰富召回结果。比如补充数据来源、原因分析和修复建议。

其次，需要对文本进行分块，如下：

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
all_splits = text_splitter.split_documents(docs)

chunk_size分块大小为 1000个字符，chunk_overlap设置为0表示片段无重叠，这两个参数会影响召回成功率，需要根据知识库和业务调用实际情况修改。比如业务调用时多用关键词，则需要把 chunk_size调小；反之，如果用的是句子或段落，则需要把chunk_size调大。

4.修改Embedding客户端

这里选取的 embedding模型是 BAAI-bge-small-zh-v1.5离线版，维度是 1024，使用 docker进行部署，适合无法联网的环境，比如内网、局域网。如果使用的是 langchain支持厂商的 embedding模型，可以跳过本步骤。

为了适配此离线 embedding模型，故基于 langchain-openai embedding客户端进行改造，修改内容如下：

// 修改 _get_len_safe_embeddings函数
// 把下面的内容去掉
for i in _iter:response = self.client.create(input=tokens[i : i + _chunk_size], **client_kwargs)if not isinstance(response, dict):response = response.model_dump()batched_embeddings.extend(r["embedding"] for r in response["data"])

这个函数主要影响的是知识库分段后的向量化过程，很重要。

其他的离线 embedding模型可以参考此方式进行修改，不一定适用所有的模型。

5.知识库召回

知识库是需要存储的，这里选用极致轻量化的方案：内存存储，对应的代价就是每次程序重启或知识库更新时需要全量初始化，比较耗时。

首先，进行知识库向量化：

embeddings = BAAIEmbeddings(base_url="http://127.0.0.1:9997/v1",  # 远程embedding服务地址api_key="no-key",model="BAAI-bge-small-zh-v1.5",
)vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents=all_splits)

BAAIEmbeddings这个类是基于 openai进行改造后的，add_documents时会调用 embedding模型进行知识库分片的向量化，并将结果存储在内存中。

至此，知识库已经完全准备好了，可以进行召回测试了，如下：

retrieved_docs = vector_store.similarity_search_with_score(query="/tmp/hadoop/mapred/staging/KS_MACHINE1666781422/.staging is not as expected",k=1,
)
print(retrieved_docs)

这里 k能够控制最终召回的数据条数，在召回成功率低的情况下可以调大。

知识库召回效果如下：

[(Document(id=‘d06c773b-fd98-4de9-b939-ce2d3e513a11’, metadata={‘source’: ‘task-records.csv’, ‘row’: 2, ‘原因分析’: ‘权限信息错误或者kinit没有正确连接’, ‘修复建议’: ‘优先联系管理员确认权限问题’}, page_content=‘报错信息: The ownership on the staging directory file:/tmp/hadoop/mapred/staging/KS_MACHINE1666781422/.staging is not as expected. It is owned by hctr. The directory must be owned by the submitter KS_MACHINE or KS_MACHINE@LAKEHOUSE.IK’), 0.8914968315763799)]

可以看到 langchain准确的找出了知识库中与问题最匹配的数据，并且可以在元数据中看到是来自task-records.csv、原因分析和修复方式，以及文本相似度。这些信息在实际项目中都是需要进行展示的，增加可信度和便于溯源。