Spaces:
Sleeping
Sleeping
File size: 3,216 Bytes
2948ced |
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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
"""Lightweight OpenAI GPT-5 client for orchestration steps (4–9 prompts, etc.)."""
from __future__ import annotations
import base64
import json
import logging
import os
from pathlib import Path
from typing import Optional
from openai import OpenAI
from constants import (
ROOT,
DEFAULT_SETTINGS,
LLM_SETTING_KEYS as SETTING_KEYS,
DEFAULT_MODEL,
DEFAULT_REASONING,
)
def load_settings(path: Path | None) -> dict:
"""Load settings.json (or a provided path) and keep only recognized keys."""
path = path or DEFAULT_SETTINGS
if not path.exists():
return {}
data = json.loads(path.read_text(encoding="utf-8"))
return {k: v for k, v in data.items() if k in SETTING_KEYS and v}
def resolve_api_key(settings: dict) -> str:
"""Resolve OPENAI_API_KEY preferring env over settings; exit if missing."""
if os.environ.get("OPENAI_API_KEY"):
return os.environ["OPENAI_API_KEY"]
if settings.get("OPENAI_API_KEY"):
return settings["OPENAI_API_KEY"]
raise SystemExit("OPENAI_API_KEY is not set (env or settings.json)")
def resolve_model(settings: dict, cli_model: Optional[str]) -> str:
"""Pick the model from CLI override, env, settings, or fallback default."""
return cli_model or os.environ.get("OPENAI_MODEL") or settings.get("OPENAI_MODEL") or DEFAULT_MODEL
def resolve_reasoning(settings: dict, cli_reasoning: Optional[str]) -> Optional[str]:
"""Pick the reasoning effort from CLI override, env, settings, or default."""
return cli_reasoning or os.environ.get("OPENAI_REASONING_EFFORT") or settings.get("OPENAI_REASONING_EFFORT") or DEFAULT_REASONING
def run(prompt: object, model: str, reasoning: Optional[str], api_key: str) -> str:
"""
Use the newer Responses API (per OpenAI 2025 guidelines).
Accepts:
- str prompt
- list/tuple [text, image_bytes] for multimodal
"""
client = OpenAI(api_key=api_key)
kwargs = {}
if reasoning:
kwargs["reasoning"] = {"effort": reasoning}
# Build input payload
if isinstance(prompt, (list, tuple)) and len(prompt) == 2 and isinstance(prompt[0], str) and isinstance(prompt[1], (bytes, bytearray)):
b64 = base64.b64encode(prompt[1]).decode("utf-8")
logging.info("[llm] multimodal input: text_len=%s image_bytes=%s", len(prompt[0]), len(prompt[1]))
input_payload = [
{
"role": "user",
"content": [
{"type": "input_text", "text": prompt[0]},
{"type": "input_image", "image_url": f"data:image/png;base64,{b64}"},
],
}
]
else:
text_prompt = prompt if isinstance(prompt, str) else str(prompt)
input_payload = [
{
"role": "user",
"content": [{"type": "input_text", "text": text_prompt}],
}
]
# Debug log full payload for traceability
logging.info("[llm] model=%s reasoning=%s payload=%s", model, reasoning, input_payload)
resp = client.responses.create(
model=model,
input=input_payload,
**kwargs,
)
return getattr(resp, "output_text", None) or str(resp)
|