结合向量检索与图谱检索
混合检索：结合向量相似度与图谱路径遍历，获取更丰富的上下文
第 3 / 4 课13 个步骤
结合向量检索与图谱检索 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
为什么使用混合检索
向量搜索可以找到语义相似的内容，但会遗漏结构化关系。图遍历可以捕获关系，但不擅长处理语义相似性。混合检索将两者结合起来，以提供更丰富的上下文。
向量搜索回顾
向量搜索会将查询和文档转换为嵌入向量（稠密向量），然后查找余弦相似度较高的文档。它可以回答哪些文档讨论的是同一主题？
import openai
import numpy as np
client = openai.OpenAI(api_key='sk-...')
def embed(text: str) -> list:
response = client.embeddings.create(
model='text-embedding-3-small',
input=text
)
return response.data[0].embedding
def cosine_similarity(a: list, b: list) -> float:
a_arr = np.array(a)
b_arr = np.array(b)
return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))
# Simple in-memory vector store
class SimpleVectorStore:
def __init__(self):
self.documents = []
def add(self, text: str, metadata: dict):
embedding = embed(text)
self.documents.append({'text': text, 'embedding': embedding, 'metadata': metadata})
def search(self, query: str, top_k: int = 5) -> list:
query_emb = embed(query)
scored = [
(cosine_similarity(query_emb, doc['embedding']), doc)
for doc in self.documents
]
scored.sort(key=lambda x: x[0], reverse=True)
return [doc for _, doc in scored[:top_k]]图检索回顾
图检索用于回答关系类问题：谁与 X 有关联？、这个人认识哪些公司？它使用显式边，而不是语义相似性。
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def get_entity_context(entity_name: str) -> dict:
with driver.session() as session:
# Get node properties
result = session.run(
'MATCH (n {name: $name}) RETURN n, labels(n) AS labels LIMIT 1',
name=entity_name
)
record = result.single()
if not record:
return {}
node_data = dict(record['n'])
node_labels = record['labels']
# Get connected entities
conn_result = session.run(
'MATCH (n {name: $name})-[r]-(connected) '
'RETURN type(r) AS rel_type, connected.name AS connected_name, labels(connected) AS connected_labels '
'LIMIT 20',
name=entity_name
)
connections = [dict(r) for r in conn_result]
return {
'name': entity_name,
'labels': node_labels,
'properties': node_data,
'connections': connections
}交错合并结果
一种融合策略是交错合并向量结果和图结果：先取向量搜索的第一个结果，再取图搜索的第一个结果，然后取向量搜索的第二个结果，依此类推。这样可以确保两个来源都发挥作用。
def interleave_results(vector_results: list, graph_results: list) -> list:
combined = []
v_idx, g_idx = 0, 0
while v_idx < len(vector_results) or g_idx < len(graph_results):
if v_idx < len(vector_results):
item = vector_results[v_idx]
item['source'] = 'vector'
combined.append(item)
v_idx += 1
if g_idx < len(graph_results):
item = graph_results[g_idx]
item['source'] = 'graph'
combined.append(item)
g_idx += 1
return combined
# Example
vector_docs = [
{'text': 'Alice led the machine learning initiative at Acme', 'score': 0.92},
{'text': 'Machine learning best practices guide', 'score': 0.85},
]
graph_context = [
{'name': 'Alice', 'type': 'Person', 'connections': ['Acme Corp', 'Bob']},
]
fused = interleave_results(vector_docs, graph_context)
for item in fused:
print(f"[{item['source']}]", item.get('text') or item.get('name'))加权组合
使用组合分数为每个结果评分：final_score = alpha * vector_score + (1-alpha) * graph_score。请根据您的使用场景中语义相似性或关系上下文哪个更重要来调整 alpha。
def weighted_fusion(vector_results: list, graph_results: list, alpha: float = 0.6) -> list:
'''
alpha: weight for vector results (0.0 = pure graph, 1.0 = pure vector)
'''
all_results = []
# Normalize vector scores (already in 0-1 range for cosine)
for i, res in enumerate(vector_results):
# Positional score: first result gets highest
positional_score = 1.0 - (i / max(len(vector_results), 1))
combined = alpha * res.get('score', positional_score)
all_results.append({
'content': res,
'source': 'vector',
'final_score': combined
})
# Graph results: score by relevance (e.g., connection count)
for i, res in enumerate(graph_results):
positional_score = 1.0 - (i / max(len(graph_results), 1))
combined = (1 - alpha) * positional_score
all_results.append({
'content': res,
'source': 'graph',
'final_score': combined
})
# Sort by final score
all_results.sort(key=lambda x: x['final_score'], reverse=True)
return all_results
print('Weighted fusion function defined (alpha=0.6 favors vector)')倒数排名融合
倒数排名融合（RRF）是一种稳健的方法，可以在无需对分数进行归一化的情况下合并多个排序列表。每个文档都会在所有列表中获得分数 sum(1 / (k + rank))。
def reciprocal_rank_fusion(result_lists: list, k: int = 60) -> list:
'''
result_lists: list of lists, each containing dicts with an 'id' field
k: constant to reduce impact of high rankings (typically 60)
'''
scores = {}
all_items = {}
for result_list in result_lists:
for rank, item in enumerate(result_list):
item_id = item.get('id') or item.get('text', '')[:50]
if item_id not in scores:
scores[item_id] = 0.0
all_items[item_id] = item
scores[item_id] += 1.0 / (k + rank + 1)
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [
{**all_items[id_], 'rrf_score': scores[id_]}
for id_ in sorted_ids
]
vector_list = [{'id': 'doc1', 'text': 'About Alice'}, {'id': 'doc3', 'text': 'About AI'}]
graph_list = [{'id': 'doc2', 'text': 'Alice connections'}, {'id': 'doc1', 'text': 'About Alice'}]
fused = reciprocal_rank_fusion([vector_list, graph_list])
for item in fused:
print(f"{item['id']}: RRF score {item['rrf_score']:.4f}")实体锚定的混合检索
一种强大的混合方法是：从查询中提取实体，使用图谱获取这些实体的上下文，然后利用该上下文增强向量搜索查询。
import spacy
nlp = spacy.load('en_core_web_sm')
def entity_anchored_retrieval(query: str, vector_store, graph_driver) -> dict:
# Step 1: Extract entities from query
doc = nlp(query)
entities = [ent.text for ent in doc.ents if ent.label_ in ['PERSON', 'ORG', 'GPE']]
# Step 2: Get graph context for entities
graph_contexts = {}
for entity in entities:
context = get_entity_context(entity)
if context:
graph_contexts[entity] = context
# Step 3: Enrich query with graph context
enriched_query = query
if graph_contexts:
context_str = ' '.join([
f"{name} works at {', '.join([c['connected_name'] for c in ctx.get('connections', [])[:3]])}"
for name, ctx in graph_contexts.items()
])
enriched_query = f'{query} Context: {context_str}'
# Step 4: Vector search with enriched query
vector_results = vector_store.search(enriched_query, top_k=5)
return {
'entities_found': entities,
'graph_contexts': graph_contexts,
'vector_results': vector_results
}构建上下文包
最后的检索步骤是将所有上下文（向量结果和图数据）打包成结构化字符串，提供给 LLM。LLM 会利用这些内容生成全面的答案。
def build_context_package(vector_results: list, graph_contexts: dict, max_tokens: int = 3000) -> str:
sections = []
# Graph entity context section
if graph_contexts:
graph_section = ['## Entity Context from Knowledge Graph']
for entity_name, context in graph_contexts.items():
connections = context.get('connections', [])
conn_summary = ', '.join([
f"{c['connected_name']} ({c['rel_type']})"
for c in connections[:5]
])
graph_section.append(f'**{entity_name}**: connected to {conn_summary}')
sections.append('\n'.join(graph_section))
# Vector search results section
if vector_results:
vector_section = ['## Relevant Documents']
for i, doc in enumerate(vector_results[:5]):
text = doc.get('text', '')[:500] # Truncate long docs
vector_section.append(f'{i+1}. {text}')
sections.append('\n'.join(vector_section))
context_package = '\n\n'.join(sections)
# Rough token estimate (1 token ~ 4 chars)
if len(context_package) > max_tokens * 4:
context_package = context_package[:max_tokens * 4]
return context_package
if __name__ == '__main__':
demo_vector = [{'text': 'Refunds are processed within 5 business days of approval.'}]
demo_graph = {'Acme Corp': {'connections': [{'connected_name': 'Jane Doe', 'rel_type': 'employs'}]}}
print(build_context_package(demo_vector, demo_graph))
异步并行检索
使用 asyncio.gather 并行运行向量检索和图检索，以尽量缩短总延迟。这样两个结果可以同时准备就绪。
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
async def async_vector_search(query: str, vector_store) -> list:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(executor, vector_store.search, query, 5)
async def async_graph_lookup(entities: list) -> dict:
loop = asyncio.get_event_loop()
results = {}
for entity in entities:
context = await loop.run_in_executor(executor, get_entity_context, entity)
if context:
results[entity] = context
return results
async def hybrid_retrieval_async(query: str, entities: list, vector_store) -> dict:
# Run vector search and graph lookup in parallel
vector_task = async_vector_search(query, vector_store)
graph_task = async_graph_lookup(entities)
vector_results, graph_contexts = await asyncio.gather(vector_task, graph_task)
return {
'vector': vector_results,
'graph': graph_contexts
}
print('Async parallel retrieval functions defined')缓存检索结果
缓存向量搜索结果和图查询结果，以避免重复的 API 调用。由于知识库变化缓慢但并非即时不变，请使用较短的 TTL（几分钟到几小时）。
import hashlib
import time
class HybridRetrievalCache:
def __init__(self, vector_ttl: int = 300, graph_ttl: int = 600):
self.vector_cache = {}
self.graph_cache = {}
self.vector_ttl = vector_ttl
self.graph_ttl = graph_ttl
def _key(self, value: str) -> str:
return hashlib.md5(value.encode()).hexdigest()[:12]
def get_vector(self, query: str):
k = self._key(query)
entry = self.vector_cache.get(k)
if entry and time.time() - entry['ts'] < self.vector_ttl:
return entry['data']
return None
def set_vector(self, query: str, results: list):
self.vector_cache[self._key(query)] = {'data': results, 'ts': time.time()}
def get_graph(self, entity: str):
k = self._key(entity)
entry = self.graph_cache.get(k)
if entry and time.time() - entry['ts'] < self.graph_ttl:
return entry['data']
return None
def set_graph(self, entity: str, context: dict):
self.graph_cache[self._key(entity)] = {'data': context, 'ts': time.time()}
cache = HybridRetrievalCache()
print('Hybrid retrieval cache initialized')选择检索权重
请根据查询类型调整 alpha 参数（向量权重与图权重）：
- 事实查询（谁创立了 OpenAI？）→ 更高的图权重
- 语义相似性查询（查找关于 AI 安全的文档）→ 更高的向量权重
- 混合查询 → 平衡权重（alpha=0.5）
def auto_tune_alpha(query: str) -> float:
query_lower = query.lower()
# High graph weight for relational questions
relational_keywords = [
'who', 'founded', 'works at', 'connected to',
'related to', 'partner', 'owns', 'acquired'
]
# High vector weight for content questions
content_keywords = [
'explain', 'describe', 'what is', 'how does',
'tell me about', 'documents about', 'find information'
]
relational_count = sum(1 for kw in relational_keywords if kw in query_lower)
content_count = sum(1 for kw in content_keywords if kw in query_lower)
if relational_count > content_count:
return 0.3 # Graph-heavy
elif content_count > relational_count:
return 0.7 # Vector-heavy
else:
return 0.5 # Balanced
queries = [
'Who founded Tesla?',
'Explain transformer architecture',
'What companies is Elon Musk connected to?'
]
for q in queries:
print(f'alpha={auto_tune_alpha(q):.1f} for: {q}')知识检查：混合检索
请测试您对结合向量检索和图检索的理解。
混合检索总结
有效的混合检索包括：使用向量搜索获取语义相似性，使用图遍历获取关系上下文，使用实体提取将查询锚定到图谱，使用融合策略（交错、加权、RRF）合并结果，以及通过异步并行执行来尽量缩短延迟。最终可以为 LLM 生成的答案提供更丰富的上下文。
免费开始
用 AI 导师学习 AI Agents — 免费
在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。
- 课程
- 60
- 课程
- 239
常见问题解答
「结合向量检索与图谱检索」课时是免费的吗？
是的 — 「结合向量检索与图谱检索」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「结合向量检索与图谱检索」这节课中我会学到什么？
混合检索：结合向量相似度与图谱路径遍历，获取更丰富的上下文 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「结合向量检索与图谱检索」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 知识图谱的实体提取
- 通过智能体工具查询 Neo4j
- 结合向量检索与图谱检索
- 构建知识增强型智能体