def __init__(self, config: OpenAIGPTConfig = OpenAIGPTConfig()):
"""
Args:
config: configuration for openai-gpt model
"""
# copy the config to avoid modifying the original; deep to decouple
# nested models while preserving their concrete subclasses
# The api_key_provider must NOT be deep-copied: the client cache is
# keyed on its identity (so a copy would defeat cache sharing), and it
# may hold non-copyable state (e.g. a threading.Lock). Clear it in a
# shallow copy first, then restore the original callable afterwards.
api_key_provider = config.api_key_provider
if api_key_provider is not None:
config = config.model_copy(update={"api_key_provider": None})
config = config.model_copy(deep=True)
if api_key_provider is not None:
config.api_key_provider = api_key_provider
super().__init__(config)
self.config: OpenAIGPTConfig = config
# save original model name such as `provider/model` before
# we strip out the `provider` - we retain the original in
# case some params are specific to a provider.
self.chat_model_orig = self.config.chat_model_orig or self.config.chat_model
# Run the first time the model is used
self.run_on_first_use = cache(self.config.run_on_first_use)
# global override of chat_model,
# to allow quick testing with other models
if settings.chat_model != "":
self.config.chat_model = settings.chat_model
self.chat_model_orig = settings.chat_model
self.config.completion_model = settings.chat_model
# Whether the caller asked for a vertexai/ route, recorded BEFORE the
# `model//formatter` split below can eat the prefix.
asked_for_vertexai = self.config.chat_model.startswith(VERTEXAI_PREFIX)
if len(parts := self.config.chat_model.split("//")) > 1:
# there is a formatter specified, e.g.
# "litellm/ollama/mistral//hf" or
# "local/localhost:8000/v1//mistral-instruct-v0.2"
formatter = parts[1]
self.config.chat_model = parts[0]
if formatter == "hf":
# e.g. "litellm/ollama/mistral//hf" -> "litellm/ollama/mistral"
formatter = find_hf_formatter(self.config.chat_model)
if formatter != "":
# e.g. "mistral"
self.config.formatter = formatter
logging.warning(
f"""
Using completions (not chat) endpoint with HuggingFace
chat_template for {formatter} for
model {self.config.chat_model}
"""
)
else:
# e.g. "local/localhost:8000/v1//mistral-instruct-v0.2"
self.config.formatter = formatter
if asked_for_vertexai and not self.config.chat_model.startswith(
VERTEXAI_PREFIX
):
# The `//formatter` suffix consumed the route itself, e.g.
# "vertexai//hf" -> the bare model "vertexai". Left alone this
# falls through to the generic branch and quietly talks to
# api.openai.com, which is never what someone typing `vertexai`
# meant. (On a VertexAIConfig it would also take a Google
# credential there -- the refusal below catches that case too, but
# this one fires for a plain OpenAIGPTConfig as well.)
raise ValueError(
f"chat_model {self.chat_model_orig!r} is not a usable "
"vertexai/ route: the //formatter suffix left no model behind. "
"Write vertexai/<publisher>/<model>//<formatter>, e.g. "
"vertexai/google/gemini-2.5-flash//hf."
)
# A vertexai/ route must not inherit OPENAI_-prefixed settings, so
# rebuild the config as a clean VertexAIConfig.
#
# Placed here deliberately, and the position is load-bearing twice
# over: AFTER the settings.chat_model override above, so the direct
# route and the override path get the same clean config; and AFTER the
# `model//formatter` split just above, so both branches below see the
# final chat_model. Before the split, "vertexai//hf" still carried the
# prefix and slipped past both branches -- then the split turned it
# into the bare model "vertexai", which routes to api.openai.com,
# taking a VERTEXAI_API_KEY with it.
if self.config.chat_model.startswith(VERTEXAI_PREFIX) and not isinstance(
self.config, VertexAIConfig
):
self.config = _as_vertexai_config(self.config)
# The conversion drops chat_model_orig (OPENAI_CHAT_MODEL_ORIG
# could otherwise decide which provider branch claims this
# request), so recompute it from the model actually in effect.
self.chat_model_orig = self.config.chat_model
elif isinstance(self.config, VertexAIConfig) and not (
self.config.chat_model.startswith(VERTEXAI_PREFIX)
):
# The converse of the above, and the same leak running backwards: a
# VertexAIConfig carries a credential from VERTEXAI_API_KEY, and
# without the vertexai/ prefix this would fall through to the
# generic branch -- api_base None, i.e. api.openai.com -- and send
# that credential there. Reachable via the settings.chat_model
# override alone (`-m gpt-4o`), so refuse rather than guess.
raise ValueError(
f"chat_model {self.config.chat_model!r} is not a vertexai/ "
"route, but the config is a VertexAIConfig, whose credential "
"is meant for Google. Use an OpenAIGPTConfig for non-Vertex "
"models. (If this came from a global chat_model override such "
"as `-m <model>`, that override cannot be applied to a "
"VertexAIConfig. If you passed an existing llm.config back in, "
"note that its vertexai/ prefix was already stripped.)"
)
if self.config.formatter is not None:
self.config.hf_formatter = HFFormatter(
HFPromptFormatterConfig(model_name=self.config.formatter)
)
self.supports_json_schema: bool = self.config.supports_json_schema or False
self.supports_strict_tools: bool = self.config.supports_strict_tools or False
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", DUMMY_API_KEY)
# read from self.config, not the `config` argument: a vertexai/ route
# replaced self.config above, and the original still holds the
# OPENAI_-prefixed key we must not use.
self.api_key = self.config.api_key
# if model name starts with "litellm",
# set the actual model name by stripping the "litellm/" prefix
# and set the litellm flag to True
if self.config.chat_model.startswith("litellm/") or self.config.litellm:
# e.g. litellm/ollama/mistral
self.config.litellm = True
self.api_base = self.config.api_base
if self.config.chat_model.startswith("litellm/"):
# strip the "litellm/" prefix
# e.g. litellm/ollama/llama2 => ollama/llama2
self.config.chat_model = self.config.chat_model.split("/", 1)[1]
elif self.config.chat_model.startswith("local/"):
# expect this to be of the form "local/localhost:8000/v1",
# depending on how the model is launched locally.
# In this case the model served locally behind an OpenAI-compatible API
# so we can just use `openai.*` methods directly,
# and don't need a adaptor library like litellm
self.config.litellm = False
self.config.seed = None # some models raise an error when seed is set
# Extract the api_base from the model name after the "local/" prefix
self.api_base = self.config.chat_model.split("/", 1)[1]
if not self.api_base.startswith("http"):
self.api_base = "http://" + self.api_base
elif self.config.chat_model.startswith("ollama/"):
self.config.ollama = True
# use api_base from config if set, else fall back on OLLAMA_BASE_URL
self.api_base = self.config.api_base or OLLAMA_BASE_URL
if self.api_key == OPENAI_API_KEY:
self.api_key = OLLAMA_API_KEY
self.config.chat_model = self.config.chat_model.replace("ollama/", "")
elif self.config.chat_model.startswith("vllm/"):
self.supports_json_schema = True
self.config.chat_model = self.config.chat_model.replace("vllm/", "")
if self.api_key == OPENAI_API_KEY:
self.api_key = os.environ.get("VLLM_API_KEY", DUMMY_API_KEY)
self.api_base = self.config.api_base or "http://localhost:8000/v1"
if not self.api_base.startswith("http"):
self.api_base = "http://" + self.api_base
if not self.api_base.endswith("/v1"):
self.api_base = self.api_base + "/v1"
elif self.config.chat_model.startswith("llamacpp/"):
self.supports_json_schema = True
self.api_base = self.config.chat_model.split("/", 1)[1]
if not self.api_base.startswith("http"):
self.api_base = "http://" + self.api_base
if self.api_key == OPENAI_API_KEY:
self.api_key = os.environ.get("LLAMA_API_KEY", DUMMY_API_KEY)
else:
self.api_base = self.config.api_base
# If api_base is unset we use OpenAI's endpoint, which supports
# these features (with JSON schema restricted to a limited set of models)
self.supports_strict_tools = self.api_base is None
self.supports_json_schema = (
self.api_base is None and self.info().has_structured_output
)
if settings.chat_model != "":
# if we're overriding chat model globally, set completion model to same
self.config.completion_model = self.config.chat_model
if self.config.formatter is not None:
# we want to format chats -> completions using this specific formatter
self.config.use_completion_for_chat = True
self.config.completion_model = self.config.chat_model
if self.config.use_completion_for_chat:
self.config.use_chat_for_completion = False
self.is_groq = self.config.chat_model.startswith("groq/")
self.is_cerebras = self.config.chat_model.startswith("cerebras/")
self.is_gemini = self.is_gemini_model()
self.is_deepseek = self.is_deepseek_model()
self.is_minimax = self.is_minimax_model()
self.is_glhf = self.config.chat_model.startswith("glhf/")
self.is_openrouter = self.config.chat_model.startswith("openrouter/")
self.is_langdb = self.config.chat_model.startswith("langdb/")
self.is_portkey = self.config.chat_model.startswith("portkey/")
self.is_vertexai = self.config.chat_model.startswith(VERTEXAI_PREFIX)
self.is_litellm_proxy = self.config.chat_model.startswith("litellm-proxy/")
if self.config.api_key_provider is not None and (
self.is_groq or self.is_cerebras or self.config.litellm
):
raise ValueError(
"api_key_provider is only supported for models served via an "
"OpenAI-compatible endpoint (using the OpenAI client); it "
"cannot be used with Groq, Cerebras, or the litellm adapter."
)
if self.is_groq:
# use groq-specific client
self.config.chat_model = self.config.chat_model.replace("groq/", "")
if self.api_key == OPENAI_API_KEY:
self.api_key = os.getenv("GROQ_API_KEY", DUMMY_API_KEY)
if self.config.use_cached_client:
self.client = get_groq_client(api_key=self.api_key)
self.async_client = get_async_groq_client(api_key=self.api_key)
else:
# Create new clients without caching
self.client = Groq(api_key=self.api_key)
self.async_client = AsyncGroq(api_key=self.api_key)
elif self.is_cerebras:
# use cerebras-specific client
self.config.chat_model = self.config.chat_model.replace("cerebras/", "")
if self.api_key == OPENAI_API_KEY:
self.api_key = os.getenv("CEREBRAS_API_KEY", DUMMY_API_KEY)
if self.config.use_cached_client:
self.client = get_cerebras_client(api_key=self.api_key)
# TODO there is not async client, so should we do anything here?
self.async_client = get_async_cerebras_client(api_key=self.api_key)
else:
# Create new clients without caching
self.client = Cerebras(api_key=self.api_key)
self.async_client = AsyncCerebras(api_key=self.api_key)
else:
# in these cases, there's no specific client: OpenAI python client suffices
#
# vertexai/ is tested FIRST, deliberately: the other branches key
# off `chat_model_orig`, and if one of them claimed a vertexai/
# model the route would silently become that provider's -- a public
# endpoint, with no ADC credential and no regional URL.
if self.is_vertexai:
# self.config was rebuilt as a VertexAIConfig above, so this
# branch cannot see an OPENAI_-prefixed api_key/headers.
vertex_config = cast(VertexAIConfig, self.config)
# "vertexai/google/gemini-2.5-flash" -> "google/gemini-2.5-flash",
# which is the model id the endpoint expects.
vertex_model = vertex_config.chat_model[len(VERTEXAI_PREFIX) :]
publisher, _, bare_model = vertex_model.partition("/")
if not publisher or not bare_model:
raise ValueError(
f"Invalid vertexai/ model {vertex_config.chat_model!r}: "
"expected vertexai/<publisher>/<model>, e.g. "
"vertexai/google/gemini-2.5-flash."
)
vertex_config.chat_model = vertex_model
if vertex_config.completion_model.startswith(VERTEXAI_PREFIX):
vertex_config.completion_model = vertex_config.completion_model[
len(VERTEXAI_PREFIX) :
]
if vertex_config.api_base:
# An explicit api_base (VERTEXAI_API_BASE, or api_base on a
# directly-constructed VertexAIConfig) targets a private/PSC
# endpoint; honor it rather than overwriting it. The
# conversion never carries api_base over from an
# OPENAI_-prefixed config, so a value here is deliberate.
self.api_base = vertex_config.api_base
else:
vertex_project = (
vertex_config.project_id
or os.getenv("GOOGLE_CLOUD_PROJECT")
or os.getenv("GCP_PROJECT")
)
if not vertex_project:
raise ValueError(
"A GCP project is required for a vertexai/ model: "
"set VertexAIConfig(project_id=...), or "
"VERTEXAI_PROJECT_ID, GOOGLE_CLOUD_PROJECT or "
"GCP_PROJECT in the environment."
)
vertex_location = (
vertex_config.location
or os.getenv("GOOGLE_CLOUD_LOCATION")
or DEFAULT_VERTEXAI_LOCATION
)
# Re-validate: the values may have come from
# GOOGLE_CLOUD_*/GCP_* above, which the field validators
# never see, and both are interpolated into the URL below.
vertex_config.project_id = _validate_vertexai_id(
"project_id", vertex_project
)
vertex_config.location = _validate_vertexai_id(
"location", vertex_location
)
self.api_base = (
f"https://{vertex_config.location}-aiplatform."
f"googleapis.com/v1beta1"
f"/projects/{vertex_config.project_id}"
f"/locations/{vertex_config.location}/endpoints/openapi"
)
no_explicit_key = vertex_config.api_key in ("", DUMMY_API_KEY)
if vertex_config.api_key_provider is None and no_explicit_key:
# No explicit Vertex credential: fall back to Google ADC,
# resolved per request so its ~1h lifetime is handled here
# rather than by the caller.
vertex_config.api_key_provider = _adc_access_token
elif self.is_litellm_proxy:
self.config.chat_model = self.config.chat_model.replace(
"litellm-proxy/", ""
)
if self.api_key == OPENAI_API_KEY:
self.api_key = self.config.litellm_proxy.api_key or self.api_key
self.api_base = self.config.litellm_proxy.api_base or self.api_base
elif self.is_gemini:
if self.api_key == OPENAI_API_KEY:
self.api_key = os.getenv("GEMINI_API_KEY", DUMMY_API_KEY)
# Prefer caller config, then Gemini env, then the default.
gemini_api_base = os.getenv("GEMINI_API_BASE", "")
explicit_api_base = (
self.config.api_base if self.config._api_base_was_supplied else None
)
self.api_base = explicit_api_base or gemini_api_base or GEMINI_BASE_URL
if (
self.config.chat_model.startswith("gemini/")
or self.api_base == GEMINI_BASE_URL
):
self.config.chat_model = self.config.chat_model.split("/", 1)[1]
elif self.is_glhf:
self.config.chat_model = self.config.chat_model.replace("glhf/", "")
if self.api_key == OPENAI_API_KEY:
self.api_key = os.getenv("GLHF_API_KEY", DUMMY_API_KEY)
self.api_base = GLHF_BASE_URL
elif self.is_openrouter:
self.config.chat_model = self.config.chat_model.replace(
"openrouter/", ""
)
if self.api_key == OPENAI_API_KEY:
self.api_key = os.getenv("OPENROUTER_API_KEY", DUMMY_API_KEY)
self.api_base = OPENROUTER_BASE_URL
elif self.is_deepseek:
self.config.chat_model = self.config.chat_model.replace("deepseek/", "")
self.api_base = DEEPSEEK_BASE_URL
if self.api_key == OPENAI_API_KEY:
self.api_key = os.getenv("DEEPSEEK_API_KEY", DUMMY_API_KEY)
elif self.is_minimax:
self.config.chat_model = self.config.chat_model.replace("minimax/", "")
# Honor caller-supplied base URL (e.g. regional endpoints,
# proxies) instead of always forcing the default.
openai_api_base = os.getenv("OPENAI_API_BASE")
explicit_api_base = (
self.config.api_base
if self.config.api_base and self.config.api_base != openai_api_base
else None
)
self.api_base = explicit_api_base or MINIMAX_BASE_URL
if self.api_key == OPENAI_API_KEY:
# Only overwrite with MINIMAX_API_KEY when it is actually
# set, so users who intentionally put their MiniMax key in
# OPENAI_API_KEY are not silently downgraded to a dummy key.
minimax_key = os.getenv("MINIMAX_API_KEY", "")
if minimax_key:
self.api_key = minimax_key
# Recompute capabilities now that the prefix has been stripped
# and self.info() can find the model in MODEL_INFO.
self.supports_strict_tools = True
self.supports_json_schema = self.info().has_structured_output
elif self.is_langdb:
self.config.chat_model = self.config.chat_model.replace("langdb/", "")
self.api_base = self.config.langdb_params.base_url
project_id = self.config.langdb_params.project_id
if project_id:
self.api_base += "/" + project_id + "/v1"
if self.api_key == OPENAI_API_KEY:
self.api_key = self.config.langdb_params.api_key or DUMMY_API_KEY
if self.config.langdb_params:
params = self.config.langdb_params
if params.project_id:
self.config.headers["x-project-id"] = params.project_id
if params.label:
self.config.headers["x-label"] = params.label
if params.run_id:
self.config.headers["x-run-id"] = params.run_id
if params.thread_id:
self.config.headers["x-thread-id"] = params.thread_id
elif self.is_portkey:
# Parse the model string and extract provider/model
provider, model = self.config.portkey_params.parse_model_string(
self.config.chat_model
)
self.config.chat_model = model
if provider:
self.config.portkey_params.provider = provider
# Set Portkey base URL
self.api_base = self.config.portkey_params.base_url + "/v1"
# Set API key - use provider's API key from env if available
if self.api_key == OPENAI_API_KEY:
self.api_key = self.config.portkey_params.get_provider_api_key(
self.config.portkey_params.provider, DUMMY_API_KEY
)
# Add Portkey-specific headers
self.config.headers.update(self.config.portkey_params.get_headers())
# Sanitize the API key: strip leading/trailing whitespace
# (including stray newlines from .env files or CI secrets).
self.api_key = self.api_key.strip()
# Create http_client if needed - Priority order:
# 1. http_client_factory (most flexibility, not cacheable)
# 2. http_client_config (cacheable, moderate flexibility)
# 3. http_verify_ssl=False (cacheable, simple SSL bypass)
http_client = None
async_http_client = None
http_client_config_used = None
if self.config.http_client_factory is not None:
# Use the factory to create http_client (not cacheable)
http_client = self.config.http_client_factory()
if isinstance(http_client, (list, tuple)):
if len(http_client) != 2:
raise ValueError(
"http_client_factory must return either a single "
"httpx.Client or a tuple of "
"(httpx.Client, httpx.AsyncClient)"
)
http_client, async_http_client = http_client
else:
# set async_http_client to None - so that it will
# be created later
async_http_client = None
elif self.config.http_client_config is not None:
# Use config dict (cacheable)
http_client_config_used = self.config.http_client_config
elif not self.config.http_verify_ssl:
# Simple SSL bypass (cacheable)
http_client_config_used = {"verify": False}
logging.warning(
"SSL verification has been disabled. This is insecure and "
"should only be used in trusted environments (e.g., "
"corporate networks with self-signed certificates)."
)
# With an api_key_provider, hand the callable itself to the
# OpenAI client, which resolves a fresh token on each request.
openai_api_key: Union[str, Callable[[], str]] = (
self.config.api_key_provider
if self.config.api_key_provider is not None
else self.api_key
)
# On a vertexai/ route, hide the SDK's own OPENAI_* channels
# for the duration of client construction -- see
# _without_openai_sdk_env. A nullcontext elsewhere, so no
# other route changes behaviour.
with _without_openai_sdk_env() if self.is_vertexai else nullcontext():
if self.config.use_cached_client:
self.client = get_openai_client(
api_key=openai_api_key,
base_url=self.api_base,
organization=self.config.organization,
timeout=Timeout(self.config.timeout),
default_headers=self.config.headers,
http_client=http_client,
http_client_config=http_client_config_used,
sdk_env_scrubbed=self.is_vertexai,
)
self.async_client = get_async_openai_client(
api_key=openai_api_key,
base_url=self.api_base,
organization=self.config.organization,
timeout=Timeout(self.config.timeout),
default_headers=self.config.headers,
http_client=async_http_client,
http_client_config=http_client_config_used,
sdk_env_scrubbed=self.is_vertexai,
)
else:
# Create new clients without caching
client_kwargs: Dict[str, Any] = dict(
api_key=openai_api_key,
base_url=self.api_base,
organization=self.config.organization,
timeout=Timeout(self.config.timeout),
default_headers=self.config.headers,
)
if http_client is not None:
client_kwargs["http_client"] = http_client
elif http_client_config_used is not None:
# Create http_client from config for non-cached scenario
try:
httpx = import_httpx()
except ImportError:
raise ValueError(missing_httpx_message())
client_kwargs["http_client"] = httpx.Client(
**http_client_config_used
)
self.client = OpenAI(**client_kwargs)
async_client_kwargs: Dict[str, Any] = dict(
api_key=(
# AsyncOpenAI awaits its api_key callable
wrap_api_key_provider_async(openai_api_key)
if callable(openai_api_key)
else openai_api_key
),
base_url=self.api_base,
organization=self.config.organization,
timeout=Timeout(self.config.timeout),
default_headers=self.config.headers,
)
if async_http_client is not None:
async_client_kwargs["http_client"] = async_http_client
elif http_client_config_used is not None:
# Create async http_client from config for non-cached scenario
try:
httpx = import_httpx()
except ImportError:
raise ValueError(missing_httpx_message())
async_client_kwargs["http_client"] = httpx.AsyncClient(
**http_client_config_used
)
self.async_client = AsyncOpenAI(**async_client_kwargs)
self.cache: CacheDB | None = None
use_cache = self.config.cache_config is not None
if "redis" in settings.cache_type and use_cache:
# read and write self.config, not the `config` argument: a
# vertexai/ route replaced self.config above, and mutating the
# original would leave self.config.cache_config disagreeing with
# the cache actually built here.
if self.config.cache_config is None or not isinstance(
self.config.cache_config,
RedisCacheConfig,
):
# switch to fresh redis config if needed
self.config.cache_config = RedisCacheConfig(
fake="fake" in settings.cache_type
)
if "fake" in settings.cache_type:
# force use of fake redis if global cache_type is "fakeredis"
self.config.cache_config.fake = True
self.cache = RedisCache(self.config.cache_config)
elif settings.cache_type != "none" and use_cache:
raise ValueError(
f"Invalid cache type {settings.cache_type}. "
"Valid types are redis, fakeredis, none"
)
self.config._validate_litellm()