Lab 03: 스트리밍

• invoke_model_with_response_stream으로 응답을 스트리밍 방식으로 받는다
• 응답이 이벤트 스트림으로 도착하는 구조를 이해한다
• 각 청크의 bytes를 JSON으로 파싱하고, type을 보고 원하는 데이터만 골라낸다
• content_block_delta의 delta.text만 추출하는 이유를 설명할 수 있다
• 글자 조각을 차례로 내보내는 제너레이터(yield) 를 구현한다
• 저수준 스트리밍 처리의 번거로움을 체감하고, 그 한계를 말할 수 있다

지금까지의 챗봇은 모델이 답변을 전부 생성한 뒤에야 화면에 텍스트가 나타났습니다. 답변이 한두 문장이면 차이를 못 느끼지만, 모델이 긴 글을 쓰는 동안에는 사용자가 몇 초씩 빈 화면을 바라보게 됩니다. 이때 사용자는 “멈춘 건가?”라고 의심하게 되고, 체감 대기 시간은 실제보다 더 길게 느껴집니다.

스트리밍은 이 문제를 체감 지연(perceived latency) 관점에서 해결합니다. 전체 답변이 끝나는 시간 자체는 크게 달라지지 않지만, 첫 글자가 화면에 나타나는 시간이 극적으로 짧아집니다. 사람은 글자가 흘러나오기 시작하면 “동작하고 있다”고 안심하고, 흘러나오는 글을 읽으며 기다리기 때문에 같은 시간도 훨씬 짧게 느낍니다.

핵심은 “전체 속도를 빠르게 만드는 것”이 아니라 “기다림을 덜 지루하게 만드는 것”입니다. 그래서 사용자 대면 챗봇에서 스트리밍은 사실상 표준입니다.

스트리밍 API의 가장 큰 차이는 응답의 모양입니다. 일반 invoke_model은 완성된 JSON 한 덩어리를 돌려주지만, invoke_model_with_response_stream은 여러 개의 이벤트(event) 가 순서대로 흘러오는 스트림을 돌려줍니다.

각 이벤트 안에는 chunk라는 조각이 들어 있고, 그 안의 bytes가 우리가 파싱해야 할 실제 데이터(JSON 바이트)입니다. 그리고 모든 청크가 같은 종류가 아닙니다. 각 청크에는 type 필드가 있어서, 지금 이 조각이 무엇을 의미하는지 알려줍니다.

graph TD
  A["invoke_model_with_response_stream 호출"] --> B["resp['body'] = 이벤트 스트림"]
  B --> C["event['chunk']['bytes'] 꺼내기"]
  C --> D["json.loads 로 파싱"]
  D --> E{"data['type'] 확인"}
  E -->|message_start| F["메시지 시작 (무시)"]
  E -->|content_block_delta| G["delta.text 추출 → yield"]
  E -->|message_stop| H["메시지 끝 (무시)"]

스트림을 따라 흘러오는 주요 type은 다음과 같습니다.

우리가 화면에 출력하고 싶은 “글자”는 오직 content_block_delta 타입의 청크 안, delta.text에만 들어 있습니다. 나머지 타입은 상태 신호일 뿐 출력할 글자가 없습니다. 그래서 우리는 모든 청크를 받되, type을 보고 필요한 것만 골라내는 분기 처리를 직접 해야 합니다.

⚠️ 만약 응답이 비어 보인다면, 대개 content_block_delta 외의 타입까지 처리하려다 꼬였거나, delta.text를 잘못된 경로에서 찾고 있는 경우입니다.

스트림에서 글자 조각을 꺼냈으면, 이것을 호출하는 쪽(챗봇 UI)에 하나씩 차례로 전달해야 합니다. 조각이 다 모일 때까지 기다렸다가 리스트로 한꺼번에 반환하면, 결국 스트리밍의 장점이 사라집니다.

파이썬의 제너레이터(generator) 가 바로 이 일에 맞습니다. 함수 안에서 return 대신 yield를 쓰면, 값을 하나 내보낼 때마다 함수가 그 자리에서 잠시 멈췄다가, 다음 값이 필요할 때 이어서 실행됩니다. 덕분에 스트림에서 글자가 도착하는 즉시 UI로 흘려보낼 수 있습니다.

def respond_stream(history: list[dict]):
    """글자 조각(delta)을 차례로 내놓는 제너레이터."""
    ...
    for event in resp["body"]:
        ...
        if data.get("type") == "content_block_delta":
            yield data["delta"].get("text", "")

• yield를 포함한 함수는 호출 즉시 실행되지 않고, 제너레이터 객체를 반환합니다.
• UI가 for chunk in respond_stream(history): 처럼 한 조각씩 꺼내 쓰면, 그때마다 함수가 다음 yield까지 실행됩니다.
• 결과적으로 모델이 글자를 만들어내는 속도에 맞춰 화면에도 글자가 흘러나오게 됩니다.

이 구조 덕분에 respond_stream은 “스트림 파싱”이라는 한 가지 책임만 지고, “화면에 어떻게 그릴지”는 UI 쪽이 맡는 깔끔한 분리가 됩니다.

이번 Lab의 핵심 파일은 lab3_streaming.py입니다. 우리가 구현할 함수는 respond_stream 하나이며, 챗봇 UI(chat_ui.py)는 이미 제공되므로 우리는 boto3로 스트림을 호출하고 파싱하는 부분에만 집중합니다.

1. 모델 호출 준비

import json
from config import get_runtime, MODEL_HAIKU, ANTHROPIC_VERSION

rt = get_runtime()

• get_runtime()은 Bedrock 런타임 클라이언트(boto3)를 반환합니다.
• MODEL_HAIKU는 이 과정 전체의 주력 모델 ID us.anthropic.claude-haiku-4-5-20251001-v1:0(인퍼런스 프로파일)입니다. 신형 모델은 us.로 시작하는 인퍼런스 프로파일 ID로 호출합니다.
• ANTHROPIC_VERSION은 Anthropic 메시지 포맷의 버전 문자열로, body에 반드시 포함해야 합니다.

2. 히스토리를 messages로 변환하고 body 구성

def respond_stream(history: list[dict]):
    """글자 조각(delta)을 차례로 내놓는 제너레이터."""
    messages = [{"role": h["role"], "content": h["text"]} for h in history]

    body = {
        "anthropic_version": ANTHROPIC_VERSION,
        "max_tokens": 512,
        "messages": messages,
    }

• history는 UI가 쌓아온 대화 기록입니다. 각 항목의 role과 text를 모델이 이해하는 messages 형식(role/content)으로 변환합니다.
• body의 3요소 — anthropic_version, max_tokens, messages — 는 Lab 1~2에서 본 것과 동일합니다. 즉, 보내는 요청의 모양은 같고, 응답을 받는 방식만 스트리밍으로 바뀝니다.

3. 스트리밍 호출

    resp = rt.invoke_model_with_response_stream(
        modelId=MODEL_HAIKU,
        body=json.dumps(body),
    )

• invoke_model 대신 invoke_model_with_response_stream을 호출합니다. 이름만 다를 뿐 인자(modelId, body)는 같습니다.
• 반환된 resp["body"]는 완성된 JSON이 아니라 이벤트들의 스트림입니다. 여기서부터가 진짜 차이입니다.

4. 스트림을 돌며 글자 조각만 골라 yield

    # resp["body"] 는 이벤트들의 스트림. 하나씩 꺼내 chunk 바이트를 파싱한다.
    for event in resp["body"]:
        chunk = event["chunk"]["bytes"]
        data = json.loads(chunk)
        if data.get("type") == "content_block_delta":
            yield data["delta"].get("text", "")

• for event in resp["body"]: — 스트림에서 이벤트를 하나씩 꺼냅니다. 모델이 새 조각을 만들 때마다 루프가 한 번 돕니다.
• event["chunk"]["bytes"] — 각 이벤트 안의 실제 데이터는 바이트 형태입니다. 사람이 읽을 수 있는 JSON이 아니라 직렬화된 바이트라는 점에 주의하세요.
• json.loads(chunk) — 그 바이트를 JSON으로 파싱하여 파이썬 딕셔너리로 만듭니다.
• if data.get("type") == "content_block_delta": — 이 청크가 실제 글자를 담은 종류인지 확인합니다. 아니면 그냥 넘어갑니다(상태 신호이므로 출력할 글자가 없음).
• yield data["delta"].get("text", "") — 글자 조각을 꺼내 호출자에게 내보냅니다. .get("text", "")로 안전하게 꺼내, 텍스트가 없는 경우에도 오류 없이 빈 문자열을 내보냅니다.

5. 챗봇 UI에 연결

if __name__ == "__main__":
    from chat_ui import run
    run(respond_stream, streaming=True, title="Lab3 · invoke_model 스트리밍")

• 제공된 콘솔 챗봇 UI 모듈 chat_ui.py의 run에 우리가 만든 제너레이터 함수를 넘깁니다.
• streaming=True로 알려주면, UI는 respond_stream이 내놓는 조각을 하나씩 받아 흘러나오듯 화면에 출력합니다.
• 우리는 boto3 호출과 스트림 파싱만 책임지고, 화면 출력은 UI가 담당합니다.

code-server 터미널에서 Lab 폴더로 이동한 뒤, 스트리밍 챗봇을 실행합니다.

# 챗봇 실습 폴더로 이동
cd ~/nxt-workshop-code/bedrock-chatbot/starter

# 스트리밍 챗봇 실행
python lab3_streaming.py

실행되면 입력 프롬프트가 나타납니다. 긴 답변이 나오도록 질문해 보세요. 예를 들어 “스트리밍이 뭔지 비유를 들어서 길게 설명해줘”처럼 답이 길어질 만한 질문이 좋습니다.

답변이 한 번에 뜨지 않고, 글자가 왼쪽에서 오른쪽으로 흘러나오는 것을 확인하세요. 이것이 content_block_delta 조각이 도착하는 즉시 yield되어 화면에 출력되는 모습입니다.

관찰 포인트

여유가 있다면, 스트림에 어떤 type들이 흘러오는지 직접 눈으로 확인해 보세요. respond_stream 안에서 파싱한 data의 type을 잠깐 출력해 보면, message_start → content_block_start → content_block_delta(여러 번) → content_block_stop → message_stop 순서로 흐르는 것이 보입니다. 우리가 글자를 뽑아낸 곳은 그중 content_block_delta뿐이었음을 다시 확인할 수 있습니다.

이번 Lab은 동작하지만, 그 과정에서 저수준 API의 번거로움을 분명히 느꼈을 것입니다.

• raw 청크를 우리가 직접 처리한다. 이벤트를 하나씩 꺼내고, 바이트를 JSON으로 파싱하고, type을 보고 분기하고, 정확한 경로(delta.text)에서 글자를 꺼내는 일을 전부 손으로 했습니다.
• type이 바뀌면 코드도 바뀐다. 응답 포맷은 모델/버전에 종속적입니다. 만약 포맷이 달라지면 우리의 파싱 분기도 따라 바뀌어야 합니다. 즉 우리가 응답 형식까지 책임지고 있는 셈입니다.
• 모델마다 body가 다른 문제는 여전히 남아 있다. Lab 1~2에서 봤듯, Anthropic 모델과 Titan은 body 구조가 완전히 다릅니다. 스트리밍에서도 이 차이는 그대로입니다.

정리하면 지금까지 우리는 세 가지 번거로움을 직접 겪었습니다 — 모델별 body 형식, 수동 히스토리 관리, 그리고 이번의 raw 스트림 청크 처리입니다. 이 모든 것을 한 번에 풀어주는 것이 다음 Lab의 주제입니다.

• 청크 중 content_block_delta 말고 **message_delta**엔 무엇이 담겨 올까요? (힌트: stop_reason, 토큰)
• 이번 응답이 쓴 토큰 수는 어느 청크에서 확인할 수 있을까요?