How to Design and Scale an Enterprise AI Bot

   # Create Azure Bot Service resource
   az bot create --resource-group ti-helpdesk-bot-rg --name tihelpdeskbot --kind webapp

   # app/main.py - MVP Version
   from fastapi import FastAPI
   from botbuilder.core import TurnContext, ActivityHandler

   app = FastAPI(title="TI Helpdesk Bot MVP", version="0.1.0")

   class BasicBot(ActivityHandler):
       async def on_message_activity(self, turn_context: TurnContext):
           await turn_context.send_activity(f"You said: {turn_context.activity.text}")

   # Simple FAQ system
   FAQ_DATABASE = {
       "password reset": "To reset your password, go to portal.company.com/reset",
       "vpn setup": "Download VPN client from it.company.com/vpn",
       "email issues": "For email problems, restart Outlook or contact IT"
   }

   def classify_intent(message: str) -> str:
       message_lower = message.lower()
       if any(word in message_lower for word in ["password", "reset"]):
           return "password_reset"
       elif any(word in message_lower for word in ["vpn", "network"]):
           return "vpn_help"
       return "unknown"

   def create_help_card():
       card = {
           "type": "AdaptiveCard",
           "body": [
               {"type": "TextBlock", "text": "How can I help you?"},
               {"type": "ActionSet", "actions": [
                   {"type": "Action.Submit", "title": "Password Reset", "data": {"action": "password"}},
                   {"type": "Action.Submit", "title": "VPN Help", "data": {"action": "vpn"}}
               ]}
           ]
       }
       return card

   async def handle_file_upload(turn_context: TurnContext):
       attachments = turn_context.activity.attachments
       for attachment in attachments:
           # Process document for knowledge base update
           await process_document(attachment)

   def process_document(file_path: str):
       # Extract text from various formats
       if file_path.endswith('.pdf'):
           text = extract_pdf_text(file_path)
       elif file_path.endswith('.docx'):
           text = extract_docx_text(file_path)

       # Simple chunking
       chunks = text.split('\n\n')
       return chunks

   def search_knowledge_base(query: str, documents: List[str]):
       # Simple keyword matching
       results = []
       for doc in documents:
           if any(word in doc.lower() for word in query.lower().split()):
               results.append(doc)
       return results[:5]  # Top 5 results

   # app/services/llm_service.py
   import openai

   class LLMService:
       def __init__(self):
           openai.api_key = settings.openai_api_key

       async def generate_response(self, query: str, context: str = "") -> str:
           prompt = f"""
           You are Pascal, a helpful IT support assistant.
           Context: {context}
           User Question: {query}

           Provide a helpful, professional response.
           """

           response = await openai.ChatCompletion.acreate(
               model="gpt-3.5-turbo",
               messages=[{"role": "user", "content": prompt}],
               max_tokens=500
           )
           return response.choices[0].message.content

   def should_use_llm(query: str) -> bool:
       simple_patterns = ["hello", "hi", "help", "menu"]
       return not any(pattern in query.lower() for pattern in simple_patterns)

   # app/services/vector_store.py
   import pinecone
   from sentence_transformers import SentenceTransformer

   class VectorStore:
       def __init__(self):
           pinecone.init(api_key=settings.pinecone_api_key)
           self.index = pinecone.Index("ti-helpdesk-bot-knowledge")
           self.encoder = SentenceTransformer('all-MiniLM-L6-v2')

       async def add_documents(self, documents: List[str]):
           embeddings = self.encoder.encode(documents)
           vectors = [(f"doc_{i}", embedding.tolist(), {"text": doc}) 
                     for i, (embedding, doc) in enumerate(zip(embeddings, documents))]
           self.index.upsert(vectors)

       async def search(self, query: str, k: int = 5):
           query_embedding = self.encoder.encode([query])
           results = self.index.query(
               vector=query_embedding[0].tolist(),
               top_k=k,
               include_metadata=True
           )
           return [match.metadata['text'] for match in results.matches]

   # app/services/rag_service.py
   class RAGService:
       def __init__(self, vector_store: VectorStore, llm_service: LLMService):
           self.vector_store = vector_store
           self.llm_service = llm_service

       async def answer_question(self, question: str) -> dict:
           # Retrieve relevant context
           context_docs = await self.vector_store.search(question, k=5)
           context = "\n".join(context_docs)

           # Generate response with context
           rag_prompt = f"""
           Based on the following context, answer the user's question accurately.
           If the context doesn't contain enough information, say so.

           Context:
           {context}

           Question: {question}

           Answer:
           """

           response = await self.llm_service.generate_response(rag_prompt)

           return {
               "answer": response,
               "sources": context_docs[:3],
               "confidence": self._calculate_confidence(context, question)
           }

   async def hybrid_search(self, query: str, filters: dict = None):
       # Semantic search
       semantic_results = await self.vector_search(query)

       # Keyword search
       keyword_results = await self.keyword_search(query)

       # Combine and rank results
       combined_results = self._combine_results(semantic_results, keyword_results)

       # Apply filters
       if filters:
           combined_results = self._apply_filters(combined_results, filters)

       return combined_results

   # app/services/auth_service.py
   class AuthService:
       def __init__(self):
           self.azure_ad_client = AzureADClient()

       async def validate_teams_token(self, token: str) -> dict:
           try:
               payload = jwt.decode(
                   token,
                   key=self.get_public_key(),
                   algorithms=["RS256"],
                   audience=settings.microsoft_app_id
               )
               return payload
           except jwt.InvalidTokenError:
               raise UnauthorizedError("Invalid token")

       async def check_user_permissions(self, user_id: str, action: str) -> bool:
           user_roles = await self.get_user_roles(user_id)
           return self.has_permission(user_roles, action)

   # app/middleware/rate_limit.py
   from slowapi import Limiter

   limiter = Limiter(key_func=get_user_id)

   @app.post("/api/messages")
   @limiter.limit("10/minute")
   async def handle_message(request: Request):
       # Process message
       pass

   # app/services/cache_service.py
   import redis

   class CacheService:
       def __init__(self):
           self.redis_client = redis.Redis(host=settings.redis_host)

       async def get_cached_response(self, query_hash: str) -> str:
           return await self.redis_client.get(f"response:{query_hash}")

       async def cache_response(self, query_hash: str, response: str, ttl: int = 3600):
           await self.redis_client.setex(f"response:{query_hash}", ttl, response)

   # app/db/database.py
   from sqlalchemy.pool import QueuePool

   engine = create_engine(
       settings.database_url,
       poolclass=QueuePool,
       pool_size=20,
       max_overflow=30,
       pool_pre_ping=True
   )

   # app/monitoring/metrics.py
   from prometheus_client import Counter, Histogram, Gauge

   REQUEST_COUNT = Counter('ti-helpdesk-bot_requests_total', 'Total requests', ['endpoint', 'method'])
   REQUEST_DURATION = Histogram('ti-helpdesk-bot_request_duration_seconds', 'Request duration')
   ACTIVE_CONVERSATIONS = Gauge('ti-helpdesk-bot_active_conversations', 'Active conversations')

   class MetricsMiddleware:
       async def __call__(self, request: Request, call_next):
           start_time = time.time()
           response = await call_next(request)
           duration = time.time() - start_time

           REQUEST_COUNT.labels(endpoint=request.url.path, method=request.method).inc()
           REQUEST_DURATION.observe(duration)

           return response

   # app/analytics/dashboard.py
   class AnalyticsDashboard:
       def get_usage_metrics(self, time_range: str) -> dict:
           return {
               "total_conversations": self.count_conversations(time_range),
               "average_response_time": self.avg_response_time(time_range),
               "user_satisfaction": self.satisfaction_score(time_range),
               "top_queries": self.top_queries(time_range),
               "resolution_rate": self.resolution_rate(time_range)
           }

   # tests/test_rag_service.py
   import pytest

   class TestRAGService:
       @pytest.mark.asyncio
       async def test_answer_accuracy(self):
           rag_service = RAGService()
           response = await rag_service.answer_question("How to reset password?")

           assert response["confidence"] > 0.8
           assert "password" in response["answer"].lower()
           assert len(response["sources"]) > 0

       @pytest.mark.asyncio
       async def test_response_time(self):
           start_time = time.time()
           response = await rag_service.answer_question("Test question")
           duration = time.time() - start_time

           assert duration < 5.0  # Must respond within 5 seconds

   # app/services/image_service.py
   import cv2
   from transformers import BlipProcessor, BlipForConditionalGeneration

   class ImageService:
       def __init__(self):
           self.processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base")
           self.model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base")

       async def analyze_image(self, image_url: str) -> dict:
           # Download and process image
           image = self.download_image(image_url)

           # Generate caption
           inputs = self.processor(image, return_tensors="pt")
           caption = self.model.generate(**inputs)

           # Extract text if present (OCR)
           text = self.extract_text_from_image(image)

           return {
               "caption": caption,
               "extracted_text": text,
               "analysis": self.analyze_technical_content(image)
           }

   # app/services/document_intelligence.py
   from azure.ai.formrecognizer import DocumentAnalysisClient

   class DocumentIntelligence:
       async def analyze_document(self, document_path: str) -> dict:
           # Extract structured data from forms, invoices, etc.
           client = DocumentAnalysisClient(endpoint=settings.azure_endpoint, credential=settings.azure_key)

           with open(document_path, "rb") as f:
               result = await client.begin_analyze_document("prebuilt-document", f).result()

           return {
               "tables": self.extract_tables(result),
               "key_value_pairs": self.extract_key_values(result),
               "text": result.content
           }

   # app/services/personalization_service.py
   class PersonalizationService:
       def __init__(self):
           self.user_profiles = {}

       async def update_user_profile(self, user_id: str, interaction: dict):
           profile = self.user_profiles.get(user_id, self.create_default_profile())

           # Update preferences based on interaction
           profile["expertise_level"] = self.infer_expertise(interaction)
           profile["preferred_detail_level"] = self.infer_detail_preference(interaction)
           profile["common_topics"] = self.update_topic_frequency(profile, interaction)

           self.user_profiles[user_id] = profile

       async def personalize_response(self, user_id: str, base_response: str) -> str:
           profile = self.user_profiles.get(user_id)
           if not profile:
               return base_response

           # Adjust response based on user preferences
           if profile["expertise_level"] == "beginner":
               return self.add_explanatory_context(base_response)
           elif profile["expertise_level"] == "expert":
               return self.add_technical_details(base_response)

           return base_response

   # app/integrations/servicenow.py
   class ServiceNowIntegration:
       def __init__(self):
           self.client = ServiceNowClient(
               instance=settings.servicenow_instance,
               username=settings.servicenow_user,
               password=settings.servicenow_password
           )

       async def create_ticket(self, issue_description: str, user_id: str) -> dict:
           ticket_data = {
               "short_description": self.extract_summary(issue_description),
               "description": issue_description,
               "caller_id": user_id,
               "category": self.classify_category(issue_description),
               "priority": self.determine_priority(issue_description)
           }

           result = await self.client.create("incident", ticket_data)
           return result

       async def check_ticket_status(self, ticket_number: str) -> dict:
           ticket = await self.client.get("incident", ticket_number)
           return {
               "status": ticket["state"],
               "assigned_to": ticket["assigned_to"],
               "last_update": ticket["sys_updated_on"]
           }

   # app/services/workflow_service.py
   class WorkflowService:
       def __init__(self):
           self.workflows = self.load_workflows()

       async def execute_workflow(self, workflow_name: str, context: dict) -> dict:
           workflow = self.workflows[workflow_name]
           result = {}

           for step in workflow["steps"]:
               step_result = await self.execute_step(step, context, result)
               result[step["name"]] = step_result

               if step_result.get("stop_workflow"):
                   break

           return result

       async def execute_step(self, step: dict, context: dict, previous_results: dict):
           step_type = step["type"]

           if step_type == "api_call":
               return await self.make_api_call(step["config"], context)
           elif step_type == "user_input":
               return await self.request_user_input(step["prompt"])
           elif step_type == "condition":
               return self.evaluate_condition(step["condition"], context)

           return {"status": "completed"}

   # app/ml/predictive_models.py
   from sklearn.ensemble import RandomForestClassifier
   import joblib

   class PredictiveAnalytics:
       def __init__(self):
           self.models = self.load_models()

       def predict_issue_escalation(self, conversation_features: dict) -> float:
           model = self.models["escalation_predictor"]
           features = self.extract_features(conversation_features)
           probability = model.predict_proba([features])[0][1]
           return probability

       def predict_user_satisfaction(self, interaction_history: list) -> float:
           model = self.models["satisfaction_predictor"]
           features = self.aggregate_interaction_features(interaction_history)
           score = model.predict([features])[0]
           return score

       def predict_knowledge_gaps(self, query_patterns: list) -> list:
           # Analyze query patterns to identify missing knowledge
           gap_analyzer = self.models["gap_analyzer"]
           gaps = gap_analyzer.identify_gaps(query_patterns)
           return gaps

   # app/ml/training_service.py
   class CustomModelTraining:
       def __init__(self):
           self.training_pipeline = TrainingPipeline()

       async def train_custom_model(self, organization_id: str, training_data: dict) -> dict:
           # Prepare organization-specific data
           processed_data = await self.preprocess_data(training_data, organization_id)

           # Select appropriate model architecture
           model_config = self.select_model_architecture(processed_data)

           # Train model
           training_job = await self.training_pipeline.start_training(
               data=processed_data,
               config=model_config,
               organization_id=organization_id
           )

           return {
               "job_id": training_job.id,
               "status": "training",
               "estimated_completion": training_job.estimated_completion
           }

       async def evaluate_model(self, model_id: str, test_data: dict) -> dict:
           model = await self.load_model(model_id)
           metrics = await model.evaluate(test_data)

           return {
               "accuracy": metrics.accuracy,
               "precision": metrics.precision,
               "recall": metrics.recall,
               "f1_score": metrics.f1_score
           }

   # app/services/tenant_service.py
   class TenantService:
       def __init__(self):
           self.tenant_configs = {}

       async def create_tenant(self, organization_info: dict) -> dict:
           tenant_id = self.generate_tenant_id()

           # Create isolated resources
           await self.create_tenant_database(tenant_id)
           await self.create_tenant_knowledge_base(tenant_id)
           await self.deploy_tenant_models(tenant_id, organization_info)

           tenant_config = {
               "tenant_id": tenant_id,
               "organization_name": organization_info["name"],
               "created_at": datetime.utcnow(),
               "subscription_tier": organization_info["tier"],
               "resource_limits": self.get_resource_limits(organization_info["tier"])
           }

           self.tenant_configs[tenant_id] = tenant_config
           return tenant_config

       async def route_request(self, request: Request) -> str:
           # Extract tenant ID from request
           tenant_id = self.extract_tenant_id(request)

           # Validate tenant exists and is active
           if not self.is_tenant_active(tenant_id):
               raise TenantNotFoundError(f"Tenant {tenant_id} not found or inactive")

           return tenant_id

   # app/governance/ai_governance.py
   class AIGovernanceService:
       def __init__(self):
           self.bias_detector = BiasDetector()
           self.explainability_engine = ExplainabilityEngine()
           self.audit_logger = AuditLogger()

       async def evaluate_model_fairness(self, model_id: str, test_data: dict) -> dict:
           # Test for various types of bias
           bias_metrics = await self.bias_detector.evaluate(model_id, test_data)

           return {
               "demographic_parity": bias_metrics.demographic_parity,
               "equalized_odds": bias_metrics.equalized_odds,
               "calibration": bias_metrics.calibration,
               "overall_fairness_score": bias_metrics.overall_score,
               "recommendations": bias_metrics.recommendations
           }

       async def explain_decision(self, model_id: str, input_data: dict) -> dict:
           explanation = await self.explainability_engine.explain(model_id, input_data)

           return {
               "decision": explanation.decision,
               "confidence": explanation.confidence,
               "key_factors": explanation.key_factors,
               "counterfactual": explanation.counterfactual_examples
           }

       async def log_ai_decision(self, decision_context: dict):
           await self.audit_logger.log({
               "timestamp": datetime.utcnow(),
               "model_id": decision_context["model_id"],
               "input_hash": self.hash_input(decision_context["input"]),
               "decision": decision_context["decision"],
               "confidence": decision_context["confidence"],
               "user_id": decision_context["user_id"]
           })