Long-Horizon 에이전트 비용 60~90% 줄이기: 캐싱·압축·라우팅 전략

import anthropic
 
client = anthropic.Anthropic()
 
LONG_CODING_GUIDELINES = "..."  # 수천 토큰짜리 코딩 가이드라인
 
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "당신은 코드 리뷰 전문 에이전트입니다.\n" + LONG_CODING_GUIDELINES,
            "cache_control": {"type": "ephemeral"}  # 이 블록을 캐시로 지정
        }
    ],
    messages=[
        {"role": "user", "content": f"다음 코드를 리뷰해줘:\n{user_code}"}
    ]
)
 
# 캐시 히트 여부 확인
usage = response.usage
print(f"캐시 히트 토큰: {usage.cache_read_input_tokens}")
print(f"캐시 생성 토큰: {usage.cache_creation_input_tokens}")

from typing import List, Dict
import anthropic
 
class ContextManager:
    def __init__(self, max_tokens: int = 50_000, summary_threshold: int = 40_000):
        self.max_tokens = max_tokens
        self.summary_threshold = summary_threshold
        self.messages: List[Dict] = []
        self.summary: str = ""
        self.client = anthropic.Anthropic()
 
    async def add_message(self, role: str, content: str, token_count: int):
        self.messages.append({
            "role": role,
            "content": content,
            "tokens": token_count
        })
 
        current_total = sum(m["tokens"] for m in self.messages)
 
        if current_total > self.summary_threshold:
            await self._compress_old_messages()
 
    async def _compress_old_messages(self):
        cutoff = len(self.messages) // 2
        to_summarize = self.messages[:cutoff]
 
        conversation_text = "\n".join(
            f"[{m['role']}]: {m['content']}" for m in to_summarize
        )
 
        # 압축 자체는 저렴한 소형 모델로 처리 — 이게 핵심 절감 포인트
        response = self.client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"다음 대화의 핵심 정보를 간결하게 요약해줘:\n\n{conversation_text}"
            }]
        )
        new_summary = response.content[0].text
 
        self.summary = f"{self.summary}\n\n{new_summary}".strip()
        self.messages = self.messages[cutoff:]
 
    def get_context(self) -> List[Dict]:
        context = []
        if self.summary:
            context.append({
                "role": "system",
                "content": f"[이전 대화 요약]\n{self.summary}"
            })
        context.extend(
            {"role": m["role"], "content": m["content"]}
            for m in self.messages
        )
        return context

from enum import Enum
 
class TaskComplexity(Enum):
    SIMPLE = "simple"      # Haiku급 — 번역, 요약, 형식 변환
    MODERATE = "moderate"  # Sonnet급 — 일반 분석, 코드 설명
    COMPLEX = "complex"    # Opus급 — 아키텍처 설계, 보안 분석
 
class ModelRouter:
    ROUTING_RULES = {
        TaskComplexity.SIMPLE: "claude-haiku-4-5-20251001",
        TaskComplexity.MODERATE: "claude-sonnet-4-6",
        TaskComplexity.COMPLEX: "claude-opus-4-8",
    }
 
    def classify_task(self, task: str, context_tokens: int) -> TaskComplexity:
        # 컨텍스트가 크면 복잡한 작업일 가능성 높음
        if context_tokens > 30_000:
            return TaskComplexity.COMPLEX
 
        # 키워드 기반 복잡도 판단 (규모가 크면 별도 분류 모델 권장)
        complex_signals = ["아키텍처 설계", "보안 취약점", "성능 최적화", "리팩토링"]
        simple_signals = ["번역", "요약", "형식 변환", "단순 질문"]
 
        if any(signal in task for signal in complex_signals):
            return TaskComplexity.COMPLEX
        if any(signal in task for signal in simple_signals):
            return TaskComplexity.SIMPLE
        return TaskComplexity.MODERATE
 
    def get_model(self, task: str, context_tokens: int) -> str:
        complexity = self.classify_task(task, context_tokens)
        return self.ROUTING_RULES[complexity]

interface AgentCheckpoint {
  taskId: string;
  step: number;
  state: Record<string, unknown>;
  completedActions: string[];
  timestamp: number;
}
 
const MAX_STEPS = 100;
 
class CheckpointedAgent {
  private checkpointInterval = 5; // 5스텝마다 저장
 
  async execute(task: string, taskId: string): Promise<void> {
    const checkpoint = await this.loadCheckpoint(taskId);
    const startStep = checkpoint?.step ?? 0;
    let state: Record<string, unknown> = checkpoint?.state ?? this.initState(task);
 
    console.log(`스텝 ${startStep}부터 재개`);
 
    for (let step = startStep; step < MAX_STEPS; step++) {
      state = await this.executeStep(step, state);
 
      if (step % this.checkpointInterval === 0) {
        await this.saveCheckpoint({
          taskId,
          step: step + 1,
          state,
          completedActions: (state.actions as string[]) ?? [],
          timestamp: Date.now()
        });
      }
 
      if (this.isComplete(state)) break;
    }
  }
 
  private async saveCheckpoint(cp: AgentCheckpoint): Promise<void> {
    await redis.setex(`checkpoint:${cp.taskId}`, 3600, JSON.stringify(cp));
  }
 
  private async loadCheckpoint(taskId: string): Promise<AgentCheckpoint | null> {
    const data = await redis.get(`checkpoint:${taskId}`);
    return data ? JSON.parse(data) : null;
  }
 
  private initState(task: string): Record<string, unknown> {
    return { task, actions: [], status: "started" };
  }
 
  private async executeStep(
    step: number,
    state: Record<string, unknown>
  ): Promise<Record<string, unknown>> {
    // 실제 스텝 실행 로직 구현
    return state;
  }
 
  private isComplete(state: Record<string, unknown>): boolean {
    return state.status === "done";
  }
}

import anthropic
from dataclasses import dataclass, field
 
@dataclass
class BugFixAgent:
    client: anthropic.Anthropic = field(default_factory=anthropic.Anthropic)
    context_manager: ContextManager = field(default_factory=ContextManager)
    router: ModelRouter = field(default_factory=ModelRouter)
 
    SYSTEM_PROMPT = """당신은 시니어 소프트웨어 엔지니어입니다.
    버그를 진단하고 수정하는 전문가입니다.
    항상 테스트를 작성하고 코드 품질을 유지합니다."""
 
    async def fix_bug(self, bug_report: str, codebase_context: str) -> str:
        # [최적화 1] 작업 복잡도에 따라 모델 선택 — 모델 라우팅
        analysis_model = self.router.get_model("버그 분석", len(codebase_context))
 
        analysis = await self.client.messages.create(
            model=analysis_model,
            max_tokens=2048,
            system=[{
                "type": "text",
                "text": self.SYSTEM_PROMPT + "\n\n" + codebase_context,
                "cache_control": {"type": "ephemeral"}  # [최적화 2] 코드베이스 캐싱
            }],
            messages=[{
                "role": "user",
                "content": f"버그 리포트: {bug_report}\n\n근본 원인을 분석해줘."
            }]
        )
 
        # [최적화 3] 복잡한 수정은 대형 모델로 전환
        # 동일한 codebase_context → 캐시 히트 발생 (캐싱 효과 재활용)
        fix = await self.client.messages.create(
            model="claude-opus-4-8",
            max_tokens=4096,
            system=[{
                "type": "text",
                "text": self.SYSTEM_PROMPT + "\n\n" + codebase_context,
                "cache_control": {"type": "ephemeral"}
            }],
            messages=[
                {"role": "user", "content": f"버그 리포트: {bug_report}"},
                {"role": "assistant", "content": analysis.content[0].text},
                {"role": "user", "content": "위 분석을 바탕으로 수정 코드를 작성해줘."}
            ]
        )
 
        return fix.content[0].text

from typing import AsyncGenerator
import anthropic
 
class ResearchAgent:
    def __init__(self):
        self.client = anthropic.Anthropic()  # 명시적으로 클래스 레벨에서 초기화
        self.context_manager = ContextManager(
            max_tokens=80_000,
            summary_threshold=60_000
        )
 
    async def research(self, query: str) -> AsyncGenerator[str, None]:
        search_results = []
 
        async for result in self._search(query):
            # 검색 결과 즉시 요약 — 원문 대신 요약본만 컨텍스트에 저장
            summary = await self._summarize_result(result, query)
            search_results.append(summary)
 
            # 토큰 수 추정 — 영어 기준 근사값
            # 한국어는 공백 기준 단어 수와 실제 토큰 수 차이가 크므로
            # tiktoken 또는 API 토큰 카운터 사용을 권장
            estimated_tokens = int(len(summary.split()) * 1.5)
 
            await self.context_manager.add_message(
                role="tool",
                content=f"[검색 결과 요약]\n{summary}",
                token_count=estimated_tokens
            )
 
            yield f"진행 중: {len(search_results)}개 결과 수집"
 
        # 최종 보고서는 대형 모델로
        report = await self._synthesize(
            query=query,
            context=self.context_manager.get_context()
        )
 
        yield report
 
    async def _summarize_result(self, result: str, query: str) -> str:
        # self.client 사용 — 전역 변수 아님
        response = self.client.messages.create(
            model="claude-haiku-4-5-20251001",  # 저렴한 소형 모델로 요약
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": f"쿼리 '{query}'와 관련된 핵심만 3문장으로 요약:\n{result[:2000]}"
            }]
        )
        return response.content[0].text
 
    async def _search(self, query: str) -> AsyncGenerator[str, None]:
        # 실제 검색 로직 구현
        pass
 
    async def _synthesize(self, query: str, context: list) -> str:
        # 최종 보고서 생성 로직 구현
        pass