datarobot_genai.drtools.core.clients.perplexity

perplexity

PerplexityError

Bases: Exception

Exception for Perplexity API errors.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

class PerplexityError(Exception):
    """Exception for Perplexity API errors."""

    def __init__(self, message: str) -> None:
        super().__init__(message)

PerplexityThinkResult

Bases: BaseModel

Source code in datarobot_genai/drtools/core/clients/perplexity.py

class PerplexityThinkResult(BaseModel):
    usage: _UsageInfo | None
    answer: str
    citations: list[str]

    model_config = ConfigDict(populate_by_name=True)

    @classmethod
    def from_perplexity_sdk(cls, result: StreamChunk) -> "PerplexityThinkResult":
        """Create a PerplexityThinkResult from perplexity sdk response data."""
        return cls(
            usage=_UsageInfo.from_perplexity_sdk(result.usage) if result.usage else None,
            answer=result.choices[0].message.content,
            citations=result.citations if result.citations else [],
        )

from_perplexity_sdk classmethod

from_perplexity_sdk(result: StreamChunk) -> PerplexityThinkResult

Create a PerplexityThinkResult from perplexity sdk response data.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

@classmethod
def from_perplexity_sdk(cls, result: StreamChunk) -> "PerplexityThinkResult":
    """Create a PerplexityThinkResult from perplexity sdk response data."""
    return cls(
        usage=_UsageInfo.from_perplexity_sdk(result.usage) if result.usage else None,
        answer=result.choices[0].message.content,
        citations=result.citations if result.citations else [],
    )

PerplexitySearchResult

Bases: BaseModel

Source code in datarobot_genai/drtools/core/clients/perplexity.py

class PerplexitySearchResult(BaseModel):
    snippet: str
    title: str
    url: str
    date: str | None = None
    last_updated: Annotated[str | None, Field(alias="lastUpdated")] = None

    model_config = ConfigDict(populate_by_name=True)

    @classmethod
    def from_perplexity_sdk(cls, result: search_create_response.Result) -> "PerplexitySearchResult":
        """Create a PerplexitySearchResult from perplexity sdk response data."""
        return cls(**result.model_dump())

from_perplexity_sdk classmethod

from_perplexity_sdk(result: Result) -> PerplexitySearchResult

Create a PerplexitySearchResult from perplexity sdk response data.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

@classmethod
def from_perplexity_sdk(cls, result: search_create_response.Result) -> "PerplexitySearchResult":
    """Create a PerplexitySearchResult from perplexity sdk response data."""
    return cls(**result.model_dump())

PerplexityClient

Client for interacting with Perplexity API. Its simple wrapper around perplexity python sdk.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

class PerplexityClient:
    """Client for interacting with Perplexity API.
    Its simple wrapper around perplexity python sdk.
    """

    def __init__(self, access_token: str) -> None:
        self._client = AsyncPerplexity(api_key=access_token)

    async def search(
        self,
        query: str | list[str],
        search_domain_filter: list[str] | None = None,
        recency: Literal["hour", "day", "week", "month", "year"] | None = None,
        max_results: int = MAX_RESULTS_DEFAULT,
        max_tokens_per_page: int = MAX_TOKENS_PER_PAGE_DEFAULT,
    ) -> list[PerplexitySearchResult]:
        """
        Search using Perplexity.

        Args:
            query: Query to filter results.
            search_domain_filter: Up to 20 domains/URLs to allowlist or denylist.
            recency: Filter results by time period.
            max_results: Number of ranked results to return.
            max_tokens_per_page: Context extraction cap per page.

        Returns
        -------
            List of Perplexity search results.
        """
        if not query:
            raise PerplexityError("Error: query cannot be empty.")
        if query and isinstance(query, str) and not query.strip():
            raise PerplexityError("Error: query cannot be empty.")
        if query and isinstance(query, list) and len(query) > MAX_QUERIES:
            raise PerplexityError(f"Error: query list cannot be bigger than {MAX_QUERIES}.")
        if query and isinstance(query, list) and not all(q.strip() for q in query):
            raise PerplexityError("Error: query cannot contain empty str.")
        if search_domain_filter and len(search_domain_filter) > MAX_SEARCH_DOMAIN_FILTER:
            raise PerplexityError("Error: maximum number of search domain filters is 20.")
        if max_results <= 0:
            raise PerplexityError("Error: max_results must be greater than 0.")
        if max_results > MAX_RESULTS:
            raise PerplexityError("Error: max_results must be smaller than or equal to 20.")
        if max_tokens_per_page <= 0:
            raise PerplexityError("Error: max_tokens_per_page must be greater than 0.")
        if max_tokens_per_page > MAX_TOKENS_PER_PAGE:
            raise PerplexityError(
                "Error: max_tokens_per_page must be smaller than or equal to 8192."
            )

        max_results = min(max_results, MAX_RESULTS)
        max_tokens_per_page = min(max_tokens_per_page, MAX_TOKENS_PER_PAGE)

        search_result = await self._client.search.create(
            query=query,
            search_domain_filter=search_domain_filter,
            search_recency_filter=recency,
            max_results=max_results,
            max_tokens_per_page=max_tokens_per_page,
        )

        return [
            PerplexitySearchResult.from_perplexity_sdk(result) for result in search_result.results
        ]

    async def think(
        self,
        prompt: str,
        model: Literal["sonar", "sonar-reasoning-pro", "sonar-deep-research"],
        json_schema: dict | None = None,
    ) -> PerplexityThinkResult:
        """
        Think using Perplexity.

        Args:
            prompt: The research prompt or instruction.
            model: Select capability:
                - 'sonar' (fast ask),
                - 'sonar-reasoning-pro' (complex logic)
                - 'sonar-deep-research' (comprehensive reports).
            json_schema: Optional JSON Schema to enforce structured output for data extraction.

        Returns
        -------
            Perplexity think result.
        """
        if not prompt or not prompt.strip():
            raise PerplexityError("Error: prompt cannot be empty.")

        response_format = None
        if json_schema:
            response_format = ResponseFormatResponseFormatJsonSchema(
                type="json_schema", json_schema={"schema": json_schema}
            )

        result = await self._client.chat.completions.create(
            messages=[ChatMessageInput(role="user", content=prompt)],
            model=model,
            response_format=response_format,
        )

        return PerplexityThinkResult.from_perplexity_sdk(result)

    async def __aenter__(self) -> "PerplexityClient":
        """Async context manager entry."""
        return self

    async def __aexit__(
        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
    ) -> None:
        """Async context manager exit."""
        await self._client.close()

search async

search(query: str | list[str], search_domain_filter: list[str] | None = None, recency: Literal['hour', 'day', 'week', 'month', 'year'] | None = None, max_results: int = MAX_RESULTS_DEFAULT, max_tokens_per_page: int = MAX_TOKENS_PER_PAGE_DEFAULT) -> list[PerplexitySearchResult]

Search using Perplexity.

Parameters:

Name	Type	Description	Default
query	str | list[str]	Query to filter results.	required
search_domain_filter	list[str] | None	Up to 20 domains/URLs to allowlist or denylist.	None
recency	Literal['hour', 'day', 'week', 'month', 'year'] | None	Filter results by time period.	None
max_results	int	Number of ranked results to return.	MAX_RESULTS_DEFAULT
max_tokens_per_page	int	Context extraction cap per page.	MAX_TOKENS_PER_PAGE_DEFAULT

Returns

List of Perplexity search results.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

async def search(
    self,
    query: str | list[str],
    search_domain_filter: list[str] | None = None,
    recency: Literal["hour", "day", "week", "month", "year"] | None = None,
    max_results: int = MAX_RESULTS_DEFAULT,
    max_tokens_per_page: int = MAX_TOKENS_PER_PAGE_DEFAULT,
) -> list[PerplexitySearchResult]:
    """
    Search using Perplexity.

    Args:
        query: Query to filter results.
        search_domain_filter: Up to 20 domains/URLs to allowlist or denylist.
        recency: Filter results by time period.
        max_results: Number of ranked results to return.
        max_tokens_per_page: Context extraction cap per page.

    Returns
    -------
        List of Perplexity search results.
    """
    if not query:
        raise PerplexityError("Error: query cannot be empty.")
    if query and isinstance(query, str) and not query.strip():
        raise PerplexityError("Error: query cannot be empty.")
    if query and isinstance(query, list) and len(query) > MAX_QUERIES:
        raise PerplexityError(f"Error: query list cannot be bigger than {MAX_QUERIES}.")
    if query and isinstance(query, list) and not all(q.strip() for q in query):
        raise PerplexityError("Error: query cannot contain empty str.")
    if search_domain_filter and len(search_domain_filter) > MAX_SEARCH_DOMAIN_FILTER:
        raise PerplexityError("Error: maximum number of search domain filters is 20.")
    if max_results <= 0:
        raise PerplexityError("Error: max_results must be greater than 0.")
    if max_results > MAX_RESULTS:
        raise PerplexityError("Error: max_results must be smaller than or equal to 20.")
    if max_tokens_per_page <= 0:
        raise PerplexityError("Error: max_tokens_per_page must be greater than 0.")
    if max_tokens_per_page > MAX_TOKENS_PER_PAGE:
        raise PerplexityError(
            "Error: max_tokens_per_page must be smaller than or equal to 8192."
        )

    max_results = min(max_results, MAX_RESULTS)
    max_tokens_per_page = min(max_tokens_per_page, MAX_TOKENS_PER_PAGE)

    search_result = await self._client.search.create(
        query=query,
        search_domain_filter=search_domain_filter,
        search_recency_filter=recency,
        max_results=max_results,
        max_tokens_per_page=max_tokens_per_page,
    )

    return [
        PerplexitySearchResult.from_perplexity_sdk(result) for result in search_result.results
    ]

think async

think(prompt: str, model: Literal['sonar', 'sonar-reasoning-pro', 'sonar-deep-research'], json_schema: dict | None = None) -> PerplexityThinkResult

Think using Perplexity.

Parameters:

Name	Type	Description	Default
prompt	str	The research prompt or instruction.	required
model	Literal['sonar', 'sonar-reasoning-pro', 'sonar-deep-research']	Select capability: - 'sonar' (fast ask), - 'sonar-reasoning-pro' (complex logic) - 'sonar-deep-research' (comprehensive reports).	required
json_schema	dict | None	Optional JSON Schema to enforce structured output for data extraction.	None

Returns

Perplexity think result.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

async def think(
    self,
    prompt: str,
    model: Literal["sonar", "sonar-reasoning-pro", "sonar-deep-research"],
    json_schema: dict | None = None,
) -> PerplexityThinkResult:
    """
    Think using Perplexity.

    Args:
        prompt: The research prompt or instruction.
        model: Select capability:
            - 'sonar' (fast ask),
            - 'sonar-reasoning-pro' (complex logic)
            - 'sonar-deep-research' (comprehensive reports).
        json_schema: Optional JSON Schema to enforce structured output for data extraction.

    Returns
    -------
        Perplexity think result.
    """
    if not prompt or not prompt.strip():
        raise PerplexityError("Error: prompt cannot be empty.")

    response_format = None
    if json_schema:
        response_format = ResponseFormatResponseFormatJsonSchema(
            type="json_schema", json_schema={"schema": json_schema}
        )

    result = await self._client.chat.completions.create(
        messages=[ChatMessageInput(role="user", content=prompt)],
        model=model,
        response_format=response_format,
    )

    return PerplexityThinkResult.from_perplexity_sdk(result)

__aenter__ async

__aenter__() -> PerplexityClient

Async context manager entry.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

async def __aenter__(self) -> "PerplexityClient":
    """Async context manager entry."""
    return self

__aexit__ async

__aexit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None

Async context manager exit.

Source code in datarobot_genai/drtools/core/clients/perplexity.py

async def __aexit__(
    self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
) -> None:
    """Async context manager exit."""
    await self._client.close()

get_perplexity_access_token async

get_perplexity_access_token() -> str | ToolError

Get Perplexity API key from HTTP headers.

At the moment of creating this fn. Perplexity does not support OAuth. It allows only API-KEY authorized flow.

Returns

Access token string on success, ToolError on failure

Example

token = await get_perplexity_access_token()
if isinstance(token, ToolError):
    # Handle error
    return token
# Use token

Source code in datarobot_genai/drtools/core/clients/perplexity.py

async def get_perplexity_access_token() -> str | ToolError:
    """
    Get Perplexity API key from HTTP headers.

    At the moment of creating this fn. Perplexity does not support OAuth.
    It allows only API-KEY authorized flow.

    Returns
    -------
        Access token string on success, ToolError on failure

    Example:
        ```python
        token = await get_perplexity_access_token()
        if isinstance(token, ToolError):
            # Handle error
            return token
        # Use token
        ```
    """
    try:
        creds = get_credentials()
        if api_key := resolve_secret("x-perplexity-api-key", creds.perplexity_api_key):
            return api_key

        logger.warning("Perplexity API key not found.")
        return ToolError(
            "Perplexity API key not found.",
            kind=ToolErrorKind.AUTHENTICATION,
        )
    except Exception as e:
        logger.error(f"Unexpected error obtaining Perplexity API key: {e}.", exc_info=e)
        return ToolError(
            "An unexpected error occurred while obtaining Perplexity API key.",
            kind=ToolErrorKind.INTERNAL,
        )