Skip to content

README.md

Latest commit

 

History

History
327 lines (252 loc) · 10 KB

README.md

File metadata and controls

327 lines (252 loc) · 10 KB

SSE Proxy Python Example

ストリーミング API レスポンスを SSE として中継する Python 実装例です。

概要

このアプリは、ストリーミング LLM API レスポンスを中継する FastAPI ベースの BFF です。クライアントは LLM API を直接呼び出さず、このアプリを呼び出します。 このアプリはサーバー側に保持した API key を使って、OpenAI 互換の /responses endpoint へリクエストを転送します。

3つの proxy endpoint は同じ request body を受け取り、いずれも text/event-stream を返します。違いは、上流 LLM API を呼び出す Python client の実装です。

上流 client Endpoint 確認できること
OpenAI Python SDK /openai-python/responses 公式 SDK の stream event を SSE に戻す実装
httpx /httpx/responses 汎用 async HTTP client で上流 SSE bytes をそのまま中継する実装
aiohttp /aiohttp/responses aiohttp の async client streaming API で上流 SSE bytes をそのまま中継する実装

LLM 接続設定はサーバー側の環境変数から読みます。ブラウザや request body から API key は受け取りません。

はじめかた

依存関係をインストールします。

uv sync

Ollama などのローカル OpenAI 互換 server を使う場合は、先に Ollama server と model を準備してから、このアプリを起動します。

ollama serve

別の terminal で model を pull し、このアプリを起動します。

ollama pull llama3.2

LLM_BASE_URL=http://localhost:11434/v1/ \
LLM_API_KEY=ollama \
LLM_MODEL=llama3.2 \
uv run uvicorn sse_proxy_python_example.app:app --reload

OpenAI API を使う場合は、次のように起動します。

LLM_BASE_URL=https://api.openai.com/v1/ \
LLM_API_KEY="$OPENAI_API_KEY" \
LLM_MODEL=gpt-5-mini \
uv run uvicorn sse_proxy_python_example.app:app --reload

別の terminal から SSE chunk が流れることを確認します。

curl -N http://127.0.0.1:8000/httpx/responses \
  -H 'Content-Type: application/json' \
  -d '{"input":"Write one short sentence about SSE."}'

設定

環境変数 既定値 説明
LLM_BASE_URL http://localhost:11434/v1/ OpenAI 互換 API の base URL
LLM_API_KEY ollama 上流 server へ送る API key
LLM_MODEL llama3.2 /responses へ送る model 名
LLM_REQUEST_TIMEOUT 120.0 上流 request timeout 秒数
LLM_PUBLIC_BASE_URL http://127.0.0.1:8000 起動ログに表示する公開 base URL

ローカル起動

uv sync
LLM_BASE_URL=http://localhost:11434/v1/ \
LLM_API_KEY=ollama \
LLM_MODEL=llama3.2 \
uv run uvicorn sse_proxy_python_example.app:app --reload

既定では http://127.0.0.1:8000 で待ち受けます。

Docker 起動

docker build -t sse-proxy-python-example .
docker run --rm -p 8000:8000 \
  -e LLM_BASE_URL=http://host.docker.internal:11434/v1/ \
  -e LLM_API_KEY=ollama \
  -e LLM_MODEL=llama3.2 \
  sse-proxy-python-example

OpenAI API を使う場合は、API key を image に埋め込まず、環境変数として渡します。

docker run --rm -p 8000:8000 \
  -e LLM_BASE_URL=https://api.openai.com/v1/ \
  -e LLM_API_KEY="$OPENAI_API_KEY" \
  -e LLM_MODEL=gpt-5-mini \
  sse-proxy-python-example

Request 例

同じ payload を3つの endpoint に投げて、SSE の出力を比較できます。 curl -N を使うと chunk が到着したタイミングで逐次表示されます。

payload='{"input":"Write one short sentence about SSE."}'

curl -N http://127.0.0.1:8000/openai-python/responses \
  -H 'Content-Type: application/json' \
  -d "$payload"

curl -N http://127.0.0.1:8000/httpx/responses \
  -H 'Content-Type: application/json' \
  -d "$payload"

curl -N http://127.0.0.1:8000/aiohttp/responses \
  -H 'Content-Type: application/json' \
  -d "$payload"

性能比較

Ollama などのローカル server に向けて、同じ request body を各 endpoint へ 繰り返し送ると、end-to-end の所要時間を比較できます。OpenAI API に対して 大量に実行すると費用や rate limit の影響があるため、まずローカル環境で確認します。

順序による偏りを避けるため、3つの endpoint の実行順を入れ替えながら測ります。

/bin/bash <<'BASH'
payload='{"input":"Write one short sentence about SSE."}'
per_round=20
orders=(
  "openai-python httpx aiohttp"
  "openai-python aiohttp httpx"
  "httpx openai-python aiohttp"
  "httpx aiohttp openai-python"
  "aiohttp openai-python httpx"
  "aiohttp httpx openai-python"
)

for order in "${orders[@]}"; do
  echo "== ${order} =="
  for endpoint in $order; do
    i=0
    while [ "$i" -lt "$per_round" ]; do
      curl -sS -o /dev/null "http://127.0.0.1:8000/$endpoint/responses" \
        -w "${endpoint} %{time_total}\n" \
        -H "Content-Type: application/json" \
        -d "$payload" || exit 1
      i=$((i + 1))
    done
  done
done
BASH

結果を見るときは、平均値だけでなく失敗回数、最小値、最大値も確認します。 上流 model の速度、初回 load、CPU/GPU、同時実行数の影響を受けるため、 client 実装だけを比較したい場合は、実 LLM ではなく固定応答を返す OpenAI 互換 mock server を使ってください。

1万回など大きい回数で見る場合は、per_round を増やします。

per_round=1667

6通りの順序で実行するため、per_round=1667 では各 endpoint 約1万回になります。

2026-06-10 にローカル Ollama (llama3.2) へ warmup を各 endpoint 10回ずつ 投げたあと、各 endpoint 120回ずつ投げた結果は次のとおりです。

Endpoint 回数 失敗 平均秒 最小秒 最大秒
/openai-python/responses 120 0 0.491 0.372 0.850
/aiohttp/responses 120 0 0.491 0.304 0.797
/httpx/responses 120 0 0.500 0.319 0.768

固定順で簡単に見るだけなら、次のようにも実行できます。

payload='{"input":"Write one short sentence about SSE."}'
count=100

for endpoint in openai-python httpx aiohttp; do
  echo "== ${endpoint} =="
  i=0
  while [ "$i" -lt "$count" ]; do
    curl -sS -o /dev/null "http://127.0.0.1:8000/$endpoint/responses" \
      -H "Content-Type: application/json" \
      -d "$payload" || exit 1
    i=$((i + 1))
  done
done

Structured Outputs 例

curl -N http://127.0.0.1:8000/httpx/responses \
  -H 'Content-Type: application/json' \
  -d '{
    "input": "Return a one sentence answer.",
    "schema_name": "answer",
    "schema": {
      "type": "object",
      "properties": {
        "answer": { "type": "string" }
      },
      "required": ["answer"],
      "additionalProperties": false
    }
  }'

Ollama 確認

OpenAI 互換 API として動く Ollama server を起動してから、このアプリを起動します。

1つ目の terminal で Ollama server を起動し、そのまま起動しておきます。

ollama serve

2つ目の terminal で、使用する model を pull します。

ollama pull llama3.2

同じ terminal で Python アプリを起動します。

LLM_BASE_URL=http://localhost:11434/v1/ \
LLM_API_KEY=ollama \
LLM_MODEL=llama3.2 \
uv run uvicorn sse_proxy_python_example.app:app --reload

起動後、上記の curl -N コマンドを実行します。

OpenAI API 確認

LLM_BASE_URL=https://api.openai.com/v1/ \
LLM_API_KEY="$OPENAI_API_KEY" \
LLM_MODEL=gpt-5-mini \
uv run uvicorn sse_proxy_python_example.app:app --reload

起動後、Request 例の3つの endpoint を同じ payload で確認します。

テスト

通常のテストでは実際の LLM API を呼びません。

uv run pytest
uv run ruff check .
uv run pyright
markdownlint-cli2 README.md task.md .markdownlint-cli2.jsonc

独自 CA を使う社内プロキシ配下での起動

LLM API へ到達するために社内プロキシを経由する必要があり、そのプロキシが 社内独自の root CA を使う場合は、この手順で起動します。上流 HTTP client は 標準の proxy 環境変数と Python SSL 環境変数を使います。

このアプリは TLS 証明書検証を有効にしたまま動作します。Python 3.13 では、 Missing Authority Key Identifier などのエラーで一部の社内プロキシ証明書が 拒否されることがあります。そのため、このアプリは Python 3.13 の厳格な X.509 検証 flag だけを外し、証明書検証自体は維持します。

proxy の環境変数を設定します。

export HTTPS_PROXY=http://proxy.example.com:8080
export HTTP_PROXY=http://proxy.example.com:8080
export NO_PROXY=localhost,127.0.0.1

社内 root CA を PEM file として用意し、Python/httpx が読む SSL_CERT_FILE に指定します。

SSL_CERT_FILE=/path/to/corporate-ca.pem \
LLM_BASE_URL=https://api.openai.com/v1/ \
LLM_API_KEY="$OPENAI_API_KEY" \
LLM_MODEL=gpt-5-mini \
uv run uvicorn sse_proxy_python_example.app:app --reload

証明書が hashed certificate directory として提供される環境では、 SSL_CERT_DIR を使います。

Docker では CA file を container に mount し、同じ環境変数を渡します。

docker run --rm -p 8000:8000 \
  -e LLM_BASE_URL=https://api.openai.com/v1/ \
  -e LLM_API_KEY="$OPENAI_API_KEY" \
  -e LLM_MODEL=gpt-5-mini \
  -e SSL_CERT_FILE=/etc/ssl/certs/corporate-ca.pem \
  -e HTTPS_PROXY \
  -e HTTP_PROXY \
  -e NO_PROXY \
  -v /path/to/corporate-ca.pem:/etc/ssl/certs/corporate-ca.pem:ro \
  sse-proxy-python-example

起動ログには SSL_CERT_FILE のパスが表示されます。event: error が返る場合は server log を確認してください。上流接続の例外は server log に出力し、client 向けの SSE error body には API key や proxy 認証情報を出しません。