mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-14 08:26:09 +00:00
Add perplexity response api class
This commit is contained in:
Sameer Kankute
@@ -0,0 +1,7 @@
|
||||
"""
|
||||
Perplexity Agentic Research API (Responses API) module
|
||||
"""
|
||||
|
||||
from .transformation import PerplexityResponsesConfig
|
||||
|
||||
__all__ = ["PerplexityResponsesConfig"]
|
||||
@@ -0,0 +1,509 @@
|
||||
"""
|
||||
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, Literal, Optional, Tuple, Union
|
||||
|
||||
import httpx
|
||||
|
||||
import litellm
|
||||
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.base_llm.responses.transformation import BaseResponsesAPIConfig
|
||||
from litellm.secret_managers.main import get_secret_str
|
||||
from litellm.types.llms.openai import (
|
||||
ResponseInputParam,
|
||||
ResponsesAPIOptionalRequestParams,
|
||||
ResponsesAPIResponse,
|
||||
ResponsesAPIStreamingResponse,
|
||||
)
|
||||
from litellm.types.responses.main import DeleteResponseResult
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.utils import LlmProviders
|
||||
|
||||
|
||||
class PerplexityResponsesConfig(BaseResponsesAPIConfig):
|
||||
"""
|
||||
Configuration for Perplexity Agentic Research API (Responses API)
|
||||
|
||||
Reference: https://docs.perplexity.ai/agentic-research/quickstart
|
||||
"""
|
||||
|
||||
@property
|
||||
def custom_llm_provider(self) -> LlmProviders:
|
||||
return "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",
|
||||
]
|
||||
|
||||
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 = {}
|
||||
|
||||
# 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
|
||||
if response_api_optional_params.get("preset"):
|
||||
mapped_params["preset"] = response_api_optional_params["preset"]
|
||||
|
||||
if response_api_optional_params.get("instructions"):
|
||||
mapped_params["instructions"] = response_api_optional_params["instructions"]
|
||||
|
||||
if response_api_optional_params.get("reasoning"):
|
||||
mapped_params["reasoning"] = response_api_optional_params["reasoning"]
|
||||
|
||||
if response_api_optional_params.get("tools"):
|
||||
mapped_params["tools"] = self._transform_tools(response_api_optional_params["tools"])
|
||||
|
||||
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 = self._transform_usage(usage_data)
|
||||
|
||||
# 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=transformed_usage,
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
"""
|
||||
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),
|
||||
}
|
||||
|
||||
# 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
|
||||
"""
|
||||
# Map Perplexity streaming chunk to OpenAI format
|
||||
return ResponsesAPIStreamingResponse(**parsed_chunk)
|
||||
|
||||
def transform_delete_response_api_request(
|
||||
self,
|
||||
response_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> Tuple[str, Dict]:
|
||||
"""Transform delete response API request"""
|
||||
# Perplexity may not support deleting responses
|
||||
# Return appropriate URL and params
|
||||
url = f"{api_base}/v1/responses/{response_id}"
|
||||
return url, {}
|
||||
|
||||
def transform_delete_response_api_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> DeleteResponseResult:
|
||||
"""Transform delete response API response"""
|
||||
try:
|
||||
response_json = raw_response.json()
|
||||
return DeleteResponseResult(
|
||||
id=response_json.get("id", ""),
|
||||
object="response.deleted",
|
||||
deleted=response_json.get("deleted", True),
|
||||
)
|
||||
except Exception as e:
|
||||
raise BaseLLMException(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"Failed to parse delete response: {str(e)}",
|
||||
)
|
||||
|
||||
def transform_get_response_api_request(
|
||||
self,
|
||||
response_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> Tuple[str, Dict]:
|
||||
"""Transform get response API request"""
|
||||
url = f"{api_base}/v1/responses/{response_id}"
|
||||
return url, {}
|
||||
|
||||
def transform_get_response_api_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> ResponsesAPIResponse:
|
||||
"""Transform get response API response"""
|
||||
return self.transform_response_api_response(
|
||||
model="", # Model will be in the response
|
||||
raw_response=raw_response,
|
||||
logging_obj=logging_obj,
|
||||
)
|
||||
|
||||
def transform_list_input_items_request(
|
||||
self,
|
||||
response_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
after: Optional[str] = None,
|
||||
before: Optional[str] = None,
|
||||
include: Optional[List[str]] = None,
|
||||
limit: int = 20,
|
||||
order: Literal["asc", "desc"] = "desc",
|
||||
) -> Tuple[str, Dict]:
|
||||
"""Transform list input items request"""
|
||||
url = f"{api_base}/v1/responses/{response_id}/input_items"
|
||||
params = {
|
||||
"limit": limit,
|
||||
"order": order,
|
||||
}
|
||||
|
||||
if after:
|
||||
params["after"] = after
|
||||
if before:
|
||||
params["before"] = before
|
||||
if include:
|
||||
params["include"] = include
|
||||
|
||||
return url, params
|
||||
|
||||
def transform_list_input_items_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> Dict:
|
||||
"""Transform list input items response"""
|
||||
try:
|
||||
return raw_response.json()
|
||||
except Exception as e:
|
||||
raise BaseLLMException(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"Failed to parse list input items response: {str(e)}",
|
||||
)
|
||||
|
||||
def get_error_class(
|
||||
self, error_message: str, status_code: int, headers: Union[Dict, httpx.Headers]
|
||||
) -> BaseLLMException:
|
||||
"""Return appropriate error class based on status code"""
|
||||
return BaseLLMException(
|
||||
status_code=status_code,
|
||||
message=error_message,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
def should_fake_stream(
|
||||
self,
|
||||
model: Optional[str],
|
||||
stream: Optional[bool],
|
||||
custom_llm_provider: Optional[str] = None,
|
||||
) -> bool:
|
||||
"""Returns True if litellm should fake a stream for the given model and stream value"""
|
||||
return False
|
||||
|
||||
#########################################################
|
||||
########## CANCEL RESPONSE API TRANSFORMATION ##########
|
||||
#########################################################
|
||||
def transform_cancel_response_api_request(
|
||||
self,
|
||||
response_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> Tuple[str, Dict]:
|
||||
"""Transform cancel response API request"""
|
||||
# Perplexity may not support canceling responses
|
||||
# Return appropriate URL and params
|
||||
url = f"{api_base}/v1/responses/{response_id}/cancel"
|
||||
return url, {}
|
||||
|
||||
def transform_cancel_response_api_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> ResponsesAPIResponse:
|
||||
"""Transform cancel response API response"""
|
||||
return self.transform_response_api_response(
|
||||
model="", # Model will be in the response
|
||||
raw_response=raw_response,
|
||||
logging_obj=logging_obj,
|
||||
)
|
||||
|
||||
#########################################################
|
||||
########## COMPACT RESPONSE API TRANSFORMATION ##########
|
||||
#########################################################
|
||||
def transform_compact_response_api_request(
|
||||
self,
|
||||
model: str,
|
||||
input: Union[str, ResponseInputParam],
|
||||
response_api_optional_request_params: Dict,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> Tuple[str, Dict]:
|
||||
"""Transform compact response API request"""
|
||||
# Perplexity may not support compact responses
|
||||
# Return standard URL and transformed request
|
||||
url = f"{api_base}/v1/responses"
|
||||
request_data = self.transform_responses_api_request(
|
||||
model=model,
|
||||
input=input,
|
||||
response_api_optional_request_params=response_api_optional_request_params,
|
||||
litellm_params=litellm_params,
|
||||
headers=headers,
|
||||
)
|
||||
return url, request_data
|
||||
|
||||
def transform_compact_response_api_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> ResponsesAPIResponse:
|
||||
"""Transform compact response API response"""
|
||||
return self.transform_response_api_response(
|
||||
model="", # Model will be in the response
|
||||
raw_response=raw_response,
|
||||
logging_obj=logging_obj,
|
||||
)
|
||||
Reference in New Issue
Block a user