LangChain 10：Milvus 入库

发表于 分类于
用 MilvusClient 完成数据库与 Collection 的 DDL，结合嵌入模型做插入、flush、扫描与相似度检索。

前言

文本块向量化之后，需要持久化到向量库，才能按相似度检索。
Milvus 适合大规模向量检索；本地可用 Standalone（常配合 Docker Desktop）。
本文讲 MilvusClient 的基本 DDL/DML/DQL：建库、建 Collection、写入向量、扫描与相似度搜索。
嵌入初始化见《LangChain 09：文档嵌入》；重排见《LangChain 11：检索重排序》；客服案例见《LangChain 12：客服知识库》。
下文默认 Milvus 地址为 http://localhost:19530，嵌入维数按 bge-m3 取 1024。
需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
先确保本机 Milvus Standalone 已启动，再安装客户端与嵌入依赖。

1
2
3
4
uv init langchain-milvus
cd langchain-milvus
uv venv --python 3.12
uv add pymilvus "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

在项目根 .env 中配置嵌入服务（示例为硅基流动）：

1
2
SILICONFLOW_API_KEY=你的硅基流动 API Key
SILICONFLOW_BASE_URL=https://api.siliconflow.cn/v1

实现

连接客户端

用 MilvusClient 连接本地或远程服务。

1
2
3
4
5
from pymilvus import MilvusClient
from rich import print as rprint

client = MilvusClient("http://localhost:19530")
rprint(client)

数据库操作

列出已有数据库；按需创建。
删除库前须先删光其下所有 Collection，否则会失败。

1
2
3
4
5
6
7
8
9
10
11
12
13
from pymilvus import MilvusClient
from rich import print as rprint

client = MilvusClient("http://localhost:19530")
existed = client.list_databases()
rprint(existed)

db_name = "rag_demo"
if db_name not in existed:
    client.create_database(db_name=db_name)

# 危险操作：确认无业务数据后再执行
# client.drop_database(db_name=db_name)

Collection 操作

先切换到目标库，再创建 Collection。
dimension 必须与嵌入模型输出维数一致；相似度度量示例用 COSINE。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from pymilvus import MilvusClient
from rich import print as rprint

client = MilvusClient("http://localhost:19530")
db_name = "rag_demo"
client.use_database(db_name=db_name)

rprint(client.list_collections())

collection_name = "docs"
if not client.has_collection(collection_name=collection_name):
    client.create_collection(
        collection_name=collection_name,
        dimension=1024,
        metric_type="COSINE",
    )

rprint(client.describe_collection(collection_name=collection_name))

# 重建前可删除
# client.drop_collection(collection_name=collection_name)

准备向量数据

初始化嵌入模型，对原始文本批量向量化，再封装为可插入的字典列表。
字段名需与 Collection schema 匹配；简易 create_collection(dimension=...) 默认含 id / vector，自定义标量字段在写入时一并带上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import os

from dotenv import load_dotenv
from langchain.embeddings import init_embeddings
from pymilvus import MilvusClient
from rich import print as rprint

load_dotenv(override=True)

client = MilvusClient("http://localhost:19530")
client.use_database(db_name="rag_demo")
collection_name = "docs"

embed_model = init_embeddings(
    model="openai:Pro/BAAI/bge-m3",
    api_key=os.environ["SILICONFLOW_API_KEY"],
    base_url=os.environ["SILICONFLOW_BASE_URL"],
)

texts = [
    "LangChain 是一个用于构建 LLM 应用的开发框架。",
    "Milvus 是一个适合 AI 应用的向量数据库。",
    "RAG 的核心是先检索相关知识，再让大模型生成答案。",
    "Docker Desktop 可以方便地在本地运行 Milvus Standalone。",
]

vectors = embed_model.embed_documents(texts)
rprint(len(vectors), len(vectors[0]), vectors[0][:5])

data = [
    {
        "id": i,
        "vector": vectors[i],
        "text": texts[i],
        "source": "demo",
    }
    for i in range(len(texts))
]

若默认 schema 不含 text / source，需按官方文档自定义 schema 后再插入；也可用仅含 id+vector 的最小字段做连通性验证。

写入与 flush

upsert 可插入或按主键更新。
Milvus 不一定立即可见写入结果，可用 flush 刷盘后再查统计。

1
2
3
4
5
6
7
8
insert_res = client.upsert(
    collection_name=collection_name,
    data=data,
)
rprint(insert_res)

client.flush(collection_name=collection_name)
rprint(client.get_collection_stats(collection_name=collection_name))

扫描与主键查询

全表扫描可用 query_iterator；按主键用 get。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from rich import print as rprint

iterator = client.query_iterator(
    collection_name=collection_name,
    filter="",
    output_fields=["*"],
)

i = 0
while True:
    rows = iterator.next()
    if not rows:
        break
    for row in rows:
        rprint(
            f"第{i + 1}条: id={row['id']}, "
            f"vector[:5]={row['vector'][:5]}, "
            f"text={row.get('text')}, source={row.get('source')}"
        )
        i += 1
iterator.close()

res = client.get(collection_name=collection_name, ids=[0, 1, 2])
rprint(len(res))
for i, row in enumerate(res):
    rprint(f"get[{i}]: id={row['id']}, text={row.get('text')}")

相似度检索

先把问题 embed_query，再 search。
limit 为返回条数；COSINE 下距离字段含义以当前客户端返回为准（常见为相似度分数，越大越相近，以你环境实测为准）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from rich import print as rprint

query = "什么是向量数据库？"
query_vector = embed_model.embed_query(query)

results = client.search(
    collection_name=collection_name,
    data=[query_vector],
    limit=3,
    output_fields=["text", "source", "id"],
)

for hit in results[0]:
    rprint(hit)

验证

  1. 本机 localhost:19530 可连，能 list_databases。
  2. 创建 rag_demo / docs 后，describe_collection 维数为 1024。
  3. upsert + flush 后统计行数大于 0。
  4. 用「什么是向量数据库？」检索，应优先命中含 Milvus 的句子。

总结

  1. DDL：库与 Collection；维数对齐嵌入模型。
  2. DML：embed_documents → 组装 data → upsert → flush。
  3. DQL：扫描 / 主键 get / search 相似度检索。
  4. 完整「知识库 + Agent」流程见《LangChain 12：客服知识库》。