向量数据库选型不是比较 “谁的功能最多”，而是确认哪些 Feature 是业务真正需要的，再用统一数据和查询负载验证。
flowchart LR A["明确检索需求"] --> B["Feature 筛选"] B --> C["形成候选名单"] C --> D["官方文档核实"] D --> E["业务数据压测"] E --> F["选型结论"]
功能对比入口
Superlinked Vector Database Comparison 汇总了 47+ 个向量数据库，可以按许可证、开发语言、过滤、Hybrid Search、BM25、Sparse Vector、托管服务、进程内运行、分片、向量维度、量化和索引类型等字段筛选。
它适合生成候选名单，但不是性能排行榜：
- 没有统一硬件和数据集；
- 没有统一索引参数；
- 没有给出 latency、吞吐和 Recall 曲线；
- 产品字段可能使用不同口径；
- 功能和限制会随版本变化。
筛选后仍需查阅各产品当前官方文档，并使用自己的数据验证。
用 Chonkie 生成数据并在 LanceDB 建表
下面读取 1706.03762v7.txt，用 Chonkie 生成 chunk、用 Qwen/Qwen3-Embedding-8B 生成向量，再让 LanceDB 根据真实记录推断表结构：
from hashlib import sha256
from pathlib import Path
import json
import urllib.request
import lancedb
from chonkie import RecursiveChunker
EMBED_URL = "http://127.0.0.1:8001/v1/embeddings"
EMBED_MODEL = "Qwen/Qwen3-Embedding-8B"
def embed(texts, batch_size=16):
vectors = []
for start in range(0, len(texts), batch_size):
body = json.dumps(
{
"model": EMBED_MODEL,
"input": texts[start : start + batch_size],
"encoding_format": "float",
}
).encode()
request = urllib.request.Request(
EMBED_URL,
data=body,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=120) as response:
data = json.load(response)
vectors.extend(
item["embedding"]
for item in sorted(data["data"], key=lambda item: item["index"])
)
return vectors
source = Path("rag/1706.03762v7.txt")
text = source.read_text(encoding="utf-8")
pages = text.rstrip("\f").split("\f")
chunker = RecursiveChunker(tokenizer="character", chunk_size=600)
page_chunks = [
(page_number, chunk)
for page_number, page_text in enumerate(pages, start=1)
for chunk in chunker.chunk(page_text)
]
vectors = embed([chunk.text for _, chunk in page_chunks])
records = []
for chunk_index, ((page, chunk), vector) in enumerate(zip(page_chunks, vectors)):
content_hash = sha256(chunk.text.encode()).hexdigest()
records.append(
{
"id": f"{source.stem}:{chunk_index}:{content_hash[:12]}",
"text": chunk.text,
"vector": vector,
"source": "rag/1706.03762v7.pdf",
"page": page,
"chunk_index": chunk_index,
"start_index": chunk.start_index,
"end_index": chunk.end_index,
"content_hash": content_hash,
}
)
db = lancedb.connect("./data/lancedb")
table = db.create_table(
"attention_paper",
records,
mode="overwrite",
)
print("rows:", table.count_rows())
print(table.schema)
# 根据检索结果定位 PDF 页，并在抽取文本中精确复原 chunk。
hit = table.to_arrow().to_pylist()[0]
print(f"{hit['source']}#page={hit['page']}")
page_text = pages[hit["page"] - 1]
assert page_text[hit["start_index"] : hit["end_index"]] == hit["text"]运行前需要启动提供 OpenAI 兼容接口的 Qwen/Qwen3-Embedding-8B 服务。这个样例保存：
| 字段 | 用途 |
|---|---|
id | 唯一标识，用于更新和删除 |
text | Embedding、Reranker 和 LLM 的原文 |
vector | Qwen3-Embedding-8B 生成的文本向量 |
source | 原始 PDF |
page | PDF 页码 |
chunk_index | chunk 在原文中的顺序 |
start_index | chunk 在该页文本中的起始字符位置 |
end_index | chunk 在该页文本中的结束字符位置 |
content_hash | 判断正文是否变化 |
pdftotext 使用换页符 \f 分隔页面，因此可以保存真实页码。page 用于打开原 PDF，start_index 和 end_index 用于在该页的抽取文本中精确定位；复杂版面如果还需要坐标高亮，应改用能返回 bounding box 的 PDF 解析工具。按业务需要还可以增加章节、版本、语言、parent_id、tenant 和 ACL。
LanceDB 与 LadybugDB 双库存储
生产系统不一定要把向量和图放在同一个数据库。可以让 LanceDB 负责向量召回，让 LadybugDB 保存图结构，两个库通过稳定的 chunk_id 关联：
flowchart LR Q["查询"] --> E["Embedding"] E --> V["LanceDB<br/>向量召回"] V --> I["chunk_id"] I --> G["LadybugDB<br/>图关系扩展"] G --> C["命中 chunk + 关联上下文"]
这种方式有两条数据写入路径：
- ** 复用已有 Embedding 数据 **：从 LanceDB 读出
id和 metadata，在 LadybugDB 中创建对应节点，不重新计算向量； - ** 从原文重新生成 Embedding**：读取
1706.03762v7.txt，执行前一个样例的切片和embed()，把同一批records分别写入两个库。
第一种适合向量库已经存在的系统，迁移成本最低。第二种适合首次建库：切片、ID 和向量只生成一次，然后双写。无论使用哪条路径，两个库都必须使用同一个稳定 id；不要使用数据库内部行号关联。
uv add --prerelease allow ladybug从 LanceDB 复用数据建立图
前一个样例已经把 1706.03762v7.txt 的 chunk、向量和 metadata 写入 attention_paper 表。下面直接读取这些记录；vector 留在 LanceDB，不再复制到图数据库。
import ladybug as lb
import lancedb
# 从现有向量库复用 chunk 和 metadata，不重新调用 Embedding 服务。
vector_db = lancedb.connect("./data/lancedb")
table = vector_db.open_table("attention_paper")
records = table.to_arrow().to_pylist()
graph_db = lb.Database("./data/attention_graph.lbdb")
graph = lb.Connection(graph_db)
graph.execute("""
CREATE NODE TABLE Document(
source STRING PRIMARY KEY
)
""")
graph.execute("""
CREATE NODE TABLE Chunk(
id STRING PRIMARY KEY,
text STRING,
source STRING,
page INT64,
chunk_index INT64
)
""")
graph.execute("CREATE REL TABLE HAS_CHUNK(FROM Document TO Chunk)")
graph.execute("CREATE REL TABLE NEXT(FROM Chunk TO Chunk)")
for source in sorted({row["source"] for row in records}):
graph.execute(
"CREATE (:Document {source: $source})",
parameters={"source": source},
)
for index, row in enumerate(records):
graph.execute(
"""
CREATE (c:Chunk {
id: $id,
text: $text,
source: $source,
page: $page,
chunk_index: $chunk_index
})
WITH c
MATCH (d:Document {source: $source})
CREATE (d)-[:HAS_CHUNK]->(c)
""",
parameters={
key: row[key] for key in ("id", "text", "source", "page", "chunk_index")
},
)
if index > 0 and records[index - 1]["source"] == row["source"]:
graph.execute(
"""
MATCH (a:Chunk {id: $left}), (b:Chunk {id: $right})
CREATE (a)-[:NEXT]->(b)
""",
parameters={"left": records[index - 1]["id"], "right": row["id"]},
)这里建立了两类关系：
Document -[:HAS_CHUNK]-> Chunk：从文档定位其所有 chunk；Chunk -[:NEXT]-> Chunk：从命中 chunk 向前后扩展连续上下文。
先向量召回，再沿图扩展
查询时仍然只生成一次 query vector。LanceDB 返回最相似的 chunk_id，再用这些 ID 到 LadybugDB 查询相邻节点：
query_vector = embed(["What is multi-head attention?"])[0]
vector_hits = table.search(query_vector).metric("cosine").limit(5).to_list()
graph_rows = (
graph.execute(
"""
UNWIND $ids AS id
MATCH (hit:Chunk {id: id})
OPTIONAL MATCH (hit)-[:NEXT]-(neighbor:Chunk)
RETURN hit.id AS id,
collect(neighbor.text) AS adjacent_context
""",
parameters={"ids": [hit["id"] for hit in vector_hits]},
)
.rows_as_dict()
.get_all()
)
context_by_id = {row["id"]: row["adjacent_context"] for row in graph_rows}
for hit in vector_hits:
print(hit["_distance"], hit["text"])
print("adjacent:", context_by_id[hit["id"]])当前图只表达文档与顺序关系，已经可以解决 “命中了一个短 chunk，但回答需要前后文” 的问题，但它还不是知识图谱。
抽取实体和关系，组成 GraphRAG
真正的 GraphRAG 还要从 chunk 中抽取实体及实体间关系。下面继续使用已有的 OpenAI 兼容 Chat API，让模型只返回 JSON：
CHAT_URL = "http://127.0.0.1:8004/v1/chat/completions"
CHAT_MODEL = "qwen-w4a8"
def extract_graph(text):
body = json.dumps(
{
"model": CHAT_MODEL,
"messages": [
{
"role": "system",
"content": (
"Extract a knowledge graph from the text. "
"Return JSON with entities and relations. "
"Each entity has name and type. Each relation has "
"source, target and type. Use canonical English names. "
"Return at most 12 entities and 12 relations; prefer "
"concepts important to understanding the text."
),
},
{"role": "user", "content": text},
],
"response_format": {"type": "json_object"},
"chat_template_kwargs": {"enable_thinking": False},
"temperature": 0,
"max_tokens": 1200,
}
).encode()
request = urllib.request.Request(
CHAT_URL,
data=body,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=120) as response:
result = json.load(response)
return normalize_graph(json.loads(result["choices"][0]["message"]["content"]))
def normalize_graph(data):
if not isinstance(data, dict):
raise ValueError("graph extraction must return a JSON object")
entities = {}
for entity in data.get("entities", []):
if not isinstance(entity, dict):
continue
name = entity.get("name")
entity_type = entity.get("type")
if isinstance(name, str) and name.strip() and isinstance(entity_type, str):
key = name.strip().casefold()
entities[key] = {
"key": key,
"name": name.strip(),
"type": entity_type.strip(),
}
relations = []
for relation in data.get("relations", []):
if not isinstance(relation, dict):
continue
source = relation.get("source")
target = relation.get("target")
kind = relation.get("type")
if not all(isinstance(value, str) for value in (source, target, kind)):
continue
source_key = source.strip().casefold()
target_key = target.strip().casefold()
if source_key in entities and target_key in entities and kind.strip():
relations.append(
{
"source": source_key,
"target": target_key,
"kind": kind.strip(),
}
)
return entities, relations
entities, relations = normalize_graph(
{
"entities": [
{"name": "Multi-Head Attention", "type": "Method"},
{"name": "Transformer", "type": "Architecture"},
],
"relations": [
{
"source": "Multi-Head Attention",
"target": "Transformer",
"type": "PART_OF",
}
],
}
)
assert len(entities) == 2 and relations[0]["kind"] == "PART_OF"normalize_graph() 是写入数据库前的边界检查：丢弃格式错误的实体，以及端点不存在的关系。这里用规范化名称作为实体 key；如果业务需要处理缩写、同名实体或别名，应增加独立的实体消歧步骤。
写入 Entity、MENTIONS 和 RELATED_TO
LadybugDB 中增加以下结构：
graph LR D["Document"] -->|HAS_CHUNK| C["Chunk"] C -->|NEXT| C2["Chunk"] C -->|MENTIONS| E["Entity"] E -->|"RELATED_TO<br/>kind=PART_OF"| E2["Entity"]
实体间关系使用统一的 RELATED_TO 边，并把模型抽取的关系类型保存到 kind，这样不需要根据模型输出动态创建表。evidence_chunk_id 记录关系来自哪个 chunk，回答时可以回到原文核验。
graph.execute("""
CREATE NODE TABLE Entity(
key STRING PRIMARY KEY,
name STRING,
type STRING
)
""")
graph.execute("CREATE REL TABLE MENTIONS(FROM Chunk TO Entity)")
graph.execute("""
CREATE REL TABLE RELATED_TO(
FROM Entity TO Entity,
kind STRING,
evidence_chunk_id STRING
)
""")
known_entities = set()
known_mentions = set()
known_relations = set()
for chunk in records:
entities, relations = extract_graph(chunk["text"])
for entity in entities.values():
if entity["key"] not in known_entities:
graph.execute(
"""
CREATE (:Entity {
key: $key,
name: $name,
type: $type
})
""",
parameters=entity,
)
known_entities.add(entity["key"])
mention = (chunk["id"], entity["key"])
if mention not in known_mentions:
graph.execute(
"""
MATCH (c:Chunk {id: $chunk_id}), (e:Entity {key: $entity_key})
CREATE (c)-[:MENTIONS]->(e)
""",
parameters={
"chunk_id": chunk["id"],
"entity_key": entity["key"],
},
)
known_mentions.add(mention)
for relation in relations:
edge = (
relation["source"],
relation["target"],
relation["kind"],
chunk["id"],
)
if edge not in known_relations:
graph.execute(
"""
MATCH (source:Entity {key: $source}),
(target:Entity {key: $target})
CREATE (source)-[:RELATED_TO {
kind: $kind,
evidence_chunk_id: $chunk_id
}]->(target)
""",
parameters={**relation, "chunk_id": chunk["id"]},
)
known_relations.add(edge)这是离线索引阶段，只在文档新增或 content_hash 变化时运行，不应放在每次用户查询的在线链路中。
从向量命中扩展到知识图谱
在线查询仍先使用 LanceDB 召回 chunk，然后从这些 chunk 的 MENTIONS 边进入实体图：
knowledge_rows = (
graph.execute(
"""
UNWIND $ids AS id
MATCH (chunk:Chunk {id: id})-[:MENTIONS]->(entity:Entity)
OPTIONAL MATCH (entity)-[relation:RELATED_TO]-(related:Entity)
RETURN chunk.id AS chunk_id,
entity.name AS entity,
relation.kind AS relation,
related.name AS related_entity,
relation.evidence_chunk_id AS evidence_chunk_id
""",
parameters={"ids": [hit["id"] for hit in vector_hits]},
)
.rows_as_dict()
.get_all()
)
for row in knowledge_rows:
print(
row["chunk_id"],
row["entity"],
row["relation"],
row["related_entity"],
"evidence:",
row["evidence_chunk_id"],
)这样得到的上下文包含三层：
- LanceDB 语义召回的原始 chunk；
NEXT关系扩展的相邻原文；MENTIONS → Entity → RELATED_TO扩展的知识关系及其证据 chunk。
把这些结果去重、限制数量并取回证据原文后，才组成最终交给 Reranker 或 LLM 的 GraphRAG context。
双库存储的代价是需要保持一致性。最小可行做法是先用 content_hash 判断 chunk 是否变化，再以 id 对两个库执行相同的新增或删除；数据量较大或更新频繁时，再引入批处理和失败重试。
数据库概览
数据库可以按数据模型分为关系数据库、文档数据库、键值数据库和图数据库等。向量存储与检索是另一项能力，并不是与图结构对应的数据模型。
| 类型 | 主要能力 |
|---|---|
| 专用向量数据库 | 以向量存储和相似度检索为核心 |
| 普通数据库 + 向量扩展 | 同时处理常规数据和向量检索 |
| 图数据库 + 向量索引 | 同时支持图关系遍历和向量相似检索 |