httpx로 Python streaming

패턴

제대로 된 Python Ollama streaming client가 챙기는 세 가지:

timeout=None — Cold model의 첫 토큰 latency가 수십 초 갈 수 있어. Default httpx timeout은 warm-up 도중에 stream을 죽여. None으로 두고 server한테 맡겨.
iter_lines() — HTTP client가 newline까지 buffer하다가 완성된 줄만 yield해줘. Raw byte를 읽어서 줄을 직접 재조립하지 마.
flush=True — 토큰 찍을 때 꼭 켜. 안 그러면 stdio buffer가 streaming 느낌을 삼켜버려.

Production에서 어디 쓰여

이게 네 backend의 streaming endpoint를 짓는 벽돌이야. 피파의 Ollama vessel이 정확히 이 패턴 — Ollama에서 NDJSON stream, 각 chunk를 universal StreamChunk로 변환, JSONL ground truth에 write, frontend로 yield.

흔한 실수

timeout=None을 깜빡함. Default timeout이 cold start에서 stream을 죽여.
줄 단위가 아니라 char 단위로 buffering함. iter_text()는 network 경계에서 잘리는데, iter_lines()가 그 경계 문제를 대신 처리해줘.
빈 마지막 chunk를 안 걸러냄. done: true는 빈 content를 달고 와. 찍기 전에 break해.

Code

Production-grade streamer·python

import httpx
import json
from typing import Iterator

def stream_chat(model: str, messages: list[dict],
                base_url: str = "http://localhost:11434",
                **options) -> Iterator[dict]:
    """NDJSON chunk를 /api/chat에서 stream. 파싱된 dict yield."""
    payload = {"model": model, "messages": messages, "stream": True}
    if options:
        payload["options"] = options

    with httpx.stream("POST", f"{base_url}/api/chat",
                      json=payload, timeout=None) as r:
        r.raise_for_status()
        for line in r.iter_lines():
            if not line:
                continue
            yield json.loads(line)

# 사용 — 토큰 도착하는 대로 print, metric 캡처
full = ""
for chunk in stream_chat("qwen2.5:7b",
                          [{"role": "user", "content": "Explain MLX in 3 sentences."}]):
    if chunk.get("done"):
        eval_count = chunk.get("eval_count", 0)
        eval_dur_ns = chunk.get("eval_duration", 1)
        tps = eval_count / (eval_dur_ns / 1e9)
        print(f"\n[{eval_count} tokens @ {tps:.1f} tok/s]")
        break
    delta = chunk.get("message", {}).get("content", "")
    full += delta
    print(delta, end="", flush=True)

httpx.AsyncClient로 async 버전·python

import httpx, json
from typing import AsyncIterator

async def stream_chat_async(model: str, messages: list[dict]) -> AsyncIterator[dict]:
    """Async streaming — FastAPI / asyncio server에 선호."""
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "POST",
            "http://localhost:11434/api/chat",
            json={"model": model, "messages": messages, "stream": True},
        ) as r:
            r.raise_for_status()
            async for line in r.aiter_lines():
                if not line:
                    continue
                yield json.loads(line)

# Async context에서 사용
async def main():
    async for chunk in stream_chat_async(
        "qwen2.5:7b",
        [{"role": "user", "content": "What is GGUF?"}],
    ):
        if chunk.get("done"):
            break
        print(chunk["message"]["content"], end="", flush=True)

패턴

Production에서 어디 쓰여

흔한 실수

Code

External links

Exercise

Progress

댓글 0