Kimi Streaming

상위 문서: ← Kimi Model Capabilities 출처: Kimi API Platform 공식 문서 — Use the Streaming Feature of the Kimi API 한글화 (2026-07-22) 공급사: Moonshot AI (月之暗面)

질문을 받은 후 Kimi LLM은 먼저 추론을 수행한 뒤 답변을 한 Token씩 생성합니다. 스트리밍은 특정 개수(보통 1 Token)의 Token이 생성되는 즉시 클라이언트로 전송하는 방식으로, 완전한 응답을 기다리지 않습니다. 완전한 응답을 기다리면 보통 수 초, 복잡한 질문과 긴 답변은 10~20초까지 걸리지만, 스트리밍을 사용하면 첫 Token을 즉시 볼 수 있어 대기 시간이 크게 줄어듭니다. Kimi AI 어시스턴트와 대화할 때 답변이 한 글자씩 나타나는 것이 바로 스트리밍입니다.

---

1. 스트리밍 출력 활성화

요청에 stream=True를 설정해 스트리밍을 활성화하세요. SDK는 iterable을 반환하며, 이를 순회하며 데이터 청크를 하나씩 읽습니다. 각 청크는 completion과 유사한 구조를 가지지만, message 필드 대신 delta 필드를 가집니다.

import os
from openai import OpenAI

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

stream = client.chat.completions.create(
    model="kimi-k3",
    messages=[
        {"role": "system", "content": "You are Kimi, an artificial intelligence assistant provided by Moonshot AI, who is better at conversing in Chinese and English. You provide users with safe, helpful, and accurate answers. At the same time, you refuse to answer any questions related to terrorism, racism, pornography, and violence. Moonshot AI is a proper noun and should not be translated into other languages."},
        {"role": "user", "content": "Hello, my name is Li Lei, what is 1+1?"}
    ],
    stream=True,  # <-- stream=True로 스트리밍 출력 모드 활성화
)

for chunk in stream:
    delta = chunk.choices[0].delta  # <-- message 필드가 delta로 교체됨

    if delta.content:
        # 문장 일관성을 위해 줄바꿈 없이 출력
        print(delta.content, end="")

---

2. SSE 응답 본문 파싱

스트리밍이 활성화되면 API는 더 이상 JSON 응답(Content-Type: application/json)을 반환하지 않고, **Content-Type: text/event-stream (SSE)**로 반환합니다 — 서버가 클라이언트에 Token을 지속적으로 push할 수 있습니다. 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":"Hello"},"finish_reason":null}]}

...

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

data: [DONE]

• 각 데이터 청크는 data: 접두사로 시작하고, 뒤이어 유효한 JSON 객체, 그리고 두 개의 newline 문자 \n\n로 끝남
• 모든 청크 전송이 끝나면 서버가 data: [DONE]을 보내 완료 표시 — 연결을 닫을 수 있음

주의: 전송 완료 판단은 **반드시 data: [DONE]**으로 하세요 — finish_reason 등으로 판단하지 마세요. finish_reason=stop이 도착했더라도 [DONE]이 오기 전까지는 불완전한 것으로 간주해야 합니다.

스트리밍 중 content 필드는 청크 단위로 전달되며, role과 usage는 매 청크마다 반복되지 않음 — role은 첫 청크에, usage는 마지막 청크에만 등장합니다.

---

3. 토큰 사용량 카운트

토큰을 카운트하는 두 가지 방법이 있습니다. 가장 직접적이고 정확한 방법은 모든 청크가 전송 완료될 때까지 기다린 뒤, 마지막 청크의 usage 필드를 읽는 것입니다 (prompt_tokens / completion_tokens / total_tokens).

...

data: {"id":"cmpl-1305b94c570f447fbde3180560736287","object":"chat.completion.chunk","created":1698999575,"model":"kimi-k3","choices":[{"index":0,"delta":{},"finish_reason":"stop","usage":{"prompt_tokens":19,"completion_tokens":13,"total_tokens":32}}]}
                                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                                               마지막 청크의 usage 필드로 토큰 수 확인
data: [DONE]

주의: usage는 청크의 최상위 레벨이 아니라 마지막 청크의 choices[0].usage에 중첩되어 있습니다. OpenAI SDK에서 chunk.usage는 None이므로 chunk.choices[0].usage를 읽거나 원시 SSE 청크를 직접 파싱해야 합니다.

하지만 네트워크 끊김이나 클라이언트 에러 등으로 스트림이 중단되면 마지막 청크가 도달하지 않아 토큰 소비량을 알 수 없습니다. 이를 피하려면 받은 모든 청크의 content를 저장해 두고, 요청 종료 후(성공/실패 무관) 토큰 카운트 엔드포인트를 호출해 실제 소비량을 계산하세요.

import os
import httpx
from openai import OpenAI

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

stream = client.chat.completions.create(
    model="kimi-k3",
    messages=[
        {"role": "system", "content": "You are Kimi, an AI assistant provided by Moonshot AI, who excels in Chinese and English conversations. You provide users with safe, helpful, and accurate answers while rejecting any questions related to terrorism, racism, or explicit content. Moonshot AI is a proper noun and should not be translated."},
        {"role": "user", "content": "Hello, my name is Li Lei. What is 1+1?"}
    ],
    stream=True,
)


def estimate_token_count(text: str) -> int:
    """토큰 계산 엔드포인트 호출"""
    headers = {"Authorization": f"Bearer {os.environ['MOONSHOT_API_KEY']}"}
    data = {"model": "kimi-k3", "messages": [{"role": "user", "content": text}]}
    r = httpx.post("https://api.moonshot.ai/v1/tokenizers/estimate-token-count",
                   headers=headers, json=data)
    r.raise_for_status()
    return r.json()["data"]["total_tokens"]


completion = []
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        completion.append(delta.content)

print("completion_tokens:", estimate_token_count("".join(completion)))

---

4. 스트리밍 출력 중단

조기에 출력을 종료하려면 HTTP 연결을 닫거나 후속 청크를 폐기하면 됩니다 — 예: 루프에서 break:

for chunk in stream:
    if condition:
        break

---

5. SDK 없이 SSE 처리

SDK가 없는 언어, 또는 SDK로 비즈니스 로직을 표현할 수 없는 경우 HTTP API를 직접 호출해 SSE 응답 본문을 처리할 수 있습니다. 다음은 SSE 응답 본문을 줄 단위로 읽고 파싱하는 예시입니다.

import os
import json
import httpx

data = {
    "model": "kimi-k3",
    "messages": [
        # 메시지
    ],
    "stream": True,
}

r = httpx.post(
    "https://api.moonshot.ai/v1/chat/completions",
    headers={"Authorization": f"Bearer {os.environ['MOONSHOT_API_KEY']}"},
    json=data,
)
if r.status_code != 200:
    raise Exception(r.text)

data: str

# iter_lines로 응답 본문을 한 줄씩 읽음
for line in r.iter_lines():
    line = line.strip()

    # 3가지 경우 처리:
    # 1) 빈 줄 → 이전 데이터 청크가 완료됨 → 역직렬화 후 처리
    # 2) 비어있지 않고 "data:"로 시작 → 데이터 청크 시작, [DONE]이 아니면 data에 저장
    # 3) 그 외 → 현재 줄이 이전 데이터 청크에 이어붙음

    if len(line) == 0:
        chunk = json.loads(data)
        choice = chunk["choices"][0]
        usage = choice.get("usage")
        if usage:
            print("total_tokens:", usage["total_tokens"])
        delta = choice["delta"]
        role = delta.get("role")
        if role:
            print("role:", role)
        content = delta.get("content")
        if content:
            print(content, end="")
        data = ""
    elif line.startswith("data: "):
        data = line[len("data: "):]
        if data == "[DONE]":
            break
    else:
        data = data + "\n" + line

---

6. 다중 응답 (n 파라미터)

현재 모델(kimi-k3, kimi-k2.7-code, kimi-k2.6)은 n을 1로 고정하고 있어, 한 요청에서 여러 응답을 반환하는 것을 지원하지 않습니다. 1보다 큰 n을 전달하면 스트리밍/비스트리밍 모두 400 에러(invalid n: only 1 is allowed for this model)를 반환합니다. 모델별 파라미터 제약은 Model Parameter Reference 참고.

---

변경 이력

날짜	변경
2026-07-22	초판 작성 (공식 Streaming 가이드 한글화)
2026-07-22	제목에 "Kimi" 접두사 추가, 본문 H1 중복 제거