세밀한 도구 스트리밍

빌드도구 인프라

세분화된 도구 스트리밍

지연 시간에 민감한 애플리케이션을 위해 도구 입력을 문자 단위로 스트리밍합니다.

This feature is eligible for Zero Data Retention (ZDR). When your organization has a ZDR arrangement, data sent through this feature is not stored after the API response is returned.

세분화된 도구 스트리밍은 모든 모델과 모든 플랫폼에서 일반적으로 사용 가능합니다. 이를 통해 버퍼링이나 JSON 검증 없이 도구 사용 매개변수 값의 스트리밍이 가능하므로, 큰 매개변수를 받기 시작하는 지연 시간을 줄일 수 있습니다.

세분화된 도구 스트리밍을 사용할 때, 잘못되었거나 부분적인 JSON 입력을 받을 수 있습니다. 코드에서 이러한 엣지 케이스를 처리하도록 해야 합니다.

세분화된 도구 스트리밍 사용 방법

세분화된 도구 스트리밍은 모든 모델과 모든 플랫폼(Claude API, Amazon Bedrock, Google Vertex AI, Microsoft Foundry)에서 사용 가능합니다. 이를 사용하려면 세분화된 스트리밍을 활성화하려는 모든 사용자 정의 도구에서 eager_input_streaming을 true로 설정하고 요청에서 스트리밍을 활성화하세요.

다음은 API에서 세분화된 도구 스트리밍을 사용하는 방법의 예입니다:

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-7",
    tools=[
        {
            "name": "make_file",
            "description": "Write text to a file",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "The filename to write text to",
                    },
                    "lines_of_text": {
                        "type": "array",
                        "description": "An array of lines of text to write to the file",
                    },
                },
                "required": ["filename", "lines_of_text"],
            },
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "Can you write a long poem and make a file called poem.txt?",
        }
    ],
) as stream:
    for event in stream:
        pass
    final_message = stream.get_final_message()

print(final_message.usage)

이 예제에서 세분화된 도구 스트리밍을 통해 Claude는 lines_of_text 매개변수가 유효한 JSON인지 검증하기 위해 버퍼링하지 않고도 긴 시의 줄들을 make_file 도구 호출로 스트리밍할 수 있습니다. 이는 전체 매개변수가 버퍼링되고 검증될 때까지 기다릴 필요 없이 매개변수가 도착할 때 스트리밍되는 것을 볼 수 있다는 의미입니다.

세분화된 도구 스트리밍을 사용하면 도구 사용 청크가 더 빠르게 스트리밍되기 시작하며, 종종 더 길고 단어 끊김이 적습니다. 이는 청킹 동작의 차이 때문입니다.

예제:

세분화된 스트리밍 없음(15초 지연):

청크 1: '{"'
청크 2: 'query": "Ty'
청크 3: 'peScri'
청크 4: 'pt 5.0 5.1 '
청크 5: '5.2 5'
청크 6: '.3'
청크 8: ' new f'
청크 9: 'eatur'
...

세분화된 스트리밍 사용(3초 지연):

청크 1: '{"query": "TypeScript 5.0 5.1 5.2 5.3'
청크 2: ' new features comparison'

세분화된 스트리밍은 버퍼링이나 JSON 검증 없이 매개변수를 전송하므로, 결과 스트림이 유효한 JSON 문자열로 완료된다는 보장이 없습니다. 특히 중지 이유 max_tokens에 도달하면 스트림이 매개변수 중간에 끝날 수 있으며 불완전할 수 있습니다. 일반적으로 max_tokens에 도달했을 때를 처리하기 위한 특정 지원을 작성해야 합니다.

도구 입력 델타 누적

tool_use 콘텐츠 블록이 스트리밍될 때, 초기 content_block_start 이벤트는 input: {} (빈 객체)를 포함합니다. 이는 자리 표시자입니다. 실제 입력은 일련의 input_json_delta 이벤트로 도착하며, 각각은 partial_json 문자열 조각을 전달합니다. 코드는 이러한 조각들을 연결하고 블록이 닫힐 때 결과를 파싱해야 합니다.

누적 계약:

type: "tool_use"인 content_block_start에서 빈 문자열을 초기화합니다: input_json = ""
type: "input_json_delta"인 각 content_block_delta에 대해 다음을 추가합니다: input_json += event.delta.partial_json
content_block_stop에서 누적된 문자열을 파싱합니다: json.loads(input_json)

초기 input: {} (객체)와 partial_json (문자열) 사이의 타입 불일치는 의도적입니다. 빈 객체는 콘텐츠 배열의 슬롯을 표시하고, 델타 문자열은 실제 값을 구축합니다.

import json
import anthropic

client = anthropic.Anthropic()

tool_inputs = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-7",
    max_tokens=1024,
    tools=[
        {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        }
    ],
    messages=[{"role": "user", "content": "Weather in Paris?"}],
) as stream:
    for event in stream:
        if (
            event.type == "content_block_start"
            and event.content_block.type == "tool_use"
        ):
            tool_inputs[event.index] = ""
        elif (
            event.type == "content_block_delta"
            and event.delta.type == "input_json_delta"
        ):
            tool_inputs[event.index] += event.delta.partial_json
        elif event.type == "content_block_stop" and event.index in tool_inputs:
            parsed = json.loads(tool_inputs[event.index])
            print(f"Tool input: {parsed}")

Python 및 TypeScript SDK는 이 누적을 수행하는 더 높은 수준의 스트림 헬퍼(stream.get_final_message(), stream.finalMessage())를 제공합니다. 블록이 닫히기 전에 부분 입력에 반응해야 하는 경우(예: 진행 표시기 렌더링 또는 다운스트림 요청 조기 시작)에만 위의 수동 패턴을 사용하세요.

도구 응답에서 잘못된 JSON 처리

세분화된 도구 스트리밍을 사용할 때, 모델로부터 잘못되었거나 불완전한 JSON을 받을 수 있습니다. 이 잘못된 JSON을 오류 응답 블록의 모델로 다시 전달해야 하는 경우, 적절한 처리를 보장하기 위해 JSON 객체로 래핑할 수 있습니다(합리적인 키 사용). 예를 들어:

{
  "INVALID_JSON": "<your invalid json string>"
}

이 접근 방식은 모델이 콘텐츠가 잘못된 JSON임을 이해하도록 도우면서 디버깅 목적으로 원본 형식이 잘못된 데이터를 보존합니다.

잘못된 JSON을 래핑할 때, 래퍼 객체에서 유효한 JSON 구조를 유지하기 위해 잘못된 JSON 문자열의 따옴표나 특수 문자를 적절히 이스케이프해야 합니다.

다음 단계

메시지 스트리밍

서버 전송 이벤트 및 스트림 이벤트 타입에 대한 전체 참조입니다.

도구 호출 처리

도구를 실행하고 필요한 메시지 형식으로 결과를 반환합니다.

도구 참조

Anthropic 스키마 도구 및 해당 버전 문자열의 전체 디렉토리입니다.

Was this page helpful?

빌드도구 인프라

세분화된 도구 스트리밍

지연 시간에 민감한 애플리케이션을 위해 도구 입력을 문자 단위로 스트리밍합니다.

This feature is eligible for Zero Data Retention (ZDR). When your organization has a ZDR arrangement, data sent through this feature is not stored after the API response is returned.

세분화된 도구 스트리밍을 사용할 때, 잘못되었거나 부분적인 JSON 입력을 받을 수 있습니다. 코드에서 이러한 엣지 케이스를 처리하도록 해야 합니다.

세분화된 도구 스트리밍 사용 방법

다음은 API에서 세분화된 도구 스트리밍을 사용하는 방법의 예입니다:

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-7",
    tools=[
        {
            "name": "make_file",
            "description": "Write text to a file",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "The filename to write text to",
                    },
                    "lines_of_text": {
                        "type": "array",
                        "description": "An array of lines of text to write to the file",
                    },
                },
                "required": ["filename", "lines_of_text"],
            },
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "Can you write a long poem and make a file called poem.txt?",
        }
    ],
) as stream:
    for event in stream:
        pass
    final_message = stream.get_final_message()

print(final_message.usage)

예제:

세분화된 스트리밍 없음(15초 지연):

청크 1: '{"'
청크 2: 'query": "Ty'
청크 3: 'peScri'
청크 4: 'pt 5.0 5.1 '
청크 5: '5.2 5'
청크 6: '.3'
청크 8: ' new f'
청크 9: 'eatur'
...

세분화된 스트리밍 사용(3초 지연):

청크 1: '{"query": "TypeScript 5.0 5.1 5.2 5.3'
청크 2: ' new features comparison'

도구 입력 델타 누적

누적 계약:

type: "tool_use"인 content_block_start에서 빈 문자열을 초기화합니다: input_json = ""
type: "input_json_delta"인 각 content_block_delta에 대해 다음을 추가합니다: input_json += event.delta.partial_json
content_block_stop에서 누적된 문자열을 파싱합니다: json.loads(input_json)

import json
import anthropic

client = anthropic.Anthropic()

tool_inputs = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-7",
    max_tokens=1024,
    tools=[
        {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        }
    ],
    messages=[{"role": "user", "content": "Weather in Paris?"}],
) as stream:
    for event in stream:
        if (
            event.type == "content_block_start"
            and event.content_block.type == "tool_use"
        ):
            tool_inputs[event.index] = ""
        elif (
            event.type == "content_block_delta"
            and event.delta.type == "input_json_delta"
        ):
            tool_inputs[event.index] += event.delta.partial_json
        elif event.type == "content_block_stop" and event.index in tool_inputs:
            parsed = json.loads(tool_inputs[event.index])
            print(f"Tool input: {parsed}")

도구 응답에서 잘못된 JSON 처리

{
  "INVALID_JSON": "<your invalid json string>"
}

이 접근 방식은 모델이 콘텐츠가 잘못된 JSON임을 이해하도록 도우면서 디버깅 목적으로 원본 형식이 잘못된 데이터를 보존합니다.

잘못된 JSON을 래핑할 때, 래퍼 객체에서 유효한 JSON 구조를 유지하기 위해 잘못된 JSON 문자열의 따옴표나 특수 문자를 적절히 이스케이프해야 합니다.

다음 단계

메시지 스트리밍

서버 전송 이벤트 및 스트림 이벤트 타입에 대한 전체 참조입니다.

도구 호출 처리

도구를 실행하고 필요한 메시지 형식으로 결과를 반환합니다.

도구 참조

Anthropic 스키마 도구 및 해당 버전 문자열의 전체 디렉토리입니다.

Was this page helpful?