import os
import sys
from openai import OpenAI, APIConnectionError, APIStatusError
mode = sys.argv[1] if len(sys.argv) == 2 else "sync"
if mode not in ("sync", "stream"):
raise SystemExit("Usage: python legacy-completions.py [sync|stream]")
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
try:
result = client.completions.create(model=os.environ["API_MODEL"],
prompt="Complete this sentence: A good API guide should", stream=mode == "stream")
text, finished = "", False
if mode == "sync":
if len(result.choices) != 1 or result.choices[0].index != 0:
raise ValueError("Expected one completion.")
choice = result.choices[0]
if choice.finish_reason != "stop" or not isinstance(choice.text, str):
raise ValueError("Incomplete completion.")
text, finished = choice.text, True
else:
with result:
for chunk in result:
if not chunk.choices:
continue # Usage-only chunk, not a completion marker.
if finished or len(chunk.choices) != 1 or chunk.choices[0].index != 0:
raise ValueError("Unexpected completion chunk.")
choice = chunk.choices[0]
if not isinstance(choice.text, str) or choice.finish_reason not in (None, "stop"):
raise ValueError("Incomplete completion.")
text += choice.text
if len(text) > 1_000_000:
raise ValueError("Example output limit exceeded.")
finished = choice.finish_reason == "stop"
if not finished or not text.strip():
raise ValueError("Missing completed text.")
print(text)
except APIStatusError as error:
raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
except APIConnectionError:
raise SystemExit("Connection failed. No automatic retry.")
except Exception:
raise SystemExit("Completion did not produce valid complete text.")