import os, json from typing import Dict, List, Optional from openai import OpenAI from mem0.llms.base import LLMBase from mem0.configs.llms.base import BaseLlmConfig class OpenAIStructuredLLM(LLMBase): def __init__(self, config: Optional[BaseLlmConfig] = None): super().__init__(config) if not self.config.model: self.config.model = "gpt-4o-2024-08-06" api_key = os.getenv("OPENAI_API_KEY") or self.config.api_key base_url = os.getenv("OPENAI_API_BASE") or self.config.openai_base_url self.client = OpenAI(api_key=api_key, base_url=base_url) def _parse_response(self, response, tools): """ Process the response based on whether tools are used or not. Args: response: The raw response from API. response_format: The format in which the response should be processed. Returns: str or dict: The processed response. """ if tools: processed_response = { "content": response.choices[0].message.content, "tool_calls": [], } if response.choices[0].message.tool_calls: for tool_call in response.choices[0].message.tool_calls: processed_response["tool_calls"].append( { "name": tool_call.function.name, "arguments": json.loads(tool_call.function.arguments), } ) return processed_response else: return response.choices[0].message.content def generate_response( self, messages: List[Dict[str, str]], response_format=None, tools: Optional[List[Dict]] = None, tool_choice: str = "auto", ): """ Generate a response based on the given messages using OpenAI. Args: messages (list): List of message dicts containing 'role' and 'content'. response_format (str or object, optional): Format of the response. Defaults to "text". tools (list, optional): List of tools that the model can call. Defaults to None. tool_choice (str, optional): Tool choice method. Defaults to "auto". Returns: str: The generated response. """ params = { "model": self.config.model, "messages": messages, "temperature": self.config.temperature, } if response_format: params["response_format"] = response_format if tools: params["tools"] = tools params["tool_choice"] = tool_choice response = self.client.beta.chat.completions.parse(**params) return self._parse_response(response, tools)