스트리밍 출력

스트리밍 출력 (Streaming)

스트리밍 출력은 모델이 토큰을 몇 개씩 생성할 때마다(보통 1개) 즉시 클라이언트로 보내는 방식이에요. 전체 답변을 끝까지 기다렸다가 한 번에 받는 대신, 첫 번째 토큰부터 바로 볼 수 있으니 첫 토큰 대기 시간이 확 줄어들어요. 복잡하고 긴 답변은 완료까지 수 초에서 20초까지 걸리기도 하는데, 스트리밍을 켜면 이 대기 시간을 체감상 크게 줄일 수 있어요. Kimi 채팅에서 답변이 한 글자씩 "튀어나오는" 효과가 바로 스트리밍이에요. 이 문서는 Kimi API 공식 문서의 使用 Kimi API 的流式输出功能 페이지를 해요체로 옮긴 거예요. 원문은 Kimi API Docs에서 확인할 수 있어요.

스트리밍 켜기

요청에 stream=true를 주면 스트리밍이 켜져요. SDK는 이때 반복 가능한(iterable) 객체를 돌려주고, 반복문으로 데이터 블록(chunk)을 하나씩 읽으면 돼요. 각 chunk의 구조는 완성형 응답과 비슷하지만 message 필드가 delta로 바뀌어요. 스트리밍 응답에서 토큰 사용량도 받고 싶다면 stream_options: {"include_usage": true}를 함께 보내는 걸 권장해요.

delta에는 세 가지 유형의 증분 데이터가 나올 수 있어요.

  • content — 본문 내용. 조각 단위로 내려와요.
  • reasoning_content — 생각 모델의 추론 내용. content와 tool_calls보다 먼저 내려와요. SDK 타입 정의에는 선언돼 있지 않아서 Python에선 hasattr/getattr로 읽어야 해요.
  • tool_calls — 도구 호출. 같은 도구 호출의 조각은 같은 index를 공유하고, id·type·function.name은 첫 번째 조각에만 한 번 나타나요. function.arguments는 JSON 문자열 조각이라 덮어쓰지 말고 이어붙여야 스트림이 끝난 뒤 JSON으로 파싱할 수 있어요.

다음은 스트리밍에서 이 세 유형의 증분 데이터를 접어 모으는 Python 예시예요.

import os
import json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.cn/v1",
)

stream = client.chat.completions.create(
    model="kimi-k3",
    messages=[
        {"role": "system", "content": "당신은 Kimi, Moonshot AI의 AI 어시스턴트입니다."},
        {"role": "user", "content": "오늘 서울 날씨는 어때요?"},
    ],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "지정한 도시의 실시간 날씨를 조회합니다.",
            "parameters": {
                "type": "object",
                "required": ["city"],
                "properties": {"city": {"type": "string", "description": "도시 이름"}},
            },
        },
    }],
    stream=True,
    stream_options={"include_usage": True},
)

usage = None
finish_reason = None
reasoning_content = ""
tool_calls = []

for chunk in stream:
    chunk_usage = chunk.usage
    if chunk_usage is None and chunk.choices:
        chunk_usage = getattr(chunk.choices[0], "usage", None)
    if chunk_usage:
        usage = chunk_usage
    if not chunk.choices:
        continue

    choice = chunk.choices[0]
    delta = choice.delta

    if choice.finish_reason:
        finish_reason = choice.finish_reason

    if hasattr(delta, "reasoning_content"):
        fragment = getattr(delta, "reasoning_content")
        if fragment:
            reasoning_content += fragment
            print(fragment, end="")

    if delta.content:
        print(delta.content, end="")

    for tc in delta.tool_calls or []:
        idx = tc.index
        while len(tool_calls) <= idx:
            tool_calls.append({"id": "", "type": "", "name": "", "arguments": ""})
        cur = tool_calls[idx]
        if tc.id: cur["id"] = tc.id
        if tc.type: cur["type"] = tc.type
        if tc.function:
            if tc.function.name: cur["name"] = tc.function.name
            if tc.function.arguments: cur["arguments"] += tc.function.arguments

if finish_reason == "tool_calls":
    for tc in tool_calls:
        args = json.loads(tc["arguments"])
        print(f"\ntool_call: {tc['id']} {tc['name']}({args})")

if usage:
    print("\ntotal_tokens:", usage.total_tokens)

finish_reason가 tool_calls면 모델이 도구 실행을 요청한 거예요. 도구를 실행한 뒤 assistant 메시지(tool_calls와 접어 모은 reasoning_content, 즉 Preserved Thinking 포함)를 도구 결과와 함께 messages에 다시 넣고 API를 재호출하면 돼요.

SSE 응답 본문 파싱하기

스트리밍을 켜면 응답이 JSON이 아니라 Content-Type: text/event-stream(SSE)으로 내려와요. SSE 본문 모양:

data: {"id":"cmpl-1305b94c570f447fbde3180560736287","object":"chat.completion.chunk","created":1698999575,"model":"kimi-k3","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}

data: {"id":"cmpl-1305b94c570f447fbde3180560736287","object":"chat.completion.chunk","created":1698999575,"model":"kimi-k3","choices":[{"index":0,"delta":{"content":"안녕"},"finish_reason":null}]}

...

data: {"id":"cmpl-1305b94c570f447fbde3180560736287","object":"chat.completion.chunk","created":1698999575,"model":"kimi-k3","choices":[],"usage":{"prompt_tokens":19,"completion_tokens":13,"total_tokens":32}}

data: [DONE]

각 데이터 블록은 data: 접두사와 유효한 JSON 객체, 그리고 두 개의 개행 \n\n으로 끝나요. 모든 블록이 전송되면 서버가 data: [DONE]을 보내고 그때 연결을 끊으면 돼요.

주의: 전송 완료 판단은 **반드시 data: [DONE]**으로 해요. finish_reason 같은 걸로 판단하지 마세요. [DONE]을 받지 못했다면 finish_reason=stop이 이미 왔어도 전송이 끝난 게 아니에요. 즉 [DONE]을 받기 전까지는 메시지가 불완전한 상태로 봐야 해요.

스트리밍 중 content는 조각 단위로 내려오고, role은 매 블록마다 반복되지 않고 첫 블록에만 나타나요. 생각 모델이면 reasoning_content도 content보다 먼저 증분 조각으로 내려와요. 모델이 도구 호출을 결정하면 delta.tool_calls에 호출 조각이 실리고, 같은 호출의 조각은 같은 index를 공유하며 id/type/function.name은 첫 조각에만, function.arguments는 이어붙여야 해요. stream_options: {"include_usage": true}를 주면 서버가 [DONE] 직전에 최종 통계 블록을 반환하는데, 이 블록은 choices가 비어 있고 이번 요청의 총 사용량이 최상위 usage 필드에 들어 있어요.

어떤 언어로 하든 기본 절차는 같아요

  1. HTTP 요청을 보내고 요청 본문의 stream을 true로 설정해요.
  2. 응답 Headers의 Content-Type이 text/event-stream이면 스트리밍 응답이에요.
  3. 응답을 한 줄씩 읽으며 데이터 블록(JSON)을 파싱해요. data: 접두사와 개행 \n으로 블록의 시작과 끝을 판단해요.
  4. 블록 내용이 [DONE]이면 전송이 완료된 거예요.

토큰 사용량 집계

토큰 계산은 두 가지 방식이 있어요. 권장하는 방법은 요청에 stream_options: {"include_usage": true}를 넣고, 모든 블록이 전송된 뒤 마지막 통계 블록 최상위의 usage 필드(prompt_tokens/completion_tokens/total_tokens)를 읽는 거예요.

여러 응답 (n 파라미터)

참고: 현재 모델(kimi-k3, kimi-k2.7-code, kimi-k2.6)의 n은 고정 1이에요. 한 번의 요청으로 여러 응답을 받는 건 아직 지원하지 않고, 1보다 큰 n을 넣으면 invalid n: only 1 is allowed for this model 400 오류가 나요. 스트리밍/비스트리밍 모두 마찬가지예요.

더 알아보기