Spaces:
Running
Running
| """ | |
| ElevenLabs Conversational AI Agent Integration | |
| ================================================ | |
| Module for interacting with ElevenLabs conversational AI agents. | |
| """ | |
| import os | |
| from dotenv import load_dotenv | |
| from elevenlabs.client import ElevenLabs | |
| from elevenlabs.conversational_ai.conversation import Conversation | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| # ============================================================================= | |
| # CONFIGURATION - Update these values with your ElevenLabs agent settings | |
| # ============================================================================= | |
| # Your ElevenLabs API key (stored in .env file) | |
| ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY") | |
| # Agent ID - Replace with your agent's ID from ElevenLabs dashboard | |
| AGENT_ID = os.getenv("ELEVENLABS_AGENT_ID", "your-agent-id-here") | |
| # Optional: Custom configuration for the agent | |
| AGENT_CONFIG = { | |
| # Add any custom configuration here if needed | |
| # Example: language, voice settings, etc. | |
| } | |
| # ============================================================================= | |
| # Agent Interaction | |
| # ============================================================================= | |
| class EULegislationAgent: | |
| """Wrapper class for ElevenLabs conversational agent.""" | |
| def __init__(self): | |
| """Initialize the ElevenLabs client and conversation.""" | |
| if not ELEVENLABS_API_KEY: | |
| raise ValueError( | |
| "ELEVENLABS_API_KEY not found. Please add it to your .env file:\n" | |
| "ELEVENLABS_API_KEY=your_api_key_here" | |
| ) | |
| self.client = ElevenLabs(api_key=ELEVENLABS_API_KEY) | |
| self.conversation = None | |
| def start_conversation(self): | |
| """Start or restart a conversation with the agent.""" | |
| try: | |
| self.conversation = Conversation( | |
| client=self.client, | |
| agent_id=AGENT_ID, | |
| requires_auth=False, # Set to True if your agent requires authentication | |
| audio_interface=None, # Set to an audio interface if using voice | |
| # Add any additional configuration here | |
| **AGENT_CONFIG | |
| ) | |
| # Start the session | |
| self.conversation.start_session() | |
| return True | |
| except Exception as e: | |
| print(f"Error starting conversation: {e}") | |
| return False | |
| def send_message(self, message: str) -> str: | |
| """ | |
| Send a message to the agent and get a response. | |
| Args: | |
| message: User's question or message | |
| Returns: | |
| str: Agent's response | |
| """ | |
| try: | |
| # Start conversation if not already started | |
| if self.conversation is None: | |
| if not self.start_conversation(): | |
| return "❌ Error: Could not connect to the agent. Please check your API key and Agent ID." | |
| # Send message and get response | |
| response = self.conversation.send_user_message(message) | |
| # Extract text from response | |
| # The response might be a generator or have different formats | |
| if hasattr(response, 'text'): | |
| return response.text | |
| elif isinstance(response, dict): | |
| if 'text' in response: | |
| return response['text'] | |
| elif 'content' in response: | |
| return response['content'] | |
| elif isinstance(response, str): | |
| return response | |
| else: | |
| # Try to convert to string | |
| return str(response) | |
| except Exception as e: | |
| error_msg = str(e) | |
| # Provide helpful error messages | |
| if "agent_id" in error_msg.lower() or "not found" in error_msg.lower(): | |
| return ( | |
| "❌ Error: Agent not found. Please verify your ELEVENLABS_AGENT_ID " | |
| "in the .env file or in src/utils/elevenlabs_agent.py" | |
| ) | |
| elif "api key" in error_msg.lower() or "unauthorized" in error_msg.lower(): | |
| return ( | |
| "❌ Error: Invalid API key. Please check your ELEVENLABS_API_KEY " | |
| "in the .env file." | |
| ) | |
| else: | |
| return f"❌ Error communicating with agent: {error_msg}" | |
| def end_conversation(self): | |
| """End the current conversation.""" | |
| if self.conversation: | |
| try: | |
| self.conversation.end_session() | |
| except Exception as e: | |
| print(f"Error ending conversation: {e}") | |
| self.conversation = None | |
| # Global agent instance | |
| _agent = None | |
| def get_agent(): | |
| """Get or create the global agent instance.""" | |
| global _agent | |
| if _agent is None: | |
| _agent = EULegislationAgent() | |
| return _agent | |
| def chat_with_agent(message: str) -> str: | |
| """ | |
| Send a message to the EU Legislation agent and get a response. | |
| This is the main function to be called from the UI. | |
| Args: | |
| message: User's question about EU agricultural legislation | |
| Returns: | |
| str: Agent's response | |
| """ | |
| agent = get_agent() | |
| return agent.send_message(message) | |
| def reset_conversation(): | |
| """Reset the conversation with the agent.""" | |
| global _agent | |
| if _agent: | |
| _agent.end_conversation() | |
| _agent = None | |