File size: 3,888 Bytes
79e32b1 d1c1b5e 79e32b1 0a3ddf8 d1c1b5e 0a3ddf8 d1c1b5e 0a3ddf8 79e32b1 d1c1b5e 79e32b1 0a3ddf8 79e32b1 d1c1b5e 79e32b1 | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import os
import httpx
from openai import AsyncAzureOpenAI, AzureOpenAI, AsyncOpenAI, OpenAI
from dotenv import load_dotenv, find_dotenv
# tries to find .env in current path, or traverses parent directories until found
dotenv_path = find_dotenv()
load_dotenv(dotenv_path)
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
AZURE_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT", "https://agenthiack.openai.azure.com/")
AZURE_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION", "2024-12-01-preview")
DEFAULT_OPENAI_COMPATIBLE_BASE_URL = "https://api.groq.com/openai/v1"
def _first_env_value(*names: str, default: str = "") -> str:
for name in names:
value = os.getenv(name)
if value:
return value.strip()
return default
def _env_bool(name: str, default: bool) -> bool:
value = os.getenv(name)
if value is None:
return default
return value.strip().lower() not in {"0", "false", "no", "off"}
def _env_int(name: str, default: int, *, min_value: int = 0) -> int:
try:
return max(min_value, int(os.getenv(name, str(default))))
except (TypeError, ValueError):
return default
def _env_float(name: str, default: float, *, min_value: float = 0.1) -> float:
try:
return max(min_value, float(os.getenv(name, str(default))))
except (TypeError, ValueError):
return default
def get_openai_compatible_config() -> tuple[str, str]:
"""Return API key and base URL for OpenAI-compatible providers.
LLM_* variables are canonical for MetaRec. OPENAI_COMPAT* aliases make it
clear that this client targets the OpenAI-compatible API surface, not the
Azure OpenAI client used by the legacy agent modules.
"""
api_key = _first_env_value(
"LLM_API_KEY",
"OPENAI_COMPAT_API_KEY",
"OPENAI_COMPATIBLE_API_KEY",
"GROQ_API_KEY",
)
base_url = _first_env_value(
"LLM_BASE_URL",
"OPENAI_COMPAT_BASE_URL",
"OPENAI_COMPATIBLE_BASE_URL",
"GROQ_BASE_URL",
default=DEFAULT_OPENAI_COMPATIBLE_BASE_URL,
).rstrip("/")
return api_key, base_url
def get_openai_compatible_transport_config() -> dict[str, object]:
return {
"timeout": _env_float("LLM_TIMEOUT_SECONDS", 30.0),
"max_retries": _env_int("LLM_SDK_MAX_RETRIES", 2),
"trust_env": _env_bool("LLM_TRUST_ENV", True),
}
def describe_openai_compatible_config(model: str | None = None) -> str:
transport = get_openai_compatible_transport_config()
return (
f"base_url={LLM_BASE_URL} "
f"model={model or os.getenv('LLM_MODEL') or '(unset)'} "
f"api_key_configured={bool(LLM_API_KEY)} "
f"timeout={transport['timeout']} "
f"max_retries={transport['max_retries']} "
f"trust_env={transport['trust_env']}"
)
def _client_kwargs(async_client: bool = False) -> dict[str, object]:
transport = get_openai_compatible_transport_config()
kwargs: dict[str, object] = {
"base_url": LLM_BASE_URL,
"api_key": LLM_API_KEY,
"max_retries": transport["max_retries"],
}
timeout = transport["timeout"]
if transport["trust_env"]:
kwargs["timeout"] = timeout
elif async_client:
kwargs["http_client"] = httpx.AsyncClient(timeout=timeout, trust_env=False)
else:
kwargs["http_client"] = httpx.Client(timeout=timeout, trust_env=False)
return kwargs
LLM_API_KEY, LLM_BASE_URL = get_openai_compatible_config()
def create_sync_client():
client = OpenAI(**_client_kwargs(async_client=False))
return client
def create_sync_azure_client():
client = AzureOpenAI(
azure_endpoint=AZURE_ENDPOINT,
api_key=OPENAI_API_KEY,
api_version=AZURE_API_VERSION,
)
return client
def create_async_client():
client = AsyncOpenAI(**_client_kwargs(async_client=True))
return client
|