向量搜索
向量搜索（Vector Search）是 Neo4j 5.15+ 引入的重要功能，它使得 Neo4j 能够支持语义搜索和 AI 应用。通过向量搜索，你可以基于数据的语义相似性来查找节点和关系，而不仅仅是精确匹配。这在构建推荐系统、语义搜索引擎、RAG（检索增强生成）应用等场景中非常有用。
什么是向量搜索
从传统搜索到语义搜索
传统的数据库搜索基于精确匹配或模式匹配。例如，搜索标题包含"数据库"的文章，使用的是字符串匹配：
-- 传统搜索：精确匹配或包含匹配
MATCH (a:Article)
WHERE a.title CONTAINS '数据库'
RETURN a
这种方式的问题在于：如果一篇文章讨论的是"数据存储技术"，但没有明确提到"数据库"这个词，传统搜索就找不到它，尽管它们的语义非常接近。
向量搜索通过将文本转换为向量（一组数字），然后计算向量之间的相似度来解决这个问题。语义相近的内容，其向量在向量空间中的距离也更近。
向量嵌入（Vector Embeddings）
向量嵌入是将数据（文本、图像、音频等）转换为数值向量的过程。这个过程通常由机器学习模型完成，如 OpenAI 的 text-embedding 模型、Sentence Transformers 等。
例如，文本"数据库管理系统"可能被转换为：
[0.12, -0.34, 0.56, 0.78, -0.23, ...]
这个向量在高维空间中代表了文本的语义。语义相似的文本，其向量之间的距离更近。
向量搜索的工作原理
向量搜索的核心是相似度计算。给定一个查询向量，系统会找出与之最相似的数据向量。常用的相似度度量方法有：
余弦相似度（Cosine Similarity）：计算两个向量夹角的余弦值，取值范围 [-1, 1]，值越大表示越相似。适用于文本嵌入，因为文本长度不影响相似度。
欧几里得距离（Euclidean Distance）：计算两个向量之间的直线距离，值越小表示越相似。适用于需要考虑向量幅度的场景。
点积（Dot Product）：计算两个向量的内积，值越大表示越相似。在归一化向量上与余弦相似度等价。
创建向量索引
在使用向量搜索之前，需要先创建向量索引。向量索引使用近似最近邻（Approximate Nearest Neighbor，ANN）算法，能够高效处理大规模向量数据。
基本语法
CREATE VECTOR INDEX [index_name] [IF NOT EXISTS]
FOR (n:LabelName) ON (n.propertyName)
OPTIONS {
indexConfig: {
`vector.dimensions`: dimension_number,
`vector.similarity_function`: 'similarity_function'
}
}
参数说明：
index_name：索引名称，建议指定以便管理LabelName：节点标签propertyName：存储向量的属性名vector.dimensions：向量维度，必须与嵌入模型的输出维度一致vector.similarity_function：相似度函数，可选值为cosine、euclidean
创建示例
-- 为电影创建向量索引（假设使用 OpenAI text-embedding-3-small 模型，1536 维）
CREATE VECTOR INDEX movie_embedding_index IF NOT EXISTS
FOR (m:Movie) ON (m.embedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 1536,
`vector.similarity_function`: 'cosine'
}
}
-- 为商品描述创建向量索引
CREATE VECTOR INDEX product_desc_vector IF NOT EXISTS
FOR (p:Product) ON (p.descriptionEmbedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 768,
`vector.similarity_function`: 'cosine'
}
}
关系向量索引
Neo4j 5.18+ 支持为关系创建向量索引：
-- 为关系的向量属性创建索引
CREATE VECTOR INDEX relation_embedding_index IF NOT EXISTS
FOR ()-[r:SIMILAR_TO]-() ON (r.embedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 256,
`vector.similarity_function`: 'euclidean'
}
}
查看向量索引
-- 列出所有索引
SHOW INDEXES
WHERE type = 'VECTOR'
-- 查看特定索引详情
SHOW INDEXES YIELD name, type, options
WHERE name = 'movie_embedding_index'
RETURN name, type, options
删除向量索引
DROP INDEX movie_embedding_index IF EXISTS
存储向量数据
使用列表存储向量
在 Neo4j 社区版中，向量可以作为浮点数列表存储：
-- 创建带有向量属性的节点
CREATE (m:Movie {
title: '盗梦空间',
year: 2010,
embedding: [0.12, -0.34, 0.56, ...] // 1536 维向量
})
-- 批量创建带向量的节点
UNWIND [
{title: '盗梦空间', embedding: [0.12, -0.34, ...]},
{title: '星际穿越', embedding: [0.23, -0.45, ...]},
{title: '黑客帝国', embedding: [0.34, -0.56, ...]}
] AS movie
CREATE (m:Movie {title: movie.title, embedding: movie.embedding})
使用 VECTOR 类型（企业版 2025.10+）
Neo4j 企业版 2025.10+ 引入了原生 VECTOR 类型，提供更高效的存储：
-- 使用 vector() 函数创建 VECTOR 类型
CREATE (m:Movie {
title: '盗梦空间',
embedding: vector([0.12, -0.34, 0.56], 3, FLOAT32)
})
-- 指定维度和数据类型
CREATE (m:Movie {
title: '盗梦空间',
embedding: vector($embeddingList, 1536, FLOAT32)
})
VECTOR 类型的优势：
- 更高效的存储（指定精度后）
- 支持多种坐标类型：INT8、INT16、INT32、INT64、FLOAT32、FLOAT64
- 维度最大支持 4096
向量搜索查询
使用 db.index.vector.queryNodes
向量搜索通过 db.index.vector.queryNodes 存储过程执行：
-- 基本向量搜索
CALL db.index.vector.queryNodes('movie_embedding_index', 10, $queryVector)
YIELD node, score
RETURN node.title AS title, score
ORDER BY score DESC
参数说明：
- 第一个参数：索引名称
- 第二个参数：返回结果数量（k 值）
- 第三个参数：查询向量
完整示例：电影推荐
假设我们有一个电影数据库，每部电影都有基于剧情描述生成的向量嵌入：
-- 1. 创建向量索引
CREATE VECTOR INDEX movie_plot_vector IF NOT EXISTS
FOR (m:Movie) ON (m.plotEmbedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 1536,
`vector.similarity_function`: 'cosine'
}
}
-- 2. 添加电影数据（嵌入向量需要通过 ML 模型生成）
CREATE (m1:Movie {
title: '盗梦空间',
plot: '一个熟练的盗梦者被委托执行一项看似不可能的任务...',
plotEmbedding: $inceptionEmbedding
})
CREATE (m2:Movie {
title: '星际穿越',
plot: '一组探险家穿越虫洞，为人类寻找新家园...',
plotEmbedding: $interstellarEmbedding
})
-- 3. 执行向量搜索（查找与查询最相似的电影）
-- queryVector 是用户查询文本的嵌入向量
CALL db.index.vector.queryNodes('movie_plot_vector', 5, $queryVector)
YIELD node, score
RETURN node.title AS title, node.plot AS plot, score
ORDER BY score DESC
结合图关系的搜索
向量搜索的强大之处在于可以与图的关系统合使用：
-- 查找与查询相似的电影，并返回其演员信息
CALL db.index.vector.queryNodes('movie_plot_vector', 10, $queryVector)
YIELD node AS movie, score
MATCH (movie)<-[:ACTED_IN]-(actor:Person)
RETURN movie.title AS movie, score, collect(actor.name) AS actors
ORDER BY score DESC
-- 基于用户偏好进行个性化推荐
MATCH (u:User {id: $userId})-[:RATED]->(rated:Movie)-[:HAS_SIMILAR_PLOT]->(similar:Movie)
WHERE NOT (u)-[:RATED]->(similar)
WITH u, similar, count(rated) AS relevance
ORDER BY relevance DESC
LIMIT 10
MATCH (similar)
CALL db.index.vector.queryNodes('movie_plot_vector', 10, u.preferenceVector)
YIELD node, score
WHERE node = similar
RETURN similar.title, relevance, score
向量相似度函数
除了使用向量索引，Neo4j 还提供了向量相似度函数，可以在没有索引的情况下计算向量相似度。这在处理小数据集或一次性查询时很有用。
余弦相似度
-- 计算两个向量的余弦相似度
MATCH (a:Movie {title: '盗梦空间'})
MATCH (b:Movie {title: '星际穿越'})
WITH a.plotEmbedding AS vecA, b.plotEmbedding AS vecB
RETURN vector.similarity.cosine(vecA, vecB) AS similarity
欧几里得距离
-- 计算两个向量的欧几里得距离
MATCH (a:Movie {title: '盗梦空间'})
MATCH (b:Movie {title: '星际穿越'})
WITH a.plotEmbedding AS vecA, b.plotEmbedding AS vecB
RETURN vector.similarity.euclidean(vecA, vecB) AS distance
点积
-- 计算两个向量的点积
MATCH (a:Movie {title: '盗梦空间'})
MATCH (b:Movie {title: '星际穿越'})
WITH a.plotEmbedding AS vecA, b.plotEmbedding AS vecB
RETURN vector.similarity.dot_product(vecA, vecB) AS dotProduct
暴力搜索（无索引）
-- 不使用索引，直接计算相似度（适合小数据集）
MATCH (m:Movie)
WHERE m.plotEmbedding IS NOT NULL
WITH m, vector.similarity.cosine(m.plotEmbedding, $queryVector) AS score
ORDER BY score DESC
LIMIT 10
RETURN m.title AS title, score
注意：暴力搜索需要计算查询向量与所有向量的相似度，对大数据集性能较差。建议始终使用向量索引。
实战案例：构建语义搜索引擎
下面我们构建一个完整的语义搜索系统，展示向量搜索在实际应用中的使用。
场景描述
假设我们有一个技术文档库，用户可以通过自然语言查询找到相关文档。
数据模型
-- 创建文档节点
CREATE (d:Document {
id: 'doc001',
title: 'Neo4j 入门指南',
content: 'Neo4j 是一个原生图数据库...',
embedding: $embedding1
})
CREATE (d2:Document {
id: 'doc002',
title: 'Cypher 查询语言详解',
content: 'Cypher 是 Neo4j 的声明式查询语言...',
embedding: $embedding2
})
-- 文档之间的引用关系
CREATE (d1:Document {id: 'doc001'})-[:REFERENCES]->(d2:Document {id: 'doc002'})
创建索引
CREATE VECTOR INDEX document_content_vector IF NOT EXISTS
FOR (d:Document) ON (d.embedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 1536,
`vector.similarity_function`: 'cosine'
}
}
搜索 API 示例
Python 实现：
from neo4j import GraphDatabase
from openai import OpenAI
class SemanticSearchEngine:
def __init__(self, neo4j_uri, neo4j_user, neo4j_password, openai_key):
self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))
self.openai_client = OpenAI(api_key=openai_key)
def get_embedding(self, text):
"""将文本转换为嵌入向量"""
response = self.openai_client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def search(self, query, top_k=10):
"""执行语义搜索"""
query_embedding = self.get_embedding(query)
with self.driver.session() as session:
result = session.run("""
CALL db.index.vector.queryNodes('document_content_vector', $top_k, $embedding)
YIELD node, score
RETURN node.id AS id, node.title AS title,
node.content AS content, score
ORDER BY score DESC
""", top_k=top_k, embedding=query_embedding)
return [record.data() for record in result]
def close(self):
self.driver.close()
# 使用示例
engine = SemanticSearchEngine(
"bolt://localhost:7687",
"neo4j",
"password",
"your-openai-key"
)
results = engine.search("如何使用图数据库进行社交网络分析")
for r in results:
print(f"{r['title']}: {r['score']:.3f}")
engine.close()
混合搜索
结合向量搜索和全文搜索，提供更准确的搜索结果：
-- 向量搜索结果
CALL db.index.vector.queryNodes('document_content_vector', 20, $queryVector)
YIELD node AS vecNode, score AS vecScore
-- 全文搜索结果
CALL db.index.fulltext.queryNodes('document_fulltext', $queryText)
YIELD node AS ftNode, score AS ftScore
-- 合并结果（假设相同节点的分数相加）
WITH collect(DISTINCT vecNode) + collect(DISTINCT ftNode) AS allNodes
UNWIND allNodes AS node
OPTIONAL MATCH (node)
CALL db.index.vector.queryNodes('document_content_vector', 1, $queryVector)
YIELD node AS vNode, score AS vScore
WHERE vNode = node
WITH node, vScore AS vecScore
OPTIONAL CALL db.index.fulltext.queryNodes('document_fulltext', $queryText)
YIELD node AS fNode, score AS fScore
WHERE fNode = node
WITH node, coalesce(vecScore, 0) * 0.7 + coalesce(fScore, 0) * 0.3 AS combinedScore
ORDER BY combinedScore DESC
LIMIT 10
RETURN node.title, node.content, combinedScore
嵌入模型选择指南
选择合适的嵌入模型是构建高质量向量搜索应用的关键。不同的模型有不同的特点和适用场景。
主流嵌入模型对比
| 模型 | 维度 | 提供商 | 特点 | 适用场景 |
|---|---|---|---|---|
| text-embedding-3-small | 1536 | OpenAI | 高性价比，多语言支持 | 通用文本搜索、推荐 |
| text-embedding-3-large | 3072 | OpenAI | 更高精度，多语言支持 | 高精度语义搜索 |
| text-embedding-ada-002 | 1536 | OpenAI | 稳定可靠，广泛使用 | 遗留项目迁移 |
| all-MiniLM-L6-v2 | 384 | Sentence Transformers | 轻量级，开源免费 | 本地部署、资源受限场景 |
| all-mpnet-base-v2 | 768 | Sentence Transformers | 高质量，开源免费 | 中英文混合、离线场景 |
| bge-large-en-v1.5 | 1024 | BGE | 英文效果好，开源 | 英文文档搜索 |
| bge-large-zh-v1.5 | 1024 | BGE | 中文效果好，开源 | 中文文档搜索 |
| embed-english-v3.0 | 1024 | Cohere | 英文效果好 | 英文语义搜索 |
| embed-multilingual-v3.0 | 1024 | Cohere | 多语言支持 | 多语言应用 |
模型选择考虑因素
1. 精度需求
精度要求高的场景（如法律文档检索、医疗知识库）应选择更高维度的模型：
# 高精度场景：使用 OpenAI large 模型
response = openai_client.embeddings.create(
model="text-embedding-3-large", # 3072 维
input=text
)
# 一般场景：使用 small 模型即可
response = openai_client.embeddings.create(
model="text-embedding-3-small", # 1536 维
input=text
)
2. 语言支持
中文为主的应用建议使用专门优化的中文模型：
# 方案一：使用 OpenAI 多语言模型（需要 API 调用）
from openai import OpenAI
client = OpenAI()
embedding = client.embeddings.create(
model="text-embedding-3-small",
input="这是一段中文文本"
).data[0].embedding
# 方案二：使用本地中文模型（开源免费）
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
embedding = model.encode("这是一段中文文本")
3. 成本考虑
| 方案 | 成本 | 延迟 | 隐私 |
|---|---|---|---|
| OpenAI API | 按使用量付费 | 网络延迟 | 数据传输到第三方 |
| Cohere API | 按使用量付费 | 网络延迟 | 数据传输到第三方 |
| 本地模型 | 免费 | 本地计算 | 数据本地处理 |
4. 维度与存储成本
向量维度直接影响存储和计算成本：
-- 1536 维向量存储大小估算
-- 每个浮点数 4 字节
-- 1536 * 4 = 6144 字节 ≈ 6KB / 向量
-- 100 万条记录的存储需求
-- 1000000 * 6KB ≈ 6GB（仅向量数据）
-- 选择低维度模型可以节省存储
-- 384 维：100 万条记录仅需要 1.5GB
本地模型部署示例
对于需要数据隐私或降低成本的应用，可以使用开源模型本地部署：
from sentence_transformers import SentenceTransformer
from neo4j import GraphDatabase
class LocalEmbeddingService:
def __init__(self, neo4j_uri, neo4j_auth, model_name='all-MiniLM-L6-v2'):
# 加载本地模型（首次会自动下载）
self.model = SentenceTransformer(model_name)
self.driver = GraphDatabase.driver(neo4j_uri, auth=neo4j_auth)
def encode(self, texts):
"""将文本列表转换为向量"""
return self.model.encode(texts, show_progress_bar=True)
def index_documents(self, documents):
"""批量索引文档"""
texts = [doc['content'] for doc in documents]
embeddings = self.encode(texts)
with self.driver.session() as session:
for doc, embedding in zip(documents, embeddings):
session.run("""
MERGE (d:Document {id: $id})
SET d.title = $title,
d.content = $content,
d.embedding = $embedding
""",
id=doc['id'],
title=doc['title'],
content=doc['content'],
embedding=embedding.tolist()
)
def search(self, query, top_k=10):
"""向量搜索"""
query_embedding = self.encode([query])[0]
with self.driver.session() as session:
result = session.run("""
CALL db.index.vector.queryNodes('document_embedding', $k, $embedding)
YIELD node, score
RETURN node.id AS id, node.title AS title,
node.content AS content, score
ORDER BY score DESC
""", k=top_k, embedding=query_embedding.tolist())
return [record.data() for record in result]
# 使用示例
service = LocalEmbeddingService(
"bolt://localhost:7687",
("neo4j", "password"),
"BAAI/bge-large-zh-v1.5" # 中文场景推荐
)
# 索引文档
documents = [
{"id": "d001", "title": "Neo4j 简介", "content": "Neo4j 是一个高性能的图数据库..."},
{"id": "d002", "title": "Cypher 查询语言", "content": "Cypher 是 Neo4j 的声明式查询语言..."}
]
service.index_documents(documents)
# 搜索
results = service.search("什么是图数据库")
for r in results:
print(f"{r['title']}: {r['score']:.3f}")
RAG 应用实战
RAG（Retrieval-Augmented Generation，检索增强生成）是将向量检索与大语言模型结合的技术架构，可以显著提升 AI 应用的回答质量和准确性。
RAG 架构概述
RAG 的核心思想是：在生成回答之前，先从知识库中检索相关文档，然后将检索结果作为上下文提供给大语言模型。这种方法可以有效解决大语言模型的知识局限性问题。
用户问题 → 向量化 → 向量搜索 → 检索相关文档 → 构建提示词 → LLM 生成回答
构建 RAG 系统
下面是一个完整的 RAG 系统实现，结合 Neo4j 向量搜索和 OpenAI GPT 模型：
from neo4j import GraphDatabase
from openai import OpenAI
from typing import List, Dict
class Neo4jRAGSystem:
def __init__(self, neo4j_uri, neo4j_user, neo4j_password, openai_key):
self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))
self.openai_client = OpenAI(api_key=openai_key)
self.embedding_model = "text-embedding-3-small"
self.chat_model = "gpt-4o-mini"
def create_vector_index(self):
"""创建向量索引（仅需执行一次）"""
with self.driver.session() as session:
session.run("""
CREATE VECTOR INDEX document_embedding IF NOT EXISTS
FOR (d:Document) ON (d.embedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 1536,
`vector.similarity_function`: 'cosine'
}
}
""")
def get_embedding(self, text: str) -> List[float]:
"""获取文本的向量表示"""
response = self.openai_client.embeddings.create(
model=self.embedding_model,
input=text
)
return response.data[0].embedding
def index_document(self, doc_id: str, title: str, content: str, metadata: Dict = None):
"""索引单个文档"""
embedding = self.get_embedding(content)
with self.driver.session() as session:
session.run("""
MERGE (d:Document {id: $doc_id})
SET d.title = $title,
d.content = $content,
d.embedding = $embedding,
d.metadata = $metadata,
d.indexedAt = datetime()
""",
doc_id=doc_id,
title=title,
content=content,
embedding=embedding,
metadata=metadata or {}
)
def retrieve_context(self, query: str, top_k: int = 5) -> List[Dict]:
"""检索相关文档"""
query_embedding = self.get_embedding(query)
with self.driver.session() as session:
result = session.run("""
CALL db.index.vector.queryNodes('document_embedding', $top_k, $embedding)
YIELD node, score
RETURN node.id AS id,
node.title AS title,
node.content AS content,
node.metadata AS metadata,
score
ORDER BY score DESC
""", top_k=top_k, embedding=query_embedding)
return [record.data() for record in result]
def build_prompt(self, query: str, contexts: List[Dict]) -> str:
"""构建包含检索上下文的提示词"""
context_text = "\n\n".join([
f"【文档 {i+1}】{ctx['title']}\n{ctx['content']}"
for i, ctx in enumerate(contexts)
])
prompt = f"""你是一个专业的技术助手。请根据以下参考文档回答用户的问题。
如果参考文档中没有相关信息，请明确告知用户。
参考文档：
{context_text}
用户问题：{query}
请基于参考文档提供准确、详细的回答："""
return prompt
def generate_answer(self, prompt: str) -> str:
"""使用 LLM 生成回答"""
response = self.openai_client.chat.completions.create(
model=self.chat_model,
messages=[
{"role": "system", "content": "你是一个专业的技术助手，擅长根据提供的上下文回答问题。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=1000
)
return response.choices[0].message.content
def answer(self, query: str, top_k: int = 5) -> Dict:
"""完整的 RAG 问答流程"""
# 1. 检索相关文档
contexts = self.retrieve_context(query, top_k)
if not contexts:
return {
"answer": "抱歉，我在知识库中没有找到相关信息来回答您的问题。",
"sources": [],
"confidence": 0
}
# 2. 构建提示词
prompt = self.build_prompt(query, contexts)
# 3. 生成回答
answer = self.generate_answer(prompt)
# 4. 返回结果
return {
"answer": answer,
"sources": [
{"title": ctx["title"], "score": ctx["score"]}
for ctx in contexts
],
"confidence": contexts[0]["score"] if contexts else 0
}
def close(self):
self.driver.close()
# 使用示例
def main():
# 初始化 RAG 系统
rag = Neo4jRAGSystem(
neo4j_uri="bolt://localhost:7687",
neo4j_user="neo4j",
neo4j_password="password",
openai_key="your-openai-api-key"
)
# 创建向量索引
rag.create_vector_index()
# 索引文档
documents = [
{
"id": "neo4j-intro",
"title": "Neo4j 图数据库简介",
"content": """
Neo4j 是目前最流行的原生图数据库，专为处理高度关联的数据而设计。
与传统关系型数据库不同，Neo4j 以图的形式存储数据，能够高效地处理复杂的关系查询。
Neo4j 使用 Cypher 作为查询语言，这是一种声明式的图查询语言，语法简洁易懂。
"""
},
{
"id": "cypher-basics",
"title": "Cypher 查询语言基础",
"content": """
Cypher 是 Neo4j 的声明式查询语言，专为图数据操作而设计。
它使用 ASCII 艺术风格的语法来表示图结构，例如 (node)-[:RELATIONSHIP]->(other)。
常用的 Cypher 子句包括 CREATE（创建）、MATCH（查询）、SET（更新）、DELETE（删除）等。
"""
},
{
"id": "graph-algorithms",
"title": "Neo4j 图算法应用",
"content": """
Neo4j Graph Data Science (GDS) 库提供了丰富的图算法，包括：
- 中心性算法：PageRank、度中心性、中介中心性
- 社区发现算法：Louvain、标签传播
- 路径查找算法：最短路径、A* 算法
- 相似度算法：Jaccard、余弦相似度
"""
}
]
for doc in documents:
rag.index_document(doc["id"], doc["title"], doc["content"])
# 问答测试
questions = [
"什么是 Neo4j？它有什么特点？",
"如何使用 Cypher 创建节点？",
"Neo4j 支持哪些图算法？"
]
for question in questions:
print(f"\n问题: {question}")
result = rag.answer(question)
print(f"回答: {result['answer']}")
print(f"参考来源: {[s['title'] for s in result['sources']]}")
rag.close()
if __name__ == "__main__":
main()
结合图关系的 RAG
Neo4j 的独特优势在于可以将向量搜索与图的关系统合，实现更智能的检索：
def retrieve_with_graph_context(self, query: str, top_k: int = 3) -> List[Dict]:
"""检索文档并利用图关系扩展上下文"""
query_embedding = self.get_embedding(query)
with self.driver.session() as session:
# 1. 向量搜索找到相关文档
# 2. 通过图关系找到关联文档
result = session.run("""
// 首先通过向量搜索找到相关文档
CALL db.index.vector.queryNodes('document_embedding', $top_k, $embedding)
YIELD node AS primary_doc, score
// 找到与主要文档相关的其他文档
OPTIONAL MATCH (primary_doc)-[:REFERENCES|RELATED_TO]-(related:Document)
// 合并结果
WITH primary_doc, score, collect(DISTINCT related) AS related_docs
RETURN primary_doc.id AS id,
primary_doc.title AS title,
primary_doc.content AS content,
score,
[doc IN related_docs | {title: doc.title, content: doc.content}] AS related
""", top_k=top_k, embedding=query_embedding)
return [record.data() for record in result]
def build_graph_enhanced_prompt(self, query: str, contexts: List[Dict]) -> str:
"""构建包含图关系的增强提示词"""
context_parts = []
for i, ctx in enumerate(contexts):
part = f"【主要文档 {i+1}】{ctx['title']}\n{ctx['content']}"
# 添加关联文档信息
if ctx.get('related'):
related_text = "\n".join([
f" - 相关: {r['title']}"
for r in ctx['related'][:2] # 限制关联文档数量
])
part += f"\n\n关联文档：\n{related_text}"
context_parts.append(part)
context_text = "\n\n---\n\n".join(context_parts)
return f"""你是一个专业的技术助手。请根据以下参考文档回答用户的问题。
注意文档之间的关联关系，可以综合多个文档的信息进行回答。
参考文档：
{context_text}
用户问题：{query}
请提供详细、准确的回答："""
RAG 系统优化建议
1. 文档切片策略
长文档需要切分成适当大小的片段：
def chunk_document(content: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""将文档切分成重叠的片段"""
words = content.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
chunks.append(chunk)
return chunks
# 索引时处理
for doc in documents:
chunks = chunk_document(doc['content'])
for i, chunk in enumerate(chunks):
rag.index_document(
doc_id=f"{doc['id']}-chunk-{i}",
title=f"{doc['title']} (第{i+1}部分)",
content=chunk,
metadata={"parent_id": doc['id'], "chunk_index": i}
)
2. 重排序（Reranking）
对初步检索结果进行重排序，提高相关性：
def rerank_results(query: str, results: List[Dict], top_n: int = 5) -> List[Dict]:
"""使用交叉编码器重排序"""
from sentence_transformers import CrossEncoder
# 使用交叉编码器计算精确相关性分数
model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
pairs = [(query, r['content']) for r in results]
scores = model.predict(pairs)
# 按新分数排序
ranked = sorted(
zip(results, scores),
key=lambda x: x[1],
reverse=True
)
return [
{**r, 'rerank_score': s}
for r, s in ranked[:top_n]
]
3. 混合检索
结合关键词搜索和向量搜索：
def hybrid_search(self, query: str, top_k: int = 10) -> List[Dict]:
"""混合检索：向量搜索 + 全文搜索"""
with self.driver.session() as session:
# 向量搜索结果
vector_results = session.run("""
CALL db.index.vector.queryNodes('document_embedding', $k, $embedding)
YIELD node, score
RETURN node.id AS id, score AS vector_score, 0 AS text_score
""", k=top_k, embedding=self.get_embedding(query))
# 全文搜索结果（需要先创建全文索引）
text_results = session.run("""
CALL db.index.fulltext.queryNodes('document_fulltext', $query)
YIELD node, score
RETURN node.id AS id, 0 AS vector_score, score AS text_score
""", query=query)
# 合并分数
scores = {}
for r in vector_results:
scores[r['id']] = {'vector_score': r['vector_score'], 'text_score': 0}
for r in text_results:
if r['id'] in scores:
scores[r['id']]['text_score'] = r['text_score']
else:
scores[r['id']] = {'vector_score': 0, 'text_score': r['text_score']}
# 计算加权分数
alpha = 0.7 # 向量搜索权重
for doc_id, s in scores.items():
s['combined'] = alpha * s['vector_score'] + (1 - alpha) * s['text_score']
# 返回排序结果
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x]['combined'], reverse=True)
# 获取文档详情
result = session.run("""
MATCH (d:Document) WHERE d.id IN $ids
RETURN d.id AS id, d.title AS title, d.content AS content
""", ids=sorted_ids[:top_k])
return [record.data() for record in result]
性能优化
索引配置优化
-- 选择合适的相似度函数
-- cosine：适合文本嵌入，忽略向量长度
-- euclidean：适合需要考虑幅度的场景
-- 维度选择：与嵌入模型匹配
-- OpenAI text-embedding-3-small: 1536
-- OpenAI text-embedding-3-large: 3072
-- Sentence Transformers: 通常 768
内存配置
向量索引需要足够的内存来保证性能：
# neo4j.conf 配置
# 页缓存大小（建议至少能容纳向量索引）
dbms.memory.pagecache.size=4G
# 堆内存
dbms.memory.heap.initial_size=2G
dbms.memory.heap.max_size=4G
批量导入优化
-- 使用 CALL IN TRANSACTIONS 批量导入
LOAD CSV WITH HEADERS FROM 'file:///documents.csv' AS row
CALL {
WITH row
CREATE (d:Document {
id: row.id,
title: row.title,
content: row.content,
embedding: row.embedding // 假设 CSV 中已有向量
})
} IN TRANSACTIONS OF 1000 ROWS
注意事项
向量维度一致性
确保存储的向量维度与索引配置一致：
-- 错误示例：向量维度不匹配
CREATE VECTOR INDEX test_vector IF NOT EXISTS
FOR (n:Test) ON (n.embedding)
OPTIONS {
indexConfig: {
`vector.dimensions`: 1536, // 索引配置 1536 维
`vector.similarity_function`: 'cosine'
}
}
-- 如果插入的向量是 768 维，会导致错误
CREATE (t:Test {embedding: [0.1, 0.2, ...]}) // 只有 768 个元素
向量值范围
确保向量值在合理范围内：
-- 向量值应该是浮点数
-- 避免使用过大的数值，可能导致精度问题
相似度函数选择
| 相似度函数 | 适用场景 | 特点 |
|---|---|---|
| cosine | 文本嵌入、语义搜索 | 忽略向量长度，关注方向 |
| euclidean | 空间数据、需要考虑幅度的场景 | 考虑向量的绝对位置 |
小结
向量搜索是 Neo4j 面向 AI 时代的重要功能：
- 语义搜索：基于内容的语义相似性进行搜索，而非关键词匹配
- 向量索引：使用近似最近邻算法高效处理大规模向量
- 灵活集成：与图的关系统合，实现复杂的推荐和搜索场景
- 多种相似度度量：支持 cosine、euclidean 等多种相似度计算方式
向量搜索在以下场景特别有价值：
- 语义搜索引擎
- 推荐系统
- RAG（检索增强生成）应用
- 相似内容发现
- 智能问答系统
参考资源
下一步
掌握了向量搜索后，可以查看 知识速查表 快速回顾所有重要概念和语法。