본문 바로가기
C.W.K.
Stream
Lesson 04 of 04 · published

OpenAI 호환성 test

~18 min · serving, compatibility, openai

Level 0Downloader
0 XP0/41 lessons0/11 achievements
0/120 XP to next level120 XP to go0% complete

OpenAI 호환성은 spectrum

모든 local engine이 OpenAI-compatible이라고 말해. 실제로도 다 거의 호환돼. 차이는 예측은 되는데 표준화는 안 돼 있어:

  • Streaming 형식. 다 SSE를 뱉는데 chunk 모양이 조금씩 달라.
  • Tool calling. Native Ollama는 args를 dict로 풀어서 주고, OpenAI는 JSON 문자열로 뱉고, 어떤 engine은 또 다르게 감싸.
  • Structured output. Ollama는 JSON Schema를 format 필드에 받고, OpenAI는 response_format을 받아. vLLM도 그걸 받고. 예전 guided_json 계열은 v0.12.0(2025년 12월)에서 빠졌어. 아래 harness가 어디에나 response_format을 보내는 이유가 그거야. 같은 아이디어에 다른 키고, 그 키들은 시간이 지나면서 서로 수렴해.
  • Token counting. token usage 필드가 조금씩 다르거나, 아예 없기도 해.

호환성 어떻게 test?

Label을 믿지 마. 네가 실제로 쓰는 기능만 골라서 돌려보는 작은 test suite를 만들어. 아래가 local-AI 호환성 test의 정석이야:

  1. 단순 message로 non-streaming chat.
  2. 같은 message로 streaming chat.
  3. Multi-turn 대화 (system + user + assistant).
  4. Tool 정의 + multi-turn tool loop.
  5. Structured output (JSON Schema).
  6. 응답의 token usage.

어떤 engine이 여섯 개를 다 통과하면 app을 안 고치고 그 engine으로 갈아탈 수 있어. 4번이나 5번이 걸리면 engine별 shim을 하나 써야 한다고 각오해.

Code

호환성 test harness·python
import httpx, json

TOOLS = [{"type": "function", "function": {
    "name": "say_ok", "description": "Reply with 'ok'.",
    "parameters": {"type": "object", "properties": {}, "required": []},
}}]

PERSON_SCHEMA = {
    "type": "object",
    "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
    "required": ["name", "age"],
}

def test_compat(base_url: str, model: str, label: str = ""):
    """위 목록의 기능 하나에 check 하나씩, 여섯 개. {n: bool}을 돌려줘."""
    print(f"\n=== {label or base_url} ===")
    kw = {"timeout": 120.0}
    results: dict[int, bool] = {}

    def report(n: int, name: str, ok: bool, detail: str = ""):
        results[n] = ok
        print(f"{n}. {name:<18} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")

    # 1. Non-streaming chat
    r = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Say 'ok' and nothing else."}],
        "stream": False,
    }, **kw)
    report(1, "non-stream", r.status_code == 200)

    # 2. Streaming chat
    with httpx.stream("POST", f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Count: 1 2 3"}],
        "stream": True,
    }, timeout=None) as r2:
        chunks = sum(1 for line in r2.iter_lines() if line.startswith("data:"))
    report(2, "stream", chunks > 1, f"({chunks} chunks)")

    # 3. Multi-turn (system + user + assistant + user)
    r3 = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [
            {"role": "system", "content": "You answer with one word."},
            {"role": "user", "content": "My code word is BANANA. Remember it."},
            {"role": "assistant", "content": "Noted."},
            {"role": "user", "content": "What is my code word?"},
        ],
        "stream": False,
    }, **kw)
    ok3 = r3.status_code == 200 and "banana" in r3.text.lower()
    report(3, "multi-turn", ok3)

    # 4. Tool 정의 + tool loop (호출하고, 결과를 다시 밀어 넣기)
    r4 = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Use the say_ok tool."}],
        "tools": TOOLS, "stream": False,
    }, **kw)
    calls = (r4.json().get("choices", [{}])[0].get("message", {}) or {}).get("tool_calls") or []
    ok4 = bool(calls)
    if ok4:
        # loop의 나머지 반: engine이 tool 결과를 되받아 줘야 해.
        r4b = httpx.post(f"{base_url}/chat/completions", json={
            "model": model,
            "messages": [
                {"role": "user", "content": "Use the say_ok tool."},
                {"role": "assistant", "content": "", "tool_calls": calls},
                {"role": "tool", "tool_call_id": calls[0].get("id", "0"), "content": "ok"},
            ],
            "tools": TOOLS, "stream": False,
        }, **kw)
        ok4 = r4b.status_code == 200
    report(4, "tool loop", ok4)

    # 5. Structured output (JSON Schema)
    r5 = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Invent a person."}],
        "response_format": {"type": "json_schema", "json_schema": {
            "name": "person", "schema": PERSON_SCHEMA}},
        "stream": False,
    }, **kw)
    ok5 = False
    if r5.status_code == 200:
        try:
            obj = json.loads(r5.json()["choices"][0]["message"]["content"])
            ok5 = isinstance(obj.get("name"), str) and isinstance(obj.get("age"), int)
        except (json.JSONDecodeError, KeyError, AttributeError):
            ok5 = False
    report(5, "structured", ok5)

    # 6. 응답의 token usage
    usage = r.json().get("usage") if r.status_code == 200 else None
    ok6 = bool(usage) and usage.get("total_tokens", 0) > 0
    report(6, "usage", ok6, f"({usage})" if usage else "")

    return results

# 여러 engine에 돌려
test_compat("http://localhost:11434/v1", "qwen2.5:7b",   label="Ollama")
test_compat("http://localhost:8080/v1",  "qwen-direct",  label="llama-server")
# test_compat("http://localhost:8000/v1", "Qwen/Qwen2.5-7B-Instruct", label="vLLM")

External links

Exercise

호환성 test harness를 Ollama랑 다른 engine 하나에 돌려봐. 여섯 기능 중 뭐가 pass고 뭐가 fail인지 list로 적어. 실패한 게 네 use case에 실제로 의미가 있는지 판단하고, 의미가 있으면 engine별 shim 계획을 세워.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인 — 댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.