AI 에이전트 비용 최적화 완전 가이드 — 품질은 유지하고 API 비용은 80% 줄이기 [2026]

실제로 발생하는 숨은 비용:
[에이전트 1회 실행 시]
- 시스템 프롬프트: 2,000 토큰 (매 호출마다 재처리)
- 도구 정의: 1,500 토큰 (매 호출마다 재처리)
- 대화 히스토리: 5,000 토큰 (계속 누적)
- 실제 쿼리: 50 토큰
총 8,550 토큰 중 실제 쿼리는 0.6%!

# cost_calculator.py
# 현재 에이전트의 실제 비용을 계산해보세요
def estimate_monthly_cost(
    daily_conversations: int,
    avg_turns_per_conversation: int,
    avg_input_tokens_per_turn: int,
    avg_output_tokens_per_turn: int,
    model: str = "sonnet"
) -> dict:
    """월간 예상 비용 계산"""
    # 2026년 가격 (per 1M tokens)
    pricing = {
        "haiku":  {"input": 1.00,  "output": 5.00},
        "sonnet": {"input": 3.00,  "output": 15.00},
        "opus":   {"input": 5.00,  "output": 25.00},
    }
    p = pricing[model]
    monthly_conversations = daily_conversations * 30
    total_turns = monthly_conversations * avg_turns_per_conversation
    monthly_input_cost  = (total_turns * avg_input_tokens_per_turn  / 1_000_000) * p["input"]
    monthly_output_cost = (total_turns * avg_output_tokens_per_turn / 1_000_000) * p["output"]
    total_cost = monthly_input_cost + monthly_output_cost
    return {
        "monthly_total": f"${total_cost:,.2f}",
        "monthly_input": f"${monthly_input_cost:,.2f}",
        "monthly_output": f"${monthly_output_cost:,.2f}",
        "per_conversation": f"${total_cost / monthly_conversations:.4f}",
        "monthly_tokens": f"{total_turns * (avg_input_tokens_per_turn + avg_output_tokens_per_turn):,}",
    }
# 예시: 고객 서비스 에이전트
result = estimate_monthly_cost(
    daily_conversations=1_000,     # 하루 1,000건
    avg_turns_per_conversation=5,  # 대화당 평균 5턴
    avg_input_tokens_per_turn=800, # 턴당 입력 800 토큰
    avg_output_tokens_per_turn=200,# 턴당 출력 200 토큰
    model="sonnet"
)
for key, val in result.items():
    print(f"  {key}: {val}")
# 출력:
# monthly_total:   $7,200.00  ← 최적화 없이 이게 현실
# monthly_input:   $3,600.00
# monthly_output:  $3,600.00
# per_conversation: $0.2400
# monthly_tokens:  30,000,000,000

# prompt_caching.py
import anthropic
import os
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
# ── 캐싱 없는 기본 호출 (비효율) ─────────────────────
def basic_agent_call(user_message: str) -> str:
    """캐싱 없이 매번 전체 시스템 프롬프트 재처리"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system="당신은 전문 고객 서비스 에이전트입니다.\n\n"
               "[회사 정책 - 5,000 토큰 분량]\n"
               "환불 정책: ...\n배송 정책: ...\n품질 보증: ...\n"
               "# 이 내용이 매 호출마다 재처리됨 (낭비!)",
        messages=[{"role": "user", "content": user_message}]
    )
    return response.content[0].text
# ── 프롬프트 캐싱 적용 (최적화) ─────────────────────
SYSTEM_PROMPT_STATIC = """당신은 전문 고객 서비스 에이전트입니다.
[회사 정책]
환불 정책: 구매 후 30일 이내 영수증 지참 시 전액 환불 가능합니다.
배송 정책: 주문 후 3~5 영업일 이내 배송. 제주/도서산간 지역은 추가 2~3일.
품질 보증: 모든 제품 1년 무상 AS 제공.
개인정보 처리: 수집된 정보는 서비스 제공 목적으로만 사용합니다.
[응대 원칙]
1. 항상 친절하고 전문적으로 응대하세요
2. 정확한 정보만 제공하고, 불확실할 경우 확인 후 안내하세요
3. 고객 불만은 먼저 공감하고, 해결책을 제시하세요
4. 복잡한 문제는 전문 팀으로 에스컬레이션하세요
"""  # 이 내용이 반복적으로 사용됨 → 캐싱 대상
def cached_agent_call(user_message: str) -> str:
    """
    프롬프트 캐싱으로 시스템 프롬프트 재처리 비용 90% 절감.
    두 번째 호출부터 캐시에서 읽어옴.
    """
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT_STATIC,
                "cache_control": {"type": "ephemeral"}
                # ↑ 이 블록을 캐시하도록 지정
                # 최소 1,024 토큰 이상이어야 캐싱됨
            }
        ],
        messages=[{"role": "user", "content": user_message}]
    )
    # 캐시 사용 현황 확인
    usage = response.usage
    print(f"  입력 토큰: {usage.input_tokens}")
    print(f"  캐시 생성 토큰: {getattr(usage, 'cache_creation_input_tokens', 0)}")
    print(f"  캐시 읽기 토큰: {getattr(usage, 'cache_read_input_tokens', 0)}")
    print(f"  출력 토큰: {usage.output_tokens}")
    return response.content[0].text
# ── 도구 정의도 캐싱 ─────────────────────────────────
def agent_with_cached_tools(user_message: str) -> str:
    """도구 정의까지 캐싱 — 도구가 많을수록 효과 극대화"""
    # 도구 정의 (여러 개일수록 캐싱 효과 큼)
    tools = [
        {
            "name": "check_order",
            "description": "주문 번호로 배송 상태를 조회합니다.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "주문 번호"}
                },
                "required": ["order_id"]
            }
        },
        {
            "name": "process_refund",
            "description": "환불 요청을 처리합니다.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string"}
                },
                "required": ["order_id", "reason"]
            }
        },
        # ... 더 많은 도구
    ]
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        tools=tools,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT_STATIC,
                "cache_control": {"type": "ephemeral"}  # 시스템 프롬프트 캐싱
            }
        ],
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": user_message,
                        # 개인화된 컨텍스트가 있다면 여기에
                    }
                ]
            }
        ]
    )
    return response.content[0].text
# 테스트
print("첫 번째 호출 (캐시 생성):")
cached_agent_call("주문 배송은 얼마나 걸리나요?")
print("\n두 번째 호출 (캐시 읽기 — 훨씬 저렴):")
cached_agent_call("환불하려면 어떻게 해야 하나요?")

# model_router.py
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
# 모델 초기화
haiku  = ChatAnthropic(model="claude-haiku-4-5-20251001",  max_tokens=512)
sonnet = ChatAnthropic(model="claude-sonnet-4-20250514",   max_tokens=1024)
opus   = ChatAnthropic(model="claude-opus-4-20250514",     max_tokens=2048)
def classify_task_complexity(query: str) -> str:
    """
    쿼리 복잡도를 분류합니다.
    Haiku로 분류하기 때문에 분류 자체 비용이 거의 없습니다.
    """
    classification_prompt = f"""
다음 사용자 요청의 복잡도를 분류하세요.
요청: "{query}"
분류 기준:
- simple: 단순 사실 확인, 정의, 간단한 계산, 감사 인사
- medium: 다단계 설명, 비교 분석, 코드 설명
- complex: 복잡한 추론, 아키텍처 설계, 윤리적 판단, 전략 수립
하나의 단어로만 응답하세요: simple, medium, complex
"""
    result = haiku.invoke([HumanMessage(content=classification_prompt)])
    return result.content.strip().lower()
def smart_agent(query: str) -> dict:
    """
    쿼리 복잡도에 따라 자동으로 최적 모델 선택.
    """
    complexity = classify_task_complexity(query)
    model_map = {
        "simple":  (haiku,  "claude-haiku"),
        "medium":  (sonnet, "claude-sonnet"),
        "complex": (opus,   "claude-opus"),
    }
    model, model_name = model_map.get(complexity, (sonnet, "claude-sonnet"))
    print(f"  복잡도: {complexity} → 모델: {model_name}")
    response = model.invoke([HumanMessage(content=query)])
    return {
        "answer": response.content,
        "model_used": model_name,
        "complexity": complexity
    }
# 테스트
queries = [
    "안녕하세요!",                              # simple → Haiku ($0.001)
    "Python의 GIL이 멀티스레딩에 미치는 영향 설명해줘",  # medium → Sonnet ($0.003)
    "10만 명이 동시 접속하는 실시간 채팅 서비스 아키텍처 설계해줘",  # complex → Opus ($0.005)
]
print("=== 모델 라우팅 테스트 ===\n")
for q in queries:
    print(f"쿼리: {q[:50]}...")
    result = smart_agent(q)
    print(f"  사용 모델: {result['model_used']}\n")

# semantic_cache.py
from langchain_anthropic import ChatAnthropic
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.messages import HumanMessage
import hashlib
import json
import os
embedding_model = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)
class SemanticCache:
    """
    의미적으로 유사한 쿼리의 응답을 캐시합니다.
    유사도 임계값(threshold) 이상이면 API 호출 없이 캐시 반환.
    """
    def __init__(self, similarity_threshold: float = 0.92):
        self.threshold = similarity_threshold
        self.vector_store = Chroma(
            collection_name="response_cache",
            embedding_function=embedding_model,
            persist_directory="./semantic_cache_db"
        )
        self.response_store: dict[str, str] = {}  # doc_id → response
        self.llm = ChatAnthropic(model="claude-sonnet-4-20250514", max_tokens=1024)
    def _get_doc_id(self, text: str) -> str:
        return hashlib.md5(text.encode()).hexdigest()
    def query(self, user_query: str) -> dict:
        """
        캐시에서 유사한 응답 검색.
        없으면 LLM 호출 후 캐시 저장.
        """
        # 1. 유사한 캐시 항목 검색
        results = self.vector_store.similarity_search_with_score(
            user_query, k=1
        )
        if results:
            doc, score = results[0]
            similarity = 1 - score  # Chroma는 거리 반환, 유사도로 변환
            if similarity >= self.threshold:
                cached_response = self.response_store.get(doc.metadata["doc_id"])
                if cached_response:
                    print(f"  ✅ 캐시 히트! (유사도: {similarity:.3f})")
                    print(f"  원본 쿼리: {doc.page_content[:50]}...")
                    return {
                        "answer": cached_response,
                        "source": "cache",
                        "similarity": similarity,
                        "cost": "$0.00"
                    }
        # 2. 캐시 미스 → LLM 호출
        print(f"  ❌ 캐시 미스 → LLM 호출")
        response = self.llm.invoke([HumanMessage(content=user_query)])
        answer = response.content
        # 3. 캐시 저장
        doc_id = self._get_doc_id(user_query)
        self.vector_store.add_texts(
            texts=[user_query],
            metadatas=[{"doc_id": doc_id}]
        )
        self.response_store[doc_id] = answer
        return {
            "answer": answer,
            "source": "llm",
            "similarity": 0.0,
            "cost": "~$0.003"
        }
# 테스트
cache = SemanticCache(similarity_threshold=0.90)
test_queries = [
    "배송은 얼마나 걸리나요?",        # 최초 → LLM 호출
    "배송 기간이 어떻게 되나요?",      # 유사 → 캐시 히트
    "물건 받는 데 며칠 걸려요?",       # 유사 → 캐시 히트
    "환불 정책이 어떻게 되나요?",      # 다른 주제 → LLM 호출
    "환불하려면 어떻게 해야 해요?",    # 유사 → 캐시 히트
]
print("=== 시맨틱 캐싱 테스트 ===\n")
for q in test_queries:
    print(f"쿼리: {q}")
    result = cache.query(q)
    print(f"  출처: {result['source']}, 비용: {result['cost']}\n")

# context_compressor.py
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import tiktoken
def count_tokens(text: str) -> int:
    """토큰 수 추정 (Claude는 공식 API가 없어 근사값 사용)"""
    return len(text.split()) * 1.3  # 단어 수 × 1.3 ≈ 토큰 수
def compress_conversation(
    messages: list,
    max_tokens: int = 3000,
    keep_recent: int = 4
) -> list:
    """
    대화가 길어지면 오래된 메시지를 압축 요약합니다.
    최근 N개 메시지는 항상 원본 유지.
    """
    total_tokens = sum(count_tokens(m.content) for m in messages)
    if total_tokens <= max_tokens:
        return messages  # 아직 짧으면 그대로
    # 최근 메시지 보존
    recent = messages[-keep_recent:]
    to_compress = messages[:-keep_recent]
    if not to_compress:
        return messages
    # 압축 대상 요약
    compression_llm = ChatAnthropic(
        model="claude-haiku-4-5-20251001",  # 요약은 저렴한 모델로!
        max_tokens=300
    )
    history_text = "\n".join(
        f"{m.__class__.__name__}: {m.content}" for m in to_compress
    )
    summary_response = compression_llm.invoke([
        HumanMessage(content=f"""
다음 대화를 3~5문장으로 핵심만 요약하세요.
사용자 의도, 중요한 결정 사항, 해결된 문제를 포함하세요.
대화:
{history_text}
""")
    ])
    summary_msg = SystemMessage(
        content=f"[이전 대화 요약]\n{summary_response.content}"
    )
    original_tokens = total_tokens
    compressed_tokens = count_tokens(summary_response.content) + sum(
        count_tokens(m.content) for m in recent
    )
    print(f"  컨텍스트 압축: {original_tokens:.0f} → {compressed_tokens:.0f} 토큰 "
          f"({(1 - compressed_tokens/original_tokens)*100:.0f}% 절감)")
    return [summary_msg] + recent
# ── 토큰 예산 설정 ────────────────────────────────────
def call_with_token_budget(
    messages: list,
    max_output_tokens: int = 500,  # 출력 제한
    verbose_mode: bool = False
) -> str:
    """
    출력 토큰에 예산을 설정합니다.
    출력 토큰은 입력보다 5배 비싸기 때문에 중요합니다.
    """
    # 불필요하게 긴 응답 방지
    system_instruction = (
        "간결하게 답변하세요. 불필요한 설명이나 반복은 피하세요."
        if not verbose_mode else
        "상세하게 설명하세요."
    )
    llm = ChatAnthropic(
        model="claude-sonnet-4-20250514",
        max_tokens=max_output_tokens,  # 출력 토큰 제한
        temperature=0
    )
    final_messages = [SystemMessage(content=system_instruction)] + messages
    response = llm.invoke(final_messages)
    return response.content

# batch_processing.py
import anthropic
import json
import time
client = anthropic.Anthropic()
def process_batch_requests(requests: list[dict]) -> list[dict]:
    """
    여러 요청을 배치로 묶어서 50% 할인 가격에 처리합니다.
    처리 시간: 수분 ~ 24시간 (비실시간)
    적합한 작업: 데이터 분석, 콘텐츠 생성, 분류 등
    """
    # 배치 요청 형식으로 변환
    batch_requests = []
    for i, req in enumerate(requests):
        batch_requests.append({
            "custom_id": f"request-{i}",
            "params": {
                "model": "claude-sonnet-4-20250514",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": req["prompt"]}]
            }
        })
    # 배치 생성
    batch = client.messages.batches.create(requests=batch_requests)
    print(f"✅ 배치 생성 완료 (ID: {batch.id})")
    print(f"   요청 수: {len(batch_requests)}개")
    print(f"   예상 비용: 일반 대비 50% 절감")
    # 완료 대기 (폴링)
    while True:
        status = client.messages.batches.retrieve(batch.id)
        print(f"   상태: {status.processing_status} "
              f"({status.request_counts.succeeded}/"
              f"{status.request_counts.processing + status.request_counts.succeeded} 완료)")
        if status.processing_status == "ended":
            break
        time.sleep(30)  # 30초마다 확인
    # 결과 수집
    results = []
    for result in client.messages.batches.results(batch.id):
        results.append({
            "id": result.custom_id,
            "content": result.result.message.content[0].text
            if result.result.type == "succeeded" else None,
            "error": result.result.error.message
            if result.result.type == "errored" else None
        })
    return results
# 사용 예시: 상품 설명 1,000개 일괄 생성
product_descriptions = [
    {"prompt": f"상품 '{i}번 상품'에 대한 SEO 최적화된 설명을 200자로 작성해줘"}
    for i in range(100)  # 100개 배치
]
# 배치 처리 (50% 할인)
# 일반 처리 시 비용: $3.00/1M × 100회 ≈ $0.30
# 배치 처리 시 비용: $1.50/1M × 100회 ≈ $0.15
results = process_batch_requests(product_descriptions)
print(f"\n✅ 완료: {len([r for r in results if r['content']])}개 성공")

# budget_manager.py
import os
from anthropic import Anthropic
from datetime import datetime, timedelta
from collections import defaultdict
client = Anthropic()
class TokenBudgetManager:
    """
    에이전트의 토큰 사용량을 추적하고 예산 초과 시 알림.
    """
    def __init__(self, daily_budget_usd: float = 50.0):
        self.daily_budget = daily_budget_usd
        self.usage_log: list[dict] = []
        self.cost_per_1m = {"input": 3.00, "output": 15.00}  # Sonnet 기준
    def track_and_call(
        self,
        messages: list,
        max_output_tokens: int = 1024,
        model: str = "claude-sonnet-4-20250514"
    ) -> str:
        """비용을 추적하면서 LLM 호출"""
        # 일일 예산 확인
        today_cost = self._get_today_cost()
        if today_cost >= self.daily_budget:
            raise Exception(
                f"일일 예산 초과: ${today_cost:.2f} / ${self.daily_budget:.2f}\n"
                f"에이전트가 오늘 더 이상 응답하지 않습니다."
            )
        # 예산 80% 도달 시 경고
        if today_cost >= self.daily_budget * 0.8:
            print(f"⚠️  예산 경고: ${today_cost:.2f} / ${self.daily_budget:.2f} "
                  f"({today_cost/self.daily_budget*100:.0f}% 사용)")
        # LLM 호출
        response = client.messages.create(
            model=model,
            max_tokens=max_output_tokens,
            messages=messages
        )
        # 비용 계산 및 기록
        input_cost  = response.usage.input_tokens  / 1_000_000 * self.cost_per_1m["input"]
        output_cost = response.usage.output_tokens / 1_000_000 * self.cost_per_1m["output"]
        total_cost  = input_cost + output_cost
        self.usage_log.append({
            "timestamp": datetime.now().isoformat(),
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "cost_usd": total_cost,
            "model": model
        })
        return response.content[0].text
    def _get_today_cost(self) -> float:
        today = datetime.now().date()
        return sum(
            log["cost_usd"] for log in self.usage_log
            if datetime.fromisoformat(log["timestamp"]).date() == today
        )
    def get_summary(self) -> dict:
        """비용 요약 리포트"""
        today_cost = self._get_today_cost()
        total_tokens = sum(
            log["input_tokens"] + log["output_tokens"] for log in self.usage_log
        )
        return {
            "오늘 비용": f"${today_cost:.4f}",
            "일일 예산": f"${self.daily_budget:.2f}",
            "예산 사용률": f"{today_cost/self.daily_budget*100:.1f}%",
            "총 호출 횟수": len(self.usage_log),
            "총 토큰": f"{total_tokens:,}",
        }
# 사용 예시
manager = TokenBudgetManager(daily_budget_usd=10.0)
for i in range(5):
    try:
        response = manager.track_and_call(
            messages=[{"role": "user", "content": f"테스트 질문 {i+1}"}],
            max_output_tokens=100
        )
        print(f"응답 {i+1}: {response[:50]}...")
    except Exception as e:
        print(f"❌ {e}")
        break
print("\n📊 비용 요약:")
for k, v in manager.get_summary().items():
    print(f"  {k}: {v}")