전통적인 규칙 기반 행동 트리(Rule-Based Behavior Tree)는 게임 NPC의 의사결정을 담당해왔습니다. 하지만 개발자들이 "상태가 너무 많아서 관리 불가능하다", "새로운 상황 추가할 때마다 트리가 폭발적으로 성장한다", "외관상 자연스러운데 왜 이렇게 비현실적으로 행동하지?"라는 고민을incessantly 합니다.
저는 3인칭 액션 게임에서 5개월간 LLM 기반 NPC AI를 구현하며 Rule Engine에서 LLM으로의 마이그레이션을 직접 경험했습니다. 이 글에서는 그 과정에서 얻은 실질적인 데이터와 코드를 바탕으로 마이그레이션의 실제 비용과 수익을 분석합니다.
HolySheep AI vs 공식 API vs 타 릴레이 서비스 비교
| 비교 항목 | HolySheep AI | 공식 OpenAI/Anthropic API | 타 릴레이 서비스 |
|---|---|---|---|
| 결제 방식 | 로컬 결제 지원 (해외 신용카드 불필요) | 해외 신용카드 필수 | 다양하지만 대부분 해외 결제 필요 |
| 지원 모델 | GPT-4.1, Claude Sonnet, Gemini 2.5, DeepSeek V3.2 등 통합 | 단일 공급사 모델만 | 제한된 모델 선택 |
| GPT-4.1 가격 | $8/MTok (입력), $24/MTok (출력) | $8/MTok (입력), $24/MTok (출력) | $10-15/MTok |
| Claude Sonnet 4.5 가격 | $15/MTok (입력), $75/MTok (출력) | $15/MTok (입력), $75/MTok (출력) | $18-22/MTok |
| Gemini 2.5 Flash 가격 | $2.50/MTok (입력), $10/MTok (출력) | $2.50/MTok (입력), $10/MTok (출력) | $4-6/MTok |
| DeepSeek V3.2 가격 | $0.42/MTok (입력), $1.68/MTok (출력) | 지원 안함 | 제한적 |
| 평균 지연 시간 | 800-1200ms | 600-1000ms | 1000-2000ms |
| 동시 접속 제한 | 유연한 할당량 정책 | 엄격한 RPM/TPM 제한 | 다양함 |
| 개발자 친숙도 | OpenAI 호환 API, 빠른 시작 가이드 | 공식 문서中心 | 다양함 |
| 무료 크레딧 | 가입 시 제공 | $5 크레딧 (신규) | varied |
이런 팀에 적합 / 비적절
✅ LLM 기반 NPC AI에 적합한 팀
- 오픈 월드나 샌드박스 게임 개발팀: 다양한 상황별 NPC 반응이 필요한 경우
- 다중 에이전트 시뮬레이션: 도시-sim, 경제 시뮬레이션 등 다수 AI 캐릭터 협업/경쟁
- 대화형 스토리 게임: 플레이어 행동에 실시간으로 반응하는 NPC 필요
- QA 자동화 테스트: 다양한 게임 시나리오를 AI가 자동 생성하고 테스트
- 신속한 프로토타이핑: 규칙 기반 시스템 구축 없이 AI에 행동 로직 위임
❌ LLM 기반 NPC AI에 부적절한 팀
- 저예산 독립 게임팀: API 호출 비용이 수익 모델 대비 과도
- 반복적 전투/퍼즐 AI: 정해진 패턴으로 충분한 경우 (예: 턴제 RPG)
- 엄격한 응답 시간 요구: 100ms 이내 결정 필요 (예: 온라인 pvp 실시간 전투)
- 완전한 결정론적 게임: 랜덤 요소 없는 예측 가능한 AI 동작 필요
- 심각한 온라인 의존성 문제: 오프라인 환경 필수인 경우
가격과 ROI 분석
저의 실제 프로젝트 데이터를 기준으로 ROI를 분석하겠습니다.
| 항목 | 규칙 기반 트리 (기존) | LLM 기반 (마이그레이션 후) |
|---|---|---|
| 개발 시간 (1인 6개월) | 약 1,200시간 | 약 600시간 |
| 행동 규칙 코드 라인 | 약 45,000줄 | 약 8,000줄 + 프롬프트 |
| 새 상황 추가 시간 | 평균 16시간/케이스 | 평균 2시간/케이스 |
| 월간 API 비용 | $0 | 약 $800-1,200 (활성 세션 기준) |
| QA 테스트覆盖率 | 약 60% | 약 85% |
| 플레이어 피드백 (자연스러움) | 평점 3.2/5 | 평점 4.4/5 |
비용 절감 포인트
# DeepSeek V3.2 사용 시 월간 비용 비교
HolySheep AI 단일 API 키로 모델 전환 가능
상황: 월간 100만 토큰 입력, 50만 토큰 출력 시뮬레이션
GPT-4.1 사용 시
gpt4_cost = (1_000_000 * 8 + 500_000 * 24) / 1_000_000 # $20/MTok equivalent
print(f"GPT-4.1 월간 비용: ${gpt4_cost * 20:.2f}") # $400
Claude Sonnet 4.5 사용 시
claude_cost = (1_000_000 * 15 + 500_000 * 75) / 1_000_000
print(f"Claude Sonnet 4.5 월간 비용: ${claude_cost * 20:.2f}") # $525
DeepSeek V3.2 사용 시 (대부분의 NPC 대화에서 충분한 품질)
deepseek_cost = (1_000_000 * 0.42 + 500_000 * 1.68) / 1_000_000
print(f"DeepSeek V3.2 월간 비용: ${deepseek_cost * 20:.2f}") # $46.8
비용 절감율: GPT-4 대비 88% 절감
savings = (400 - 46.8) / 400 * 100
print(f"DeepSeek V3.2 절감율: {savings:.1f}%")
왜 HolySheep AI를 선택해야 하나
1. 단일 API 키로 모델 전환 유연성
게임의 상황에 따라 AI 모델을 유연하게 전환해야 합니다:
- 중요 NPC 대화: GPT-4.1 또는 Claude Sonnet 4.5
- 일상 NPC 행동: Gemini 2.5 Flash
- 배경 npc 대규모 시뮬레이션: DeepSeek V3.2
HolySheep AI는 지금 가입하면 이 모든 모델을 단일 API 키로 관리할 수 있습니다. 여러 공급사의 API 키를 각각 관리하는 번거로움과 결제 복잡성이 해소됩니다.
2. 로컬 결제 지원
해외 신용카드 없이도 원활하게 결제할 수 있어 한국, 일본, 동남아시아 개발팀에게 최적입니다. 개발 초기에는 소규모 테스트용으로 시작했다가 게임 출시 후 트래픽 증가에 따라 유연하게 확장할 수 있습니다.
3. 비용 최적화
DeepSeek V3.2의 경우 $0.42/MTok 입력으로 기존 모델 대비 최대 95% 비용 절감이 가능합니다. 배경 NPC처럼 대량으로 처리해야 하지만 품질 요구가 상대적으로 낮은 작업에 적합합니다.
구현 아키텍처
하이브리드 행동 트리 설계
import os
import json
from openai import OpenAI
HolySheep AI 설정 - 단일 API 키로 모든 모델 통합
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class HybridBehaviorTree:
"""규칙 기반 트리 + LLM 하이브리드 NPC AI 시스템"""
def __init__(self):
self.rule_engine = RuleBasedBehaviors()
self.llm_prompts = self._load_prompts()
def _load_prompts(self):
return {
"complex_decision": """당신은 게임 NPC입니다.
현재 상황: {situation}
NPC 성격: {personality}
NPC 목표: {goal}
최근 대화: {recent_dialogue}
다음 행동 중 가장 적절한 것을 선택하고 이유를 설명하세요:
1. 직접적으로 응답
2. 다른 NPC에게 도움 요청
3. 거리를 두며 관찰
4. 물리적으로 반응 (도망/공격/방어)
5. 도구를 사용
응답 형식: {{"action": "선택번호", "reasoning": "이유", "dialogue": "대사"}}""",
"emotional_response": """NPC 감정 상태: {emotion}
플레어어 행동: {player_action}
현재 장소: {location}
이 상황에서 NPC가 느끼는 감정과 적절한 반응을 생성하세요."""
}
def decide_action(self, npc_id, context):
"""행동 결정 로직"""
# 1단계: 규칙 엔진으로 빠른 결정
rule_result = self.rule_engine.evaluate(npc_id, context)
if rule_result.confidence > 0.8:
# 규칙 엔진이 높은 확신도 있으면 즉시 실행
return rule_result.action
# 2단계: 복잡한 상황은 LLM에 위임
if rule_result.confidence < 0.5 or context.novel_situation:
return self._llm_decision(npc_id, context)
# 3단계: 하이브리드 - 규칙 기반으로 후보 생성, LLM으로 최종 선택
candidates = self.rule_engine.generate_candidates(npc_id, context)
return self._llm_select_best(npc_id, context, candidates)
def _llm_decision(self, npc_id, context):
"""LLM 기반 복잡한 의사결정"""
# 중요도에 따라 모델 선택
if context.situation_importance == "critical":
model = "gpt-4.1"
elif context.situation_importance == "important":
model = "claude-sonnet-4-5"
else:
model = "gemini-2.5-flash" # 비용 효율적
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": self.llm_prompts["complex_decision"]},
{"role": "user", "content": json.dumps(context.to_dict(), ensure_ascii=False)}
],
temperature=0.7,
max_tokens=500
)
return json.loads(response.choices[0].message.content)
def _llm_select_best(self, npc_id, context, candidates):
"""규칙 엔진 후보 중 LLM이 최적 선택"""
prompt = f"""현재 상황: {context.description}
가능한 행동 후보:
{json.dumps(candidates, ensure_ascii=False, indent=2)}
위 후보 중 가장 적절한 행동을 선택하고 이유를 설명하세요."""
response = client.chat.completions.create(
model="gemini-2.5-flash", # 빠른 선택에는 가성비 모델
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=200
)
return {"action": response.choices[0].message.content, "source": "hybrid"}
class RuleBasedBehaviors:
"""기존 규칙 기반 행동 시스템 - 빠른 결정용"""
def __init__(self):
self.behavior_trees = {}
self.behavior_trees["aggressive"] = self._build_aggressive_tree()
self.behavior_trees["defensive"] = self._build_defensive_tree()
self.behavior_trees["neutral"] = self._build_neutral_tree()
def evaluate(self, npc_id, context):
"""규칙 기반 평가 - 1ms 내 응답"""
npc_type = context.npc.personality_type
if npc_type in self.behavior_trees:
return self.behavior_trees[npc_type].execute(context)
return BehaviorResult(action="idle", confidence=0.3)
def generate_candidates(self, npc_id, context):
"""규칙 기반으로 행동 후보 생성"""
return [
{"id": 1, "action": "greet", "weight": 0.8},
{"id": 2, "action": "ignore", "weight": 0.6},
{"id": 3, "action": "ask_question", "weight": 0.7}
]
def _build_aggressive_tree(self):
return BehaviorTree([
Condition("player_in_range", lambda ctx: ctx.distance < 5),
Sequence([
Action("chase_player"),
Condition("can_attack", lambda ctx: ctx.cooldown_ready),
Action("attack")
])
])
def _build_defensive_tree(self):
return BehaviorTree([
Condition("health_low", lambda ctx: ctx.health < 30),
Selector([
Sequence([
Action("find_cover"),
Action("heal")
]),
Action("retreat")
])
])
def _build_neutral_tree(self):
return BehaviorTree([
Selector([
Sequence([
Condition("player_approaching", True),
Action("greet")
]),
Action("idle")
])
])
class BehaviorTree:
def __init__(self, nodes):
self.nodes = nodes
def execute(self, context):
for node in self.nodes:
result = node.evaluate(context)
if result.success:
return result
return BehaviorResult(action="idle", confidence=0.5)
사용 예시
if __name__ == "__main__":
ai_system = HybridBehaviorTree()
# 테스트 컨텍스트
context = GameContext(
npc=NPC(name="村长", personality_type="neutral", personality="겁이 많고 신중한"),
player_action="asked_about_secret",
situation="위험한 보스를 발견했다",
location="작은 마을",
distance=3.5,
health=85,
novel_situation=True,
situation_importance="important"
)
result = ai_system.decide_action("npc_001", context)
print(f"선택된 행동: {result}")
배경 NPC 대규모 시뮬레이션용 배치 처리
import asyncio
import aiohttp
import json
from datetime import datetime
from collections import defaultdict
class BatchNPCSimulator:
"""배경 NPC 대량 시뮬레이션 - 비용 최적화 버전"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.session = None
self.costs = defaultdict(float)
async def initialize(self):
"""aiohttp 세션 초기화"""
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
async def close(self):
if self.session:
await self.session.close()
async def simulate_npc_batch(self, npcs: list, tick: int):
"""한 프레임에서 여러 NPC 행동 동시 처리"""
# DeepSeek V3.2 사용 - 비용 효율적
# 프롬프트 최적화로 토큰 사용량 최소화
prompt_template = """NPC:{name}
성격:{personality}
상황:{situation}
주변NPC:{nearby}
[{tick}틱] 행동:"""
messages = []
for npc in npcs[:50]: # 배치당 최대 50개
nearby = [n.name for n in npcs if n.id != npc.id and n.position.distance_to(npc.position) < 10]
content = prompt_template.format(
name=npc.name,
personality=npc.personality,
situation=npc.current_situation,
nearby=", ".join(nearby[:3]) if nearby else "없음",
tick=tick
)
messages.append({
"custom_id": npc.id,
"method": "POST",
"url": "/chat/completions",
"body": {
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": content}],
"max_tokens": 50, # 짧은 응답으로 토큰 절감
"temperature": 0.5
}
})
# HolySheep AI 배치 API 활용
async with self.session.post(
f"{self.base_url}/batch",
json={"input_file_content": json.dumps(messages, ensure_ascii=False)},
params={"completion_window": "24h"}
) as response:
result = await response.json()
return result
async def run_city_simulation(self, npcs: list, total_ticks: int):
"""도시 전체 NPC 시뮬레이션 실행"""
await self.initialize()
start_time = datetime.now()
all_results = []
for tick in range(total_ticks):
print(f"[{tick}/{total_ticks}] 시뮬레이션 진행 중...")
batch_results = await self.simulate_npc_batch(npcs, tick)
all_results.extend(batch_results.get("results", []))
# 100 틱마다 비용 보고
if tick % 100 == 0:
self._report_costs(tick, len(all_results))
# Rate limiting 방지
await asyncio.sleep(0.1)
await self.close()
total_cost = sum(self.costs.values())
print(f"\n시뮬레이션 완료!")
print(f"총 처리: {len(all_results)} NPC 행동")
print(f"총 비용: ${total_cost:.2f}")
print(f"평균 비용/NPC: ${total_cost/len(npcs):.4f}")
return all_results
def _report_costs(self, tick: int, total_results: int):
"""비용 보고"""
print(f"\n [{tick}틱] 현재 비용:")
for model, cost in self.costs.items():
print(f" {model}: ${cost:.4f}")
print(f" 합계: ${sum(self.costs.values()):.4f}")
실행 예시
async def main():
from dataclasses import dataclass
from typing import List
@dataclass
class NPC:
id: str
name: str
personality: str
current_situation: str
position: any
# 테스트용 NPC 데이터 (실제로는 DB 또는 파일에서 로드)
test_npcs = [
NPC(id=f"npc_{i}", name=f"주민{i}", personality="평화로운",
current_situation="시장에 있음", position=None)
for i in range(100)
]
simulator = BatchNPCSimulator(api_key="YOUR_HOLYSHEEP_API_KEY")
results = await simulator.run_city_simulation(test_npcs, total_ticks=1000)
if __name__ == "__main__":
asyncio.run(main())
자주 발생하는 오류와 해결책
오류 1: "Rate limit exceeded" - 동시 요청 초과
# 문제: 동시에 여러 NPC의 행동을 결정할 때 Rate Limit 발생
해결: 요청 큐와 재시도 로직 구현
import time
from tenacity import retry, stop_after_attempt, wait_exponential
class RateLimitedClient:
"""Rate Limit을 자동으로 처리하는 클라이언트 래퍼"""
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.api_key = api_key
self.request_times = []
self.max_requests_per_minute = 500 # HolySheep 기본 제한
def _wait_for_rate_limit(self):
"""과거 1분간의 요청 수 확인 및 대기"""
current_time = time.time()
cutoff_time = current_time - 60
# 1분 이전의 요청 기록 제거
self.request_times = [t for t in self.request_times if t > cutoff_time]
if len(self.request_times) >= self.max_requests_per_minute:
# 가장 오래된 요청 이후 1분이 지나야 재요청 가능
oldest = min(self.request_times)
wait_time = 60 - (current_time - oldest) + 1
print(f"Rate limit 대기: {wait_time:.1f}초")
time.sleep(wait_time)
self.request_times.append(time.time())
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def chat_completion(self, messages, model="gemini-2.5-flash"):
"""재시도 로직이 포함된 Chat Completion"""
self._wait_for_rate_limit()
try:
response = self._make_request(messages, model)
return response
except Exception as e:
if "rate_limit" in str(e).lower():
print(f"Rate limit 발생, 재시도 중... (남은 시도: {3 - self._retry_state.attempt_number})")
raise
raise
def _make_request(self, messages, model):
"""실제 API 요청"""
import aiohttp
import asyncio
async def _async_request():
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={"model": model, "messages": messages, "max_tokens": 200}
) as resp:
if resp.status == 429:
raise Exception("Rate limit exceeded")
return await resp.json()
return asyncio.run(_async_request())
사용 예시
client = RateLimitedClient(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
여러 NPC 동시 처리
tasks = [client.chat_completion([{"role": "user", "content": f"NPC_{i} 행동"})]) for i in range(100)]
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
results = list(executor.map(lambda t: client.chat_completion(*t), tasks))
오류 2: "Invalid response format" - LLM 응답 파싱 실패
# 문제: LLM이 요청한 JSON 형식 대신 일반 텍스트 반환
해결: 강력한 파싱 및 폴백 로직 구현
import json
import re
from typing import Optional, Dict, Any
class RobustResponseParser:
"""다양한 LLM 응답을 안정적으로 파싱"""
def __init__(self, client):
self.client = client
self.default_response = {
"action": "idle",
"dialogue": "...",
"reasoning": "파싱 실패로 기본값 사용"
}
def get_behavior(self, npc_id: str, context: str, max_retries: int = 3) -> Dict[str, Any]:
"""재시도 로직이 포함된 행동 결정"""
prompt = f"""현재 상황: {context}
응답은 반드시 다음 JSON 형식이어야 합니다:
{{"action": "행동", "dialogue": "대사", "reasoning": "이유"}}
가능한 행동: attack, defend, flee, talk, idle, trade, help
JSON 응답:"""
for attempt in range(max_retries):
try:
response = self.client.chat_completion([
{"role": "user", "content": prompt}
])
parsed = self._parse_json_response(response)
if parsed:
return parsed
except Exception as e:
print(f"파싱 시도 {attempt + 1} 실패: {e}")
# 모든 시도 실패 시 기본값 반환
print(f"NPC {npc_id}: 기본 행동으로 폴백")
return self.default_response.copy()
def _parse_json_response(self, response: str) -> Optional[Dict[str, Any]]:
"""다양한 형식의 응답에서 JSON 추출"""
if isinstance(response, dict):
content = response.get("choices", [{}])[0].get("message", {}).get("content", "")
else:
content = str(response)
# 방법 1: ```json 블록에서 추출
json_match = re.search(r'``json\s*(.*?)\s*``', content, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(1))
except json.JSONDecodeError:
pass
# 방법 2: ``` 블록에서 추출
json_match = re.search(r'``\s*(.*?)\s*``', content, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(1))
except json.JSONDecodeError:
pass
# 방법 3: 중괄호로 감싸진 JSON 추출
json_match = re.search(r'\{[^{}]*"action"[^{}]*\}', content, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(0))
except json.JSONDecodeError:
pass
# 방법 4: 첫 번째 {에서 마지막 }까지
if '{' in content and '}' in content:
start = content.find('{')
end = content.rfind('}') + 1
try:
return json.loads(content[start:end])
except json.JSONDecodeError:
pass
# 방법 5: 일반 텍스트에서 키워드 추출
return self._fallback_parse(content)
def _fallback_parse(self, text: str) -> Optional[Dict[str, Any]]:
"""키워드 기반 폴백 파싱"""
text_lower = text.lower()
# 행동 키워드 매핑
action_keywords = {
"attack": ["공격", "attack", "attacking", "적"],
"defend": ["방어", "defend", "shield", "guard"],
"flee": ["도망", "flee", "run", "escape", "회피"],
"talk": ["대화", "talk", "speak", "말", "말하다"],
"trade": ["거래", "trade", "exchange", "교환"],
"help": ["도움", "help", "assist", "support"]
}
detected_action = "idle"
for action, keywords in action_keywords.items():
if any(kw in text_lower for kw in keywords):
detected_action = action
break
return {
"action": detected_action,
"dialogue": text[:100] if len(text) > 100 else text,
"reasoning": "키워드 기반 폴백 파싱"
}
테스트
parser = RobustResponseParser(client)
result = parser.get_behavior("npc_001", "플레이어가 다가오며 적대적인 자세")
print(f"파싱 결과: {result}")
오류 3: "Context window exceeded" - 컨텍스트 길이 초과
# 문제: 긴 대화 이력과 상황 설명이 컨텍스트 제한 초과
해결: 대화 요약 및 스마트 컨텍스트 관리
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
@dataclass
class ConversationMessage:
role: str
content: str
timestamp: datetime
importance: float = 1.0 # 0.0-1.0
@dataclass
class SmartContextManager:
"""컨텍스트 창을 효율적으로 관리하는 클래스"""
max_tokens: int = 4000 # DeepSeek V3.2 기본 제한 고려
summary_interval: int = 10 # 10턴마다 요약
def __post_init__(self):
self.messages: List[ConversationMessage] = []
self.summaries: List[str] = []
self.summary_count = 0
def add_message(self, role: str, content: str, importance: float = 1.0):
"""새 메시지 추가 및 자동 요약"""
self.messages.append(ConversationMessage(
role=role,
content=content,
timestamp=datetime.now(),
importance=importance
))
# 요약 필요 시 자동 실행
if len(self.messages) >= self.summary_interval:
self._create_summary()
def _create_summary(self):
"""이전 대화 요약 및 메모리 압축"""
if not self.messages:
return
# 중요도 높은 메시지만 유지
important_messages = [
m for m in self.messages
if m.importance > 0.5
]
# 요약 프롬프트 생성
summary_prompt = """다음 대화를 3문장 이내로 요약하세요.
핵심 정보와 감정 변화를 포함하세요.
대화:
""" + "\n".join([f"{m.role}: {m.content}" for m in self.messages[-10:]])
# 요약 요청 (별도 API 호출 또는 간단한 로컬 처리)
summary = self._local_summary(summary_prompt)
self.summaries.append(summary)
self.messages = important_messages[-5:] # 최근 중요 메시지만 유지
self.summary_count += 1
print(f"대화 요약 #{self.summary_count}: {summary[:50]}...")
def _local_summary(self, text: str) -> str:
"""로컬 기반 간단 요약 (비용 절감)"""
# 실제로는 LLM 호출 또는 로컬 모델 사용 가능
sentences = text.split(".")
return ".".join(sentences[:3]) if len(sentences) > 3 else text
def build_context_prompt(self, current_situation: str) -> List[dict]:
"""현재 상황에 맞는 컨텍스트 프롬프트 구축"""
# 이전 요약 포함
system_messages = []
if self.summaries:
system_messages.append({
"role": "system",
"content": f"이전 대화 요약:\n" + "\n".join([f"- {s}" for s in self.summaries[-3:]])
})
# 최근 대화 포함
recent_messages = [
{"role": m.role, "content": m.content}
for m in self.messages[-5:]
]
# 현재 상황 추가
current_message = {
"role": "user",
"content": f"현재 상황: {current_situation}\n\nNPC의 다음 행동을 결정하세요."
}
return system_messages + recent_messages + [current_message]
def estimate_tokens(self, messages: List[dict]) -> int:
"""토큰 수 추정 (정확하지 않지만 빠른估算)"""
total_chars = sum(len(m.get("content", "")) for m in messages)
return int(total_chars / 4) # 대략적 토큰 변환율
def truncate_if_needed(self, messages: List[dict], model: str) -> List[dict]:
"""필요시 메시지 자르기"""
# 모델별 컨텍스트 제한
limits = {
"gpt-4.1": 128000,
"claude-sonnet-4-5": 200000,
"gemini-2.5-flash": 100000,
"deepseek-v3.2": 64000
}
limit = limits.get(model, 8000)
current_tokens = self.estimate_tokens(messages)
if current_tokens <= limit:
return messages
# 오래된 메시지부터 제거
while current_tokens > limit and len(messages