1e98b384e6
Previous commit (b5aa6ffd) dropped top_k for every Anthropic call,
but only Claude 4.7 (Opus/Sonnet/Haiku) actually rejects it. 4.6, 4.5,
and the 3.x line still accept top_k and use it as documented.
Backend: _stream_anthropic matches the model id against
^claude-(opus|sonnet|haiku)-4-7(-|.|$) and only strips top_k when it
hits. Every other Claude generation continues to receive the value
from the chat settings panel.
Frontend: anthropic.topK is restored to true so the Top K slider is
visible again — the backend handles the per-model drop, and the
4.7 case is silent (request still succeeds without top_k).
759 lines
31 KiB
Python
759 lines
31 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""
|
|
Async HTTP client for proxying chat completions to external LLM providers.
|
|
|
|
Most registry providers expose OpenAI-compatible /v1/chat/completions endpoints;
|
|
Anthropic uses native Messages API with translation in this client.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any, AsyncGenerator, Optional
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Claude 4.7 (Opus/Sonnet/Haiku) deprecated top_k and returns 400
|
|
# "top_k is deprecated for this model" when it is set. 3.x and 4.5/4.6
|
|
# still accept it. Match the 4-7 line specifically so we keep the knob
|
|
# live on every other Claude generation.
|
|
_ANTHROPIC_TOP_K_DEPRECATED = re.compile(r"^claude-(?:opus|sonnet|haiku)-4-7(?:[-.]|$)")
|
|
|
|
# Shared client reused across all requests for HTTP connection pooling.
|
|
# Auth headers and timeouts are passed per-request, so a single client
|
|
# handles every provider without storing credentials.
|
|
_http_client = httpx.AsyncClient()
|
|
|
|
|
|
class ExternalProviderClient:
|
|
"""Async proxy for OpenAI-compatible external LLM APIs."""
|
|
|
|
def __init__(
|
|
self,
|
|
provider_type: str,
|
|
base_url: str,
|
|
api_key: str,
|
|
timeout: float = 120.0,
|
|
):
|
|
self.provider_type = provider_type
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
self._timeout = httpx.Timeout(timeout, connect = 10.0)
|
|
|
|
def _auth_headers(self) -> dict[str, str]:
|
|
"""Build authentication headers using the provider's registry config."""
|
|
from core.inference.providers import get_provider_info
|
|
|
|
provider_info = get_provider_info(self.provider_type) or {}
|
|
auth_header = provider_info.get("auth_header", "Authorization")
|
|
auth_prefix = provider_info.get("auth_prefix", "Bearer ")
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
auth_header: f"{auth_prefix}{self.api_key}",
|
|
}
|
|
# Merge any provider-specific extra headers (e.g. anthropic-version, OpenRouter attribution)
|
|
headers.update(provider_info.get("extra_headers", {}))
|
|
return headers
|
|
|
|
def _is_openai_compatible(self) -> bool:
|
|
"""Return False for providers that need request/response translation (e.g. Anthropic)."""
|
|
from core.inference.providers import get_provider_info
|
|
|
|
info = get_provider_info(self.provider_type) or {}
|
|
return info.get("openai_compatible", True)
|
|
|
|
async def stream_chat_completion(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
model: str,
|
|
temperature: float = 0.7,
|
|
top_p: float = 0.95,
|
|
max_tokens: Optional[int] = None,
|
|
presence_penalty: float = 0.0,
|
|
top_k: Optional[int] = None,
|
|
stream: bool = True,
|
|
) -> AsyncGenerator[str, None]:
|
|
"""
|
|
Yield OpenAI-format SSE lines from the external provider.
|
|
|
|
For OpenAI-compatible providers, lines are forwarded verbatim.
|
|
For Anthropic, the native Messages API SSE is translated to OpenAI format.
|
|
|
|
``top_k`` and ``presence_penalty`` are forwarded only when the caller
|
|
supplies a value the provider accepts — the frontend's
|
|
provider-capability map already filters these per provider, so we
|
|
treat them as opt-in here.
|
|
"""
|
|
if not self._is_openai_compatible():
|
|
async for line in self._stream_anthropic(
|
|
messages, model, temperature, top_p, max_tokens, top_k
|
|
):
|
|
yield line
|
|
return
|
|
|
|
# OpenAI moved their flagship models (gpt-5.x) off /v1/chat/completions
|
|
# — those endpoints return 404 with "This is not a chat model" for the
|
|
# new families. Route all OpenAI traffic through /v1/responses instead;
|
|
# we translate the Responses SSE back into Chat Completions chunks so
|
|
# the frontend stays endpoint-agnostic.
|
|
if self.provider_type == "openai":
|
|
async for line in self._stream_openai_responses(
|
|
messages, model, temperature, top_p, max_tokens
|
|
):
|
|
yield line
|
|
return
|
|
|
|
body: dict[str, Any] = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"stream": stream,
|
|
"temperature": temperature,
|
|
"top_p": top_p,
|
|
"presence_penalty": presence_penalty,
|
|
}
|
|
if max_tokens is not None:
|
|
# OpenAI newer models (gpt-4o, gpt-5.x) reject max_tokens
|
|
if self.provider_type == "openai":
|
|
body["max_completion_tokens"] = max_tokens
|
|
else:
|
|
body["max_tokens"] = max_tokens
|
|
|
|
url = f"{self.base_url}/chat/completions"
|
|
logger.info(
|
|
"Proxying chat completion to %s (provider=%s, model=%s)",
|
|
url,
|
|
self.provider_type,
|
|
model,
|
|
)
|
|
|
|
try:
|
|
async with _http_client.stream(
|
|
"POST",
|
|
url,
|
|
json = body,
|
|
headers = self._auth_headers(),
|
|
timeout = self._timeout,
|
|
) as response:
|
|
if response.status_code != 200:
|
|
error_body = await response.aread()
|
|
error_text = error_body.decode("utf-8", errors = "replace")
|
|
logger.error(
|
|
"External provider returned %d: %s",
|
|
response.status_code,
|
|
error_text[:500],
|
|
)
|
|
yield _error_sse_line(
|
|
response.status_code, error_text, self.provider_type
|
|
)
|
|
return
|
|
|
|
# NOTE: manual __anext__ loop instead of `async for` is intentional.
|
|
# On Python 3.13 + httpcore 1.0.x, `async for` auto-calls aclose() on
|
|
# early exit (break/return/GeneratorExit) BEFORE our finally block runs.
|
|
# That propagates GeneratorExit into PoolByteStream.__aiter__() while it
|
|
# calls `await self.aclose()` inside `with AsyncShieldCancellation()`,
|
|
# triggering "RuntimeError: async generator ignored GeneratorExit".
|
|
# Fix: call response.aclose() FIRST (sets PoolByteStream._closed=True),
|
|
# then lines_gen.aclose() is a no-op and GeneratorExit re-raises cleanly.
|
|
lines_gen = response.aiter_lines().__aiter__()
|
|
try:
|
|
while True:
|
|
try:
|
|
line = await lines_gen.__anext__()
|
|
except StopAsyncIteration:
|
|
break
|
|
if line.strip():
|
|
yield line
|
|
except GeneratorExit:
|
|
await response.aclose() # set PoolByteStream._closed=True FIRST
|
|
await lines_gen.aclose() # now safe — aclose() is a no-op
|
|
raise
|
|
finally:
|
|
await response.aclose()
|
|
await lines_gen.aclose()
|
|
|
|
except httpx.ConnectError as exc:
|
|
logger.error("Connection error to %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
502,
|
|
f"Failed to connect to {self.provider_type}: {exc}",
|
|
self.provider_type,
|
|
)
|
|
except httpx.ReadTimeout as exc:
|
|
logger.error("Read timeout from %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
504,
|
|
f"Timeout waiting for {self.provider_type} response",
|
|
self.provider_type,
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
logger.error("HTTP error from %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
502,
|
|
f"Error communicating with {self.provider_type}: {exc}",
|
|
self.provider_type,
|
|
)
|
|
|
|
async def _stream_anthropic(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
model: str,
|
|
temperature: float,
|
|
top_p: float,
|
|
max_tokens: Optional[int],
|
|
top_k: Optional[int] = None,
|
|
) -> AsyncGenerator[str, None]:
|
|
"""
|
|
Call the Anthropic Messages API and translate its SSE to OpenAI format.
|
|
|
|
Anthropic SSE event types:
|
|
content_block_delta → OpenAI chunk with delta.content
|
|
message_delta → OpenAI chunk with finish_reason
|
|
message_stop → data: [DONE]
|
|
(all others skipped)
|
|
"""
|
|
import json as _json
|
|
|
|
# Extract system prompt and translate image_url parts to Anthropic format
|
|
system: Optional[str] = None
|
|
filtered: list[dict[str, Any]] = []
|
|
for msg in messages:
|
|
if msg.get("role") == "system":
|
|
content = msg.get("content", "")
|
|
system = (
|
|
content
|
|
if isinstance(content, str)
|
|
else "\n".join(
|
|
p["text"] for p in content if p.get("type") == "text"
|
|
)
|
|
)
|
|
continue
|
|
|
|
content = msg.get("content")
|
|
if isinstance(content, list):
|
|
# Translate OpenAI image_url parts → Anthropic native image format
|
|
anthropic_parts: list[dict[str, Any]] = []
|
|
for part in content:
|
|
if part.get("type") == "text":
|
|
anthropic_parts.append({"type": "text", "text": part["text"]})
|
|
elif part.get("type") == "image_url":
|
|
url = part.get("image_url", {}).get("url", "")
|
|
if url.startswith("data:"):
|
|
# data:image/png;base64,<DATA> → split header and data
|
|
header, _, b64data = url.partition(",")
|
|
media_type = (
|
|
header.split(";")[0].replace("data:", "")
|
|
or "image/jpeg"
|
|
)
|
|
anthropic_parts.append(
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": media_type,
|
|
"data": b64data,
|
|
},
|
|
}
|
|
)
|
|
else:
|
|
# Remote URL — Anthropic supports url source type natively.
|
|
# See: https://docs.anthropic.com/en/docs/build-with-claude/vision#url-based-images
|
|
anthropic_parts.append(
|
|
{
|
|
"type": "image",
|
|
"source": {
|
|
"type": "url",
|
|
"url": url,
|
|
},
|
|
}
|
|
)
|
|
filtered.append({"role": msg["role"], "content": anthropic_parts})
|
|
else:
|
|
filtered.append(msg)
|
|
|
|
body: dict[str, Any] = {
|
|
"model": model,
|
|
"messages": filtered,
|
|
"max_tokens": max_tokens or 1024, # required by Anthropic
|
|
"temperature": temperature,
|
|
# Anthropic rejects requests that set both temperature and top_p
|
|
"stream": True,
|
|
}
|
|
# top_k is deprecated on Claude 4.7 (Opus/Sonnet/Haiku) — the API
|
|
# returns 400 "top_k is deprecated for this model" when it is set.
|
|
# 3.x and 4.5/4.6 still accept it, so gate strictly on the 4.7 ids.
|
|
if (
|
|
top_k is not None
|
|
and top_k > 0
|
|
and not _ANTHROPIC_TOP_K_DEPRECATED.match(model)
|
|
):
|
|
body["top_k"] = top_k
|
|
if system:
|
|
body["system"] = system
|
|
|
|
url = f"{self.base_url}/messages"
|
|
completion_id = f"chatcmpl-anthropic-{model.replace('/', '-')}"
|
|
|
|
_finish_reason_map = {
|
|
"end_turn": "stop",
|
|
"max_tokens": "length",
|
|
"stop_sequence": "stop",
|
|
}
|
|
|
|
logger.info("Proxying Anthropic Messages API to %s (model=%s)", url, model)
|
|
|
|
try:
|
|
async with _http_client.stream(
|
|
"POST",
|
|
url,
|
|
json = body,
|
|
headers = self._auth_headers(),
|
|
timeout = self._timeout,
|
|
) as response:
|
|
if response.status_code != 200:
|
|
error_body = await response.aread()
|
|
error_text = error_body.decode("utf-8", errors = "replace")
|
|
logger.error(
|
|
"Anthropic returned %d: %s",
|
|
response.status_code,
|
|
error_text[:500],
|
|
)
|
|
yield _error_sse_line(
|
|
response.status_code, error_text, self.provider_type
|
|
)
|
|
return
|
|
|
|
# NOTE: same manual __anext__ loop as stream_chat_completion — see comment there.
|
|
lines_gen = response.aiter_lines().__aiter__()
|
|
try:
|
|
while True:
|
|
try:
|
|
line = await lines_gen.__anext__()
|
|
except StopAsyncIteration:
|
|
break
|
|
if not line or line.startswith("event:"):
|
|
continue
|
|
if not line.startswith("data:"):
|
|
continue
|
|
|
|
data_str = line[len("data:") :].strip()
|
|
if not data_str:
|
|
continue
|
|
|
|
try:
|
|
event = _json.loads(data_str)
|
|
except _json.JSONDecodeError:
|
|
continue
|
|
|
|
event_type = event.get("type")
|
|
|
|
if event_type == "content_block_delta":
|
|
delta = event.get("delta", {})
|
|
if delta.get("type") == "text_delta":
|
|
chunk = {
|
|
"id": completion_id,
|
|
"object": "chat.completion.chunk",
|
|
"choices": [
|
|
{
|
|
"index": 0,
|
|
"delta": {"content": delta.get("text", "")},
|
|
"finish_reason": None,
|
|
}
|
|
],
|
|
}
|
|
yield f"data: {_json.dumps(chunk)}"
|
|
|
|
elif event_type == "message_delta":
|
|
stop_reason = event.get("delta", {}).get("stop_reason")
|
|
if stop_reason:
|
|
chunk = {
|
|
"id": completion_id,
|
|
"object": "chat.completion.chunk",
|
|
"choices": [
|
|
{
|
|
"index": 0,
|
|
"delta": {},
|
|
"finish_reason": _finish_reason_map.get(
|
|
stop_reason, "stop"
|
|
),
|
|
}
|
|
],
|
|
}
|
|
yield f"data: {_json.dumps(chunk)}"
|
|
|
|
elif event_type == "message_stop":
|
|
yield "data: [DONE]"
|
|
await (
|
|
response.aclose()
|
|
) # set PoolByteStream._closed=True FIRST
|
|
break
|
|
except GeneratorExit:
|
|
await response.aclose() # set PoolByteStream._closed=True FIRST
|
|
await lines_gen.aclose() # now safe — aclose() is a no-op
|
|
raise
|
|
finally:
|
|
await response.aclose()
|
|
await lines_gen.aclose()
|
|
|
|
except httpx.ConnectError as exc:
|
|
logger.error("Connection error to %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
502,
|
|
f"Failed to connect to {self.provider_type}: {exc}",
|
|
self.provider_type,
|
|
)
|
|
except httpx.ReadTimeout as exc:
|
|
logger.error("Read timeout from %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
504,
|
|
f"Timeout waiting for {self.provider_type} response",
|
|
self.provider_type,
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
logger.error("HTTP error from %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
502,
|
|
f"Error communicating with {self.provider_type}: {exc}",
|
|
self.provider_type,
|
|
)
|
|
|
|
async def _stream_openai_responses(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
model: str,
|
|
temperature: float,
|
|
top_p: float,
|
|
max_tokens: Optional[int],
|
|
) -> AsyncGenerator[str, None]:
|
|
"""
|
|
Call OpenAI's /v1/responses endpoint and translate its SSE stream back
|
|
into OpenAI Chat Completions chunk format.
|
|
|
|
The Responses API uses a different request shape (``input`` instead of
|
|
``messages``, ``instructions`` for system prompts, ``max_output_tokens``
|
|
for the budget) and emits event-typed SSE frames (e.g.
|
|
``response.output_text.delta``) rather than chat-completion chunks.
|
|
``presence_penalty`` / ``top_k`` are not part of the Responses contract
|
|
and are dropped here intentionally.
|
|
"""
|
|
import json as _json
|
|
|
|
# Split system messages out into a single `instructions` string and
|
|
# translate user/assistant messages into the Responses input shape.
|
|
instructions_parts: list[str] = []
|
|
input_items: list[dict[str, Any]] = []
|
|
for msg in messages:
|
|
role = msg.get("role")
|
|
content = msg.get("content", "")
|
|
|
|
if role == "system":
|
|
if isinstance(content, str):
|
|
if content:
|
|
instructions_parts.append(content)
|
|
elif isinstance(content, list):
|
|
for part in content:
|
|
if part.get("type") == "text" and part.get("text"):
|
|
instructions_parts.append(part["text"])
|
|
continue
|
|
|
|
if isinstance(content, str):
|
|
input_items.append({"role": role, "content": content})
|
|
continue
|
|
|
|
if isinstance(content, list):
|
|
translated_parts: list[dict[str, Any]] = []
|
|
for part in content:
|
|
part_type = part.get("type")
|
|
if part_type == "text":
|
|
translated_parts.append(
|
|
{"type": "input_text", "text": part.get("text", "")}
|
|
)
|
|
elif part_type == "image_url":
|
|
url = part.get("image_url", {}).get("url", "")
|
|
if url:
|
|
# Responses takes image_url as a flat string (both
|
|
# https:// URLs and data: URLs are accepted).
|
|
translated_parts.append(
|
|
{"type": "input_image", "image_url": url}
|
|
)
|
|
if translated_parts:
|
|
input_items.append({"role": role, "content": translated_parts})
|
|
|
|
# NOTE: gpt-5.x / o3 / gpt-4.5 are reasoning-class models. They reject
|
|
# temperature and top_p with `Unsupported parameter` 400s on
|
|
# /v1/responses (and on /v1/chat/completions for the same families).
|
|
# The PROVIDER_REGISTRY['openai'] model_id_allowlist already scopes
|
|
# the picker to those families, so we never need to send sampling
|
|
# knobs here. ``reasoning.effort`` defaults to "medium" server-side
|
|
# if omitted — surface it in a future commit if a knob is wanted.
|
|
del temperature, top_p # explicit drop — params are accepted for
|
|
# API symmetry with the other stream methods but not forwarded.
|
|
|
|
body: dict[str, Any] = {
|
|
"model": model,
|
|
"input": input_items,
|
|
"stream": True,
|
|
}
|
|
if instructions_parts:
|
|
body["instructions"] = "\n\n".join(instructions_parts)
|
|
if max_tokens is not None:
|
|
body["max_output_tokens"] = max_tokens
|
|
|
|
url = f"{self.base_url}/responses"
|
|
completion_id = f"chatcmpl-openai-{model.replace('/', '-')}"
|
|
|
|
logger.info("Proxying OpenAI Responses API to %s (model=%s)", url, model)
|
|
|
|
try:
|
|
async with _http_client.stream(
|
|
"POST",
|
|
url,
|
|
json = body,
|
|
headers = self._auth_headers(),
|
|
timeout = self._timeout,
|
|
) as response:
|
|
if response.status_code != 200:
|
|
error_body = await response.aread()
|
|
error_text = error_body.decode("utf-8", errors = "replace")
|
|
logger.error(
|
|
"OpenAI Responses returned %d: %s",
|
|
response.status_code,
|
|
error_text[:500],
|
|
)
|
|
yield _error_sse_line(
|
|
response.status_code, error_text, self.provider_type
|
|
)
|
|
return
|
|
|
|
# NOTE: same manual __anext__ loop as stream_chat_completion —
|
|
# see comment there for the GeneratorExit / aclose ordering.
|
|
lines_gen = response.aiter_lines().__aiter__()
|
|
done_emitted = False
|
|
try:
|
|
while True:
|
|
try:
|
|
line = await lines_gen.__anext__()
|
|
except StopAsyncIteration:
|
|
break
|
|
if not line or line.startswith("event:"):
|
|
continue
|
|
if not line.startswith("data:"):
|
|
continue
|
|
|
|
data_str = line[len("data:") :].strip()
|
|
if not data_str:
|
|
continue
|
|
if data_str == "[DONE]":
|
|
if not done_emitted:
|
|
yield "data: [DONE]"
|
|
done_emitted = True
|
|
break
|
|
|
|
try:
|
|
event = _json.loads(data_str)
|
|
except _json.JSONDecodeError:
|
|
continue
|
|
|
|
event_type = event.get("type")
|
|
|
|
if event_type == "response.output_text.delta":
|
|
delta_text = event.get("delta", "")
|
|
if delta_text:
|
|
chunk = {
|
|
"id": completion_id,
|
|
"object": "chat.completion.chunk",
|
|
"choices": [
|
|
{
|
|
"index": 0,
|
|
"delta": {"content": delta_text},
|
|
"finish_reason": None,
|
|
}
|
|
],
|
|
}
|
|
yield f"data: {_json.dumps(chunk)}"
|
|
|
|
elif event_type == "response.completed":
|
|
chunk = {
|
|
"id": completion_id,
|
|
"object": "chat.completion.chunk",
|
|
"choices": [
|
|
{
|
|
"index": 0,
|
|
"delta": {},
|
|
"finish_reason": "stop",
|
|
}
|
|
],
|
|
}
|
|
yield f"data: {_json.dumps(chunk)}"
|
|
|
|
elif event_type == "response.incomplete":
|
|
chunk = {
|
|
"id": completion_id,
|
|
"object": "chat.completion.chunk",
|
|
"choices": [
|
|
{
|
|
"index": 0,
|
|
"delta": {},
|
|
"finish_reason": "length",
|
|
}
|
|
],
|
|
}
|
|
yield f"data: {_json.dumps(chunk)}"
|
|
|
|
elif event_type in ("response.failed", "error"):
|
|
# Surface the failure to the client; let the
|
|
# outer route emit [DONE] as part of its cleanup.
|
|
error_payload = event.get("response", {}).get(
|
|
"error", {}
|
|
) or {
|
|
"message": event.get("message", "Unknown error"),
|
|
"code": event.get("code"),
|
|
}
|
|
yield _error_sse_line(
|
|
502,
|
|
_json.dumps(error_payload),
|
|
self.provider_type,
|
|
)
|
|
break
|
|
except GeneratorExit:
|
|
await response.aclose()
|
|
await lines_gen.aclose()
|
|
raise
|
|
finally:
|
|
await response.aclose()
|
|
await lines_gen.aclose()
|
|
|
|
except httpx.ConnectError as exc:
|
|
logger.error("Connection error to %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
502,
|
|
f"Failed to connect to {self.provider_type}: {exc}",
|
|
self.provider_type,
|
|
)
|
|
except httpx.ReadTimeout as exc:
|
|
logger.error("Read timeout from %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
504,
|
|
f"Timeout waiting for {self.provider_type} response",
|
|
self.provider_type,
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
logger.error("HTTP error from %s: %s", self.provider_type, exc)
|
|
yield _error_sse_line(
|
|
502,
|
|
f"Error communicating with {self.provider_type}: {exc}",
|
|
self.provider_type,
|
|
)
|
|
|
|
async def chat_completion(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
model: str,
|
|
temperature: float = 0.7,
|
|
top_p: float = 0.95,
|
|
max_tokens: Optional[int] = None,
|
|
presence_penalty: float = 0.0,
|
|
) -> dict[str, Any]:
|
|
"""Non-streaming chat completion. Returns the full response dict.
|
|
|
|
Note: only valid for OpenAI-compatible providers. Anthropic requires its
|
|
own Messages API; use stream_chat_completion (with stream=False) instead
|
|
if a non-streaming Anthropic path is needed in the future.
|
|
"""
|
|
body: dict[str, Any] = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"stream": False,
|
|
"temperature": temperature,
|
|
"top_p": top_p,
|
|
"presence_penalty": presence_penalty,
|
|
}
|
|
if max_tokens is not None:
|
|
if self.provider_type == "openai":
|
|
body["max_completion_tokens"] = max_tokens
|
|
else:
|
|
body["max_tokens"] = max_tokens
|
|
|
|
response = await _http_client.post(
|
|
f"{self.base_url}/chat/completions",
|
|
json = body,
|
|
headers = self._auth_headers(),
|
|
timeout = self._timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def list_models(self) -> list[dict[str, Any]]:
|
|
"""
|
|
Call GET /models on the provider to discover available models.
|
|
|
|
Returns a list of model dicts with at least 'id' and optionally
|
|
'created', 'owned_by', etc.
|
|
|
|
All supported providers expose a /models endpoint:
|
|
- OpenAI-compatible: standard {"data": [...]} response
|
|
- Anthropic: https://api.anthropic.com/v1/models — same {"data": [...]} shape
|
|
"""
|
|
try:
|
|
response = await _http_client.get(
|
|
f"{self.base_url}/models",
|
|
headers = self._auth_headers(),
|
|
timeout = self._timeout,
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
# OpenAI format: {"data": [{"id": "...", ...}, ...]}
|
|
models = data.get("data", [])
|
|
return models
|
|
except httpx.HTTPError as exc:
|
|
logger.error("Failed to list models from %s: %s", self.provider_type, exc)
|
|
raise
|
|
|
|
async def verify_models_endpoint_lightweight(self) -> None:
|
|
"""
|
|
Confirm GET /models returns 200 without buffering the full response body.
|
|
|
|
Used for providers with enormous catalogs (e.g. OpenRouter, Hugging Face router)
|
|
where downloading the full JSON would be prohibitive.
|
|
"""
|
|
url = f"{self.base_url}/models"
|
|
try:
|
|
async with _http_client.stream(
|
|
"GET",
|
|
url,
|
|
headers = self._auth_headers(),
|
|
timeout = self._timeout,
|
|
) as response:
|
|
if response.status_code != 200:
|
|
response.raise_for_status()
|
|
async for _chunk in response.aiter_bytes(chunk_size = 2048):
|
|
break
|
|
except httpx.HTTPError as exc:
|
|
logger.error(
|
|
"Lightweight /models check failed for %s: %s",
|
|
self.provider_type,
|
|
exc,
|
|
)
|
|
raise
|
|
|
|
async def close(self) -> None:
|
|
"""No-op — the underlying client is shared across requests."""
|
|
|
|
|
|
def _error_sse_line(status_code: int, message: str, provider_type: str) -> str:
|
|
"""Format an error as an SSE data line in OpenAI error format."""
|
|
import json
|
|
|
|
error_obj = {
|
|
"error": {
|
|
"message": message,
|
|
"type": "provider_error",
|
|
"code": str(status_code),
|
|
"provider": provider_type,
|
|
}
|
|
}
|
|
return f"data: {json.dumps(error_obj)}"
|