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

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

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初始化成功！")

# 构建测试文档列表
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)}")

# 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("❌ 批量插入失败！")

# 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插入失败！")

# 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("❌ 增量追加文档失败！")

# 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删除失败！")

# 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删除失败！")

# 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("❌ 过滤条件删除失败！")

# 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("❌ 知识库动态维护流程失败！")