AI 에이전트 LLM 비용이 폭증하는 구조와 60~80% 절감 전략

import autogen
 
# 종료 조건 없이 설정하면 비용이 조용히 폭증합니다
groupchat = autogen.GroupChat(
    agents=[user_proxy, planner, coder, reviewer],
    messages=[],
    # max_round 미설정 → 루프가 자동 종료되지 않습니다
)
 
# 권장: 아키텍처 수준에서 명시적으로 상한을 선언합니다
groupchat_safe = autogen.GroupChat(
    agents=[user_proxy, planner, coder, reviewer],
    messages=[],
    max_round=10,
)

from litellm import completion
 
def cascade_route(task: str, context: dict) -> str:
    complexity = estimate_complexity(task)  # 0.0 ~ 1.0
 
    if complexity < 0.3:
        model = "claude-haiku-4-5"      # 단순 분류, 형식 변환, 요약
    elif complexity < 0.7:
        model = "claude-sonnet-4-6"     # 코드 생성, 멀티스텝 추론
    else:
        model = "claude-opus-4-8"       # 복잡한 아키텍처 설계, 모호한 요구사항 분석
 
    response = completion(
        model=model,
        messages=[{"role": "user", "content": task}],
        metadata={"cost_tag": f"complexity_{complexity:.1f}"}  # 비용 추적용
    )
    return response.choices[0].message.content

class TrajectoryCompressor:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
 
    def compress(self, trajectory: list[dict]) -> list[dict]:
        compressed = []
        for step in trajectory:
            if step["status"] == "completed":
                # step["summary"]는 별도 LLM 호출로 미리 생성해둔 요약입니다.
                # 요약 생성 비용이 따로 발생하지만, 이후 호출에서 절감하는 토큰이
                # 훨씬 크기 때문에 전체 비용은 줄어듭니다.
                compressed.append({
                    "role": "assistant",
                    "content": f"[완료] {step['summary']}",
                    "status": "compressed"
                })
            else:
                compressed.append(step)
        return compressed

import hashlib
import json
 
class AgentPlanCache:
    def __init__(self):
        self._cache: dict[str, dict] = {}
 
    def _task_fingerprint(self, task_type: str, context_keys: list[str]) -> str:
        payload = json.dumps({"type": task_type, "keys": sorted(context_keys)})
        return hashlib.sha256(payload.encode()).hexdigest()[:16]
 
    def get_plan(self, task_type: str, context_keys: list[str]) -> dict | None:
        key = self._task_fingerprint(task_type, context_keys)
        return self._cache.get(key)
 
    def store_plan(self, task_type: str, context_keys: list[str], plan: dict, score: float):
        if score >= 0.95:  # 성공률 95% 이상인 플랜만 캐싱
            key = self._task_fingerprint(task_type, context_keys)
            self._cache[key] = {"plan": plan, "score": score}
 
# 결과: 서빙 비용 50.31% 절감, 레이턴시 27.28% 절감, 최적 성능의 96.61% 유지

import anthropic
 
client = anthropic.Anthropic()
 
# 전략 1: 전체 컨텍스트 캐싱
# 구현이 단순하지만, 컨텍스트 일부가 바뀌면 캐시 전체가 무효화됩니다
response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": full_context,
        "cache_control": {"type": "ephemeral"}
    }],
    messages=[{"role": "user", "content": user_task}]
)
 
# 전략 2: 시스템 프롬프트만 캐싱 — 가장 범용적이고 대부분의 상황에 적합합니다
response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": system_prompt,           # 고정된 시스템 프롬프트만 캐싱
        "cache_control": {"type": "ephemeral"}
    }],
    messages=[{"role": "user", "content": user_task}]  # 동적 부분은 캐싱 제외
)
 
# 전략 3: 동적 도구 결과 제외 캐싱
# 시스템 프롬프트와 고정 지침은 캐싱하고, 자주 바뀌는 도구 호출 결과는 제외합니다
response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": system_prompt,
        "cache_control": {"type": "ephemeral"}
    }],
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": static_instructions, "cache_control": {"type": "ephemeral"}},
            {"type": "text", "text": dynamic_tool_results}  # 캐싱 범위에서 제외
        ]
    }]
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
 
class AgentState(TypedDict):
    task: str
    plan: list[str]
    completed_steps: Annotated[list[str], operator.add]
    token_budget: int   # 비용 상한을 상태로 관리
    tokens_used: int
 
def check_budget(state: AgentState) -> str:
    if state["tokens_used"] >= state["token_budget"]:
        return "budget_exceeded"
    if not state["plan"]:
        return END
    return "execute_step"
 
builder = StateGraph(AgentState)
builder.add_node("planner", plan_task)
builder.add_node("executor", execute_step)
builder.add_node("synthesizer", synthesize_result)
 
# 비용 제어 분기를 그래프 구조에 내장합니다
builder.add_conditional_edges("executor", check_budget, {
    "execute_step": "executor",
    "budget_exceeded": "synthesizer",   # 예산 초과 시 중간 결과로 종료
    END: "synthesizer"
})