Spaces:
Sleeping
Sleeping
| import os | |
| import openai | |
| import logging | |
| from typing import Optional | |
| class ResponseManager: | |
| """ | |
| This class initializes the OpenAI client and provides methods to create responses, | |
| maintain conversation history, and handle user queries. | |
| Attributes: | |
| DEFAULT_META_PROMPT_FILE (str): Default path to the meta prompt file. | |
| DEFAULT_MODEL (str): Default OpenAI model to use. | |
| DEFAULT_TEMPERATURE (float): Default temperature for response generation. | |
| DEFAULT_MAX_OUTPUT_TOKENS (int): Default maximum number of output tokens. | |
| DEFAULT_MAX_NUM_RESULTS (int): Default maximum number of search results. | |
| Methods: | |
| __init__(vector_store_id: Optional[str], api_key: Optional[str], meta_prompt_file: Optional[str]): | |
| Initializes the ResponseManager with a vector store ID, API key, and meta prompt file. | |
| _load_meta_prompt(meta_prompt_file: str) -> str: | |
| Loads the meta prompt from the specified file. | |
| create_response(query: str, model: Optional[str], temperature: Optional[float], | |
| max_output_tokens: Optional[int], max_num_results: Optional[int]) -> str: | |
| Creates a response to a user query using the OpenAI API. | |
| conversation(query: str, model: Optional[str], temperature: Optional[float], | |
| max_output_tokens: Optional[int], max_num_results: Optional[int], | |
| Handles chatbot interaction and maintains conversation history. | |
| """ | |
| DEFAULT_META_PROMPT_FILE = 'config/meta_prompt.txt' | |
| DEFAULT_MODEL = "gpt-4o-mini" | |
| DEFAULT_TEMPERATURE = 0 | |
| DEFAULT_MAX_OUTPUT_TOKENS = 800 | |
| DEFAULT_MAX_NUM_RESULTS = 15 | |
| def __init__(self, vector_store_id: Optional[str] = None, api_key: Optional[str] = None, meta_prompt_file: Optional[str] = None): | |
| """ | |
| Initialize the ResponseManager with a vector store ID, API key, and meta prompt file. | |
| :param vector_store_id: The ID of the vector store to use for file search. | |
| :param api_key: The OpenAI API key for authentication. | |
| :param meta_prompt_file: Path to the meta prompt file (default: 'config/meta_prompt.txt'). | |
| """ | |
| # Load vector_store_id and api_key from environment variables if not provided | |
| self.vector_store_id = vector_store_id or os.getenv('VECTOR_STORE_ID') | |
| if not self.vector_store_id: | |
| logging.error("VECTOR_STORE_ID is not provided or set in the environment.") | |
| raise ValueError("VECTOR_STORE_ID is required.") | |
| self.api_key = api_key or os.getenv('OPENAI_API_KEY') | |
| if not self.api_key: | |
| logging.error("OPENAI_API_KEY is not provided or set in the environment.") | |
| raise ValueError("OPENAI_API_KEY is required.") | |
| # Initialize other attributes | |
| self.meta_prompt_file = meta_prompt_file or self.DEFAULT_META_PROMPT_FILE | |
| self.previous_response_id = None | |
| # Initialize the OpenAI client | |
| self.client = openai.OpenAI(api_key=self.api_key) | |
| # Load the meta prompt from the specified file | |
| self.meta_prompt = self._load_meta_prompt(self.meta_prompt_file) | |
| def _load_meta_prompt(self, meta_prompt_file: str) -> str: | |
| """ | |
| Load the meta prompt from the specified file. | |
| :param meta_prompt_file: Path to the meta prompt file. | |
| :return: The meta prompt as a string. | |
| """ | |
| if not os.path.exists(meta_prompt_file): | |
| logging.error(f"Meta prompt file '{meta_prompt_file}' not found.") | |
| raise FileNotFoundError(f"Meta prompt file '{meta_prompt_file}' not found.") | |
| with open(meta_prompt_file, 'r', encoding='utf-8') as file: | |
| meta_prompt = file.read().strip() | |
| logging.info(f"Meta prompt loaded successfully from '{meta_prompt_file}'.") | |
| return meta_prompt | |
| def generate_response(self, query: str, history: list) -> list: | |
| """ | |
| Generate a response to a user query using the OpenAI API. | |
| This method interacts with the OpenAI API to create a response based on the user's query. | |
| It supports optional parameters for model configuration and handles errors gracefully. | |
| Args: | |
| query (str): The user query to respond to. | |
| history (list): The conversation history from the chatbot. | |
| Returns: | |
| list: A list of dictionaries representing the conversation, including the generated response. | |
| """ | |
| # Prepare the input for the API call | |
| input_data = [{"role": "developer", "content": self.meta_prompt}] if self.previous_response_id is None else [] | |
| input_data.append({"role": "user", "content": query}) | |
| # Validate the query | |
| if not query.strip(): | |
| logging.warning("Empty or invalid query received.") | |
| warning_message = "Please enter a valid query." | |
| input_data.append({"role": "assistant", "content": warning_message}) | |
| return history + input_data | |
| try: | |
| logging.info("Sending request to OpenAI API...") | |
| response = self.client.responses.create( | |
| model=self.DEFAULT_MODEL, | |
| previous_response_id=self.previous_response_id, | |
| input=input_data, | |
| tools=[{ | |
| "type": "file_search", | |
| "vector_store_ids": [self.vector_store_id], | |
| "max_num_results": self.DEFAULT_MAX_NUM_RESULTS | |
| }], | |
| temperature=self.DEFAULT_TEMPERATURE, | |
| max_output_tokens=self.DEFAULT_MAX_OUTPUT_TOKENS | |
| ) | |
| self.previous_response_id = response.id | |
| logging.info("Response received successfully.") | |
| input_data.append({"role": "assistant", "content": response.output_text}) | |
| return history + input_data | |
| except Exception as e: | |
| logging.error(f"An error occurred while generating a response: {e}") | |
| error_message = "Sorry, I couldn't generate a response at this time. Please try again later." | |
| input_data.append({"role": "assistant", "content": error_message}) | |
| return history + input_data | |