🎯 Strands 프로덕션

프로덕션 전환

PoC를 프로덕션으로 올릴 때 마주하는 세 가지 최적화 과제

프로덕션 최적화 3축

PoC에서 프로덕션으로 전환할 때 반드시 해결해야 하는 비용·속도·인프라 과제

1. 1
   비용 최적화

   프롬프트 캐싱으로 반복 토큰 비용 90% 절감. 시스템 프롬프트·도구 정의처럼 매 호출 동일한 prefix를 캐시하여 입력 토큰 과금 최소화.

2. 2
   속도 최적화

   스트리밍으로 TTFT(Time To First Token)를 최소화. 전체 응답 완료를 기다리지 않고 토큰 단위로 즉시 전달하여 체감 지연 제거.

3. 3
   인프라 최적화

   워크로드 특성에 맞는 배포 타겟 선택. Lambda·Fargate·AgentCore Runtime 중 콜드스타트·비용·확장성 균형점을 찾아 운영 부담 최소화.

프로덕션 핵심 지표

캐싱·스트리밍·멀티 프로바이더가 만드는 정량적 개선 효과

최적화 축별 도구 매핑

각 축에서 Strands SDK가 제공하는 구체적 메커니즘과 적용 대상

호출 단위 모델 라우팅

같은 에이전트에서 요청별로 다른 모델 지정 — 경량 작업은 싼 모델, 복잡한 작업만 비싼 모델

핵심 포인트

agent(prompt, model=)

호출 시점에 모델 오버라이드 — 에이전트 재생성 없이 요청별 지정 (v1.50 이후 사용 가능)

기본 모델 유지

model= 없는 호출은 Agent 생성 시 지정한 모델 그대로

맥락 공유

같은 에이전트의 messages[]를 이어 씀 — 모델만 바뀌고 대화는 유지

비용 최적화

분류·요약 같은 경량 요청을 저가 모델로 라우팅 — 캐싱과 겹쳐 쓰는 별개의 절감 축

코드

from strands import Agent
from strands.models import BedrockModel

cheap = BedrockModel(model_id="us.anthropic.claude-haiku-4-5")
smart = BedrockModel(model_id="us.anthropic.claude-sonnet-4-6")

# 기본 모델은 smart — tools·system_prompt는 하나로 유지
agent = Agent(model=smart, tools=[search_docs])

# 경량 작업 — 이 호출만 싼 모델로 라우팅
agent("이 문의를 카테고리로 분류해줘", model=cheap)

# 복잡한 작업 — 기본 모델 그대로 (같은 대화 맥락 공유)
agent("분류 결과를 바탕으로 대응 방안을 설계해줘")

프롬프트 캐싱

매 호출 반복되는 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 장면 — 토큰이 흐르는 화면

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

배포 타겟

워크로드 특성에 맞춘 세 가지 컴퓨팅 타겟의 선택 기준

배포 옵션 3종

Lambda는 경량 단발, Fargate는 장시간 세션, AgentCore Runtime은 관리형 — 강점과 제약의 대비

배포 타겟 비교표

콜드스타트·비용 모델·확장성·운영 부담 4축 정량 비교

AgentCore Runtime 배포 예시

Strands 에이전트를 AgentCore에 배포하는 최소 설정과 CLI 워크플로우

# app/MyProdAgent/main.py — agentcore create가 생성한 스캐폴드를 아래처럼 수정
from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

agent = Agent(
    model=BedrockModel(
        model_id="us.anthropic.claude-sonnet-4-6",
        cache_tools="default",
        streaming=True,
    ),
    system_prompt="You are a production assistant.",
    tools=[search_docs, query_db, send_notification],
)

# @app.entrypoint — AgentCore Runtime이 호출하는 진입점
@app.entrypoint
def handler(payload):
    result = agent(payload.get("prompt"))
    return {"result": str(result)}
# ─── CLI 배포 워크플로우 ───
# 1. agentcore create --name MyProdAgent — 스캐폴드 생성(main.py + pyproject.toml)
# 2. 생성된 app/MyProdAgent/main.py를 위 내용으로 수정
# 3. agentcore deploy — 코드 패키징 + 배포(버전·엔드포인트는 Runtime이 관리)