AI 에이전트 블로그
컴퓨터 액세스 시스템

에이전트 응답 스트리밍: 멀티스텝 워크플로우를 위한 실시간 출력

streaming agent responsesreal-time agentsprogressive agent outputagent UXstreaming tool callsClaude APIserver-sent events

에이전트 응답 스트리밍: 멀티스텝 워크플로우를 위한 실시간 출력

에이전트가 질문을 리서치하고, 세 가지 도구를 호출하고, 답변을 종합하는 데 20초가 걸린다고 해봅시다. 사용자는 “질문하기”를 클릭하고 20초 동안 스피너만 바라봅니다. 제대로 작동하는 건지 알 수 없습니다. 페이지를 새로고침할까 고민합니다. 처음부터 다시 시작해야 하나 생각합니다.

이제 다른 상황을 상상해보세요. 사용자가 “질문하기”를 클릭하면 500밀리초 안에 에이전트의 첫 단어가 나타납니다. 에이전트가 정보를 검색하는 모습—”🔍 주문 데이터베이스 검색 중…”—을 보고, 결과가 실시간으로 도착하는 걸 확인합니다. 답변이 문장 단위로 작성되는 걸 읽어나갑니다. 걸리는 시간은 똑같이 20초지만, 경험은 완전히 다릅니다.

스트리밍은 사용자 대면 에이전트에서 있으면 좋은 기능이 아닙니다—UX의 필수 요소입니다. 첫 토큰까지의 시간(Time-to-first-token)은 에이전트 인터페이스에서 가장 중요한 지연 시간 지표입니다. 진행 상황을 보는 사용자는 인내심이 있습니다. 아무것도 보이지 않으면 이탈합니다. 웹 성능에 관한 수많은 연구에서 체감 지연 시간이 실제 지연 시간보다 더 중요하다는 것이 밝혀졌으며, 스트리밍은 이 둘의 격차를 줄이는 가장 강력한 도구입니다.

이 글에서는 멀티스텝 AI 에이전트에 실시간 스트리밍을 구현하는 방법을 알아봅니다. 기본적인 토큰 단위 전송부터 도구 호출 투명성, 상태 업데이트, 점진적 공개, 오류 복구, 전송 계층 선택까지 다룹니다.

섹션 1: Claude 스트리밍 API 기초

에이전트를 스트리밍하기 전에, 단일 Claude 응답을 스트리밍하는 방법부터 시작해야 합니다. 기본 개념을 살펴봅시다.

스트림 vs. 비스트림

비스트리밍 API 호출은 전체 응답이 생성될 때까지 블로킹 상태를 유지한 후, 한 번에 모든 내용을 반환합니다. 스트리밍 호출은 응답이 생성되는 동안 첫 번째 토큰부터 시작하여 이벤트 시퀀스를 반환합니다.

코드 상의 차이는 미미합니다. 사용자 경험의 차이는 엄청납니다.

이벤트 유형

Claude 스트리밍 API는 구조화된 서버-전송 이벤트(server-sent events) 시퀀스를 내보냅니다:

  1. message_start — 메타데이터(모델, 역할, 사용량)가 포함된 초기 Message 객체를 담습니다.
  2. content_block_start — 콘텐츠 블록(텍스트 또는 tool_use)의 시작을 알립니다.
  3. content_block_delta — 텍스트 조각이나 부분적인 도구 입력 JSON 등 증분 콘텐츠를 담습니다.
  4. content_block_stop — 콘텐츠 블록의 끝을 알립니다.
  5. message_delta — 메시지에 대한 최종 업데이트(중단 이유, 최종 사용량)를 담습니다.
  6. message_stop — 스트림이 완료되었음을 알립니다.

텍스트 응답의 경우, 각각 소량의 텍스트 청크(보통 몇 개의 토큰)를 담은 content_block_delta 이벤트를 여러 번 받게 됩니다.

기본 스트리밍 구현

다음은 Anthropic Python SDK를 사용한 동기식 스트리밍 예시입니다:

import anthropic


client = anthropic.Anthropic()


def stream_basic_response(user_message: str):
    """Stream a basic Claude response token by token."""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # Newline after stream completes


stream_basic_response("Explain quantum entanglement in simple terms.")

그리고 프로덕션 웹 서버에서 사용할 비동기 버전입니다:

import anthropic
import asyncio


async_client = anthropic.AsyncAnthropic()


async def stream_basic_response_async(user_message: str):
    """Async streaming with the Anthropic SDK."""
    async with async_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()


asyncio.run(stream_basic_response_async("Explain quantum entanglement in simple terms."))

부분 토큰과 UTF-8 경계 처리

SDK가 UTF-8 디코딩을 자동으로 처리하지만, 원시 HTTP 스트림을 직접 다루는 경우에는 멀티바이트 문자가 청크 경계에서 분리될 수 있다는 점에 주의하세요. 항상 원시 바이트를 버퍼링하고 완전한 UTF-8 시퀀스가 확보됐을 때만 디코딩하세요. anthropic SDK의 text_stream 이터레이터가 이를 자동으로 처리합니다—원시 SSE 스트림을 직접 파싱하는 대신 SDK를 사용해야 하는 또 다른 이유입니다.

섹션 2: 도구 호출 스트리밍

기본 텍스트 스트리밍은 기본 중의 기본입니다. 진짜 도전—그리고 진짜 가치—은 에이전트가 도구를 사용할 때 발생합니다. 세 가지 도구를 순서대로 호출하는 멀티스텝 에이전트는 도구 실행 중에 무슨 일이 일어나고 있는지 보여주지 않으면 죽어있는 것처럼 느껴질 수 있습니다.

스트림에서 도구 호출이 나타나는 방식

Claude가 도구를 사용하기로 결정하면, 스트림은 type: "tool_use"가 포함된 content_block_start 이벤트를 내보내고, 이어서 도구의 입력 JSON 조각이 담긴 content_block_delta 이벤트들이 뒤따릅니다. 완전한 도구 호출이 조립되면 도구를 실행하고, 결과를 주입하여 대화를 계속합니다.

핵심 인사이트: content_block_start가 도착하는 즉시 도구 이름을 알 수 있습니다, 입력 JSON이 완성되기 전에도 마찬가지입니다. 즉, 전체 도구 호출을 기다리지 않고도 즉시 ”🔍 주문 데이터베이스 검색 중…”과 같은 내용을 사용자에게 보여줄 수 있습니다.

완전한 스트리밍 에이전트 루프

다음은 도구 호출을 실시간으로 표시하는 완전한 스트리밍 에이전트 루프입니다:

import anthropic
import json


client = anthropic.Anthropic()


# Define tools
tools = [
    {
        "name": "search_orders",
        "description": "Search customer orders by order ID or customer email.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Order ID or email"},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_shipping_status",
        "description": "Get real-time shipping status for an order.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "The order ID"},
            },
            "required": ["order_id"],
        },
    },
]




def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "search_orders":
        # Simulated database lookup
        return json.dumps({
            "order_id": "ORD-12345",
            "customer": "jane@example.com",
            "items": ["Blue Widget x2", "Red Gadget x1"],
            "total": "$47.99",
            "status": "shipped",
        })
    elif tool_name == "get_shipping_status":
        return json.dumps({
            "order_id": tool_


---


## 관련 기사


- [도구 사용 패턴: 신뢰할 수 있는 에이전트-도구 인터페이스 구축](/ko/blog/agent-tool-use-patterns/)
- [멀티에이전트 패턴: 오케스트레이터, 워커, 파이프라인](/ko/blog/multi-agent-patterns/)
- [에이전트 오류 복구: 프로덕션 신뢰성을 위한 5가지 패턴](/ko/blog/agent-error-recovery-patterns/)
- [웹 자동화 에이전트: Claude와 Computer Use를 활용한 브라우저 제어](/ko/blog/web-automation-agents-browser-control-with-claude-and-computer-use/)