Stream in Python

# pip install openai
import os
import sys
 
from openai import OpenAI
 
client = OpenAI(base_url="https://app.syntic.ai/v1", api_key=os.environ["SYNTIC_API_KEY"])
 
stream = client.chat.completions.create(
    model="Amara S1.2",
    stream=True,
    messages=[{"role": "user", "content": "Explain how a bicycle derailleur works in 5 sentences."}],
)
 
usage = None
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        sys.stdout.write(chunk.choices[0].delta.content)
        sys.stdout.flush()
    if getattr(chunk, "usage", None):
        usage = chunk.usage
print()
print("usage:", usage)

# pip install requests
import json
import os
import sys
 
import requests
 
resp = requests.post(
    "https://app.syntic.ai/v1/chat/completions",
    headers={"Authorization": f"Bearer {os.environ['SYNTIC_API_KEY']}"},
    json={
        "model": "Amara S1.2",
        "messages": [{"role": "user", "content": "Explain how a bicycle derailleur works in 5 sentences."}],
    },
    stream=True,
    timeout=(10, 300),          # connect timeout, read timeout between bytes
)
if resp.status_code != 200:
    raise SystemExit(f"{resp.status_code}: {resp.text}")
print("request id:", resp.headers.get("x-syntic-request-id"), file=sys.stderr)
 
finished = False
for line in resp.iter_lines(decode_unicode=True):
    if not line or not line.startswith("data: "):
        continue
    payload = line[len("data: "):]
    if payload == "[DONE]":
        finished = True
        break
    event = json.loads(payload)
    if "error" in event:
        raise RuntimeError(f"stream failed: {event.get('detail')}")
    choices = event.get("choices") or []
    if choices and choices[0].get("delta", {}).get("content"):
        sys.stdout.write(choices[0]["delta"]["content"])
        sys.stdout.flush()
    if event.get("usage"):
        print("\nusage:", event["usage"], file=sys.stderr)
if not finished:
    print("\n[stream ended early; reply is incomplete]", file=sys.stderr)

# pip install httpx
import asyncio
import json
import os
 
import httpx
 
 
async def main():
    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, read=300.0)) as http:
        async with http.stream(
            "POST",
            "https://app.syntic.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['SYNTIC_API_KEY']}"},
            json={"model": "Amara Quick", "messages": [{"role": "user", "content": "Name five rivers in Africa."}]},
        ) as resp:
            if resp.status_code != 200:
                print(resp.status_code, await resp.aread())
                return
            async for line in resp.aiter_lines():
                if line.startswith("data: ") and line != "data: [DONE]":
                    event = json.loads(line[6:])
                    piece = (event.get("choices") or [{}])[0].get("delta", {}).get("content")
                    if piece:
                        print(piece, end="", flush=True)
    print()
 
asyncio.run(main())