{"aiPlatform":"claude-code@2025.06","category":"development","commandName":"/ai-assistant","content":"---\nname: AI Assistant Development\ndescription: Expert tool for creating intelligent conversational interfaces, chatbots, and AI-powered applications with natural language understanding and context management.\nallowed_tools:\n  - filesystem      # Access training data and model files\n  - memory          # Track conversation patterns and user interactions\n  - sqlite          # Store conversation history and analytics\ntags:\n  - ai-assistant\n  - chatbot\n  - natural-language\n  - conversational-ai\n  - machine-learning\ncategory: development\nversion: 1.0.0\nauthor: AI Commands Team\n---\n\n# AI Assistant Development\n\nYou are an AI assistant development expert specializing in creating intelligent conversational interfaces, chatbots, and AI-powered applications. Design comprehensive AI assistant solutions with natural language understanding, context management, and seamless integrations.\n\n## Context\nThe user needs to develop an AI assistant or chatbot with natural language capabilities, intelligent responses, and practical functionality. Focus on creating production-ready assistants that provide real value to users.\n\n## Requirements\n$ARGUMENTS\n\n## Instructions\n\n### 1. AI Assistant Architecture\n\nDesign comprehensive assistant architecture:\n\n**Assistant Architecture Framework**\n```python\nfrom typing import Dict, List, Optional, Any\nfrom dataclasses import dataclass\nfrom abc import ABC, abstractmethod\nimport asyncio\n\n@dataclass\nclass ConversationContext:\n    \"\"\"Maintains conversation state and context\"\"\"\n    user_id: str\n    session_id: str\n    messages: List[Dict[str, Any]]\n    user_profile: Dict[str, Any]\n    conversation_state: Dict[str, Any]\n    metadata: Dict[str, Any]\n\nclass AIAssistantArchitecture:\n    def __init__(self, config: Dict[str, Any]):\n        self.config = config\n        self.components = self._initialize_components()\n        \n    def design_architecture(self):\n        \"\"\"Design comprehensive AI assistant architecture\"\"\"\n        return {\n            'core_components': {\n                'nlu': self._design_nlu_component(),\n                'dialog_manager': self._design_dialog_manager(),\n                'response_generator': self._design_response_generator(),\n                'context_manager': self._design_context_manager(),\n                'integration_layer': self._design_integration_layer()\n            },\n            'data_flow': self._design_data_flow(),\n            'deployment': self._design_deployment_architecture(),\n            'scalability': self._design_scalability_features()\n        }\n    \n    def _design_nlu_component(self):\n        \"\"\"Natural Language Understanding component\"\"\"\n        return {\n            'intent_recognition': {\n                'model': 'transformer-based classifier',\n                'features': [\n                    'Multi-intent detection',\n                    'Confidence scoring',\n                    'Fallback handling'\n                ],\n                'implementation': '''\nclass IntentClassifier:\n    def __init__(self, model_path: str):\n        self.model = self.load_model(model_path)\n        self.intents = self.load_intent_schema()\n    \n    async def classify(self, text: str) -> Dict[str, Any]:\n        # Preprocess text\n        processed = self.preprocess(text)\n        \n        # Get model predictions\n        predictions = await self.model.predict(processed)\n        \n        # Extract intents with confidence\n        intents = []\n        for intent, confidence in predictions:\n            if confidence > self.config['threshold']:\n                intents.append({\n                    'name': intent,\n                    'confidence': confidence,\n                    'parameters': self.extract_parameters(text, intent)\n                })\n        \n        return {\n            'intents': intents,\n            'primary_intent': intents[0] if intents else None,\n            'requires_clarification': len(intents) > 1\n        }\n'''\n            },\n            'entity_extraction': {\n                'model': 'NER with custom entities',\n                'features': [\n                    'Domain-specific entities',\n                    'Contextual extraction',\n                    'Entity resolution'\n                ]\n            },\n            'sentiment_analysis': {\n                'model': 'Fine-tuned sentiment classifier',\n                'features': [\n                    'Emotion detection',\n                    'Urgency classification',\n                    'User satisfaction tracking'\n                ]\n            }\n        }\n    \n    def _design_dialog_manager(self):\n        \"\"\"Dialog management system\"\"\"\n        return '''\nclass DialogManager:\n    \"\"\"Manages conversation flow and state\"\"\"\n    \n    def __init__(self):\n        self.state_machine = ConversationStateMachine()\n        self.policy_network = DialogPolicy()\n        \n    async def process_turn(self, \n                          context: ConversationContext, \n                          nlu_result: Dict[str, Any]) -> Dict[str, Any]:\n        # Determine current state\n        current_state = self.state_machine.get_state(context)\n        \n        # Apply dialog policy\n        action = await self.policy_network.select_action(\n            current_state, \n            nlu_result, \n            context\n        )\n        \n        # Execute action\n        result = await self.execute_action(action, context)\n        \n        # Update state\n        new_state = self.state_machine.transition(\n            current_state, \n            action, \n            result\n        )\n        \n        return {\n            'action': action,\n            'new_state': new_state,\n            'response_data': result\n        }\n    \n    async def execute_action(self, action: str, context: ConversationContext):\n        \"\"\"Execute dialog action\"\"\"\n        action_handlers = {\n            'greet': self.handle_greeting,\n            'provide_info': self.handle_information_request,\n            'clarify': self.handle_clarification,\n            'confirm': self.handle_confirmation,\n            'execute_task': self.handle_task_execution,\n            'end_conversation': self.handle_conversation_end\n        }\n        \n        handler = action_handlers.get(action, self.handle_unknown)\n        return await handler(context)\n'''\n```\n\n### 2. Natural Language Processing\n\nImplement advanced NLP capabilities:\n\n**NLP Pipeline Implementation**\n```python\nclass NLPPipeline:\n    def __init__(self):\n        self.tokenizer = self._initialize_tokenizer()\n        self.embedder = self._initialize_embedder()\n        self.models = self._load_models()\n    \n    async def process_message(self, message: str, context: ConversationContext):\n        \"\"\"Process user message through NLP pipeline\"\"\"\n        # Tokenization and preprocessing\n        tokens = self.tokenizer.tokenize(message)\n        \n        # Generate embeddings\n        embeddings = await self.embedder.embed(tokens)\n        \n        # Parallel processing of NLP tasks\n        tasks = [\n            self.detect_intent(embeddings),\n            self.extract_entities(tokens, embeddings),\n            self.analyze_sentiment(embeddings),\n            self.detect_language(tokens),\n            self.check_spelling(tokens)\n        ]\n        \n        results = await asyncio.gather(*tasks)\n        \n        return {\n            'intent': results[0],\n            'entities': results[1],\n            'sentiment': results[2],\n            'language': results[3],\n            'corrections': results[4],\n            'original_message': message,\n            'processed_tokens': tokens\n        }\n    \n    async def detect_intent(self, embeddings):\n        \"\"\"Advanced intent detection\"\"\"\n        # Multi-label classification\n        intent_scores = await self.models['intent_classifier'].predict(embeddings)\n        \n        # Hierarchical intent detection\n        primary_intent = self.get_primary_intent(intent_scores)\n        sub_intents = self.get_sub_intents(primary_intent, embeddings)\n        \n        return {\n            'primary': primary_intent,\n            'secondary': sub_intents,\n            'confidence': max(intent_scores.values()),\n            'all_scores': intent_scores\n        }\n    \n    def extract_entities(self, tokens, embeddings):\n        \"\"\"Extract and resolve entities\"\"\"\n        # Named Entity Recognition\n        entities = self.models['ner'].extract(tokens, embeddings)\n        \n        # Entity linking and resolution\n        resolved_entities = []\n        for entity in entities:\n            resolved = self.resolve_entity(entity)\n            resolved_entities.append({\n                'text': entity['text'],\n                'type': entity['type'],\n                'resolved_value': resolved['value'],\n                'confidence': resolved['confidence'],\n                'alternatives': resolved.get('alternatives', [])\n            })\n        \n        return resolved_entities\n    \n    def build_semantic_understanding(self, nlu_result, context):\n        \"\"\"Build semantic representation of user intent\"\"\"\n        return {\n            'user_goal': self.infer_user_goal(nlu_result, context),\n            'required_information': self.identify_missing_info(nlu_result),\n            'constraints': self.extract_constraints(nlu_result),\n            'preferences': self.extract_preferences(nlu_result, context)\n        }\n```\n\n### 3. Conversation Flow Design\n\nDesign intelligent conversation flows:\n\n**Conversation Flow Engine**\n```python\nclass ConversationFlowEngine:\n    def __init__(self):\n        self.flows = self._load_conversation_flows()\n        self.state_tracker = StateTracker()\n        \n    def design_conversation_flow(self):\n        \"\"\"Design multi-turn conversation flows\"\"\"\n        return {\n            'greeting_flow': {\n                'triggers': ['hello', 'hi', 'greetings'],\n                'nodes': [\n                    {\n                        'id': 'greet_user',\n                        'type': 'response',\n                        'content': self.personalized_greeting,\n                        'next': 'ask_how_to_help'\n                    },\n                    {\n                        'id': 'ask_how_to_help',\n                        'type': 'question',\n                        'content': \"How can I assist you today?\",\n                        'expected_intents': ['request_help', 'ask_question'],\n                        'timeout': 30,\n                        'timeout_action': 'offer_suggestions'\n                    }\n                ]\n            },\n            'task_completion_flow': {\n                'triggers': ['task_request'],\n                'nodes': [\n                    {\n                        'id': 'understand_task',\n                        'type': 'nlu_processing',\n                        'extract': ['task_type', 'parameters'],\n                        'next': 'check_requirements'\n                    },\n                    {\n                        'id': 'check_requirements',\n                        'type': 'validation',\n                        'validate': self.validate_task_requirements,\n                        'on_success': 'confirm_task',\n                        'on_missing': 'request_missing_info'\n                    },\n                    {\n                        'id': 'request_missing_info',\n                        'type': 'slot_filling',\n                        'slots': self.get_required_slots,\n                        'prompts': self.get_slot_prompts,\n                        'next': 'confirm_task'\n                    },\n                    {\n                        'id': 'confirm_task',\n                        'type': 'confirmation',\n                        'content': self.generate_task_summary,\n                        'on_confirm': 'execute_task',\n                        'on_deny': 'clarify_task'\n                    }\n                ]\n            }\n        }\n    \n    async def execute_flow(self, flow_id: str, context: ConversationContext):\n        \"\"\"Execute a conversation flow\"\"\"\n        flow = self.flows[flow_id]\n        current_node = flow['nodes'][0]\n        \n        while current_node:\n            result = await self.execute_node(current_node, context)\n            \n            # Determine next node\n            if result.get('user_input'):\n                next_node_id = self.determine_next_node(\n                    current_node, \n                    result['user_input'],\n                    context\n                )\n            else:\n                next_node_id = current_node.get('next')\n            \n            current_node = self.get_node(flow, next_node_id)\n            \n            # Update context\n            context.conversation_state.update(result.get('state_updates', {}))\n        \n        return context\n```\n\n### 4. Response Generation\n\nCreate intelligent response generation:\n\n**Response Generator**\n```python\nclass ResponseGenerator:\n    def __init__(self, llm_client=None):\n        self.llm = llm_client\n        self.templates = self._load_response_templates()\n        self.personality = self._load_personality_config()\n        \n    async def generate_response(self, \n                               intent: str, \n                               context: ConversationContext,\n                               data: Dict[str, Any]) -> str:\n        \"\"\"Generate contextual responses\"\"\"\n        \n        # Select response strategy\n        if self.should_use_template(intent):\n            response = self.generate_from_template(intent, data)\n        elif self.should_use_llm(intent, context):\n            response = await self.generate_with_llm(intent, context, data)\n        else:\n            response = self.generate_hybrid_response(intent, context, data)\n        \n        # Apply personality and tone\n        response = self.apply_personality(response, context)\n        \n        # Ensure response appropriateness\n        response = self.validate_response(response, context)\n        \n        return response\n    \n    async def generate_with_llm(self, intent, context, data):\n        \"\"\"Generate response using LLM\"\"\"\n        # Construct prompt\n        prompt = self.build_llm_prompt(intent, context, data)\n        \n        # Set generation parameters\n        params = {\n            'temperature': self.get_temperature(intent),\n            'max_tokens': 150,\n            'stop_sequences': ['\\n\\n', 'User:', 'Human:']\n        }\n        \n        # Generate response\n        response = await self.llm.generate(prompt, **params)\n        \n        # Post-process response\n        return self.post_process_llm_response(response)\n    \n    def build_llm_prompt(self, intent, context, data):\n        \"\"\"Build context-aware prompt for LLM\"\"\"\n        return f\"\"\"\nYou are a helpful AI assistant with the following characteristics:\n{self.personality.description}\n\nConversation history:\n{self.format_conversation_history(context.messages[-5:])}\n\nUser intent: {intent}\nRelevant data: {json.dumps(data, indent=2)}\n\nGenerate a helpful, concise response that:\n1. Addresses the user's intent\n2. Uses the provided data appropriately\n3. Maintains conversation continuity\n4. Follows the personality guidelines\n\nResponse:\"\"\"\n    \n    def generate_from_template(self, intent, data):\n        \"\"\"Generate response from templates\"\"\"\n        template = self.templates.get(intent)\n        if not template:\n            return self.get_fallback_response()\n        \n        # Select template variant\n        variant = self.select_template_variant(template, data)\n        \n        # Fill template slots\n        response = variant\n        for key, value in data.items():\n            response = response.replace(f\"{{{key}}}\", str(value))\n        \n        return response\n    \n    def apply_personality(self, response, context):\n        \"\"\"Apply personality traits to response\"\"\"\n        # Add personality markers\n        if self.personality.get('friendly'):\n            response = self.add_friendly_markers(response)\n        \n        if self.personality.get('professional'):\n            response = self.ensure_professional_tone(response)\n        \n        # Adjust based on user preferences\n        if context.user_profile.get('prefers_brief'):\n            response = self.make_concise(response)\n        \n        return response\n```\n\n### 5. Context Management\n\nImplement sophisticated context management:\n\n**Context Management System**\n```python\nclass ContextManager:\n    def __init__(self):\n        self.short_term_memory = ShortTermMemory()\n        self.long_term_memory = LongTermMemory()\n        self.working_memory = WorkingMemory()\n        \n    async def manage_context(self, \n                            new_input: Dict[str, Any],\n                            current_context: ConversationContext) -> ConversationContext:\n        \"\"\"Manage conversation context\"\"\"\n        \n        # Update conversation history\n        current_context.messages.append({\n            'role': 'user',\n            'content': new_input['message'],\n            'timestamp': datetime.now(),\n            'metadata': new_input.get('metadata', {})\n        })\n        \n        # Resolve references\n        resolved_input = await self.resolve_references(new_input, current_context)\n        \n        # Update working memory\n        self.working_memory.update(resolved_input, current_context)\n        \n        # Detect topic changes\n        topic_shift = self.detect_topic_shift(resolved_input, current_context)\n        if topic_shift:\n            current_context = self.handle_topic_shift(topic_shift, current_context)\n        \n        # Maintain entity state\n        current_context = self.update_entity_state(resolved_input, current_context)\n        \n        # Prune old context if needed\n        if len(current_context.messages) > self.config['max_context_length']:\n            current_context = self.prune_context(current_context)\n        \n        return current_context\n    \n    async def resolve_references(self, input_data, context):\n        \"\"\"Resolve pronouns and references\"\"\"\n        text = input_data['message']\n        \n        # Pronoun resolution\n        pronouns = self.extract_pronouns(text)\n        for pronoun in pronouns:\n            referent = self.find_referent(pronoun, context)\n            if referent:\n                text = text.replace(pronoun['text'], referent['resolved'])\n        \n        # Temporal reference resolution\n        temporal_refs = self.extract_temporal_references(text)\n        for ref in temporal_refs:\n            resolved_time = self.resolve_temporal_reference(ref, context)\n            text = text.replace(ref['text'], str(resolved_time))\n        \n        input_data['resolved_message'] = text\n        return input_data\n    \n    def maintain_entity_state(self):\n        \"\"\"Track entity states across conversation\"\"\"\n        return '''\nclass EntityStateTracker:\n    def __init__(self):\n        self.entities = {}\n        \n    def update_entity(self, entity_id: str, updates: Dict[str, Any]):\n        \"\"\"Update entity state\"\"\"\n        if entity_id not in self.entities:\n            self.entities[entity_id] = {\n                'id': entity_id,\n                'type': updates.get('type'),\n                'attributes': {},\n                'history': []\n            }\n        \n        # Record history\n        self.entities[entity_id]['history'].append({\n            'timestamp': datetime.now(),\n            'updates': updates\n        })\n        \n        # Apply updates\n        self.entities[entity_id]['attributes'].update(updates)\n    \n    def get_entity_state(self, entity_id: str) -> Optional[Dict[str, Any]]:\n        \"\"\"Get current entity state\"\"\"\n        return self.entities.get(entity_id)\n    \n    def query_entities(self, entity_type: str = None, **filters):\n        \"\"\"Query entities by type and attributes\"\"\"\n        results = []\n        for entity in self.entities.values():\n            if entity_type and entity['type'] != entity_type:\n                continue\n            \n            matches = True\n            for key, value in filters.items():\n                if entity['attributes'].get(key) != value:\n                    matches = False\n                    break\n            \n            if matches:\n                results.append(entity)\n        \n        return results\n'''\n```\n\n### 6. Integration with LLMs\n\nIntegrate with various LLM providers:\n\n**LLM Integration Layer**\n```python\nclass LLMIntegrationLayer:\n    def __init__(self):\n        self.providers = {\n            'openai': OpenAIProvider(),\n            'anthropic': AnthropicProvider(),\n            'local': LocalLLMProvider()\n        }\n        self.current_provider = None\n        \n    async def setup_llm_integration(self, provider: str, config: Dict[str, Any]):\n        \"\"\"Setup LLM integration\"\"\"\n        self.current_provider = self.providers[provider]\n        await self.current_provider.initialize(config)\n        \n        return {\n            'provider': provider,\n            'capabilities': self.current_provider.get_capabilities(),\n            'rate_limits': self.current_provider.get_rate_limits()\n        }\n    \n    async def generate_completion(self, \n                                 prompt: str,\n                                 system_prompt: str = None,\n                                 **kwargs):\n        \"\"\"Generate completion with fallback handling\"\"\"\n        try:\n            # Primary attempt\n            response = await self.current_provider.complete(\n                prompt=prompt,\n                system_prompt=system_prompt,\n                **kwargs\n            )\n            \n            # Validate response\n            if self.is_valid_response(response):\n                return response\n            else:\n                return await self.handle_invalid_response(prompt, response)\n                \n        except RateLimitError:\n            # Switch to fallback provider\n            return await self.use_fallback_provider(prompt, system_prompt, **kwargs)\n        except Exception as e:\n            # Log error and use cached response if available\n            return self.get_cached_response(prompt) or self.get_default_response()\n    \n    def create_function_calling_interface(self):\n        \"\"\"Create function calling interface for LLMs\"\"\"\n        return '''\nclass FunctionCallingInterface:\n    def __init__(self):\n        self.functions = {}\n        \n    def register_function(self, \n                         name: str,\n                         func: callable,\n                         description: str,\n                         parameters: Dict[str, Any]):\n        \"\"\"Register a function for LLM to call\"\"\"\n        self.functions[name] = {\n            'function': func,\n            'description': description,\n            'parameters': parameters\n        }\n    \n    async def process_function_call(self, llm_response):\n        \"\"\"Process function calls from LLM\"\"\"\n        if 'function_call' not in llm_response:\n            return llm_response\n        \n        function_name = llm_response['function_call']['name']\n        arguments = llm_response['function_call']['arguments']\n        \n        if function_name not in self.functions:\n            return {'error': f'Unknown function: {function_name}'}\n        \n        # Validate arguments\n        validated_args = self.validate_arguments(\n            function_name, \n            arguments\n        )\n        \n        # Execute function\n        result = await self.functions[function_name]['function'](**validated_args)\n        \n        # Return result for LLM to process\n        return {\n            'function_result': result,\n            'function_name': function_name\n        }\n'''\n```\n\n### 7. Testing Conversational AI\n\nImplement comprehensive testing:\n\n**Conversation Testing Framework**\n```python\nclass ConversationTestFramework:\n    def __init__(self):\n        self.test_suites = []\n        self.metrics = ConversationMetrics()\n        \n    def create_test_suite(self):\n        \"\"\"Create comprehensive test suite\"\"\"\n        return {\n            'unit_tests': self._create_unit_tests(),\n            'integration_tests': self._create_integration_tests(),\n            'conversation_tests': self._create_conversation_tests(),\n            'performance_tests': self._create_performance_tests(),\n            'user_simulation': self._create_user_simulation()\n        }\n    \n    def _create_conversation_tests(self):\n        \"\"\"Test multi-turn conversations\"\"\"\n        return '''\nclass ConversationTest:\n    async def test_multi_turn_conversation(self):\n        \"\"\"Test complete conversation flow\"\"\"\n        assistant = AIAssistant()\n        context = ConversationContext(user_id=\"test_user\")\n        \n        # Conversation script\n        conversation = [\n            {\n                'user': \"Hello, I need help with my order\",\n                'expected_intent': 'order_help',\n                'expected_action': 'ask_order_details'\n            },\n            {\n                'user': \"My order number is 12345\",\n                'expected_entities': [{'type': 'order_id', 'value': '12345'}],\n                'expected_action': 'retrieve_order'\n            },\n            {\n                'user': \"When will it arrive?\",\n                'expected_intent': 'delivery_inquiry',\n                'should_use_context': True\n            }\n        ]\n        \n        for turn in conversation:\n            # Send user message\n            response = await assistant.process_message(\n                turn['user'], \n                context\n            )\n            \n            # Validate intent detection\n            if 'expected_intent' in turn:\n                assert response['intent'] == turn['expected_intent']\n            \n            # Validate entity extraction\n            if 'expected_entities' in turn:\n                self.validate_entities(\n                    response['entities'], \n                    turn['expected_entities']\n                )\n            \n            # Validate context usage\n            if turn.get('should_use_context'):\n                assert 'order_id' in response['context_used']\n    \n    def test_error_handling(self):\n        \"\"\"Test error scenarios\"\"\"\n        error_cases = [\n            {\n                'input': \"askdjfkajsdf\",\n                'expected_behavior': 'fallback_response'\n            },\n            {\n                'input': \"I want to [REDACTED]\",\n                'expected_behavior': 'safety_response'\n            },\n            {\n                'input': \"Tell me about \" + \"x\" * 1000,\n                'expected_behavior': 'length_limit_response'\n            }\n        ]\n        \n        for case in error_cases:\n            response = assistant.process_message(case['input'])\n            assert response['behavior'] == case['expected_behavior']\n'''\n    \n    def create_automated_testing(self):\n        \"\"\"Automated conversation testing\"\"\"\n        return '''\nclass AutomatedConversationTester:\n    def __init__(self):\n        self.test_generator = TestCaseGenerator()\n        self.evaluator = ResponseEvaluator()\n        \n    async def run_automated_tests(self, num_tests: int = 100):\n        \"\"\"Run automated conversation tests\"\"\"\n        results = {\n            'total_tests': num_tests,\n            'passed': 0,\n            'failed': 0,\n            'metrics': {}\n        }\n        \n        for i in range(num_tests):\n            # Generate test case\n            test_case = self.test_generator.generate()\n            \n            # Run conversation\n            conversation_log = await self.run_conversation(test_case)\n            \n            # Evaluate results\n            evaluation = self.evaluator.evaluate(\n                conversation_log,\n                test_case['expectations']\n            )\n            \n            if evaluation['passed']:\n                results['passed'] += 1\n            else:\n                results['failed'] += 1\n                \n            # Collect metrics\n            self.update_metrics(results['metrics'], evaluation['metrics'])\n        \n        return results\n    \n    def generate_adversarial_tests(self):\n        \"\"\"Generate adversarial test cases\"\"\"\n        return [\n            # Ambiguous inputs\n            \"I want that thing we discussed\",\n            \n            # Context switching\n            \"Actually, forget that. Tell me about the weather\",\n            \n            # Multiple intents\n            \"Cancel my order and also update my address\",\n            \n            # Incomplete information\n            \"Book a flight\",\n            \n            # Contradictions\n            \"I want a vegetarian meal with bacon\"\n        ]\n'''\n```\n\n### 8. Deployment and Scaling\n\nDeploy and scale AI assistants:\n\n**Deployment Architecture**\n```python\nclass AssistantDeployment:\n    def create_deployment_architecture(self):\n        \"\"\"Create scalable deployment architecture\"\"\"\n        return {\n            'containerization': '''\n# Dockerfile for AI Assistant\nFROM python:3.11-slim\n\nWORKDIR /app\n\n# Install dependencies\nCOPY requirements.txt .\nRUN pip install --no-cache-dir -r requirements.txt\n\n# Copy application\nCOPY . .\n\n# Load models at build time\nRUN python -m app.model_loader\n\n# Expose port\nEXPOSE 8080\n\n# Health check\nHEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \\\n  CMD python -m app.health_check\n\n# Run application\nCMD [\"gunicorn\", \"--worker-class\", \"uvicorn.workers.UvicornWorker\", \\\n     \"--workers\", \"4\", \"--bind\", \"0.0.0.0:8080\", \"app.main:app\"]\n''',\n            'kubernetes_deployment': '''\napiVersion: apps/v1\nkind: Deployment\nmetadata:\n  name: ai-assistant\nspec:\n  replicas: 3\n  selector:\n    matchLabels:\n      app: ai-assistant\n  template:\n    metadata:\n      labels:\n        app: ai-assistant\n    spec:\n      containers:\n      - name: assistant\n        image: ai-assistant:latest\n        ports:\n        - containerPort: 8080\n        resources:\n          requests:\n            memory: \"2Gi\"\n            cpu: \"1000m\"\n          limits:\n            memory: \"4Gi\"\n            cpu: \"2000m\"\n        env:\n        - name: MODEL_CACHE_SIZE\n          value: \"1000\"\n        - name: MAX_CONCURRENT_SESSIONS\n          value: \"100\"\n        livenessProbe:\n          httpGet:\n            path: /health\n            port: 8080\n          periodSeconds: 10\n        readinessProbe:\n          httpGet:\n            path: /ready\n            port: 8080\n          periodSeconds: 5\n---\napiVersion: v1\nkind: Service\nmetadata:\n  name: ai-assistant-service\nspec:\n  selector:\n    app: ai-assistant\n  ports:\n  - port: 80\n    targetPort: 8080\n  type: LoadBalancer\n---\napiVersion: autoscaling/v2\nkind: HorizontalPodAutoscaler\nmetadata:\n  name: ai-assistant-hpa\nspec:\n  scaleTargetRef:\n    apiVersion: apps/v1\n    kind: Deployment\n    name: ai-assistant\n  minReplicas: 3\n  maxReplicas: 10\n  metrics:\n  - type: Resource\n    resource:\n      name: cpu\n      target:\n        type: Utilization\n        averageUtilization: 70\n  - type: Resource\n    resource:\n      name: memory\n      target:\n        type: Utilization\n        averageUtilization: 80\n''',\n            'caching_strategy': self._design_caching_strategy(),\n            'load_balancing': self._design_load_balancing()\n        }\n    \n    def _design_caching_strategy(self):\n        \"\"\"Design caching for performance\"\"\"\n        return '''\nclass AssistantCache:\n    def __init__(self):\n        self.response_cache = ResponseCache()\n        self.model_cache = ModelCache()\n        self.context_cache = ContextCache()\n        \n    async def get_cached_response(self, \n                                 message: str, \n                                 context_hash: str) -> Optional[str]:\n        \"\"\"Get cached response if available\"\"\"\n        cache_key = self.generate_cache_key(message, context_hash)\n        \n        # Check response cache\n        cached = await self.response_cache.get(cache_key)\n        if cached and not self.is_expired(cached):\n            return cached['response']\n        \n        return None\n    \n    def cache_response(self, \n                      message: str,\n                      context_hash: str,\n                      response: str,\n                      ttl: int = 3600):\n        \"\"\"Cache response with TTL\"\"\"\n        cache_key = self.generate_cache_key(message, context_hash)\n        \n        self.response_cache.set(\n            cache_key,\n            {\n                'response': response,\n                'timestamp': datetime.now(),\n                'ttl': ttl\n            }\n        )\n    \n    def preload_model_cache(self):\n        \"\"\"Preload frequently used models\"\"\"\n        models_to_cache = [\n            'intent_classifier',\n            'entity_extractor',\n            'response_generator'\n        ]\n        \n        for model_name in models_to_cache:\n            model = load_model(model_name)\n            self.model_cache.store(model_name, model)\n'''\n```\n\n### 9. Monitoring and Analytics\n\nMonitor assistant performance:\n\n**Assistant Analytics System**\n```python\nclass AssistantAnalytics:\n    def __init__(self):\n        self.metrics_collector = MetricsCollector()\n        self.analytics_engine = AnalyticsEngine()\n        \n    def create_monitoring_dashboard(self):\n        \"\"\"Create monitoring dashboard configuration\"\"\"\n        return {\n            'real_time_metrics': {\n                'active_sessions': 'gauge',\n                'messages_per_second': 'counter',\n                'response_time_p95': 'histogram',\n                'intent_accuracy': 'gauge',\n                'fallback_rate': 'gauge'\n            },\n            'conversation_metrics': {\n                'avg_conversation_length': 'gauge',\n                'completion_rate': 'gauge',\n                'user_satisfaction': 'gauge',\n                'escalation_rate': 'gauge'\n            },\n            'system_metrics': {\n                'model_inference_time': 'histogram',\n                'cache_hit_rate': 'gauge',\n                'error_rate': 'counter',\n                'resource_utilization': 'gauge'\n            },\n            'alerts': [\n                {\n                    'name': 'high_fallback_rate',\n                    'condition': 'fallback_rate > 0.2',\n                    'severity': 'warning'\n                },\n                {\n                    'name': 'slow_response_time',\n                    'condition': 'response_time_p95 > 2000',\n                    'severity': 'critical'\n                }\n            ]\n        }\n    \n    def analyze_conversation_quality(self):\n        \"\"\"Analyze conversation quality metrics\"\"\"\n        return '''\nclass ConversationQualityAnalyzer:\n    def analyze_conversations(self, time_range: str):\n        \"\"\"Analyze conversation quality\"\"\"\n        conversations = self.fetch_conversations(time_range)\n        \n        metrics = {\n            'intent_recognition': self.analyze_intent_accuracy(conversations),\n            'response_relevance': self.analyze_response_relevance(conversations),\n            'conversation_flow': self.analyze_conversation_flow(conversations),\n            'user_satisfaction': self.analyze_satisfaction(conversations),\n            'error_patterns': self.identify_error_patterns(conversations)\n        }\n        \n        return self.generate_quality_report(metrics)\n    \n    def identify_improvement_areas(self, analysis):\n        \"\"\"Identify areas for improvement\"\"\"\n        improvements = []\n        \n        # Low intent accuracy\n        if analysis['intent_recognition']['accuracy'] < 0.85:\n            improvements.append({\n                'area': 'Intent Recognition',\n                'issue': 'Low accuracy in intent detection',\n                'recommendation': 'Retrain intent classifier with more examples',\n                'priority': 'high'\n            })\n        \n        # High fallback rate\n        if analysis['conversation_flow']['fallback_rate'] > 0.15:\n            improvements.append({\n                'area': 'Coverage',\n                'issue': 'High fallback rate',\n                'recommendation': 'Expand training data for uncovered intents',\n                'priority': 'medium'\n            })\n        \n        return improvements\n'''\n```\n\n### 10. Continuous Improvement\n\nImplement continuous improvement cycle:\n\n**Improvement Pipeline**\n```python\nclass ContinuousImprovement:\n    def create_improvement_pipeline(self):\n        \"\"\"Create continuous improvement pipeline\"\"\"\n        return {\n            'data_collection': '''\nclass ConversationDataCollector:\n    async def collect_feedback(self, session_id: str):\n        \"\"\"Collect user feedback\"\"\"\n        feedback_prompt = {\n            'satisfaction': 'How satisfied were you with this conversation? (1-5)',\n            'resolved': 'Was your issue resolved?',\n            'improvements': 'How could we improve?'\n        }\n        \n        feedback = await self.prompt_user_feedback(\n            session_id, \n            feedback_prompt\n        )\n        \n        # Store feedback\n        await self.store_feedback({\n            'session_id': session_id,\n            'timestamp': datetime.now(),\n            'feedback': feedback,\n            'conversation_metadata': self.get_session_metadata(session_id)\n        })\n        \n        return feedback\n    \n    def identify_training_opportunities(self):\n        \"\"\"Identify conversations for training\"\"\"\n        # Find low-confidence interactions\n        low_confidence = self.find_low_confidence_interactions()\n        \n        # Find failed conversations\n        failed = self.find_failed_conversations()\n        \n        # Find highly-rated conversations\n        exemplary = self.find_exemplary_conversations()\n        \n        return {\n            'needs_improvement': low_confidence + failed,\n            'good_examples': exemplary\n        }\n''',\n            'model_retraining': '''\nclass ModelRetrainer:\n    async def retrain_models(self, new_data):\n        \"\"\"Retrain models with new data\"\"\"\n        # Prepare training data\n        training_data = self.prepare_training_data(new_data)\n        \n        # Validate data quality\n        validation_result = self.validate_training_data(training_data)\n        if not validation_result['passed']:\n            return {'error': 'Data quality check failed', 'issues': validation_result['issues']}\n        \n        # Retrain models\n        models_to_retrain = ['intent_classifier', 'entity_extractor']\n        \n        for model_name in models_to_retrain:\n            # Load current model\n            current_model = self.load_model(model_name)\n            \n            # Create new version\n            new_model = await self.train_model(\n                model_name,\n                training_data,\n                base_model=current_model\n            )\n            \n            # Evaluate new model\n            evaluation = await self.evaluate_model(\n                new_model,\n                self.get_test_set()\n            )\n            \n            # Deploy if improved\n            if evaluation['performance'] > current_model.performance:\n                await self.deploy_model(new_model, model_name)\n        \n        return {'status': 'completed', 'models_updated': models_to_retrain}\n''',\n            'a_b_testing': '''\nclass ABTestingFramework:\n    def create_ab_test(self, \n                      test_name: str,\n                      variants: List[Dict[str, Any]],\n                      metrics: List[str]):\n        \"\"\"Create A/B test for assistant improvements\"\"\"\n        test = {\n            'id': generate_test_id(),\n            'name': test_name,\n            'variants': variants,\n            'metrics': metrics,\n            'allocation': self.calculate_traffic_allocation(variants),\n            'duration': self.estimate_test_duration(metrics)\n        }\n        \n        # Deploy test\n        self.deploy_test(test)\n        \n        return test\n    \n    async def analyze_test_results(self, test_id: str):\n        \"\"\"Analyze A/B test results\"\"\"\n        data = await self.collect_test_data(test_id)\n        \n        results = {}\n        for metric in data['metrics']:\n            # Statistical analysis\n            analysis = self.statistical_analysis(\n                data['control'][metric],\n                data['variant'][metric]\n            )\n            \n            results[metric] = {\n                'control_mean': analysis['control_mean'],\n                'variant_mean': analysis['variant_mean'],\n                'lift': analysis['lift'],\n                'p_value': analysis['p_value'],\n                'significant': analysis['p_value'] < 0.05\n            }\n        \n        return results\n'''\n        }\n```\n\n## Output Format\n\n1. **Architecture Design**: Complete AI assistant architecture with components\n2. **NLP Implementation**: Natural language processing pipeline and models\n3. **Conversation Flows**: Dialog management and flow design\n4. **Response Generation**: Intelligent response creation with LLM integration\n5. **Context Management**: Sophisticated context and state management\n6. **Testing Framework**: Comprehensive testing for conversational AI\n7. **Deployment Guide**: Scalable deployment architecture\n8. **Monitoring Setup**: Analytics and performance monitoring\n9. **Improvement Pipeline**: Continuous improvement processes\n\nFocus on creating production-ready AI assistants that provide real value through natural conversations, intelligent responses, and continuous learning from user interactions.","contentHash":"79ebff0af493f10d8f4925519544c3b3305f64fe7b89335feb235d797725e5c6","copies":0,"createdAt":"2025-08-12T16:09:35.134Z","description":"Build production-ready AI assistants and chatbots","github":{"repoUrl":"https://github.com/Commands-com/commands","lastSyncDirection":"from-github","metadata":{"importedFrom":"github_repository","repoPrivate":false,"repoDefaultBranch":"main","connectedAt":"2025-08-12T16:09:35.134Z"},"importedAt":"2025-08-12T16:09:35.134Z","lastSyncAt":"2025-08-17T17:57:46.971Z","fileMapping":{"license":null,"readme":null,"assets":[],"mainFile":"tools/ai-assistant.md"},"selectedCommand":"ai-assistant","fileShas":{"mainFile":"23ed7af147310ed4fdb99b2ddfbe8efd714ab1be","yamlPath":"313647b1fb381389da33b7913e95baf617c4b392"},"branch":"main","connectionType":"commands_yaml","connected":true,"lastSyncCommit":"01591bc061d236bde47bf23b0f47e8afcf1a5144","importSource":"repository_import","installationId":"69232615","syncStatus":"synced"},"githubRepoUrl":"https://github.com/Commands-com/commands","id":"824dbfc6-5c0f-458f-b4c5-12ecdd77a14a","inputParameters":[{"defaultValue":"openai","name":"ai_provider","options":["openai","anthropic","google","azure","aws-bedrock","huggingface"],"description":"LLM provider to use","label":"AI Provider","type":"select","required":false},{"defaultValue":"chatbot","name":"assistant_type","options":["chatbot","code-assistant","support-agent","knowledge-base","task-automation"],"description":"Type of AI assistant","label":"Assistant Type","type":"select","required":false}],"instructions":"Build production-ready AI assistants and chatbots","likes":0,"mcp_search_content":"","organizationUsername":"commands-com","price":"free","search_content":"ai assistant build production-ready ai assistants and chatbots /ai-assistant development claude-code@2025.06","title":"AI Assistant","type":"command","updatedAt":"2025-08-17T17:57:46.971Z","userId":"W0V8NAw5AhWRwcuwSoFLOi1Yem83","visibility":"public","name":"ai-assistant","userInteraction":{"userHasStarred":false}}