78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
import logging
|
|
from contextlib import contextmanager
|
|
from contextvars import ContextVar
|
|
from collections.abc import Iterator
|
|
from band_aider.ai_client.openai_client_kind import OpenAIClientKind
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CompletionScope:
|
|
"""Identify the actual completion owner, e.g. CompletionScope('IntentCheck')."""
|
|
|
|
check_name: str = ""
|
|
files: tuple[str, ...] = ()
|
|
|
|
|
|
completion_scope: ContextVar[CompletionScope] = ContextVar("completion_scope", default=CompletionScope())
|
|
|
|
|
|
@contextmanager
|
|
def scoped_completion(check_name: str, files: tuple[str, ...] = ()) -> Iterator[None]:
|
|
"""Bind one call's authoritative scope and restore its caller even after failure."""
|
|
token = completion_scope.set(CompletionScope(check_name, files))
|
|
try:
|
|
yield
|
|
finally:
|
|
completion_scope.reset(token)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OpenAIConfig:
|
|
"""OpenAIConfig stores one reconstructable OpenAI-compatible client configuration."""
|
|
|
|
model: str
|
|
api_base: str
|
|
api_key: str | None = None
|
|
client: OpenAIClientKind = OpenAIClientKind.COMPLETION
|
|
temperature: float = 0
|
|
max_tokens: int | None = None
|
|
max_attempts: int = 8
|
|
retry_temperature_step: float = 0.2
|
|
max_retry_temperature: float = 0.8
|
|
verify_ssl: bool = True
|
|
|
|
|
|
class CompletionClient(ABC):
|
|
"""CompletionClient provides prompt-to-text completion behavior."""
|
|
|
|
@property
|
|
def manages_response_validation(self) -> bool:
|
|
"""Preserve an owning workflow's single-response retry policy through decorators."""
|
|
|
|
return self.inner.manages_response_validation if self.inner is not None else False
|
|
|
|
@property
|
|
@abstractmethod
|
|
def inner(self) -> "CompletionClient | None":
|
|
"""Return the wrapped client for decorators, or None for root clients."""
|
|
|
|
@abstractmethod
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return model text for one prompt."""
|
|
|
|
@property
|
|
@abstractmethod
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Return read-only OpenAI-compatible configs available for escalation."""
|
|
|
|
|
|
class HumanResponse(str):
|
|
"""Carry dictated text through completion adapters without cleanup or model repair."""
|