milvus向量入库

注意地方：

1. embedding的维度和milvus解析入库的维度必须是一样的

2. Milvus 插入格式要求（关键！） 以列表的列表的形式插入（单个字段）
insert_data = [
[text1, text2, …], # 所有 text 字段值组成的列表
[title_path1, title_path2, …],
[embedding1, embedding2, …] # 每个 embedding 是 float 列表
]

集合collection创建步骤

1. 创建集合名称：创建一个名为 “ai_whitebook_2025_v2” 的集合 【COLLECTION_NAME】，
2. 并设置字段【fields】：id（主键）、text（文本）、title_path（标题路径）、embedding（向量）。
3. 创建集合schema：创建集合的架构，并指定字段和数据类型。
4. 创建集合实例：指定集合名称和集合架构schema。
5. 为向量字段创建索引：使用 IVF_FLAT 索引类型和 IP 测量类型创建索引，并设置 nlist 参数为 128。
6. 插入数据：使用列表的列表格式插入数据，确保数据类型和格式正确。

1. Milvus 向量数据库数据入库脚本

"""
Milvus 向量数据库数据入库脚本

功能：
  - 从 JSON 文件加载带嵌入向量的文本 chunks
  - 自动检测向量维度
  - 创建 Milvus 集合（Collection）并配置索引
  - 批量插入数据（采用 Milvus 推荐的“按字段分组”格式）
  - 自动加载集合到内存以支持后续检索

数据格式要求（JSON）：
  [
    {
      "text": "完整文本内容",
      "title_path": "文件名 / 一级标题 / 二级标题",
      "embedding": [0.1, -0.5, ..., 0.8]  # float 列表
    },
    ...
  ]

Milvus 插入格式要求（关键！）：
  insert_data = [
    [text1, text2, ...],        # 所有 text 字段值组成的列表
    [title_path1, title_path2, ...],
    [embedding1, embedding2, ...]  # 每个 embedding 是 float 列表
  ]
  即：**按字段分组，而非按记录分组**
"""

import json
from pymilvus import (
    connections,
    FieldSchema,
    CollectionSchema,
    DataType,
    Collection,
    utility
)

# ==================== 配置区 ====================
MILVUS_HOST = "localhost"
MILVUS_PORT = "19530"
COLLECTION_NAME = "kongjiang_research_report_chunks_with_embeddings_20251003063221"
JSON_PATH = "/Users/*****/Projects/my_project_text/RAG_PRO/chunk_datas/20251003063221_空间智能研究报告chunks_with_embeddings.json"
# ==============================================


def main():
    """
    主流程：连接 → 加载数据 → 创建集合 → 插入 → 加载到内存
    """
    # 1️⃣ 连接 Milvus 服务
    print("🔌 连接 Milvus...")
    try:
        connections.connect("default", host=MILVUS_HOST, port=MILVUS_PORT)
        print("✅ Milvus 连接成功")
    except Exception as e:
        print(f"❌ Milvus 连接失败: {e}")
        return

    # 2️⃣ 加载带嵌入的 chunks 数据
    print(f"📥 加载数据: {JSON_PATH}")
    try:
        with open(JSON_PATH, 'r', encoding='utf-8') as f:
            chunks = json.load(f)
        if not chunks:
            print("❌ 数据为空")
            return
        print(f"📊 共 {len(chunks)} 个 chunks")
    except Exception as e:
        print(f"❌ 数据加载失败: {e}")
        return

    # 3️⃣ 自动检测向量维度（从第一条数据推断）
    VECTOR_DIM = len(chunks[0]["embedding"])
    print(f"📏 检测到向量维度: {VECTOR_DIM}")

    # 4️⃣ 清理旧集合并创建新集合
    try:
        # 删除已存在的同名集合（避免冲突）
        if utility.has_collection(COLLECTION_NAME):
            print(f"🗑️  删除旧集合: {COLLECTION_NAME}")
            utility.drop_collection(COLLECTION_NAME)

        # 定义字段 Schema
        fields = [
            # 主键：自增整数（auto_id=True）
            FieldSchema("id", DataType.INT64, is_primary=True, auto_id=True),
            # 文本内容：VARCHAR，max_length 必须足够大（65535 是 Milvus 当前最大值）
            FieldSchema("text", DataType.VARCHAR, max_length=65535),
            # 标题路径：用于展示上下文
            FieldSchema("title_path", DataType.VARCHAR, max_length=2000),
            # 向量字段：FLOAT_VECTOR，必须指定 dim
            FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
        ]
        
        # 创建集合 Schema 和实例
        schema = CollectionSchema(fields, description="空间智能研究报告 chunks")
        collection = Collection(COLLECTION_NAME, schema)
        
        # 为向量字段创建索引（必须！否则无法搜索）
        index_params = {
            "index_type": "IVF_FLAT",   # 精确搜索（小数据集推荐），也可选 IVF_SQ8/HNSW
            "metric_type": "IP",        # 内积（等价于余弦相似度，前提是向量已归一化）
            "params": {"nlist": 128}    # 聚类中心数，建议为 sqrt(总数据量)
        }
        collection.create_index("embedding", index_params)
        print("✅ 集合与索引创建成功")
        
    except Exception as e:
        print(f"❌ 集合操作失败: {e}")
        return

    # 5️⃣ 准备插入数据（关键：按字段分组！）
    print("🔄 准备数据...")
    texts = []
    title_paths = []
    embeddings = []
    
    for chunk in chunks:
        # 强制类型转换，避免 Milvus 类型错误
        texts.append(str(chunk["text"]))
        title_paths.append(str(chunk["title_path"]))
        # 确保 embedding 是 float 列表（避免 numpy.int/float）
        embeddings.append([float(x) for x in chunk["embedding"]])

    # 验证数据一致性
    assert len(texts) == len(title_paths) == len(embeddings), "字段长度不一致！"
    print(f"🔍 数据验证通过: {len(texts)} 条记录")

    # 6️⃣ 执行插入（Milvus 要求格式：[field1_list, field2_list, ...]）
    print("📤 正在插入数据...")
    try:
        insert_data = [texts, title_paths, embeddings]
        collection.insert(insert_data)
        
        # flush() 确保数据落盘（非必须，但推荐）
        collection.flush()
        print(f"✅ 成功插入 {collection.num_entities} 条记录")
        
    except Exception as e:
        print(f"❌ 数据插入失败: {e}")
        return

    # 7️⃣ 加载集合到内存（必须！否则无法执行搜索）
    try:
        collection.load()
        print("🎉 数据插入完成，集合已加载到内存，可执行检索")
    except Exception as e:
        print(f"⚠️  集合加载警告: {e}")


if __name__ == "__main__":
    main()

2. 通用 Milvus 入库代码模板（推荐收藏）

# milvus_ingest_template.py
import json
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility

# 配置
MILVUS_HOST = "localhost"
MILVUS_PORT = "19530"
COLLECTION_NAME = "your_collection_name"
JSON_PATH = "your_data.json"

def ingest_to_milvus():
    # 1. 连接
    connections.connect("default", host=MILVUS_HOST, port=MILVUS_PORT)
    
    # 2. 加载数据
    with open(JSON_PATH, 'r') as f:
        data = json.load(f)
    
    # 3. 检测维度
    dim = len(data[0]["vector"])
    
    # 4. 创建集合（先删后建）
    if utility.has_collection(COLLECTION_NAME):
        utility.drop_collection(COLLECTION_NAME)
    
    fields = [
        FieldSchema("id", DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema("content", DataType.VARCHAR, max_length=65535),
        FieldSchema("vector", DataType.FLOAT_VECTOR, dim=dim)
    ]
    schema = CollectionSchema(fields, "Your description")
    collection = Collection(COLLECTION_NAME, schema)
    
    # 5. 创建索引
    collection.create_index("vector", {
        "index_type": "IVF_FLAT",
        "metric_type": "IP",
        "params": {"nlist": 128}
    })
    
    # 6. 准备数据（按字段分组！）
    contents = [str(item["content"]) for item in data]
    vectors = [[float(x) for x in item["vector"]] for item in data]
    
    # 7. 插入
    collection.insert([contents, vectors])
    collection.flush()
    
    # 8. 加载
    collection.load()
    print(f"✅ Inserted {collection.num_entities} entities.")

if __name__ == "__main__":
    ingest_to_milvus()