showtan001
60ee157e97
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
|
|
DEFAULT_PROVIDER = "anthropic"
|
|
DEFAULT_MODEL_BY_PROVIDER = {
|
|
"anthropic": "claude-sonnet-5",
|
|
"openai": "gpt-5.6",
|
|
"openai_compatible": "local-model",
|
|
}
|
|
DEFAULT_TIMEOUT_SECONDS = 30.0
|
|
DEFAULT_HISTORY_MAX_TURNS = 20
|
|
|
|
ENABLED_ENV = "AI_PLANNER_ENABLED"
|
|
PROVIDER_ENV = "AI_PLANNER_PROVIDER"
|
|
MODEL_ENV = "AI_PLANNER_MODEL"
|
|
TIMEOUT_ENV = "AI_PLANNER_TIMEOUT_SECONDS"
|
|
THINKING_BUDGET_ENV = "AI_PLANNER_THINKING_BUDGET_TOKENS"
|
|
API_KEY_ENV = "AI_PLANNER_API_KEY"
|
|
BASE_URL_ENV = "AI_PLANNER_BASE_URL"
|
|
MULTIMODAL_ENV = "AI_PLANNER_MULTIMODAL"
|
|
HISTORY_MAX_TURNS_ENV = "AI_PLANNER_HISTORY_MAX_TURNS"
|
|
|
|
SUPPORTED_PROVIDERS = frozenset(DEFAULT_MODEL_BY_PROVIDER)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlannerConfig:
|
|
enabled: bool = False
|
|
provider: str = DEFAULT_PROVIDER
|
|
model: str = ""
|
|
timeout: float = DEFAULT_TIMEOUT_SECONDS
|
|
thinking_budget_tokens: int | None = None
|
|
api_key: str | None = None
|
|
base_url: str | None = None
|
|
multimodal: bool = False
|
|
history_max_turns: int = DEFAULT_HISTORY_MAX_TURNS
|
|
|
|
def resolved_model(self) -> str:
|
|
return self.model or DEFAULT_MODEL_BY_PROVIDER[self.provider]
|
|
|
|
|
|
def load_config(env: Mapping[str, str] | None = None) -> PlannerConfig:
|
|
values = env or os.environ
|
|
return PlannerConfig(
|
|
enabled=_parse_bool(values.get(ENABLED_ENV), default=False),
|
|
provider=_parse_provider(values.get(PROVIDER_ENV)),
|
|
model=values.get(MODEL_ENV) or "",
|
|
timeout=_parse_timeout(values.get(TIMEOUT_ENV)),
|
|
thinking_budget_tokens=_parse_thinking_budget(values.get(THINKING_BUDGET_ENV)),
|
|
api_key=values.get(API_KEY_ENV) or _provider_key(values),
|
|
base_url=values.get(BASE_URL_ENV) or None,
|
|
multimodal=_parse_bool(values.get(MULTIMODAL_ENV), default=False),
|
|
history_max_turns=_parse_positive_int(
|
|
values.get(HISTORY_MAX_TURNS_ENV),
|
|
default=DEFAULT_HISTORY_MAX_TURNS,
|
|
),
|
|
)
|
|
|
|
|
|
def _parse_bool(value: str | None, *, default: bool) -> bool:
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in {"1", "true", "yes", "on", "enabled"}
|
|
|
|
|
|
def _parse_provider(value: str | None) -> str:
|
|
if value is None:
|
|
return DEFAULT_PROVIDER
|
|
provider = value.strip().lower()
|
|
if provider in {"openai-compatible", "openai_compatible", "local"}:
|
|
return "openai_compatible"
|
|
return provider if provider in SUPPORTED_PROVIDERS else DEFAULT_PROVIDER
|
|
|
|
|
|
def _parse_timeout(value: str | None) -> float:
|
|
if value is None:
|
|
return DEFAULT_TIMEOUT_SECONDS
|
|
try:
|
|
timeout = float(value)
|
|
except ValueError:
|
|
return DEFAULT_TIMEOUT_SECONDS
|
|
return timeout if timeout > 0 else DEFAULT_TIMEOUT_SECONDS
|
|
|
|
|
|
def _parse_thinking_budget(value: str | None) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
budget = int(value)
|
|
except ValueError:
|
|
return None
|
|
return budget if budget > 0 else None
|
|
|
|
|
|
def _parse_positive_int(value: str | None, *, default: int) -> int:
|
|
if value is None:
|
|
return default
|
|
try:
|
|
parsed = int(value)
|
|
except ValueError:
|
|
return default
|
|
return parsed if parsed > 0 else default
|
|
|
|
|
|
def _provider_key(values: Mapping[str, str]) -> str | None:
|
|
provider = (values.get(PROVIDER_ENV) or DEFAULT_PROVIDER).strip().lower()
|
|
if provider == "openai":
|
|
return values.get("OPENAI_API_KEY") or None
|
|
if provider == "anthropic":
|
|
return values.get("ANTHROPIC_API_KEY") or None
|
|
return None
|