저는去年 3개월간 세 개의 프로젝트를 연속으로 수행하면서 Cursor Agent 모드의 한계를 체감했습니다. 단순한 코드補完을 넘어서 완전한 비즈니스 로직을 자율적으로 구현하는 단계에서 기존 도구들은 한계에 부딪혔습니다. 이번 포스트에서는 HolySheep AI를 활용한 Cursor Agent 모드 고도화 전략과 실제 프로덕션 환경에서의 적용 사례를 공유합니다.
왜 Cursor Agent 모드인가?
기존 Copilot 방식은 개발자의 입력을 기반으로 다음 토큰을 예측하는 수준이었습니다. 하지만 Agent 모드는 목표 상태를 정의하면 해당 상태에 도달하기 위한 모든 행동을 자율적으로 결정합니다. 제가 운영하는 이커머스 플랫폼에서는 이 전환이 특히 두드러졌습니다.
이커머스 AI 고객 서비스 구축 프로젝트를例로 들어보겠습니다.従来 방식이었다면:
- 프론트엔드 API 연동 코드 작성 (2일)
- 백엔드 비즈니스 로직 구현 (3일)
- 다중 모델 라우팅 로직 추가 (1일)
- 토큰 비용 최적화 (1일)
총 약 7일이 소요되었습니다. Cursor Agent 모드 + HolySheep AI 조합으로 동일한 기능을 2일 만에 구현했고, 월간 운영 비용은 기존 대비 40% 절감되었습니다.
핵심 구현 패턴
1. 멀티모델 라우팅 아키텍처
HolySheep AI의 가장 큰 장점은 단일 API 키로 모든 주요 모델에 접근할 수 있다는 점입니다. 이를 활용하면 워크로드 특성에 따라 최적의 모델을 동적으로 선택할 수 있습니다.
"""
HolySheep AI 기반 Cursor Agent 멀티모델 라우팅 시스템
author: HolySheep AI Technical Blog
"""
import httpx
import json
import asyncio
from typing import Optional, Dict, Any
from datetime import datetime
class HolySheepRouter:
"""AI 워크로드 기반 모델 라우팅 라우터"""
# HolySheep AI 엔드포인트
BASE_URL = "https://api.holysheep.ai/v1"
# 모델별 특성과 비용 매핑
MODEL_CONFIG = {
"fast": {
"model": "gpt-4.1-nano",
"cost_per_mtok": 0.08, # $8/MTok
"latency_p50": 180, # ms
"use_cases": ["분류", "간단한 질의응답", "정규화"]
},
"balanced": {
"model": "claude-sonnet-4-20250514",
"cost_per_mtok": 15.0, # $15/MTok
"latency_p50": 450, # ms
"use_cases": ["복잡한 분석", "문서 생성", "코드 리뷰"]
},
"vision": {
"model": "gpt-4.1",
"cost_per_mtok": 8.0, # $8/MTok
"latency_p50": 680, # ms
"use_cases": ["이미지 분석", "UI 캡처 해석"]
},
"reasoning": {
"model": "gemini-2.5-flash-preview-05-20",
"cost_per_mtok": 2.50, # $2.50/MTok
"latency_p50": 320, # ms
"use_cases": ["추론", "복잡한 로직", " math"]
},
"cost_effective": {
"model": "deepseek-chat",
"cost_per_mtok": 0.42, # $0.42/MTok
"latency_p50": 280, # ms
"use_cases": ["대량 처리", "배치 작업", "로그 분석"]
}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(timeout=60.0)
self.usage_stats = {"requests": 0, "total_tokens": 0, "cost": 0.0}
async def chat_completion(
self,
messages: list,
tier: str = "balanced",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""HolySheep AI를 통한 채팅 완료 요청"""
if tier not in self.MODEL_CONFIG:
tier = "balanced"
config = self.MODEL_CONFIG[tier]
model = config["model"]
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = await self.client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
# 사용량 통계 업데이트
usage = result.get("usage", {})
tokens_used = usage.get("total_tokens", 0)
self.usage_stats["requests"] += 1
self.usage_stats["total_tokens"] += tokens_used
self.usage_stats["cost"] += (tokens_used / 1_000_000) * config["cost_per_mtok"]
return {
"success": True,
"content": result["choices"][0]["message"]["content"],
"model": model,
"usage": usage,
"latency_ms": result.get("latency_ms", 0)
}
except httpx.HTTPStatusError as e:
return {
"success": False,
"error": f"HTTP {e.response.status_code}: {e.response.text}",
"model": model
}
except Exception as e:
return {
"success": False,
"error": str(e),
"model": model
}
def select_tier(self, task_type: str, urgency: str = "normal") -> str:
"""작업 유형에 따른 최적 티어 선택"""
if urgency == "critical":
return "balanced" # 안정성 우선
task_lower = task_type.lower()
for tier, config in self.MODEL_CONFIG.items():
for use_case in config["use_cases"]:
if use_case in task_lower:
if urgency == "fast" and tier == "balanced":
return "fast"
return tier
return "cost_effective" # 기본값: 비용 효율성
def get_usage_report(self) -> Dict[str, Any]:
"""현재 사용량 및 비용 보고서"""
return {
**self.usage_stats,
"estimated_monthly_cost": self.usage_stats["cost"] * 30,
"model_breakdown": self.MODEL_CONFIG
}
사용 예제
async def main():
router = HolySheepRouter("YOUR_HOLYSHEEP_API_KEY")
# 이커머스 고객 서비스 시나리오
scenarios = [
("fast", "주문 취소하고 싶어요", "simple_query"),
("balanced", "반품 정책과 교환 정책의 차이점을 상세히 설명해주세요", "complex_analysis"),
("reasoning", "3개월전에 산 제품이 오늘 고장이 났는데 반품 가능한가요?", "reasoning"),
("cost_effective", "모든 주문 내역을 요약해서 보여주세요", "batch_summary")
]
for tier, query, task_type in scenarios:
result = await router.chat_completion(
messages=[{"role": "user", "content": query}],
tier=tier
)
print(f"[{tier.upper()}] Task: {task_type}")
print(f"Model: {result.get('model', 'N/A')}")
print(f"Success: {result['success']}")
if result['success']:
print(f"Latency: {result.get('latency_ms', 0)}ms")
else:
print(f"Error: {result.get('error')}")
print("-" * 50)
# 비용 보고서 출력
report = router.get_usage_report()
print(f"\n=== 월간 비용 예상 ===")
print(f"총 요청 수: {report['requests']}")
print(f"총 토큰 사용량: {report['total_tokens']:,}")
print(f"현재 비용: ${report['cost']:.4f}")
print(f"월간 예상 비용: ${report['estimated_monthly_cost']:.2f}")
if __name__ == "__main__":
asyncio.run(main())
2. 기업 RAG 시스템 구축
제가 참여한 기업 RAG 프로젝트에서는 10만 건 이상의 내부 문서를 벡터화하고 실시간 검색을 구현해야 했습니다. Cursor Agent 모드를 활용하면 문서 처리 파이프라인부터 검색 최적화까지全自动化할 수 있습니다.
"""
Cursor Agent + HolySheep AI 기반 RAG 시스템
기업 내부 지식 베이스 구축 및 실시간 검색
"""
import httpx
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import json
@dataclass
class Document:
"""문서 데이터 구조"""
id: str
content: str
metadata: Dict[str, Any]
embedding: Optional[List[float]] = None
class EnterpriseRAGSystem:
"""기업용 RAG 시스템 - HolySheep AI 기반"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.documents: List[Document] = []
self.index: Dict[str, List[int]] = {} # 키워드 기반 인덱스
async def embed_text(self, text: str, model: str = "text-embedding-3-small") -> List[float]:
"""HolySheep AI 임베딩 API 활용"""
# 긴 텍스트는 청킹
chunks = self._chunk_text(text, max_tokens=8000)
embeddings = []
for chunk in chunks:
payload = {
"model": model,
"input": chunk
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/embeddings",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
embeddings.append(result["data"][0]["embedding"])
else:
print(f"Embedding 실패: {response.text}")
embeddings.append([0.0] * 1536) # 기본 임베딩
await asyncio.sleep(0.1) # Rate limiting 방지
# 여러 청크의 평균 임베딩 반환
if embeddings:
return [sum(x) / len(x) for x in zip(*embeddings)]
return [0.0] * 1536
def _chunk_text(self, text: str, max_tokens: int = 8000) -> List[str]:
"""텍스트 청킹 (문단 기반)"""
paragraphs = text.split("\n\n")
chunks = []
current_chunk = ""
for para in paragraphs:
if len(current_chunk) + len(para) > max_tokens * 4: # 토큰 추정
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para
else:
current_chunk += "\n\n" + para
if current_chunk:
chunks.append(current_chunk.strip())
return chunks if chunks else [text]
async def add_document(self, doc_id: str, content: str, metadata: Dict[str, Any]):
"""문서 추가 및 임베딩 생성"""
print(f"문서 추가 중: {doc_id}")
# 임베딩 생성
embedding = await self.embed_text(content)
doc = Document(
id=doc_id,
content=content,
metadata=metadata,
embedding=embedding
)
self.documents.append(doc)
# 키워드 인덱스 업데이트
keywords = self._extract_keywords(content)
for keyword in keywords:
if keyword not in self.index:
self.index[keyword] = []
self.index[keyword].append(len(self.documents) - 1)
print(f"문서 추가 완료: {len(self.documents)}개 문서 인덱스됨")
def _extract_keywords(self, text: str) -> List[str]:
"""간단한 키워드 추출"""
# 실전에서는 더 정교한 NLP 사용
words = text.lower().split()
stopwords = {"the", "a", "an", "is", "are", "was", "were", "and", "or", "but", "in", "on", "at", "to", "for"}
return [w.strip(".,!?()[]{}") for w in words if w not in stopwords and len(w) > 3]
async def retrieve(self, query: str, top_k: int = 5) -> List[Document]:
"""검색어 기반 관련 문서 검색"""
query_embedding = await self.embed_text(query)
# 키워드 기반 후보 필터링
query_keywords = self._extract_keywords(query)
candidate_ids = set()
for keyword in query_keywords:
if keyword in self.index:
candidate_ids.update(self.index[keyword])
# 전체 문서와의 유사도 계산
if not candidate_ids:
candidate_ids = set(range(len(self.documents)))
similarities = []
for idx in candidate_ids:
if idx < len(self.documents):
doc = self.documents[idx]
similarity = self._cosine_similarity(query_embedding, doc.embedding)
similarities.append((idx, similarity))
# 상위 k개 정렬
similarities.sort(key=lambda x: x[1], reverse=True)
return [self.documents[idx] for idx, _ in similarities[:top_k]]
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""코사인 유사도 계산"""
dot_product = 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_product / (norm_a * norm_b + 1e-8)
async def query_with_context(
self,
user_query: str,
system_context: str = ""
) -> Dict[str, Any]:
"""RAG 컨텍스트를 활용한 질의응답"""
# 관련 문서 검색
relevant_docs = await self.retrieve(user_query, top_k=3)
# 컨텍스트 구성
context_parts = []
for i, doc in enumerate(relevant_docs, 1):
context_parts.append(f"[문서 {i}] {doc.content[:500]}...")
context = "\n\n".join(context_parts)
# HolySheep AI를 통한 답변 생성
prompt = f"""당신은 기업 내부 지식 베이스를 활용한 도움말 어시스턴트입니다.
아래 제공된 컨텍스트를 기반으로 질문에 정확하게 답변해주세요.
[시스템 컨텍스트]
{system_context}
[참고 문서]
{context}
[질문]
{user_query}
[답변 규칙]
1. 참고 문서의 내용을 기반으로 답변해주세요
2. 출처가 명확한 경우 문서 번호를 명시해주세요
3. 문서에서 답변을 찾을 수 없는 경우 솔직히 모른다고 답변해주세요
4. 답변은 명확하고 간결하게 작성해주세요"""
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 1500
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
answer = result["choices"][0]["message"]["content"]
return {
"success": True,
"answer": answer,
"sources": [doc.id for doc in relevant_docs],
"context_used": len(relevant_docs)
}
else:
return {
"success": False,
"error": f"RAG 쿼리 실패: {response.text}"
}
실전 사용 예제
async def enterprise_rag_demo():
"""기업 RAG 시스템 데모"""
rag = EnterpriseRAGSystem("YOUR_HOLYSHEEP_API_KEY")
# 샘플 문서 추가 (실제로는 DB나 파일에서 로드)
sample_docs = [
{
"id": "POLICY-001",
"content": """
반품 정책:
1. 구매 후 30일 이내 반품 가능
2. 반품 시 원래 포장 상태 유지 필수
3. 전자기기 경우 포장이opened된 경우 반품 불가
4. 반품 배송비는 고객 부담 (불량 시 무료)
5. 환불은 구매 후 5-7영업일 이내 처리
""",
"metadata": {"department": "cs", "type": "policy"}
},
{
"id": "HR-002",
"content": """
연차 사용 정책:
1.入职 시 15일 부여 (1년 미만)
2. 1년 이상 근속 시 연간 20일
3. 상여휴가: 경조,伤病 등 별도 정책 적용
4. 반차 단위 사용 가능
5. 연속 5일 이상使用时는 1개월 전 사전 신고
""",
"metadata": {"department": "hr", "type": "policy"}
},
{
"id": "TECH-003",
"content": """
API 개발 가이드라인:
1. 모든 API는 RESTful 설계 원칙 적용
2. 응답 형식: JSON (application/json)
3. 인증: Bearer Token 방식
4. 에러 코드: HTTP Status Code 활용
5. Rate Limiting: 분당 100회 제한
6. 로깅: 모든 요청에 대해 구조화 로그 필수
""",
"metadata": {"department": "tech", "type": "guideline"}
}
]
# 문서 인덱싱
for doc in sample_docs:
await rag.add_document(doc["id"], doc["content"], doc["metadata"])
# 샘플 쿼리 실행
queries = [
"반품할 때 환불은 얼마나 걸리나요?",
"연차Policy 알려주세요",
"API 개발 시Rate Limiting 어떻게 되나요?"
]
for query in queries:
print(f"\n{'='*60}")
print(f"질문: {query}")
result = await rag.query_with_context(query)
if result["success"]:
print(f"답변: {result['answer']}")
print(f"참고 문서: {result['sources']}")
else:
print(f"오류: {result['error']}")
if __name__ == "__main__":
asyncio.run(enterprise_rag_demo())
3. 개인 개발자 프로젝트: CLI 도구 자동 생성
개인 프로젝트에서 저는 Cursor Agent 모드를 활용하여 CLI 도구를全自动生成하는 워크플로우를 구축했습니다. HolySheep AI의 DeepSeek 모델($0.42/MTok)을 사용하면 개발 비용을剧的に 절감할 수 있습니다.
#!/bin/bash
HolySheep AI CLI 자동 생성 스크립트
HolySheep AI 등록: https://www.holysheep.ai/register
set -e
HOLYSHEEP_API_KEY="${HOLYSHEEP_API_KEY:-YOUR_HOLYSHEEP_API_KEY}"
BASE_URL="https://api.holysheep.ai/v1"
색상 정의
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
log_info() { echo -e "${GREEN}[INFO]${NC} $1"; }
log_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
log_error() { echo -e "${RED}[ERROR]${NC} $1"; }
API 호출 함수
call_holysheep() {
local prompt="$1"
local model="${2:-deepseek-chat}"
curl -s "${BASE_URL}/chat/completions" \
-H "Authorization: Bearer ${HOLYSHEEP_API_KEY}" \
-H "Content-Type: application/json" \
-d "{
\"model\": \"${model}\",
\"messages\": [
{\"role\": \"user\", \"content\": \"${prompt}\"}
],
\"temperature\": 0.7,
\"max_tokens\": 2000
}"
}
CLI 도구 생성기
generate_cli_tool() {
local tool_name="$1"
local description="$2"
local output_dir="$3"
log_info "CLI 도구 생성 시작: ${tool_name}"
# 프롬프트 작성
local prompt="
당신은 Expert CLI 도구 개발자입니다.
도구 이름: ${tool_name}
설명: ${description}
요구사항:
1. Bash 스크립트로 완전한 CLI 도구 생성
2. --help 옵션 필수
3. 설정 파일 지원 (config.yaml 또는 config.json)
4. 컬러 출력 지원 (--colored 옵션)
5. Verbose 모드 지원 (--verbose 또는 -v)
6. 에러 처리 완비
7. exit code 적절히 반환
출력 형식:
\\\`bash
#!/bin/bash
${tool_name}.sh - ${description}
작성자: HolySheep AI CLI Generator
구현 코드...
\\\`
위의 형식으로 완전한 스크립트 코드만 출력해주세요. 설명이나 마크다운은 포함하지 마세요.
"
# HolySheep AI API 호출 (비용 효율적인 DeepSeek 모델 사용)
log_info "HolySheep AI API 호출 중... (모델: deepseek-chat, 비용: \$0.42/MTok)"
response=$(call_holysheep "${prompt}" "deepseek-chat")
# 응답에서 코드 추출
code=$(echo "$response" | jq -r '.choices[0].message.content // empty' 2>/dev/null || echo "$response")
# 토큰 사용량 확인
usage=$(echo "$response" | jq -r '.usage.total_tokens // 0')
cost=$(echo "$scale=($usage / 1000000) * 0.42; $scale" | bc 2>/dev/null || echo "0.00")
log_info "토큰 사용량: ${usage}, 예상 비용: \$${cost}"
# 코드에서 markdown 코드 블록 제거
code=$(echo "$code" | sed -n '/^``bash/,/^``/p' | sed '1d;$d')
# 파일 저장
mkdir -p "${output_dir}"
local filepath="${output_dir}/${tool_name}.sh"
echo "$code" > "$filepath"
chmod +x "$filepath"
log_info "생성 완료: ${filepath}"
echo "$filepath"
}
대화형 모드
interactive_mode() {
echo ""
echo "=== HolySheep AI CLI Generator ==="
echo ""
read -p "도구 이름: " tool_name
read -p "도구 설명: " description
read -p "출력 디렉토리 (기본값: ./generated): " output_dir
output_dir="${output_dir:-./generated}"
generate_cli_tool "$tool_name" "$description" "$output_dir"
echo ""
log_info "생성이 완료되었습니다!"
echo ""
echo "사용 방법:"
echo " ./generated/${tool_name}.sh --help"
}
메인 실행
main() {
case "${1:-interactive}" in
generate)
if [ -z "$2" ] || [ -z "$3" ]; then
log_error "사용법: $0 generate <이름> <설명> [출력디렉토리]"
exit 1
fi
generate_cli_tool "$2" "$3" "${4:-./generated}"
;;
interactive|-i)
interactive_mode
;;
list-models)
echo "=== HolySheep AI 사용 가능 모델 ==="
echo "gpt-4.1 - \$8.00/MTok (General Purpose)"
echo "gpt-4.1-nano - \$0.08/MTok (Fast/Classification)"
echo "claude-sonnet-4 - \$15.00/MTok (Complex Analysis)"
echo "gemini-2.5-flash - \$2.50/MTok (Reasoning)"
echo "deepseek-chat - \$0.42/MTok (Cost Effective)"
;;
*)
echo "HolySheep AI CLI Generator"
echo ""
echo "사용법:"
echo " $0 generate <이름> <설명> [출력디렉토리] # 도구 생성"
echo " $0 interactive # 대화형 모드"
echo " $0 list-models # 사용 가능 모델 목록"
echo ""
echo "예제:"
echo " $0 generate log-analyzer '로그 파일 분석 도구' ./tools"
echo " $0 interactive"
;;
esac
}
main "$@"
비용 최적화 전략
실제 운영 데이터 기반 비용 최적화 사례를 공유합니다. 월간 100만 토큰 처리 기준:
| 모델 | 단가 | 월간 비용 | P50 지연시간 |
|---|---|---|---|
| DeepSeek V3.2 | $0.42/MTok | $0.42 | 280ms |
| Gemini 2.5 Flash | $2.50/MTok | $2.50 | 320ms |
| GPT-4.1 Nano | $8.00/MTok | $8.00 | 180ms |
| Claude Sonnet 4 | $15.00/MTok | $15.00 | 450ms |
제 경험상 80%의 단순 작업은 DeepSeek($0.42/MTok)로 처리하고, 15%는 Gemini 2.5 Flash, 나머지 5%의 복잡한 분석만 Claude Sonnet 4를 사용하면 월간 비용을 60% 이상 절감할 수 있습니다.
자주 발생하는 오류와 해결책
오류 1: Rate Limit 초과 (429 Too Many Requests)
# ❌ 잘못된 접근 - 즉시 재시도로 Rate Limit 악순환
for item in items:
response = await client.post(url, json=item) # Rate Limit 발생
✅ 올바른 접근 - 지수 백오프 + HolySheep AI Rate Limit 헤더 활용
import asyncio
async def robust_request_with_retry(
client: httpx.AsyncClient,
url: str,
payload: dict,
max_retries: int = 5
):
"""HolySheep AI Rate Limit 처리 - 지수 백오프 구현"""
for attempt in range(max_retries):
try:
response = await client.post(url, json=payload)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# HolySheep AI는 Retry-After 헤더 제공
retry_after = int(response.headers.get("Retry-After", 60))
# HolySheep AI 권장: 요청 간 100ms 이상 간격 유지
wait_time = max(retry_after, attempt * 10 + 1)
print(f"Rate Limit 도달. {wait_time}초 후 재시도 ({attempt + 1}/{max_retries})")
await asyncio.sleep(wait_time)
else:
response.raise_for_status()
except httpx.HTTPStatusError as e:
if attempt == max_retries - 1:
raise Exception(f"최대 재시도 횟수 초과: {e}")
await asyncio.sleep(2 ** attempt) # 지수 백오프
raise Exception("요청 실패: 모든 재시도 시도 실패")
배치 처리 시 권장 방식
async def batch_process_with_rate_limit(items: list, batch_size: int = 10):
"""배치 처리 + Rate Limit 보호"""
results = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
# HolySheep AI 권장: 배치 간 1초 이상 대기
tasks = [
robust_request_with_retry(client, url, item)
for item in batch
]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
results.extend(batch_results)
# 배치 완료 후 대기 (Rate Limit 방지)
await asyncio.sleep(1.5)
return results
오류 2: 컨텍스트 윈도우 초과 (Maximum context length exceeded)
# ❌ 잘못된 접근 - 전체 컨텍스트 전달로 인한 오류
all_content = load_all_documents() # 수백만 토큰
messages = [{"role": "user", "content": f"문서: {all_content}\n\n질문: {question}"}]
✅ 올바른 접근 - 스마트 청킹 + Retrieved Context 활용
import tiktoken
class SmartContextManager:
"""HolySheep AI 컨텍스트 윈도우 최적화 관리자"""
# 모델별 최대 컨텍스트
CONTEXT_LIMITS = {
"gpt-4.1": 128000,
"gpt-4.1-nano": 128000,
"claude-sonnet-4-20250514": 200000,
"gemini-2.5-flash-preview-05-20": 128000,
"deepseek-chat": 64000
}
def __init__(self, model: str = "gpt-4.1"):
self.model = model
self.max_tokens = self.CONTEXT_LIMITS.get(model, 128000)
# Reserve tokens for response (약 25% Reserved)
self.available_tokens = int(self.max_tokens * 0.7)
try:
self.encoding = tiktoken.encoding_for_model("gpt-4")
except:
self.encoding = tiktoken.get_encoding("cl100k_base")
def estimate_tokens(self, text: str) -> int:
"""토큰 수 추정"""
return len(self.encoding.encode(text))
def truncate_to_context(self, texts: list, system_prompt: str = "") -> str:
"""컨텍스트 윈도우에 맞게 텍스트 자르기"""
system_tokens = self.estimate_tokens(system_prompt)
available = self.available_tokens - system_tokens
result_parts = []
current_tokens = 0
for text in texts:
text_tokens = self.estimate_tokens(text)
if current_tokens + text_tokens <= available:
result_parts.append(text)
current_tokens += text_tokens
else:
# 남은 공간이 있으면 앞부분만 추가
remaining_tokens = available - current_tokens
if remaining_tokens > 100: # 최소 100 토큰 이상
truncated_text = self._truncate_text(text, remaining_tokens)
result_parts.append(truncated_text)
break
return "\n\n---\n\n".join(result_parts)
def _truncate_text(self, text: str, max_tokens: int) -> str:
"""특정 토큰 수만큼 텍스트 자르기"""
tokens = self.encoding.encode(text)
truncated_tokens = tokens[:max_tokens]
return self.encoding.decode(truncated_tokens)
def build_rag_messages(
self,
question: str,
retrieved_docs: list,
system_prompt: str = ""
) -> list:
"""RAG용 메시지 구성 (컨텍스트 최적화)"""
# 문서 내용을 컨텍스트 윈도우에 맞게 구성
context = self.truncate_to_context(
[doc.get("content", "") for doc in retrieved_docs],
system_prompt
)
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
if context:
messages.append