Spaces:
Paused
Paused
File size: 2,346 Bytes
bb6e650 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """
Shared OpenAI HTTP client — single implementation of the chat-completions call.
Replaces duplicated urllib boilerplate in gpt_reasoning, relevance,
mission_parser, and threat_chat.
"""
import json
import logging
import os
import urllib.request
import urllib.error
from typing import Dict, Optional, Tuple
logger = logging.getLogger(__name__)
_API_URL = "https://api.openai.com/v1/chat/completions"
class OpenAIAPIError(Exception):
"""Raised when the OpenAI API call fails (HTTP or network error)."""
def __init__(self, message: str, status_code: Optional[int] = None):
self.status_code = status_code
super().__init__(message)
def get_api_key() -> Optional[str]:
"""Return the OpenAI API key from the environment, or None."""
return os.environ.get("OPENAI_API_KEY")
def chat_completion(payload: Dict, *, timeout: int = 30) -> Dict:
"""Send a chat-completion request and return the parsed JSON response.
Args:
payload: Full request body (model, messages, etc.).
timeout: HTTP timeout in seconds.
Returns:
Parsed response dict.
Raises:
OpenAIAPIError: On HTTP or network failure.
"""
api_key = get_api_key()
if not api_key:
raise OpenAIAPIError("OPENAI_API_KEY not set")
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
}
try:
req = urllib.request.Request(
_API_URL,
data=json.dumps(payload).encode("utf-8"),
headers=headers,
method="POST",
)
with urllib.request.urlopen(req, timeout=timeout) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as e:
raise OpenAIAPIError(
f"HTTP {e.code}: {e.reason}", status_code=e.code
) from e
except urllib.error.URLError as e:
raise OpenAIAPIError(f"URL error: {e.reason}") from e
def extract_content(resp_data: Dict) -> Tuple[Optional[str], Optional[str]]:
"""Safely extract content and refusal from a chat-completion response.
Returns:
(content, refusal) — either may be None.
"""
choice = resp_data.get("choices", [{}])[0]
message = choice.get("message", {})
return message.get("content"), message.get("refusal")
|