Streaming Responses

from perplexity import Perplexity  # Initialize the client (uses PERPLEXITY_API_KEY environment variable) client = Perplexity()  # Create streaming completion stream = client.chat.completions.create(  model="sonar",  messages=[{"role": "user", "content": "What is the latest in AI research?"}],  stream=True )  # Process streaming response for chunk in stream:  if chunk.choices[0].delta.content:  print(chunk.choices[0].delta.content, end="") 

from perplexity import Perplexity  def stream_with_metadata():  client = Perplexity()    stream = client.chat.completions.create(  model="sonar",  messages=[{"role": "user", "content": "Explain quantum computing"}],  stream=True  )    content = ""  search_results = []  usage_info = None    for chunk in stream:  # Process content  if chunk.choices[0].delta.content:  content_piece = chunk.choices[0].delta.content  content += content_piece  print(content_piece, end='', flush=True)    # Collect metadata from final chunks  if hasattr(chunk, 'search_results') and chunk.search_results:  search_results = chunk.search_results    if hasattr(chunk, 'usage') and chunk.usage:  usage_info = chunk.usage    # Check if streaming is complete  if chunk.choices[0].finish_reason:  print(f"\n\nSearch Results: {search_results}")  print(f"Usage: {usage_info}")    return content, search_results, usage_info  stream_with_metadata() 

import perplexity from perplexity import Perplexity  client = Perplexity()  try:  stream = client.chat.completions.create(  model="sonar-pro",  messages=[  {"role": "user", "content": "Explain machine learning concepts"}  ],  stream=True  )    content = ""  for chunk in stream:  if chunk.choices[0].delta.content:  content_chunk = chunk.choices[0].delta.content  content += content_chunk  print(content_chunk, end="")   except perplexity.APIConnectionError as e:  print(f"Network connection failed: {e}") except perplexity.RateLimitError as e:  print(f"Rate limit exceeded, please retry later: {e}") except perplexity.APIStatusError as e:  print(f"API error {e.status_code}: {e.response}") except Exception as e:  print(f"Unexpected error: {e}") 

# pip install sseclient-py import sseclient import requests import json  def stream_with_proper_sse():  url = "https://api.perplexity.ai/chat/completions"  headers = {  "Authorization": "Bearer YOUR_API_KEY",  "Content-Type": "application/json"  }  payload = {  "model": "sonar",  "messages": [{"role": "user", "content": "Explain quantum computing"}],  "stream": True  }    response = requests.post(url, headers=headers, json=payload, stream=True)  client = sseclient.SSEClient(response)    for event in client.events():  if event.data == '[DONE]':  break  try:  chunk_data = json.loads(event.data)  content = chunk_data['choices'][0]['delta'].get('content', '')  if content:  print(content, end='')  except json.JSONDecodeError:  continue  stream_with_proper_sse() 

from perplexity import Perplexity import time  def buffered_streaming(buffer_size=50, flush_interval=1.0):  client = Perplexity()    stream = client.chat.completions.create(  model="sonar",  messages=[{"role": "user", "content": "Write a detailed explanation of machine learning"}],  stream=True  )    buffer = ""  last_flush = time.time()    for chunk in stream:  if chunk.choices[0].delta.content:  buffer += chunk.choices[0].delta.content    # Flush buffer if it's full or enough time has passed  if len(buffer) >= buffer_size or (time.time() - last_flush) >= flush_interval:  print(buffer, end='', flush=True)  buffer = ""  last_flush = time.time()    # Flush remaining buffer  if buffer:  print(buffer, end='', flush=True)  buffered_streaming() 

from perplexity import Perplexity from typing import Callable, Optional  def stream_with_callbacks(  query: str,  on_content: Optional[Callable[[str], None]] = None,  on_search_results: Optional[Callable[[list], None]] = None,  on_complete: Optional[Callable[[str, dict], None]] = None ):  client = Perplexity()    stream = client.chat.completions.create(  model="sonar",  messages=[{"role": "user", "content": query}],  stream=True  )    full_content = ""  metadata = {}    for chunk in stream:  # Handle content chunks  if chunk.choices[0].delta.content:  content_piece = chunk.choices[0].delta.content  full_content += content_piece  if on_content:  on_content(content_piece)    # Handle search results  if hasattr(chunk, 'search_results') and chunk.search_results:  metadata['search_results'] = chunk.search_results  if on_search_results:  on_search_results(chunk.search_results)    # Handle other metadata  if hasattr(chunk, 'usage') and chunk.usage:  metadata['usage'] = chunk.usage    # Handle completion  if chunk.choices[0].finish_reason:  if on_complete:  on_complete(full_content, metadata)    return full_content, metadata  # Usage example def print_content(content: str):  print(content, end='', flush=True)  def handle_search_results(results: list):  print(f"\n[Found {len(results)} sources]", end='')  def handle_completion(content: str, metadata: dict):  print(f"\n\nCompleted. Total length: {len(content)} characters")  if 'usage' in metadata:  print(f"Token usage: {metadata['usage']}")  stream_with_callbacks(  "Explain the latest developments in renewable energy",  on_content=print_content,  on_search_results=handle_search_results,  on_complete=handle_completion )