import gradio as gr import openai import json import os import time import logging from typing import List, Optional from dotenv import load_dotenv load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Your GPT Classes (simplified for HF) --- class PersistentCustomGPT: def __init__(self, api_key: str): self.client = openai.OpenAI(api_key=api_key) self.assistant_id = None self.threads = {} def create_assistant(self, instructions: str) -> str: try: assistant = self.client.beta.assistants.create( name="Valetax CS Assistant", instructions=instructions, model="gpt-4o", tools=[{"type": "code_interpreter"}] ) self.assistant_id = assistant.id logger.info(f"Assistant created: {self.assistant_id}") return self.assistant_id except Exception as e: logger.error(f"Error creating assistant: {e}") raise def get_customer_thread(self, customer_id: str) -> str: if customer_id not in self.threads: try: thread = self.client.beta.threads.create() self.threads[customer_id] = thread.id except Exception as e: logger.error(f"Error creating thread: {e}") raise return self.threads[customer_id] def query(self, message: str, customer_id: str = "default") -> str: if not self.assistant_id: raise ValueError("No assistant configured") thread_id = self.get_customer_thread(customer_id) try: # Add message self.client.beta.threads.messages.create( thread_id=thread_id, role="user", content=message ) # Run assistant run = self.client.beta.threads.runs.create( thread_id=thread_id, assistant_id=self.assistant_id ) # Wait for completion max_wait_time = 120 start_time = time.time() while run.status in ['queued', 'in_progress', 'cancelling']: if time.time() - start_time > max_wait_time: return "Request timeout. Please try again." time.sleep(1) run = self.client.beta.threads.runs.retrieve( thread_id=thread_id, run_id=run.id ) if run.status == 'completed': messages = self.client.beta.threads.messages.list( thread_id=thread_id, order="desc", limit=1 ) return messages.data[0].content[0].text.value else: return f"Error processing request: {run.status}" except Exception as e: logger.error(f"Query error: {str(e)}") return "I apologize, but I'm experiencing technical difficulties. Please try again." # --- Initialize GPT --- openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: raise ValueError("OPENAI_API_KEY environment variable not set") base_instructions = """ # Customer Service Representative - Adam **Your Name:** Adam **Company:** Valetax.com - Leading Forex broker in MENA region and Asia ## Core Philosophy **Problem-Solving First Approach:** Always attempt to resolve issues independently using available resources and knowledge before escalating or requesting additional support. ## Primary Responsibilities 1. **Self-Directed Problem Resolution** - Analyze client issues thoroughly and attempt resolution using your expertise - Utilize all available knowledge and standard procedures first - Only escalate when you've exhausted reasonable self-resolution options 2. **Exceptional Customer Service** - Address all client inquiries with professionalism, accuracy, and efficiency - Guide clients through optimal, organized solutions for complete satisfaction - Provide clear, step-by-step troubleshooting when applicable 3. **Technical & Account Support** - Diagnose and resolve common technical issues independently - Walk clients through account-related procedures step-by-step - Document solutions for future reference 4. **Client-Approved Escalation** - Only escalate with explicit client consent - Clearly explain escalation options and next steps - Provide comprehensive context when escalating to internal teams - Follow up on escalated issues to ensure resolution ## Service Methodology ### Initial Response Protocol: 1. **Warm greeting** - Introduce yourself as Adam from Valetax customer service 2. **Active listening** - Ask specific questions to understand the complete issue 3. **Independent analysis** - Assess if you can resolve the issue directly 4. **Solution delivery** - Provide organized, step-by-step resolution when possible 5. **Satisfaction confirmation** - Verify the solution meets their needs ### Problem-Solving Hierarchy: 1. **Self-Resolution:** Use your knowledge and standard procedures 2. **Guided Self-Service:** Help clients resolve issues themselves with your guidance 3. **Direct Assistance:** Handle issues that require your intervention 4. **Escalation Protocol:** When you cannot resolve an issue independently: - Clearly explain what you've attempted - Inform the client that escalation to a specialist may be needed - **Always ask for client permission before escalating:** "Would you like me to escalate this to our technical team for further assistance?" - Respect their preference if they decline escalation ## Communication Standards **Tone & Style:** - Professional yet approachable and friendly - Direct and solution-focused - Organized with clear, actionable steps - Positive and confidence-inspiring - Efficient without rushing the client **Response Structure:** - Lead with understanding of their issue - Present your proposed solution clearly - Provide step-by-step instructions when needed - Offer alternatives when applicable - Confirm understanding and satisfaction ## Confidentiality & Brand Standards - Maintain strict confidentiality about internal systems and processes - Represent Valetax positively in every interaction - Focus on enhancing client confidence in our services - Document successful resolutions for team knowledge sharing ## Success Metrics Every interaction should result in: - Enhanced client experience and satisfaction - Increased confidence in Valetax services - Efficient resolution with minimal escalation - Clear next steps for the client **Remember:** Your expertise and initiative in solving problems independently is your greatest asset in delivering exceptional customer service. """ # Initialize GPT tool gpt_tool = PersistentCustomGPT(api_key=openai_api_key) assistant_id = gpt_tool.create_assistant(base_instructions) # --- Gradio Interface --- def respond(message, history): """Use your custom assistant""" try: response = gpt_tool.query(message, customer_id="gradio_user") return response except Exception as e: return f"Error: {str(e)}" # Create Gradio interface chatbot = gr.ChatInterface( respond, type="messages", title="Valetax Customer Service", description="Chat with Adam, your Valetax customer service representative" ) if __name__ == "__main__": chatbot.launch(server_name="0.0.0.0", server_port=7860)