Chatbots entreprise : de la conception au déploiement

┌─────────────────────────────────────────────────────────┐
│                    INTERFACES UTILISATEUR                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌─────────┐ │
│  │  Slack   │  │  Teams   │  │   Web    │  │  Email  │ │
│  └──────────┘  └──────────┘  └──────────┘  └─────────┘ │
└────────────────────────┬────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────┐
│                  ORCHESTRATION LAYER                     │
│  ┌────────────────────────────────────────────────────┐ │
│  │  Router / Intent Classification                    │ │
│  │  (Déterminer : RAG, Agent, Handoff human)         │ │
│  └────────────────────────────────────────────────────┘ │
│                                                          │
│  ┌──────────────────┐         ┌────────────────────┐   │
│  │   RAG Engine     │         │   Agentic Layer    │   │
│  │ - Retrieval      │         │ - Function calling │   │
│  │ - Reranking      │         │ - Tools/APIs       │   │
│  │ - Generation     │         │ - Multi-step       │   │
│  └──────────────────┘         └────────────────────┘   │
│                                                          │
│  ┌────────────────────────────────────────────────────┐ │
│  │  Guardrails (PII detection, moderation, limits)   │ │
│  └────────────────────────────────────────────────────┘ │
└────────────────────────┬────────────────────────────────┘
                         │
        ┌────────────────┼────────────────┐
        │                │                │
        ▼                ▼                ▼
┌──────────────┐  ┌─────────────┐  ┌──────────────┐
│  Vector DB   │  │   LLM API   │  │  Tools/APIs  │
│  (Pinecone,  │  │ (OpenAI,    │  │  - Jira      │
│   Qdrant)    │  │  Anthropic) │  │  - Slack API │
└──────────────┘  └─────────────┘  │  - CRM       │
                                    │  - HRIS      │
                                    └──────────────┘

# Exemple avec LangChain et Pinecone
from langchain.document_loaders import (
    ConfluenceLoader,
    NotionDBLoader,
    SlackDirectoryLoader,
    GoogleDriveLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from pinecone import Pinecone

# 1. Charger les documents depuis sources multiples
confluence_loader = ConfluenceLoader(
    url="https://votre-confluence.atlassian.net",
    username="[email protected]",
    api_key="xxx"
)
confluence_docs = confluence_loader.load()

notion_loader = NotionDBLoader(
    integration_token="secret_xxx",
    database_id="xxx"
)
notion_docs = notion_loader.load()

# 2. Fusionner toutes les sources
all_docs = confluence_docs + notion_docs

# 3. Chunking intelligent
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = text_splitter.split_documents(all_docs)

# 4. Enrichir avec métadonnées
for chunk in chunks:
    chunk.metadata["source_type"] = "confluence"  # ou "notion", etc.
    chunk.metadata["department"] = extract_department(chunk)  # RH, IT, etc.
    chunk.metadata["indexed_at"] = datetime.now().isoformat()

# 5. Créer embeddings et indexer
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
pc = Pinecone(api_key="xxx")
index = pc.Index("chatbot-entreprise")

# Batch upsert pour performance
batch_size = 100
for i in range(0, len(chunks), batch_size):
    batch = chunks[i:i+batch_size]
    texts = [c.page_content for c in batch]
    metadatas = [c.metadata for c in batch]

    vectors = embeddings.embed_documents(texts)
    ids = [f"doc_{i+j}" for j in range(len(batch))]

    index.upsert(vectors=list(zip(ids, vectors, metadatas)))

print(f"✅ {len(chunks)} chunks indexés dans Pinecone")

metadata = {
    "source": "confluence",
    "url": "https://...",
    "title": "Procédure de congés",
    "department": "RH",
    "last_updated": "2025-01-15",
    "author": "[email protected]",
    "access_level": "all",  # ou "managers_only", etc.
    "language": "fr",
    "doc_type": "procedure"  # vs "faq", "policy", etc.
}

def sync_confluence_incremental():
    """Synchroniser seulement les pages modifiées depuis dernière sync"""
    last_sync = get_last_sync_timestamp()

    # Récupérer pages modifiées
    modified_pages = confluence.cql(
        f"lastModified >= '{last_sync}' order by lastModified"
    )

    for page in modified_pages:
        # Supprimer anciens chunks de cette page
        index.delete(filter={"confluence_page_id": page.id})

        # Réindexer
        new_chunks = process_page(page)
        index.upsert(new_chunks)

    update_last_sync_timestamp()

from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import PromptTemplate

class EnterpriseChatbot:
    def __init__(self):
        # LLM principal (génération)
        self.llm = ChatOpenAI(
            model="gpt-4-turbo",
            temperature=0.3,  # Peu créatif, factuel
            max_tokens=1000
        )

        # Embeddings
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-large"
        )

        # Vector store
        self.vectorstore = Pinecone.from_existing_index(
            index_name="chatbot-entreprise",
            embedding=self.embeddings
        )

        # Mémoire conversationnelle (5 derniers tours)
        self.memory = ConversationBufferWindowMemory(
            k=5,
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )

        # Prompt système
        self.system_prompt = PromptTemplate(
            template="""Tu es l'assistant virtuel de TechCorp, spécialisé dans les questions RH et IT.

Contexte des documents pertinents :
{context}

Historique de conversation :
{chat_history}

Question actuelle :
{question}

Instructions :
1. Réponds UNIQUEMENT avec les informations du contexte fourni
2. Si l'info n'est pas dans le contexte, dis "Je n'ai pas trouvé cette information dans notre base de connaissance"
3. Cite toujours ta source (ex: "D'après le document X...")
4. Sois concis et professionnel
5. Si la question nécessite une action (créer ticket, etc.), dis-le clairement

Réponse:""",
            input_variables=["context", "chat_history", "question"]
        )

        # Chain RAG
        self.chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(
                search_type="mmr",  # Maximum Marginal Relevance
                search_kwargs={
                    "k": 5,  # Récupérer 5 chunks
                    "fetch_k": 20,  # Parmi 20 candidats (pour diversité)
                    "lambda_mult": 0.7  # Balance pertinence/diversité
                }
            ),
            memory=self.memory,
            combine_docs_chain_kwargs={"prompt": self.system_prompt},
            return_source_documents=True,
            verbose=True
        )

    def chat(self, user_message: str, user_id: str = None) -> dict:
        """
        Traiter un message utilisateur

        Returns:
            {
                "answer": str,
                "sources": List[dict],
                "confidence": float
            }
        """
        # Appliquer filtres métadonnées si besoin
        # (ex: un employé ne voit que docs publics)
        filters = self._get_filters_for_user(user_id)

        # Exécuter RAG
        response = self.chain({
            "question": user_message,
            "filter": filters  # Métadonnées filtering
        })

        # Extraire sources
        sources = []
        for doc in response["source_documents"]:
            sources.append({
                "title": doc.metadata.get("title"),
                "url": doc.metadata.get("url"),
                "excerpt": doc.page_content[:200]
            })

        # Calculer score de confiance (basé sur similarité)
        confidence = self._calculate_confidence(response["source_documents"])

        return {
            "answer": response["answer"],
            "sources": sources,
            "confidence": confidence
        }

    def _get_filters_for_user(self, user_id: str) -> dict:
        """Filtrer docs selon permissions utilisateur"""
        user_role = get_user_role(user_id)  # de votre DB

        if user_role == "employee":
            return {"access_level": "all"}
        elif user_role == "manager":
            return {"access_level": {"$in": ["all", "managers"]}}
        elif user_role == "admin":
            return {}  # Pas de filtre

    def _calculate_confidence(self, docs: list) -> float:
        """Score de confiance basé sur similarité"""
        if not docs:
            return 0.0

        # Supposons que docs[0] a le meilleur score
        # (Pinecone retourne par ordre de similarité)
        top_score = docs[0].metadata.get("score", 0.5)

        if top_score > 0.85:
            return 0.95
        elif top_score > 0.75:
            return 0.80
        elif top_score > 0.65:
            return 0.60
        else:
            return 0.40

from cohere import Client

class EnterpriseChatbotWithReranking(EnterpriseChatbot):
    def __init__(self):
        super().__init__()
        self.cohere = Client(api_key="xxx")

    def chat(self, user_message: str, user_id: str = None) -> dict:
        # 1. Retrieval initial (large, 20 chunks)
        initial_docs = self.vectorstore.similarity_search(
            user_message,
            k=20,
            filter=self._get_filters_for_user(user_id)
        )

        # 2. Reranking avec Cohere
        reranked = self.cohere.rerank(
            query=user_message,
            documents=[doc.page_content for doc in initial_docs],
            top_n=5,  # Garder top 5
            model="rerank-multilingual-v2.0"
        )

        # 3. Extraire top docs
        top_docs = [initial_docs[r.index] for r in reranked.results]

        # 4. Générer réponse avec contexte reranké
        context = "\n\n---\n\n".join([doc.page_content for doc in top_docs])

        response = self.llm.invoke(
            self.system_prompt.format(
                context=context,
                chat_history=self.memory.load_memory_variables({})["chat_history"],
                question=user_message
            )
        )

        return {
            "answer": response.content,
            "sources": self._format_sources(top_docs),
            "confidence": reranked.results[0].relevance_score
        }

# Mémoire avec résumé automatique pour longues conversations
from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-3.5-turbo"),
    max_token_limit=500,  # Résumer si > 500 tokens
    memory_key="chat_history",
    return_messages=True
)

# Exemple de conversation
"""
User: Comment demander mes congés ?
Bot: Tu peux les demander via l'outil RH dans l'intranet...

User: Et si je veux annuler ?
Bot: [Comprend que "annuler" = annuler congés grâce au contexte]

User: Combien de jours me restent ?
Bot: [Sait qu'on parle toujours de congés]
"""

from langchain.agents import Tool
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

# Tool 1: Consulter solde de congés
class LeaveBalanceInput(BaseModel):
    employee_id: str = Field(description="ID de l'employé")

def get_leave_balance(employee_id: str) -> str:
    """Consulter le solde de congés d'un employé"""
    # Appel à votre API RH
    response = requests.get(
        f"https://api-rh.company.com/employees/{employee_id}/leave-balance",
        headers={"Authorization": f"Bearer {RH_API_KEY}"}
    )
    data = response.json()

    return f"Solde de congés: {data['remaining_days']} jours restants sur {data['annual_allowance']} jours annuels."

leave_balance_tool = StructuredTool.from_function(
    func=get_leave_balance,
    name="get_leave_balance",
    description="Consulte le solde de congés d'un employé. À utiliser quand l'utilisateur demande combien de jours de congés il lui reste.",
    args_schema=LeaveBalanceInput
)

# Tool 2: Créer un ticket IT
class CreateTicketInput(BaseModel):
    title: str = Field(description="Titre du ticket")
    description: str = Field(description="Description détaillée du problème")
    priority: str = Field(description="Priorité: low, medium, high", default="medium")
    category: str = Field(description="Catégorie: hardware, software, network, access")

def create_it_ticket(title: str, description: str, priority: str, category: str) -> str:
    """Créer un ticket IT dans Jira"""
    jira = JIRA(
        server="https://company.atlassian.net",
        basic_auth=("[email protected]", JIRA_API_TOKEN)
    )

    issue = jira.create_issue(
        project="IT",
        summary=title,
        description=description,
        issuetype={"name": "Support"},
        priority={"name": priority.capitalize()},
        customfield_10050=category  # custom field "Category"
    )

    return f"✅ Ticket créé avec succès ! Numéro: {issue.key}\nSuivi: https://company.atlassian.net/browse/{issue.key}"

create_ticket_tool = StructuredTool.from_function(
    func=create_it_ticket,
    name="create_it_ticket",
    description="Créer un ticket de support IT. À utiliser quand l'utilisateur signale un problème technique.",
    args_schema=CreateTicketInput
)

# Tool 3: Rechercher un collègue
def search_colleague(query: str) -> str:
    """Rechercher un collègue dans l'annuaire"""
    # Recherche dans Active Directory / LDAP / DB
    results = search_in_directory(query)

    if not results:
        return "Aucun collègue trouvé pour cette recherche."

    output = "Voici les résultats:\n"
    for person in results[:5]:  # Max 5 résultats
        output += f"\n👤 {person['name']} - {person['title']}\n"
        output += f"   📧 {person['email']}\n"
        output += f"   📱 {person['phone']}\n"
        output += f"   🏢 {person['department']}\n"

    return output

search_colleague_tool = Tool(
    name="search_colleague",
    func=search_colleague,
    description="Recherche un collègue dans l'annuaire de l'entreprise par nom, département ou fonction."
)

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# Définir le prompt de l'agent
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", """Tu es l'assistant RH+IT de TechCorp.

Tu as accès à plusieurs outils pour aider les employés :
- Consulter le solde de congés
- Créer des tickets IT
- Rechercher des collègues dans l'annuaire

Avant d'utiliser un outil, confirme avec l'utilisateur si nécessaire.
Sois concis et professionnel.
Si tu ne peux pas aider, escalade vers un humain."""),

    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# Créer l'agent
tools = [leave_balance_tool, create_ticket_tool, search_colleague_tool]

agent = create_openai_functions_agent(
    llm=ChatOpenAI(model="gpt-4-turbo", temperature=0),
    tools=tools,
    prompt=agent_prompt
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,  # Éviter boucles infinies
    early_stopping_method="generate"  # Stopper si bloqué
)

# Utiliser l'agent
response = agent_executor.invoke({
    "input": "Combien de jours de congés me restent ? Mon employee_id est EMP12345",
    "chat_history": []
})

print(response["output"])
# Output: "Vous avez 18 jours de congés restants sur vos 25 jours annuels."

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class TicketWorkflowState(TypedDict):
    user_message: str
    ticket_created: bool
    ticket_id: str
    manager_notified: bool
    priority: str

def create_ticket_node(state: TicketWorkflowState) -> TicketWorkflowState:
    """Étape 1: Créer le ticket"""
    ticket_id = create_jira_ticket(state["user_message"], state["priority"])
    return {
        **state,
        "ticket_created": True,
        "ticket_id": ticket_id
    }

def notify_manager_node(state: TicketWorkflowState) -> TicketWorkflowState:
    """Étape 2: Notifier le manager si priorité haute"""
    if state["priority"] == "high":
        send_slack_message(
            channel="#it-urgent",
            message=f"🚨 Nouveau ticket urgent: {state['ticket_id']}"
        )
        return {**state, "manager_notified": True}
    return state

def route_after_ticket(state: TicketWorkflowState) -> str:
    """Router selon priorité"""
    if state["priority"] == "high":
        return "notify_manager"
    else:
        return END

# Construire le graphe
workflow = StateGraph(TicketWorkflowState)

workflow.add_node("create_ticket", create_ticket_node)
workflow.add_node("notify_manager", notify_manager_node)

workflow.set_entry_point("create_ticket")
workflow.add_conditional_edges(
    "create_ticket",
    route_after_ticket,
    {
        "notify_manager": "notify_manager",
        END: END
    }
)
workflow.add_edge("notify_manager", END)

app = workflow.compile()

# Exécuter le workflow
result = app.invoke({
    "user_message": "Mon ordinateur ne démarre plus",
    "ticket_created": False,
    "priority": "high"
})

SYSTEM_PROMPT = """Tu es Alex, l'assistant virtuel de TechCorp.

## Personnalité {#personnalité}
- 🤝 Amical mais professionnel (tu vs vous selon contexte)
- 💡 Proactif : suggère des solutions avant qu'on te les demande
- 🎯 Efficace : réponds directement, pas de blabla inutile
- 😊 Positif : même pour les mauvaises nouvelles, reste constructif
- 🧠 Humble : si tu ne sais pas, tu le dis (pas d'invention)

## Ton {#ton}
- Utilise "tu" pour collègues, "vous" pour C-level
- Émojis autorisés mais avec parcimonie (1-2 max par message)
- Pas de jargon technique sauf si nécessaire
- Phrases courtes (< 20 mots idéalement)

## Structure de réponse {#structure-de-réponse}
1. Réponse directe en 1-2 phrases
2. Détails si pertinent
3. Prochaines étapes / actions suggérées
4. Sources (avec liens)

## Exemples {#exemples}
❌ "Conformément aux dispositions de l'article 3.2 du règlement intérieur..."
✅ "D'après notre règlement intérieur, tu as droit à 25 jours de congés par an."

❌ "Je vais effectuer une requête dans la base de données vectorielle..."
✅ "Je cherche ça pour toi... [2 secondes] Voici ce que j'ai trouvé !"

## Quand escalader vers un humain {#quand-escalader-vers-un-humain}
- L'utilisateur est frustré (3+ messages sans résolution)
- Question hors de ton scope (ex: négociation salariale)
- Demande explicite ("Je veux parler à quelqu'un")
- Urgence critique (ex: "Mon ordi est en feu")
"""

def get_personalized_prompt(user_id: str) -> str:
    """Adapter le prompt selon profil utilisateur"""
    user = get_user_profile(user_id)

    base_prompt = SYSTEM_PROMPT

    # Adapter le tutoiement/vouvoiement
    if user["seniority"] in ["C-level", "VP"]:
        base_prompt += "\n\n⚠️ Cet utilisateur est senior. Utilise 'vous' et sois particulièrement concis."

    # Adapter la langue
    if user["language"] == "en":
        base_prompt = translate_to_english(base_prompt)

    # Historique de satisfaction
    if user["avg_csat"] < 3:
        base_prompt += "\n\n⚠️ Cet utilisateur a eu des expériences négatives. Sois extra attentif et propose rapidement d'escalader vers humain si besoin."

    return base_prompt

import re
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

class PIIGuardrail:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def detect_and_redact_pii(self, text: str) -> dict:
        """Détecter et anonymiser les PII"""
        # Analyse
        results = self.analyzer.analyze(
            text=text,
            language="fr",
            entities=[
                "PHONE_NUMBER",
                "EMAIL_ADDRESS",
                "CREDIT_CARD",
                "IBAN_CODE",
                "FR_SSN",  # Numéro de sécurité sociale français
                "PERSON",
                "LOCATION"
            ]
        )

        # Anonymiser
        anonymized = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results
        )

        return {
            "has_pii": len(results) > 0,
            "pii_types": [r.entity_type for r in results],
            "original": text,
            "anonymized": anonymized.text
        }

    def check_before_logging(self, text: str) -> bool:
        """Vérifier avant de logger (RGPD compliance)"""
        result = self.detect_and_redact_pii(text)

        if result["has_pii"]:
            # Logger version anonymisée seulement
            log_to_analytics(result["anonymized"])
            return False  # Ne pas logger l'original
        else:
            log_to_analytics(text)
            return True

# Utilisation
guardrail = PIIGuardrail()

user_message = "Mon email est [email protected] et mon tel 06 12 34 56 78"
result = guardrail.detect_and_redact_pii(user_message)

print(result["anonymized"])
# Output: "Mon email est <EMAIL_ADDRESS> et mon tel <PHONE_NUMBER>"

from openai import OpenAI

class ContentModerationGuardrail:
    def __init__(self):
        self.client = OpenAI()

    def is_safe(self, text: str) -> dict:
        """Vérifier si le contenu est approprié"""
        response = self.client.moderations.create(input=text)
        result = response.results[0]

        return {
            "is_safe": not result.flagged,
            "categories": {
                cat: score
                for cat, score in result.category_scores.items()
                if score > 0.01  # Seuil
            },
            "flagged_categories": [
                cat for cat, flagged in result.categories.items()
                if flagged
            ]
        }

    def check_and_block(self, text: str) -> tuple[bool, str]:
        """Bloquer si contenu inapproprié"""
        moderation = self.is_safe(text)

        if not moderation["is_safe"]:
            flagged = moderation["flagged_categories"]

            if "harassment" in flagged:
                return False, "❌ Message bloqué : langage inapproprié détecté."
            elif "violence" in flagged:
                return False, "❌ Message bloqué : contenu violent détecté."
            else:
                return False, "❌ Message bloqué : contenu non conforme à notre politique."

        return True, ""

# Utilisation
moderation = ContentModerationGuardrail()
is_ok, message = moderation.check_and_block("Message de l'utilisateur...")
if not is_ok:
    return {"error": message}

from redis import Redis
from datetime import datetime, timedelta

class RateLimitGuardrail:
    def __init__(self):
        self.redis = Redis(host='localhost', port=6379, db=0)

    def check_rate_limit(
        self,
        user_id: str,
        limit_per_hour: int = 50,
        limit_per_day: int = 200
    ) -> tuple[bool, str]:
        """Vérifier si l'utilisateur n'abuse pas"""
        now = datetime.now()
        hour_key = f"rate_limit:hour:{user_id}:{now.strftime('%Y%m%d%H')}"
        day_key = f"rate_limit:day:{user_id}:{now.strftime('%Y%m%d')}"

        # Incrémenter compteurs
        hour_count = self.redis.incr(hour_key)
        day_count = self.redis.incr(day_key)

        # Set expiration si nouveau
        if hour_count == 1:
            self.redis.expire(hour_key, 3600)  # 1h
        if day_count == 1:
            self.redis.expire(day_key, 86400)  # 24h

        # Vérifier limites
        if hour_count > limit_per_hour:
            return False, f"⏸️ Limite horaire atteinte ({limit_per_hour} messages/h). Réessaye dans {60 - now.minute} minutes."

        if day_count > limit_per_day:
            return False, f"⏸️ Limite journalière atteinte ({limit_per_day} messages/jour). Réessaye demain."

        return True, ""

    def check_cost_limit(
        self,
        user_id: str,
        estimated_tokens: int,
        cost_per_1k_tokens: float = 0.01
    ) -> tuple[bool, str]:
        """Vérifier budget mensuel utilisateur"""
        month_key = f"cost:{user_id}:{datetime.now().strftime('%Y%m')}"

        # Calculer coût
        cost = (estimated_tokens / 1000) * cost_per_1k_tokens

        # Ajouter au total mensuel
        total_cost = float(self.redis.get(month_key) or 0) + cost
        self.redis.setex(month_key, 86400 * 31, total_cost)  # 31 jours

        # Limite mensuelle : 50€ par utilisateur
        if total_cost > 50:
            return False, "💰 Budget mensuel dépassé. Contacte l'admin."

        return True, ""

# Middleware d'application
def chatbot_middleware(user_id: str, message: str):
    """Vérifier tous les guardrails avant de traiter"""
    rate_limiter = RateLimitGuardrail()
    moderation = ContentModerationGuardrail()
    pii = PIIGuardrail()

    # 1. Rate limiting
    ok, error = rate_limiter.check_rate_limit(user_id)
    if not ok:
        return {"error": error}

    # 2. Modération
    ok, error = moderation.check_and_block(message)
    if not ok:
        return {"error": error}

    # 3. PII detection (warning seulement)
    pii_result = pii.detect_and_redact_pii(message)
    if pii_result["has_pii"]:
        log_warning(f"PII detected from {user_id}: {pii_result['pii_types']}")

    # 4. Estimer coût
    estimated_tokens = len(message.split()) * 1.3  # Rough estimate
    ok, error = rate_limiter.check_cost_limit(user_id, estimated_tokens)
    if not ok:
        return {"error": error}

    # Tout OK, continuer
    return {"ok": True}

def detect_prompt_injection(user_input: str) -> bool:
    """Détecter tentatives de prompt injection"""

    # Patterns suspects
    injection_patterns = [
        r"ignore (previous|all) instructions",
        r"you are now",
        r"new instructions:",
        r"system:",
        r"<\|im_start\|>",
        r"<\|im_end\|>",
        r"disregard",
        r"forget everything",
        r"roleplay as",
    ]

    user_lower = user_input.lower()

    for pattern in injection_patterns:
        if re.search(pattern, user_lower):
            log_security_alert(f"Prompt injection attempt: {user_input}")
            return True

    return False

# Dans votre chatbot
if detect_prompt_injection(user_message):
    return {
        "answer": "❌ Ta demande semble suspecte. Si c'est une erreur, reformule-la.",
        "blocked": True
    }

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

# Initialiser l'app Slack
app = App(token=SLACK_BOT_TOKEN)

@app.event("app_mention")
def handle_mention(event, say, client):
    """Répondre quand le bot est mentionné"""
    user_id = event["user"]
    text = event["text"]
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])

    # Retirer la mention du bot du texte
    bot_user_id = client.auth_test()["user_id"]
    clean_text = re.sub(f"<@{bot_user_id}>", "", text).strip()

    # Indiquer que le bot est en train de taper
    client.chat_postMessage(
        channel=channel,
        thread_ts=thread_ts,
        text="🤔 Laisse-moi chercher ça pour toi..."
    )

    try:
        # Appeler votre chatbot
        response = chatbot.chat(clean_text, user_id=user_id)

        # Formater la réponse avec sources
        blocks = [
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": response["answer"]}
            }
        ]

        # Ajouter les sources
        if response["sources"]:
            sources_text = "*📚 Sources:*\n" + "\n".join([
                f"• <{s['url']}|{s['title']}>"
                for s in response["sources"]
            ])
            blocks.append({
                "type": "context",
                "elements": [{"type": "mrkdwn", "text": sources_text}]
            })

        # Ajouter boutons d'action
        blocks.append({
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "👍 Utile"},
                    "action_id": "feedback_positive",
                    "value": response.get("conversation_id")
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "👎 Pas utile"},
                    "action_id": "feedback_negative",
                    "value": response.get("conversation_id")
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "👤 Parler à un humain"},
                    "action_id": "escalate_to_human",
                    "value": response.get("conversation_id"),
                    "style": "danger"
                }
            ]
        })

        # Envoyer la réponse
        client.chat_update(
            channel=channel,
            ts=thread_ts,
            text=response["answer"],  # Fallback text
            blocks=blocks
        )

    except Exception as e:
        logging.error(f"Error in Slack handler: {e}")
        say(
            text="😞 Désolé, j'ai rencontré un problème. Un humain va t'aider.",
            thread_ts=thread_ts
        )
        notify_support_team(channel, thread_ts, clean_text, user_id)

@app.action("feedback_positive")
def handle_positive_feedback(ack, body, client):
    """Gérer feedback positif"""
    ack()
    conversation_id = body["actions"][0]["value"]
    log_feedback(conversation_id, rating=5)

    client.chat_postMessage(
        channel=body["channel"]["id"],
        thread_ts=body["message"]["ts"],
        text="Merci pour ton retour ! 🙏"
    )

@app.action("escalate_to_human")
def handle_escalation(ack, body, client):
    """Escalader vers support humain"""
    ack()

    user_id = body["user"]["id"]
    channel = body["channel"]["id"]
    thread_ts = body["message"]["ts"]

    # Notifier équipe support dans canal privé
    client.chat_postMessage(
        channel="#support-escalations",
        text=f"🆘 <@{user_id}> demande de l'aide humaine",
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Nouvelle escalation*\n<#{channel}> | <https://company.slack.com/archives/{channel}/p{thread_ts.replace('.', '')}|Voir la conversation>"
                }
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "✋ Je prends en charge"},
                        "action_id": "claim_escalation",
                        "value": f"{channel}:{thread_ts}:{user_id}"
                    }
                ]
            }
        ]
    )

    # Confirmer à l'utilisateur
    client.chat_postMessage(
        channel=channel,
        thread_ts=thread_ts,
        text="✅ Un membre de l'équipe va te répondre sous peu !"
    )

# Lancer le bot
if __name__ == "__main__":
    handler = SocketModeHandler(app, SLACK_APP_TOKEN)
    handler.start()

from botbuilder.core import BotFrameworkAdapter, BotFrameworkAdapterSettings, TurnContext
from botbuilder.schema import Activity, ActivityTypes
from aiohttp import web

# Configuration
SETTINGS = BotFrameworkAdapterSettings(
    app_id=TEAMS_APP_ID,
    app_password=TEAMS_APP_PASSWORD
)
ADAPTER = BotFrameworkAdapter(SETTINGS)

async def on_message_activity(turn_context: TurnContext):
    """Gérer message Teams"""
    user_message = turn_context.activity.text
    user_id = turn_context.activity.from_property.id

    # Typing indicator
    await turn_context.send_activity(Activity(type=ActivityTypes.typing))

    try:
        # Appeler chatbot
        response = chatbot.chat(user_message, user_id=user_id)

        # Créer carte adaptative pour réponse
        card = {
            "type": "AdaptiveCard",
            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
            "version": "1.4",
            "body": [
                {
                    "type": "TextBlock",
                    "text": response["answer"],
                    "wrap": True
                }
            ]
        }

        # Ajouter sources si disponibles
        if response["sources"]:
            card["body"].append({
                "type": "TextBlock",
                "text": "**Sources:**",
                "weight": "Bolder",
                "spacing": "Medium"
            })
            for source in response["sources"]:
                card["body"].append({
                    "type": "TextBlock",
                    "text": f"[{source['title']}]({source['url']})",
                    "wrap": True
                })

        # Ajouter actions
        card["actions"] = [
            {
                "type": "Action.Submit",
                "title": "👍 Utile",
                "data": {"feedback": "positive", "conv_id": response.get("conversation_id")}
            },
            {
                "type": "Action.Submit",
                "title": "👎 Pas utile",
                "data": {"feedback": "negative", "conv_id": response.get("conversation_id")}
            }
        ]

        # Envoyer
        await turn_context.send_activity(
            Activity(
                type=ActivityTypes.message,
                attachments=[{
                    "contentType": "application/vnd.microsoft.card.adaptive",
                    "content": card
                }]
            )
        )

    except Exception as e:
        logging.error(f"Teams bot error: {e}")
        await turn_context.send_activity("😞 Désolé, une erreur s'est produite.")

async def messages(req: web.Request) -> web.Response:
    """Endpoint pour Teams"""
    body = await req.json()
    activity = Activity().deserialize(body)

    auth_header = req.headers.get("Authorization", "")

    async def call_bot(turn_context: TurnContext):
        if turn_context.activity.type == ActivityTypes.message:
            await on_message_activity(turn_context)

    await ADAPTER.process_activity(activity, auth_header, call_bot)
    return web.Response(status=200)

# Serveur web
app = web.Application()
app.router.add_post("/api/messages", messages)
web.run_app(app, host="0.0.0.0", port=3978)

<!-- Embed dans votre site -->
<!DOCTYPE html>
<html>
<head>
    <title>Chatbot Support</title>
    <style>
        /* Widget en bas à droite */
        #chatbot-widget {
            position: fixed;
            bottom: 20px;
            right: 20px;
            width: 400px;
            height: 600px;
            border: 1px solid #ccc;
            border-radius: 12px;
            box-shadow: 0 4px 20px rgba(0,0,0,0.15);
            display: flex;
            flex-direction: column;
            background: white;
            z-index: 9999;
        }

        #chatbot-header {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white;
            padding: 16px;
            border-radius: 12px 12px 0 0;
            display: flex;
            justify-content: space-between;
            align-items: center;
        }

        #chatbot-messages {
            flex: 1;
            overflow-y: auto;
            padding: 16px;
            display: flex;
            flex-direction: column;
            gap: 12px;
        }

        .message {
            max-width: 80%;
            padding: 12px 16px;
            border-radius: 18px;
            line-height: 1.4;
        }

        .message.user {
            background: #667eea;
            color: white;
            align-self: flex-end;
            border-bottom-right-radius: 4px;
        }

        .message.bot {
            background: #f1f3f5;
            color: #333;
            align-self: flex-start;
            border-bottom-left-radius: 4px;
        }

        .message.bot .sources {
            margin-top: 8px;
            padding-top: 8px;
            border-top: 1px solid #ddd;
            font-size: 0.85em;
        }

        #chatbot-input {
            padding: 16px;
            border-top: 1px solid #e0e0e0;
            display: flex;
            gap: 8px;
        }

        #chatbot-input input {
            flex: 1;
            padding: 12px;
            border: 1px solid #ddd;
            border-radius: 24px;
            outline: none;
        }

        #chatbot-input button {
            padding: 12px 24px;
            background: #667eea;
            color: white;
            border: none;
            border-radius: 24px;
            cursor: pointer;
            font-weight: 600;
        }

        #chatbot-input button:hover {
            background: #5568d3;
        }

        .typing-indicator {
            display: inline-block;
        }

        .typing-indicator span {
            display: inline-block;
            width: 8px;
            height: 8px;
            border-radius: 50%;
            background: #999;
            margin: 0 2px;
            animation: typing 1.4s infinite;
        }

        .typing-indicator span:nth-child(2) { animation-delay: 0.2s; }
        .typing-indicator span:nth-child(3) { animation-delay: 0.4s; }

        @keyframes typing {
            0%, 60%, 100% { transform: translateY(0); }
            30% { transform: translateY(-10px); }
        }
    </style>
</head>
<body>
    <div id="chatbot-widget">
        <div id="chatbot-header">
            <div>
                <h3 style="margin:0;">💬 Support Assistant</h3>
                <small>En ligne - Répond en ~3s</small>
            </div>
            <button onclick="closeWidget()" style="background:none;border:none;color:white;font-size:24px;cursor:pointer;">×</button>
        </div>

        <div id="chatbot-messages">
            <div class="message bot">
                👋 Salut ! Comment puis-je t'aider aujourd'hui ?
            </div>
        </div>

        <div id="chatbot-input">
            <input type="text" id="user-input" placeholder="Pose ta question..." onkeypress="handleKeyPress(event)">
            <button onclick="sendMessage()">Envoyer</button>
        </div>
    </div>

    <script>
        const messagesDiv = document.getElementById('chatbot-messages');
        const userInput = document.getElementById('user-input');

        function addMessage(text, isUser = false, sources = []) {
            const msgDiv = document.createElement('div');
            msgDiv.className = `message ${isUser ? 'user' : 'bot'}`;

            let html = text;

            // Ajouter sources si disponibles
            if (!isUser && sources.length > 0) {
                html += '<div class="sources">📚 <strong>Sources:</strong><br>';
                sources.forEach(s => {
                    html += `• <a href="${s.url}" target="_blank">${s.title}</a><br>`;
                });
                html += '</div>';
            }

            msgDiv.innerHTML = html;
            messagesDiv.appendChild(msgDiv);
            messagesDiv.scrollTop = messagesDiv.scrollHeight;
        }

        function showTyping() {
            const typingDiv = document.createElement('div');
            typingDiv.className = 'message bot typing-indicator';
            typingDiv.id = 'typing-indicator';
            typingDiv.innerHTML = '<span></span><span></span><span></span>';
            messagesDiv.appendChild(typingDiv);
            messagesDiv.scrollTop = messagesDiv.scrollHeight;
        }

        function hideTyping() {
            const typing = document.getElementById('typing-indicator');
            if (typing) typing.remove();
        }

        async function sendMessage() {
            const text = userInput.value.trim();
            if (!text) return;

            // Afficher message utilisateur
            addMessage(text, true);
            userInput.value = '';

            // Typing indicator
            showTyping();

            try {
                // Appeler votre API
                const response = await fetch('/api/chat', {
                    method: 'POST',
                    headers: {'Content-Type': 'application/json'},
                    body: JSON.stringify({
                        message: text,
                        user_id: getUserId()  // Cookie/localStorage
                    })
                });

                const data = await response.json();

                hideTyping();
                addMessage(data.answer, false, data.sources || []);

            } catch (error) {
                hideTyping();
                addMessage('😞 Désolé, une erreur s\'est produite. Réessaye dans un moment.', false);
            }
        }

        function handleKeyPress(e) {
            if (e.key === 'Enter') sendMessage();
        }

        function closeWidget() {
            document.getElementById('chatbot-widget').style.display = 'none';
        }

        function getUserId() {
            // Récupérer ou créer ID utilisateur (cookie/localStorage)
            let userId = localStorage.getItem('chatbot_user_id');
            if (!userId) {
                userId = 'user_' + Math.random().toString(36).substr(2, 9);
                localStorage.setItem('chatbot_user_id', userId);
            }
            return userId;
        }
    </script>
</body>
</html>

class EscalationDetector:
    def should_escalate(
        self,
        conversation_history: list,
        last_response: dict,
        user_message: str
    ) -> tuple[bool, str]:
        """Détecter si escalade nécessaire"""

        # 1. Demande explicite
        escalation_keywords = [
            "parler à quelqu'un",
            "agent humain",
            "vrai personne",
            "manager",
            "support",
            "pas satisfait"
        ]
        if any(kw in user_message.lower() for kw in escalation_keywords):
            return True, "explicit_request"

        # 2. Frustration détectée
        frustration_keywords = [
            "inutile",
            "ça marche pas",
            "encore",
            "toujours pas",
            "en boucle"
        ]
        if any(kw in user_message.lower() for kw in frustration_keywords):
            return True, "frustration"

        # 3. Confiance faible du bot
        if last_response.get("confidence", 1.0) < 0.4:
            return True, "low_confidence"

        # 4. Boucle (mêmes questions répétées)
        if len(conversation_history) >= 6:
            recent_questions = [msg["user"] for msg in conversation_history[-6:]]
            # Similarité entre questions
            if self._detect_loop(recent_questions):
                return True, "conversation_loop"

        # 5. Hors scope
        out_of_scope_indicators = [
            "salaire",
            "augmentation",
            "licenciement",
            "confidentiel"
        ]
        if any(ind in user_message.lower() for ind in out_of_scope_indicators):
            return True, "out_of_scope"

        return False, ""

    def _detect_loop(self, questions: list) -> bool:
        """Détecter si utilisateur répète les mêmes questions"""
        from difflib import SequenceMatcher

        for i in range(len(questions)-1):
            for j in range(i+1, len(questions)):
                similarity = SequenceMatcher(None, questions[i], questions[j]).ratio()
                if similarity > 0.7:
                    return True
        return False

# Dans votre chatbot
def chat_with_escalation(user_message: str, conversation_id: str) -> dict:
    """Chat avec détection d'escalade"""

    # Récupérer historique
    history = get_conversation_history(conversation_id)

    # Générer réponse
    response = chatbot.chat(user_message)

    # Vérifier escalade
    escalation = EscalationDetector()
    should_escalate, reason = escalation.should_escalate(
        history,
        response,
        user_message
    )

    if should_escalate:
        # Notifier équipe support
        ticket_id = create_support_ticket(
            conversation_id=conversation_id,
            reason=reason,
            history=history
        )

        # Réponse d'escalade
        return {
            "answer": "Je vais te mettre en contact avec un membre de l'équipe. Ticket créé : " + ticket_id,
            "escalated": True,
            "ticket_id": ticket_id,
            "reason": reason
        }

    return response

import requests

def create_support_ticket(
    conversation_id: str,
    reason: str,
    history: list
) -> str:
    """Créer ticket Zendesk avec contexte"""

    # Formater historique
    conversation_text = "\n\n".join([
        f"{'User' if msg['role'] == 'user' else 'Bot'}: {msg['content']}"
        for msg in history
    ])

    # Créer ticket
    ticket_data = {
        "ticket": {
            "subject": f"Escalation chatbot ({reason})",
            "comment": {
                "body": f"""Escalation automatique du chatbot.

Raison : {reason}
Conversation ID : {conversation_id}

=== Historique de conversation ===
{conversation_text}

=== Contexte ===
- Heure escalade : {datetime.now().isoformat()}
- Confiance dernier message : {history[-1].get('confidence', 'N/A')}
"""
            },
            "priority": "normal" if reason != "frustration" else "high",
            "tags": ["chatbot_escalation", reason],
            "custom_fields": [
                {"id": 123456, "value": conversation_id}  # Custom field "Conv ID"
            ]
        }
    }

    response = requests.post(
        "https://company.zendesk.com/api/v2/tickets.json",
        json=ticket_data,
        auth=(ZENDESK_EMAIL, ZENDESK_API_TOKEN)
    )

    ticket = response.json()["ticket"]
    return ticket["id"]