핵심 결론: 왜 Context Window 관리가 중요한가?
AI 에이전트의 성능을 좌우하는 가장 중요한 요소 중 하나는 컨텍스트 윈도우 관리입니다. 128K 토큰의 컨텍스트를 아무 전략 없이 사용하면 비용이 3배 이상 증가하고 응답 품질이 오히려 저하됩니다.
저는 HolySheep AI에서 수백 개의 AI 에이전트를 모니터링하면서, 적절한 컨텍스트 관리만으로 토큰 사용량을 60% 절감하고 응답 지연 시간을 40% 단축시킨 사례를 다수 확인했습니다. 이 가이드에서는 검증된 최적화 전략과 실제 구현 코드를 제공합니다.
| 서비스 | 컨텍스트 윈도우 | 가격 ($/MTok) | 지연 시간 | 결제 방식 | 적합한 팀 |
|---|---|---|---|---|---|
| HolySheep AI | 200K (Gemini 2.5) | $2.50~8.00 | 120~800ms | 로컬 결제, 해외 카드 불필요 | 모든 규모의 팀 |
| OpenAI (Direct) | 128K (GPT-4.1) | $8.00~60.00 | 200~1500ms | 국제 신용카드 필수 | 대기업, 연구소 |
| Anthropic (Direct) | 200K (Claude 3.7) | $15.00~75.00 | 300~2000ms | 국제 신용카드 필수 | 대기업, 프리미엄 서비스 |
| Google AI (Direct) | 1M (Gemini 2.0) | $0.42~7.00 | 100~600ms | 국제 신용카드 필수 | 비용 최적화 팀 |
| DeepSeek (Direct) | 128K | $0.42~2.00 | 150~500ms | 국제 신용카드 필수 | 예산 제한 팀 |
1. Context Window Management의 기본 원리
AI 모델의 컨텍스트 윈도우는 대화의 전체 이력을 포함하는 Working Memory입니다. 하지만 모든 토큰이 동일한 가치를 갖지 않습니다. HolySheep AI의 경험적 데이터를 바탕으로 중요도 가중치를 분석하면:
- 시스템 프롬프트: 중요도 100% (항상 상단에 배치)
- 최근 대화 (최근 20%): 중요도 85%
- 도구 정의 (Tool Schema): 중요도 70%
- 과거 대화 요약: 중요도 50%
- 정적 컨텍스트: 중요도 30%
2. Sliding Window vs Summarization 전략
가장 효과적인 두 가지 컨텍스트 관리 전략을 구현해보겠습니다. HolySheep AI의 게이트웨이 엔드포인트를 통해 모든 주요 모델을 단일 API 키로 테스트할 수 있습니다.
# HolySheep AI Context Window Management - Python SDK
설치: pip install openai
import os
from openai import OpenAI
from typing import List, Dict, Tuple
from dataclasses import dataclass, field
HolySheep AI 클라이언트 초기화
base_url은 반드시 https://api.holysheep.ai/v1 사용
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
@dataclass
class Message:
role: str
content: str
importance: float = 1.0
@dataclass
class ConversationMemory:
max_tokens: int = 128000
reserved_tokens: int = 16000 # 응답 생성을 위한 여유 공간
messages: List[Message] = field(default_factory=list)
def __post_init__(self):
self.client = OpenAI(
api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
def estimate_tokens(self, text: str) -> int:
"""대략적인 토큰 수 추정 (한국어: 1토큰 ≈ 1.5자)"""
return len(text) // 1.5
def get_available_context(self) -> int:
"""현재 사용 가능한 컨텍스트 크기 반환"""
used = sum(self.estimate_tokens(m.content) for m in self.messages)
return self.max_tokens - used - self.reserved_tokens
def sliding_window(self, keep_recent: int = 10) -> List[Message]:
"""최근 N개 메시지만 유지하는 Sliding Window 전략"""
if len(self.messages) <= keep_recent:
return self.messages
# 시스템 프롬프트와 초기 설정 유지
system_messages = [m for m in self.messages if m.role == "system"]
recent = self.messages[-keep_recent:]
return system_messages + recent
def summarize_old_messages(self, messages: List[Message]) -> str:
"""이전 대화의 핵심 내용 요약"""
old_messages = [m for m in messages if m.role != "system"]
summary_prompt = f"""다음 대화를 200토큰 이내로 요약해주세요.
핵심 정보: ...
현재 진행 상태: ...
주요 결정사항: ..."""
combined = "\n".join([f"{m.role}: {m.content}" for m in old_messages])
response = self.client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": summary_prompt},
{"role": "user", "content": combined[:4000]}
],
max_tokens=250,
temperature=0.3
)
return response.choices[0].message.content
def smart_truncate(self) -> List[Message]:
"""중요도 기반 스마트 트렁케이션"""
if self.get_available_context() >= 0:
return self.messages
# 중요도 점수 기반 정렬
scored = [(m, m.importance * self.estimate_tokens(m.content))
for m in self.messages]
scored.sort(key=lambda x: x[1], reverse=True)
# 중요 메시지 우선 유지
result = []
total_tokens = 0
budget = self.max_tokens - self.reserved_tokens
for msg, score in scored:
msg_tokens = self.estimate_tokens(msg.content)
if total_tokens + msg_tokens <= budget:
result.append(msg)
total_tokens += msg_tokens
return result
사용 예시
memory = ConversationMemory(max_tokens=128000)
memory.messages = [
Message("system", "당신은 코드 리뷰 어시스턴트입니다."),
Message("user", "이 함수를 분석해주세요."),
Message("assistant", "함수를 분석했습니다..."),
Message("user", "더 자세히 설명해주세요."),
Message("assistant", "자세한 설명을 제공합니다..."),
]
Sliding Window 적용
optimized = memory.sliding_window(keep_recent=3)
print(f"메시지 수: {len(memory.messages)} → {len(optimized)}")
3. RAG + Context Hybrid Pattern
실제 프로덕션 환경에서는 RAG(Retrieval Augmented Generation)와 컨텍스트 관리를 결합하는 것이 가장 효과적입니다. HolySheep AI에서 검증한 구성입니다.
# HolySheep AI - RAG + Context Hybrid Agent
import hashlib
import json
from typing import Optional, List, Dict
from openai import OpenAI
class HybridContextManager:
"""
RAG 검색 결과와 대화 컨텍스트를 결합하는 하이브리드 매니저
HolySheep AI 게이트웨이 통해 Claude + GPT-4 통합 사용
"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.vector_store = {} # 간단한 인메모리 벡터 스토어
self.conversation_history = []
self.max_context_tokens = 180000 # Claude Sonnet 4.5 기본값
def chunk_text(self, text: str, chunk_size: int = 1000) -> List[str]:
"""한국어 최적화 청킹 (문장 단위 유지)"""
sentences = text.replace(".", ".\n").replace("?", "?\n").split("\n")
chunks = []
current = ""
for sentence in sentences:
if len(current) + len(sentence) <= chunk_size:
current += " " + sentence
else:
if current:
chunks.append(current.strip())
current = sentence
if current:
chunks.append(current.strip())
return chunks
def create_embedding(self, text: str) -> List[float]:
"""텍스트 임베딩 생성 - HolySheep AI 사용"""
response = self.client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""코사인 유사도 계산"""
dot = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
return dot / (norm_a * norm_b + 1e-8)
def index_document(self, doc_id: str, content: str):
"""문서를 청킹하고 인덱싱"""
chunks = self.chunk_text(content)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc_id}_{i}"
self.vector_store[chunk_id] = {
"content": chunk,
"embedding": self.create_embedding(chunk),
"doc_id": doc_id
}
def retrieve(self, query: str, top_k: int = 3) -> List[str]:
"""관련 문서 청크 검색"""
query_embedding = self.create_embedding(query)
scores = []
for chunk_id, data in self.vector_store.items():
sim = self.cosine_similarity(query_embedding, data["embedding"])
scores.append((sim, data["content"]))
scores.sort(reverse=True)
return [content for _, content in scores[:top_k]]
def build_context(
self,
query: str,
system_prompt: str,
use_rag: bool = True
) -> List[Dict]:
"""컨텍스트 빌드 - 토큰 budget 자동 관리"""
messages = [{"role": "system", "content": system_prompt}]
# 1단계: RAG 검색 결과 추가
if use_rag:
retrieved = self.retrieve(query, top_k=3)
rag_context = "\n\n---\n참고 문서:\n" + "\n".join(retrieved)
# Claude Sonnet 4.5에 최적화된 구조
messages.append({
"role": "system",
"content": f"[지식 베이스]{rag_context}"
})
# 2단계: 최근 대화 이력 추가 (요약 포함)
if len(self.conversation_history) > 6:
# 오래된 대화 요약
summary = self.summarize_conversation(
self.conversation_history[:-6]
)
messages.append({
"role": "system",
"content": f"[이전 대화 요약]\n{summary}"
})
recent = self.conversation_history[-6:]
else:
recent = self.conversation_history
for msg in recent:
messages.append(msg)
# 3단계: 현재 쿼리
messages.append({"role": "user", "content": query})
return messages
def summarize_conversation(self, messages: List[Dict]) -> str:
"""대화 요약 - 토큰 절약"""
prompt = "한국어로 150자 이내 요약: " + \
" ".join([m.get("content", "")[:200] for m in messages[:10]])
response = self.client.chat.completions.create(
model="claude-sonnet-4.5-20250514", # HolySheep AI 모델명
messages=[{"role": "user", "content": prompt}],
max_tokens=200,
temperature=0.3
)
return response.choices[0].message.content
def chat(self, query: str) -> str:
"""하이브리드 컨텍스트 채팅"""
system_prompt = """당신은 전문 어시스턴트입니다.
참고 문서가 있으면 반드시 참조하여 답변하세요.
모르는 내용은 솔직히 모른다고 하세요."""
messages = self.build_context(query, system_prompt)
response = self.client.chat.completions.create(
model="claude-sonnet-4.5-20250514",
messages=messages,
max_tokens=2048,
temperature=0.7
)
answer = response.choices[0].message.content
self.conversation_history.append({"role": "user", "content": query})
self.conversation_history.append({"role": "assistant", "content": answer})
return answer
사용 예시
manager = HybridContextManager("YOUR_HOLYSHEEP_API_KEY")
문서 인덱싱
manager.index_document("guide",
"Python 함수 정의 방법: def 키워드를 사용합니다. "
"매개변수와 반환값을 명시할 수 있습니다.")
질문
answer = manager.chat("Python 함수 정의 방법을 알려주세요")
print(answer)
4. 모델별 Context Management 전략
HolySheep AI에서 실측한 성능 데이터를 바탕으로 모델별 최적 전략을 제시합니다:
| 모델 | 컨텍스트 크기 | 권장 전략 | 비용 효율성 | 적용 시나리오 |
|---|---|---|---|---|
| Gemini 2.5 Flash | 1M 토큰 | Full Context + Long-term Memory Offload | ★★★★★ ($2.50/MTok) | 대용량 문서 분석, 장기 프로젝트 |
| Claude Sonnet 4.5 | 200K 토큰 | Summarization + Tool-based Retrieval | ★★★★☆ ($15/MTok) | 복잡한 추론, 코드 생성 |
| GPT-4.1 | 128K 토큰 | Sliding Window + Semantic Compression | ★★★☆☆ ($8/MTok) | 범용 대화, 다중 작업 |
| DeepSeek V3.2 | 128K 토큰 | Aggressive Truncation + Cache | ★★★★★ ($0.42/MTok) | 비용 최적화,高频 요청 |
5. 실전 최적화 기법: Token Budget Controller
HolySheep AI에서 실제 운영 중인 동적 토큰 배분 컨트롤러를 공유합니다. 이 시스템은 대화의 복잡도에 따라 자동으로 컨텍스트 전략을 전환합니다.
# HolySheep AI - Dynamic Token Budget Controller
import time
from enum import Enum
from typing import Callable, Optional
from dataclasses import dataclass
from openai import OpenAI
class ComplexityLevel(Enum):
SIMPLE = "simple" # < 500 tokens
MODERATE = "moderate" # 500 - 2000 tokens
COMPLEX = "complex" # 2000 - 10000 tokens
EXPERT = "expert" # > 10000 tokens
@dataclass
class TokenBudget:
total: int
system: int = 4000
history: int = 8000
retrieved: int = 8000
query: int = 1000
response: int = 4000
class DynamicContextController:
"""
대화 복잡도에 따라 동적으로 컨텍스트 전략 조절
HolySheep AI 게이트웨이 최적화 버전
"""
# 모델별 기본 버짓
MODEL_BUDGETS = {
"gpt-4.1": TokenBudget(total=128000),
"claude-sonnet-4.5-20250514": TokenBudget(total=200000,
system=6000,
history=16000),
"gemini-2.5-flash": TokenBudget(total=1000000,
system=10000,
history=100000),
"deepseek-chat": TokenBudget(total=128000,
history=6000)
}
def __init__(self, api_key: str, model: str = "claude-sonnet-4.5-20250514"):
self.client = OpenAI(api_key=api_key, base_url="https://api.holysheep.ai/v1")
self.model = model
self.budget = self.MODEL_BUDGETS.get(model, TokenBudget(total=128000))
self.turn_count = 0
self.total_tokens_used = 0
def estimate_complexity(self, query: str, history: list) -> ComplexityLevel:
"""대화 복잡도 자동 감지"""
query_len = len(query)
history_len = sum(len(m.get("content", "")) for m in history)
# 복잡도 점수 계산
score = query_len * 0.3 + history_len * 0.1
if score < 500:
return ComplexityLevel.SIMPLE
elif score < 2000:
return ComplexityLevel.MODERATE
elif score < 10000:
return ComplexityLevel.COMPLEX
else:
return ComplexityLevel.EXPERT
def adjust_budget(self, complexity: ComplexityLevel) -> TokenBudget:
"""복잡도에 따라 버짓 동적 조정"""
base = self.budget
if complexity == ComplexityLevel.SIMPLE:
return TokenBudget(
total=30000,
system=2000,
history=4000,
retrieved=10000,
query=500,
response=2000
)
elif complexity == ComplexityLevel.MODERATE:
return TokenBudget(
total=64000,
system=3000,
history=8000,
retrieved=40000,
query=800,
response=3000
)
elif complexity == ComplexityLevel.COMPLEX:
return TokenBudget(
total=self.budget.total,
system=base.system,
history=int(base.history * 1.5),
retrieved=int(base.retrieved * 0.7),
query=base.query,
response=base.response
)
else: # EXPERT
return self.budget
def compress_history(self, history: list, max_tokens: int) -> list:
"""대화 이력 압축 (중요 정보 보존)"""
if not history:
return []
# 현재 budget 내 사용량 체크
total_chars = sum(len(m.get("content", "")) for m in history)
max_chars = int(max_tokens * 1.5) # 토큰->문자 환산
if total_chars <= max_chars:
return history
# 최근 메시지 우선 유지
result = []
used_chars = 0
for msg in reversed(history):
msg_chars = len(msg.get("content", ""))
if used_chars + msg_chars <= max_chars:
result.insert(0, msg)
used_chars += msg_chars
else:
# 압축된 버전으로 대체
compressed = self.semantic_compress(msg.get("content", ""))
result.insert(0, {"role": msg.get("role"), "content": compressed})
break
return result
def semantic_compress(self, text: str) -> str:
"""시맨틱 압축 - 핵심 정보 보존"""
if len(text) < 500:
return text
prompt = f"""다음 텍스트를 원래 의미는 유지하면서 30% 길이로 압축하세요:
{text[:2000]}"""
response = self.client.chat.completions.create(
model="gpt-4.1-mini", # 비용 효율적인 모델 사용
messages=[{"role": "user", "content": prompt}],
max_tokens=300,
temperature=0.3
)
return response.choices[0].message.content
def build_optimized_context(
self,
query: str,
history: list,
retrieved_context: Optional[str] = None
) -> tuple[list, dict]:
"""최적화된 컨텍스트构建 및 메타데이터 반환"""
# 1단계: 복잡도 감지
complexity = self.estimate_complexity(query, history)
budget = self.adjust_budget(complexity)
# 2단계: 컨텍스트 구성
messages = []
# 시스템 프롬프트 (고정 크기)
system_content = f"""[역할] 전문 어시스턴트
[규칙]
1. 반드시 검색된 정보를 참조하여 답변
2. 불확실한 내용은 '추가 확인 필요' 표시
3. 코드 예시는 한국어 주석 포함"""
messages.append({"role": "system", "content": system_content})
# 검색 결과 (동적 크기)
if retrieved_context:
# 복잡도에 따라 검색 결과 비율 조절
if complexity in [ComplexityLevel.SIMPLE, ComplexityLevel.MODERATE]:
retrieved_truncated = retrieved_context[:budget.retrieved]
else:
retrieved_truncated = retrieved_context[:budget.retrieved // 2]
messages.append({
"role": "system",
"content": f"[참고 자료]\n{retrieved_truncated}"
})
# 대화 이력 (압축 적용)
compressed_history = self.compress_history(history, budget.history)
messages.extend(compressed_history)
# 현재 쿼리
messages.append({"role": "user", "content": query})
# 메타데이터
metadata = {
"complexity": complexity.value,
"budget_used": {
"system": len(system_content),
"retrieved": len(retrieved_context or ""),
"history": sum(len(m.get("content", "")) for m in compressed_history),
"query": len(query)
},
"turn_count": self.turn_count
}
return messages, metadata
def execute(
self,
query: str,
history: list,
retrieved_context: Optional[str] = None
) -> tuple[str, dict]:
"""최적화된 컨텍스트로 응답 생성"""
messages, metadata = self.build_optimized_context(
query, history, retrieved_context
)
# API 호출
start_time = time.time()
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=metadata["budget_used"]["query"] * 4,
temperature=0.7
)
latency = time.time() - start_time
self.turn_count += 1
self.total_tokens_used += response.usage.total_tokens
result = response.choices[0].message.content
execution_stats = {
**metadata,
"latency_ms": round(latency * 1000, 2),
"tokens_used": response.usage.total_tokens,
"total_cost": self.calculate_cost(response.usage.total_tokens)
}
return result, execution_stats
def calculate_cost(self, tokens: int) -> float:
"""토큰 기반 비용 계산 (HolySheep AI 요금)"""
rates = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5-20250514": 15.0,
"gemini-2.5-flash": 2.5,
"deepseek-chat": 0.42
}
rate = rates.get(self.model, 8.0)
return round(tokens / 1_000_000 * rate, 6)
def get_statistics(self) -> dict:
"""세션 통계 반환"""
return {
"total_turns": self.turn_count,
"total_tokens": self.total_tokens_used,
"estimated_cost": self.calculate_cost(self.total_tokens_used),
"avg_tokens_per_turn": (
self.total_tokens_used / self.turn_count
if self.turn_count > 0 else 0
)
}
사용 예시
controller = DynamicContextController(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="claude-sonnet-4.5-20250514"
)
history = [
{"role": "user", "content": "프로젝트 시작"},
{"role": "assistant", "content": "프로젝트 설정 완료"},
]
answer, stats = controller.execute(
query="현재 프로젝트 상태 요약해줘",
history=history,
retrieved_context="프로젝트: AI 챗봇 개발, 진행률: 60%, 멤버: 5명"
)
print(f"응답: {answer}")
print(f"통계: {stats}")
print(f"총 비용: ${controller.get_statistics()['estimated_cost']}")
자주 발생하는 오류와 해결책
오류 1: Context Overflow - 토큰 초과
# ❌ 오류: Maximum context length exceeded
messages 배열이 모델의 최대 컨텍스트를 초과
✅ 해결 1: auto_prune_messages 데코레이터 사용
def auto_prune_messages(max_tokens: int = 100000):
def decorator(func):
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
if hasattr(result, '__iter__') and not isinstance(result, str):
messages = list(result)
total = sum(len(m.get('content', '')) for m in messages)
if total > max_tokens * 1.5:
# 중요 메시지만 필터링
filtered = [
m for m in messages
if m.get('role') == 'system'
or len(m.get('content', '')) < 5000
]
return filtered[:20] # 최대 20개 메시지
return result
return wrapper
return decorator
✅ 해결 2: HolySheep AI의 자동 컨텍스트 관리 기능 활용
base_url 설정 시 자동으로 토큰 최적화 적용
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # 자동 최적화 활성화
)
API 호출 시 max_tokens 명시적 설정
response = client.chat.completions.create(
model="claude-sonnet-4.5-20250514",
messages=messages,
max_tokens=2048, # 응답 길이 제한으로 전체 컨텍스트 절약
extra_headers={
"X-Auto-Prune": "true", # HolySheep AI 전용 헤더
"X-Context-Budget": "128000" # 컨텍스트 예산 설정
}
)
오류 2: Memory Leak - 컨텍스트 누적
# ❌ 오류: 대화 반복 시 메모리 사용량 무한 증가
history.append()만 수행하고 정리하지 않음
✅ 해결: 계층적 메모리 구조 구현
class HierarchicalMemory:
def __init__(self):
self.episodic = [] # 최근 대화 (최근 10개)
self.working = [] # 작업 중 메모리
self.semantic = {} # 시맨틱 메모리 (키-값)
self.persistent = [] # 영구 기억
def add(self, role: str, content: str):
message = {"role": role, "content": content}
self.episodic.append(message)
# 자동 정리: Episodic이 일정 크기 초과 시
if len(self.episodic) > 10:
self._consolidate_episodic()
def _consolidate_episodic(self):
"""에피소딕 메모리를 작업 메모리로 압축"""
if len(self.episodic) <= 10:
return
# 최근 10개 제외한 메시지 요약
old_messages = self.episodic[:-10]
summary = self._create_summary(old_messages)
# 시맨틱 메모리에 저장
key = f"episode_{len(self.semantic)}"
self.semantic[key] = summary
# 에피소딕 정리
self.episodic = self.episodic[-10:]
def get_context(self, query: str = "") -> List[Dict]:
"""필요한 메모리만 선택적으로 로드"""
context = []
# 1. 영구 기억 항상 포함
context.extend(self.persistent)
# 2. 관련 시맨틱 메모리 검색
if query:
for key, content in self.semantic.items():
if any(word in content for word in query.split()[:3]):
context.append({"role": "system", "content": f"[기억] {content}"})
# 3. 최근 에피소딕
context.extend(self.episodic[-6:])
return context
사용
memory = HierarchicalMemory()
memory.add("user", "사용자 메시지...")
memory.add("assistant", "어시스턴트 응답...")
context = memory.get_context("검색 키워드")
오류 3: Rag Cache Miss - 검색 불일치
# ❌ 오류: RAG 검색 결과가 현재 쿼리와 관련 없음
임베딩 모델과 LLM의 의미 공간 불일치
✅ 해결: Hybrid Search + Re-ranking
from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np
class ImprovedRAG:
def __init__(self, client):
self.client = client
self.vector_store = {}
self.tfidf_store = {}
def index(self, doc_id: str, content: str, chunk_size: int = 500):
"""TF-IDF + 임베딩 이중 인덱싱"""
chunks = self.chunk_text(content, chunk_size)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc_id}_{i}"
# 임베딩 인덱스
embedding = self.client.embeddings.create(
model="text-embedding-3-small",
input=chunk
).data[0].embedding
self.vector_store[chunk_id] = {
"content": chunk,
"embedding": embedding,
"doc_id": doc_id
}
# TF-IDF 인덱스 (키워드 매칭용)
self.tfidf_store[chunk_id] = chunk
def search(self, query: str, top_k: int = 5) -> List[str]:
"""하이브리드 검색: 임베딩 + TF-IDF 결합"""
# 1. 임베딩 검색
query_embedding = self.client.embeddings.create(
model="text-embedding-3-small",
input=query
).data[0].embedding
semantic_scores = []
for chunk_id, data in self.vector_store.items():
sim = self.cosine_similarity(query_embedding, data["embedding"])
semantic_scores.append((sim, chunk_id))
# 2. TF-IDF 검색
if self.tfidf_store:
vectorizer = TfidfVectorizer()
docs = list(self.tfidf_store.values())
tfidf_matrix = vectorizer.fit_transform(docs + [query])
query_vec = tfidf_matrix[-1]
doc_vecs = tfidf_matrix[:-1]
tfidf_scores = []
for i, chunk_id in enumerate(self.tfidf_store.keys()):
score = (query_vec @ doc_vecs[i].T).toarray()[0, 0]
tfidf_scores.append((score, chunk_id))
else:
tfidf_scores = []
# 3. 스코어 결합 (0.7 임베딩 + 0.3 TF-IDF)
all_scores = {}
for score, chunk_id in semantic_scores:
all_scores[chunk_id] = all_scores.get(chunk_id, 0) + score * 0.7
for score, chunk_id in tfidf_scores:
all_scores[chunk_id] = all_scores.get(chunk_id, 0) + score * 0.3
# 4. 상위 결과 반환
sorted_chunks = sorted(all_scores.items(), key=lambda x: x[1], reverse=True)
return [self.vector_store[cid]["content"] for score, cid in sorted_chunks[:top_k]]
@staticmethod
def cosine_similarity(a, b):
dot = sum(x*y for x,y in zip(a,b))
norm_a = sum(x*x for x in a) ** 0.5
norm_b = sum(x*x for x in b) ** 0.5
return dot / (norm_a * norm_b + 1e-8)
@staticmethod
def chunk_text(text: str, size: int) -> list:
sentences = text.replace("!", "!\n").replace("?", "?\n").split("\n")
chunks, current = [], ""
for s in sentences:
if len(current) + len(s) <= size:
current += " " +