大家好，我是工藤学编程 🦉	一个正在努力学习的小博主，期待你的关注
实战代码系列最新文章😉	C++实现图书管理系统（Qt C++ GUI界面版）
SpringBoot实战系列🐷	【SpringBoot实战系列】SpringBoot3.X 整合 MinIO 存储原生方案
分库分表	分库分表之实战-sharding-JDBC分库分表执行流程原理剖析
消息队列	深入浅出 RabbitMQ-RabbitMQ消息确认机制（ACK）
AI大模型	零基础学AI大模型之相似度Search与MMR最大边界相关搜索实战

---

前情摘要

1、零基础学AI大模型之读懂AI大模型
2、零基础学AI大模型之从0到1调用大模型API
3、零基础学AI大模型之SpringAI
4、零基础学AI大模型之AI大模型常见概念
5、零基础学AI大模型之大模型私有化部署全指南
6、零基础学AI大模型之AI大模型可视化界面
7、零基础学AI大模型之LangChain
8、零基础学AI大模型之LangChain六大核心模块与大模型IO交互链路
9、零基础学AI大模型之Prompt提示词工程
10、零基础学AI大模型之LangChain-PromptTemplate
11、零基础学AI大模型之ChatModel聊天模型与ChatPromptTemplate实战
12、零基础学AI大模型之LangChain链
13、零基础学AI大模型之Stream流式输出实战
14、零基础学AI大模型之LangChain Output Parser
15、零基础学AI大模型之解析器PydanticOutputParser
16、零基础学AI大模型之大模型的“幻觉”
17、零基础学AI大模型之RAG技术
18、零基础学AI大模型之RAG系统链路解析与Document Loaders多案例实战
19、零基础学AI大模型之LangChain PyPDFLoader实战与PDF图片提取全解析
20、零基础学AI大模型之LangChain WebBaseLoader与Docx2txtLoader实战
21、零基础学AI大模型之RAG系统链路构建：文档切割转换全解析
22、零基础学AI大模型之LangChain 文本分割器实战：CharacterTextSplitter 与 RecursiveCharacterTextSplitter 全解析
23、零基础学AI大模型之Embedding与LLM大模型对比全解析
24、零基础学AI大模型之LangChain Embedding框架全解析
25、零基础学AI大模型之嵌入模型性能优化
26、零基础学AI大模型之向量数据库介绍与技术选型思考
27、零基础学AI大模型之Milvus向量数据库全解析
28、零基础学AI大模型之Milvus核心：分区-分片-段结构全解+最佳实践
29、零基础学AI大模型之Milvus部署架构选型+Linux实战：Docker一键部署+WebUI使用
30、零基础学AI大模型之Milvus实战：Attu可视化安装+Python整合全案例
31、零基础学AI大模型之Milvus索引实战
32、零基础学AI大模型之Milvus DML实战
33、零基础学AI大模型之Milvus向量Search查询综合案例实战
33、零基础学AI大模型之新版LangChain向量数据库VectorStore设计全解析
34、零基础学AI大模型之相似度Search与MMR最大边界相关搜索实战

零基础学AI大模型之LangChain整合Milvus：新增与删除数据实战

一、实战核心目标

1. 掌握LangChain整合Milvus的3种数据新增方式（批量插入、指定ID插入、增量追加）
2. 精通Milvus数据删除的核心场景（按ID删除、批量删除、过滤删除）
3. 理解新增/删除操作的返回结果含义与状态校验方法
4. 解决新增后查不到、删除失败等常见问题
5. 实现知识库动态维护的端到端实战（新增→查询验证→删除→二次验证）

二、环境准备与基础配置

2.1 安装依赖

确保已安装LangChain核心库、Milvus集成包和嵌入模型依赖：

# 核心依赖（LangChain+Milvus）
pip install langchain-core langchain-milvus
# 嵌入模型依赖（本文用DashScope，也可替换为其他模型）
pip install dashscope

2.2 基础配置

初始化嵌入模型和Milvus向量存储实例，后续所有操作都基于此实例：

from langchain_milvus import Milvus
from langchain_core.documents import Document
from langchain_community.embeddings import DashScopeEmbeddings
from uuid import uuid4

# 1. 初始化嵌入模型（支持替换为SentenceTransformer、OpenAI等）
embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",  # 阿里云通义千问嵌入模型
    max_retries=3,  # 失败重试次数
    dashscope_api_key="你的API密钥"  # 替换为自己的密钥，无密钥可换本地模型
)

# 2. 初始化Milvus VectorStore实例
vector_store = Milvus(
    embedding=embeddings,
    connection_args={"uri": "http://192.168.229.128:19530"},  # 你的Milvus服务地址
    collection_name="langchain_milvus_crud",  # 集合名称
    drop_old=True  # 启动时删除同名集合（测试环境用，生产环境关闭）
)

print("Milvus VectorStore初始化成功！")

2.3 准备测试数据

定义10条Document测试数据，包含不同来源（tweet/news/website）的文本和元数据：

# 构建测试文档列表
documents = [
    Document(
        page_content="I had chocolate chip pancakes and scrambled eggs for breakfast this morning.",
        metadata={"source": "tweet"}
    ),
    Document(
        page_content="The weather forecast for tomorrow is cloudy and overcast, with a high of 62 degrees.",
        metadata={"source": "news"}
    ),
    Document(
        page_content="Building an exciting new project with LangChain - come check it out!",
        metadata={"source": "tweet"}
    ),
    Document(
        page_content="Robbers broke into the city bank and stole $1 million in cash.",
        metadata={"source": "news"}
    ),
    Document(
        page_content="Wow! That was an amazing movie. I can't wait to see it again.",
        metadata={"source": "tweet"}
    ),
    Document(
        page_content="Is the new iPhone worth the price? Read this review to find out.",
        metadata={"source": "website"}
    ),
    Document(
        page_content="The top 10 soccer players in the world right now.",
        metadata={"source": "website"}
    ),
    Document(
        page_content="LangGraph is the best framework for building stateful, agentic applications!",
        metadata={"source": "tweet"}
    ),
    Document(
        page_content="The stock market is down 500 points today due to fears of a recession.",
        metadata={"source": "news"}
    ),
    Document(
        page_content="I have a bad feeling I am going to get deleted :(",
        metadata={"source": "tweet"}
    )
]

print(f"准备测试文档数量：{len(documents)}")

三、数据新增实战（3种核心场景）

LangChain的Milvus集成支持多种新增方式，可根据是否指定ID、是否批量插入等场景灵活选择：

3.1 场景1：批量插入（指定自定义ID）

手动指定每条文档的唯一ID，方便后续精准删除或更新，适合有固定ID标识的场景（如文档ID、订单ID）：

# 1. 生成自定义ID（这里用1-10的字符串，实际可替换为业务ID）
ids = [str(i + 1) for i in range(len(documents))]
print(f"自定义文档ID列表：{ids}")

# 2. 批量插入文档（指定ids参数）
insert_result = vector_store.add_documents(
    documents=documents,
    ids=ids  # 绑定文档与ID的映射关系
)

# 3. 解析插入结果
print(f"\n插入操作返回结果：{insert_result}")
print(f"成功插入的文档ID：{insert_result}")  # 返回插入成功的ID列表

# 4. 验证插入结果（通过ID查询）
def verify_inserted(ids):
    """验证文档是否插入成功"""
    results = vector_store.get(ids=ids)  # 根据ID查询文档
    return len(results["documents"]) == len(ids)

if verify_inserted(ids):
    print("✅ 批量插入（指定ID）成功！")
else:
    print("❌ 批量插入失败！")

3.2 场景2：批量插入（自动生成ID）

无需手动指定ID，LangChain会自动生成UUID作为文档唯一标识，适合无固定ID的场景：

# 1. 新增1条测试文档
new_document = Document(
    page_content="LangChain整合Milvus的新增删除操作真简单！",
    metadata={"source": "blog"}
)

# 2. 插入文档（不指定ids参数，自动生成UUID）
auto_id_result = vector_store.add_documents(documents=[new_document])

# 3. 解析结果
print(f"\n自动生成ID插入结果：{auto_id_result}")
print(f"自动生成的文档ID：{auto_id_result[0]}")  # 格式为UUID字符串

# 4. 验证
if verify_inserted(auto_id_result):
    print("✅ 自动生成ID插入成功！")
else:
    print("❌ 自动生成ID插入失败！")

3.3 场景3：增量追加文档（批量）

向已存在的集合中追加多条文档，适合知识库增量更新（如每日新增文档入库）：

# 1. 准备增量文档
incremental_docs = [
    Document(
        page_content="Milvus支持高并发的向量插入和查询操作",
        metadata={"source": "tech"}
    ),
    Document(
        page_content="RAG系统的知识库需要定期增量更新和去重",
        metadata={"source": "tech"}
    )
]

# 2. 增量追加（自动生成ID）
increment_result = vector_store.add_documents(documents=incremental_docs)

print(f"\n增量追加文档ID：{increment_result}")
if verify_inserted(increment_result):
    print("✅ 增量追加文档成功！")
else:
    print("❌ 增量追加文档失败！")

3.4 新增操作核心说明

操作方式	适用场景	关键参数	优势
指定自定义ID批量插入	有固定业务ID的场景	documents、ids	便于后续精准操作（删除/更新）
自动生成ID插入	无固定ID的场景	documents	无需手动管理ID，简化开发
增量追加批量插入	知识库增量更新	documents	支持批量导入，效率高

四、数据删除实战（3种核心场景）

Milvus的删除操作主要支持“按ID删除”和“按过滤条件删除”，LangChain封装后使用更简洁：

4.1 场景1：按单个ID删除

根据文档的唯一ID删除指定文档，是最常用的删除场景（如删除错误文档）：

# 1. 删除之前插入的ID为"1"的文档（场景1中插入）
delete_single_result = vector_store.delete(ids=["1"])

# 2. 解析删除结果（Milvus返回的操作统计）
print(f"\n单个ID删除结果：{delete_single_result}")
print(f"删除成功数量：{delete_single_result.delete_count}")  # 成功删除返回1，失败返回0

# 3. 验证删除（查询已删除的ID，应返回空）
deleted_check = vector_store.get(ids=["1"])
if len(deleted_check["documents"]) == 0:
    print("✅ 单个ID删除成功！")
else:
    print("❌ 单个ID删除失败！")

4.2 场景2：批量删除（多个ID）

一次性删除多个文档，适合批量清理过期数据（如批量删除某类失效文档）：

# 1. 准备要删除的ID列表（场景1中的ID"2"和"3"）
batch_delete_ids = ["2", "3"]

# 2. 批量删除
delete_batch_result = vector_store.delete(ids=batch_delete_ids)

print(f"\n批量删除结果：{delete_batch_result}")
print(f"批量删除成功数量：{delete_batch_result.delete_count}")

# 3. 验证
batch_check = vector_store.get(ids=batch_delete_ids)
if len(batch_check["documents"]) == 0:
    print("✅ 批量ID删除成功！")
else:
    print("❌ 批量ID删除失败！")

4.3 场景3：按过滤条件删除（元数据过滤）

根据元数据条件批量删除文档，适合按分类清理数据（如删除所有来源为"tweet"的文档）：

# 1. 按元数据过滤删除（删除source="tweet"的所有文档）
delete_filter_result = vector_store.delete(
    filter="source == 'tweet'"  # 过滤表达式，语法与Milvus一致
)

print(f"\n过滤条件删除结果：{delete_filter_result}")
print(f"过滤删除成功数量：{delete_filter_result.delete_count}")

# 2. 验证（查询source="tweet"的文档，应返回空）
filter_check = vector_store.similarity_search(
    query="任意查询",
    k=10,
    filter="source == 'tweet'"
)
if len(filter_check) == 0:
    print("✅ 过滤条件删除成功！")
else:
    print("❌ 过滤条件删除失败！")

4.4 删除操作核心说明

• 返回结果解读：delete()方法返回Milvus的操作响应对象，关键字段包括delete_count（成功删除数量）、success_count（操作成功次数）、err_count（错误次数）。
• 删除不可逆：删除操作会直接删除向量和对应的元数据，无法恢复，生产环境需谨慎操作（建议先备份）。
• 过滤条件支持：支持==、!=、>、<、IN等运算符，多条件用AND/OR连接（如source == 'news' AND publish_time < '2024-01-01'）。

五、端到端实战：知识库动态维护流程

整合新增、查询、删除操作，实现完整的知识库维护流程：

# 1. 新增一批业务文档
business_docs = [
    Document(
        page_content="企业知识库需定期清理过期文档",
        metadata={"source": "business", "expire_time": "2025-12-31"}
    ),
    Document(
        page_content="LangChain+Milvus是RAG系统的优选方案",
        metadata={"source": "business", "expire_time": "2026-12-31"}
    )
]
business_ids = vector_store.add_documents(documents=business_docs)
print(f"\n新增业务文档ID：{business_ids}")

# 2. 查询验证（确认新增成功）
query_result = vector_store.similarity_search(
    query="RAG系统优选方案",
    k=2,
    filter="source == 'business'"
)
print(f"\n查询到的业务文档数量：{len(query_result)}")

# 3. 删除过期文档（模拟删除expire_time < '2026-01-01'的文档）
expire_delete_result = vector_store.delete(
    filter="source == 'business' AND expire_time < '2026-01-01'"
)
print(f"\n删除过期文档数量：{expire_delete_result.delete_count}")

# 4. 最终验证（确认过期文档已删除）
final_check = vector_store.similarity_search(
    query="RAG系统优选方案",
    k=2,
    filter="source == 'business' AND expire_time < '2026-01-01'"
)
if len(final_check) == 0:
    print("✅ 知识库动态维护流程（新增→查询→删除→验证）完成！")
else:
    print("❌ 知识库动态维护流程失败！")

六、关键注意事项与避坑指南

1. ID唯一性：插入时指定的ID必须唯一，重复ID会导致插入失败（Milvus主键冲突）。
2. 嵌入模型一致性：新增和查询时必须使用同一嵌入模型，否则向量维度不匹配，无法检索到结果。
3. 删除操作谨慎：生产环境删除前建议先通过get()或similarity_search()验证目标文档，避免误删。
4. 批量操作性能：批量插入/删除时，单次数据量建议控制在1000条以内，过大可能导致超时（可分批次处理）。
5. 过滤条件语法：delete()的filter参数语法与Milvus一致，字符串需用单引号包裹，多条件用AND/OR连接。
6. 集合加载状态：删除操作无需手动加载集合（LangChain自动处理），但需确保Milvus服务正常运行。

七、常见问题排查

1. 新增后查不到文档？

   • 检查嵌入模型是否一致，是否使用了不同的模型或模型版本。
   • 验证插入时的ids是否与查询时的ID匹配，或是否误删。
   • 检查Milvus集合是否存在（可通过vector_store._collection.exists()验证）。

2. 删除操作返回delete_count=0？

   • 检查指定的ID是否存在（通过get()方法验证）。
   • 过滤条件是否有误，导致未匹配到任何文档。
   • 确认Milvus服务是否正常，网络是否通畅。

3. 插入时提示“主键冲突”？

   • 确保自定义ID不重复，或关闭drop_old=True后清理重复ID。
   • 改用自动生成ID的方式（不指定ids参数）。

如果本文对你有帮助，欢迎点赞+关注，后续会持续输出AI大模型与LangChain的实战内容～

我的博客即将同步至腾讯云开发者社区，邀请大家一同入驻：https://cloud.tencent.com/developer/support-plan?invite_code=adofqgu8vvn

---