import json import os from typing import Dict, List, Optional from openai import OpenAI from mem0.configs.llms.base import BaseLlmConfig from mem0.llms.base import LLMBase class OpenAILLM(LLMBase): """ A class to interact with OpenAI or OpenRouter APIs for generating responses using LLMs. """ def __init__(self, config: Optional[BaseLlmConfig] = None): """ Initializes the OpenAILLM instance. Args: config (Optional[BaseLlmConfig]): Configuration for the LLM, including model, API key, and base URLs. """ super().__init__(config) if not self.config.model: self.config.model = "gpt-4o-mini" if os.environ.get("OPENROUTER_API_KEY"): # Use OpenRouter self.client = OpenAI( api_key=os.environ.get("OPENROUTER_API_KEY"), base_url=self.config.openrouter_base_url or os.getenv("OPENROUTER_API_BASE") or "https://openrouter.ai/api/v1", ) else: api_key = self.config.api_key or os.getenv("OPENAI_API_KEY") base_url = ( self.config.openai_base_url or os.getenv("OPENAI_API_BASE") or "https://api.openai.com/v1" ) self.client = OpenAI(api_key=api_key, base_url=base_url) def generate_response( self, messages: List[Dict[str, str]], response_format: Optional[str] = None, ) -> str: """ Generates a response based on the provided messages using OpenAI or OpenRouter. Args: messages (List[Dict[str, str]]): A list of message dictionaries containing 'role' and 'content'. response_format (Optional[str]): The format of the response. Defaults to None. Returns: str: The generated response from the model. """ params = { "model": self.config.model, "messages": messages, "temperature": self.config.temperature, "max_tokens": self.config.max_tokens, "top_p": self.config.top_p, } if os.getenv("OPENROUTER_API_KEY"): openrouter_params = {} if self.config.models: openrouter_params["models"] = self.config.models openrouter_params["route"] = self.config.route params.pop("model") if self.config.site_url and self.config.app_name: extra_headers = { "HTTP-Referer": self.config.site_url, "X-Title": self.config.app_name, } openrouter_params["extra_headers"] = extra_headers params.update(**openrouter_params) if response_format: params["response_format"] = response_format response = self.client.chat.completions.create(**params) return response.choices[0].message.content