LLM 에이전트 파이프라인 토큰 비용 50% 줄이기: 요약 에이전트 vs 청크 주입 vs 프롬프트 캐싱 실전 비교

from anthropic import Anthropic
 
client = Anthropic()
 
def summarize_history(messages: list[dict], max_tokens: int = 8000) -> list[dict]:
    """누적 메시지가 임계치를 넘으면 요약본으로 압축한다."""
    # 토큰 수 추정 (실제 구현에서는 tiktoken 등으로 정확히 계산)
    estimated_tokens = sum(len(m["content"].split()) * 1.3 for m in messages)
 
    if estimated_tokens < max_tokens:
        return messages
 
    # 최근 2개 메시지는 유지, 나머지를 요약
    recent = messages[-2:]
    to_summarize = messages[:-2]
 
    summary_response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="당신은 대화 요약 전문가입니다. 핵심 사실, 결정 사항, 맥락만 보존하여 간결하게 요약하세요.",
        messages=[
            {
                "role": "user",
                "content": f"다음 대화를 500토큰 이내로 요약하세요:\n\n{str(to_summarize)}"
            }
        ]
    )
 
    summary_text = summary_response.content[0].text
 
    # 요약본을 첫 메시지로 교체
    return [
        {"role": "user", "content": f"[이전 대화 요약]\n{summary_text}"},
        {"role": "assistant", "content": "요약된 맥락을 파악했습니다. 계속하겠습니다."},
        *recent
    ]
 
# 에이전트 루프 예시
messages = []
for step in pipeline_steps:
    messages = summarize_history(messages)  # 매 단계 전 압축 체크
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=messages + [{"role": "user", "content": step}]
    )
    messages.append({"role": "user", "content": step})
    messages.append({"role": "assistant", "content": response.content[0].text})

from langchain_anthropic import ChatAnthropic
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor, LLMChainFilter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
 
llm = ChatAnthropic(model="claude-sonnet-4-6")
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)
 
# 1단계: 관련 없는 문서 전체를 저비용으로 제거
filter_compressor = LLMChainFilter.from_llm(llm)
 
# 2단계: 남은 문서에서 쿼리 관련 문장만 추출 (고비용이지만 이미 걸러진 후)
extractor_compressor = LLMChainExtractor.from_llm(llm)
 
# 계층화: filter 먼저, extract 나중
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
pipeline_compressor = DocumentCompressorPipeline(
    transformers=[filter_compressor, extractor_compressor]
)
 
compression_retriever = ContextualCompressionRetriever(
    base_compressor=pipeline_compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)
 
# 사용 예시
query = "프롬프트 캐싱 적용 방법"
compressed_docs = compression_retriever.get_relevant_documents(query)
# 10개 문서 검색 → 쿼리 관련 핵심 문장만 반환 (평균 40~60% 토큰 감소)

from langchain_core.documents import Document
 
def filter_by_similarity(docs_with_scores: list[tuple], threshold: float = 0.75) -> list[Document]:
    """코사인 유사도 임계치 이하 문서 제거 (BM25 점수와 조합 가능)"""
    return [doc for doc, score in docs_with_scores if score >= threshold]
 
# 벡터스토어에서 유사도 점수와 함께 검색
docs_with_scores = vectorstore.similarity_search_with_score(query, k=10)
filtered_docs = filter_by_similarity(docs_with_scores, threshold=0.75)

import anthropic
 
client = anthropic.Anthropic()
 
# 긴 시스템 프롬프트나 문서를 캐싱 (최소 1,024 토큰 필요)
SYSTEM_PROMPT = """당신은 코드 리뷰 전문가입니다...
[2,000 토큰 이상의 상세한 가이드라인]
"""
 
def analyze_with_cache(code_snippet: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"}  # 5분 TTL 캐시
            }
        ],
        messages=[{"role": "user", "content": f"다음 코드를 리뷰해주세요:\n{code_snippet}"}]
    )
 
    # 캐시 히트율 확인
    usage = response.usage
    print(f"캐시 읽기 토큰: {usage.cache_read_input_tokens}")
    print(f"캐시 쓰기 토큰: {usage.cache_creation_input_tokens}")
    print(f"일반 입력 토큰: {usage.input_tokens}")
 
    cache_hit_rate = usage.cache_read_input_tokens / (
        usage.cache_read_input_tokens + usage.cache_creation_input_tokens + usage.input_tokens
    ) * 100
    print(f"캐시 히트율: {cache_hit_rate:.1f}%")
 
    return response.content[0].text
 
# 동일 시스템 프롬프트로 반복 호출 — 2번째 요청부터 캐시 히트
for code in code_snippets:
    result = analyze_with_cache(code)

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
// 2,000 토큰짜리 시스템 프롬프트를 캐싱
const CACHED_SYSTEM = `당신은 기술 문서 작성 전문가입니다...
[상세 가이드라인 — 최소 1,024 토큰]`;
 
async function generateWithCache(userPrompt: string): Promise<string> {
  const response = await client.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 2048,
    system: [
      {
        type: "text",
        text: CACHED_SYSTEM,
        cache_control: { type: "ephemeral" },
      },
    ],
    messages: [{ role: "user", content: userPrompt }],
  });
 
  const usage = response.usage as any;
  console.log(`캐시 읽기: ${usage.cache_read_input_tokens ?? 0} 토큰`);
 
  return response.content[0].type === "text" ? response.content[0].text : "";
}