NDJSON → SSE proxy

왜 이 proxy 만들어?

대부분의 cloud-shape chat UI랑 SDK (OpenAI Python SDK, Vercel AI SDK, LangChain streaming hook)는 SSE를 먹고 살아. Frontend를 다시 안 짜고 Ollama를 drop-in하고 싶으면, NDJSON을 SSE로 옮겨주는 작은 proxy를 하나 세워.

변환의 모양

Ollama는 {json}\n을 뱉고, SSE는 data: {json}\n\n을 원해.
OpenAI-compatible SSE client는 끝에 data: [DONE]\n\n sentinel이 오길 기대하고.
data: 안의 chunk 모양도 보통 OpenAI delta 모양으로 바꿔줘야 해: {"choices": [{"delta": {"content": "..."}}]}.

OpenAI 호환성 reinvent하지 마

Ollama 이미 /v1/chat/completions를 OpenAI-compatible endpoint로 노출해. "drop-in OpenAI SDK"만 필요하면 SDK의 base_url을 http://localhost:11434/v1로 맞춰두는 걸로 끝이야. proxy는 건너뛰어도 돼. Ollama의 compat endpoint가 안 해주는 변환 (커스텀 auth, observability, request shaping)이 필요할 때만 직접 proxy를 세워.

Code

FastAPI proxy: NDJSON → OpenAI delta shape SSE·python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx, json

app = FastAPI()

async def ollama_to_openai_sse(model: str, messages: list[dict]):
    """Ollama NDJSON을 OpenAI-shape SSE로 bridge."""
    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:
            async for line in r.aiter_lines():
                if not line:
                    continue
                chunk = json.loads(line)
                content = chunk.get("message", {}).get("content", "")
                done = chunk.get("done", False)

                payload = {
                    "id": "ollama-stream",
                    "object": "chat.completion.chunk",
                    "model": model,
                    "choices": [{
                        "index": 0,
                        "delta": {"content": content} if content else {},
                        "finish_reason": "stop" if done else None,
                    }],
                }
                yield f"data: {json.dumps(payload)}\n\n"
                if done:
                    yield "data: [DONE]\n\n"
                    return

@app.post("/v1/chat/completions")
async def chat_completions(req: dict):
    return StreamingResponse(
        ollama_to_openai_sse(req["model"], req["messages"]),
        media_type="text/event-stream",
    )

아니면 Ollama 내장 OpenAI compat 그냥 써·bash

# Drop-in OpenAI 호환만 원하면 proxy 안 필요
# OpenAI client를 이렇게 가리키기:
#   base_url = http://localhost:11434/v1
#   api_key = "ollama"   (빈 값 아닌 아무거나)

# OpenAI Python SDK
python3 - <<'PY'
from openai import OpenAI
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
stream = client.chat.completions.create(
    model="qwen2.5:7b",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="", flush=True)
PY

왜 이 proxy 만들어?

변환의 모양

OpenAI 호환성 reinvent하지 마

Code

External links

Exercise

Progress

댓글 0