Version: 1.0.0 Last Updated: 2026-05-14 Status: Active

bast-ai-openai¶

Version: 0.1.6 Module: bast_ai_openai Path: src/python/openai

Overview¶

OpenAI API integration library providing chat completions, embeddings, and assistant management for Bast AI services.

Features¶

Chat Completions¶

GPT-4o, GPT-4, GPT-3.5-turbo support
Streaming responses
Function/tool calling
JSON mode responses

Embeddings¶

text-embedding-ada-002
text-embedding-3-small
text-embedding-3-large
Batch embedding support

Assistants API¶

Assistant creation and management
Thread management
File attachments
Code interpreter

Additional Integrations¶

Perplexity API support
Response validation
Token counting
Cost estimation

Dependencies¶

Internal Dependencies¶

Package	Purpose
bast-ai-config	Configuration management

External Dependencies¶

Package	Version	Purpose
openai	^1.97	OpenAI Python SDK

Usage Examples¶

Chat Completions¶

from bast_ai_openai import OpenAIService

openai_svc = OpenAIService(catname="mycat")

# Simple completion
response = openai_svc.chat_completion(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is machine learning?"}
    ],
    model="gpt-4o",
    temperature=0.7
)

print(response.content)

Streaming Responses¶

# Streaming completion
for chunk in openai_svc.chat_completion_stream(
    messages=[{"role": "user", "content": "Tell me a story"}],
    model="gpt-4o"
):
    print(chunk.content, end="", flush=True)

Tool/Function Calling¶

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }
    }
]

response = openai_svc.chat_completion(
    messages=[{"role": "user", "content": "What's the weather in NYC?"}],
    tools=tools,
    model="gpt-4o"
)

if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"Call: {tool_call.function.name}")
        print(f"Args: {tool_call.function.arguments}")

Embeddings¶

# Single embedding
embedding = openai_svc.create_embedding(
    text="machine learning algorithms",
    model="text-embedding-3-small"
)

# Batch embeddings
texts = ["text 1", "text 2", "text 3"]
embeddings = openai_svc.create_embeddings(
    texts=texts,
    model="text-embedding-3-small"
)

JSON Mode¶

response = openai_svc.chat_completion(
    messages=[
        {"role": "system", "content": "Return JSON with name and age."},
        {"role": "user", "content": "John is 30 years old"}
    ],
    model="gpt-4o",
    response_format={"type": "json_object"}
)

import json
data = json.loads(response.content)

Assistants API¶

# Create assistant
assistant = openai_svc.create_assistant(
    name="Data Analyst",
    instructions="You analyze data and provide insights.",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}]
)

# Create thread and run
thread = openai_svc.create_thread()
openai_svc.add_message(thread.id, "Analyze this dataset: [1,2,3,4,5]")
run = openai_svc.run_assistant(thread.id, assistant.id)

# Get response
messages = openai_svc.get_messages(thread.id)

Configuration¶

# resources/config/{catname}.yaml
openai:
  api_key: ${OPENAI_API_KEY}
  organization: ${OPENAI_ORG_ID:}
  model: gpt-4o
  embedding_model: text-embedding-3-small
  temperature: 0.7
  max_tokens: 1000
  timeout: 60

  # Rate limiting
  rate_limit:
    requests_per_minute: 500
    tokens_per_minute: 150000

Testing¶

Unit Tests (3 files)¶

test_svc_api.py - Service API tests
test_perplexity.py - Perplexity integration tests
test_ideal.py - Ideal wrapper tests

Running Tests¶

pants test src/python/openai:: -- -s --no-header

Issues & Improvements¶

Current Issues¶

No automatic retries: API failures not automatically retried
Limited cost tracking: Token usage not tracked
No response caching: Identical requests hit API

Suggested Improvements¶

Retry logic: Add exponential backoff for API errors
Cost tracking: Track and report token usage/costs
Response caching: Cache embeddings and repeated queries
Rate limiting: Built-in rate limit handling
Fallback models: Automatic fallback to cheaper models
Prompt templates: Add templating support

Best Practices¶

Error Handling¶

from openai import RateLimitError, APIError

try:
    response = openai_svc.chat_completion(messages=messages)
except RateLimitError:
    logger.warning("Rate limited, waiting...")
    time.sleep(60)
except APIError as e:
    logger.error(f"API error: {e}")

Token Management¶

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

# Check before sending
if count_tokens(prompt) > 128000:
    # Truncate or chunk
    pass

Embedding Caching¶

import hashlib

def get_cached_embedding(text: str) -> list[float]:
    cache_key = hashlib.md5(text.encode()).hexdigest()
    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)

    embedding = openai_svc.create_embedding(text)
    redis.set(cache_key, json.dumps(embedding), ex=86400)
    return embedding

Changelog¶

v0.1.6 (Current)¶

Updated openai SDK to ^1.97
Added GPT-4o support
Improved streaming

v0.1.5¶

Added Perplexity API support
Better error handling

v0.1.4¶

Added Assistants API support
Batch embedding support