"""Perplexity AI client for intelligent OSINT synthesis."""
import httpx
from typing import Optional, Any
from dataclasses import dataclass
@dataclass
class PerplexityClient:
"""Client for Perplexity AI API."""
api_key: Optional[str] = None
base_url: str = "https://api.perplexity.ai"
model: str = "sonar"
_client: Optional[httpx.AsyncClient] = None
async def start(self):
"""Initialize HTTP client."""
if not self._client and self.api_key:
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
timeout=60.0,
)
async def close(self):
"""Close HTTP client."""
if self._client:
await self._client.aclose()
self._client = None
async def query(
self,
prompt: str,
system_prompt: Optional[str] = None,
search_focus: str = "internet",
) -> dict[str, Any]:
"""
Query Perplexity for AI-powered research.
Args:
prompt: Research question or query
system_prompt: System context for the query
search_focus: "internet" or "academic"
Returns:
AI response with citations
"""
if not self.api_key:
return {"error": "Perplexity API key not configured"}
await self.start()
try:
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
resp = await self._client.post(
"/chat/completions",
json={
"model": self.model,
"messages": messages,
"temperature": 0.2,
"top_p": 0.9,
"search_domain_filter": [],
"return_citations": True,
"return_images": False,
"return_related_questions": True,
"search_recency_filter": "month",
},
)
resp.raise_for_status()
data = resp.json()
choice = data.get("choices", [{}])[0]
return {
"response": choice.get("message", {}).get("content"),
"citations": data.get("citations", []),
"related_questions": data.get("related_questions", []),
"model": data.get("model"),
"usage": data.get("usage"),
}
except httpx.HTTPStatusError as e:
return {"error": f"Perplexity API error: {e.response.status_code}"}
except Exception as e:
return {"error": str(e)}
async def research_person(self, name: str, context: Optional[str] = None) -> dict[str, Any]:
"""
Research a person using Perplexity.
Args:
name: Person's name
context: Additional context (company, role, etc.)
Returns:
AI-synthesized person brief
"""
system_prompt = """You are an OSINT researcher. Provide factual, verified information only.
Structure your response as:
- Background & Current Role
- Career History
- Notable Achievements or News
- Online Presence
- Key Connections
Be concise and cite your sources."""
query = f"Research {name}"
if context:
query += f" ({context})"
query += ". Who are they? What is their background? Any recent news?"
return await self.query(query, system_prompt)
async def research_company(self, company: str, domain: Optional[str] = None) -> dict[str, Any]:
"""
Research a company using Perplexity.
Args:
company: Company name
domain: Company domain for disambiguation
Returns:
AI-synthesized company brief
"""
system_prompt = """You are a business intelligence researcher. Provide factual, verified information.
Structure your response as:
- Company Overview
- Leadership & Key People
- Products/Services
- Recent News & Developments
- Market Position & Competitors
Be concise and cite your sources."""
query = f"Research {company}"
if domain:
query += f" ({domain})"
query += ". Company overview, leadership, recent news, and market position."
return await self.query(query, system_prompt)
