""" title: Perplexity Citations Manifold Pipe author: flynnoct, nikolaskn, justinh-rahb, moblangeois, almajo123 author_url: https://github.com/open-webui funding_url: https://github.com/open-webui version: 0.2.1 license: MIT """ # This version uses async requests and logging library. # OpenWebUI does not natively support Preplexity AI outputs with their citations yet. # Function until this changes (https://github.com/open-webui/open-webui/discussions/9046) import json import logging from typing import Generator, Iterator, Union import aiohttp from open_webui.utils.misc import pop_system_message from pydantic import BaseModel, Field logger = logging.getLogger(__name__) class Pipe: class Valves(BaseModel): NAME_PREFIX: str = Field( default="perplexity/", description="The prefix applied before the model names.", ) PERPLEXITY_API_BASE_URL: str = Field( default="https://api.perplexity.ai", description="The base URL for Perplexity API endpoints.", ) PERPLEXITY_API_KEY: str = Field( default="", description="Required API key to access Perplexity services.", ) def __init__(self): self.type = "manifold" self.valves = self.Valves() def pipes(self): return [ { "id": "perplexity/sonar-pro", "name": f"{self.valves.NAME_PREFIX}Sonar Pro", } ] async def pipe(self, body: dict, __user__: dict) -> Union[str, Generator, Iterator]: logger.debug(f"pipe:{__name__}") if not self.valves.PERPLEXITY_API_KEY: raise Exception("PERPLEXITY_API_KEY not provided in the valves.") headers = { "Authorization": f"Bearer {self.valves.PERPLEXITY_API_KEY}", "Content-Type": "application/json", "accept": "application/json", } system_message, messages = pop_system_message(body.get("messages", [])) system_prompt = "You are a helpful assistant." if system_message is not None: system_prompt = system_message["content"] model_id = body["model"] if model_id.startswith("perplexity_citations_manifold."): model_id = model_id[len("perplexity_citations_manifold.") :] model_id = model_id.removeprefix(self.valves.NAME_PREFIX) payload = { "model": model_id, "messages": [{"role": "system", "content": system_prompt}, *messages], "stream": body.get("stream", True), "return_citations": True, "return_images": True, } url = f"{self.valves.PERPLEXITY_API_BASE_URL}/chat/completions" try: if body.get("stream", False): return self.stream_response(url, headers, payload) else: return await self.non_stream_response(url, headers, payload) except aiohttp.ClientError as e: logger.error(f"Request failed: {e}") return f"Error: Request failed: {e}" except Exception as e: logger.error(f"Error in pipe method: {e}") return f"Error: {e}" def extract_domain(self, url: str) -> str: """Extract domain from URL.""" try: from urllib.parse import urlparse return urlparse(url).netloc except: return url def format_citations(self, citations: list) -> str: """Format citations in markdown.""" if not citations: return "" citations_list = [ f"[\[{i+1}\] {self.extract_domain(url)}]({url})" for i, url in enumerate(citations) ] return "\n\n**Referenzen:**\n" + "\n".join(citations_list) async def stream_response(self, url, headers, payload): try: async with aiohttp.ClientSession() as session: async with session.post( url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=60), ) as response: if response.status != 200: text = await response.text() raise Exception(f"HTTP Error {response.status}: {text}") data = None citations = None async for line in response.content: line = line.decode("utf-8").strip() if line and line.startswith("data: "): try: data = json.loads(line[6:]) if citations is None and "citations" in data: citations = data["citations"] yield data["choices"][0]["delta"]["content"] except json.JSONDecodeError: logger.error(f"Failed to parse JSON: {line}") except KeyError as e: logger.error(f"Unexpected data structure: {e}") logger.error(f"Full data: {data}") if citations: yield self.format_citations(citations) except aiohttp.ClientError as e: logger.error(f"Request failed: {e}") yield f"Error: Request failed: {e}" except Exception as e: logger.error(f"General error in stream_response method: {e}") yield f"Error: {e}" async def non_stream_response(self, url, headers, payload): try: async with aiohttp.ClientSession() as session: async with session.post( url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=60), ) as response: if response.status != 200: text = await response.text() raise Exception(f"HTTP Error {response.status}: {text}") res = await response.json() citations = res.get("citations", []) content = res["choices"][0]["message"]["content"] return content + self.format_citations(citations) except aiohttp.ClientError as e: logger.error(f"Failed non-stream request: {e}") return f"Error: {e}" except Exception as e: logger.error(f"General error in non_stream_response method: {e}") return f"Error: {e}"