import json import os import tempfile import time from typing import Callable, List, Optional import openai def _summarize_batch(batch_status) -> dict: """Pull the user-relevant bits out of an OpenAI Batch object.""" counts = getattr(batch_status, "request_counts", None) usage = getattr(batch_status, "usage", None) or {} completed = getattr(counts, "completed", 0) if counts else 0 failed = getattr(counts, "failed", 0) if counts else 0 total_requests = getattr(counts, "total", 0) if counts else 0 return { "batch_id": getattr(batch_status, "id", None), "status": getattr(batch_status, "status", None), "created_at": getattr(batch_status, "created_at", None), "in_progress_at": getattr(batch_status, "in_progress_at", None), "completed_at": getattr(batch_status, "completed_at", None), "completed": completed, "failed": failed, "total_requests": total_requests, "usage": dict(usage) if isinstance(usage, dict) else {}, } class GPT: def __init__(self, model_name="o3-mini", api_key=None): self.model_name = model_name self.api_key = api_key or os.getenv("OPENAI_API_KEY") if not self.api_key: raise ValueError("API key not provided and OPENAI_API_KEY environment variable not set.") # Per-instance client so concurrent users with different keys never share auth. self.client = openai.OpenAI(api_key=self.api_key) def predict( self, prompts: List[str], status_callback: Optional[Callable[[dict], None]] = None, ) -> List[str]: # Use a per-call temp file so concurrent requests can't clobber each other's input. with tempfile.NamedTemporaryFile( mode="w", suffix=".jsonl", delete=False, encoding="utf-8" ) as tmp: jsonl_path = tmp.name try: self._generate_batch_jsonl(prompts, jsonl_path) # Upload the JSONL file for batch processing with open(jsonl_path, "rb") as f: file_resp = self.client.files.create(file=f, purpose="batch") file_id = file_resp.id finally: try: os.remove(jsonl_path) except OSError: pass # Create a batch job batch = self.client.batches.create( input_file_id=file_id, endpoint="/v1/chat/completions", completion_window="24h" ) batch_id = batch.id # Wait for completion while True: batch_status = self.client.batches.retrieve(batch_id) print(batch_status) if status_callback is not None: try: status_callback(_summarize_batch(batch_status)) except Exception as cb_err: print(f"status_callback error (ignored): {cb_err}") if batch_status.status == "completed": break elif batch_status.status in {"failed", "expired", "cancelled"}: raise RuntimeError(f"Batch job failed with status: {batch_status.status}") time.sleep(10) # Handle errors if any if batch_status.error_file_id: error_stream = self.client.files.content(batch_status.error_file_id) error_log = error_stream.read().decode("utf-8") raise RuntimeError(f"Batch job failed. Error log:\n{error_log}") # Retrieve output output_file_id = batch_status.output_file_id file_stream = self.client.files.content(output_file_id) content = file_stream.read().decode("utf-8") # Parse responses responses_by_id = {} for line in content.splitlines(): obj = json.loads(line) idx = int(obj["custom_id"]) text = obj["response"]["body"]["choices"][0]["message"]["content"] responses_by_id[idx] = text return [responses_by_id.get(i, "No response") for i in range(len(prompts))] def _generate_batch_jsonl(self, prompts: List[str], output_path: str): with open(output_path, "w", encoding="utf-8") as f: for idx, prompt in enumerate(prompts): entry = { "custom_id": str(idx), "method": "POST", "url": "/v1/chat/completions", "body": { "model": self.model_name, "messages": [ {"role": "user", "content": prompt} ] } } f.write(json.dumps(entry) + "\n")