9.3 向量索引构建示例
文档进入向量库前，通常需要先经过清洗、切分、打标签和嵌入处理，最后写入索引。下面展示一个最简流程的实现，真实生产环境中可替换为 Milvus 或 Qdrant 的 SDK。
from dataclasses import dataclass
from typing import Iterable
import hashlib
@dataclass
class Chunk:
chunk_id: str
text: str
metadata: dict
def chunk_document(doc_id: str, title: str, text: str, source_type: str) -> list[Chunk]:
# 按段落切分并过滤空行
parts = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks = []
for i, part in enumerate(parts):
# 使用哈希生成唯一 ID，避免重复
cid = hashlib.md5(f"{doc_id}-{i}-{part[:100]}".encode()).hexdigest()
chunks.append(
Chunk(
chunk_id=cid,
text=f"[{title}]\n{part}",
metadata={"doc_id": doc_id, "source_type": source_type, "seq": i},
)
)
chunks
():
rows = []
c chunks:
vec = embed_fn(c.text)
rows.append({: c.chunk_id, : vec, : c.metadata | {: c.text}})
store.upsert(rows)
