95 lines
5.1 KiB
Python
95 lines
5.1 KiB
Python
import asyncio
|
|
import http.client
|
|
import os
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes
|
|
|
|
from .errors import LLMError, ProviderUnavailable
|
|
from .provider_base import ProviderResponse
|
|
from .diagnostics import envelope_details, parse_response_object
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
class OpenAICompatibleProvider:
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
return await asyncio.to_thread(self._generate, config, request, max_response_bytes)
|
|
|
|
@staticmethod
|
|
def _generate(config, request, limit):
|
|
payload = {"model": config.model, "messages": request["messages"],
|
|
config.token_parameter: config.max_output_tokens, "stream": False}
|
|
if config.reasoning_effort is not None:
|
|
payload["reasoning_effort"] = config.reasoning_effort
|
|
if config.response_format == "json_schema":
|
|
payload["response_format"] = {"type": "json_schema", "json_schema": {
|
|
"name": "case_output", "schema": request["output_schema"], "strict": True}}
|
|
elif config.response_format == "json_object":
|
|
payload["response_format"] = {"type": "json_object"}
|
|
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
|
if config.api_key_env:
|
|
secret = os.environ.get(config.api_key_env)
|
|
if not secret or any(ord(c) < 32 or ord(c) == 127 for c in secret):
|
|
raise LLMError("LLM_CREDENTIALS")
|
|
headers["Authorization"] = "Bearer " + secret
|
|
req = urllib.request.Request(config.base_url.rstrip("/") + "/chat/completions", data=json_bytes(payload), headers=headers)
|
|
# Do not inherit ambient proxies or forward credentials through redirects.
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect())
|
|
deadline = time.monotonic() + config.timeout_seconds
|
|
try:
|
|
with opener.open(req, timeout=config.timeout_seconds) as response:
|
|
chunks = []
|
|
size = 0
|
|
while True:
|
|
if time.monotonic() >= deadline:
|
|
raise ProviderUnavailable("LLM_TIMEOUT")
|
|
chunk = response.read(min(65536, limit + 1 - size))
|
|
if not chunk:
|
|
break
|
|
size += len(chunk)
|
|
if size > limit:
|
|
raise LLMError("LLM_RESPONSE_LIMIT")
|
|
chunks.append(chunk)
|
|
raw = b"".join(chunks)
|
|
except urllib.error.HTTPError as error:
|
|
error.close()
|
|
if error.code in {408, 429} or 500 <= error.code < 600:
|
|
raise ProviderUnavailable("LLM_HTTP_TEMPORARY", http_status=error.code) from None
|
|
raise LLMError("LLM_HTTP_CONFIGURATION", http_status=error.code) from None
|
|
except (urllib.error.URLError, TimeoutError, OSError, http.client.HTTPException):
|
|
raise ProviderUnavailable("LLM_CONNECTION") from None
|
|
diagnostics = {'response_bytes':len(raw),'max_output_tokens':config.max_output_tokens}
|
|
try:
|
|
envelope = parse_response_object(raw, stage='http_envelope')
|
|
diagnostics = envelope_details(envelope,config.max_output_tokens,len(raw))
|
|
choice = envelope["choices"][0]
|
|
message = choice["message"]
|
|
if choice.get("finish_reason") != "stop" or message.get("refusal") or message.get("tool_calls"):
|
|
raise LLMError("LLM_INCOMPLETE_OUTPUT",diagnostics=diagnostics)
|
|
content = message["content"]
|
|
if not isinstance(content, str):
|
|
raise LLMError("LLM_RESPONSE",diagnostics=diagnostics)
|
|
document = parse_response_object(content.encode("utf-8"), stage='output_json')
|
|
model = envelope.get("model", config.model)
|
|
if not isinstance(model, str) or not model.strip() or len(model) > 256 or any(ord(c) < 32 for c in model):
|
|
raise LLMError("LLM_RESPONSE",diagnostics=diagnostics)
|
|
usage = envelope.get("usage") or {}
|
|
input_tokens, output_tokens = usage.get("prompt_tokens"), usage.get("completion_tokens")
|
|
for count in (input_tokens, output_tokens):
|
|
if count is not None and (isinstance(count, bool) or not isinstance(count, int) or count < 0):
|
|
raise LLMError("LLM_RESPONSE",diagnostics=diagnostics)
|
|
return ProviderResponse(document, model, input_tokens, output_tokens, diagnostics,
|
|
diagnostics.get('cached_input_tokens'), diagnostics.get('reasoning_tokens'))
|
|
except LLMError as error:
|
|
error.diagnostics = diagnostics | error.diagnostics
|
|
raise
|
|
except (ValueError, TypeError, KeyError, IndexError, AttributeError, RecursionError):
|
|
raise LLMError("LLM_RESPONSE",diagnostics=diagnostics |
|
|
{'response_stage':'http_envelope','response_error':'invalid_structure'}) from None
|