Skip to content

Python (OpenAI SDK)

Tokens-এর সাথে official openai Python package ব্যবহার করুন: client setup, sync ও async call, streaming, tool call, timeout, retry আর error handling।

যেসব tool-এ কাজ করেOpenAI Python SDK

সর্বশেষ আপডেট 11 অক্টোবর 2026

Markdown-এ দেখুন
এই পাতায়

Official openai Python package-টা Tokens-এর সাথে কোনো বদল ছাড়াই চলে। শুধু এটাকে https://tokens.bd/v1-এ দেখিয়ে দিন, আপনার Tokens key দিন, আর Tokens catalog-এর model ID ব্যবহার করুন। বাকি সব (streaming, tool call, async) আপনার চেনা SDK-ই থাকবে।

OpenAI Python SDK install ও configure করুন#

bash
pip install --upgrade openai
export TOKENS_API_KEY="tok_live_your_key"
hello.py
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://tokens.bd/v1",
    api_key=os.environ["TOKENS_API_KEY"],
)

completion = client.chat.completions.create(
    model="deepseek/deepseek-v4.1-flash",
    messages=[
        {"role": "system", "content": "Answer in one short paragraph."},
        {"role": "user", "content": "When should I use a dataclass instead of a dict?"},
    ],
    max_tokens=400,
)

print(completion.choices[0].message.content)
print(completion.usage)

Model ID হুবহু /models পেজ বা client.models.list() থেকে নিন। ID-র ধরন provider/model, আর বানান ভুল হলে 404 model_not_found আসবে।

বিকল্প: OPENAI_BASE_URL আর OPENAI_API_KEY#

আপনি নিজে না দিলে SDK এই দুটো environment variable থেকে OPENAI_BASE_URL আর OPENAI_API_KEY পড়ে নেয়। তাই পুরোনো code এক লাইনও না বদলে Tokens-এ চলে আসতে পারে:

bash
export OPENAI_BASE_URL="https://tokens.bd/v1"
export OPENAI_API_KEY="$TOKENS_API_KEY"
python
from openai import OpenAI

client = OpenAI()  # picks up OPENAI_BASE_URL and OPENAI_API_KEY

সুবিধা আছে, কিন্তু এই variable-গুলো পুরো shell জুড়ে চলে। ওই shell-এ যে tool-ই এগুলো পড়ে (Aider, কিছু agent, অন্য script), সবার traffic-ই Tokens-এ যাবে। এটা শুধু একটা জায়গায় আটকে রাখতে চাইলে প্রথম উদাহরণের মতো base_url আর api_key সরাসরি দিন।

Response stream করুন#

python
stream = client.chat.completions.create(
    model="deepseek/deepseek-v4.1-flash",
    messages=[{"role": "user", "content": "Write a haiku about merge conflicts."}],
    stream=True,
    stream_options={"include_usage": True},
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
    if chunk.usage:
        print(f"\n\n{chunk.usage.prompt_tokens} in, {chunk.usage.completion_tokens} out")

if chunk.choices check-টা বাদ দেবেন না। include_usage চালু থাকলে শেষ chunk-এর choices list ফাঁকা থাকে, শুধু usage থাকে। আবার include_usage না দিলে stream-এ কোনো token count-ই আসে না। আরও জানতে Streaming দেখুন।

Async client ব্যবহার করুন#

AsyncOpenAI একই argument নেয়। FastAPI, aiohttp বা যেখানেই event loop চলছে, সেখানে এটা ব্যবহার করুন।

python
import asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://tokens.bd/v1",
    api_key=os.environ["TOKENS_API_KEY"],
)

async def main() -> None:
    stream = await client.chat.completions.create(
        model="deepseek/deepseek-v4.1-flash",
        messages=[{"role": "user", "content": "Name three uses for asyncio.Semaphore."}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

asyncio.run(main())

asyncio.gather দিয়ে অনেক request একসাথে ছাড়লে semaphore দিয়ে সংখ্যাটা আটকে দিন। Tokens প্রতি অ্যাকাউন্টে একসাথে চলা request-এর সংখ্যা সীমিত রাখে (সীমাটা আসে আপনার plan থেকে), আর সেটা পেরোলে 429 concurrency_limit আসে। দেখুন Rate limits।

Tool call#

যে model tool calling সাপোর্ট করে, শুধু সেটাতেই এটা চলে। ভরসা করার আগে /models-এ model-এর পেজটা দেখে নিন।

python
import json

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Current weather for a city",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
}]

messages = [{"role": "user", "content": "Is it raining in Dhaka?"}]
first = client.chat.completions.create(
    model="deepseek/deepseek-v4.1-flash", messages=messages, tools=tools
)
msg = first.choices[0].message

if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = {"city": args["city"], "condition": "light rain", "temp_c": 29}  # your real lookup here
        messages.append({"role": "tool", "tool_call_id": call.id, "content": json.dumps(result)})
    final = client.chat.completions.create(
        model="deepseek/deepseek-v4.1-flash", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)

Request আর response-এর পুরো গঠন আছে Tool calling পেজে।

Timeout ও retry ঠিক করুন#

কিছু ব্যর্থতায় (connection error, 408, 409, 429 আর 5xx) SDK নিজে থেকেই backoff দিয়ে default-এ দুইবার retry করে। Default timeout 10 মিনিট। দুটোই client-এ বা আলাদা করে প্রতি call-এ বদলানো যায়:

python
import os
import httpx
from openai import OpenAI

client = OpenAI(
    base_url="https://tokens.bd/v1",
    api_key=os.environ["TOKENS_API_KEY"],
    timeout=httpx.Timeout(120.0, connect=10.0),
    max_retries=3,
)

# Override for one call
client.with_options(timeout=30.0, max_retries=0).chat.completions.create(...)

Tokens-এর ক্ষেত্রে কয়েকটা কথা মনে রাখুন:

  • Server error, 429, timeout বা connection error হলে gateway আপনাকে উত্তর দেওয়ার আগেই অন্য upstream source-এ চেষ্টা করে দেখে। তাই আপনি যদি 5xx পান, তার মানে সেই failover-ও কাজ করেনি। SDK-র দু-একবার retry-ই যথেষ্ট।
  • লম্বা generation নিয়ে চিন্তা নেই। Response header আসার জন্য gateway 600 সেকেন্ড পর্যন্ত অপেক্ষা করে। লম্বা output হলে stream করুন, যাতে connection অকারণে ফাঁকা পড়ে না থাকে।
  • 429 window_exhausted হলে window reset না হওয়া পর্যন্ত retry করে লাভ নেই। Retry-After বলে দেয় কত সেকেন্ড বাকি, আর সেটা কয়েক ঘণ্টাও হতে পারে।

Tokens-এর error code সামলান#

Error-এর গঠন OpenAI-র মতোই, তাই openai তার চেনা exception class-ই raise করে। Tokens-এর জন্য কাজের জিনিস হলো error.code।

python
import openai

try:
    client.chat.completions.create(
        model="deepseek/deepseek-v4.1-flash",
        messages=[{"role": "user", "content": "hi"}],
    )
except openai.APIStatusError as e:
    request_id = e.response.headers.get("x-tokens-request-id")
    print(e.status_code, e.code, e.message, request_id)
    if e.code == "insufficient_credits":
        print("Top up at https://tokens.bd/dashboard/billing")
except openai.APIConnectionError as e:
    print("Network problem:", e)

প্রতিটি ব্যর্থতার সাথে x-tokens-request-id log করে রাখুন। ওই ID ধরে Support আপনার request খুঁজে বের করতে পারে। সাধারণত যে code-গুলো সামনে আসে: invalid_api_key (401), model_not_allowed_on_key ও tier_permission_denied (403), insufficient_credits (402), আর rate_limited, concurrency_limit ও window_exhausted (429)। সবগুলোর তালিকা Errors পেজে, আর প্রতিটার সমাধান Troubleshooting পেজে।

Warning

Key রাখুন environment variable-এ বা secrets manager-এ। কোনো commit করা file বা share করা notebook-এ key চলে গেলে /dashboard/keys থেকে সেটা rotate করুন। পুরোনো secret সাথে সাথে অকেজো হয়ে যায়।

এই পাতাটা কি কাজে লেগেছে?

এখনো আটকে আছেন? Support ticket খুলুন

আপনার agent set up করতে সাহায্য লাগবে?

Connection tester দিয়ে সংযোগ পরীক্ষা করে নিন, অথবা একটা API key তৈরি করুন।