File size: 3,479 Bytes
28d8100 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
import asyncio
from typing import Dict, Any
import httpx
from pydantic import BaseModel
from model_registry import model_card
perpelxity_card = model_card["perplexity"]
class QueryRequest(BaseModel):
query: str
class PerplexityClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = perpelxity_card["url"]
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def generate_response(self, query: str) -> Dict[str, Any]:
payload = {
"model": perpelxity_card["model"],
"messages": [
{
"role": "system",
"content": perpelxity_card["prompt"]
},
{
"role": "user",
"content": query
}
],
"max_tokens": perpelxity_card["inference_config"]["max_tokens"],
"temperature": perpelxity_card["inference_config"]["temperature"],
"stream": False, # We'll handle streaming separately
}
async with httpx.AsyncClient() as client:
try:
async with httpx.AsyncClient(
timeout=httpx.Timeout(
connect=10.0, # Connection timeout
read=45.0, # Read timeout
write=10.0, # Write timeout
pool=10.0 # Connection pool timeout
)
) as client:
response = await client.post(
self.base_url,
headers=self.headers,
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
print(f"HTTP error occurred: {e}")
print(f"Response text: {e.response.text}")
raise
except httpx.RequestError as e:
print(f"Request error occurred: {e}")
raise
except Exception as e:
print(f"An unexpected error occurred: {e}")
raise
def parse_perplexity_response(response: Dict[str, Any]) -> Dict[str, Any]:
"""
Parse the Perplexity API response and extract key information.
Args:
response (Dict[str, Any]): Raw response from Perplexity API
Returns:
Dict[str, Any]: Parsed response with content and citations
"""
print("parse_perplexity_response called...")
# Basic citation extraction (this is a simple implementation)
# In a real-world scenario, you might want a more sophisticated citation extraction
citations = []
# default content to stream if no response from the Perplexity AI
default_content = (
"I'm sorry, I couldn't find any relevant information for your query from the available sources. "
"If you'd like, you can try rephrasing your question or provide more context to help refine the search. "
"Alternatively, let me know if you'd like assistance in a different area."
)
citations = response.get("citations", [])
content = response.get("choices", default_content)[0]["message"]["content"]
return {
"content": content,
"citations": citations
}
|