scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 207

Merge 2c9c4eccd3 into 4ce724e3df

Browse source
This commit is contained in:
shrey 2026-10-29 20:50:18 +00:00 • committed by GitHub
commit 61514a1704
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 993 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/__init__.py
View file

@ -42,6 +42,7 @@ from litellm._logging import (
from litellm._logging import suppress_startup_output as suppress_startup_output
import re
from litellm.constants import (
DEFAULT_MODEL_COST_MAP_URL,
DEFAULT_BATCH_SIZE,
DEFAULT_FLUSH_INTERVAL_SECONDS,
ROUTER_MAX_FALLBACKS,
@ -411,7 +412,7 @@ aclient_session: Optional[Union[httpx.AsyncClient, DefaultAsyncHttpxClient]] = N
model_fallbacks: Optional[List] = None # Deprecated for 'litellm.fallbacks'
model_cost_map_url: str = os.getenv(
"LITELLM_MODEL_COST_MAP_URL",
"https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json",
DEFAULT_MODEL_COST_MAP_URL,
)
blog_posts_url: str = os.getenv(
"LITELLM_BLOG_POSTS_URL",

3
litellm/constants.py
View file

@ -1740,6 +1740,9 @@ ALLOWED_VERTEX_AI_PASSTHROUGH_HEADERS: Final = {
PASS_THROUGH_HEADER_PREFIX: Final = "x-pass-"
TYPESAFE_DANGER_ZONE_HEADER_PREFIX: Final = "typesafe-danger-zone-"
INTERNAL_KWARG_PREFIX: Final = "_litellm_"
DEFAULT_MODEL_COST_MAP_URL: Final = (
"https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json"
)
CONTROL_OPTIONS_KEY: Final = f"{INTERNAL_KWARG_PREFIX}control"
SESSION_ID_FROM_TRACE_ID_PARAM_KEY: Final = f"{INTERNAL_KWARG_PREFIX}session_id_from_trace_id"

9
litellm/cost_calculator.py
View file

@ -25,7 +25,7 @@ from litellm.constants import (
from litellm.litellm_core_utils.core_helpers import as_str_mapping, billed_usage_cost
from litellm.litellm_core_utils.cost_discount import resolve_cost_discount
from litellm.litellm_core_utils.decisions_pricing import decisions_pricing_fields, is_decisions_call_type
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, get_hidden_params
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
StandardBuiltInToolCostTracking,
)
@ -710,6 +710,7 @@ def cost_per_token(
custom_llm_provider=custom_llm_provider,
number_of_queries=number_of_queries or 1,
optional_params=(getattr(response, "_hidden_params", None) if response else None),
response=response,
)
elif (
per_second_cost := _per_second_pricing_cost(
@ -1947,7 +1948,11 @@ def completion_cost(
model=search_model,
custom_llm_provider=custom_llm_provider,
number_of_queries=number_of_queries,
optional_params=optional_params,
response=_ATTRIBUTE_VALUE.validate_python(completion_response),
optional_params={
**(get_hidden_params(completion_response) or {}),
**(optional_params or {}),
},
)
)

144
litellm/llms/exa_ai/search/cost_calculator.py Normal file
View file

@ -0,0 +1,144 @@
from collections.abc import Mapping
from types import MappingProxyType
from typing import Annotated, Final
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError
from litellm.constants import DEFAULT_MODEL_COST_MAP_URL, INTERNAL_KWARG_PREFIX
from litellm.litellm_core_utils.get_model_cost_map import (
AliasedCostEntry,
GetModelCostMap,
get_model_cost_map_source_info,
)
from litellm.llms.base_llm.search.transformation import SearchResponse
EXA_AI_COST_DOLLARS_PARAM: Final = f"{INTERNAL_KWARG_PREFIX}exa_ai_cost_dollars"
_PRICE_FIELDS: Final = frozenset({"input_cost_per_query", "tiered_pricing"})
_ENTRY: Final = TypeAdapter(Mapping[str, object])
EXA_AI_CATALOG_PRICES: Final[Mapping[str, float]] = MappingProxyType(
{
"exa_ai/search": 0.007,
"exa_ai/search-lite": 0.004,
"exa_ai/search-instant": 0.004,
"exa_ai/search-deep": 0.012,
"exa_ai/search-deep-lite": 0.012,
"exa_ai/search-deep-reasoning": 0.015,
}
)
class _ExaCostDollars(BaseModel):
model_config = ConfigDict(frozen=True)
total: Annotated[float, Field(strict=True, ge=0, allow_inf_nan=False)]
class ExaAITier(BaseModel):
model_config = ConfigDict(frozen=True)
input_cost_per_query: float
max_results_range: tuple[int, int]
_TIERS: Final = TypeAdapter(tuple[ExaAITier, ...])
EXA_AI_LEGACY_TIERS: Final = (
ExaAITier(input_cost_per_query=0.005, max_results_range=(0, 25)),
ExaAITier(input_cost_per_query=0.025, max_results_range=(26, 100)),
)
"""The tiers LiteLLM ships on `exa_ai/search`. Only older releases bill them, so they never change."""
def exa_ai_reported_cost(params: Mapping[str, object], response: object = None) -> float | None:
"""The `costDollars.total` Exa reported for the request, when it is a finite non-negative number."""
try:
payload: Final = response.model_extra if isinstance(response, SearchResponse) else response
metadata: Final = _ENTRY.validate_python(payload or {}).get("provider_metadata")
receipt: Final = _ENTRY.validate_python(metadata or {}).get("costDollars")
return _ExaCostDollars.model_validate(
receipt if receipt is not None else params.get(EXA_AI_COST_DOLLARS_PARAM)
).total
except ValidationError:
return None
def _legacy_tiers(tiers: object) -> bool:
try:
return _TIERS.validate_python(tiers) == EXA_AI_LEGACY_TIERS
except ValidationError:
return False
def exa_ai_price_fields(model: str, entry: object) -> frozenset[str]:
if not isinstance(entry, Mapping):
return frozenset()
fields: Final = _ENTRY.validate_python(entry)
registered: Final[frozenset[str]] = (
entry.registered_fields & _PRICE_FIELDS if isinstance(entry, AliasedCostEntry) else frozenset()
)
if registered:
return registered
source: Final = get_model_cost_map_source_info()
if source["source"] == "remote" and source["url"] == DEFAULT_MODEL_COST_MAP_URL:
loaded: Final = GetModelCostMap.loaded_model_cost_map().get(model, {})
return frozenset(
name for name in _PRICE_FIELDS if fields.get(name) is not None and fields[name] != loaded.get(name)
)
return frozenset(
name
for name in _PRICE_FIELDS
if fields.get(name) is not None
and (
fields[name] != EXA_AI_CATALOG_PRICES.get(model)
if name == "input_cost_per_query"
else not _legacy_tiers(fields[name])
)
)
EXA_AI_INCLUDED_RESULTS: Final = 10
EXA_AI_ADDITIONAL_RESULT_COST: Final = 0.001
EXA_AI_SUMMARY_COST_PER_PAGE: Final = 0.001
EXA_AI_TYPES_WITHOUT_SUMMARIES: Final = frozenset({"lite"})
def _non_negative_int(value: object) -> int | None:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
return None
return value
def exa_ai_requested_results(optional_params: Mapping[str, object]) -> int:
return next(
(
requested
for key in ("max_results", "numResults")
if (requested := _non_negative_int(optional_params.get(key))) is not None
),
EXA_AI_INCLUDED_RESULTS,
)
def _requests_summaries(optional_params: Mapping[str, object]) -> bool:
if optional_params.get("type") in EXA_AI_TYPES_WITHOUT_SUMMARIES:
return False
contents: Final = optional_params.get("contents")
return isinstance(contents, Mapping) and _ENTRY.validate_python(contents).get("summary") not in (None, False)
def exa_ai_list_price(base_price: float, optional_params: Mapping[str, object]) -> float:
"""
List price of one Exa /search request, for responses that carry no `costDollars.total`.
Multi-query searches are joined into a single request, so this prices one request:
the search type's base price, a per-result charge above the results that price includes,
and a per-page charge for each requested result when summaries are requested on a type that offers them.
"""
results: Final = exa_ai_requested_results(optional_params)
additional_results: Final = max(results - EXA_AI_INCLUDED_RESULTS, 0)
summarized_pages: Final = results if _requests_summaries(optional_params) else 0
return (
base_price
+ additional_results * EXA_AI_ADDITIONAL_RESULT_COST
+ summarized_pages * EXA_AI_SUMMARY_COST_PER_PAGE
)

160
litellm/llms/exa_ai/search/transformation.py
View file

@ -5,11 +5,12 @@ Exa AI API Reference: https://docs.exa.ai/reference/search
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.hidden_params import set_hidden_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.base_llm.search.transformation import (
@ -17,41 +18,16 @@ from litellm.llms.base_llm.search.transformation import (
SearchResponse,
SearchResult,
)
from litellm.llms.exa_ai.search.cost_calculator import EXA_AI_COST_DOLLARS_PARAM
from litellm.llms.exa_ai.search.types import ExaAISearchRequest
from litellm.secret_managers.main import get_secret_str
__all__ = ["ExaAISearchConfig", "ExaAISearchRequest"]
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
_STRS: Final = TypeAdapter(Iterable[str], config=ConfigDict(hide_input_in_errors=True))
_RESULT_EXTRAS: Final = (("highlights", "highlights"), ("highlightScores", "highlight_scores"), ("score", "score"))
class _ExaAISearchRequestRequired(TypedDict):
"""Required fields for Exa AI Search API request."""
query: str # Required - search query
class ExaAISearchRequest(_ExaAISearchRequestRequired, total=False):
"""
Exa AI Search API request format.
Based on: https://docs.exa.ai/reference/search
"""
type: str # Optional - search type ('keyword', 'neural', 'fast', 'auto'), default 'auto'
category: str # Optional - data category ('company', 'research paper', 'news', 'pdf', 'github', 'tweet', 'personal site', 'linkedin profile', 'financial report')
userLocation: str # Optional - two-letter ISO country code
numResults: int # Optional - number of results (max 100), default 10
includeDomains: list[str] # Optional - list of domains to include
excludeDomains: list[str] # Optional - list of domains to exclude
startCrawlDate: str # Optional - crawl date filter (ISO 8601 format)
endCrawlDate: str # Optional - crawl date filter (ISO 8601 format)
startPublishedDate: str # Optional - published date filter (ISO 8601 format)
endPublishedDate: str # Optional - published date filter (ISO 8601 format)
includeText: list[str] # Optional - strings that must be present in webpage text
excludeText: list[str] # Optional - strings that must not be present in webpage text
context: bool | dict # Optional - format results for LLMs
moderation: bool # Optional - enable content moderation, default false
contents: dict # Optional - content retrieval options
_RENAMED_RESULT_FIELDS: Final = frozenset({"text", "publishedDate", "highlightScores"})
class ExaAISearchConfig(BaseSearchConfig):
@ -99,9 +75,9 @@ class ExaAISearchConfig(BaseSearchConfig):
def transform_search_request(
self,
query: str | list[str],
optional_params: dict,
**kwargs,
) -> dict:
optional_params: Mapping[str, object],
**kwargs: object,
) -> dict[str, object]:
"""
Transform Search request to Exa AI API format.
@ -121,44 +97,32 @@ class ExaAISearchConfig(BaseSearchConfig):
Returns:
Dict with typed request data following ExaAISearchRequest spec
"""
if isinstance(query, list):
# Exa AI only supports single string queries, join with spaces
query = " ".join(query)
request_data: Final[ExaAISearchRequest] = {
"query": query,
mapped: Final = {
target: optional_params[source]
for source, target in (
("max_results", "numResults"),
("search_domain_filter", "includeDomains"),
("country", "userLocation"),
)
if source in optional_params
}
provider_params: Final = {
key: value
for key, value in optional_params.items()
if key not in self.get_supported_perplexity_optional_params() and key not in mapped
}
return {
"query": " ".join(query) if isinstance(query, list) else query,
**mapped,
**provider_params,
**({"contents": {"text": True}} if "contents" not in provider_params else {}),
}
# Transform Perplexity unified spec parameters to Exa format
if "max_results" in optional_params:
request_data["numResults"] = optional_params["max_results"]
if "search_domain_filter" in optional_params:
request_data["includeDomains"] = optional_params["search_domain_filter"]
if "country" in optional_params:
request_data["userLocation"] = optional_params["country"]
# Convert to dict before dynamic key assignments
result_data: Final = dict(request_data)
# pass through all other parameters as-is
for param, value in optional_params.items():
if param not in self.get_supported_perplexity_optional_params() and param not in result_data:
result_data[param] = value
# By default, request text content if not explicitly specified
# Exa AI doesn't return content/text unless explicitly requested
if "contents" not in result_data:
result_data["contents"] = {"text": True}
return result_data
def transform_search_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
**kwargs,
**kwargs: object,
) -> SearchResponse:
"""
Transform Exa AI API response to LiteLLM unified SearchResponse format.
@ -171,9 +135,11 @@ class ExaAISearchConfig(BaseSearchConfig):
- results[].summary → SearchResult.snippet (fallback when "text" and "highlights" are absent)
- results[].publishedDate → SearchResult.date
- No last_updated field in Exa AI response (set to None)
- results[].highlights / highlightScores / score → SearchResult.highlights / highlight_scores / score,
each set only when Exa returned it
- results[].highlightScores → SearchResult.highlight_scores, set only when Exa returned it
- every other results[] field Exa returned (id, author, image, summary, highlights, score, ...)
→ the SearchResult as is
- every top-level key except results (requestId, costDollars, ...) → SearchResponse.provider_metadata
- costDollars → the response's hidden params, where search cost tracking reads Exa's reported total
Args:
raw_response: Raw httpx response from Exa AI API
@ -183,30 +149,44 @@ class ExaAISearchConfig(BaseSearchConfig):
SearchResponse with standardized format
"""
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
for raw_result in _JSON_VALUES.validate_python(response_json.get("results", [])):
result = _JSON_OBJECT.validate_python(raw_result)
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": result.get("text")
or "\n\n".join(_STRS.validate_python(result.get("highlights") or []))
or result.get("summary")
or "",
"date": result.get("publishedDate"), # ISO 8601 datetime string
"last_updated": None, # Exa AI doesn't provide last_updated in response
**{field: result[key] for key, field in _RESULT_EXTRAS if key in result},
}
)
results.append(search_result)
return SearchResponse.model_validate(
response: Final = SearchResponse.model_validate(
{
"results": results,
"results": [
_transform_result(value)
for value in _JSON_VALUES.validate_python(response_json.get("results") or [])
],
"object": "search",
"provider_metadata": {key: value for key, value in response_json.items() if key != "results"},
}
)
params: Final = _JSON_OBJECT.validate_python(kwargs.get("optional_params") or {})
set_hidden_params(
response,
{
**{
key: value
for key, value in params.items()
if key in ("type", "max_results", "numResults", "contents")
},
EXA_AI_COST_DOLLARS_PARAM: response_json.get("costDollars"),
},
)
return response
def _transform_result(raw_result: object) -> SearchResult:
result: Final = _JSON_OBJECT.validate_python(raw_result)
return SearchResult.model_validate(
{
**{key: value for key, value in result.items() if key not in _RENAMED_RESULT_FIELDS},
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": result.get("text")
or "\n\n".join(_STRS.validate_python(result.get("highlights") or []))
or result.get("summary")
or "",
"date": result.get("publishedDate"),
"last_updated": None,
**({"highlight_scores": result["highlightScores"]} if "highlightScores" in result else {}),
}
)

99
litellm/llms/exa_ai/search/types.py Normal file
View file

@ -0,0 +1,99 @@
"""Exa /search request contract, https://exa.ai/docs/reference/search."""
from collections.abc import Mapping, Sequence
from typing import Literal, TypeAlias
from typing_extensions import ReadOnly, Required, TypedDict
ExaSearchMode: TypeAlias = Literal["lite", "instant", "fast", "auto", "deep-lite", "deep", "deep-reasoning"]
ExaPageSection: TypeAlias = Literal["header", "navigation", "banner", "body", "sidebar", "footer", "metadata"]
class ExaTextOptions(TypedDict, total=False):
maxCharacters: ReadOnly[int | None]
includeHtmlTags: ReadOnly[bool | None]
verbosity: ReadOnly[Literal["compact", "standard", "full"] | None]
includeSections: ReadOnly[Sequence[ExaPageSection] | None]
excludeSections: ReadOnly[Sequence[ExaPageSection] | None]
class ExaHighlightsOptions(TypedDict, total=False):
"""numSentences and highlightsPerUrl are deprecated compatibility fields."""
query: ReadOnly[str | None]
verbosity: ReadOnly[Literal["low", "medium", "high"] | None]
dynamic: ReadOnly[bool | None]
maxCharacters: ReadOnly[int | None]
numSentences: ReadOnly[int | None]
highlightsPerUrl: ReadOnly[int | None]
class ExaSummaryOptions(TypedDict, total=False):
query: ReadOnly[str | None]
schema: ReadOnly[Mapping[str, object] | None]
class ExaExtrasOptions(TypedDict, total=False):
links: ReadOnly[int | None]
imageLinks: ReadOnly[int | None]
richImageLinks: ReadOnly[int | None]
richLinks: ReadOnly[int | None]
codeBlocks: ReadOnly[int | None]
class ExaContextOptions(TypedDict, total=False):
maxCharacters: ReadOnly[int]
class ExaContentsOptions(TypedDict, total=False):
"""context, livecrawl, and crawledBeforeDate are deprecated compatibility fields."""
text: ReadOnly[bool | ExaTextOptions | None]
highlights: ReadOnly[bool | ExaHighlightsOptions | None]
summary: ReadOnly[ExaSummaryOptions | None]
extras: ReadOnly[ExaExtrasOptions | None]
livecrawlTimeout: ReadOnly[int | None]
maxAgeHours: ReadOnly[int | None]
snapshotAsOf: ReadOnly[str | None]
subpages: ReadOnly[int | None]
subpageTarget: ReadOnly[str | Sequence[str] | None]
context: ReadOnly[bool | ExaContextOptions | None]
livecrawl: ReadOnly[Literal["never", "always", "fallback", "preferred"] | None]
crawledBeforeDate: ReadOnly[str | None]
class ExaCountryLocation(TypedDict):
country: ReadOnly[str]
class ExaCoordinateLocation(TypedDict, total=False):
country: ReadOnly[str]
latitude: Required[ReadOnly[float]]
longitude: Required[ReadOnly[float]]
class ExaAISearchRequest(TypedDict, total=False):
"""Crawl-date filters, includeText, excludeText, and context are deprecated."""
query: Required[ReadOnly[str]]
type: ReadOnly[ExaSearchMode | None]
category: ReadOnly[str | None]
additionalQueries: ReadOnly[Sequence[str] | None]
userLocation: ReadOnly[str | ExaCountryLocation | ExaCoordinateLocation | None]
numResults: ReadOnly[int | None]
includeDomains: ReadOnly[Sequence[str] | None]
excludeDomains: ReadOnly[Sequence[str] | None]
startPublishedDate: ReadOnly[str | None]
endPublishedDate: ReadOnly[str | None]
moderation: ReadOnly[bool | None]
contents: ReadOnly[ExaContentsOptions | None]
outputSchema: ReadOnly[Mapping[str, object] | None]
objective: ReadOnly[str | None]
systemPrompt: ReadOnly[str | None]
userIdentifier: ReadOnly[str | None]
compliance: ReadOnly[Literal["hipaa"] | None]
startCrawlDate: ReadOnly[str | None]
endCrawlDate: ReadOnly[str | None]
includeText: ReadOnly[Sequence[str] | None]
excludeText: ReadOnly[Sequence[str] | None]
context: ReadOnly[bool | ExaContextOptions | None]

44
litellm/model_prices_and_context_window_backup.json
View file

@ -24387,7 +24387,11 @@
"output_vector_size": 2560
},
"exa_ai/search": {
"input_cost_per_query": 0.007,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request, covering up to 10 results. Each result above 10 adds $0.001, and each summarized page adds $0.001. Exa's reported costDollars.total takes precedence when present, unless an operator priced this entry. tiered_pricing keeps what older releases bill for this entry. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search",
"tiered_pricing": [
{
@ -24406,6 +24410,46 @@
}
]
},
"exa_ai/search-lite": {
"input_cost_per_query": 0.004,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type lite, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-22)"
},
"mode": "search"
},
"exa_ai/search-instant": {
"input_cost_per_query": 0.004,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type instant, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search"
},
"exa_ai/search-deep": {
"input_cost_per_query": 0.012,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type deep, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search"
},
"exa_ai/search-deep-lite": {
"input_cost_per_query": 0.012,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type deep-lite, covering up to 10 results. Source: https://exa.ai/docs/reference/pricing (2026-10-27)"
},
"mode": "search"
},
"exa_ai/search-deep-reasoning": {
"input_cost_per_query": 0.015,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type deep-reasoning, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search"
},
"firecrawl/search": {
"litellm_provider": "firecrawl",
"mode": "search",

63
litellm/search/cost_calculator.py
View file

@ -10,6 +10,12 @@ from pydantic import TypeAdapter, ValidationError
import litellm
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.llms.exa_ai.search.cost_calculator import (
exa_ai_list_price,
exa_ai_price_fields,
exa_ai_reported_cost,
exa_ai_requested_results,
)
from litellm.utils import get_model_info
PROVIDER_USAGE_ADAPTER: Final[TypeAdapter[tuple[Mapping[str, object], ...]]] = TypeAdapter(
@ -17,7 +23,7 @@ PROVIDER_USAGE_ADAPTER: Final[TypeAdapter[tuple[Mapping[str, object], ...]]] = T
)
EMPTY_OPTIONAL_PARAMS: Final[Mapping[str, object]] = MappingProxyType({})
SEARCH_MODE_PARAM_BY_PROVIDER: Final[Mapping[str, str]] = MappingProxyType(
{"tavily": "search_depth", "linkup": "depth"}
{"tavily": "search_depth", "linkup": "depth", "exa_ai": "type"}
)
@ -33,10 +39,14 @@ def _provider_usage(
return None
def _registered_at_runtime(entry: str) -> bool:
"""Whether `entry`'s live price differs from the catalog as loaded, i.e. `register_model` changed it."""
live: Final[object] = litellm.model_cost.get(entry) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # registry is dynamically typed
return live != GetModelCostMap.loaded_model_cost_map().get(entry) # pyright: ignore[reportUnknownVariableType] # `live` comes from the untyped registry
def _live_entry(entry: str) -> object:
return litellm.model_cost.get(entry) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # registry is dynamically typed
def _operator_priced(model: str, provider: str | None) -> bool:
if provider == "exa_ai":
return bool(exa_ai_price_fields(model, _live_entry(model)))
return _live_entry(model) != GetModelCostMap.loaded_model_cost_map().get(model)
def _pricing_model(
@ -51,10 +61,14 @@ def _pricing_model(
generic_model: Final = f"{custom_llm_provider}/search"
if mode_param is None or model not in (generic_model, f"{custom_llm_provider}/{generic_model}"):
return model
mode: Final = (optional_params if optional_params is not None else EMPTY_OPTIONAL_PARAMS).get(mode_param)
mode: Final = (optional_params if optional_params is not None else EMPTY_OPTIONAL_PARAMS).get(
mode_param, "auto" if custom_llm_provider == "exa_ai" else None
)
mode_model: Final = f"{custom_llm_provider}/search-{mode}"
priced: Final = mode_model in litellm.model_cost # pyright: ignore[reportUnknownMemberType] # registry is dynamically typed
if not priced or (_registered_at_runtime(generic_model) and not _registered_at_runtime(mode_model)):
if not priced or (
_operator_priced(generic_model, custom_llm_provider) and not _operator_priced(mode_model, custom_llm_provider)
):
return model
return mode_model
@ -64,19 +78,22 @@ def search_provider_cost_per_query(
custom_llm_provider: str | None = None,
number_of_queries: int = 1,
optional_params: Mapping[str, object] | None = None,
response: object = None,
) -> tuple[float, float]:
"""
Calculate cost for search-only providers.
Returns (input_cost, output_cost) where input_cost = queries * cost_per_query
Supports tiered pricing based on max_results parameter, and per-mode price entries selected by the
provider's mode parameter (Tavily `search_depth`, Linkup `depth`).
provider's mode parameter (Tavily `search_depth`, Linkup `depth`, Exa `type`). An Exa request uses an operator's
price, otherwise its reported total, otherwise estimated list price.
Args:
model: Model name (e.g., "exa_ai/search", "tavily/search")
custom_llm_provider: Provider name (e.g., "exa_ai", "tavily")
number_of_queries: Number of search queries performed (default: 1)
optional_params: Optional parameters including max_results for tiered pricing
response: Search response carrying Exa's reported cost
Returns:
Tuple of (input_cost, output_cost) where output_cost is always 0.0
@ -93,15 +110,35 @@ def search_provider_cost_per_query(
)
return (input_cost, 0.0)
model_info: Final = get_model_info(
model=_pricing_model(model, custom_llm_provider, optional_params),
custom_llm_provider=custom_llm_provider,
pricing_model: Final = _pricing_model(model, custom_llm_provider, optional_params)
exa_model: Final = (
pricing_model.removeprefix("exa_ai/") if pricing_model == "exa_ai/exa_ai/search" else pricing_model
)
params: Final = optional_params if optional_params is not None else EMPTY_OPTIONAL_PARAMS
exa_price_fields: Final[frozenset[str]] = (
exa_ai_price_fields(exa_model, _live_entry(exa_model)) if custom_llm_provider == "exa_ai" else frozenset()
)
reported_cost: Final = (
exa_ai_reported_cost(params, response) if custom_llm_provider == "exa_ai" and not exa_price_fields else None
)
if reported_cost is not None:
return (reported_cost, 0.0)
model_info: Final = get_model_info(model=pricing_model, custom_llm_provider=custom_llm_provider)
if custom_llm_provider == "exa_ai" and not exa_price_fields:
return (exa_ai_list_price(float(model_info.get("input_cost_per_query") or 0.0), params), 0.0)
# Check for tiered pricing (e.g., Exa AI based on max_results)
tiered_pricing: Final = model_info.get("tiered_pricing")
if tiered_pricing and isinstance(tiered_pricing, list):
max_results: Final = (optional_params or {}).get("max_results", 10) # default 10 results
use_tiers: Final = (
"tiered_pricing" in exa_price_fields and "input_cost_per_query" not in exa_price_fields
if custom_llm_provider == "exa_ai"
else model_info.get("input_cost_per_query") is None
)
if tiered_pricing and isinstance(tiered_pricing, list) and use_tiers:
max_results: Final = (
exa_ai_requested_results(params) if custom_llm_provider == "exa_ai" else params.get("max_results", 10)
)
cost_per_query = 0.0
for tier in tiered_pricing:

8
litellm/utils.py
View file

@ -3776,9 +3776,13 @@ def _replace_cost_map_entry(
previous_fields: Final[frozenset[str]] = (
current.registered_fields if isinstance(current, AliasedCostEntry) else frozenset()
)
registered_fields: Final = previous_fields | frozenset(
name for name, value in registration.items() if value is not None
supplied_fields: Final = frozenset(name for name, value in registration.items() if value is not None)
replaced_fields: Final[frozenset[str]] = (
frozenset({"input_cost_per_query", "tiered_pricing"})
if merged.get("litellm_provider") == "exa_ai" and supplied_fields & {"input_cost_per_query", "tiered_pricing"}
else frozenset()
)
registered_fields: Final = (previous_fields - replaced_fields) | supplied_fields
sharing_names: Final = tuple(name for name in names if cost_map.get(name) is current)
cost_map.update(dict.fromkeys((key, *sharing_names), AliasedCostEntry(merged, names, registered_fields)))

44
model_prices_and_context_window.json
View file

@ -24387,7 +24387,11 @@
"output_vector_size": 2560
},
"exa_ai/search": {
"input_cost_per_query": 0.007,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request, covering up to 10 results. Each result above 10 adds $0.001, and each summarized page adds $0.001. Exa's reported costDollars.total takes precedence when present, unless an operator priced this entry. tiered_pricing keeps what older releases bill for this entry. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search",
"tiered_pricing": [
{
@ -24406,6 +24410,46 @@
}
]
},
"exa_ai/search-lite": {
"input_cost_per_query": 0.004,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type lite, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-22)"
},
"mode": "search"
},
"exa_ai/search-instant": {
"input_cost_per_query": 0.004,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type instant, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search"
},
"exa_ai/search-deep": {
"input_cost_per_query": 0.012,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type deep, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search"
},
"exa_ai/search-deep-lite": {
"input_cost_per_query": 0.012,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type deep-lite, covering up to 10 results. Source: https://exa.ai/docs/reference/pricing (2026-10-27)"
},
"mode": "search"
},
"exa_ai/search-deep-reasoning": {
"input_cost_per_query": 0.015,
"litellm_provider": "exa_ai",
"metadata": {
"notes": "Base price per /search request with type deep-reasoning, covering up to 10 results. Source: https://exa.ai/pricing (2026-10-15)"
},
"mode": "search"
},
"firecrawl/search": {
"litellm_provider": "firecrawl",
"mode": "search",

117
tests/integration/sdk/test_exa_search_envelope_sdk.py
View file

@ -1,5 +1,8 @@
import asyncio
import json
import uuid
import threading
from collections.abc import Mapping
from datetime import datetime
from typing import Final
import pytest
@ -7,15 +10,18 @@ from integration._support.wire import Reply, Request, wire_server
from pydantic import JsonValue
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.llms.base_llm.search.transformation import SearchResponse
pytestmark: Final = pytest.mark.timeout(60)
_REPORTED_COST: Final = 0.005
_ENVELOPE: Final[dict[str, JsonValue]] = {
"requestId": "exa-request-sdk",
"output": {"content": {"answer": "verified"}, "grounding": []},
"resolvedSearchType": "neural",
"searchTime": 211.0,
"costDollars": {"total": 0.005, "search": {"neural": 0.005}},
"costDollars": {"total": _REPORTED_COST, "search": {"neural": _REPORTED_COST}},
}
_RESULT: Final[dict[str, JsonValue]] = {
"id": "https://result.invalid/sdk",
@ -23,6 +29,7 @@ _RESULT: Final[dict[str, JsonValue]] = {
"url": "https://result.invalid/sdk",
"publishedDate": "2026-02-01T00:00:00.000Z",
"score": 0.73,
"author": None,
"highlights": ["First line.", "Second line."],
"highlightScores": [0.61, 0.42],
}
@ -30,6 +37,7 @@ _EXPECTED: Final[dict[str, JsonValue]] = {
"object": "search",
"results": [
{
"id": "https://result.invalid/sdk",
"title": "SDK result",
"url": "https://result.invalid/sdk",
"snippet": "First line.\n\nSecond line.",
@ -38,6 +46,7 @@ _EXPECTED: Final[dict[str, JsonValue]] = {
"highlights": ["First line.", "Second line."],
"highlight_scores": [0.61, 0.42],
"score": 0.73,
"author": None,
}
],
"provider_metadata": _ENVELOPE,
@ -49,34 +58,84 @@ def _respond(request: Request) -> Reply:
return Reply(body=json.dumps({**_ENVELOPE, "results": [_RESULT]}).encode())
def test_sync_exa_search_returns_highlights_scores_and_the_response_envelope() -> None:
with wire_server(_respond) as wire:
response: Final = litellm.search(
query=f"sdk sync {uuid.uuid4().hex}",
search_provider="exa_ai",
api_key="synthetic-exa-key",
api_base=f"{wire.url}/exa",
max_results=1,
contents={"highlights": {"numSentences": 2}},
)
assert len(wire.drain()) == 1
assert isinstance(response, SearchResponse), type(response)
assert response.model_dump(mode="json") == _EXPECTED
PARAMS: Final = {
"contents": {"highlights": {"dynamic": True, "verbosity": "low"}},
"outputSchema": {"type": "object", "properties": {"answer": {"type": "string"}}},
"systemPrompt": "Use evidence",
"type": "auto",
}
@pytest.mark.asyncio
async def test_async_exa_search_returns_highlights_scores_and_the_response_envelope() -> None:
with wire_server(_respond) as wire:
response: Final = await litellm.asearch(
query=f"sdk async {uuid.uuid4().hex}",
search_provider="exa_ai",
api_key="synthetic-exa-key",
api_base=f"{wire.url}/exa",
max_results=1,
contents={"highlights": {"numSentences": 2}},
)
assert len(wire.drain()) == 1
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("schema", [False, True])
@pytest.mark.parametrize("price_override", [None, 0.0, 0.123])
@pytest.mark.parametrize("global_callback", [False, True])
async def test_exa_search_request_response_and_logged_cost_cross_real_http(
monkeypatch: pytest.MonkeyPatch,
asynchronous: bool,
schema: bool,
price_override: float | None,
global_callback: bool,
) -> None:
class Recorder(CustomLogger):
def __init__(self) -> None:
self.finished: Final = threading.Event()
self.records: tuple[tuple[Mapping[str, object], object], ...] = ()
assert isinstance(response, SearchResponse), type(response)
assert response.model_dump(mode="json") == _EXPECTED
def log_success_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime
) -> None:
self.records = (*self.records, (kwargs, response_obj))
self.finished.set()
async def async_log_success_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime
) -> None:
self.log_success_event(kwargs, response_obj, start_time, end_time)
recorder: Final = Recorder()
if global_callback:
monkeypatch.setattr(litellm, "callbacks", [recorder])
provider_params: Final = {
**{key: value for key, value in PARAMS.items() if schema or key != "outputSchema"},
"contents": PARAMS["contents"] if schema else {"highlights": {"numSentences": 2}},
}
def respond(request: Request) -> Reply:
assert request.headers["x-api-key"] == "test-key"
assert request.headers["exa-beta"] == "dynamic-highlights-2026-08-28"
assert json.loads(request.body) == {"query": "question", "numResults": 1, **provider_params}
return _respond(request)
with wire_server(respond) as server:
params: Final = {
"query": "question",
"search_provider": "exa_ai",
"api_base": f"{server.url}/exa",
"max_results": 1,
"api_key": "test-key",
"extra_headers": {"Exa-Beta": "dynamic-highlights-2026-08-28"},
**({"input_cost_per_query": price_override} if price_override is not None else {}),
**(
{}
if global_callback
else {"success_callback": [recorder.async_log_success_event if asynchronous else recorder]}
),
**provider_params,
}
response: Final = await litellm.asearch(**params) if asynchronous else litellm.search(**params)
assert len(server.drain()) == 1
assert isinstance(response, SearchResponse)
assert response.model_dump(mode="json") == _EXPECTED
assert await asyncio.to_thread(recorder.finished.wait, 5)
assert len(recorder.records) == 1
details, logged = recorder.records[0]
assert isinstance(logged, SearchResponse)
assert logged.model_dump(mode="json") == _EXPECTED
assert isinstance(details["standard_logging_object"], Mapping)
expected_cost: Final = _REPORTED_COST if price_override is None else price_override
assert response.hidden_params["response_cost"] == expected_cost
assert details["response_cost"] == expected_cost
assert details["standard_logging_object"]["response_cost"] == expected_cost
assert details["stream"] is False

65
tests/unit/llms/exa_ai/search/test_cost_calculator.py Normal file
View file

@ -0,0 +1,65 @@
import json
from collections.abc import Mapping
from pathlib import Path
from typing import Final
import pytest
from pydantic import TypeAdapter
from litellm.llms.exa_ai.search.cost_calculator import (
EXA_AI_ADDITIONAL_RESULT_COST,
EXA_AI_CATALOG_PRICES,
EXA_AI_INCLUDED_RESULTS,
EXA_AI_LEGACY_TIERS,
EXA_AI_SUMMARY_COST_PER_PAGE,
ExaAITier,
exa_ai_list_price,
)
REPO_ROOT: Final = Path(__file__).resolve().parents[5]
SHIPPED_COST_MAPS: Final = (
"model_prices_and_context_window.json",
"litellm/model_prices_and_context_window_backup.json",
)
def _search_entries(path: str) -> dict[str, dict[str, object]]:
return {
key: entry
for key, entry in json.loads((REPO_ROOT / path).read_text()).items()
if isinstance(entry, dict) and entry.get("mode") == "search"
}
@pytest.mark.parametrize("path", SHIPPED_COST_MAPS)
def test_shipped_exa_prices_agree_with_the_receipt_override_baseline(path: str) -> None:
entries: Final = _search_entries(path)
assert {
key: entry["input_cost_per_query"] for key, entry in entries.items() if key.startswith("exa_ai/")
} == EXA_AI_CATALOG_PRICES
assert (
TypeAdapter(tuple[ExaAITier, ...]).validate_python(entries["exa_ai/search"]["tiered_pricing"])
== EXA_AI_LEGACY_TIERS
)
@pytest.mark.parametrize(
("params", "extra_results", "summaries"),
(
({}, 0, 0),
({"max_results": None}, 0, 0),
({"max_results": True, "numResults": EXA_AI_INCLUDED_RESULTS + 3}, 3, 0),
({"numResults": EXA_AI_INCLUDED_RESULTS + 3}, 3, 0),
({"max_results": EXA_AI_INCLUDED_RESULTS + 2, "numResults": 100}, 2, 0),
({"max_results": 3, "contents": {"summary": {}}}, 0, 3),
({"max_results": 3, "contents": {"summary": False}}, 0, 0),
({"type": "lite", "max_results": 3, "contents": {"summary": {}}}, 0, 0),
),
)
def test_fallback_adds_only_requested_results_and_summaries(
params: Mapping[str, object], extra_results: int, summaries: int
) -> None:
base_price: Final = 0.123
assert exa_ai_list_price(base_price, params) == pytest.approx(
base_price + extra_results * EXA_AI_ADDITIONAL_RESULT_COST + summaries * EXA_AI_SUMMARY_COST_PER_PAGE
)

120
tests/unit/llms/exa_ai/search/test_transformation.py
View file

@ -9,9 +9,12 @@ from pydantic import ValidationError
import litellm
from litellm.llms.base_llm.search.transformation import SearchResponse, SearchResult
from litellm.llms.exa_ai.search.cost_calculator import EXA_AI_COST_DOLLARS_PARAM
from litellm.llms.exa_ai.search.transformation import ExaAISearchConfig
from litellm.search.cost_calculator import search_provider_cost_per_query
EXA_SEARCH_URL: Final = "https://api.exa.ai/search"
EXA_REPORTED_COST: Final = 0.123456
@pytest.fixture
@ -125,13 +128,16 @@ def test_transform_search_response_maps_every_exa_result_in_provider_order():
assert response.results == [
SearchResult.model_validate(
{
"id": "https://example.com/first",
"title": "First",
"url": "https://example.com/first",
"snippet": "First text",
"date": "2024-01-05T00:00:00.000Z",
"author": "Ada",
"highlights": ["ignored when text is present"],
"highlight_scores": [0.5],
"score": 0.42,
"image": None,
}
),
SearchResult.model_validate(
@ -142,7 +148,9 @@ def test_transform_search_response_maps_every_exa_result_in_provider_order():
"highlights": ["one", "two"],
}
),
SearchResult(title="Third", url="https://example.com/third", snippet="Summary"),
SearchResult.model_validate(
{"title": "Third", "url": "https://example.com/third", "snippet": "Summary", "summary": "Summary"}
),
]
assert response.model_extra == {
"provider_metadata": {
@ -162,7 +170,15 @@ def test_transform_search_response_maps_every_exa_result_in_provider_order():
({"highlightScores": [0.1]}, {"highlight_scores": [0.1]}),
({"score": None}, {"score": None}),
({"highlights": [], "highlightScores": [], "score": 0}, {"highlights": [], "highlight_scores": [], "score": 0}),
({"id": "https://example.com", "author": "Ada", "image": None, "summary": "a summary"}, {}),
({"publishedDate": "2024-01-05T00:00:00.000Z"}, {}),
(
{"id": "https://example.com", "author": "Ada", "image": None, "summary": "a summary"},
{"id": "https://example.com", "author": "Ada", "image": None, "summary": "a summary"},
),
(
{"favicon": "https://example.com/favicon.ico", "subpages": []},
{"favicon": "https://example.com/favicon.ico", "subpages": []},
),
],
)
def test_transform_search_response_attaches_only_the_exa_result_fields_it_returned(
@ -232,7 +248,9 @@ def test_transform_search_response_joins_whatever_the_highlights_value_iterates_
assert response.results[0].snippet == expected_snippet
@pytest.mark.parametrize("payload", [{}, {"results": []}, {"results": ""}, {"results": {}}, {"requestId": "abc"}])
@pytest.mark.parametrize(
"payload", [{}, {"results": []}, {"results": None}, {"results": ""}, {"results": {}}, {"requestId": "abc"}]
)
def test_transform_search_response_without_results_is_empty(payload: dict[str, object]):
assert _transform(payload).results == []
@ -283,3 +301,99 @@ def test_get_complete_url_ignores_a_trailing_slash_on_api_base(api_base: str) ->
ExaAISearchConfig().get_complete_url(api_base=api_base, optional_params={})
== "https://gateway.example.com/exa/search"
)
def _catalog_price(max_results: int) -> float:
price, _ = search_provider_cost_per_query(
model="exa_ai/search", custom_llm_provider="exa_ai", optional_params={"max_results": max_results}
)
return price
@pytest.mark.usefixtures("exa_api_key")
@pytest.mark.parametrize("reported_cost", [EXA_REPORTED_COST, 0, 3])
def test_search_bills_the_cost_exa_reports(respx_mock: respx.MockRouter, reported_cost: float) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(
json={"results": [], "costDollars": {"total": reported_cost, "search": {"neural": reported_cost}}}
)
assert _catalog_price(25) != reported_cost
response: Final = litellm.search(query="asyncio TaskGroup", search_provider="exa_ai", max_results=25)
assert response.hidden_params["response_cost"] == reported_cost
@pytest.mark.usefixtures("exa_api_key")
@pytest.mark.parametrize(
"body",
[
'{"results": []}',
'{"results": [], "costDollars": {}}',
'{"results": [], "costDollars": {"total": null}}',
'{"results": [], "costDollars": {"total": -0.01}}',
'{"results": [], "costDollars": {"total": "0.022"}}',
'{"results": [], "costDollars": {"total": true}}',
'{"results": [], "costDollars": {"total": NaN}}',
'{"results": [], "costDollars": {"total": Infinity}}',
'{"results": [], "costDollars": "0.022"}',
'{"results": [], "costDollars": [0.022]}',
],
)
def test_search_without_a_usable_exa_cost_bills_the_catalog_price(respx_mock: respx.MockRouter, body: str) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(content=body, headers={"content-type": "application/json"})
response: Final = litellm.search(query="asyncio TaskGroup", search_provider="exa_ai", max_results=25)
assert response.hidden_params["response_cost"] == pytest.approx(_catalog_price(25))
@pytest.mark.usefixtures("exa_api_key")
@pytest.mark.parametrize(
("config", "expected_cost"),
[("cost_discount_config", EXA_REPORTED_COST * 0.9), ("cost_margin_config", EXA_REPORTED_COST * 1.1)],
)
def test_an_exa_discount_or_margin_applies_on_top_of_the_cost_exa_reports(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, config: str, expected_cost: float
) -> None:
monkeypatch.setattr(litellm, config, {"exa_ai": 0.1})
respx_mock.post(EXA_SEARCH_URL).respond(json={"results": [], "costDollars": {"total": EXA_REPORTED_COST}})
response: Final = litellm.search(query="asyncio TaskGroup", search_provider="exa_ai", max_results=25)
assert response.hidden_params["response_cost"] == pytest.approx(expected_cost)
@pytest.mark.usefixtures("exa_api_key")
def test_a_caller_cannot_set_the_cost_exa_reports(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post(EXA_SEARCH_URL).respond(json={"results": []})
response: Final = litellm.search(
query="asyncio TaskGroup",
search_provider="exa_ai",
max_results=25,
**{EXA_AI_COST_DOLLARS_PARAM: {"total": 0}},
)
assert EXA_AI_COST_DOLLARS_PARAM not in json.loads(route.calls.last.request.content)
assert response.hidden_params["response_cost"] == pytest.approx(_catalog_price(25))
@pytest.mark.usefixtures("exa_api_key")
@pytest.mark.parametrize("serialized", (False, True))
def test_costing_an_exa_reply_directly_returns_the_cost_exa_reports(
respx_mock: respx.MockRouter, serialized: bool
) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json={"results": [], "costDollars": {"total": EXA_REPORTED_COST}})
response: Final = litellm.search(query="asyncio TaskGroup", search_provider="exa_ai", max_results=25)
assert _catalog_price(25) != EXA_REPORTED_COST
costed_response: Final = response.model_dump() if serialized else response
completion_cost: Final = litellm.completion_cost(
completion_response=costed_response, model="exa_ai/search", custom_llm_provider="exa_ai", call_type="search"
)
per_token: Final = litellm.cost_per_token(
model="exa_ai/search", custom_llm_provider="exa_ai", call_type="search", response=costed_response
)
assert completion_cost == pytest.approx(EXA_REPORTED_COST)
assert per_token == pytest.approx((EXA_REPORTED_COST, 0.0))

246
tests/unit/search/test_main.py
View file

@ -1,6 +1,6 @@
import datetime
import json
from collections.abc import Mapping
from collections.abc import Iterator, Mapping
from typing import Final
import httpx
@ -8,6 +8,7 @@ import pytest
import respx
import litellm
from litellm.constants import DEFAULT_MODEL_COST_MAP_URL
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.search.cost_calculator import search_provider_cost_per_query
@ -15,6 +16,11 @@ from litellm.search.cost_calculator import search_provider_cost_per_query
EXA_SEARCH_URL: Final = "https://api.exa.ai/search"
TAVILY_SEARCH_URL: Final = "https://api.tavily.com/search"
LINKUP_SEARCH_URL: Final = "https://api.linkup.so/v1/search"
EXA_REPORTED_COST: Final = 0.123456
EXA_REPLY_WITH_COST: Final = {
"results": [{"title": "t", "url": "https://example.com"}],
"costDollars": {"total": EXA_REPORTED_COST},
}
SEARCH_MODE_PRICES: Final = {
"tavily/search": 0.002,
"tavily/search-advanced": 0.007,
@ -57,8 +63,13 @@ def _proxy_style_logging_obj() -> Logging:
@pytest.mark.usefixtures("_local_model_cost_map")
def test_search_bills_search_tool_input_cost_per_query(respx_mock: respx.MockRouter) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json={"results": [{"title": "t", "url": "https://example.com"}]})
@pytest.mark.parametrize("reported_cost", (None, EXA_REPORTED_COST))
def test_search_bills_search_tool_input_cost_per_query(
respx_mock: respx.MockRouter, reported_cost: float | None
) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(
json={"results": [], **({"costDollars": {"total": reported_cost}} if reported_cost is not None else {})}
)
map_price, _ = search_provider_cost_per_query(model="exa_ai/search", custom_llm_provider="exa_ai")
configured_price: Final = 0.05
assert map_price not in (0.0, configured_price)
@ -77,19 +88,29 @@ def test_search_bills_search_tool_input_cost_per_query(respx_mock: respx.MockRou
litellm_logging_obj=_proxy_style_logging_obj(),
)
assert map_priced._hidden_params["response_cost"] == pytest.approx(map_price)
assert map_priced._hidden_params["response_cost"] == pytest.approx(
map_price if reported_cost is None else reported_cost
)
assert tool_priced._hidden_params["response_cost"] == pytest.approx(configured_price)
assert respx_mock.calls.last.request.content.find(b"input_cost_per_query") == -1
@pytest.mark.usefixtures("_local_model_cost_map")
@pytest.mark.parametrize("reported_cost", (None, EXA_REPORTED_COST))
def test_unpriced_fallback_search_tool_is_billed_at_its_own_price_not_the_failed_primarys(
respx_mock: respx.MockRouter,
reported_cost: float | None,
) -> None:
respx_mock.post(EXA_SEARCH_URL).mock(
side_effect=(
httpx.Response(500, json={"error": "primary down"}),
httpx.Response(200, json={"results": [{"title": "t", "url": "https://example.com"}]}),
httpx.Response(
200,
json={
"results": [],
**({"costDollars": {"total": reported_cost}} if reported_cost is not None else {}),
},
),
)
)
map_price, _ = search_provider_cost_per_query(model="exa_ai/search", custom_llm_provider="exa_ai")
@ -112,7 +133,130 @@ def test_unpriced_fallback_search_tool_is_billed_at_its_own_price_not_the_failed
litellm_logging_obj=shared_by_every_fallback_attempt,
)
assert fallback._hidden_params["response_cost"] == pytest.approx(map_price)
assert fallback._hidden_params["response_cost"] == pytest.approx(
map_price if reported_cost is None else reported_cost
)
@pytest.mark.usefixtures("_local_model_cost_map")
@pytest.mark.parametrize("registered", (False, True))
@pytest.mark.parametrize("price_field", ("input_cost_per_query", "tiered_pricing"))
@pytest.mark.parametrize(
"exa_reply", [EXA_REPLY_WITH_COST, {"results": []}], ids=["exa-reports-a-cost", "exa-reports-no-cost"]
)
def test_an_exa_price_someone_set_overrides_the_cost_exa_reports_and_litellms_tiers(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
registered: bool,
price_field: str,
exa_reply: Mapping[str, object],
) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json=exa_reply)
tier_price, _ = search_provider_cost_per_query(
model="exa_ai/search", custom_llm_provider="exa_ai", optional_params={"max_results": 25}
)
set_price: Final = 0.003
assert set_price not in (tier_price, EXA_REPORTED_COST)
price: Final = (
set_price
if price_field == "input_cost_per_query"
else [{"input_cost_per_query": set_price, "max_results_range": [0, 100]}]
)
if registered:
litellm.register_model({"exa_ai/search": {price_field: price, "litellm_provider": "exa_ai", "mode": "search"}})
else:
monkeypatch.setitem(
litellm.model_cost, "exa_ai/search", {**litellm.model_cost["exa_ai/search"], price_field: price}
)
response: Final = litellm.search(
query="litellm release notes", search_provider="exa_ai", api_key="test-exa-key", max_results=25
)
assert response.hidden_params["response_cost"] == pytest.approx(set_price)
@pytest.mark.usefixtures("_local_model_cost_map")
def test_registering_litellms_own_exa_tiers_still_overrides_the_cost_exa_reports(respx_mock: respx.MockRouter) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json=EXA_REPLY_WITH_COST)
tier_price: Final = litellm.model_cost["exa_ai/search"]["tiered_pricing"][0]["input_cost_per_query"]
assert tier_price != EXA_REPORTED_COST
litellm.register_model(
{
"exa_ai/search": {
"tiered_pricing": litellm.model_cost["exa_ai/search"]["tiered_pricing"],
"litellm_provider": "exa_ai",
"mode": "search",
}
}
)
response: Final = litellm.search(
query="litellm release notes", search_provider="exa_ai", api_key="test-exa-key", max_results=25
)
assert response.hidden_params["response_cost"] == pytest.approx(tier_price)
@pytest.mark.usefixtures("_local_model_cost_map")
@pytest.mark.parametrize(("search_type", "expected_cost"), [("lite", 0.002), ("auto", EXA_REPORTED_COST)])
def test_a_price_set_for_one_exa_search_type_applies_only_to_that_type(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, search_type: str, expected_cost: float
) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json=EXA_REPLY_WITH_COST)
monkeypatch.setitem(
litellm.model_cost,
"exa_ai/search-lite",
{"input_cost_per_query": 0.002, "litellm_provider": "exa_ai", "mode": "search"},
)
response: Final = litellm.search(
query="litellm release notes", search_provider="exa_ai", api_key="test-exa-key", type=search_type
)
assert json.loads(respx_mock.calls.last.request.content)["type"] == search_type
assert response.hidden_params["response_cost"] == pytest.approx(expected_cost)
@pytest.mark.usefixtures("_local_model_cost_map")
def test_a_flat_price_registered_on_a_tiered_search_entry_beats_its_tiers() -> None:
tier_price, _ = search_provider_cost_per_query(
model="firecrawl/search", custom_llm_provider="firecrawl", optional_params={"max_results": 15}
)
registered_price: Final = 0.002
assert tier_price != registered_price
litellm.register_model(
{
"firecrawl/search": {
"input_cost_per_query": registered_price,
"litellm_provider": "firecrawl",
"mode": "search",
}
}
)
assert search_provider_cost_per_query(
model="firecrawl/search", custom_llm_provider="firecrawl", optional_params={"max_results": 15}
) == pytest.approx((registered_price, 0.0))
@pytest.mark.usefixtures("_local_model_cost_map")
@pytest.mark.parametrize("mode", ("auto", "deep"))
def test_the_cost_exa_reports_is_billed_when_the_cost_map_has_no_exa_entry(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, mode: str
) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json=EXA_REPLY_WITH_COST)
monkeypatch.setattr(
litellm,
"model_cost",
{key: value for key, value in litellm.model_cost.items() if not key.startswith("exa_ai/")},
)
response: Final = litellm.search(
query="litellm release notes", search_provider="exa_ai", api_key="test-exa-key", type=mode
)
assert response.hidden_params["response_cost"] == pytest.approx(EXA_REPORTED_COST)
@pytest.mark.usefixtures("_search_mode_prices")
@ -255,6 +399,96 @@ def test_a_runtime_price_on_the_mode_entry_wins_over_a_runtime_price_on_the_gene
assert cost == pytest.approx((0.011, 0.0))
@pytest.mark.usefixtures("_local_model_cost_map")
@pytest.mark.parametrize("params", ({"type": "deep"}, {}), ids=("explicit-mode", "default-mode"))
def test_exa_pricing_apis_keep_the_requested_mode(respx_mock: respx.MockRouter, params: Mapping[str, object]) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json=EXA_REPLY_WITH_COST)
mode: Final = params.get("type", "auto")
price: Final = 0.321
litellm.register_model(
{f"exa_ai/search-{mode}": {"input_cost_per_query": price, "litellm_provider": "exa_ai", "mode": "search"}},
persist_across_reloads=False,
)
response: Final = litellm.search(query="mode", search_provider="exa_ai", api_key="test-key", **params)
assert response.hidden_params["response_cost"] == pytest.approx(price)
assert sum(
litellm.cost_per_token(
model="exa_ai/search", custom_llm_provider="exa_ai", call_type="search", response=response
)
) == pytest.approx(price)
assert litellm.completion_cost(
completion_response=response, model="exa_ai/search", custom_llm_provider="exa_ai", call_type="search"
) == pytest.approx(price)
@pytest.mark.usefixtures("_local_model_cost_map")
def test_exa_native_result_count_bills_the_same_operator_tier(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post(EXA_SEARCH_URL).respond(json=EXA_REPLY_WITH_COST)
price: Final = 0.456
litellm.register_model(
{
"exa_ai/search": {
"tiered_pricing": [
{"input_cost_per_query": 0.123, "max_results_range": [0, 10]},
{"input_cost_per_query": price, "max_results_range": [11, 100]},
]
}
},
persist_across_reloads=False,
)
for params in ({"max_results": 50}, {"numResults": 50}, {"max_results": None, "numResults": 50}):
response: Final = litellm.search(query="count", search_provider="exa_ai", api_key="test-key", **params)
assert json.loads(route.calls.last.request.content)["numResults"] == 50
assert response.hidden_params["response_cost"] == pytest.approx(price)
@pytest.fixture
def _restore_catalog_source(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
yield
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=DEFAULT_MODEL_COST_MAP_URL))
@pytest.mark.usefixtures("_restore_catalog_source")
@pytest.mark.parametrize(
("source", "price", "expected_cost"),
(
(DEFAULT_MODEL_COST_MAP_URL, 0.321, EXA_REPORTED_COST),
("https://prices.invalid/map.json", 0.321, 0.321),
("https://prices.invalid/map.json", None, EXA_REPORTED_COST),
),
ids=("updated-official-catalog", "operator-map", "mirror"),
)
@pytest.mark.parametrize("mode", ("auto", "lite", "deep"))
def test_exa_catalog_source_preserves_receipts_and_generic_operator_prices(
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
source: str,
price: float | None,
expected_cost: float,
mode: str,
) -> None:
respx_mock.post(EXA_SEARCH_URL).respond(json=EXA_REPLY_WITH_COST)
catalog: Final = GetModelCostMap.load_local_model_cost_map()
served: Final = {
**catalog,
"exa_ai/search": {
**catalog["exa_ai/search"],
**({"input_cost_per_query": price} if price is not None else {}),
},
}
respx_mock.get(source).respond(json=served)
monkeypatch.delenv("LITELLM_LOCAL_MODEL_COST_MAP", raising=False)
monkeypatch.setattr(
litellm,
"model_cost",
litellm.get_model_cost_map(url=source, max_attempts=1),
)
response: Final = litellm.search(query="catalog", search_provider="exa_ai", api_key="test-key", type=mode)
assert response.hidden_params["response_cost"] == pytest.approx(expected_cost)
@pytest.mark.usefixtures("_search_mode_prices")
def test_sync_tavily_advanced_search_reports_the_advanced_price(respx_mock: respx.MockRouter) -> None:
respx_mock.post(TAVILY_SEARCH_URL).respond(json={"results": [{"title": "t", "url": "https://example.com"}]})

14
tests/unit/test_utils.py
View file

@ -2331,6 +2331,20 @@ def test_register_model_price_is_used_for_completion_cost() -> None:
assert cost == pytest.approx(0.02)
@pytest.mark.parametrize("flat_first", (True, False))
def test_latest_exa_price_registration_replaces_the_previous_price_kind(
local_model_cost_map: None, flat_first: bool
) -> None:
flat: Final = {"input_cost_per_query": 0.321}
tiered: Final = {"tiered_pricing": [{"input_cost_per_query": 0.456, "max_results_range": [0, 100]}]}
for price in (flat, tiered) if flat_first else (tiered, flat):
litellm.register_model({"exa_ai/search": price}, persist_across_reloads=False)
assert litellm.completion_cost(
model="exa_ai/search", custom_llm_provider="exa_ai", call_type="search"
) == pytest.approx(0.456 if flat_first else 0.321)
@respx.mock
@pytest.mark.usefixtures("isolated_openai_model_sets", "local_model_cost_map")
def test_completion_registers_test_owned_model_prices() -> None: