# ChatGPTのサブスクを使った版
# oauth_codex は認証にしかつかっていない
import json
from typing import List, Dict, Generator
import httpx
from oauth_codex import Client
_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"
SYSTEM_PROMPT = {
"role": "system",
"content": "あなたは親切なAIチャットボットです。日本語で回答してください。"
}
# モデル設定
# MODEL : "gpt-5.5" / "gpt-5.4" / "gpt-5.4-mini"
# REASONING : "low"(高速・軽量) / "medium"(バランス) / "high"(深い推論) / "xhigh"(最大、デフォルト)
MODEL = "gpt-5.4"
REASONING = {"effort": "medium"}
class _Delta:
def __init__(self, content: str) -> None:
self.content = content
class _Choice:
def __init__(self, content: str) -> None:
self.delta = _Delta(content)
class _Chunk:
def __init__(self, content: str) -> None:
self.choices = [_Choice(content)]
# APIキーは不要。初回にブラウザでChatGPTへログイン
client = Client()
client.authenticate()
def chat_completion_stream(messages: List[Dict[str, str]]) -> Generator:
headers = dict(client._engine._client.auth.get_headers())
headers["Content-Type"] = "application/json"
payload = {"model": MODEL, "stream": True, "store": False,
"input": messages, "reasoning": REASONING}
with httpx.stream("POST", _CODEX_URL, headers=headers, json=payload, timeout=60) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
return
obj = json.loads(data)
if obj.get("type") == "response.output_text.delta":
yield _Chunk(obj.get("delta", ""))
response = chat_completion_stream([SYSTEM_PROMPT, {"role": "user", "content": "カレーライスの作り方を教えてください。"}])
print("".join([chunk.choices[0].delta.content or "" for chunk in response]))