在構建基于大語言模型的應用時向量數(shù)據(jù)庫是連接非結構化文本與模型理解能力的關鍵橋梁。然而很多開發(fā)者在完成向量存儲后往往忽略了數(shù)據(jù)管理的重要性導致應用難以維護和迭代。本文將聚焦于 LangChain 與 Milvus 的深度集成手把手帶你完成從數(shù)據(jù)插入、更新到刪除的完整 DML數(shù)據(jù)操作語言實戰(zhàn)確保你的 AI 應用數(shù)據(jù)層既高效又健壯。無論你是剛接觸向量數(shù)據(jù)庫的新手還是希望優(yōu)化現(xiàn)有 LangChain 工作流的開發(fā)者本文都將提供一套可直接復用的代碼方案和清晰的工程實踐思路。1. 背景與核心概念為什么需要關注 DML在 LangChain 生態(tài)中Milvus 因其高性能、可擴展性以及對海量向量數(shù)據(jù)的友好支持成為了許多開發(fā)者的首選向量數(shù)據(jù)庫。我們通常關注如何將文檔切分、嵌入并存入 Milvus即“寫”操作但這僅僅是開始。一個成熟的 AI 應用其知識庫必然是動態(tài)的數(shù)據(jù)更新源文檔內容修訂后需要同步更新向量庫中的對應條目。數(shù)據(jù)刪除清理過時、無效或敏感的數(shù)據(jù)。條件操作需要根據(jù)元數(shù)據(jù)如文檔ID、類別、創(chuàng)建時間進行批量更新或刪除。這些操作統(tǒng)稱為 DML。在 Milvus 中DML 主要圍繞insert、upsert、delete等操作展開。LangChain 的Milvus向量存儲封裝類雖然提供了基礎的add_documents和delete方法但要實現(xiàn)靈活、精準的數(shù)據(jù)管理我們需要深入其底層連接直接使用 Milvus SDK 的強大功能。本文將深入探討如何結合 LangChain 的高級抽象與 Milvus SDK 的精細控制構建一個可維護的向量數(shù)據(jù)管理流程。2. 環(huán)境準備與版本說明在開始實戰(zhàn)之前請確保你的開發(fā)環(huán)境已就緒。以下版本為本文撰寫時的穩(wěn)定版本實際操作時請根據(jù)你的項目需求進行微調。核心組件版本Python: 3.8LangChain: 0.1.x (本文示例基于0.1.10的語法請注意 LangChain 版本迭代較快部分導入路徑可能變化)Milvus: 2.3.x 或以上 (Standalone 或 Cluster 模式均可)pymilvus: 2.3.x (Milvus 的 Python SDK)Embedding 模型: 本文使用text-embedding-ada-002的 OpenAI 接口你也可以替換為sentence-transformers等本地模型。項目依賴 (requirements.txt):langchain0.1.10 langchain-openai0.0.5 # 用于調用OpenAI Embedding pymilvus2.3.6 openai1.12.0 python-dotenv1.0.0 # 用于管理環(huán)境變量關鍵環(huán)境變量 (.env文件):# 如果你的Milvus需要認證 MILVUS_URIhttp://localhost:19530 MILVUS_USERusername MILVUS_PASSWORDpassword MILVUS_DB_NAMEdefault # OpenAI (或其他Embedding服務) OPENAI_API_KEYyour_openai_api_key_here啟動 Milvus 服務:如果你使用 Docker 運行 Standalone 模式的 Milvus可以使用以下命令docker run -d --name milvus-standalone \ -p 19530:19530 \ -p 9091:9091 \ -v /path/to/milvus/data:/var/lib/milvus \ -v /path/to/milvus/conf:/etc/milvus \ milvusdb/milvus:v2.3.6-standalone啟動后確保可以通過19530端口連接到 Milvus。3. 核心原理與 LangChain 集成拆解3.1 LangChain 中 Milvus 向量存儲的運作方式LangChain 的Milvus類通常從langchain.vectorstores導入是一個高級封裝。它的核心工作流程是接收文檔接收Document對象列表每個Document包含page_content和metadata。生成嵌入通過指定的Embeddings模型如OpenAIEmbeddings將page_content轉換為向量。與 Milvus 交互通過pymilvusSDK將向量和元數(shù)據(jù)插入到指定的 Milvus 集合Collection中。創(chuàng)建索引可選在插入數(shù)據(jù)后為向量字段創(chuàng)建索引以加速檢索。Milvus.from_documents()方法封裝了創(chuàng)建集合、插入數(shù)據(jù)、建立索引的完整過程。3.2 DML 操作的底層支撐pymilvus SDK為了實現(xiàn)更精細的 DML 控制我們必須理解其底層依賴——pymilvus。幾個關鍵對象Collection: 代表 Milvus 中的一個數(shù)據(jù)集合是操作的入口。MutationResult: 執(zhí)行插入、刪除、更新操作后返回的結果包含影響的行數(shù)等信息。主鍵 (Primary Key): Milvus 集合必須有一個主鍵字段通常是int64或varchar。在 LangChain 默認配置中它會自動生成一個pk字段。為了支持更新和刪除我們必須自定義一個有意義且唯一的主鍵如doc_id并將其存入metadata。3.3 自定義主鍵的策略這是實現(xiàn)可靠 DML 的基石。LangChain 默認使用自動生成的 ID這不利于追蹤。我們的策略是在創(chuàng)建集合時顯式定義一個主鍵字段例如doc_id(VARCHAR類型)。在文檔的metadata中必須包含這個doc_id。插入數(shù)據(jù)時將doc_id作為主鍵值傳入。后續(xù)的更新和刪除操作都可以通過這個doc_id來精確定位數(shù)據(jù)。4. 完整實戰(zhàn)構建支持 CRUD 的向量數(shù)據(jù)管理模塊接下來我們將一步步構建一個完整的示例。假設我們管理一個“技術文章”知識庫每篇文章有唯一ID、標題、內容和分類。4.1 初始化創(chuàng)建帶自定義主鍵的 Milvus 集合首先我們不再完全依賴 LangChain 的自動創(chuàng)建而是先通過pymilvus明確定義集合結構。# file: init_milvus_collection.py from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility from langchain_openai import OpenAIEmbeddings import os from dotenv import load_dotenv load_dotenv() # 1. 連接 Milvus connections.connect( aliasdefault, urios.getenv(MILVUS_URI, http://localhost:19530), useros.getenv(MILVUS_USER, ), # 如果未設置認證則為空字符串 passwordos.getenv(MILVUS_PASSWORD, ), db_nameos.getenv(MILVUS_DB_NAME, default) ) # 2. 定義集合名稱和 Embedding 維度 collection_name tech_articles_with_pk embedding_dim 1536 # OpenAI text-embedding-ada-002 的維度 # 3. 刪除已存在的同名集合僅用于演示生產環(huán)境慎用 if utility.has_collection(collection_name): utility.drop_collection(collection_name) # 4. 定義字段 Schema fields [ FieldSchema(namepk, dtypeDataType.VARCHAR, is_primaryTrue, auto_idFalse, max_length100), # 自定義主鍵不自增 FieldSchema(namedoc_id, dtypeDataType.VARCHAR, max_length100), # 與pk保持一致方便查詢 FieldSchema(nametitle, dtypeDataType.VARCHAR, max_length500), FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length65535), FieldSchema(namecategory, dtypeDataType.VARCHAR, max_length50), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimembedding_dim), FieldSchema(nameupdate_time, dtypeDataType.INT64), # 用于記錄更新時間戳 ] # 5. 創(chuàng)建集合 Schema schema CollectionSchema(fields, description技術文章向量庫支持通過doc_id更新) # 6. 創(chuàng)建集合 collection Collection(namecollection_name, schemaschema) print(f集合 {collection_name} 創(chuàng)建成功主鍵字段為 pk。) # 7. 為向量字段創(chuàng)建索引HNSW是常用索引 index_params { metric_type: L2, index_type: HNSW, params: {M: 8, efConstruction: 200}, } collection.create_index(embedding, index_params) print(向量索引創(chuàng)建成功。) # 8. 加載集合到內存執(zhí)行搜索前必須加載 collection.load() print(集合已加載。)4.2 封裝數(shù)據(jù)操作類我們將創(chuàng)建一個工具類封裝 LangChain 的文檔添加以及基于pymilvus的更新、刪除操作。# file: milvus_dml_manager.py from typing import List, Optional, Dict, Any from langchain.schema import Document from langchain.vectorstores import Milvus from langchain_openai import OpenAIEmbeddings from pymilvus import Collection, connections import openai import time import os class MilvusDMLManager: def __init__(self, collection_name: str, embedding_model: OpenAIEmbeddings): 初始化管理器。 :param collection_name: Milvus 集合名稱 :param embedding_model: LangChain Embeddings 模型實例 self.collection_name collection_name self.embedding_model embedding_model # 獲取 Milvus 集合對象 self.collection Collection(collection_name) # 初始化 LangChain 的 Milvus 向量存儲用于檢索和基礎添加 # 注意這里我們傳入已存在的集合名并禁用自動創(chuàng)建schema self.vector_store Milvus( embedding_functionself.embedding_model, collection_nameself.collection_name, connection_args{ uri: os.getenv(MILVUS_URI, http://localhost:19530), user: os.getenv(MILVUS_USER, ), password: os.getenv(MILVUS_PASSWORD, ), db_name: os.getenv(MILVUS_DB_NAME, default) }, auto_schemaFalse, # 關鍵告訴 LangChain 不要自動創(chuàng)建schema ) def add_documents(self, documents: List[Document]) - List[str]: 使用 LangChain 添加文檔并確保主鍵被正確設置。 LangChain 的 add_documents 會調用 embedding 模型并插入數(shù)據(jù)。 但我們需要確保 Document 的 metadata 中包含我們定義的 doc_id 并且這個 doc_id 會被用作 Milvus 的主鍵 pk。 # 在插入前可以驗證 documents 是否包含 doc_id for doc in documents: if doc_id not in doc.metadata: raise ValueError(f文檔缺失 doc_id metadata: {doc.page_content[:100]}...) # 調用 LangChain 的添加方法。 # Milvus 類內部會處理 embedding 生成并將 metadata 中的所有字段插入。 # 根據(jù)我們的集合定義它會把 metadata 中的 doc_id 值同時賦給 doc_id 字段和主鍵 pk 字段。 # 這是通過在創(chuàng)建集合時將 pk 字段的 auto_id 設為 False 實現(xiàn)的插入時必須提供主鍵值。 # LangChain 的 Milvus 模塊會將 id 或指定的字段作為主鍵。 # 我們需要確保初始化 Milvus 向量存儲時通過 primary_field 參數(shù)指定主鍵字段為 pk。 # 但更直接的方式是在初始化 self.vector_store 時確保其配置與我們的集合匹配。 # 一個更穩(wěn)妥的做法是直接使用 pymilvus 插入但為了利用 LangChain 的 embedding 流程我們稍作調整 # 實際上LangChain 的 Milvus 類在插入時會嘗試將每個 Document 的 id (如果不存在則生成) 作為主鍵。 # 我們需要將我們的 doc_id 賦值給 Document 的 id 屬性。 for doc in documents: doc.id doc.metadata[doc_id] # 添加文檔 pks self.vector_store.add_documents(documents) print(f成功添加 {len(pks)} 個文檔主鍵列表: {pks[:5]}...) # 打印前5個 return pks def upsert_document(self, document: Document): 更新或插入文檔。 邏輯如果存在相同 doc_id 的文檔則先刪除再插入Milvus 的 Upsert 在 2.3 版本可用這里提供通用方法。 :param document: LangChain Document 對象其 metadata 必須包含 doc_id doc_id document.metadata.get(doc_id) if not doc_id: raise ValueError(文檔必須包含 doc_id metadata 以支持 upsert。) # 1. 生成嵌入向量 embedding self.embedding_model.embed_query(document.page_content) # 2. 準備數(shù)據(jù)行字段順序需與集合定義一致 data [ [doc_id], # pk [doc_id], # doc_id [document.metadata.get(title, )], # title [document.page_content], # content [document.metadata.get(category, )], # category [embedding], # embedding [int(time.time())] # update_time ] # 3. 構建刪除表達式 expr fpk {doc_id} # 4. 執(zhí)行刪除如果存在 delete_result self.collection.delete(expr) print(f嘗試刪除 doc_id{doc_id}, 刪除條數(shù): {delete_result.delete_count}) # 5. 插入新數(shù)據(jù) insert_result self.collection.insert(data) print(f插入 doc_id{doc_id} 成功主鍵: {insert_result.primary_keys}) # 6. 刷新集合使更改立即可見對于小規(guī)模操作可以最后統(tǒng)一刷新 self.collection.flush() return insert_result def delete_by_doc_id(self, doc_id: str) - int: 根據(jù)文檔ID刪除數(shù)據(jù)。 :param doc_id: 要刪除的文檔ID :return: 被刪除的行數(shù) expr fpk {doc_id} # 或 fdoc_id {doc_id}兩者值相同 delete_result self.collection.delete(expr) print(f刪除表達式: {expr}, 影響行數(shù): {delete_result.delete_count}) self.collection.flush() return delete_result.delete_count def delete_by_condition(self, condition_expr: str) - int: 根據(jù)條件表達式批量刪除。 :param condition_expr: Milvus 布爾表達式例如 category deprecated :return: 被刪除的行數(shù) # 注意刪除前最好先查詢確認避免誤刪 query_result self.collection.query(exprcondition_expr, output_fields[pk, title]) if query_result: print(f即將刪除以下 {len(query_result)} 條記錄:) for item in query_result[:5]: # 預覽前5條 print(f - PK: {item[pk]}, Title: {item.get(title)}) if len(query_result) 5: print(f ... 以及另外 {len(query_result)-5} 條記錄) # 這里可以添加人工確認邏輯生產環(huán)境建議有審批流程 delete_result self.collection.delete(condition_expr) print(f刪除完成影響行數(shù): {delete_result.delete_count}) self.collection.flush() return delete_result.delete_count def search_similar(self, query: str, k: int 3) - List[Dict]: 使用 LangChain 的向量存儲進行相似性搜索。 docs self.vector_store.similarity_search(query, kk) results [] for doc in docs: results.append({ content: doc.page_content, metadata: doc.metadata, score: doc.metadata.get(score, 0) # LangChain 可能不直接返回分數(shù) }) return results4.3 主程序執(zhí)行完整的 DML 操作鏈現(xiàn)在我們編寫一個主程序來使用這個管理器。# file: main_demo.py from milvus_dml_manager import MilvusDMLManager from langchain.schema import Document from langchain_openai import OpenAIEmbeddings import os from dotenv import load_dotenv load_dotenv() # 初始化 Embedding 模型 embeddings OpenAIEmbeddings( modeltext-embedding-ada-002, openai_api_keyos.getenv(OPENAI_API_KEY) ) # 初始化管理器 manager MilvusDMLManager( collection_nametech_articles_with_pk, embedding_modelembeddings ) print( 1. 添加初始文檔 ) initial_docs [ Document( page_contentLangChain是一個用于開發(fā)由語言模型驅動的應用程序的框架。, metadata{doc_id: doc_001, title: LangChain簡介, category: framework} ), Document( page_contentMilvus是一個開源向量數(shù)據(jù)庫專為海量向量相似性搜索而設計。, metadata{doc_id: doc_002, title: Milvus概述, category: database} ), Document( page_contentDML包括INSERT、UPDATE、DELETE等數(shù)據(jù)操作命令。, metadata{doc_id: doc_003, title: 數(shù)據(jù)庫DML, category: database} ), ] added_ids manager.add_documents(initial_docs) print(f添加的文檔ID: {added_ids}\n) print( 2. 相似性搜索測試 ) search_results manager.search_similar(什么是向量數(shù)據(jù)庫, k2) for res in search_results: print(f- 內容: {res[content][:80]}...) print(f 元數(shù)據(jù): {res[metadata]}\n) print( 3. 更新文檔 (Upsert) ) # 假設 doc_002 的內容需要更新 updated_doc Document( page_contentMilvus是一個云原生開源向量數(shù)據(jù)庫支持秒級檢索萬億級向量廣泛應用于AI、推薦、搜索等領域。, metadata{doc_id: doc_002, title: Milvus詳解, category: vector-db} ) upsert_result manager.upsert_document(updated_doc) print(fUpsert 操作完成。\n) print( 4. 驗證更新結果 ) # 再次搜索查看更新后的內容 search_results_after_update manager.search_similar(云原生向量數(shù)據(jù)庫, k1) for res in search_results_after_update: print(f更新后搜索結果: {res[content]}) print(f對應元數(shù)據(jù): {res[metadata]}\n) print( 5. 條件刪除 ) # 刪除分類為 framework 的文檔這里會刪除 doc_001 deleted_count manager.delete_by_condition(category framework) print(f條件刪除完成共刪除 {deleted_count} 條記錄。\n) print( 6. 按 ID 刪除 ) # 刪除 doc_003 deleted_count_id manager.delete_by_doc_id(doc_003) print(f按ID刪除完成共刪除 {deleted_count_id} 條記錄。\n) print( 7. 最終搜索驗證 ) # 現(xiàn)在集合中應該只剩下 doc_002 final_results manager.search_similar(數(shù)據(jù)庫, k5) print(f剩余文檔數(shù)量: {len(final_results)}) for res in final_results: print(f- 剩余文檔: {res[metadata].get(title)} (ID: {res[metadata].get(doc_id)}))4.4 運行與驗證確保 Milvus 服務正在運行。按順序執(zhí)行腳本python init_milvus_collection.py python main_demo.py觀察控制臺輸出你應該能看到完整的“添加 - 搜索 - 更新 - 再搜索 - 條件刪除 - 按ID刪除 - 最終驗證”流程。預期輸出關鍵點初始添加成功并返回主鍵列表。第一次搜索能查到“Milvus是一個開源向量數(shù)據(jù)庫...”。Upsert 后再次搜索“云原生向量數(shù)據(jù)庫”返回的內容應變?yōu)楦潞蟮拈L文本。條件刪除后分類為framework的文檔被移除。按ID刪除后doc_003被移除。最終集合中僅剩更新后的doc_002。5. 常見問題與排查思路在集成 LangChain 與 Milvus 進行 DML 操作時你可能會遇到以下典型問題問題現(xiàn)象可能原因排查思路與解決方案插入失敗提示主鍵沖突1. 主鍵字段auto_idTrue但插入時又提供了值。2. 多次插入相同的doc_id。1. 檢查集合 Schema 定義確保is_primaryTrue的字段其auto_idFalse。2. 實現(xiàn) Upsert 邏輯先刪后插或檢查業(yè)務邏輯避免重復插入。更新后搜索不到最新數(shù)據(jù)1. 數(shù)據(jù)未刷新 (flush)。2. 索引未重建對于大量更新后。3. 搜索時未加載集合。1. 在插入/刪除操作后調用collection.flush()。2. 對于大量數(shù)據(jù)變更考慮在后臺異步重建索引。3. 確保在執(zhí)行搜索前collection.load()。LangChain 的add_documents不生效1.auto_schemaTrue但與已存在集合的 Schema 沖突。2. Document 的metadata字段與集合 Schema 不匹配。3. 連接參數(shù)如 URI、端口錯誤。1. 對于已存在的集合初始化Milvus時設置auto_schemaFalse。2. 檢查集合已有字段確保metadata中的鍵能對應上或使用collection.create_partition管理。3. 使用connections.list_connections()檢查連接狀態(tài)。按條件刪除 (delete_by_condition) 報錯1. 表達式語法錯誤。2. 嘗試在未加載的集合上執(zhí)行刪除。3. 字段名或值類型不匹配。1. 使用簡單的表達式測試如pk id1。Milvus 表達式語法參考其文檔。2. 確保collection.load()已被調用。3. 使用collection.query(expr, output_fields[*])先驗證表達式是否能正確查詢到目標數(shù)據(jù)。嵌入維度不匹配1. 創(chuàng)建集合時定義的dim與 Embedding 模型實際產生的維度不一致。1. 確認 Embedding 模型的輸出維度如text-embedding-ada-002是 1536。2. 重建集合或使用collection.alter修改字段此操作復雜通常建議重建。性能問題插入/更新慢1. 單條插入。2. 索引類型不適合寫多讀少的場景。3. 網絡延遲或資源不足。1. 采用批量插入如每次插入 100-500 條數(shù)據(jù)。2. 評估索引類型IVF_FLAT比HNSW寫入更快但搜索精度可能稍低。3. 監(jiān)控 Milvus 集群資源CPU、內存、磁盤IO。6. 最佳實踐與工程建議將 LangChain 與 Milvus 用于生產環(huán)境時除了功能實現(xiàn)更應關注可靠性、可維護性和性能。主鍵設計是基石全局唯一且業(yè)務相關不要使用無意義的自增ID或UUID。使用能標識數(shù)據(jù)實體的字段如user_id:content_hash的組合便于直接定位和業(yè)務關聯(lián)。類型選擇根據(jù)數(shù)據(jù)量選擇VARCHAR或INT64。VARCHAR更靈活INT64性能稍好。實現(xiàn)數(shù)據(jù)版本化管理在元數(shù)據(jù)中增加version、update_time字段。實施“軟刪除”增加is_deleted標志位而不是物理刪除。這樣便于數(shù)據(jù)追溯和恢復。重大更新時可以采用“插入新版本標記舊版本失效”的策略而非原地更新便于AB測試或回滾。批處理與異步操作對于大規(guī)模數(shù)據(jù)導入或更新務必使用批量操作。pymilvus的insert方法接受多行數(shù)據(jù)。考慮將耗時的 DML 操作放入異步任務隊列如 Celery、Dramatiq避免阻塞主應用線程。# 批量插入示例 def batch_insert_docs(doc_list: List[Document], batch_size200): for i in range(0, len(doc_list), batch_size): batch doc_list[i:ibatch_size] # ... 處理并插入 batch self.collection.insert(batch_data) self.collection.flush()操作前備份與驗證在執(zhí)行批量刪除或更新前務必先執(zhí)行查詢確認影響范圍如示例中的delete_by_condition方法所示。對于核心數(shù)據(jù)定期為 Milvus 集合創(chuàng)建快照如果使用 Milvus 企業(yè)版或云服務。編寫數(shù)據(jù)遷移腳本時應有“試運行”Dry-Run模式只打印將要執(zhí)行的操作而不實際執(zhí)行。監(jiān)控與日志記錄所有 DML 操作的關鍵信息操作類型、影響的主鍵、操作時間、執(zhí)行人或服務、耗時。監(jiān)控 Milvus 的關鍵指標集合中的實體數(shù)量、索引狀態(tài)、查詢/插入 QPS、內存使用率。為刪除操作設置更高級別的日志如 WARNING 或 ERROR 級別。與 LangChain 生態(tài)的優(yōu)雅結合將MilvusDMLManager這樣的管理類封裝成獨立的服務或模塊與 LangChain 的 Chain 或 Agent 通過 API 交互。利用 LangChain 的Retriever抽象。你可以自定義一個Retriever它在內部調用你的管理器并可能加入緩存、過濾等邏輯?？紤]使用langchain.indexes模塊來維護向量存儲的增量更新它提供了SQLRecordManager來追蹤文檔的哈希狀態(tài)。通過以上實踐你可以構建一個不僅功能強大而且穩(wěn)定、可觀測、易于維護的向量數(shù)據(jù)管理系統(tǒng)使其成為你 AI 應用堅實的數(shù)據(jù)底座。