Panduan Integrasi Python SDK (openai)

1. Instalasi Paket & Variabel Lingkungan

pip install --upgrade openai

# Set environment variable di shell (.env)
export TOKENFAJRI_API_KEY="sk-tok-xxxxxxxxxxxxxxxxxxxxxxxx"
export TOKENFAJRI_BASE_URL="https://token.fajriag.ai.id/v1"

2. Permintaan Sinkron Standar (Chat Completions)

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["TOKENFAJRI_API_KEY"],
    base_url=os.environ.get("TOKENFAJRI_BASE_URL", "https://token.fajriag.ai.id/v1"),
)

response = client.chat.completions.create(
    model="deepseek-v4.1-flash-1017",
    messages=[
        {"role": "system", "content": "Jawab dengan bahasa Indonesia yang ringkas dan teknis."},
        {"role": "user", "content": "Berikan contoh query PostgreSQL dengan indeks komposit."},
    ],
    temperature=0.2,
)

print(response.choices[0].message.content)
print("Total token terpakai:", response.usage.total_tokens)

3. Streaming Real-Time (Server-Sent Events / SSE)

Untuk respons interaktif dengan Time-To-First-Token (TTFT) rendah, aktifkan stream=True:

stream = client.chat.completions.create(
    model="deepseek-v4.1-flash-1017",
    messages=[{"role": "user", "content": "Tulis fungsi quicksort di Python beserta kompleksitasnya."}],
    stream=True,
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

4. Konkurensi Tinggi dengan AsyncOpenAI (FastAPI / Asyncio)

Pada server backend produksi berbasis FastAPI atau worker asinkron, gunakan AsyncOpenAI agar thread event loop tidak terblokir:

import asyncio
import os
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key=os.environ["TOKENFAJRI_API_KEY"],
    base_url="https://token.fajriag.ai.id/v1",
    max_retries=2,
    timeout=45.0,
)

async def ask_model(prompt: str) -> str:
    res = await async_client.chat.completions.create(
        model="deepseek-v4.1-flash-1017",
        messages=[{"role": "user", "content": prompt}],
    )
    return res.choices[0].message.content or ""

if __name__ == "__main__":
    print(asyncio.run(ask_model("Apa keuntungan connection pooling di PostgreSQL?")))