벡터 검색과 그래프 검색 결합
하이브리드 검색으로 벡터 유사도와 그래프 경로 순회를 결합해 더 풍부한 맥락을 확보합니다.
벡터 검색과 그래프 검색 결합은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 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. 사용 사례에서 의미적 유사성과 관계형 문맥 중 어느 쪽이 더 중요한지에 따라 알파를 조정하십시오.
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')검색 가중치 선택
질문 유형에 따라 알파 매개변수(벡터와 그래프의 가중치)를 조정하십시오:
- 사실 조회 질문(OpenAI를 설립한 사람은 누구인가요?) → 그래프 가중치를 높게 설정
- 의미적 유사성 질문(AI 안전에 관한 문서를 찾아 주세요) → 벡터 가중치를 높게 설정
- 혼합 질문 → 균형 잡힌 가중치(알파=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 답변에 더 풍부한 문맥을 제공할 수 있습니다.
자주 묻는 질문
“벡터 검색과 그래프 검색 결합” 강의는 무료인가요?
네 — “벡터 검색과 그래프 검색 결합” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“벡터 검색과 그래프 검색 결합”에서 뭘 배우나요?
하이브리드 검색으로 벡터 유사도와 그래프 경로 순회를 결합해 더 풍부한 맥락을 확보합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.
“벡터 검색과 그래프 검색 결합” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 지식 그래프를 위한 엔터티 추출
- 에이전트 도구에서 Neo4j 쿼리 실행
- 벡터 검색과 그래프 검색 결합
- 지식 증강 에이전트 구축