지식 증강 에이전트 구축
엔터티 연결 → 그래프 쿼리 → 답변 종합의 전 과정을 다룹니다.
지식 증강 에이전트 구축은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
지식 증강 에이전트란 무엇인가요?
지식 증강 에이전트는 지식 기반을 사용하여 답변을 더욱 풍부하게 만듭니다. 질문이 들어오면 에이전트는 엔터티를 추출하고, 지식 그래프에서 이를 조회하고, 벡터 검색으로 관련 문서를 찾은 다음, 풍부하고 근거 있는 답변을 생성할 수 있도록 모든 문맥을 LLM에 제공합니다.
전체 검색 파이프라인
에이전트 파이프라인은 다음과 같습니다. 1 질문 수신 → 2 엔터티 추출 → 3 엔터티 문맥을 위한 그래프 조회 → 4 관련 문서를 위한 벡터 검색 → 5 모든 문맥 결합 → 6 LLM이 답변 생성.
from dataclasses import dataclass, field
from typing import List, Dict, Any
@dataclass
class RetrievalContext:
question: str
entities: List[str] = field(default_factory=list)
graph_context: Dict[str, Any] = field(default_factory=dict)
vector_documents: List[Dict] = field(default_factory=list)
combined_context: str = ''
answer: str = ''
sources_used: List[str] = field(default_factory=list)
# The agent will populate this object as it works through the pipeline
ctx = RetrievalContext(question='What AI projects is Sam Altman known for?')
print('RetrievalContext created:', ctx.question)1단계: 엔터티 추출
질문에서 이름이 붙은 엔터티를 추출하십시오. 이 엔터티는 그래프 조회의 기준점이 됩니다. 속도를 위해 spaCy를 사용하고, 까다로운 경우나 분야별 엔터티에는 LLM을 사용하십시오.
import spacy
nlp = spacy.load('en_core_web_sm')
def extract_question_entities(question: str) -> List[str]:
doc = nlp(question)
entities = list({
ent.text for ent in doc.ents
if ent.label_ in ['PERSON', 'ORG', 'GPE', 'PRODUCT', 'WORK_OF_ART']
})
return entities
def extract_entities_with_llm_fallback(question: str, client) -> List[str]:
spacy_entities = extract_question_entities(question)
if spacy_entities:
return spacy_entities
# Fallback to LLM for questions where spaCy finds nothing
import json
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{
'role': 'user',
'content': f'Extract named entities (people, companies, technologies) from: "{question}". Return JSON: {{"entities": ["name1", "name2"]}}'
}],
response_format={'type': 'json_object'}
)
result = json.loads(response.choices[0].message.content)
return result.get('entities', [])
question = 'What AI projects is Sam Altman known for?'
entities = extract_question_entities(question)
print('Extracted entities:', entities)2단계: 그래프 조회
추출한 각 엔터티에 대해 지식 그래프를 조회하여 속성과 관계를 가져오십시오. 이렇게 하면 LLM이 환각으로 만들어 낼 수 없는 배경 사실을 제공할 수 있습니다.
from neo4j import GraphDatabase
driver = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', 'password'))
def get_rich_entity_context(entity_name: str) -> dict:
with driver.session() as session:
# Get entity + its relationships
result = session.run(
'MATCH (n {name: $name}) '
'OPTIONAL MATCH (n)-[r]->(target) '
'RETURN n, labels(n) AS labels, '
'COLLECT({rel: type(r), target_name: target.name, target_label: labels(target)}) AS outgoing '
'LIMIT 1',
name=entity_name
)
record = result.single()
if not record:
return {'found': False, 'name': entity_name}
return {
'found': True,
'name': entity_name,
'labels': record['labels'],
'properties': dict(record['n']),
'connections': [
c for c in record['outgoing'] if c.get('target_name')
][:10]
}
def format_entity_context_for_llm(entity_ctx: dict) -> str:
if not entity_ctx.get('found'):
return f'No knowledge graph data found for "{entity_ctx["name"]}"'
props = entity_ctx.get('properties', {})
connections = entity_ctx.get('connections', [])
conn_strs = [f"{c['rel']} -> {c['target_name']}" for c in connections[:5]]
return (
f"Entity: {entity_ctx['name']} ({', '.join(entity_ctx['labels'])})\n"
f"Properties: {props}\n"
f"Relationships: {'; '.join(conn_strs)}"
)3단계: 벡터 검색
원래 질문으로 벡터 검색을 실행하여 지식 기반에서 의미적으로 가장 관련성이 높은 문서를 찾으십시오. 이러한 문서는 답변을 뒷받침하는 근거를 제공합니다.
import chromadb
import openai
client = openai.OpenAI(api_key='sk-...')
chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection('knowledge_base')
def vector_search(query: str, top_k: int = 5) -> list:
response = client.embeddings.create(
model='text-embedding-3-small',
input=query
)
query_embedding = response.data[0].embedding
results = collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
include=['documents', 'metadatas', 'distances']
)
documents = []
for i in range(len(results['ids'][0])):
documents.append({
'text': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'distance': results['distances'][0][i],
'relevance': 1 - results['distances'][0][i] # Convert distance to similarity
})
return documents
print('Vector search function defined')4단계: 문맥 결합
그래프 문맥과 벡터 문서를 하나의 잘 구조화된 문맥 문자열로 모으십시오. 순서가 중요합니다. 정밀도가 높은 그래프 사실을 먼저 배치하고, 폭넓은 내용을 제공하는 벡터 문서를 그다음에 배치하십시오.
def combine_context(question: str, graph_contexts: dict, vector_docs: list) -> str:
sections = []
# Graph facts section
if graph_contexts:
graph_parts = ['### Knowledge Graph Facts']
for entity_name, ctx in graph_contexts.items():
graph_parts.append(format_entity_context_for_llm(ctx))
sections.append('\n'.join(graph_parts))
# Vector documents section
if vector_docs:
doc_parts = ['### Relevant Documents']
for i, doc in enumerate(vector_docs[:4]):
title = doc.get('metadata', {}).get('title', f'Document {i+1}')
text = doc['text'][:800] # Limit per document
relevance = doc.get('relevance', 0)
doc_parts.append(f'**{title}** (relevance: {relevance:.2f})\n{text}')
sections.append('\n'.join(doc_parts))
context = '\n\n'.join(sections)
# Total context budget: ~8000 tokens ~ 32000 chars
if len(context) > 32000:
context = context[:32000]
return context5단계: LLM 답변 생성
결합한 문맥을 시스템 메시지나 사용자 문맥으로 LLM에 전달하십시오. 제공된 정보를 사용하고 각 사실의 출처를 인용하도록 지시하십시오.
import openai
client = openai.OpenAI(api_key='sk-...')
def generate_answer(question: str, combined_context: str) -> str:
system_prompt = (
'You are a knowledgeable assistant. Answer the question using ONLY the provided context. '
'Cite your sources by mentioning whether a fact came from the knowledge graph or a specific document. '
'If the context does not contain enough information, say so clearly.'
)
user_message = (
f'Context:\n{combined_context}\n\n'
f'Question: {question}\n\n'
'Please answer based on the context above.'
)
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': user_message}
],
temperature=0.1 # Low temperature for factual answers
)
return response.choices[0].message.content전체 에이전트 오케스트레이터
오케스트레이터 함수는 모든 단계를 하나로 연결합니다. 질문을 받아 전체 파이프라인을 실행하고, 답변과 사용된 문맥이 포함된 구조화된 결과를 반환합니다.
async def knowledge_augmented_agent(question: str) -> RetrievalContext:
ctx = RetrievalContext(question=question)
# Step 1: Extract entities
ctx.entities = extract_question_entities(question)
print(f'Entities: {ctx.entities}')
# Steps 2 & 3: Graph + Vector in parallel
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
loop = asyncio.get_event_loop()
async def graph_step():
contexts = {}
for entity in ctx.entities:
context = await loop.run_in_executor(executor, get_rich_entity_context, entity)
if context.get('found'):
contexts[entity] = context
return contexts
async def vector_step():
return await loop.run_in_executor(executor, vector_search, question, 5)
ctx.graph_context, ctx.vector_documents = await asyncio.gather(
graph_step(), vector_step()
)
# Step 4: Combine
ctx.combined_context = combine_context(
question, ctx.graph_context, ctx.vector_documents
)
# Step 5: Generate answer
ctx.answer = generate_answer(question, ctx.combined_context)
return ctx검색 결과가 없을 때 처리하기
지식 기반에 관련 정보가 없을 때는 에이전트가 환각으로 답을 만들어 내지 말고 그 사실을 명확하게 밝혀야 합니다. LLM을 호출하기 전에 검색 결과에 유용한 내용이 포함되어 있는지 확인하십시오.
def check_retrieval_quality(graph_contexts: dict, vector_docs: list, threshold: float = 0.7) -> dict:
has_graph = len(graph_contexts) > 0
# Filter vector docs below relevance threshold
high_quality_docs = [d for d in vector_docs if d.get('relevance', 0) >= threshold]
return {
'has_graph_context': has_graph,
'graph_entity_count': len(graph_contexts),
'vector_doc_count': len(high_quality_docs),
'retrieval_quality': 'high' if (has_graph or len(high_quality_docs) >= 2) else 'low',
'usable_docs': high_quality_docs
}
def answer_with_fallback(question: str, graph_contexts: dict, vector_docs: list) -> str:
quality = check_retrieval_quality(graph_contexts, vector_docs)
if quality['retrieval_quality'] == 'low':
return (
f'I don\'t have enough information in my knowledge base to answer '
f'"{question}" confidently. '
'Please ensure relevant documents are indexed or the knowledge graph '
'contains the required entities.'
)
context = combine_context(question, graph_contexts, quality['usable_docs'])
return generate_answer(question, context)검색 결과 캐싱
비슷한 질문에 대해 비용이 많이 드는 API 호출을 반복하지 않도록 엔터티 조회 결과와 벡터 검색 결과를 캐시하십시오. 오래된 데이터가 주기적으로 갱신되도록 TTL을 사용하십시오.
import hashlib
import json
from datetime import datetime, timedelta
class RetrievalCache:
def __init__(self, ttl_minutes: int = 60):
self.cache = {}
self.ttl = timedelta(minutes=ttl_minutes)
def _key(self, namespace: str, value: str) -> str:
return hashlib.md5(f'{namespace}:{value}'.encode()).hexdigest()
def get(self, namespace: str, value: str):
key = self._key(namespace, value)
entry = self.cache.get(key)
if entry and datetime.now() - entry['ts'] < self.ttl:
return entry['data']
return None
def set(self, namespace: str, value: str, data):
key = self._key(namespace, value)
self.cache[key] = {'data': data, 'ts': datetime.now()}
cache = RetrievalCache(ttl_minutes=30)
def cached_graph_lookup(entity: str) -> dict:
cached = cache.get('graph', entity)
if cached:
print(f'Cache hit for entity: {entity}')
return cached
result = get_rich_entity_context(entity)
cache.set('graph', entity, result)
return result
print('Retrieval cache initialized')로그 기록과 관찰 가능성
답변이 좋거나 나빴던 이유를 진단할 수 있도록 모든 검색 단계를 기록하십시오. 발견된 엔터티, 검색된 문서 수, 각 문서의 관련성 점수, 최종 답변을 기록하십시오.
import logging
import json
from datetime import datetime
logger = logging.getLogger('ka_agent')
def log_agent_run(ctx: 'RetrievalContext', duration_ms: float):
logger.info(json.dumps({
'timestamp': datetime.utcnow().isoformat(),
'question': ctx.question,
'entities_found': ctx.entities,
'graph_entities_resolved': list(ctx.graph_context.keys()),
'vector_docs_retrieved': len(ctx.vector_documents),
'vector_doc_relevances': [
round(d.get('relevance', 0), 3)
for d in ctx.vector_documents
],
'context_length_chars': len(ctx.combined_context),
'answer_length_chars': len(ctx.answer),
'duration_ms': round(duration_ms, 1)
}))
print('Observability logging configured')지식 증강 에이전트 이해도 확인
지식 증강 에이전트 구축에 대한 이해도를 확인합니다.
지식 증강 에이전트 요약
지식 증강 에이전트는 엔터티 추출, 그래프 순회, 벡터 검색을 하나의 파이프라인으로 결합하여 LLM에 풍부하고 근거가 있는 컨텍스트를 제공합니다. 그 결과 지식 기반의 실제 데이터로 뒷받침되는, 더 정확하고 환각이 적은 답변을 얻을 수 있습니다. 주요 추가 요소는 캐싱, 검색 결과가 비어 있을 때의 대체 처리, 관찰 가능성을 위한 구조화된 기록입니다.
자주 묻는 질문
“지식 증강 에이전트 구축” 강의는 무료인가요?
네 — “지식 증강 에이전트 구축” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“지식 증강 에이전트 구축”에서 뭘 배우나요?
엔터티 연결 → 그래프 쿼리 → 답변 종합의 전 과정을 다룹니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.
“지식 증강 에이전트 구축” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.