據(jù)庫(kù)DML實(shí)戰(zhàn)：構(gòu)建動(dòng)態(tài)可維護(hù)的RAG應(yīng)用)
如果你正在構(gòu)建一個(gè)基于大語(yǔ)言模型LLM的智能應(yīng)用比如一個(gè)能回答專業(yè)問(wèn)題的知識(shí)庫(kù)助手你很可能遇到過(guò)這個(gè)核心難題如何讓模型“記住”并快速“找到”海量的私有文檔信息直接讓模型去“閱讀”成千上萬(wàn)份PDF、Word文檔是不現(xiàn)實(shí)的這會(huì)導(dǎo)致響應(yīng)極慢、成本高昂且容易超出模型的上下文長(zhǎng)度限制。主流的解決方案是“檢索增強(qiáng)生成”RAG其核心在于一個(gè)高效的向量數(shù)據(jù)庫(kù)。而 Milvus正是這個(gè)領(lǐng)域備受矚目的開(kāi)源明星。但很多開(kāi)發(fā)者止步于“Hello World”式的示例把文檔灌進(jìn)去問(wèn)個(gè)簡(jiǎn)單問(wèn)題得到答案就覺(jué)得大功告成。然而真實(shí)的生產(chǎn)環(huán)境是動(dòng)態(tài)的知識(shí)需要更新錯(cuò)誤的文檔需要?jiǎng)h除過(guò)時(shí)的信息需要修正。這時(shí)你就會(huì)遇到本文要解決的核心問(wèn)題如何對(duì) Milvus 向量數(shù)據(jù)庫(kù)中的數(shù)據(jù)進(jìn)行可靠的增、刪、改DML操作并確保你的 LangChain RAG 應(yīng)用能隨之動(dòng)態(tài)、準(zhǔn)確地響應(yīng)本文將帶你超越基礎(chǔ)教程深入langchain-milvus集成的 DML 實(shí)戰(zhàn)。我們不僅會(huì)演示如何插入、刪除和更新數(shù)據(jù)更會(huì)剖析這些操作背后的原理、在 LangChain 鏈中的最佳集成位置以及你必須避開(kāi)的那些“坑”——比如刪除操作的實(shí)際行為、元數(shù)據(jù)過(guò)濾的陷阱以及如何保證數(shù)據(jù)變更后檢索的實(shí)時(shí)性。讀完本文你將能構(gòu)建一個(gè)真正可維護(hù)、可演化的 RAG 應(yīng)用而不僅僅是一個(gè)靜態(tài)的演示原型。1. 為什么 Milvus 的 DML 操作是 RAG 應(yīng)用的“生命線”在原型階段我們往往只關(guān)心“寫入”和“查詢”。但一旦應(yīng)用上線數(shù)據(jù)就活了。想象以下幾個(gè)場(chǎng)景知識(shí)更新公司發(fā)布了新的產(chǎn)品手冊(cè)你需要將新文檔的向量添加到數(shù)據(jù)庫(kù)中而不是重建整個(gè)庫(kù)。錯(cuò)誤修正發(fā)現(xiàn)之前導(dǎo)入的一份技術(shù)規(guī)格表有錯(cuò)誤你需要定位并刪除或更新對(duì)應(yīng)的錯(cuò)誤信息片段。數(shù)據(jù)隔離你的系統(tǒng)服務(wù)于多個(gè)客戶多租戶需要能安全地刪除某個(gè)客戶的全部數(shù)據(jù)。合規(guī)要求根據(jù)用戶請(qǐng)求需要從知識(shí)庫(kù)中徹底抹去某些敏感信息。如果缺乏可靠的 DML數(shù)據(jù)操作語(yǔ)言支持上述每一個(gè)場(chǎng)景都會(huì)讓你陷入困境要么全量重建向量庫(kù)耗時(shí)耗力要么容忍數(shù)據(jù)的不一致和錯(cuò)誤。Milvus 的核心價(jià)值之一就是它提供了完整、高效的 DML 接口insert,delete,upsert并且與 LangChain 的VectorStore抽象層進(jìn)行了深度集成。這使得在 LangChain 應(yīng)用中進(jìn)行數(shù)據(jù)維護(hù)變得像操作傳統(tǒng)數(shù)據(jù)庫(kù)一樣直觀但背后卻涉及向量索引的復(fù)雜管理。本文將聚焦于langchain-milvus這個(gè)官方集成包通過(guò)實(shí)戰(zhàn)代碼讓你掌握這條“生命線”的操控方法。2. 核心概念與工具準(zhǔn)備在開(kāi)始實(shí)戰(zhàn)前我們需要統(tǒng)一幾個(gè)關(guān)鍵概念并準(zhǔn)備好戰(zhàn)場(chǎng)。2.1 關(guān)鍵概念澄清Milvus一個(gè)高性能、云原生的開(kāi)源向量數(shù)據(jù)庫(kù)專為海量向量數(shù)據(jù)的存儲(chǔ)、索引和檢索而設(shè)計(jì)。它支持標(biāo)量過(guò)濾、動(dòng)態(tài) Schema、數(shù)據(jù)分片等高級(jí)特性。LangChain一個(gè)用于構(gòu)建由 LLM 驅(qū)動(dòng)的應(yīng)用程序的框架。它通過(guò)“鏈”Chain將模型、提示、記憶、檢索器等組件連接起來(lái)。其VectorStore模塊提供了與各種向量數(shù)據(jù)庫(kù)交互的統(tǒng)一接口。langchain-milvus這是 LangChain 官方維護(hù)的 Milvus 集成包。它實(shí)現(xiàn)了VectorStore接口封裝了 Milvus Python SDK 的復(fù)雜調(diào)用讓我們可以用更 LangChain 風(fēng)格的方式操作 Milvus。DML in Vector DB與傳統(tǒng)數(shù)據(jù)庫(kù)的增刪改查類似但在向量數(shù)據(jù)庫(kù)中插入Insert將文本經(jīng)過(guò)嵌入模型Embedding Model轉(zhuǎn)化為向量后連同元數(shù)據(jù)metadata一起存入集合Collection。刪除Delete通過(guò)指定主鍵如id或復(fù)雜的元數(shù)據(jù)過(guò)濾條件刪除對(duì)應(yīng)的向量實(shí)體。特別注意Milvus 的刪除是邏輯刪除軟刪除數(shù)據(jù)不會(huì)立即從磁盤物理清除而是在后續(xù)的壓縮Compaction過(guò)程中清理。刪除的實(shí)體在之后的檢索中不可見(jiàn)。更新UpsertMilvus 沒(méi)有直接的update操作。upsert是“更新或插入”的合并操作。如果提供的主鍵已存在則用新數(shù)據(jù)覆蓋舊數(shù)據(jù)先刪除舊再插入新如果不存在則執(zhí)行插入。這是實(shí)現(xiàn)“修改”功能的主要方式。集合Collection與分區(qū)PartitionMilvus 中Collection類似于數(shù)據(jù)庫(kù)的表Partition是表內(nèi)的數(shù)據(jù)分區(qū)常用于數(shù)據(jù)隔離如按租戶、按類別分區(qū)。在 LangChain 中我們通常在一個(gè)Collection內(nèi)操作partition_key可作為元數(shù)據(jù)的一部分用于高效過(guò)濾。2.2 環(huán)境與工具準(zhǔn)備我們將在一個(gè)干凈的 Python 環(huán)境中進(jìn)行。請(qǐng)確保已安裝以下依賴# 創(chuàng)建并激活虛擬環(huán)境 (可選但推薦) python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安裝核心依賴 pip install langchain langchain-milvus langchain-openai pymilvus # 如果你使用其他嵌入模型如 sentence-transformers # pip install sentence-transformers版本說(shuō)明本文基于langchain-milvus0.2.0它提供了更完善的 DML 方法封裝。請(qǐng)確保你的版本足夠新。Milvus 服務(wù)你需要一個(gè)運(yùn)行中的 Milvus 實(shí)例。有以下幾種方式Docker推薦用于本地開(kāi)發(fā)# 拉取并運(yùn)行 Milvus 單機(jī)版 docker run -d --name milvus-standalone \ -p 19530:19530 -p 9091:9091 \ milvusdb/milvus:latest-standalone安裝包從 Milvus 官網(wǎng) 下載對(duì)應(yīng)系統(tǒng)的安裝包。云服務(wù)使用 Zilliz Cloud 等托管服務(wù)。本文假設(shè)你使用本地 Docker 運(yùn)行的 Milvus連接地址為localhost:19530。3. 初始化連接 Milvus 與創(chuàng)建集合一切操作始于一個(gè)穩(wěn)定的連接和一張定義好的“表”Collection。3.1 建立連接與 VectorStore 對(duì)象我們使用Milvus類來(lái)連接數(shù)據(jù)庫(kù)并創(chuàng)建一個(gè) VectorStore 對(duì)象。這里我們使用 OpenAI 的嵌入模型你需要準(zhǔn)備一個(gè)OPENAI_API_KEY。# 文件init_milvus.py import os from langchain_milvus import Milvus from langchain_openai import OpenAIEmbeddings # 設(shè)置環(huán)境變量請(qǐng)?zhí)鎿Q為你的實(shí)際密鑰 os.environ[OPENAI_API_KEY] your-openai-api-key # 1. 定義嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 使用較小、較快的模型 # 2. 連接參數(shù) connection_args { uri: http://localhost:19530, # Milvus 服務(wù)地址 # 如果 Milvus 啟用了認(rèn)證需要添加以下參數(shù) # user: username, # password: password, # token: your-token, } # 3. 初始化 Milvus VectorStore # 如果集合不存在會(huì)自動(dòng)創(chuàng)建。我們指定集合名為 langchain_demo。 vector_store Milvus( embedding_functionembeddings, collection_namelangchain_demo, connection_argsconnection_args, drop_oldFalse, # 重要如果集合已存在不要?jiǎng)h除它。設(shè)為 True 會(huì)清空現(xiàn)有數(shù)據(jù) auto_idTrue, # 讓 Milvus 自動(dòng)生成唯一主鍵 id ) print(fMilvus VectorStore 初始化成功連接到集合: {vector_store.collection_name})關(guān)鍵參數(shù)解釋collection_name: 集合名稱相當(dāng)于表名。drop_old:務(wù)必謹(jǐn)慎設(shè)置為True會(huì)在初始化時(shí)刪除已存在的同名集合并新建一個(gè)空集合導(dǎo)致數(shù)據(jù)丟失。生產(chǎn)環(huán)境通常設(shè)為False。auto_id: 設(shè)置為True時(shí)Milvus 會(huì)為每條數(shù)據(jù)自動(dòng)生成一個(gè)整數(shù)型主鍵。如果你有自己的業(yè)務(wù) ID如字符串類型的文檔ID可以設(shè)為False并在插入數(shù)據(jù)時(shí)通過(guò)ids參數(shù)指定。connection_args: 除了uri還可能包含secure是否使用 TLS、client_key_path等高級(jí)參數(shù)。運(yùn)行此腳本如果沒(méi)有報(bào)錯(cuò)說(shuō)明連接成功。你可以通過(guò) Milvus 的管理工具如 Attu來(lái)查看是否創(chuàng)建了langchain_demo集合。4. 核心 DML 操作實(shí)戰(zhàn)現(xiàn)在進(jìn)入正題。我們將模擬一個(gè)企業(yè)知識(shí)庫(kù)的常見(jiàn)維護(hù)操作。4.1 插入數(shù)據(jù)構(gòu)建初始知識(shí)庫(kù)首先我們向空集合中插入第一批文檔。# 文件insert_data.py from init_milvus import vector_store # 導(dǎo)入上一步初始化的 vector_store # 準(zhǔn)備要插入的文檔 documents [ LangChain 是一個(gè)用于開(kāi)發(fā)由大語(yǔ)言模型驅(qū)動(dòng)的應(yīng)用程序的框架。, Milvus 是一個(gè)高性能、開(kāi)源向量數(shù)據(jù)庫(kù)專為海量向量相似性搜索而設(shè)計(jì)。, 檢索增強(qiáng)生成RAG通過(guò)結(jié)合檢索器和生成模型來(lái)提升回答的準(zhǔn)確性和相關(guān)性。, Python 是一種廣泛使用的高級(jí)編程語(yǔ)言以其簡(jiǎn)潔的語(yǔ)法和強(qiáng)大的庫(kù)生態(tài)系統(tǒng)而聞名。, ] # 準(zhǔn)備對(duì)應(yīng)的元數(shù)據(jù)。元數(shù)據(jù)用于過(guò)濾和精煉檢索結(jié)果。 metadatas [ {source: framework_doc, category: langchain, version: 1.0}, {source: db_doc, category: vector_db, version: 2.0}, {source: technique_doc, category: rag, version: 1.0}, {source: language_doc, category: programming, version: 1.0}, ] # 執(zhí)行插入操作 # add_texts 是 LangChain VectorStore 接口的標(biāo)準(zhǔn)方法內(nèi)部會(huì)調(diào)用 Milvus 的 insert。 inserted_ids vector_store.add_texts(textsdocuments, metadatasmetadatas) print(f成功插入 {len(inserted_ids)} 條文檔。) print(f生成的文檔ID列表前5個(gè): {inserted_ids[:5]})執(zhí)行與驗(yàn)證 運(yùn)行腳本后會(huì)輸出插入的文檔數(shù)量。這些文檔會(huì)被OpenAIEmbeddings模型轉(zhuǎn)換為向量然后連同文本和元數(shù)據(jù)一起存儲(chǔ)到 Milvus 的langchain_demo集合中。你可以立即進(jìn)行一次檢索來(lái)驗(yàn)證# 接在插入腳本后或新建一個(gè)文件 query 什么是向量數(shù)據(jù)庫(kù) results vector_store.similarity_search(query, k2) print(f查詢: {query}) for i, doc in enumerate(results): print(f[結(jié)果 {i1}] {doc.page_content} | 元數(shù)據(jù): {doc.metadata})你應(yīng)該能看到與Milvus相關(guān)的文檔被檢索出來(lái)。4.2 刪除數(shù)據(jù)清理錯(cuò)誤或過(guò)時(shí)信息假設(shè)我們發(fā)現(xiàn)關(guān)于Python的那條文檔描述過(guò)于簡(jiǎn)單想要?jiǎng)h除它。在 Milvus 中刪除需要依據(jù)主鍵id或通過(guò)元數(shù)據(jù)表達(dá)式進(jìn)行過(guò)濾。場(chǎng)景一通過(guò)主鍵 ID 刪除在上一步的插入操作中add_texts返回了插入數(shù)據(jù)的 ID 列表。我們可以用這個(gè) ID 來(lái)刪除。# 文件delete_by_id.py from init_milvus import vector_store # 假設(shè)我們知道要?jiǎng)h除的文檔 ID 是 inserted_ids 列表中的最后一個(gè)。 # 在實(shí)際應(yīng)用中這個(gè)ID可能來(lái)自之前的操作記錄或查詢結(jié)果。 # 這里我們模擬已知ID為 444具體值需根據(jù)你實(shí)際運(yùn)行 insert_data.py 的輸出調(diào)整。 target_id 444 # 請(qǐng)?zhí)鎿Q為實(shí)際ID # 方法一使用 Milvus 提供的 delete 方法更底層更靈活 # 首先需要獲取 collection 對(duì)象 collection vector_store.collection # 構(gòu)建刪除表達(dá)式id in [target_id] expr fid in [{target_id}] # 執(zhí)行刪除 delete_result collection.delete(expr) print(f通過(guò)表達(dá)式刪除結(jié)果: {delete_result}) # 方法二使用 LangChain-Milvus 封裝的 delete 方法更簡(jiǎn)潔但可能依賴版本 # 注意langchain-milvus 的 delete 方法可能直接接收ID列表或表達(dá)式。 # 請(qǐng)查閱你所使用版本的文檔。一個(gè)常見(jiàn)的用法是 try: # 某些版本支持直接傳 ids vector_store.delete(ids[target_id]) print(f已通過(guò)ID刪除文檔: {target_id}) except Exception as e: print(f使用 vector_store.delete 時(shí)出錯(cuò)可能方法簽名不同: {e}) # 驗(yàn)證刪除再次查詢“Python” results vector_store.similarity_search(Python, k5) print(f\n刪除后查詢‘Python’結(jié)果數(shù): {len(results)}) if len(results) 0: for doc in results: print(f - {doc.page_content[:50]}...) else: print( 未找到相關(guān)文檔刪除成功。)場(chǎng)景二通過(guò)元數(shù)據(jù)過(guò)濾條件刪除更常見(jiàn)的場(chǎng)景是批量刪除例如刪除所有category為programming的舊版本文檔。# 文件delete_by_metadata.py from init_milvus import vector_store # 構(gòu)建刪除表達(dá)式。Milvus 的表達(dá)式語(yǔ)法類似于簡(jiǎn)單的 SQL WHERE 子句。 # 注意字段名需要用雙引號(hào)括起來(lái)字符串值用單引號(hào)。 delete_expr category programming and version 1.0 collection vector_store.collection delete_result collection.delete(delete_expr) print(f通過(guò)元數(shù)據(jù)表達(dá)式刪除結(jié)果: {delete_result}) print(f表達(dá)式: {delete_expr}) # 重要提醒刪除操作是異步的 # 執(zhí)行刪除后索引可能不會(huì)立即更新。在緊接著的檢索中被刪除的數(shù)據(jù)可能暫時(shí)還能被查到取決于一致性級(jí)別。 # 對(duì)于強(qiáng)一致性要求可以在刪除后執(zhí)行 flush() 并等待索引重建但這在 LangChain 日常使用中較少見(jiàn)。 # collection.flush()關(guān)于刪除的深度理解邏輯刪除Milvus 的刪除是標(biāo)記刪除并非立即物理擦除。這有利于性能但意味著存儲(chǔ)空間不會(huì)立即釋放。一致性刪除操作對(duì)后續(xù)查詢的可見(jiàn)性取決于集合的一致性級(jí)別Consistency Level。默認(rèn)級(jí)別Bounded下刪除可能不會(huì)立即在所有查詢節(jié)點(diǎn)上生效存在短暫延遲。壓縮后臺(tái)的自動(dòng)壓縮Compaction進(jìn)程會(huì)定期清理被標(biāo)記刪除的數(shù)據(jù)回收存儲(chǔ)空間。4.3 更新數(shù)據(jù)使用 Upsert 修正信息如前所述Milvus 沒(méi)有直接的update。我們需要使用upsert操作。upsert的邏輯是如果存在則替換如果不存在則插入。假設(shè)我們要更新LangChain的文檔將其版本號(hào)從1.0更新到2.0并修正內(nèi)容。# 文件upsert_data.py from init_milvus import vector_store # 1. 首先我們需要找到要更新的文檔的 ID。 # 我們可以通過(guò)元數(shù)據(jù)查詢來(lái)獲取。 search_expr source framework_doc and category langchain # 注意Milvus 的 query 接口用于根據(jù)表達(dá)式查詢數(shù)據(jù)返回完整的實(shí)體包括向量、標(biāo)量字段。 # 但 langchain-milvus 可能沒(méi)有直接暴露 query 方法。我們可以使用 collection.query。 collection vector_store.collection # 加載集合到內(nèi)存對(duì)于查詢是必要的 collection.load() # 執(zhí)行查詢指定要輸出的字段 query_results collection.query( exprsearch_expr, output_fields[id, text, source, category, version] # “text”是存儲(chǔ)原始文本的字段名需確認(rèn) ) # 注意在 langchain-milvus 中文本存儲(chǔ)的字段名默認(rèn)可能是 text 或由 text_field 參數(shù)指定。 print(查詢到的待更新文檔:) for res in query_results: print(res) old_id res[id] # 獲取主鍵ID old_text res.get(text) break # 假設(shè)只更新第一個(gè)匹配的文檔 if not query_results: print(未找到要更新的文檔將執(zhí)行插入操作。) old_id None old_text None # 2. 準(zhǔn)備新的文檔內(nèi)容和元數(shù)據(jù) new_text LangChain 是一個(gè)強(qiáng)大的框架用于構(gòu)建由大語(yǔ)言模型驅(qū)動(dòng)的應(yīng)用程序。它提供了組件化和鏈?zhǔn)秸{(diào)用的能力極大簡(jiǎn)化了LLM應(yīng)用的開(kāi)發(fā)。 new_metadata {source: framework_doc_v2, category: langchain, version: 2.0} # 3. 執(zhí)行 Upsert。 # 在 Milvus 層面upsert 需要提供完整的數(shù)據(jù)行包括主鍵。 # 如果 old_id 存在則用新數(shù)據(jù)覆蓋該ID的記錄如果為 None則插入新記錄。 # 由于 langchain-milvus 的 add_texts 在 auto_idTrue 時(shí)無(wú)法指定ID我們需要用更底層的方式。 # 這里演示使用 pymilvus 直接操作。 from pymilvus import DataType # 假設(shè)我們集合的 Schema 包含id (int64), vector (float vector), text (varchar), source, category, version 等標(biāo)量字段。 # 我們需要構(gòu)造與 Schema 匹配的數(shù)據(jù)。 data [ [old_id if old_id else 0], # ID列。如果更新用舊ID如果插入可以傳0或不傳取決于auto_id但upsert要求提供。這里用0占位實(shí)際由系統(tǒng)處理或覆蓋。 [new_text], # 文本列 [new_metadata[source]], # source列 [new_metadata[category]], # category列 [new_metadata[version]], # version列 # 注意我們?nèi)鄙傧蛄苛衭psert 需要提供完整的向量或者能根據(jù)文本生成向量。 ] # 直接使用 pymilvus 的 upsert 較為復(fù)雜需要精確對(duì)齊 Schema。 # 更實(shí)用的方法在 LangChain 層面我們可以采用“先刪后插”來(lái)模擬更新。 # 這是一個(gè)更通用、更不易出錯(cuò)的方法。 print(\n--- 采用‘先刪后插’策略模擬更新 ---) if old_id: # 步驟A刪除舊文檔 delete_expr fid in [{old_id}] collection.delete(delete_expr) print(f已刪除舊文檔 ID: {old_id}) # 步驟B插入新文檔使用 add_texts它會(huì)自動(dòng)生成向量 new_ids vector_store.add_texts(texts[new_text], metadatas[new_metadata]) print(f已插入新文檔ID: {new_ids[0]}) # 驗(yàn)證更新 print(\n驗(yàn)證更新后的查詢結(jié)果) results vector_store.similarity_search(LangChain 框架, k2) for doc in results: print(f內(nèi)容: {doc.page_content[:80]}... | 版本: {doc.metadata.get(version)})關(guān)鍵點(diǎn)upsert的復(fù)雜性直接使用 Milvus 的upsert需要手動(dòng)處理向量生成和數(shù)據(jù)對(duì)齊在 LangChain 工作流中并不方便?！跋葎h后插”策略對(duì)于 LangChain 應(yīng)用更清晰、更可控的“更新”流程是1) 根據(jù)業(yè)務(wù)邏輯定位要更新的文檔ID2) 刪除這些ID的文檔3) 使用add_texts插入新的文檔含新向量。這雖然涉及兩步操作但邏輯簡(jiǎn)單且能充分利用 LangChain 的嵌入模型自動(dòng)生成向量。版本控制在元數(shù)據(jù)中加入version、update_time等字段是管理文檔演變歷史的良好實(shí)踐。5. 在 LangChain 鏈中集成動(dòng)態(tài)數(shù)據(jù)管理DML 操作不應(yīng)該只是孤立的腳本。在真實(shí)的 RAG 應(yīng)用中它們需要被集成到業(yè)務(wù)邏輯中。以下是一個(gè)模擬的“知識(shí)庫(kù)管理后臺(tái)”的簡(jiǎn)單鏈條設(shè)計(jì)。# 文件rag_with_management.py from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI from init_milvus import vector_store, embeddings from langchain.text_splitter import RecursiveCharacterTextSplitter class KnowledgeBaseManager: def __init__(self, vector_store): self.vector_store vector_store self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrievervector_store.as_retriever(search_kwargs{k: 3}), return_source_documentsTrue ) self.text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) def add_document(self, full_text, metadata): 向知識(shí)庫(kù)添加新文檔自動(dòng)分塊 print(f正在處理文檔長(zhǎng)度: {len(full_text)} 字符) chunks self.text_splitter.split_text(full_text) metadatas [metadata.copy() for _ in chunks] # 可以為每個(gè)塊添加一個(gè) chunk_id 到元數(shù)據(jù)中便于追蹤 for i, meta in enumerate(metadatas): meta[chunk_index] i ids self.vector_store.add_texts(textschunks, metadatasmetadatas) print(f文檔已分塊并插入共 {len(ids)} 個(gè)塊。) return ids def delete_by_source(self, source_value): 根據(jù)來(lái)源刪除所有相關(guān)文檔塊 expr fsource {source_value} collection self.vector_store.collection # 注意先查詢一下有多少條確認(rèn)范圍 count_expr fcount(*) where {expr} # 這里簡(jiǎn)化處理實(shí)際應(yīng)使用 count 接口 print(f準(zhǔn)備刪除 source 為 {source_value} 的所有文檔...) try: # 執(zhí)行刪除 res collection.delete(expr) print(f刪除請(qǐng)求已提交。) # 在實(shí)際生產(chǎn)環(huán)境中這里可能需要記錄刪除操作日志或觸發(fā)后續(xù)清理任務(wù)。 except Exception as e: print(f刪除操作失敗: {e}) return False return True def query_knowledge_base(self, question): 查詢知識(shí)庫(kù)并獲取答案 result self.qa_chain.invoke({query: question}) answer result[result] sources [doc.metadata.get(source, Unknown) for doc in result[source_documents]] print(f\n問(wèn)題: {question}) print(f答案: {answer}) print(f參考來(lái)源: {list(set(sources))}) # 去重 return answer, result[source_documents] # 使用示例 if __name__ __main__: manager KnowledgeBaseManager(vector_store) # 場(chǎng)景添加一份新政策文檔 new_policy 公司最新差旅政策于2023年10月更新。主要內(nèi)容包括 1. 經(jīng)濟(jì)艙機(jī)票需提前7天預(yù)訂。 2. 住宿標(biāo)準(zhǔn)為一線城市每晚不超過(guò)800元二線城市不超過(guò)600元。 3. 所有報(bào)銷需通過(guò)財(cái)務(wù)系統(tǒng)在行程結(jié)束后30天內(nèi)提交。 manager.add_document(new_policy, {source: travel_policy_202310, doc_type: policy, effective_date: 2023-10-01}) # 場(chǎng)景查詢 manager.query_knowledge_base(公司的差旅住宿標(biāo)準(zhǔn)是多少) # 場(chǎng)景刪除舊的、過(guò)期的政策 # 假設(shè)我們知道舊政策的 source 標(biāo)識(shí)符 # manager.delete_by_source(travel_policy_202301)這個(gè)KnowledgeBaseManager類將 DML 操作封裝成了業(yè)務(wù)方法并與檢索問(wèn)答鏈結(jié)合形成了一個(gè)簡(jiǎn)易但完整的知識(shí)庫(kù)后臺(tái)管理原型。6. 運(yùn)行驗(yàn)證與效果觀察運(yùn)行上述rag_with_management.py腳本你應(yīng)該能看到以下輸出插入過(guò)程顯示文檔被分塊處理并成功插入。查詢過(guò)程LLM 基于從 Milvus 檢索到的政策文檔塊生成關(guān)于差旅標(biāo)準(zhǔn)的答案并列出參考來(lái)源 (travel_policy_202310)。刪除過(guò)程如果取消注釋會(huì)提交刪除舊政策文檔的請(qǐng)求。關(guān)鍵驗(yàn)證點(diǎn)插入后即時(shí)檢索新插入的文檔能否在接下來(lái)的查詢中被立即檢索到是的在默認(rèn)一致性級(jí)別下通?？梢?。刪除后檢索被刪除的文檔是否從后續(xù)的檢索結(jié)果中消失是的但可能有短暫延遲。元數(shù)據(jù)過(guò)濾在Retriever中可以通過(guò)search_kwargs設(shè)置filter實(shí)現(xiàn)基于元數(shù)據(jù)的檢索。例如只檢索特定doc_type或特定日期之后的文檔。這本身不是 DML但卻是利用 DML 操作如更新版本號(hào)所維護(hù)的元數(shù)據(jù)進(jìn)行精準(zhǔn)查詢的關(guān)鍵。# 在創(chuàng)建 Retriever 時(shí)添加過(guò)濾器 from langchain.vectorstores import Milvus # 假設(shè)我們只想查詢 policy 類型的文檔 retriever vector_store.as_retriever( search_kwargs{ k: 4, filter: doc_type policy # Milvus 布爾表達(dá)式 } )7. 常見(jiàn)問(wèn)題與排查思路問(wèn)題現(xiàn)象可能原因排查方式解決方案add_texts成功但查詢不到新數(shù)據(jù)1. 集合未加載Load。2. 索引未構(gòu)建或構(gòu)建中。3. 一致性級(jí)別導(dǎo)致延遲可見(jiàn)。1. 檢查collection.load()是否已調(diào)用。2. 通過(guò) Attu 或collection.index()查看索引狀態(tài)。3. 查詢后稍等片刻再試。1. 在查詢前顯式調(diào)用collection.load()。2. 確保索引構(gòu)建完成狀態(tài)為Finished。3. 對(duì)于生產(chǎn)環(huán)境理解并設(shè)置合適的consistency_level。刪除數(shù)據(jù)后仍能被檢索到1. 邏輯刪除未完成壓縮。2. 查詢時(shí)未使用強(qiáng)一致性。3. 表達(dá)式寫錯(cuò)未匹配到目標(biāo)數(shù)據(jù)。1. 檢查刪除操作的返回結(jié)果delete_count。2. 在查詢時(shí)指定consistency_levelStrong性能有損。3. 使用collection.query驗(yàn)證刪除表達(dá)式是否能查出數(shù)據(jù)。1. 理解這是邏輯刪除的正?，F(xiàn)象或手動(dòng)觸發(fā)compaction。2. 對(duì)于需要立即生效的場(chǎng)景使用強(qiáng)一致性查詢。3. 仔細(xì)檢查刪除表達(dá)式語(yǔ)法使用 Attu 工具測(cè)試。upsert或“先刪后插”后出現(xiàn)重復(fù)或數(shù)據(jù)不一致1. 刪除和插入不是原子操作中間可能有并發(fā)查詢。2. 插入時(shí)生成了新的 ID導(dǎo)致舊ID被刪新ID插入但業(yè)務(wù)邏輯上它們代表同一實(shí)體。1. 檢查是否有其他進(jìn)程同時(shí)在讀寫。2. 審查業(yè)務(wù)邏輯確保“更新”語(yǔ)義正確。是否應(yīng)該保留歷史版本1. 考慮引入分布式鎖或更細(xì)粒度的版本控制如增加is_latest標(biāo)志。2. 使用upsert并確保能穩(wěn)定生成業(yè)務(wù)主鍵如doc_id而非依賴自動(dòng)ID。嵌入模型變更導(dǎo)致檢索效果差插入和查詢使用了不同的嵌入模型向量空間不一致。確認(rèn)Milvus初始化時(shí)傳入的embedding_function始終是同一個(gè)模型實(shí)例。確保整個(gè)應(yīng)用生命周期使用相同的嵌入模型。如需切換模型通常需要重建整個(gè)向量庫(kù)。連接失敗 (Cannot connect to Milvus)1. Milvus 服務(wù)未啟動(dòng)。2. 網(wǎng)絡(luò)或端口不通。3. 認(rèn)證信息錯(cuò)誤。1. 運(yùn)行docker ps檢查容器狀態(tài)。2. 使用telnet localhost 19530測(cè)試端口。3. 檢查connection_args中的uri、token等。1. 啟動(dòng) Milvus 服務(wù)。2. 檢查防火墻和網(wǎng)絡(luò)配置。3. 核對(duì)連接參數(shù)。8. 最佳實(shí)踐與工程建議主鍵設(shè)計(jì)如果使用auto_idFalse請(qǐng)使用有業(yè)務(wù)意義的、全局唯一的字符串或數(shù)字作為主鍵如doc_id:uuid。這能極大簡(jiǎn)化后續(xù)的更新和刪除操作。元數(shù)據(jù)規(guī)劃精心設(shè)計(jì)元數(shù)據(jù) Schema。常見(jiàn)的字段包括source文檔來(lái)源、doc_id原文檔ID、chunk_index塊索引、version、created_time、updated_time、owner租戶/用戶ID等。良好的元數(shù)據(jù)是高效過(guò)濾和管理的基石。批量操作無(wú)論是插入還是刪除都應(yīng)盡量批量進(jìn)行以減少網(wǎng)絡(luò)往返和事務(wù)開(kāi)銷。add_texts本身支持批量。刪除時(shí)盡量用id in [id1, id2, ...]表達(dá)式一次性刪除多個(gè)。錯(cuò)誤處理與重試網(wǎng)絡(luò)操作和數(shù)據(jù)庫(kù)操作都可能失敗。在生產(chǎn)代碼中務(wù)必對(duì)add_texts、delete等操作添加重試邏輯和異常捕獲。監(jiān)控與日志記錄重要的 DML 操作尤其是刪除和批量更新的詳細(xì)信息包括操作者、時(shí)間、影響的數(shù)據(jù)ID或條件。這對(duì)于數(shù)據(jù)審計(jì)和問(wèn)題排查至關(guān)重要。測(cè)試環(huán)境先行任何刪除和更新腳本都應(yīng)在測(cè)試環(huán)境充分驗(yàn)證后再上生產(chǎn)?？梢钥紤]在刪除前先執(zhí)行一次query預(yù)覽將要影響的數(shù)據(jù)。理解一致性權(quán)衡Milvus 提供了不同的一致性級(jí)別。在追求極高查詢性能的場(chǎng)景可能接受最終一致性在要求數(shù)據(jù)強(qiáng)一致性的金融或合規(guī)場(chǎng)景則需要選擇Strong一致性并承受相應(yīng)的性能損耗。版本化與歷史數(shù)據(jù)對(duì)于重要的知識(shí)文檔單純的“覆蓋式”更新可能不夠。可以考慮版本化方案插入新版本文檔并通過(guò)元數(shù)據(jù)如is_latestTrue標(biāo)記當(dāng)前生效版本而不是物理刪除舊版本。這保留了歷史記錄并支持回滾。掌握 LangChain 與 Milvus 的 DML 操作意味著你能夠構(gòu)建的 RAG 應(yīng)用從靜態(tài)的“只讀”演示進(jìn)化成了動(dòng)態(tài)的、可維護(hù)的、貼近真實(shí)業(yè)務(wù)需求的系統(tǒng)。數(shù)據(jù)不再是初始化后一成不變的背景板而是可以隨著業(yè)務(wù)流動(dòng)和演化的核心資產(chǎn)。從插入第一份文檔到精準(zhǔn)刪除一條錯(cuò)誤記錄再到平滑更新整個(gè)知識(shí)體系每一步操作都關(guān)乎著最終應(yīng)用的可信度和智能水平。 LangChain與Milvus向量數(shù)據(jù)庫(kù)DML實(shí)戰(zhàn)：構(gòu)建動(dòng)態(tài)可維護(hù)的RAG應(yīng)用
如果你正在構(gòu)建一個(gè)基于大語(yǔ)言模型LLM的智能應(yīng)用比如一個(gè)能回答專業(yè)問(wèn)題的知識(shí)庫(kù)助手你很可能遇到過(guò)這個(gè)核心難題如何讓模型“記住”并快速“找到”海量的私有文檔信息直接讓模型去“閱讀”成千上萬(wàn)份PDF、Word文檔是不現(xiàn)實(shí)的這會(huì)導(dǎo)致響應(yīng)極慢、成本高昂且容易超出模型的上下文長(zhǎng)度限制。主流的解決方案是“檢索增強(qiáng)生成”RAG其核心在于一個(gè)高效的向量數(shù)據(jù)庫(kù)。而 Milvus正是這個(gè)領(lǐng)域備受矚目的開(kāi)源明星。但很多開(kāi)發(fā)者止步于“Hello World”式的示例把文檔灌進(jìn)去問(wèn)個(gè)簡(jiǎn)單問(wèn)題得到答案就覺(jué)得大功告成。然而真實(shí)的生產(chǎn)環(huán)境是動(dòng)態(tài)的知識(shí)需要更新錯(cuò)誤的文檔需要?jiǎng)h除過(guò)時(shí)的信息需要修正。這時(shí)你就會(huì)遇到本文要解決的核心問(wèn)題如何對(duì) Milvus 向量數(shù)據(jù)庫(kù)中的數(shù)據(jù)進(jìn)行可靠的增、刪、改DML操作并確保你的 LangChain RAG 應(yīng)用能隨之動(dòng)態(tài)、準(zhǔn)確地響應(yīng)本文將帶你超越基礎(chǔ)教程深入langchain-milvus集成的 DML 實(shí)戰(zhàn)。我們不僅會(huì)演示如何插入、刪除和更新數(shù)據(jù)更會(huì)剖析這些操作背后的原理、在 LangChain 鏈中的最佳集成位置以及你必須避開(kāi)的那些“坑”——比如刪除操作的實(shí)際行為、元數(shù)據(jù)過(guò)濾的陷阱以及如何保證數(shù)據(jù)變更后檢索的實(shí)時(shí)性。讀完本文你將能構(gòu)建一個(gè)真正可維護(hù)、可演化的 RAG 應(yīng)用而不僅僅是一個(gè)靜態(tài)的演示原型。1. 為什么 Milvus 的 DML 操作是 RAG 應(yīng)用的“生命線”在原型階段我們往往只關(guān)心“寫入”和“查詢”。但一旦應(yīng)用上線數(shù)據(jù)就活了。想象以下幾個(gè)場(chǎng)景知識(shí)更新公司發(fā)布了新的產(chǎn)品手冊(cè)你需要將新文檔的向量添加到數(shù)據(jù)庫(kù)中而不是重建整個(gè)庫(kù)。錯(cuò)誤修正發(fā)現(xiàn)之前導(dǎo)入的一份技術(shù)規(guī)格表有錯(cuò)誤你需要定位并刪除或更新對(duì)應(yīng)的錯(cuò)誤信息片段。數(shù)據(jù)隔離你的系統(tǒng)服務(wù)于多個(gè)客戶多租戶需要能安全地刪除某個(gè)客戶的全部數(shù)據(jù)。合規(guī)要求根據(jù)用戶請(qǐng)求需要從知識(shí)庫(kù)中徹底抹去某些敏感信息。如果缺乏可靠的 DML數(shù)據(jù)操作語(yǔ)言支持上述每一個(gè)場(chǎng)景都會(huì)讓你陷入困境要么全量重建向量庫(kù)耗時(shí)耗力要么容忍數(shù)據(jù)的不一致和錯(cuò)誤。Milvus 的核心價(jià)值之一就是它提供了完整、高效的 DML 接口insert,delete,upsert并且與 LangChain 的VectorStore抽象層進(jìn)行了深度集成。這使得在 LangChain 應(yīng)用中進(jìn)行數(shù)據(jù)維護(hù)變得像操作傳統(tǒng)數(shù)據(jù)庫(kù)一樣直觀但背后卻涉及向量索引的復(fù)雜管理。本文將聚焦于langchain-milvus這個(gè)官方集成包通過(guò)實(shí)戰(zhàn)代碼讓你掌握這條“生命線”的操控方法。2. 核心概念與工具準(zhǔn)備在開(kāi)始實(shí)戰(zhàn)前我們需要統(tǒng)一幾個(gè)關(guān)鍵概念并準(zhǔn)備好戰(zhàn)場(chǎng)。2.1 關(guān)鍵概念澄清Milvus一個(gè)高性能、云原生的開(kāi)源向量數(shù)據(jù)庫(kù)專為海量向量數(shù)據(jù)的存儲(chǔ)、索引和檢索而設(shè)計(jì)。它支持標(biāo)量過(guò)濾、動(dòng)態(tài) Schema、數(shù)據(jù)分片等高級(jí)特性。LangChain一個(gè)用于構(gòu)建由 LLM 驅(qū)動(dòng)的應(yīng)用程序的框架。它通過(guò)“鏈”Chain將模型、提示、記憶、檢索器等組件連接起來(lái)。其VectorStore模塊提供了與各種向量數(shù)據(jù)庫(kù)交互的統(tǒng)一接口。langchain-milvus這是 LangChain 官方維護(hù)的 Milvus 集成包。它實(shí)現(xiàn)了VectorStore接口封裝了 Milvus Python SDK 的復(fù)雜調(diào)用讓我們可以用更 LangChain 風(fēng)格的方式操作 Milvus。DML in Vector DB與傳統(tǒng)數(shù)據(jù)庫(kù)的增刪改查類似但在向量數(shù)據(jù)庫(kù)中插入Insert將文本經(jīng)過(guò)嵌入模型Embedding Model轉(zhuǎn)化為向量后連同元數(shù)據(jù)metadata一起存入集合Collection。刪除Delete通過(guò)指定主鍵如id或復(fù)雜的元數(shù)據(jù)過(guò)濾條件刪除對(duì)應(yīng)的向量實(shí)體。特別注意Milvus 的刪除是邏輯刪除軟刪除數(shù)據(jù)不會(huì)立即從磁盤物理清除而是在后續(xù)的壓縮Compaction過(guò)程中清理。刪除的實(shí)體在之后的檢索中不可見(jiàn)。更新UpsertMilvus 沒(méi)有直接的update操作。upsert是“更新或插入”的合并操作。如果提供的主鍵已存在則用新數(shù)據(jù)覆蓋舊數(shù)據(jù)先刪除舊再插入新如果不存在則執(zhí)行插入。這是實(shí)現(xiàn)“修改”功能的主要方式。集合Collection與分區(qū)PartitionMilvus 中Collection類似于數(shù)據(jù)庫(kù)的表Partition是表內(nèi)的數(shù)據(jù)分區(qū)常用于數(shù)據(jù)隔離如按租戶、按類別分區(qū)。在 LangChain 中我們通常在一個(gè)Collection內(nèi)操作partition_key可作為元數(shù)據(jù)的一部分用于高效過(guò)濾。2.2 環(huán)境與工具準(zhǔn)備我們將在一個(gè)干凈的 Python 環(huán)境中進(jìn)行。請(qǐng)確保已安裝以下依賴# 創(chuàng)建并激活虛擬環(huán)境 (可選但推薦) python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安裝核心依賴 pip install langchain langchain-milvus langchain-openai pymilvus # 如果你使用其他嵌入模型如 sentence-transformers # pip install sentence-transformers版本說(shuō)明本文基于langchain-milvus0.2.0它提供了更完善的 DML 方法封裝。請(qǐng)確保你的版本足夠新。Milvus 服務(wù)你需要一個(gè)運(yùn)行中的 Milvus 實(shí)例。有以下幾種方式Docker推薦用于本地開(kāi)發(fā)# 拉取并運(yùn)行 Milvus 單機(jī)版 docker run -d --name milvus-standalone \ -p 19530:19530 -p 9091:9091 \ milvusdb/milvus:latest-standalone安裝包從 Milvus 官網(wǎng) 下載對(duì)應(yīng)系統(tǒng)的安裝包。云服務(wù)使用 Zilliz Cloud 等托管服務(wù)。本文假設(shè)你使用本地 Docker 運(yùn)行的 Milvus連接地址為localhost:19530。3. 初始化連接 Milvus 與創(chuàng)建集合一切操作始于一個(gè)穩(wěn)定的連接和一張定義好的“表”Collection。3.1 建立連接與 VectorStore 對(duì)象我們使用Milvus類來(lái)連接數(shù)據(jù)庫(kù)并創(chuàng)建一個(gè) VectorStore 對(duì)象。這里我們使用 OpenAI 的嵌入模型你需要準(zhǔn)備一個(gè)OPENAI_API_KEY。# 文件init_milvus.py import os from langchain_milvus import Milvus from langchain_openai import OpenAIEmbeddings # 設(shè)置環(huán)境變量請(qǐng)?zhí)鎿Q為你的實(shí)際密鑰 os.environ[OPENAI_API_KEY] your-openai-api-key # 1. 定義嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 使用較小、較快的模型 # 2. 連接參數(shù) connection_args { uri: http://localhost:19530, # Milvus 服務(wù)地址 # 如果 Milvus 啟用了認(rèn)證需要添加以下參數(shù) # user: username, # password: password, # token: your-token, } # 3. 初始化 Milvus VectorStore # 如果集合不存在會(huì)自動(dòng)創(chuàng)建。我們指定集合名為 langchain_demo。 vector_store Milvus( embedding_functionembeddings, collection_namelangchain_demo, connection_argsconnection_args, drop_oldFalse, # 重要如果集合已存在不要?jiǎng)h除它。設(shè)為 True 會(huì)清空現(xiàn)有數(shù)據(jù) auto_idTrue, # 讓 Milvus 自動(dòng)生成唯一主鍵 id ) print(fMilvus VectorStore 初始化成功連接到集合: {vector_store.collection_name})關(guān)鍵參數(shù)解釋collection_name: 集合名稱相當(dāng)于表名。drop_old:務(wù)必謹(jǐn)慎設(shè)置為True會(huì)在初始化時(shí)刪除已存在的同名集合并新建一個(gè)空集合導(dǎo)致數(shù)據(jù)丟失。生產(chǎn)環(huán)境通常設(shè)為False。auto_id: 設(shè)置為True時(shí)Milvus 會(huì)為每條數(shù)據(jù)自動(dòng)生成一個(gè)整數(shù)型主鍵。如果你有自己的業(yè)務(wù) ID如字符串類型的文檔ID可以設(shè)為False并在插入數(shù)據(jù)時(shí)通過(guò)ids參數(shù)指定。connection_args: 除了uri還可能包含secure是否使用 TLS、client_key_path等高級(jí)參數(shù)。運(yùn)行此腳本如果沒(méi)有報(bào)錯(cuò)說(shuō)明連接成功。你可以通過(guò) Milvus 的管理工具如 Attu來(lái)查看是否創(chuàng)建了langchain_demo集合。4. 核心 DML 操作實(shí)戰(zhàn)現(xiàn)在進(jìn)入正題。我們將模擬一個(gè)企業(yè)知識(shí)庫(kù)的常見(jiàn)維護(hù)操作。4.1 插入數(shù)據(jù)構(gòu)建初始知識(shí)庫(kù)首先我們向空集合中插入第一批文檔。# 文件insert_data.py from init_milvus import vector_store # 導(dǎo)入上一步初始化的 vector_store # 準(zhǔn)備要插入的文檔 documents [ LangChain 是一個(gè)用于開(kāi)發(fā)由大語(yǔ)言模型驅(qū)動(dòng)的應(yīng)用程序的框架。, Milvus 是一個(gè)高性能、開(kāi)源向量數(shù)據(jù)庫(kù)專為海量向量相似性搜索而設(shè)計(jì)。, 檢索增強(qiáng)生成RAG通過(guò)結(jié)合檢索器和生成模型來(lái)提升回答的準(zhǔn)確性和相關(guān)性。, Python 是一種廣泛使用的高級(jí)編程語(yǔ)言以其簡(jiǎn)潔的語(yǔ)法和強(qiáng)大的庫(kù)生態(tài)系統(tǒng)而聞名。, ] # 準(zhǔn)備對(duì)應(yīng)的元數(shù)據(jù)。元數(shù)據(jù)用于過(guò)濾和精煉檢索結(jié)果。 metadatas [ {source: framework_doc, category: langchain, version: 1.0}, {source: db_doc, category: vector_db, version: 2.0}, {source: technique_doc, category: rag, version: 1.0}, {source: language_doc, category: programming, version: 1.0}, ] # 執(zhí)行插入操作 # add_texts 是 LangChain VectorStore 接口的標(biāo)準(zhǔn)方法內(nèi)部會(huì)調(diào)用 Milvus 的 insert。 inserted_ids vector_store.add_texts(textsdocuments, metadatasmetadatas) print(f成功插入 {len(inserted_ids)} 條文檔。) print(f生成的文檔ID列表前5個(gè): {inserted_ids[:5]})執(zhí)行與驗(yàn)證 運(yùn)行腳本后會(huì)輸出插入的文檔數(shù)量。這些文檔會(huì)被OpenAIEmbeddings模型轉(zhuǎn)換為向量然后連同文本和元數(shù)據(jù)一起存儲(chǔ)到 Milvus 的langchain_demo集合中。你可以立即進(jìn)行一次檢索來(lái)驗(yàn)證# 接在插入腳本后或新建一個(gè)文件 query 什么是向量數(shù)據(jù)庫(kù) results vector_store.similarity_search(query, k2) print(f查詢: {query}) for i, doc in enumerate(results): print(f[結(jié)果 {i1}] {doc.page_content} | 元數(shù)據(jù): {doc.metadata})你應(yīng)該能看到與Milvus相關(guān)的文檔被檢索出來(lái)。4.2 刪除數(shù)據(jù)清理錯(cuò)誤或過(guò)時(shí)信息假設(shè)我們發(fā)現(xiàn)關(guān)于Python的那條文檔描述過(guò)于簡(jiǎn)單想要?jiǎng)h除它。在 Milvus 中刪除需要依據(jù)主鍵id或通過(guò)元數(shù)據(jù)表達(dá)式進(jìn)行過(guò)濾。場(chǎng)景一通過(guò)主鍵 ID 刪除在上一步的插入操作中add_texts返回了插入數(shù)據(jù)的 ID 列表。我們可以用這個(gè) ID 來(lái)刪除。# 文件delete_by_id.py from init_milvus import vector_store # 假設(shè)我們知道要?jiǎng)h除的文檔 ID 是 inserted_ids 列表中的最后一個(gè)。 # 在實(shí)際應(yīng)用中這個(gè)ID可能來(lái)自之前的操作記錄或查詢結(jié)果。 # 這里我們模擬已知ID為 444具體值需根據(jù)你實(shí)際運(yùn)行 insert_data.py 的輸出調(diào)整。 target_id 444 # 請(qǐng)?zhí)鎿Q為實(shí)際ID # 方法一使用 Milvus 提供的 delete 方法更底層更靈活 # 首先需要獲取 collection 對(duì)象 collection vector_store.collection # 構(gòu)建刪除表達(dá)式id in [target_id] expr fid in [{target_id}] # 執(zhí)行刪除 delete_result collection.delete(expr) print(f通過(guò)表達(dá)式刪除結(jié)果: {delete_result}) # 方法二使用 LangChain-Milvus 封裝的 delete 方法更簡(jiǎn)潔但可能依賴版本 # 注意langchain-milvus 的 delete 方法可能直接接收ID列表或表達(dá)式。 # 請(qǐng)查閱你所使用版本的文檔。一個(gè)常見(jiàn)的用法是 try: # 某些版本支持直接傳 ids vector_store.delete(ids[target_id]) print(f已通過(guò)ID刪除文檔: {target_id}) except Exception as e: print(f使用 vector_store.delete 時(shí)出錯(cuò)可能方法簽名不同: {e}) # 驗(yàn)證刪除再次查詢“Python” results vector_store.similarity_search(Python, k5) print(f\n刪除后查詢‘Python’結(jié)果數(shù): {len(results)}) if len(results) 0: for doc in results: print(f - {doc.page_content[:50]}...) else: print( 未找到相關(guān)文檔刪除成功。)場(chǎng)景二通過(guò)元數(shù)據(jù)過(guò)濾條件刪除更常見(jiàn)的場(chǎng)景是批量刪除例如刪除所有category為programming的舊版本文檔。# 文件delete_by_metadata.py from init_milvus import vector_store # 構(gòu)建刪除表達(dá)式。Milvus 的表達(dá)式語(yǔ)法類似于簡(jiǎn)單的 SQL WHERE 子句。 # 注意字段名需要用雙引號(hào)括起來(lái)字符串值用單引號(hào)。 delete_expr category programming and version 1.0 collection vector_store.collection delete_result collection.delete(delete_expr) print(f通過(guò)元數(shù)據(jù)表達(dá)式刪除結(jié)果: {delete_result}) print(f表達(dá)式: {delete_expr}) # 重要提醒刪除操作是異步的 # 執(zhí)行刪除后索引可能不會(huì)立即更新。在緊接著的檢索中被刪除的數(shù)據(jù)可能暫時(shí)還能被查到取決于一致性級(jí)別。 # 對(duì)于強(qiáng)一致性要求可以在刪除后執(zhí)行 flush() 并等待索引重建但這在 LangChain 日常使用中較少見(jiàn)。 # collection.flush()關(guān)于刪除的深度理解邏輯刪除Milvus 的刪除是標(biāo)記刪除并非立即物理擦除。這有利于性能但意味著存儲(chǔ)空間不會(huì)立即釋放。一致性刪除操作對(duì)后續(xù)查詢的可見(jiàn)性取決于集合的一致性級(jí)別Consistency Level。默認(rèn)級(jí)別Bounded下刪除可能不會(huì)立即在所有查詢節(jié)點(diǎn)上生效存在短暫延遲。壓縮后臺(tái)的自動(dòng)壓縮Compaction進(jìn)程會(huì)定期清理被標(biāo)記刪除的數(shù)據(jù)回收存儲(chǔ)空間。4.3 更新數(shù)據(jù)使用 Upsert 修正信息如前所述Milvus 沒(méi)有直接的update。我們需要使用upsert操作。upsert的邏輯是如果存在則替換如果不存在則插入。假設(shè)我們要更新LangChain的文檔將其版本號(hào)從1.0更新到2.0并修正內(nèi)容。# 文件upsert_data.py from init_milvus import vector_store # 1. 首先我們需要找到要更新的文檔的 ID。 # 我們可以通過(guò)元數(shù)據(jù)查詢來(lái)獲取。 search_expr source framework_doc and category langchain # 注意Milvus 的 query 接口用于根據(jù)表達(dá)式查詢數(shù)據(jù)返回完整的實(shí)體包括向量、標(biāo)量字段。 # 但 langchain-milvus 可能沒(méi)有直接暴露 query 方法。我們可以使用 collection.query。 collection vector_store.collection # 加載集合到內(nèi)存對(duì)于查詢是必要的 collection.load() # 執(zhí)行查詢指定要輸出的字段 query_results collection.query( exprsearch_expr, output_fields[id, text, source, category, version] # “text”是存儲(chǔ)原始文本的字段名需確認(rèn) ) # 注意在 langchain-milvus 中文本存儲(chǔ)的字段名默認(rèn)可能是 text 或由 text_field 參數(shù)指定。 print(查詢到的待更新文檔:) for res in query_results: print(res) old_id res[id] # 獲取主鍵ID old_text res.get(text) break # 假設(shè)只更新第一個(gè)匹配的文檔 if not query_results: print(未找到要更新的文檔將執(zhí)行插入操作。) old_id None old_text None # 2. 準(zhǔn)備新的文檔內(nèi)容和元數(shù)據(jù) new_text LangChain 是一個(gè)強(qiáng)大的框架用于構(gòu)建由大語(yǔ)言模型驅(qū)動(dòng)的應(yīng)用程序。它提供了組件化和鏈?zhǔn)秸{(diào)用的能力極大簡(jiǎn)化了LLM應(yīng)用的開(kāi)發(fā)。 new_metadata {source: framework_doc_v2, category: langchain, version: 2.0} # 3. 執(zhí)行 Upsert。 # 在 Milvus 層面upsert 需要提供完整的數(shù)據(jù)行包括主鍵。 # 如果 old_id 存在則用新數(shù)據(jù)覆蓋該ID的記錄如果為 None則插入新記錄。 # 由于 langchain-milvus 的 add_texts 在 auto_idTrue 時(shí)無(wú)法指定ID我們需要用更底層的方式。 # 這里演示使用 pymilvus 直接操作。 from pymilvus import DataType # 假設(shè)我們集合的 Schema 包含id (int64), vector (float vector), text (varchar), source, category, version 等標(biāo)量字段。 # 我們需要構(gòu)造與 Schema 匹配的數(shù)據(jù)。 data [ [old_id if old_id else 0], # ID列。如果更新用舊ID如果插入可以傳0或不傳取決于auto_id但upsert要求提供。這里用0占位實(shí)際由系統(tǒng)處理或覆蓋。 [new_text], # 文本列 [new_metadata[source]], # source列 [new_metadata[category]], # category列 [new_metadata[version]], # version列 # 注意我們?nèi)鄙傧蛄苛衭psert 需要提供完整的向量或者能根據(jù)文本生成向量。 ] # 直接使用 pymilvus 的 upsert 較為復(fù)雜需要精確對(duì)齊 Schema。 # 更實(shí)用的方法在 LangChain 層面我們可以采用“先刪后插”來(lái)模擬更新。 # 這是一個(gè)更通用、更不易出錯(cuò)的方法。 print(\n--- 采用‘先刪后插’策略模擬更新 ---) if old_id: # 步驟A刪除舊文檔 delete_expr fid in [{old_id}] collection.delete(delete_expr) print(f已刪除舊文檔 ID: {old_id}) # 步驟B插入新文檔使用 add_texts它會(huì)自動(dòng)生成向量 new_ids vector_store.add_texts(texts[new_text], metadatas[new_metadata]) print(f已插入新文檔ID: {new_ids[0]}) # 驗(yàn)證更新 print(\n驗(yàn)證更新后的查詢結(jié)果) results vector_store.similarity_search(LangChain 框架, k2) for doc in results: print(f內(nèi)容: {doc.page_content[:80]}... | 版本: {doc.metadata.get(version)})關(guān)鍵點(diǎn)upsert的復(fù)雜性直接使用 Milvus 的upsert需要手動(dòng)處理向量生成和數(shù)據(jù)對(duì)齊在 LangChain 工作流中并不方便?！跋葎h后插”策略對(duì)于 LangChain 應(yīng)用更清晰、更可控的“更新”流程是1) 根據(jù)業(yè)務(wù)邏輯定位要更新的文檔ID2) 刪除這些ID的文檔3) 使用add_texts插入新的文檔含新向量。這雖然涉及兩步操作但邏輯簡(jiǎn)單且能充分利用 LangChain 的嵌入模型自動(dòng)生成向量。版本控制在元數(shù)據(jù)中加入version、update_time等字段是管理文檔演變歷史的良好實(shí)踐。5. 在 LangChain 鏈中集成動(dòng)態(tài)數(shù)據(jù)管理DML 操作不應(yīng)該只是孤立的腳本。在真實(shí)的 RAG 應(yīng)用中它們需要被集成到業(yè)務(wù)邏輯中。以下是一個(gè)模擬的“知識(shí)庫(kù)管理后臺(tái)”的簡(jiǎn)單鏈條設(shè)計(jì)。# 文件rag_with_management.py from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI from init_milvus import vector_store, embeddings from langchain.text_splitter import RecursiveCharacterTextSplitter class KnowledgeBaseManager: def __init__(self, vector_store): self.vector_store vector_store self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrievervector_store.as_retriever(search_kwargs{k: 3}), return_source_documentsTrue ) self.text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) def add_document(self, full_text, metadata): 向知識(shí)庫(kù)添加新文檔自動(dòng)分塊 print(f正在處理文檔長(zhǎng)度: {len(full_text)} 字符) chunks self.text_splitter.split_text(full_text) metadatas [metadata.copy() for _ in chunks] # 可以為每個(gè)塊添加一個(gè) chunk_id 到元數(shù)據(jù)中便于追蹤 for i, meta in enumerate(metadatas): meta[chunk_index] i ids self.vector_store.add_texts(textschunks, metadatasmetadatas) print(f文檔已分塊并插入共 {len(ids)} 個(gè)塊。) return ids def delete_by_source(self, source_value): 根據(jù)來(lái)源刪除所有相關(guān)文檔塊 expr fsource {source_value} collection self.vector_store.collection # 注意先查詢一下有多少條確認(rèn)范圍 count_expr fcount(*) where {expr} # 這里簡(jiǎn)化處理實(shí)際應(yīng)使用 count 接口 print(f準(zhǔn)備刪除 source 為 {source_value} 的所有文檔...) try: # 執(zhí)行刪除 res collection.delete(expr) print(f刪除請(qǐng)求已提交。) # 在實(shí)際生產(chǎn)環(huán)境中這里可能需要記錄刪除操作日志或觸發(fā)后續(xù)清理任務(wù)。 except Exception as e: print(f刪除操作失敗: {e}) return False return True def query_knowledge_base(self, question): 查詢知識(shí)庫(kù)并獲取答案 result self.qa_chain.invoke({query: question}) answer result[result] sources [doc.metadata.get(source, Unknown) for doc in result[source_documents]] print(f\n問(wèn)題: {question}) print(f答案: {answer}) print(f參考來(lái)源: {list(set(sources))}) # 去重 return answer, result[source_documents] # 使用示例 if __name__ __main__: manager KnowledgeBaseManager(vector_store) # 場(chǎng)景添加一份新政策文檔 new_policy 公司最新差旅政策于2023年10月更新。主要內(nèi)容包括 1. 經(jīng)濟(jì)艙機(jī)票需提前7天預(yù)訂。 2. 住宿標(biāo)準(zhǔn)為一線城市每晚不超過(guò)800元二線城市不超過(guò)600元。 3. 所有報(bào)銷需通過(guò)財(cái)務(wù)系統(tǒng)在行程結(jié)束后30天內(nèi)提交。 manager.add_document(new_policy, {source: travel_policy_202310, doc_type: policy, effective_date: 2023-10-01}) # 場(chǎng)景查詢 manager.query_knowledge_base(公司的差旅住宿標(biāo)準(zhǔn)是多少) # 場(chǎng)景刪除舊的、過(guò)期的政策 # 假設(shè)我們知道舊政策的 source 標(biāo)識(shí)符 # manager.delete_by_source(travel_policy_202301)這個(gè)KnowledgeBaseManager類將 DML 操作封裝成了業(yè)務(wù)方法并與檢索問(wèn)答鏈結(jié)合形成了一個(gè)簡(jiǎn)易但完整的知識(shí)庫(kù)后臺(tái)管理原型。6. 運(yùn)行驗(yàn)證與效果觀察運(yùn)行上述rag_with_management.py腳本你應(yīng)該能看到以下輸出插入過(guò)程顯示文檔被分塊處理并成功插入。查詢過(guò)程LLM 基于從 Milvus 檢索到的政策文檔塊生成關(guān)于差旅標(biāo)準(zhǔn)的答案并列出參考來(lái)源 (travel_policy_202310)。刪除過(guò)程如果取消注釋會(huì)提交刪除舊政策文檔的請(qǐng)求。關(guān)鍵驗(yàn)證點(diǎn)插入后即時(shí)檢索新插入的文檔能否在接下來(lái)的查詢中被立即檢索到是的在默認(rèn)一致性級(jí)別下通?？梢?。刪除后檢索被刪除的文檔是否從后續(xù)的檢索結(jié)果中消失是的但可能有短暫延遲。元數(shù)據(jù)過(guò)濾在Retriever中可以通過(guò)search_kwargs設(shè)置filter實(shí)現(xiàn)基于元數(shù)據(jù)的檢索。例如只檢索特定doc_type或特定日期之后的文檔。這本身不是 DML但卻是利用 DML 操作如更新版本號(hào)所維護(hù)的元數(shù)據(jù)進(jìn)行精準(zhǔn)查詢的關(guān)鍵。# 在創(chuàng)建 Retriever 時(shí)添加過(guò)濾器 from langchain.vectorstores import Milvus # 假設(shè)我們只想查詢 policy 類型的文檔 retriever vector_store.as_retriever( search_kwargs{ k: 4, filter: doc_type policy # Milvus 布爾表達(dá)式 } )7. 常見(jiàn)問(wèn)題與排查思路問(wèn)題現(xiàn)象可能原因排查方式解決方案add_texts成功但查詢不到新數(shù)據(jù)1. 集合未加載Load。2. 索引未構(gòu)建或構(gòu)建中。3. 一致性級(jí)別導(dǎo)致延遲可見(jiàn)。1. 檢查collection.load()是否已調(diào)用。2. 通過(guò) Attu 或collection.index()查看索引狀態(tài)。3. 查詢后稍等片刻再試。1. 在查詢前顯式調(diào)用collection.load()。2. 確保索引構(gòu)建完成狀態(tài)為Finished。3. 對(duì)于生產(chǎn)環(huán)境理解并設(shè)置合適的consistency_level。刪除數(shù)據(jù)后仍能被檢索到1. 邏輯刪除未完成壓縮。2. 查詢時(shí)未使用強(qiáng)一致性。3. 表達(dá)式寫錯(cuò)未匹配到目標(biāo)數(shù)據(jù)。1. 檢查刪除操作的返回結(jié)果delete_count。2. 在查詢時(shí)指定consistency_levelStrong性能有損。3. 使用collection.query驗(yàn)證刪除表達(dá)式是否能查出數(shù)據(jù)。1. 理解這是邏輯刪除的正?，F(xiàn)象或手動(dòng)觸發(fā)compaction。2. 對(duì)于需要立即生效的場(chǎng)景使用強(qiáng)一致性查詢。3. 仔細(xì)檢查刪除表達(dá)式語(yǔ)法使用 Attu 工具測(cè)試。upsert或“先刪后插”后出現(xiàn)重復(fù)或數(shù)據(jù)不一致1. 刪除和插入不是原子操作中間可能有并發(fā)查詢。2. 插入時(shí)生成了新的 ID導(dǎo)致舊ID被刪新ID插入但業(yè)務(wù)邏輯上它們代表同一實(shí)體。1. 檢查是否有其他進(jìn)程同時(shí)在讀寫。2. 審查業(yè)務(wù)邏輯確保“更新”語(yǔ)義正確。是否應(yīng)該保留歷史版本1. 考慮引入分布式鎖或更細(xì)粒度的版本控制如增加is_latest標(biāo)志。2. 使用upsert并確保能穩(wěn)定生成業(yè)務(wù)主鍵如doc_id而非依賴自動(dòng)ID。嵌入模型變更導(dǎo)致檢索效果差插入和查詢使用了不同的嵌入模型向量空間不一致。確認(rèn)Milvus初始化時(shí)傳入的embedding_function始終是同一個(gè)模型實(shí)例。確保整個(gè)應(yīng)用生命周期使用相同的嵌入模型。如需切換模型通常需要重建整個(gè)向量庫(kù)。連接失敗 (Cannot connect to Milvus)1. Milvus 服務(wù)未啟動(dòng)。2. 網(wǎng)絡(luò)或端口不通。3. 認(rèn)證信息錯(cuò)誤。1. 運(yùn)行docker ps檢查容器狀態(tài)。2. 使用telnet localhost 19530測(cè)試端口。3. 檢查connection_args中的uri、token等。1. 啟動(dòng) Milvus 服務(wù)。2. 檢查防火墻和網(wǎng)絡(luò)配置。3. 核對(duì)連接參數(shù)。8. 最佳實(shí)踐與工程建議主鍵設(shè)計(jì)如果使用auto_idFalse請(qǐng)使用有業(yè)務(wù)意義的、全局唯一的字符串或數(shù)字作為主鍵如doc_id:uuid。這能極大簡(jiǎn)化后續(xù)的更新和刪除操作。元數(shù)據(jù)規(guī)劃精心設(shè)計(jì)元數(shù)據(jù) Schema。常見(jiàn)的字段包括source文檔來(lái)源、doc_id原文檔ID、chunk_index塊索引、version、created_time、updated_time、owner租戶/用戶ID等。良好的元數(shù)據(jù)是高效過(guò)濾和管理的基石。批量操作無(wú)論是插入還是刪除都應(yīng)盡量批量進(jìn)行以減少網(wǎng)絡(luò)往返和事務(wù)開(kāi)銷。add_texts本身支持批量。刪除時(shí)盡量用id in [id1, id2, ...]表達(dá)式一次性刪除多個(gè)。錯(cuò)誤處理與重試網(wǎng)絡(luò)操作和數(shù)據(jù)庫(kù)操作都可能失敗。在生產(chǎn)代碼中務(wù)必對(duì)add_texts、delete等操作添加重試邏輯和異常捕獲。監(jiān)控與日志記錄重要的 DML 操作尤其是刪除和批量更新的詳細(xì)信息包括操作者、時(shí)間、影響的數(shù)據(jù)ID或條件。這對(duì)于數(shù)據(jù)審計(jì)和問(wèn)題排查至關(guān)重要。測(cè)試環(huán)境先行任何刪除和更新腳本都應(yīng)在測(cè)試環(huán)境充分驗(yàn)證后再上生產(chǎn)?？梢钥紤]在刪除前先執(zhí)行一次query預(yù)覽將要影響的數(shù)據(jù)。理解一致性權(quán)衡Milvus 提供了不同的一致性級(jí)別。在追求極高查詢性能的場(chǎng)景可能接受最終一致性在要求數(shù)據(jù)強(qiáng)一致性的金融或合規(guī)場(chǎng)景則需要選擇Strong一致性并承受相應(yīng)的性能損耗。版本化與歷史數(shù)據(jù)對(duì)于重要的知識(shí)文檔單純的“覆蓋式”更新可能不夠。可以考慮版本化方案插入新版本文檔并通過(guò)元數(shù)據(jù)如is_latestTrue標(biāo)記當(dāng)前生效版本而不是物理刪除舊版本。這保留了歷史記錄并支持回滾。掌握 LangChain 與 Milvus 的 DML 操作意味著你能夠構(gòu)建的 RAG 應(yīng)用從靜態(tài)的“只讀”演示進(jìn)化成了動(dòng)態(tài)的、可維護(hù)的、貼近真實(shí)業(yè)務(wù)需求的系統(tǒng)。數(shù)據(jù)不再是初始化后一成不變的背景板而是可以隨著業(yè)務(wù)流動(dòng)和演化的核心資產(chǎn)。從插入第一份文檔到精準(zhǔn)刪除一條錯(cuò)誤記錄再到平滑更新整個(gè)知識(shí)體系每一步操作都關(guān)乎著最終應(yīng)用的可信度和智能水平。