Building a chatbot with multiple AI providers using AgentControl configs

1
"""
2
Simple AI Chatbot
3
Multi-provider support: Anthropic, OpenAI, and Google
4
Direct API integration with automatic provider selection
5
"""
6
7
import os
8
import logging
9
from typing import Dict, List, Optional
10
from abc import ABC, abstractmethod
11
import dotenv
12
13
# AI Provider imports
14
import anthropic
15
import openai
16
import google.genai as genai
17
18
# Set up logging
19
logging.basicConfig(
20
    level=logging.INFO,
21
    format='%(asctime)s - %(levelname)s - %(message)s'
22
)
23
logger = logging.getLogger(__name__)
24
25
# Suppress HTTP request logs from libraries
26
logging.getLogger("httpx").setLevel(logging.WARNING)
27
logging.getLogger("httpcore").setLevel(logging.WARNING)
28
logging.getLogger("openai").setLevel(logging.WARNING)
29
logging.getLogger("anthropic").setLevel(logging.WARNING)
30
31
# Load environment variables
32
dotenv.load_dotenv()
33
34
35
class BaseAIProvider(ABC):
36
    """Base class for AI providers"""
37
38
    def __init__(self, api_key: Optional[str] = None):
39
        self.api_key = api_key
40
        self.client = self._initialize_client() if api_key else None
41
42
    @abstractmethod
43
    def _initialize_client(self):
44
        """Initialize the provider's client"""
45
        pass
46
47
    @abstractmethod
48
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
49
        """Send message to the AI provider"""
50
        pass
51
52
    def format_messages(self, messages: List[Dict], system_prompt: str) -> List[Dict]:
53
        """Default message formatting (can be overridden by providers)"""
54
        formatted = [{"role": "system", "content": system_prompt}] if system_prompt else []
55
        formatted.extend([{"role": msg["role"], "content": msg["content"]} for msg in messages])
56
        return formatted
57
58
    def extract_params(self, params: Dict) -> Dict:
59
        """Extract common parameters"""
60
        return {
61
            "temperature": params.get("temperature", 0.7),
62
            "max_tokens": params.get("max_tokens", 500)
63
        }
64
65
66
class AnthropicProvider(BaseAIProvider):
67
    """Anthropic Claude provider"""
68
69
    def _initialize_client(self):
70
        return anthropic.Anthropic(api_key=self.api_key)
71
72
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
73
        if not self.client:
74
            raise ValueError("Anthropic API key not configured")
75
76
        extracted_params = self.extract_params(params)
77
78
        response = self.client.messages.create(
79
            model=model,
80
            max_tokens=extracted_params["max_tokens"],
81
            temperature=extracted_params["temperature"],
82
            system=system_prompt,
83
            messages=messages
84
        )
85
86
        return response.content[0].text
87
88
89
class OpenAIProvider(BaseAIProvider):
90
    """OpenAI GPT provider"""
91
92
    def _initialize_client(self):
93
        return openai.OpenAI(api_key=self.api_key)
94
95
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
96
        if not self.client:
97
            raise ValueError("OpenAI API key not configured")
98
99
        formatted_messages = self.format_messages(messages, system_prompt)
100
        extracted_params = self.extract_params(params)
101
102
        response = self.client.chat.completions.create(
103
            model=model,
104
            messages=formatted_messages,
105
            **extracted_params
106
        )
107
108
        return response.choices[0].message.content
109
110
111
class GoogleProvider(BaseAIProvider):
112
    """Google Gemini provider"""
113
114
    def _initialize_client(self):
115
        # New SDK uses client instantiation with API key
116
        # The environment variable GEMINI_API_KEY is automatically picked up
117
        if self.api_key:
118
            import os
119
            os.environ['GEMINI_API_KEY'] = self.api_key
120
        return genai.Client()
121
122
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
123
        if not self.client:
124
            raise ValueError("Google API key not configured")
125
126
        extracted_params = self.extract_params(params)
127
128
        # Format conversation with system prompt
129
        contents = []
130
131
        # Add system prompt as context
132
        if system_prompt:
133
            contents.append(f"{system_prompt}\n")
134
135
        # Add conversation history
136
        for msg in messages:
137
            role = "User" if msg["role"] == "user" else "Assistant"
138
            contents.append(f"{role}: {msg['content']}")
139
140
        full_prompt = "\n".join(contents)
141
142
        # Use the new client API
143
        response = self.client.models.generate_content(
144
            model=model,
145
            contents=full_prompt,
146
            config={
147
                "temperature": extracted_params["temperature"],
148
                "max_output_tokens": extracted_params["max_tokens"],
149
            }
150
        )
151
152
        return response.text
153
154
155
class AIProviderRegistry:
156
    """Registry for AI providers with automatic initialization"""
157
158
    def __init__(self):
159
        self.providers = {
160
            "anthropic": AnthropicProvider(os.getenv("ANTHROPIC_API_KEY")),
161
            "openai": OpenAIProvider(os.getenv("OPENAI_API_KEY")),
162
            "google": GoogleProvider(os.getenv("GEMINI_API_KEY"))
163
        }
164
165
    def send_message(self, provider: str, model_id: str, messages: List[Dict],
166
                     system_prompt: str, parameters: Dict) -> str:
167
        """Route message to appropriate provider"""
168
        provider_name = provider.lower()
169
170
        if provider_name not in self.providers:
171
            raise ValueError(f"Unsupported provider: {provider}")
172
173
        provider_instance = self.providers[provider_name]
174
        return provider_instance.send_message(model_id, messages, system_prompt, parameters)
175
176
    def get_available_providers(self) -> List[str]:
177
        """Get list of configured providers"""
178
        return [name for name, provider in self.providers.items() if provider.api_key]
179
180
    def get_default_provider(self) -> tuple:
181
        """Get the default provider based on available API keys"""
182
        if os.getenv("ANTHROPIC_API_KEY"):
183
            return "anthropic", "claude-3-haiku-20240307"
184
        elif os.getenv("OPENAI_API_KEY"):
185
            return "openai", "chatgpt-4o-latest"
186
        elif os.getenv("GEMINI_API_KEY"):
187
            return "google", "gemini-2.5-flash-lite"
188
        else:
189
            raise ValueError("No AI provider API keys found")
190
191
192
def run_chatbot():
193
    """Main chatbot loop"""
194
    print("=" * 70)
195
    print("  Simple AI Chatbot")
196
    print("=" * 70)
197
    print("\nSupporting: Anthropic Claude, OpenAI GPT, Google Gemini")
198
    print("Type 'exit' or 'quit' to end the conversation\n")
199
200
    # Initialize AI provider registry
201
    try:
202
        ai_registry = AIProviderRegistry()
203
        available = ai_registry.get_available_providers()
204
205
        if not available:
206
            logger.error("No AI provider API keys found. Please configure at least one provider.")
207
            return
208
209
        # Get default provider
210
        provider, model_id = ai_registry.get_default_provider()
211
        logger.info(f"✓ Using {provider} with model {model_id}")
212
        logger.info(f"Available providers: {', '.join(available)}")
213
214
    except Exception as e:
215
        logger.error(f"Failed to initialize AI providers: {e}")
216
        return
217
218
    # Default system prompt
219
    system_prompt = "You are a helpful AI assistant. Provide clear, concise, and friendly responses."
220
221
    # Default parameters
222
    parameters = {
223
        "temperature": 0.7,
224
        "max_tokens": 500
225
    }
226
227
    conversation_history = []
228
229
    # Main chat loop
230
    while True:
231
        try:
232
            user_input = input("You: ").strip()
233
234
            if user_input.lower() in ['exit', 'quit', 'q']:
235
                print("\nGoodbye! Thanks for chatting.")
236
                break
237
238
            if not user_input:
239
                continue
240
241
            # Add user message to history
242
            conversation_history.append({"role": "user", "content": user_input})
243
244
            # Send to AI provider
245
            print("\nAssistant: ", end="", flush=True)
246
247
            response = ai_registry.send_message(
248
                provider=provider,
249
                model_id=model_id,
250
                messages=conversation_history,
251
                system_prompt=system_prompt,
252
                parameters=parameters
253
            )
254
255
            print(response)
256
257
            # Add assistant response to history
258
            conversation_history.append({"role": "assistant", "content": response})
259
260
        except KeyboardInterrupt:
261
            print("\n\nInterrupted. Goodbye!")
262
            break
263
        except Exception as e:
264
            logger.error(f"Error in chat loop: {e}")
265
            print(f"\nError: {e}")
266
267
            # Provide helpful guidance for common errors
268
            if "API key not valid" in str(e) and "googleapis.com" in str(e):
269
                print("\n💡 Tip: For Google Gemini, you need an API key from Google AI Studio:")
270
                print("   1. Go to https://aistudio.google.com/app/apikey")
271
                print("   2. Click 'Get API Key' and create a new key")
272
                print("   3. Add it to your .env file as GEMINI_API_KEY=your-key-here")
273
            elif "API key" in str(e).lower():
274
                print("\n💡 Tip: Check that your API key is correct and has the necessary permissions.")
275
276
277
if __name__ == "__main__":
278
    # Check for at least one AI provider key
279
    provider_keys = ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY"]
280
    if not any(os.getenv(key) for key in provider_keys):
281
        logger.error("No AI provider API keys found. Please add at least one:")
282
        for key in provider_keys:
283
            logger.error(f"  - {key}")
284
        exit(1)
285
286
    # Run the chatbot
287
    run_chatbot()

1
"""
2
Simple AI Chatbot with LaunchDarkly Targeting
3
Dynamic configuration and feature flagging
4
Supports user context-based provider and model selection
5
"""
6
7
import os
8
import logging
9
from typing import Dict, List, Any, Tuple, Optional
10
from abc import ABC, abstractmethod
11
import dotenv
12
13
# LaunchDarkly imports
14
import ldclient
15
from ldclient import Context
16
from ldclient.config import Config
17
from ldai import LDAIClient, AICompletionConfig, AICompletionConfigDefault, ModelConfig, LDMessage, ProviderConfig
18
19
# AI Provider imports
20
import anthropic
21
import openai
22
import google.genai as genai
23
24
# Set up logging
25
logging.basicConfig(
26
    level=logging.INFO,
27
    format='%(asctime)s - %(levelname)s - %(message)s'
28
)
29
logger = logging.getLogger(__name__)
30
31
# Suppress HTTP request logs from libraries
32
logging.getLogger("httpx").setLevel(logging.WARNING)
33
logging.getLogger("httpcore").setLevel(logging.WARNING)
34
logging.getLogger("openai").setLevel(logging.WARNING)
35
logging.getLogger("anthropic").setLevel(logging.WARNING)
36
37
# Load environment variables
38
dotenv.load_dotenv()
39
40
41
class LaunchDarklyAIClient:
42
    """Manages LaunchDarkly AgentControl configuration"""
43
44
    def __init__(self, sdk_key: str, agent_config_key: str):
45
        """Initialize LaunchDarkly client"""
46
        self.sdk_key = sdk_key
47
        self.agent_config_key = agent_config_key
48
        self.ld_client = None
49
        self.ai_client = None
50
51
        # Only initialize if we have a valid SDK key
52
        if sdk_key and sdk_key != "your-launchdarkly-sdk-key" and not sdk_key.startswith("your-"):
53
            try:
54
                ldclient.set_config(Config(sdk_key))
55
                self.ld_client = ldclient.get()
56
                self.ai_client = LDAIClient(self.ld_client)
57
                # Check if client initialized successfully
58
                if not self.ld_client.is_initialized():
59
                    logger.info("LaunchDarkly client not initialized, will use fallback configuration")
60
                    self.ld_client = None
61
                    self.ai_client = None
62
            except Exception as e:
63
                logger.info(f"LaunchDarkly initialization skipped: {e}")
64
                self.ld_client = None
65
                self.ai_client = None
66
        else:
67
            logger.info("No valid LaunchDarkly SDK key provided, using fallback configuration")
68
69
    def get_ai_config(self, user_context: Context, variables: Dict[str, Any] = None) -> AICompletionConfig:
70
        """Get AgentControl configuration for a specific user context"""
71
        fallback_config = self._get_fallback_config()
72
73
        if not self.ai_client:
74
            return fallback_config
75
76
        config = self.ai_client.completion_config(
77
            self.agent_config_key,
78
            user_context,
79
            fallback_config,
80
            variables or {}
81
        )
82
        return config
83
84
    def _get_fallback_config(self) -> AICompletionConfigDefault:
85
        """Fallback configuration when LaunchDarkly is unavailable"""
86
        # Detect which provider is available
87
        provider_name = "anthropic"  # default
88
        model_name = "claude-3-haiku-20240307"
89
90
        if os.getenv("ANTHROPIC_API_KEY"):
91
            provider_name = "anthropic"
92
            model_name = "claude-3.5-haiku-20241022"
93
        elif os.getenv("OPENAI_API_KEY"):
94
            provider_name = "openai"
95
            model_name = "chatgpt-4o-latest"
96
        elif os.getenv("GEMINI_API_KEY"):
97
            provider_name = "google"
98
            model_name = "gemini-2.5-flash-lite"
99
        else:
100
            logger.warning("No AI provider API keys found for fallback configuration")
101
102
        return AICompletionConfigDefault(
103
            enabled=True,
104
            model=ModelConfig(
105
                name=model_name,
106
                parameters={"temperature": 0.7, "max_tokens": 500}
107
            ),
108
            messages=[LDMessage(
109
                role="system",
110
                content="You are a helpful AI assistant. Provide clear, concise, and friendly responses."
111
            )],
112
            provider=ProviderConfig(name=provider_name)
113
        )
114
115
116
class BaseAIProvider(ABC):
117
    """Base class for AI providers"""
118
119
    def __init__(self, api_key: Optional[str] = None):
120
        self.api_key = api_key
121
        self.client = self._initialize_client() if api_key else None
122
123
    @abstractmethod
124
    def _initialize_client(self):
125
        """Initialize the provider's client"""
126
        pass
127
128
    @abstractmethod
129
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
130
        """Send message to the AI provider"""
131
        pass
132
133
    def format_messages(self, messages: List[Dict], system_prompt: str) -> List[Dict]:
134
        """Default message formatting (can be overridden by providers)"""
135
        formatted = [{"role": "system", "content": system_prompt}] if system_prompt else []
136
        formatted.extend([{"role": msg["role"], "content": msg["content"]} for msg in messages])
137
        return formatted
138
139
    def extract_params(self, params: Dict) -> Dict:
140
        """Extract common parameters"""
141
        return {
142
            "temperature": params.get("temperature", 0.7),
143
            "max_tokens": params.get("max_tokens", 500)
144
        }
145
146
147
class AnthropicProvider(BaseAIProvider):
148
    """Anthropic Claude provider"""
149
150
    def _initialize_client(self):
151
        return anthropic.Anthropic(api_key=self.api_key)
152
153
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
154
        if not self.client:
155
            raise ValueError("Anthropic API key not configured")
156
157
        extracted_params = self.extract_params(params)
158
159
        response = self.client.messages.create(
160
            model=model,
161
            max_tokens=extracted_params["max_tokens"],
162
            temperature=extracted_params["temperature"],
163
            system=system_prompt,
164
            messages=messages
165
        )
166
167
        return response.content[0].text
168
169
170
class OpenAIProvider(BaseAIProvider):
171
    """OpenAI GPT provider"""
172
173
    def _initialize_client(self):
174
        return openai.OpenAI(api_key=self.api_key)
175
176
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
177
        if not self.client:
178
            raise ValueError("OpenAI API key not configured")
179
180
        formatted_messages = self.format_messages(messages, system_prompt)
181
        extracted_params = self.extract_params(params)
182
183
        response = self.client.chat.completions.create(
184
            model=model,
185
            messages=formatted_messages,
186
            **extracted_params
187
        )
188
189
        return response.choices[0].message.content
190
191
192
class GoogleProvider(BaseAIProvider):
193
    """Google Gemini provider"""
194
195
    def _initialize_client(self):
196
        # New SDK uses client instantiation with API key
197
        # The environment variable GEMINI_API_KEY is automatically picked up
198
        if self.api_key:
199
            import os
200
            os.environ['GEMINI_API_KEY'] = self.api_key
201
        return genai.Client()
202
203
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict) -> str:
204
        if not self.client:
205
            raise ValueError("Google API key not configured")
206
207
        extracted_params = self.extract_params(params)
208
209
        # Format conversation with system prompt
210
        contents = []
211
212
        # Add system prompt as context
213
        if system_prompt:
214
            contents.append(f"{system_prompt}\n")
215
216
        # Add conversation history
217
        for msg in messages:
218
            role = "User" if msg["role"] == "user" else "Assistant"
219
            contents.append(f"{role}: {msg['content']}")
220
221
        full_prompt = "\n".join(contents)
222
223
        # Use the new client API
224
        response = self.client.models.generate_content(
225
            model=model,
226
            contents=full_prompt,
227
            config={
228
                "temperature": extracted_params["temperature"],
229
                "max_output_tokens": extracted_params["max_tokens"],
230
            }
231
        )
232
233
        return response.text
234
235
236
class AIProviderRegistry:
237
    """Registry for AI providers with automatic initialization"""
238
239
    def __init__(self):
240
        self.providers = {
241
            "anthropic": AnthropicProvider(os.getenv("ANTHROPIC_API_KEY")),
242
            "openai": OpenAIProvider(os.getenv("OPENAI_API_KEY")),
243
            "google": GoogleProvider(os.getenv("GEMINI_API_KEY"))
244
        }
245
246
    def send_message(self, provider: str, model_id: str, messages: List[Dict],
247
                     system_prompt: str, parameters: Dict) -> str:
248
        """Route message to appropriate provider"""
249
        provider_name = provider.lower()
250
251
        if provider_name not in self.providers:
252
            raise ValueError(f"Unsupported provider: {provider}")
253
254
        provider_instance = self.providers[provider_name]
255
        return provider_instance.send_message(model_id, messages, system_prompt, parameters)
256
257
    def get_available_providers(self) -> List[str]:
258
        """Get list of configured providers"""
259
        return [name for name, provider in self.providers.items() if provider.api_key]
260
261
262
def create_user_context(user_id: str, attributes: Dict[str, Any] = None) -> Context:
263
    """Create a LaunchDarkly context for a user"""
264
    builder = Context.builder(user_id)
265
    if attributes:
266
        for key, value in attributes.items():
267
            builder.set(key, value)
268
    return builder.build()
269
270
271
def run_chatbot():
272
    """Main chatbot loop"""
273
    print("=" * 70)
274
    print("  Simple AI Chatbot with LaunchDarkly Targeting")
275
    print("=" * 70)
276
    print("\nSupporting: Anthropic Claude, OpenAI GPT, Google Gemini")
277
    print("Type 'exit' or 'quit' to end the conversation\n")
278
279
    # Initialize clients
280
    try:
281
        ld_ai_client = LaunchDarklyAIClient(
282
            sdk_key=os.getenv("LD_SDK_KEY", ""),
283
            agent_config_key=os.getenv("LAUNCHDARKLY_AGENT_CONFIG_KEY", "default-config")
284
        )
285
        ai_registry = AIProviderRegistry()
286
287
        available = ai_registry.get_available_providers()
288
        logger.info(f"✓ Clients initialized. Available providers: {', '.join(available)}")
289
    except Exception as e:
290
        logger.error(f"Failed to initialize clients: {e}")
291
        return
292
293
    # Create user context
294
    user_context = create_user_context(
295
        user_id="demo-user-001",
296
        attributes={"name": "Demo User", "environment": "development"}
297
    )
298
299
    # Get initial config to show provider/model
300
    config = ld_ai_client.get_ai_config(user_context)
301
    logger.info(f"✓ Using {config.provider.name} with model {config.model.name}")
302
303
    conversation_history = []
304
305
    # Main chat loop
306
    while True:
307
        try:
308
            user_input = input("You: ").strip()
309
310
            if user_input.lower() in ['exit', 'quit', 'q']:
311
                print("\nGoodbye! Thanks for chatting.")
312
                break
313
314
            if not user_input:
315
                continue
316
317
            # Fetch fresh config from LaunchDarkly for each message
318
            config = ld_ai_client.get_ai_config(user_context)
319
320
            # Extract configuration
321
            provider = config.provider.name
322
            model_id = config.model.name
323
            system_prompt = config.messages[0].content if config.messages else "You are a helpful assistant."
324
325
326
            # Get model parameters
327
            model_params = config.model.parameters if hasattr(config.model, 'parameters') and config.model.parameters else {}
328
            parameters = {
329
                "temperature": model_params.get("temperature", 0.7),
330
                "max_tokens": model_params.get("max_tokens", 500)
331
            }
332
333
            # Add user message to history
334
            conversation_history.append({"role": "user", "content": user_input})
335
336
            # Send to AI provider
337
            print("\nAssistant: ", end="", flush=True)
338
339
            response = ai_registry.send_message(
340
                provider=provider,
341
                model_id=model_id,
342
                messages=conversation_history,
343
                system_prompt=system_prompt,
344
                parameters=parameters
345
            )
346
347
            print(response)
348
349
            # Add assistant response to history
350
            conversation_history.append({"role": "assistant", "content": response})
351
352
        except KeyboardInterrupt:
353
            print("\n\nInterrupted. Goodbye!")
354
            break
355
        except Exception as e:
356
            logger.error(f"Error in chat loop: {e}")
357
            print(f"\nError: {e}")
358
359
            # Provide helpful guidance for common errors
360
            if "API key not valid" in str(e) and "googleapis.com" in str(e):
361
                print("\n💡 Tip: For Google Gemini, you need an API key from Google AI Studio:")
362
                print("   1. Go to https://aistudio.google.com/app/apikey")
363
                print("   2. Click 'Get API Key' and create a new key")
364
                print("   3. Add it to your .env file as GEMINI_API_KEY=your-key-here")
365
            elif "API key" in str(e).lower():
366
                print("\n💡 Tip: Check that your API key is correct and has the necessary permissions.")
367
368
def run_chatbot_with_persona(persona: str = "business"):
369
    """
370
    Run chatbot with a specific persona context
371
372
    Args:
373
        persona: The persona to use (business, creative, or default)
374
    """
375
    print("=" * 70)
376
    print(f"  AI Chatbot - Persona: {persona.upper()}")
377
    print("=" * 70)
378
    print("\nType 'exit' to quit, 'switch' to change persona\n")
379
380
    # Initialize clients
381
    try:
382
        ld_ai_client = LaunchDarklyAIClient(
383
            sdk_key=os.getenv("LD_SDK_KEY", ""),
384
            agent_config_key=os.getenv("LAUNCHDARKLY_AGENT_CONFIG_KEY", "default-config")
385
        )
386
        ai_registry = AIProviderRegistry()
387
388
        available = ai_registry.get_available_providers()
389
        logger.info(f"✓ Clients initialized. Available providers: {', '.join(available)}")
390
    except Exception as e:
391
        logger.error(f"Failed to initialize clients: {e}")
392
        return False
393
394
    # Create user context with persona attribute
395
    user_context = create_user_context(
396
        user_id=f"{persona}-user-001",
397
        attributes={
398
            "persona": persona,
399
            "name": f"{persona.title()} User"
400
        }
401
    )
402
403
    # Conversation loop
404
    conversation_history = []
405
406
    while True:
407
        try:
408
            user_input = input("You: ").strip()
409
410
            if user_input.lower() in ['exit', 'quit']:
411
                print("\n👋 Goodbye!\n")
412
                break
413
414
            if user_input.lower() == 'switch':
415
                print("\n🔄 Switching persona...\n")
416
                return True
417
418
            if not user_input:
419
                continue
420
421
            # Fetch fresh config from LaunchDarkly for each message
422
            config = ld_ai_client.get_ai_config(user_context)
423
424
            # Extract configuration
425
            provider = config.provider.name
426
            model_id = config.model.name
427
            system_prompt = config.messages[0].content if config.messages else "You are a helpful assistant."
428
429
430
            # Get model parameters
431
            model_params = config.model.parameters if hasattr(config.model, 'parameters') and config.model.parameters else {}
432
            parameters = {
433
                "temperature": model_params.get("temperature", 0.7),
434
                "max_tokens": model_params.get("max_tokens", 500)
435
            }
436
437
            # Add user message to history
438
            conversation_history.append({"role": "user", "content": user_input})
439
440
            # Send to AI provider
441
            print("\nAssistant: ", end="", flush=True)
442
443
            response = ai_registry.send_message(
444
                provider=provider,
445
                model_id=model_id,
446
                messages=conversation_history,
447
                system_prompt=system_prompt,
448
                parameters=parameters
449
            )
450
451
            print(response + "\n")
452
453
            # Add assistant response to history
454
            conversation_history.append({"role": "assistant", "content": response})
455
456
        except KeyboardInterrupt:
457
            print("\n\n👋 Goodbye!\n")
458
            break
459
        except Exception as e:
460
            logger.error(f"Error in chat loop: {e}")
461
            print(f"\n❌ Error: {e}\n")
462
463
    return False
464
465
466
def main_with_personas():
467
    """Main entry point with persona selection"""
468
    print("\n" + "=" * 70)
469
    print("  LaunchDarkly AgentControl config - Persona Demo")
470
    print("=" * 70)
471
472
    personas = {
473
        "1": "business",
474
        "2": "creative",
475
        "3": None  # Default
476
    }
477
478
    while True:
479
        print("\nSelect a persona:")
480
        print("  1. Business (professional and concise)")
481
        print("  2. Creative (imaginative and engaging)")
482
        print("  3. Default (friendly and helpful)")
483
        print("  q. Quit")
484
485
        choice = input("\nYour choice (1-3, q): ").strip()
486
487
        if choice.lower() == 'q':
488
            print("\n👋 Goodbye!\n")
489
            break
490
491
        if choice not in personas:
492
            print("❌ Invalid choice. Please select 1-3 or q.")
493
            continue
494
495
        persona = personas[choice]
496
497
        # Run with selected persona or default
498
        if persona:
499
            should_switch = run_chatbot_with_persona(persona)
500
            if not should_switch:
501
                break
502
            # If should_switch is True, loop continues to persona selection
503
        else:
504
            run_chatbot()  # Run default chatbot (no persona context)
505
            break  # Default mode exits after session ends
506
507
if __name__ == "__main__":
508
    import sys
509
510
    # Check for LaunchDarkly configuration (optional - will use fallback if not provided)
511
    sdk_key = os.getenv("LD_SDK_KEY")
512
    config_key = os.getenv("LAUNCHDARKLY_AGENT_CONFIG_KEY", "default-config")
513
514
    if not sdk_key or sdk_key == "your-launchdarkly-sdk-key":
515
        logger.info("No LaunchDarkly SDK key found - using fallback configuration")
516
        logger.info("To use LaunchDarkly features, add LD_SDK_KEY to your .env file")
517
518
    # Check for at least one AI provider key
519
    provider_keys = ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY"]
520
    if not any(os.getenv(key) for key in provider_keys):
521
        logger.error("No AI provider API keys found. Please add at least one:")
522
        for key in provider_keys:
523
            logger.error(f"  - {key}")
524
        exit(1)
525
526
    # Check if persona mode is requested
527
    if len(sys.argv) > 1 and sys.argv[1] == "--personas":
528
        main_with_personas()
529
    else:
530
        # Default behavior - run original chatbot
531
        run_chatbot()

1
"""
2
Simple AI Chatbot with LaunchDarkly AgentControl config
3
Complete LaunchDarkly integration with targeting and metrics
4
Tracks token usage, response times, and success rates
5
"""
6
7
import os
8
import logging
9
import time
10
from typing import Dict, List, Any, Tuple, Optional
11
from abc import ABC, abstractmethod
12
import dotenv
13
14
# LaunchDarkly imports
15
import ldclient
16
from ldclient import Context
17
from ldclient.config import Config
18
from ldai import LDAIClient, AICompletionConfig, AICompletionConfigDefault, ModelConfig, LDMessage, ProviderConfig
19
from ldai_openai import get_ai_metrics_from_response
20
21
# AI Provider imports
22
import anthropic
23
import openai
24
import google.genai as genai
25
26
# Set up logging
27
logging.basicConfig(
28
    level=logging.INFO,
29
    format='%(asctime)s - %(levelname)s - %(message)s'
30
)
31
logger = logging.getLogger(__name__)
32
33
# Suppress HTTP request logs from libraries
34
logging.getLogger("httpx").setLevel(logging.WARNING)
35
logging.getLogger("httpcore").setLevel(logging.WARNING)
36
logging.getLogger("openai").setLevel(logging.WARNING)
37
logging.getLogger("anthropic").setLevel(logging.WARNING)
38
39
# Load environment variables
40
dotenv.load_dotenv()
41
42
43
class LaunchDarklyAIClient:
44
    """Manages LaunchDarkly AgentControl configuration"""
45
46
    def __init__(self, sdk_key: str, agent_config_key: str):
47
        """Initialize LaunchDarkly client"""
48
        self.sdk_key = sdk_key
49
        self.agent_config_key = agent_config_key
50
        self.ld_client = None
51
        self.ai_client = None
52
53
        # Only initialize if we have a valid SDK key
54
        if sdk_key and sdk_key != "your-launchdarkly-sdk-key" and not sdk_key.startswith("your-"):
55
            try:
56
                ldclient.set_config(Config(sdk_key))
57
                self.ld_client = ldclient.get()
58
                self.ai_client = LDAIClient(self.ld_client)
59
                # Check if client initialized successfully
60
                if not self.ld_client.is_initialized():
61
                    logger.info("LaunchDarkly client not initialized, will use fallback configuration")
62
                    self.ld_client = None
63
                    self.ai_client = None
64
            except Exception as e:
65
                logger.info(f"LaunchDarkly initialization skipped: {e}")
66
                self.ld_client = None
67
                self.ai_client = None
68
        else:
69
            logger.info("No valid LaunchDarkly SDK key provided, using fallback configuration")
70
71
    def get_ai_config(self, user_context: Context, variables: Dict[str, Any] = None) -> AICompletionConfig:
72
        """Get AgentControl configuration for a specific user context"""
73
        fallback_config = self._get_fallback_config()
74
75
        if not self.ai_client:
76
            return fallback_config
77
78
        config = self.ai_client.completion_config(
79
            self.agent_config_key,
80
            user_context,
81
            fallback_config,
82
            variables or {}
83
        )
84
        return config
85
86
    def _get_fallback_config(self) -> AICompletionConfigDefault:
87
        """Fallback configuration when LaunchDarkly is unavailable"""
88
        # Detect which provider is available
89
        provider_name = "anthropic"  # default
90
        model_name = "claude-3-haiku-20240307"
91
92
        if os.getenv("ANTHROPIC_API_KEY"):
93
            provider_name = "anthropic"
94
            model_name = "claude-3.5-haiku-20241022"
95
        elif os.getenv("OPENAI_API_KEY"):
96
            provider_name = "openai"
97
            model_name = "chatgpt-4o-latest"
98
        elif os.getenv("GEMINI_API_KEY"):
99
            provider_name = "google"
100
            model_name = "gemini-2.5-flash-lite"
101
        else:
102
            logger.warning("No AI provider API keys found for fallback configuration")
103
104
        return AICompletionConfigDefault(
105
            enabled=True,
106
            model=ModelConfig(
107
                name=model_name,
108
                parameters={"temperature": 0.7, "max_tokens": 500}
109
            ),
110
            messages=[LDMessage(
111
                role="system",
112
                content="You are a helpful AI assistant. Provide clear, concise, and friendly responses."
113
            )],
114
            provider=ProviderConfig(name=provider_name)
115
        )
116
117
118
class BaseAIProvider(ABC):
119
    """Base class for AI providers"""
120
121
    def __init__(self, api_key: Optional[str] = None):
122
        self.api_key = api_key
123
        self.client = self._initialize_client() if api_key else None
124
125
    @abstractmethod
126
    def _initialize_client(self):
127
        """Initialize the provider's client"""
128
        pass
129
130
    @abstractmethod
131
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict, tracker=None) -> str:
132
        """Send message to the AI provider"""
133
        pass
134
135
    def format_messages(self, messages: List[Dict], system_prompt: str) -> List[Dict]:
136
        """Default message formatting (can be overridden by providers)"""
137
        formatted = [{"role": "system", "content": system_prompt}] if system_prompt else []
138
        formatted.extend([{"role": msg["role"], "content": msg["content"]} for msg in messages])
139
        return formatted
140
141
    def extract_params(self, params: Dict) -> Dict:
142
        """Extract common parameters"""
143
        return {
144
            "temperature": params.get("temperature", 0.7),
145
            "max_tokens": params.get("max_tokens", 500)
146
        }
147
148
149
class AnthropicProvider(BaseAIProvider):
150
    """Anthropic Claude provider"""
151
152
    def _initialize_client(self):
153
        return anthropic.Anthropic(api_key=self.api_key)
154
155
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict, tracker=None) -> str:
156
        if not self.client:
157
            raise ValueError("Anthropic API key not configured")
158
159
        extracted_params = self.extract_params(params)
160
161
        if tracker:
162
            # Track API call duration and response metrics
163
            response = tracker.track_duration_of(
164
                lambda: self.client.messages.create(
165
                    model=model,
166
                    max_tokens=extracted_params["max_tokens"],
167
                    temperature=extracted_params["temperature"],
168
                    system=system_prompt,
169
                    messages=messages
170
                )
171
            )
172
173
            # Track token usage if available
174
            if hasattr(response, 'usage'):
175
                from ldai.tracker import TokenUsage
176
                token_usage = TokenUsage(
177
                    input=response.usage.input_tokens,
178
                    output=response.usage.output_tokens,
179
                    total=response.usage.input_tokens + response.usage.output_tokens
180
                )
181
                tracker.track_tokens(token_usage)
182
183
            tracker.track_success()
184
        else:
185
            response = self.client.messages.create(
186
                model=model,
187
                max_tokens=extracted_params["max_tokens"],
188
                temperature=extracted_params["temperature"],
189
                system=system_prompt,
190
                messages=messages
191
            )
192
193
        return response.content[0].text
194
195
196
class OpenAIProvider(BaseAIProvider):
197
    """OpenAI GPT provider"""
198
199
    def _initialize_client(self):
200
        return openai.OpenAI(api_key=self.api_key)
201
202
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict, tracker=None) -> str:
203
        if not self.client:
204
            raise ValueError("OpenAI API key not configured")
205
206
        formatted_messages = self.format_messages(messages, system_prompt)
207
        extracted_params = self.extract_params(params)
208
209
        if tracker:
210
            response = tracker.track_metrics_of(
211
                get_ai_metrics_from_response,
212
                lambda: self.client.chat.completions.create(
213
                    model=model,
214
                    messages=formatted_messages,
215
                    **extracted_params
216
                )
217
            )
218
        else:
219
            response = self.client.chat.completions.create(
220
                model=model,
221
                messages=formatted_messages,
222
                **extracted_params
223
            )
224
225
        return response.choices[0].message.content
226
227
228
class GoogleProvider(BaseAIProvider):
229
    """Google Gemini provider"""
230
231
    def _initialize_client(self):
232
        # New SDK uses client instantiation with API key
233
        # The environment variable GEMINI_API_KEY is automatically picked up
234
        if self.api_key:
235
            import os
236
            os.environ['GEMINI_API_KEY'] = self.api_key
237
        return genai.Client()
238
239
    def send_message(self, model: str, messages: List[Dict], system_prompt: str, params: Dict, tracker=None) -> str:
240
        if not self.client:
241
            raise ValueError("Google API key not configured")
242
243
        extracted_params = self.extract_params(params)
244
245
        # Format conversation with system prompt
246
        contents = []
247
248
        # Add system prompt as context
249
        if system_prompt:
250
            contents.append(f"{system_prompt}\n")
251
252
        # Add conversation history
253
        for msg in messages:
254
            role = "User" if msg["role"] == "user" else "Assistant"
255
            contents.append(f"{role}: {msg['content']}")
256
257
        full_prompt = "\n".join(contents)
258
259
        # Manual metrics tracking for Google Gemini
260
        start_time = time.time()
261
262
        # Use the new client API
263
        response = self.client.models.generate_content(
264
            model=model,
265
            contents=full_prompt,
266
            config={
267
                "temperature": extracted_params["temperature"],
268
                "max_output_tokens": extracted_params["max_tokens"],
269
            }
270
        )
271
272
        duration = time.time() - start_time
273
274
        if tracker:
275
            # Track duration and success
276
            tracker.track_duration(duration)
277
            tracker.track_success()
278
279
        return response.text
280
281
282
class AIProviderRegistry:
283
    """Registry for AI providers with automatic initialization"""
284
285
    def __init__(self):
286
        self.providers = {
287
            "anthropic": AnthropicProvider(os.getenv("ANTHROPIC_API_KEY")),
288
            "openai": OpenAIProvider(os.getenv("OPENAI_API_KEY")),
289
            "google": GoogleProvider(os.getenv("GEMINI_API_KEY"))
290
        }
291
292
    def send_message(self, provider: str, model_id: str, messages: List[Dict],
293
                     system_prompt: str, parameters: Dict, tracker=None) -> str:
294
        """Route message to appropriate provider"""
295
        provider_name = provider.lower()
296
297
        if provider_name not in self.providers:
298
            raise ValueError(f"Unsupported provider: {provider}")
299
300
        provider_instance = self.providers[provider_name]
301
        return provider_instance.send_message(model_id, messages, system_prompt, parameters, tracker)
302
303
    def get_available_providers(self) -> List[str]:
304
        """Get list of configured providers"""
305
        return [name for name, provider in self.providers.items() if provider.api_key]
306
307
308
def create_user_context(user_id: str, attributes: Dict[str, Any] = None) -> Context:
309
    """Create a LaunchDarkly context for a user"""
310
    builder = Context.builder(user_id)
311
    if attributes:
312
        for key, value in attributes.items():
313
            builder.set(key, value)
314
    return builder.build()
315
316
317
def run_chatbot():
318
    """Main chatbot loop with full tracking"""
319
    print("=" * 70)
320
    print("  Simple AI Chatbot with LaunchDarkly AgentControl config")
321
    print("=" * 70)
322
    print("\nSupporting: Anthropic Claude, OpenAI GPT, Google Gemini")
323
    print("Type 'exit' or 'quit' to end the conversation\n")
324
325
    # Initialize clients
326
    try:
327
        ld_ai_client = LaunchDarklyAIClient(
328
            sdk_key=os.getenv("LD_SDK_KEY", ""),
329
            agent_config_key=os.getenv("LAUNCHDARKLY_AGENT_CONFIG_KEY", "default-config")
330
        )
331
        ai_registry = AIProviderRegistry()
332
333
        available = ai_registry.get_available_providers()
334
        logger.info(f"✓ Clients initialized. Available providers: {', '.join(available)}")
335
    except Exception as e:
336
        logger.error(f"Failed to initialize clients: {e}")
337
        return
338
339
    # Create user context
340
    user_context = create_user_context(
341
        user_id="demo-user-001",
342
        attributes={"name": "Demo User", "environment": "development"}
343
    )
344
345
    # Get initial config to show provider/model
346
    config = ld_ai_client.get_ai_config(user_context)
347
    logger.info(f"✓ Using {config.provider.name} with model {config.model.name}")
348
349
    conversation_history = []
350
351
    # Main chat loop
352
    while True:
353
        try:
354
            user_input = input("You: ").strip()
355
356
            if user_input.lower() in ['exit', 'quit', 'q']:
357
                print("\nGoodbye! Thanks for chatting.")
358
                break
359
360
            if not user_input:
361
                continue
362
363
            # Fetch fresh config from LaunchDarkly for each message
364
            config = ld_ai_client.get_ai_config(user_context)
365
366
            # Extract configuration
367
            provider = config.provider.name
368
            model_id = config.model.name
369
            system_prompt = config.messages[0].content if config.messages else "You are a helpful assistant."
370
            tracker = config.create_tracker() if config.enabled else None
371
372
            # Get model parameters
373
            model_params = config.model.parameters if hasattr(config.model, 'parameters') and config.model.parameters else {}
374
            parameters = {
375
                "temperature": model_params.get("temperature", 0.7),
376
                "max_tokens": model_params.get("max_tokens", 500)
377
            }
378
379
            # Add user message to history
380
            conversation_history.append({"role": "user", "content": user_input})
381
382
            # Send to AI provider
383
            print("\nAssistant: ", end="", flush=True)
384
385
            response = ai_registry.send_message(
386
                provider=provider,
387
                model_id=model_id,
388
                messages=conversation_history,
389
                system_prompt=system_prompt,
390
                parameters=parameters,
391
                tracker=tracker
392
            )
393
394
            print(response)
395
396
            # Add assistant response to history
397
            conversation_history.append({"role": "assistant", "content": response})
398
399
        except KeyboardInterrupt:
400
            print("\n\nInterrupted. Goodbye!")
401
            break
402
        except Exception as e:
403
            logger.error(f"Error in chat loop: {e}")
404
            print(f"\nError: {e}")
405
406
            # Provide helpful guidance for common errors
407
            if "API key not valid" in str(e) and "googleapis.com" in str(e):
408
                print("\n💡 Tip: For Google Gemini, you need an API key from Google AI Studio:")
409
                print("   1. Go to https://aistudio.google.com/app/apikey")
410
                print("   2. Click 'Get API Key' and create a new key")
411
                print("   3. Add it to your .env file as GEMINI_API_KEY=your-key-here")
412
            elif "API key" in str(e).lower():
413
                print("\n💡 Tip: Check that your API key is correct and has the necessary permissions.")
414
415
def run_chatbot_with_persona(persona: str = "business"):
416
    """
417
    Run chatbot with a specific persona context
418
419
    Args:
420
        persona: The persona to use (business, creative, or default)
421
    """
422
    print("=" * 70)
423
    print(f"  AI Chatbot - Persona: {persona.upper()}")
424
    print("=" * 70)
425
    print("\nType 'exit' to quit, 'switch' to change persona\n")
426
427
    # Initialize clients
428
    try:
429
        ld_ai_client = LaunchDarklyAIClient(
430
            sdk_key=os.getenv("LD_SDK_KEY", ""),
431
            agent_config_key=os.getenv("LAUNCHDARKLY_AGENT_CONFIG_KEY", "default-config")
432
        )
433
        ai_registry = AIProviderRegistry()
434
435
        available = ai_registry.get_available_providers()
436
        logger.info(f"✓ Clients initialized. Available providers: {', '.join(available)}")
437
    except Exception as e:
438
        logger.error(f"Failed to initialize clients: {e}")
439
        return False
440
441
    # Create user context with persona attribute
442
    user_context = create_user_context(
443
        user_id=f"{persona}-user-001",
444
        attributes={
445
            "persona": persona,
446
            "name": f"{persona.title()} User"
447
        }
448
    )
449
450
    # Conversation loop
451
    conversation_history = []
452
453
    while True:
454
        try:
455
            user_input = input("You: ").strip()
456
457
            if user_input.lower() in ['exit', 'quit']:
458
                print("\n👋 Goodbye!\n")
459
                break
460
461
            if user_input.lower() == 'switch':
462
                print("\n🔄 Switching persona...\n")
463
                return True
464
465
            if not user_input:
466
                continue
467
468
            # Fetch fresh config from LaunchDarkly for each message
469
            config = ld_ai_client.get_ai_config(user_context)
470
471
            # Extract configuration
472
            provider = config.provider.name
473
            model_id = config.model.name
474
            system_prompt = config.messages[0].content if config.messages else "You are a helpful assistant."
475
            tracker = config.create_tracker() if config.enabled else None
476
477
            # Get model parameters
478
            model_params = config.model.parameters if hasattr(config.model, 'parameters') and config.model.parameters else {}
479
            parameters = {
480
                "temperature": model_params.get("temperature", 0.7),
481
                "max_tokens": model_params.get("max_tokens", 500)
482
            }
483
484
            # Add user message to history
485
            conversation_history.append({"role": "user", "content": user_input})
486
487
            # Send to AI provider
488
            print("\nAssistant: ", end="", flush=True)
489
490
            response = ai_registry.send_message(
491
                provider=provider,
492
                model_id=model_id,
493
                messages=conversation_history,
494
                system_prompt=system_prompt,
495
                parameters=parameters,
496
                tracker=tracker
497
            )
498
499
            print(response + "\n")
500
501
            # Add assistant response to history
502
            conversation_history.append({"role": "assistant", "content": response})
503
504
        except KeyboardInterrupt:
505
            print("\n\n👋 Goodbye!\n")
506
            break
507
        except Exception as e:
508
            logger.error(f"Error in chat loop: {e}")
509
            print(f"\n❌ Error: {e}\n")
510
511
    return False
512
513
514
def main_with_personas():
515
    """Main entry point with persona selection"""
516
    print("\n" + "=" * 70)
517
    print("  LaunchDarkly AgentControl config - Persona Demo")
518
    print("=" * 70)
519
520
    personas = {
521
        "1": "business",
522
        "2": "creative",
523
        "3": None  # Default
524
    }
525
526
    while True:
527
        print("\nSelect a persona:")
528
        print("  1. Business (professional and concise)")
529
        print("  2. Creative (imaginative and engaging)")
530
        print("  3. Default (friendly and helpful)")
531
        print("  q. Quit")
532
533
        choice = input("\nYour choice (1-3, q): ").strip()
534
535
        if choice.lower() == 'q':
536
            print("\n👋 Goodbye!\n")
537
            break
538
539
        if choice not in personas:
540
            print("❌ Invalid choice. Please select 1-3 or q.")
541
            continue
542
543
        persona = personas[choice]
544
545
        # Run with selected persona or default
546
        if persona:
547
            should_switch = run_chatbot_with_persona(persona)
548
            if not should_switch:
549
                break
550
            # If should_switch is True, loop continues to persona selection
551
        else:
552
            run_chatbot()  # Run default chatbot (no persona context)
553
            break  # Default mode exits after session ends
554
555
if __name__ == "__main__":
556
    import sys
557
558
    # Check for LaunchDarkly configuration (optional - will use fallback if not provided)
559
    sdk_key = os.getenv("LD_SDK_KEY")
560
    config_key = os.getenv("LAUNCHDARKLY_AGENT_CONFIG_KEY", "default-config")
561
562
    if not sdk_key or sdk_key == "your-launchdarkly-sdk-key":
563
        logger.info("No LaunchDarkly SDK key found - using fallback configuration")
564
        logger.info("To use LaunchDarkly features, add LD_SDK_KEY to your .env file")
565
566
    # Check for at least one AI provider key
567
    provider_keys = ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY"]
568
    if not any(os.getenv(key) for key in provider_keys):
569
        logger.error("No AI provider API keys found. Please add at least one:")
570
        for key in provider_keys:
571
            logger.error(f"  - {key}")
572
        exit(1)
573
574
    # Check if persona mode is requested
575
    if len(sys.argv) > 1 and sys.argv[1] == "--personas":
576
        main_with_personas()
577
    else:
578
        # Default behavior - run original chatbot
579
        run_chatbot()