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、原因分析和修復方式，以及文本相似度。這些信息在實際項目中都是需要進行展示的，增加可信度和便于溯源。
6.總結
本文描述了如何使用 langchain構建一個輕量級的知識庫，步驟如下：
- 設計csv格式的知識庫
- 加載csv文件
- 文本分片
- 文本向量化與存儲
- 文本召回
其實基于其他的存儲，比如 pg、es，步驟類似，其次知識庫的工程化不能忽視，否則會極大的影響召回成功率。
在實際項目中，召回數據通常會有很多條，一般生產環境會默認允許召回 20條、30條甚至更多，故還可以增加 rerank模型進行重排，進一步增加知識庫檢索準確率。
