스트리밍과 재시도

프롬프트 캐싱

매 호출 반복되는 prefix를 캐시해 입력 토큰 비용을 절감하는 메커니즘

cache_tools로 도구 정의 캐싱

도구 정의는 한 줄 옵션 — 멀티턴 자동(CacheConfig)·시스템 프롬프트 명시(cachePoint)까지 대상별 선언

from strands import Agent
from strands.models import BedrockModel, CacheConfig
from strands.types.content import SystemContentBlock

model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-6",
    cache_tools="default",  # 도구 정의 전체를 캐시 (5분 TTL)
    # 멀티턴 메시지 자동 캐싱 — 마지막 user 메시지 끝에 캐시 포인트 자동 배치
    cache_config=CacheConfig(strategy="auto"),
)

agent = Agent(
    model=model,
    # 시스템 프롬프트 명시 캐싱 — cachePoint 마커까지가 캐시 prefix
    system_prompt=[
        SystemContentBlock(text="""You are a production data analyst.
    You have access to SQL databases, S3 data lakes, and monitoring dashboards.
    Always validate queries before execution and explain your reasoning."""),
        SystemContentBlock(cachePoint={"type": "default"}),
    ],
    tools=[query_database, read_s3, check_metrics, generate_report],
)

# 첫 호출: cache write (약간의 추가 비용)
# 이후 호출: cache hit → 시스템 프롬프트 + 도구 정의 비용 90% 절감
result = agent("지난주 매출 리포트를 생성해줘")

캐싱 동작 원리

Cache Point 마커 기준으로 prefix 일치 여부를 판단하는 hit/miss 흐름

1. 1
   Cache Point 선언

   SystemContentBlock에 cachePoint: { type: "default" }를 삽입. 이 마커까지의 토큰이 캐시 키의 prefix가 됨.

2. 2
   Prefix 해싱

   프로바이더가 시스템 프롬프트 + 도구 정의를 해싱하여 캐시 키 생성. 동일 prefix면 같은 키로 매핑.

3. 3
   Cache Hit 판정

   TTL 내에 동일 키가 존재하면 hit — 캐시된 KV 상태를 재사용하여 prefix 토큰 재처리 생략.

4. 4
   비용 정산

   Hit 시 캐시 읽기 비용만 과금 (일반 입력 토큰의 10%). Miss 시 cache write 비용 25% 추가 후 다음 호출부터 절감.

캐싱 조건 비교

Anthropic API 직접 호출과 Bedrock 경유 시 캐싱 동작 차이

스트리밍

토큰 단위 수신 인터페이스부터 브라우저 전달까지의 실시간 경로

AsyncIterator + Callback 스트리밍

토큰 단위 수신을 위한 두 가지 인터페이스 — 이터레이터와 이벤트 콜백

import asyncio
from strands import Agent

queue = asyncio.Queue()

# 방법 1: Callback Handler — callable 하나로 이벤트 수신
def sse_handler(**kwargs):
    """토큰을 SSE 형식으로 클라이언트에 푸시"""
    if "data" in kwargs:  # 텍스트 델타
        queue.put_nowait(f"data: {kwargs['data']}\n\n")
    tool_use = kwargs.get("current_tool_use")
    if tool_use:          # 도구 호출 이벤트
        queue.put_nowait(f"event: tool\ndata: {tool_use['name']}\n\n")

agent = Agent(callback_handler=sse_handler)

# 방법 2: AsyncIterator — async for 루프 (이벤트는 dict)
async def stream_response(agent: Agent, prompt: str):
    async for event in agent.stream_async(prompt):
        if "data" in event:
            yield f"data: {event['data']}\n\n"
        elif event.get("current_tool_use"):
            name = event["current_tool_use"]["name"]
            yield f"event: tool\ndata: {name}\n\n"
    yield "data: [DONE]\n\n"

Polling vs Streaming 비교

응답 수신 방식에 따른 지연·리소스·UX 차이

SSE 파이프라인 구조

Agent에서 브라우저까지 토큰이 전달되는 4단계 흐름

1. 1
   Agent 추론

   Strands Agent가 모델 API를 스트리밍 모드로 호출. 모델이 토큰을 생성할 때마다 chunk 이벤트 발생.

2. 2
   Token Callback

   callback_handler로 전달한 함수가 "data" 키의 텍스트 델타를 수신. SSE 형식(data: ...)으로 변환하여 응답 큐에 적재.

3. 3
   SSE Transport

   FastAPI StreamingResponse / Lambda Function URL이 Content-Type: text/event-stream으로 청크 전송. Keep-alive 유지.

4. 4
   Client Render

   EventSource API가 토큰을 수신할 때마다 DOM에 append. 타이핑 효과로 실시간 표시. 연결 끊김 시 자동 재연결.

SSE 장면 — 토큰이 흐르는 화면

같은 순간의 양쪽 모습 — 와이어의 델타 이벤트와 브라우저의 부분 출력