"""
Transformation logic for Perplexity Agentic Research API (Responses API)
This module handles the translation between OpenAI's Responses API format
and Perplexity's Responses API format, which supports:
- Third-party model access (OpenAI, Anthropic, Google, xAI, etc.)
- Presets for optimized configurations
- Web search and URL fetching tools
- Reasoning effort control
- Instructions parameter for system-level guidance
"""
from typing import Any, Dict, List, Optional, Union
import httpx
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import (
ResponseAPIUsage,
ResponseInputParam,
ResponsesAPIOptionalRequestParams,
ResponsesAPIResponse,
ResponsesAPIStreamingResponse,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
class PerplexityResponsesConfig(OpenAIResponsesAPIConfig):
"""
Configuration for Perplexity Agentic Research API (Responses API)
Reference: https://docs.perplexity.ai/agentic-research/quickstart
"""
@property
def custom_llm_provider(self) -> LlmProviders:
return LlmProviders.PERPLEXITY
def get_supported_openai_params(self, model: str) -> list:
"""
Perplexity Responses API supports a different set of parameters
Ref: https://docs.perplexity.ai/api-reference/responses-post
"""
return [
"max_output_tokens",
"stream",
"temperature",
"top_p",
"tools",
"reasoning",
"preset",
"instructions",
"models", # Model fallback support
]
def validate_environment(
self, headers: dict, model: str, litellm_params: Optional[GenericLiteLLMParams]
) -> dict:
"""Validate environment and set up headers"""
# Get API key from environment
api_key = (
get_secret_str("PERPLEXITYAI_API_KEY")
or get_secret_str("PERPLEXITY_API_KEY")
)
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
headers["Content-Type"] = "application/json"
return headers
def get_complete_url(
self,
api_base: Optional[str],
litellm_params: dict,
) -> str:
"""Get the complete URL for the Perplexity Responses API"""
if api_base is None:
api_base = get_secret_str("PERPLEXITY_API_BASE") or "https://api.perplexity.ai"
# Ensure api_base doesn't end with a slash
api_base = api_base.rstrip("/")
# Add the responses endpoint
return f"{api_base}/v1/responses"
def map_openai_params(
self,
response_api_optional_params: ResponsesAPIOptionalRequestParams,
model: str,
drop_params: bool,
) -> Dict:
"""
Map OpenAI Responses API parameters to Perplexity format
Key differences:
- Supports 'preset' parameter for predefined configurations
- Supports 'instructions' parameter for system-level guidance
- Tools are specified differently (web_search, fetch_url)
"""
mapped_params: Dict[str, Any] = {}
# Map standard parameters
if response_api_optional_params.get("max_output_tokens"):
mapped_params["max_output_tokens"] = response_api_optional_params["max_output_tokens"]
if response_api_optional_params.get("temperature"):
mapped_params["temperature"] = response_api_optional_params["temperature"]
if response_api_optional_params.get("top_p"):
mapped_params["top_p"] = response_api_optional_params["top_p"]
if response_api_optional_params.get("stream"):
mapped_params["stream"] = response_api_optional_params["stream"]
if response_api_optional_params.get("stream_options"):
mapped_params["stream_options"] = response_api_optional_params["stream_options"]
# Map Perplexity-specific parameters (using .get() with Any dict access)
preset = response_api_optional_params.get("preset") # type: ignore
if preset:
mapped_params["preset"] = preset
instructions = response_api_optional_params.get("instructions") # type: ignore
if instructions:
mapped_params["instructions"] = instructions
if response_api_optional_params.get("reasoning"):
mapped_params["reasoning"] = response_api_optional_params["reasoning"]
tools = response_api_optional_params.get("tools")
if tools:
# Convert tools to list of dicts for transformation
tools_list = [dict(tool) if hasattr(tool, '__dict__') else tool for tool in tools] # type: ignore
mapped_params["tools"] = self._transform_tools(tools_list) # type: ignore
return mapped_params
def _transform_tools(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Transform tools to Perplexity format
Perplexity supports:
- web_search: Performs web searches
- fetch_url: Fetches content from URLs
"""
perplexity_tools = []
for tool in tools:
if isinstance(tool, dict):
tool_type = tool.get("type")
# Direct Perplexity tool format
if tool_type in ["web_search", "fetch_url"]:
perplexity_tools.append(tool)
# OpenAI function format - try to map to Perplexity tools
elif tool_type == "function":
function = tool.get("function", {})
function_name = function.get("name", "")
if function_name == "web_search" or "search" in function_name.lower():
perplexity_tools.append({"type": "web_search"})
elif function_name == "fetch_url" or "fetch" in function_name.lower():
perplexity_tools.append({"type": "fetch_url"})
return perplexity_tools
def transform_responses_api_request(
self,
model: str,
input: Union[str, ResponseInputParam],
response_api_optional_request_params: Dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Dict:
"""
Transform request to Perplexity Responses API format
"""
# Check if the model is a preset (format: preset/preset-name)
if model.startswith("preset/"):
preset_name = model.replace("preset/", "")
data = {
"preset": preset_name,
"input": self._format_input(input),
}
# Check if preset is explicitly provided in params
elif response_api_optional_request_params.get("preset"):
data = {
"preset": response_api_optional_request_params.pop("preset"),
"input": self._format_input(input),
}
else:
# Full request format for third-party models
data = {
"model": model,
"input": self._format_input(input),
}
# Add all optional parameters
for key, value in response_api_optional_request_params.items():
data[key] = value
return data
def _format_input(self, input: Union[str, ResponseInputParam]) -> Union[str, List[Dict[str, Any]]]:
"""
Format input for Perplexity Responses API
The API accepts either:
- A simple string for single-turn queries
- An array of message objects for multi-turn conversations
"""
if isinstance(input, str):
return input
# Handle ResponseInputParam format
if isinstance(input, list):
formatted_messages = []
for item in input:
if isinstance(item, dict):
formatted_message = {
"type": "message",
"role": item.get("role"),
"content": item.get("content", ""),
}
formatted_messages.append(formatted_message)
return formatted_messages
return str(input)
def transform_response_api_response(
self,
model: str,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIResponse:
"""
Transform Perplexity Responses API response to OpenAI Responses API format
"""
try:
raw_response_json = raw_response.json()
except Exception as e:
raise BaseLLMException(
status_code=raw_response.status_code,
message=f"Failed to parse response: {str(e)}",
)
# Check for error status
status = raw_response_json.get("status")
if status == "failed":
error = raw_response_json.get("error", {})
error_message = error.get("message", "Unknown error")
raise BaseLLMException(
status_code=raw_response.status_code,
message=error_message,
)
# Transform usage to handle Perplexity's cost structure
usage_data = raw_response_json.get("usage", {})
transformed_usage_dict = self._transform_usage(usage_data)
# Convert usage dict to ResponseAPIUsage object
usage_obj = ResponseAPIUsage(**transformed_usage_dict) if transformed_usage_dict else None
# Map Perplexity response to OpenAI Responses API format
response = ResponsesAPIResponse(
id=raw_response_json.get("id", ""),
object="response",
created_at=raw_response_json.get("created_at", 0),
status=raw_response_json.get("status", "completed"),
model=raw_response_json.get("model", model),
output=raw_response_json.get("output", []),
usage=usage_obj,
)
return response
def _transform_usage(self, usage_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Transform Perplexity usage data to OpenAI format
Perplexity returns:
{
"input_tokens": 100,
"output_tokens": 200,
"total_tokens": 300,
"cost": {
"currency": "USD",
"input_cost": 0.0001,
"output_cost": 0.0002,
"total_cost": 0.0003
}
}
OpenAI expects:
{
"input_tokens": 100,
"output_tokens": 200,
"total_tokens": 300,
"cost": 0.0003
}
"""
transformed = {
"input_tokens": usage_data.get("input_tokens", 0),
"output_tokens": usage_data.get("output_tokens", 0),
"total_tokens": usage_data.get("total_tokens", 0),
}
# Transform cost from Perplexity format (dict) to OpenAI format (float)
cost_obj = usage_data.get("cost")
if isinstance(cost_obj, dict) and "total_cost" in cost_obj:
transformed["cost"] = cost_obj["total_cost"]
verbose_logger.debug(
"Transformed Perplexity cost object to float: %s -> %s",
cost_obj,
cost_obj["total_cost"]
)
elif cost_obj is not None:
# If cost is already a float/number, use it as-is
transformed["cost"] = cost_obj
# Add input_tokens_details if present
if "input_tokens_details" in usage_data:
transformed["input_tokens_details"] = usage_data["input_tokens_details"]
# Add output_tokens_details if present
if "output_tokens_details" in usage_data:
transformed["output_tokens_details"] = usage_data["output_tokens_details"]
return transformed
def transform_streaming_response(
self,
model: str,
parsed_chunk: dict,
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIStreamingResponse:
"""
Transform a parsed streaming response chunk into a ResponsesAPIStreamingResponse
"""
# Get the event type from the chunk
verbose_logger.debug("Raw Perplexity Chunk=%s", parsed_chunk)
event_type = str(parsed_chunk.get("type"))
event_pydantic_model = PerplexityResponsesConfig.get_event_model_class(
event_type=event_type
)
# Transform Perplexity-specific fields to OpenAI format
parsed_chunk = self._transform_perplexity_chunk(parsed_chunk)
# Defensive: Handle error.code being null (similar to OpenAI implementation)
try:
error_obj = parsed_chunk.get("error")
if isinstance(error_obj, dict) and error_obj.get("code") is None:
# Preserve other fields, but ensure `code` is a non-null string
parsed_chunk = dict(parsed_chunk)
parsed_chunk["error"] = dict(error_obj)
parsed_chunk["error"]["code"] = "unknown_error"
except Exception:
# If anything unexpected happens here, fall back to attempting
# instantiation and let higher-level handlers manage errors.
verbose_logger.debug("Failed to coalesce error.code in parsed_chunk")
return event_pydantic_model(**parsed_chunk)
def _transform_perplexity_chunk(self, chunk: dict) -> dict:
"""
Transform Perplexity-specific fields in a streaming chunk to OpenAI format.
This handles:
- Converting Perplexity's cost object to a simple float
"""
# Make a copy to avoid modifying the original
chunk = dict(chunk)
# Transform usage.cost from Perplexity format to OpenAI format
# Perplexity: {"currency": "USD", "input_cost": 0.0001, "output_cost": 0.0002, "total_cost": 0.0003}
# OpenAI: 0.0003 (just the total_cost as a float)
try:
response_obj = chunk.get("response")
if isinstance(response_obj, dict):
usage_obj = response_obj.get("usage")
if isinstance(usage_obj, dict):
cost_obj = usage_obj.get("cost")
if isinstance(cost_obj, dict) and "total_cost" in cost_obj:
# Replace the cost object with just the total_cost value
chunk = dict(chunk)
chunk["response"] = dict(response_obj)
chunk["response"]["usage"] = dict(usage_obj)
chunk["response"]["usage"]["cost"] = cost_obj["total_cost"]
verbose_logger.debug(
"Transformed Perplexity cost object to float: %s -> %s",
cost_obj,
cost_obj["total_cost"]
)
except Exception as e:
# If transformation fails, log and continue with original chunk
verbose_logger.debug("Failed to transform Perplexity cost object: %s", e)
return chunk