راهنمای کامل OpenAI Python SDK با API پیکو هوش

استریم

stream = client.chat.completions.create(
    model="claude-sonnet-5-5",
    messages=[{"role": "user", "content": "یک مقدمه‌ی کوتاه درباره‌ی Docker بنویس"}],
    stream=True,
    stream_options={"include_usage": True},  # برای گرفتن usage در رویداد آخر
)
for event in stream:
    if event.choices and event.choices[0].delta.content:
        print(event.choices[0].delta.content, end="", flush=True)
print()

با stream_options={"include_usage": True} (پارامتر استاندارد API چت) رویداد آخر شامل مصرف است؛ ممکن است رویدادی با choices خالی و فقط usage ببینید، به همین دلیل شرط event.choices لازم است. ساختار خام در راهنمای SSE.

Async

import asyncio
from openai import AsyncOpenAI

aclient = AsyncOpenAI(base_url="https://api.picoai.ir/v1", api_key=os.environ["PICO_API_KEY"])

async def ask(q: str) -> str:
    r = await aclient.chat.completions.create(
        model="claude-haiku-5-5",
        messages=[{"role": "user", "content": q}],
        max_tokens=200,
    )
    return r.choices[0].message.content

async def main():
    answers = await asyncio.gather(*(ask(q) for q in ["DNS؟", "CDN؟", "TLS؟"]))
    print(answers)

asyncio.run(main())

برای کنترل سقف درخواست در دقیقه، تعداد هم‌زمانی را با asyncio.Semaphore محدود کنید.

مدیریت خطا

import openai

try:
    r = client.chat.completions.create(model="gpt-6-sol",
        messages=[{"role": "user", "content": "سلام"}])
except openai.AuthenticationError:          # 401
    print("کلید نامعتبر است")
except openai.PermissionDeniedError as e:   # 403: مثلاً سقف ماهانه‌ی کلید
    print("دسترسی/سقف:", e.message)
except openai.RateLimitError as e:          # 429
    print("Retry-After:", e.response.headers.get("retry-after"))
except openai.APIStatusError as e:          # سایر وضعیت‌ها (402، 404، 5xx)
    print(e.status_code, e.response.json())
except openai.APIConnectionError:
    print("مشکل شبکه")

نکته: خطای 402 (کمبود موجودی) کلاس اختصاصی ندارد و در APIStatusError می‌آید. بدنه‌ی خطا شامل error.code (مثل insufficient_balance) و یک message_fa فارسی است. جدول کامل کدها و راهبرد تلاش مجدد در کدهای خطا و Retry.

Retry و Timeout

SDK به‌صورت پیش‌فرض ۲ بار برای خطاهای اتصال و وضعیت‌های 408، 409، 429 و 5xx تلاش مجدد می‌کند (timeout پیش‌فرض خود SDK ۱۰ دقیقه است؛ ما در نمونه ۶۰ ثانیه گذاشته‌ایم). تنظیم برای یک درخواست:

r = client.with_options(max_retries=5, timeout=120.0).chat.completions.create(
    model="gpt-6-sol",
    messages=[{"role": "user", "content": "یک گزارش بلند بنویس"}],
)

از آنجا که درخواست ناموفق هزینه‌ای ندارد، تلاش مجدد از نظر مالی بی‌خطر است. اما اگر استریم وسط راه قطع شود، متن دریافت‌شده را نگه دارید و خودتان تصمیم بگیرید.

دسترسی به هدرها و ID درخواست

raw = client.chat.completions.with_raw_response.create(
    model="gpt-6-luna", messages=[{"role": "user", "content": "سلام"}])
print(raw.headers.get("content-type"))
completion = raw.parse()

تصویر و گفتار

مسیرهای /v1/images/generations و /v1/audio/speech هم با همان کلاینت قابل استفاده‌اند (client.images.generate و client.audio.speech.create) با مدل‌های مربوطه از کاتالوگ. فرمت دقیق پارامترها را در مستندات ببینید. اندپوینت embeddings موجود نیست، پس client.embeddings.create خطا می‌دهد.

نکات تولید

1. یک کلاینت را در کل برنامه به اشتراک بگذارید؛ ساخت مجدد در هر درخواست اتصال را هدر می‌دهد.
2. max_tokens را همیشه بگذارید.
3. مصرف usage را لاگ کنید و با فرمول هزینه مقایسه کنید.
4. برای هر محیط کلید جدا داشته باشید.
5. پیام‌های بلند را کوتاه یا خلاصه کنید تا هزینه‌ی ورودی کم شود.

عیب‌یابی

الگوی کلاس کلاینت برای پروژه‌ها

در پروژه‌های واقعی بهتر است همه‌ی تنظیمات یک‌جا باشد:

from functools import lru_cache
from openai import OpenAI
import os

@lru_cache(maxsize=1)
def get_client() -> OpenAI:
    return OpenAI(
        base_url=os.getenv("PICO_BASE_URL", "https://api.picoai.ir/v1"),
        api_key=os.environ["PICO_API_KEY"],
        timeout=60.0,
        max_retries=3,
    )

def summarize(text: str, model: str = "claude-haiku-5-5") -> str:
    r = get_client().chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "متن را در سه بولت فارسی خلاصه کن."},
            {"role": "user", "content": text[:6000]},
        ],
        max_tokens=300,
        temperature=0.2,
    )
    return r.choices[0].message.content

کوتاه‌کردن ورودی (text[:6000]) و max_tokens دو محافظ ساده‌ی هزینه‌اند. برای تست واحد، تابع get_client را mock کنید تا آزمون‌ها درخواست واقعی نفرستند.

خروجی JSON

برای دریافت JSON قابل پارس، در پرامپت قالب را دقیق بنویسید و پاسخ را با json.loads در try/except پارس کنید. اگر مدل از response_format={"type": "json_object"} پشتیبانی کند می‌توانید آن را بفرستید؛ این پارامتر به مدل منتقل می‌شود و نتیجه به پشتیبانی مدل بستگی دارد، پس آن را با مدل انتخابی خود آزمایش کنید و پارس را همیشه محافظت‌شده نگه دارید.

ورودی تصویر (بینایی)

مدل‌هایی که بینایی دارند (در کاتالوگ مشخص شده) تصویر را در content به‌صورت آرایه می‌پذیرند:

r = client.chat.completions.create(
    model="gpt-6-sol",
    messages=[{"role": "user", "content": [
        {"type": "text", "text": "این نمودار چه می‌گوید؟"},
        {"type": "image_url", "image_url": {"url": "https://example.com/chart.png"}},
    ]}],
)

تصویر به توکن ورودی تبدیل می‌شود و هزینه دارد؛ ابعاد را در حد نیاز نگه دارید.

سوالات متداول

آیا باید نسخه‌ی خاصی از openai نصب کنم؟

نسخه‌ی ۱.x به بعد (با کلاس OpenAI()). README رسمی فعلی Python 3.10 یا بالاتر را لازم می‌داند. نمونه‌ها برای همین سبک‌اند. توجه: OpenAI برای مدل‌های خودش Responses API را API اصلی معرفی می‌کند و Chat Completions را «برای همیشه پشتیبانی‌شده» می‌داند؛ پیکو هوش مسیر /v1/chat/completions را ارائه می‌دهد، پس از client.chat.completions استفاده کنید نه client.responses.

آیا structured output (response_format) کار می‌کند؟

به مدل بستگی دارد؛ پارامتر را به مدل ارسال می‌کنیم اما پشتیبانی نهایی با مدل زیرین است. با مدل موردنظر تست کنید.

آیا می‌توانم مدل‌های Claude را با همین کلاینت صدا بزنم؟

بله، model="claude-sonnet-5-5".

مرجع؟

OpenAI Python در یکپارچه‌سازی‌ها، pricing، مستندات.