openai-async-utils

Lightweight async helpers for OpenAI chat completions:

• call_openai_api: one-shot request with retries.
• stream_openai_api: yields response chunks with retries.

Install

pip install openai-async-utils
export OPENAI_API_KEY="sk-…"

You can also pass api_key="sk-…" directly to functions.

Quick Example

import asyncio
from openai_utils import call_openai_api, stream_openai_api

async def main():
    sys = "You are a helpful assistant."
    prev = [{"role": "user", "content": "Hello!"}]

    # Full response
    text, usage = await call_openai_api(sys, prev)
    print(text, usage)

    # Streaming response
    async for chunk in stream_openai_api(sys, prev):
        print(chunk, end="")

asyncio.run(main())

API

• call_openai_api(system_prompt, previous_prompts, api_key=None, validation_schema=None, model="gpt-3.5-turbo", max_attempts=3) Returns (response_str, {"prompt_tokens":…, "completion_tokens":…, "total_tokens":…}).

• stream_openai_api(system_prompt, previous_prompts, api_key=None, model="gpt-3.5-turbo", max_attempts=3) Async generator yielding text chunks.

License

MIT