在構建基于大語言模型的應用時向量數據庫是連接非結構化文本與模型理解能力的關鍵橋梁。然而很多開發者在完成向量存儲后往往忽略了數據管理的重要性導致應用難以維護和迭代。本文將聚焦于 LangChain 與 Milvus 的深度集成手把手帶你完成從數據插入、更新到刪除的完整 DML數據操作語言實戰確保你的 AI 應用數據層既高效又健壯。無論你是剛接觸向量數據庫的新手還是希望優化現有 LangChain 工作流的開發者本文都將提供一套可直接復用的代碼方案和清晰的工程實踐思路。1. 背景與核心概念為什么需要關注 DML在 LangChain 生態中Milvus 因其高性能、可擴展性以及對海量向量數據的友好支持成為了許多開發者的首選向量數據庫。我們通常關注如何將文檔切分、嵌入并存入 Milvus即“寫”操作但這僅僅是開始。一個成熟的 AI 應用其知識庫必然是動態的數據更新源文檔內容修訂后需要同步更新向量庫中的對應條目。數據刪除清理過時、無效或敏感的數據。條件操作需要根據元數據如文檔ID、類別、創建時間進行批量更新或刪除。這些操作統稱為 DML。在 Milvus 中DML 主要圍繞insert、upsert、delete等操作展開。LangChain 的Milvus向量存儲封裝類雖然提供了基礎的add_documents和delete方法但要實現靈活、精準的數據管理我們需要深入其底層連接直接使用 Milvus SDK 的強大功能。本文將深入探討如何結合 LangChain 的高級抽象與 Milvus SDK 的精細控制構建一個可維護的向量數據管理流程。2. 環境準備與版本說明在開始實戰之前請確保你的開發環境已就緒。以下版本為本文撰寫時的穩定版本實際操作時請根據你的項目需求進行微調。核心組件版本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 # 用于管理環境變量關鍵環境變量 (.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將向量和元數據插入到指定的 Milvus 集合Collection中。創建索引可選在插入數據后為向量字段創建索引以加速檢索。Milvus.from_documents()方法封裝了創建集合、插入數據、建立索引的完整過程。3.2 DML 操作的底層支撐pymilvus SDK為了實現更精細的 DML 控制我們必須理解其底層依賴——pymilvus。幾個關鍵對象Collection: 代表 Milvus 中的一個數據集合是操作的入口。MutationResult: 執行插入、刪除、更新操作后返回的結果包含影響的行數等信息。主鍵 (Primary Key): Milvus 集合必須有一個主鍵字段通常是int64或varchar。在 LangChain 默認配置中它會自動生成一個pk字段。為了支持更新和刪除我們必須自定義一個有意義且唯一的主鍵如doc_id并將其存入metadata。3.3 自定義主鍵的策略這是實現可靠 DML 的基石。LangChain 默認使用自動生成的 ID這不利于追蹤。我們的策略是在創建集合時顯式定義一個主鍵字段例如doc_id(VARCHAR類型)。在文檔的metadata中必須包含這個doc_id。插入數據時將doc_id作為主鍵值傳入。后續的更新和刪除操作都可以通過這個doc_id來精確定位數據。4. 完整實戰構建支持 CRUD 的向量數據管理模塊接下來我們將一步步構建一個完整的示例。假設我們管理一個“技術文章”知識庫每篇文章有唯一ID、標題、內容和分類。4.1 初始化創建帶自定義主鍵的 Milvus 集合首先我們不再完全依賴 LangChain 的自動創建而是先通過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. 刪除已存在的同名集合僅用于演示生產環境慎用 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. 創建集合 Schema schema CollectionSchema(fields, description技術文章向量庫支持通過doc_id更新) # 6. 創建集合 collection Collection(namecollection_name, schemaschema) print(f集合 {collection_name} 創建成功主鍵字段為 pk。) # 7. 為向量字段創建索引HNSW是常用索引 index_params { metric_type: L2, index_type: HNSW, params: {M: 8, efConstruction: 200}, } collection.create_index(embedding, index_params) print(向量索引創建成功。) # 8. 加載集合到內存執行搜索前必須加載 collection.load() print(集合已加載。)4.2 封裝數據操作類我們將創建一個工具類封裝 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 向量存儲用于檢索和基礎添加 # 注意這里我們傳入已存在的集合名并禁用自動創建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 不要自動創建schema ) def add_documents(self, documents: List[Document]) - List[str]: 使用 LangChain 添加文檔并確保主鍵被正確設置。 LangChain 的 add_documents 會調用 embedding 模型并插入數據。 但我們需要確保 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 中的所有字段插入。 # 根據我們的集合定義它會把 metadata 中的 doc_id 值同時賦給 doc_id 字段和主鍵 pk 字段。 # 這是通過在創建集合時將 pk 字段的 auto_id 設為 False 實現的插入時必須提供主鍵值。 # LangChain 的 Milvus 模塊會將 id 或指定的字段作為主鍵。 # 我們需要確保初始化 Milvus 向量存儲時通過 primary_field 參數指定主鍵字段為 pk。 # 但更直接的方式是在初始化 self.vector_store 時確保其配置與我們的集合匹配。 # 一個更穩妥的做法是直接使用 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. 準備數據行字段順序需與集合定義一致 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. 執行刪除如果存在 delete_result self.collection.delete(expr) print(f嘗試刪除 doc_id{doc_id}, 刪除條數: {delete_result.delete_count}) # 5. 插入新數據 insert_result self.collection.insert(data) print(f插入 doc_id{doc_id} 成功主鍵: {insert_result.primary_keys}) # 6. 刷新集合使更改立即可見對于小規模操作可以最后統一刷新 self.collection.flush() return insert_result def delete_by_doc_id(self, doc_id: str) - int: 根據文檔ID刪除數據。 :param doc_id: 要刪除的文檔ID :return: 被刪除的行數 expr fpk {doc_id} # 或 fdoc_id {doc_id}兩者值相同 delete_result self.collection.delete(expr) print(f刪除表達式: {expr}, 影響行數: {delete_result.delete_count}) self.collection.flush() return delete_result.delete_count def delete_by_condition(self, condition_expr: str) - int: 根據條件表達式批量刪除。 :param condition_expr: Milvus 布爾表達式例如 category deprecated :return: 被刪除的行數 # 注意刪除前最好先查詢確認避免誤刪 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} 條記錄) # 這里可以添加人工確認邏輯生產環境建議有審批流程 delete_result self.collection.delete(condition_expr) print(f刪除完成影響行數: {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 可能不直接返回分數 }) return results4.3 主程序執行完整的 DML 操作鏈現在我們編寫一個主程序來使用這個管理器。# 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是一個用于開發由語言模型驅動的應用程序的框架。, metadata{doc_id: doc_001, title: LangChain簡介, category: framework} ), Document( page_contentMilvus是一個開源向量數據庫專為海量向量相似性搜索而設計。, metadata{doc_id: doc_002, title: Milvus概述, category: database} ), Document( page_contentDML包括INSERT、UPDATE、DELETE等數據操作命令。, metadata{doc_id: doc_003, title: 數據庫DML, category: database} ), ] added_ids manager.add_documents(initial_docs) print(f添加的文檔ID: {added_ids}\n) print( 2. 相似性搜索測試 ) search_results manager.search_similar(什么是向量數據庫, k2) for res in search_results: print(f- 內容: {res[content][:80]}...) print(f 元數據: {res[metadata]}\n) print( 3. 更新文檔 (Upsert) ) # 假設 doc_002 的內容需要更新 updated_doc Document( page_contentMilvus是一個云原生開源向量數據庫支持秒級檢索萬億級向量廣泛應用于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(云原生向量數據庫, k1) for res in search_results_after_update: print(f更新后搜索結果: {res[content]}) print(f對應元數據: {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. 最終搜索驗證 ) # 現在集合中應該只剩下 doc_002 final_results manager.search_similar(數據庫, k5) print(f剩余文檔數量: {len(final_results)}) for res in final_results: print(f- 剩余文檔: {res[metadata].get(title)} (ID: {res[metadata].get(doc_id)}))4.4 運行與驗證確保 Milvus 服務正在運行。按順序執行腳本python init_milvus_collection.py python main_demo.py觀察控制臺輸出你應該能看到完整的“添加 - 搜索 - 更新 - 再搜索 - 條件刪除 - 按ID刪除 - 最終驗證”流程。預期輸出關鍵點初始添加成功并返回主鍵列表。第一次搜索能查到“Milvus是一個開源向量數據庫...”。Upsert 后再次搜索“云原生向量數據庫”返回的內容應變為更新后的長文本。條件刪除后分類為framework的文檔被移除。按ID刪除后doc_003被移除。最終集合中僅剩更新后的doc_002。5. 常見問題與排查思路在集成 LangChain 與 Milvus 進行 DML 操作時你可能會遇到以下典型問題問題現象可能原因排查思路與解決方案插入失敗提示主鍵沖突1. 主鍵字段auto_idTrue但插入時又提供了值。2. 多次插入相同的doc_id。1. 檢查集合 Schema 定義確保is_primaryTrue的字段其auto_idFalse。2. 實現 Upsert 邏輯先刪后插或檢查業務邏輯避免重復插入。更新后搜索不到最新數據1. 數據未刷新 (flush)。2. 索引未重建對于大量更新后。3. 搜索時未加載集合。1. 在插入/刪除操作后調用collection.flush()。2. 對于大量數據變更考慮在后臺異步重建索引。3. 確保在執行搜索前collection.load()。LangChain 的add_documents不生效1.auto_schemaTrue但與已存在集合的 Schema 沖突。2. Document 的metadata字段與集合 Schema 不匹配。3. 連接參數如 URI、端口錯誤。1. 對于已存在的集合初始化Milvus時設置auto_schemaFalse。2. 檢查集合已有字段確保metadata中的鍵能對應上或使用collection.create_partition管理。3. 使用connections.list_connections()檢查連接狀態。按條件刪除 (delete_by_condition) 報錯1. 表達式語法錯誤。2. 嘗試在未加載的集合上執行刪除。3. 字段名或值類型不匹配。1. 使用簡單的表達式測試如pk id1。Milvus 表達式語法參考其文檔。2. 確保collection.load()已被調用。3. 使用collection.query(expr, output_fields[*])先驗證表達式是否能正確查詢到目標數據。嵌入維度不匹配1. 創建集合時定義的dim與 Embedding 模型實際產生的維度不一致。1. 確認 Embedding 模型的輸出維度如text-embedding-ada-002是 1536。2. 重建集合或使用collection.alter修改字段此操作復雜通常建議重建。性能問題插入/更新慢1. 單條插入。2. 索引類型不適合寫多讀少的場景。3. 網絡延遲或資源不足。1. 采用批量插入如每次插入 100-500 條數據。2. 評估索引類型IVF_FLAT比HNSW寫入更快但搜索精度可能稍低。3. 監控 Milvus 集群資源CPU、內存、磁盤IO。6. 最佳實踐與工程建議將 LangChain 與 Milvus 用于生產環境時除了功能實現更應關注可靠性、可維護性和性能。主鍵設計是基石全局唯一且業務相關不要使用無意義的自增ID或UUID。使用能標識數據實體的字段如user_id:content_hash的組合便于直接定位和業務關聯。類型選擇根據數據量選擇VARCHAR或INT64。VARCHAR更靈活INT64性能稍好。實現數據版本化管理在元數據中增加version、update_time字段。實施“軟刪除”增加is_deleted標志位而不是物理刪除。這樣便于數據追溯和恢復。重大更新時可以采用“插入新版本標記舊版本失效”的策略而非原地更新便于AB測試或回滾。批處理與異步操作對于大規模數據導入或更新務必使用批量操作。pymilvus的insert方法接受多行數據。考慮將耗時的 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()操作前備份與驗證在執行批量刪除或更新前務必先執行查詢確認影響范圍如示例中的delete_by_condition方法所示。對于核心數據定期為 Milvus 集合創建快照如果使用 Milvus 企業版或云服務。編寫數據遷移腳本時應有“試運行”Dry-Run模式只打印將要執行的操作而不實際執行。監控與日志記錄所有 DML 操作的關鍵信息操作類型、影響的主鍵、操作時間、執行人或服務、耗時。監控 Milvus 的關鍵指標集合中的實體數量、索引狀態、查詢/插入 QPS、內存使用率。為刪除操作設置更高級別的日志如 WARNING 或 ERROR 級別。與 LangChain 生態的優雅結合將MilvusDMLManager這樣的管理類封裝成獨立的服務或模塊與 LangChain 的 Chain 或 Agent 通過 API 交互。利用 LangChain 的Retriever抽象。你可以自定義一個Retriever它在內部調用你的管理器并可能加入緩存、過濾等邏輯。考慮使用langchain.indexes模塊來維護向量存儲的增量更新它提供了SQLRecordManager來追蹤文檔的哈希狀態。通過以上實踐你可以構建一個不僅功能強大而且穩定、可觀測、易于維護的向量數據管理系統使其成為你 AI 應用堅實的數據底座。 LangChain與Milvus向量數據庫DML實戰：數據更新與刪除操作指南
在構建基于大語言模型的應用時向量數據庫是連接非結構化文本與模型理解能力的關鍵橋梁。然而很多開發者在完成向量存儲后往往忽略了數據管理的重要性導致應用難以維護和迭代。本文將聚焦于 LangChain 與 Milvus 的深度集成手把手帶你完成從數據插入、更新到刪除的完整 DML數據操作語言實戰確保你的 AI 應用數據層既高效又健壯。無論你是剛接觸向量數據庫的新手還是希望優化現有 LangChain 工作流的開發者本文都將提供一套可直接復用的代碼方案和清晰的工程實踐思路。1. 背景與核心概念為什么需要關注 DML在 LangChain 生態中Milvus 因其高性能、可擴展性以及對海量向量數據的友好支持成為了許多開發者的首選向量數據庫。我們通常關注如何將文檔切分、嵌入并存入 Milvus即“寫”操作但這僅僅是開始。一個成熟的 AI 應用其知識庫必然是動態的數據更新源文檔內容修訂后需要同步更新向量庫中的對應條目。數據刪除清理過時、無效或敏感的數據。條件操作需要根據元數據如文檔ID、類別、創建時間進行批量更新或刪除。這些操作統稱為 DML。在 Milvus 中DML 主要圍繞insert、upsert、delete等操作展開。LangChain 的Milvus向量存儲封裝類雖然提供了基礎的add_documents和delete方法但要實現靈活、精準的數據管理我們需要深入其底層連接直接使用 Milvus SDK 的強大功能。本文將深入探討如何結合 LangChain 的高級抽象與 Milvus SDK 的精細控制構建一個可維護的向量數據管理流程。2. 環境準備與版本說明在開始實戰之前請確保你的開發環境已就緒。以下版本為本文撰寫時的穩定版本實際操作時請根據你的項目需求進行微調。核心組件版本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 # 用于管理環境變量關鍵環境變量 (.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將向量和元數據插入到指定的 Milvus 集合Collection中。創建索引可選在插入數據后為向量字段創建索引以加速檢索。Milvus.from_documents()方法封裝了創建集合、插入數據、建立索引的完整過程。3.2 DML 操作的底層支撐pymilvus SDK為了實現更精細的 DML 控制我們必須理解其底層依賴——pymilvus。幾個關鍵對象Collection: 代表 Milvus 中的一個數據集合是操作的入口。MutationResult: 執行插入、刪除、更新操作后返回的結果包含影響的行數等信息。主鍵 (Primary Key): Milvus 集合必須有一個主鍵字段通常是int64或varchar。在 LangChain 默認配置中它會自動生成一個pk字段。為了支持更新和刪除我們必須自定義一個有意義且唯一的主鍵如doc_id并將其存入metadata。3.3 自定義主鍵的策略這是實現可靠 DML 的基石。LangChain 默認使用自動生成的 ID這不利于追蹤。我們的策略是在創建集合時顯式定義一個主鍵字段例如doc_id(VARCHAR類型)。在文檔的metadata中必須包含這個doc_id。插入數據時將doc_id作為主鍵值傳入。后續的更新和刪除操作都可以通過這個doc_id來精確定位數據。4. 完整實戰構建支持 CRUD 的向量數據管理模塊接下來我們將一步步構建一個完整的示例。假設我們管理一個“技術文章”知識庫每篇文章有唯一ID、標題、內容和分類。4.1 初始化創建帶自定義主鍵的 Milvus 集合首先我們不再完全依賴 LangChain 的自動創建而是先通過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. 刪除已存在的同名集合僅用于演示生產環境慎用 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. 創建集合 Schema schema CollectionSchema(fields, description技術文章向量庫支持通過doc_id更新) # 6. 創建集合 collection Collection(namecollection_name, schemaschema) print(f集合 {collection_name} 創建成功主鍵字段為 pk。) # 7. 為向量字段創建索引HNSW是常用索引 index_params { metric_type: L2, index_type: HNSW, params: {M: 8, efConstruction: 200}, } collection.create_index(embedding, index_params) print(向量索引創建成功。) # 8. 加載集合到內存執行搜索前必須加載 collection.load() print(集合已加載。)4.2 封裝數據操作類我們將創建一個工具類封裝 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 向量存儲用于檢索和基礎添加 # 注意這里我們傳入已存在的集合名并禁用自動創建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 不要自動創建schema ) def add_documents(self, documents: List[Document]) - List[str]: 使用 LangChain 添加文檔并確保主鍵被正確設置。 LangChain 的 add_documents 會調用 embedding 模型并插入數據。 但我們需要確保 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 中的所有字段插入。 # 根據我們的集合定義它會把 metadata 中的 doc_id 值同時賦給 doc_id 字段和主鍵 pk 字段。 # 這是通過在創建集合時將 pk 字段的 auto_id 設為 False 實現的插入時必須提供主鍵值。 # LangChain 的 Milvus 模塊會將 id 或指定的字段作為主鍵。 # 我們需要確保初始化 Milvus 向量存儲時通過 primary_field 參數指定主鍵字段為 pk。 # 但更直接的方式是在初始化 self.vector_store 時確保其配置與我們的集合匹配。 # 一個更穩妥的做法是直接使用 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. 準備數據行字段順序需與集合定義一致 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. 執行刪除如果存在 delete_result self.collection.delete(expr) print(f嘗試刪除 doc_id{doc_id}, 刪除條數: {delete_result.delete_count}) # 5. 插入新數據 insert_result self.collection.insert(data) print(f插入 doc_id{doc_id} 成功主鍵: {insert_result.primary_keys}) # 6. 刷新集合使更改立即可見對于小規模操作可以最后統一刷新 self.collection.flush() return insert_result def delete_by_doc_id(self, doc_id: str) - int: 根據文檔ID刪除數據。 :param doc_id: 要刪除的文檔ID :return: 被刪除的行數 expr fpk {doc_id} # 或 fdoc_id {doc_id}兩者值相同 delete_result self.collection.delete(expr) print(f刪除表達式: {expr}, 影響行數: {delete_result.delete_count}) self.collection.flush() return delete_result.delete_count def delete_by_condition(self, condition_expr: str) - int: 根據條件表達式批量刪除。 :param condition_expr: Milvus 布爾表達式例如 category deprecated :return: 被刪除的行數 # 注意刪除前最好先查詢確認避免誤刪 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} 條記錄) # 這里可以添加人工確認邏輯生產環境建議有審批流程 delete_result self.collection.delete(condition_expr) print(f刪除完成影響行數: {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 可能不直接返回分數 }) return results4.3 主程序執行完整的 DML 操作鏈現在我們編寫一個主程序來使用這個管理器。# 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是一個用于開發由語言模型驅動的應用程序的框架。, metadata{doc_id: doc_001, title: LangChain簡介, category: framework} ), Document( page_contentMilvus是一個開源向量數據庫專為海量向量相似性搜索而設計。, metadata{doc_id: doc_002, title: Milvus概述, category: database} ), Document( page_contentDML包括INSERT、UPDATE、DELETE等數據操作命令。, metadata{doc_id: doc_003, title: 數據庫DML, category: database} ), ] added_ids manager.add_documents(initial_docs) print(f添加的文檔ID: {added_ids}\n) print( 2. 相似性搜索測試 ) search_results manager.search_similar(什么是向量數據庫, k2) for res in search_results: print(f- 內容: {res[content][:80]}...) print(f 元數據: {res[metadata]}\n) print( 3. 更新文檔 (Upsert) ) # 假設 doc_002 的內容需要更新 updated_doc Document( page_contentMilvus是一個云原生開源向量數據庫支持秒級檢索萬億級向量廣泛應用于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(云原生向量數據庫, k1) for res in search_results_after_update: print(f更新后搜索結果: {res[content]}) print(f對應元數據: {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. 最終搜索驗證 ) # 現在集合中應該只剩下 doc_002 final_results manager.search_similar(數據庫, k5) print(f剩余文檔數量: {len(final_results)}) for res in final_results: print(f- 剩余文檔: {res[metadata].get(title)} (ID: {res[metadata].get(doc_id)}))4.4 運行與驗證確保 Milvus 服務正在運行。按順序執行腳本python init_milvus_collection.py python main_demo.py觀察控制臺輸出你應該能看到完整的“添加 - 搜索 - 更新 - 再搜索 - 條件刪除 - 按ID刪除 - 最終驗證”流程。預期輸出關鍵點初始添加成功并返回主鍵列表。第一次搜索能查到“Milvus是一個開源向量數據庫...”。Upsert 后再次搜索“云原生向量數據庫”返回的內容應變為更新后的長文本。條件刪除后分類為framework的文檔被移除。按ID刪除后doc_003被移除。最終集合中僅剩更新后的doc_002。5. 常見問題與排查思路在集成 LangChain 與 Milvus 進行 DML 操作時你可能會遇到以下典型問題問題現象可能原因排查思路與解決方案插入失敗提示主鍵沖突1. 主鍵字段auto_idTrue但插入時又提供了值。2. 多次插入相同的doc_id。1. 檢查集合 Schema 定義確保is_primaryTrue的字段其auto_idFalse。2. 實現 Upsert 邏輯先刪后插或檢查業務邏輯避免重復插入。更新后搜索不到最新數據1. 數據未刷新 (flush)。2. 索引未重建對于大量更新后。3. 搜索時未加載集合。1. 在插入/刪除操作后調用collection.flush()。2. 對于大量數據變更考慮在后臺異步重建索引。3. 確保在執行搜索前collection.load()。LangChain 的add_documents不生效1.auto_schemaTrue但與已存在集合的 Schema 沖突。2. Document 的metadata字段與集合 Schema 不匹配。3. 連接參數如 URI、端口錯誤。1. 對于已存在的集合初始化Milvus時設置auto_schemaFalse。2. 檢查集合已有字段確保metadata中的鍵能對應上或使用collection.create_partition管理。3. 使用connections.list_connections()檢查連接狀態。按條件刪除 (delete_by_condition) 報錯1. 表達式語法錯誤。2. 嘗試在未加載的集合上執行刪除。3. 字段名或值類型不匹配。1. 使用簡單的表達式測試如pk id1。Milvus 表達式語法參考其文檔。2. 確保collection.load()已被調用。3. 使用collection.query(expr, output_fields[*])先驗證表達式是否能正確查詢到目標數據。嵌入維度不匹配1. 創建集合時定義的dim與 Embedding 模型實際產生的維度不一致。1. 確認 Embedding 模型的輸出維度如text-embedding-ada-002是 1536。2. 重建集合或使用collection.alter修改字段此操作復雜通常建議重建。性能問題插入/更新慢1. 單條插入。2. 索引類型不適合寫多讀少的場景。3. 網絡延遲或資源不足。1. 采用批量插入如每次插入 100-500 條數據。2. 評估索引類型IVF_FLAT比HNSW寫入更快但搜索精度可能稍低。3. 監控 Milvus 集群資源CPU、內存、磁盤IO。6. 最佳實踐與工程建議將 LangChain 與 Milvus 用于生產環境時除了功能實現更應關注可靠性、可維護性和性能。主鍵設計是基石全局唯一且業務相關不要使用無意義的自增ID或UUID。使用能標識數據實體的字段如user_id:content_hash的組合便于直接定位和業務關聯。類型選擇根據數據量選擇VARCHAR或INT64。VARCHAR更靈活INT64性能稍好。實現數據版本化管理在元數據中增加version、update_time字段。實施“軟刪除”增加is_deleted標志位而不是物理刪除。這樣便于數據追溯和恢復。重大更新時可以采用“插入新版本標記舊版本失效”的策略而非原地更新便于AB測試或回滾。批處理與異步操作對于大規模數據導入或更新務必使用批量操作。pymilvus的insert方法接受多行數據。考慮將耗時的 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()操作前備份與驗證在執行批量刪除或更新前務必先執行查詢確認影響范圍如示例中的delete_by_condition方法所示。對于核心數據定期為 Milvus 集合創建快照如果使用 Milvus 企業版或云服務。編寫數據遷移腳本時應有“試運行”Dry-Run模式只打印將要執行的操作而不實際執行。監控與日志記錄所有 DML 操作的關鍵信息操作類型、影響的主鍵、操作時間、執行人或服務、耗時。監控 Milvus 的關鍵指標集合中的實體數量、索引狀態、查詢/插入 QPS、內存使用率。為刪除操作設置更高級別的日志如 WARNING 或 ERROR 級別。與 LangChain 生態的優雅結合將MilvusDMLManager這樣的管理類封裝成獨立的服務或模塊與 LangChain 的 Chain 或 Agent 通過 API 交互。利用 LangChain 的Retriever抽象。你可以自定義一個Retriever它在內部調用你的管理器并可能加入緩存、過濾等邏輯。考慮使用langchain.indexes模塊來維護向量存儲的增量更新它提供了SQLRecordManager來追蹤文檔的哈希狀態。通過以上實踐你可以構建一個不僅功能強大而且穩定、可觀測、易于維護的向量數據管理系統使其成為你 AI 應用堅實的數據底座。