class PortkeyClient(BasePortkeyClient):
def __init__(self, config: PortkeyConfig = PortkeyConfig(), provider: str = "deepinfra"):
super().__init__(config, provider)
self._client = Portkey(api_key=self.config.api_key, virtual_key=self.config.virtual_key)
@retry(
stop=stop_after_attempt(3),
wait=wait_fixed(2),
retry=retry_if_result(should_retry_content)
)
def invoke(self, messages: List[Union[SystemMessage, AIMessage, HumanMessage]],
model: str = "llama_70b",
**kwargs):
self._client.config = PortkeyConfigObjects().invoke # Contains the ID of the config obj on portkey
model_id = self.models.get_model_id(model)
converted_messages = self._convert_langchain_messages(messages)
try:
response = self._client.chat.completions.create(
messages=converted_messages,
model=model_id,
**kwargs
)
return response.choices[0].message
except Exception as e:
logger.error(f"Error in chat completion: {str(e)}")
return f"An error occurred: {str(e)}"