fromllama_index.llms.perplexityimportPerplexityfromllama_index.core.llmsimportChatMessagepplx_api_key="your-perplexity-api-key"llm=Perplexity(api_key=pplx_api_key,model="mistral-7b-instruct",temperature=0.5)messages_dict=[{"role":"system","content":"Be precise and concise."},{"role":"user","content":"Tell me 5 sentences about Perplexity."},]messages=[ChatMessage(**msg)formsginmessages_dict]response=llm.chat(messages)print(str(response))
Source code in llama-index-integrations/llms/llama-index-llms-perplexity/llama_index/llms/perplexity/base.py
classPerplexity(LLM):"""Perplexity LLM. Examples: `pip install llama-index-llms-perplexity` ```python from llama_index.llms.perplexity import Perplexity from llama_index.core.llms import ChatMessage pplx_api_key = "your-perplexity-api-key" llm = Perplexity( api_key=pplx_api_key, model="mistral-7b-instruct", temperature=0.5 ) messages_dict = [ {"role": "system", "content": "Be precise and concise."}, {"role": "user", "content": "Tell me 5 sentences about Perplexity."}, ] messages = [ChatMessage(**msg) for msg in messages_dict] response = llm.chat(messages) print(str(response)) ``` """model:str=Field(description="The Perplexity model to use.")temperature:float=Field(description="The temperature to use during generation.")max_tokens:Optional[int]=Field(default=None,description="The maximum number of tokens to generate.",)context_window:Optional[int]=Field(default=None,description="The context window to use during generation.",)api_key:str=Field(default=None,description="The Perplexity API key.",exclude=True)api_base:str=Field(default="https://api.perplexity.ai",description="The base URL for Perplexity API.",)additional_kwargs:Dict[str,Any]=Field(default_factory=dict,description="Additional kwargs for the Perplexity API.")max_retries:int=Field(default=10,description="The maximum number of API retries.")headers:Dict[str,str]=Field(default_factory=dict,description="Headers for API requests.")def__init__(self,model:str="mistral-7b-instruct",temperature:float=0.1,max_tokens:Optional[int]=None,api_key:Optional[str]=None,api_base:Optional[str]="https://api.perplexity.ai",additional_kwargs:Optional[Dict[str,Any]]=None,max_retries:int=10,context_window:Optional[int]=None,callback_manager:Optional[CallbackManager]=None,system_prompt:Optional[str]=None,messages_to_prompt:Optional[Callable[[Sequence[ChatMessage]],str]]=None,completion_to_prompt:Optional[Callable[[str],str]]=None,pydantic_program_mode:PydanticProgramMode=PydanticProgramMode.DEFAULT,output_parser:Optional[BaseOutputParser]=None,**kwargs:Any,)->None:additional_kwargs=additional_kwargsor{}headers={"accept":"application/json","content-type":"application/json","authorization":f"Bearer {api_key}",}super().__init__(model=model,temperature=temperature,max_tokens=max_tokens,additional_kwargs=additional_kwargs,max_retries=max_retries,callback_manager=callback_manager,api_key=api_key,api_base=api_base,headers=headers,context_window=context_window,system_prompt=system_prompt,messages_to_prompt=messages_to_prompt,completion_to_prompt=completion_to_prompt,pydantic_program_mode=pydantic_program_mode,output_parser=output_parser,**kwargs,)@classmethoddefclass_name(cls)->str:return"perplexity_llm"@propertydefmetadata(self)->LLMMetadata:returnLLMMetadata(context_window=(self.context_windowifself.context_windowisnotNoneelseself._get_context_window()),num_output=self.max_tokensor-1,# You can replace this with the appropriate valueis_chat_model=self._is_chat_model(),model_name=self.model,)def_get_context_window(self)->int:model_context_windows={"sonar-small-chat":16384,"sonar-small-online":12000,"sonar-medium-chat":16384,"sonar-medium-online":12000,"codellama-34b-instruct":16384,"mistral-7b-instruct":16384,"mixtral-8x7b-instruct":16384,}returnmodel_context_windows.get(self.model,4096)# Default to 4096 if model not founddef_is_chat_model(self)->bool:chat_models={"sonar-small-chat","sonar-small-online","sonar-medium-chat","sonar-medium-online""codellama-34b-instruct","mistral-7b-instruct","mixtral-8x7b-instruct",}returnself.modelinchat_modelsdef_get_all_kwargs(self,**kwargs:Any)->Dict[str,Any]:"""Get all data for the request as a dictionary."""base_kwargs={"model":self.model,"temperature":self.temperature,}ifself.max_tokensisnotNone:base_kwargs["max_tokens"]=self.max_tokensreturn{**base_kwargs,**self.additional_kwargs,**kwargs}def_complete(self,prompt:str,**kwargs:Any)->CompletionResponse:url=f"{self.api_base}/chat/completions"payload={"model":self.model,"messages":[{"role":"system","content":self.system_prompt},{"role":"user","content":prompt,},],**self._get_all_kwargs(**kwargs),}response=requests.post(url,json=payload,headers=self.headers)response.raise_for_status()data=response.json()returnCompletionResponse(text=data["choices"][0]["message"],raw=data)@llm_completion_callback()defcomplete(self,prompt:str,formatted:bool=False,**kwargs:Any)->CompletionResponse:ifself._is_chat_model():raiseValueError("The complete method is not supported for chat models.")returnself._complete(prompt,**kwargs)def_chat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponse:url=f"{self.api_base}/chat/completions"payload={"model":self.model,"messages":[message.dict(exclude={"additional_kwargs"})formessageinmessages],**self._get_all_kwargs(**kwargs),}response=requests.post(url,json=payload,headers=self.headers)response.raise_for_status()data=response.json()message=ChatMessage(role="assistant",content=data["choices"][0]["message"]["content"])returnChatResponse(message=message,raw=data)@llm_chat_callback()defchat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponse:returnself._chat(messages,**kwargs)asyncdef_acomplete(self,prompt:str,**kwargs:Any)->CompletionResponse:url=f"{self.api_base}/chat/completions"payload={"model":self.model,"prompt":prompt,**self._get_all_kwargs(**kwargs),}asyncwithhttpx.AsyncClient()asclient:response=awaitclient.post(url,json=payload,headers=self.headers)response.raise_for_status()data=response.json()returnCompletionResponse(text=data["choices"][0]["text"],raw=data)@llm_completion_callback()asyncdefacomplete(self,prompt:str,formatted:bool=False,**kwargs:Any)->CompletionResponse:ifself._is_chat_model():raiseValueError("The complete method is not supported for chat models.")returnawaitself._acomplete(prompt,**kwargs)asyncdef_achat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponse:url=f"{self.api_base}/chat/completions"payload={"model":self.model,"messages":[message.dict(exclude={"additional_kwargs"})formessageinmessages],**self._get_all_kwargs(**kwargs),}asyncwithhttpx.AsyncClient()asclient:response=awaitclient.post(url,json=payload,headers=self.headers)response.raise_for_status()data=response.json()message=ChatMessage(role="assistant",content=data["choices"][0]["message"]["content"])returnChatResponse(message=message,raw=data)@llm_chat_callback()asyncdefachat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponse:returnawaitself._achat(messages,**kwargs)def_stream_complete(self,prompt:str,**kwargs:Any)->CompletionResponseGen:url=f"{self.api_base}/chat/completions"payload={"model":self.model,"prompt":prompt,"stream":True,**self._get_all_kwargs(**kwargs),}defgen()->CompletionResponseGen:withrequests.Session()assession:withsession.post(url,json=payload,headers=self.headers,stream=True)asresponse:response.raise_for_status()text=""forlineinresponse.iter_lines(decode_unicode=True):# decode lines to Unicodeifline.startswith("data:"):data=json.loads(line[5:])delta=data["choices"][0]["text"]text+=deltayieldCompletionResponse(delta=delta,text=text,raw=data)returngen()@llm_completion_callback()defstream_complete(self,prompt:str,formatted:bool=False,**kwargs:Any)->CompletionResponseGen:ifself._is_chat_model():raiseValueError("The complete method is not supported for chat models.")stream_complete_fn=self._stream_completereturnstream_complete_fn(prompt,**kwargs)asyncdef_astream_complete(self,prompt:str,**kwargs:Any)->CompletionResponseAsyncGen:importaiohttpurl=f"{self.api_base}/chat/completions"payload={"model":self.model,"prompt":prompt,"stream":True,**self._get_all_kwargs(**kwargs),}asyncdefgen()->CompletionResponseAsyncGen:asyncwithaiohttp.ClientSession()assession:asyncwithsession.post(url,json=payload,headers=self.headers)asresponse:response.raise_for_status()text=""asyncforlineinresponse.content:line_text=line.decode("utf-8").strip()ifline_text.startswith("data:"):data=json.loads(line_text[5:])delta=data["choices"][0]["text"]text+=deltayieldCompletionResponse(delta=delta,text=text,raw=data)returngen()@llm_completion_callback()asyncdefastream_complete(self,prompt:str,formatted:bool=False,**kwargs:Any)->CompletionResponseAsyncGen:ifself._is_chat_model():raiseValueError("The complete method is not supported for chat models.")returnawaitself._astream_complete(prompt,**kwargs)def_stream_chat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponseGen:url=f"{self.api_base}/chat/completions"payload={"model":self.model,"messages":[message.dict(exclude={"additional_kwargs"})formessageinmessages],"stream":True,**self._get_all_kwargs(**kwargs),}defgen()->ChatResponseGen:content=""withrequests.Session()assession:withsession.post(url,json=payload,headers=self.headers,stream=True)asresponse:response.raise_for_status()forlineinresponse.iter_lines(decode_unicode=True):# decode lines to Unicodeifline.startswith("data:"):data=json.loads(line[5:])delta=data["choices"][0]["delta"]["content"]content+=deltamessage=ChatMessage(role="assistant",content=content,raw=data)yieldChatResponse(message=message,delta=delta,raw=data)returngen()@llm_chat_callback()defstream_chat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponseGen:returnself._stream_chat(messages,**kwargs)asyncdef_astream_chat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponseAsyncGen:importaiohttpurl=f"{self.api_base}/chat/completions"payload={"model":self.model,"messages":[message.dict(exclude={"additional_kwargs"})formessageinmessages],"stream":True,**self._get_all_kwargs(**kwargs),}asyncdefgen()->ChatResponseAsyncGen:asyncwithaiohttp.ClientSession()assession:asyncwithsession.post(url,json=payload,headers=self.headers)asresponse:response.raise_for_status()content=""asyncforlineinresponse.content:line_text=line.decode("utf-8").strip()ifline_text.startswith("data:"):data=json.loads(line_text[5:])delta=data["choices"][0]["delta"]["content"]content+=deltamessage=ChatMessage(role="assistant",content=content,raw=data)yieldChatResponse(message=message,delta=delta,raw=data)returngen()@llm_chat_callback()asyncdefastream_chat(self,messages:Sequence[ChatMessage],**kwargs:Any)->ChatResponseAsyncGen:returnawaitself._astream_chat(messages,**kwargs)