Chatbots entwickeln: Open-Source Framework + Azure OpenAI + AI Search für Enterprise Chatbots 2025

graph TB
    A[User Input] --> B[Frontend Interface]
    B --> C[API Gateway]
    C --> D[Authentication Service]
    D --> E[Open-Source Framework]
    E --> F[Azure AI Search]
    F --> G[Azure OpenAI]
    G --> H[Response Generation]
    H --> I[Monitoring & Analytics]
    E --> J[Vector Database]
    J --> K[Knowledge Base]

# Python Environment Setup
python --version # 3.11+
pip install langchain openai azure-identity azure-search-documents
pip install llama-index azure-storage-blob python-dotenv

# Azure CLI Installation
curl -sL https://aka.ms/InstallAzureCLIDebian | sudo bash
az login

# Resource Group erstellen
az group create --name chatbot-rg --location germanywestcentral

# Azure OpenAI Service
az cognitiveservices account create \
  --name openai-chatbot \
  --resource-group chatbot-rg \
  --kind OpenAI \
  --sku S0 \
  --location germanywestcentral

# Azure AI Search
az search service create \
  --name ai-search-chatbot \
  --resource-group chatbot-rg \
  --sku Standard \
  --partition-count 1 \
  --replica-count 1 \
  --location germanywestcentral

# Azure Storage für Dokumente
az storage account create \
  --name chatbotstorage2025 \
  --resource-group chatbot-rg \
  --location germanywestcentral \
  --sku Standard_LRS \
  --kind StorageV2

# langchain_azure_openai_integration.py
import os
from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

class AzureOpenAIChatbot:
    def __init__(self, vault_url: str, openai_endpoint: str):
        self.credential = DefaultAzureCredential()
        self.secret_client = SecretClient(vault_url=vault_url, credential=self.credential)

        # Secrets aus Azure Key Vault laden
        self.openai_key = self.secret_client.get_secret("openai-api-key").value
        self.search_key = self.secret_client.get_secret("ai-search-key").value

        # Azure OpenAI Model initialisieren
        self.llm = AzureChatOpenAI(
            azure_endpoint=openai_endpoint,
            azure_deployment="gpt-4-turbo",
            api_version="2024-02-01",
            api_key=self.openai_key,
            temperature=0.7,
            max_tokens=1000
        )

        # Prompt Template für deutsche Unternehmenskommunikation
        self.prompt = ChatPromptTemplate.from_template("""
        Du bist ein professioneller KI-Assistent für deutsche Unternehmen.
        Antworte auf Deutsch und berücksichtige deutsche Geschäftsstandards.

        Kontext: {context}

        Frage: {question}

        Antwort:
        """)

        self.chain = self.prompt | self.llm | StrOutputParser()

    async def generate_response(self, question: str, context: str = "") -> str:
        """Generiert eine kontextbezogene Antwort"""
        try:
            response = await self.chain.ainvoke({
                "question": question,
                "context": context
            })
            return response
        except Exception as e:
            return f"Fehler bei der Antwortgenerierung: {str(e)}"

# llama_index_document_processing.py
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.storage import StorageContext
from llama_index.vector_stores.azureaisearch import AzureAISearchVectorStore
from azure.search.documents.indexes import SearchIndexClient
from azure.core.credentials import AzureKeyCredential

class DocumentProcessor:
    def __init__(self, search_endpoint: str, search_key: str, index_name: str):
        self.search_endpoint = search_endpoint
        self.search_key = search_key
        self.index_name = index_name

        # Azure AI Search Vector Store
        self.vector_store = AzureAISearchVectorStore(
            endpoint=search_endpoint,
            credential=AzureKeyCredential(search_key),
            index_name=index_name,
            embedding_field_name="content_vector",
            text_field_name="content"
        )

    def process_documents(self, documents_path: str):
        """Verarbeitet Dokumente und erstellt Vector Index"""
        # Dokumente laden
        documents = SimpleDirectoryReader(documents_path).load_data()

        # Text in Chunks aufteilen
        text_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
        nodes = text_splitter.get_nodes_from_documents(documents)

        # Index mit Azure AI Search erstellen
        storage_context = StorageContext.from_defaults(vector_store=self.vector_store)
        index = VectorStoreIndex(nodes, storage_context=storage_context)

        return index

    async def search_relevant_documents(self, query: str, top_k: int = 5):
        """Sucht relevante Dokumente für eine Query"""
        query_engine = self.vector_store.as_query_engine(similarity_top_k=top_k)
        response = await query_engine.aquery(query)
        return response

# azure_search_index_schema.py
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    SearchFieldDataType,
    SimpleField,
    SearchableField,
    VectorSearch,
    VectorSearchAlgorithmConfiguration,
    HnswAlgorithmConfiguration
)

def create_search_index(search_endpoint: str, search_key: str, index_name: str):
    """Erstellt Azure AI Search Index mit Vector-Support"""

    client = SearchIndexClient(search_endpoint, AzureKeyCredential(search_key))

    # Vector Search Konfiguration
    vector_search = VectorSearch(
        algorithms=[
            HnswAlgorithmConfiguration(
                name="hnsw-config",
                kind="hnsw",
                parameters={
                    "m": 4,
                    "efConstruction": 400,
                    "efSearch": 500,
                    "metric": "cosine"
                }
            )
        ]
    )

    # Index Schema definieren
    fields = [
        SimpleField(name="id", type=SearchFieldDataType.String, key=True),
        SearchableField(name="title", type=SearchFieldDataType.String),
        SearchableField(name="content", type=SearchFieldDataType.String),
        SearchableField(name="category", type=SearchFieldDataType.String),
        SimpleField(name="last_modified", type=SearchFieldDataType.DateTimeOffset),
        SimpleField(name="file_path", type=SearchFieldDataType.String),
        SearchField(
            name="content_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            vector_search_dimensions=1536,  # text-embedding-ada-002
            vector_search_configuration="hnsw-config"
        )
    ]

    index = SearchIndex(
        name=index_name,
        fields=fields,
        vector_search=vector_search
    )

    client.create_index(index)
    return index

# azure_openai_embeddings.py
import asyncio
from openai import AzureOpenAI
from azure.search.documents import SearchClient
from azure.core.credentials import AzureKeyCredential
from typing import List, Dict, Any

class EmbeddingService:
    def __init__(self, openai_endpoint: str, openai_key: str, deployment_name: str):
        self.client = AzureOpenAI(
            azure_endpoint=openai_endpoint,
            api_key=openai_key,
            api_version="2024-02-01"
        )
        self.deployment_name = deployment_name

    async def generate_embeddings_batch(self, texts: List[str]) -> List[List[float]]:
        """Generiert Embeddings für mehrere Texte batchweise"""
        embeddings = []

        # Batch-Verarbeitung für Performance
        batch_size = 10
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]

            try:
                response = await self.client.embeddings.create(
                    model=self.deployment_name,
                    input=batch
                )

                batch_embeddings = [data.embedding for data in response.data]
                embeddings.extend(batch_embeddings)

                # Rate Limiting berücksichtigen
                await asyncio.sleep(0.1)

            except Exception as e:
                print(f"Fehler bei Embedding-Generierung: {e}")
                # Fallback: Leere Embeddings
                embeddings.extend([[0.0] * 1536 for _ in batch])

        return embeddings

    async def index_documents_with_embeddings(
        self,
        documents: List[Dict[str, Any]],
        search_endpoint: str,
        search_key: str,
        index_name: str
    ):
        """Indiziert Dokumente mit Embeddings in Azure AI Search"""

        search_client = SearchClient(
            endpoint=search_endpoint,
            index_name=index_name,
            credential=AzureKeyCredential(search_key)
        )

        # Texte für Embedding extrahieren
        texts = [doc.get('content', '') for doc in documents]

        # Embeddings generieren
        embeddings = await self.generate_embeddings_batch(texts)

        # Dokumente mit Embeddings vorbereiten
        documents_with_embeddings = []
        for doc, embedding in zip(documents, embeddings):
            doc_with_embedding = {
                **doc,
                'content_vector': embedding
            }
            documents_with_embeddings.append(doc_with_embedding)

        # Batch Upload zu Azure AI Search
        batch_size = 100
        for i in range(0, len(documents_with_embeddings), batch_size):
            batch = documents_with_embeddings[i:i + batch_size]
            await search_client.upload_documents(documents=batch)

        print(f"Indiziert {len(documents_with_embeddings)} Dokumente erfolgreich")

# enterprise_chatbot.py
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class ChatbotConfig:
    azure_openai_endpoint: str
    azure_search_endpoint: str
    key_vault_url: str
    index_name: str
    temperature: float = 0.7
    max_tokens: int = 1000
    context_window: int = 5

class EnterpriseChatbot:
    def __init__(self, config: ChatbotConfig):
        self.config = config
        self.conversation_history: List[Dict] = []
        self.azure_openai = AzureOpenAIChatbot(
            vault_url=config.key_vault_url,
            openai_endpoint=config.azure_openai_endpoint
        )
        self.document_processor = DocumentProcessor(
            search_endpoint=config.azure_search_endpoint,
            search_key=self.azure_openai.search_key,
            index_name=config.index_name
        )
        self.embedding_service = EmbeddingService(
            openai_endpoint=config.azure_openai_endpoint,
            openai_key=self.azure_openai.openai_key,
            deployment_name="text-embedding-ada-002"
        )

    async def process_message(self, user_message: str) -> Dict[str, Any]:
        """Verarbeitet eine Benutzer-Nachricht und generiert Antwort"""

        # Conversation History verwalten
        self.conversation_history.append({
            "role": "user",
            "message": user_message,
            "timestamp": datetime.now()
        })

        # Context aus Conversation extrahieren
        context = self._extract_conversation_context()

        # Relevante Dokumente suchen
        search_results = await self.document_processor.search_relevant_documents(
            user_message,
            top_k=3
        )

        # Kontext aus Dokumenten extrahieren
        document_context = ""
        if search_results and hasattr(search_results, 'source_nodes'):
            document_context = "\n".join([
                node.node.text for node in search_results.source_nodes
            ])

        # Vollständigen Context zusammenstellen
        full_context = f"""
        Gesprächsverlauf: {context}
        Relevante Unternehmensdokumente: {document_context}
        """

        # Antwort generieren
        response = await self.azure_openai.generate_response(
            user_message,
            full_context
        )

        # Antwort in History speichern
        self.conversation_history.append({
            "role": "assistant",
            "message": response,
            "timestamp": datetime.now()
        })

        # Context Window verwalten
        if len(self.conversation_history) > self.config.context_window * 2:
            self.conversation_history = self.conversation_history[-self.config.context_window * 2:]

        return {
            "response": response,
            "confidence": getattr(search_results, 'score', 0.0) if search_results else 0.0,
            "sources": [node.node.metadata for node in search_results.source_nodes] if search_results and hasattr(search_results, 'source_nodes') else []
        }

    def _extract_conversation_context(self) -> str:
        """Extrahiert relevanten Context aus Conversation History"""
        recent_messages = self.conversation_history[-self.config.context_window:]
        context_parts = []

        for msg in recent_messages:
            role = "Benutzer" if msg["role"] == "user" else "Assistent"
            context_parts.append(f"{role}: {msg['message']}")

        return "\n".join(context_parts)

    async def add_knowledge_base(self, documents_path: str):
        """Fügt neue Dokumente zur Knowledge Base hinzu"""
        # Dokumente verarbeiten
        documents = SimpleDirectoryReader(documents_path).load_data()

        # Dokumente für Index vorbereiten
        docs_for_index = []
        for doc in documents:
            docs_for_index.append({
                'id': doc.id_,
                'title': doc.metadata.get('title', 'Unbekannt'),
                'content': doc.text,
                'category': doc.metadata.get('category', 'Allgemein'),
                'last_modified': datetime.now().isoformat(),
                'file_path': doc.metadata.get('file_path', '')
            })

        # Embeddings generieren und indizieren
        await self.embedding_service.index_documents_with_embeddings(
            docs_for_index,
            self.config.azure_search_endpoint,
            self.azure_openai.search_key,
            self.config.index_name
        )

        print(f"Knowledge Base erweitert um {len(documents)} Dokumente")

# chatbot_api.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional
import uvicorn
from enterprise_chatbot import EnterpriseChatbot, ChatbotConfig

app = FastAPI(title="Enterprise Chatbot API", version="1.0.0")
security = HTTPBearer()

class ChatRequest(BaseModel):
    message: str
    session_id: Optional[str] = None

class ChatResponse(BaseModel):
    response: str
    confidence: float
    sources: list
    session_id: str

# Global Chatbot Instance
chatbot_config = ChatbotConfig(
    azure_openai_endpoint="https://your-openai.openai.azure.com/",
    azure_search_endpoint="https://your-search.search.windows.net/",
    key_vault_url="https://your-keyvault.vault.azure.net/",
    index_name="enterprise-knowledge-base"
)

chatbot = EnterpriseChatbot(chatbot_config)

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(
    request: ChatRequest,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Chatbot API Endpoint"""

    try:
        # Authentifizierung prüfen (implementieren Sie Ihre Logik)
        # verify_token(credentials.credentials)

        # Nachricht verarbeiten
        result = await chatbot.process_message(request.message)

        return ChatResponse(
            response=result["response"],
            confidence=result["confidence"],
            sources=result["sources"],
            session_id=request.session_id or "default"
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Chatbot Fehler: {str(e)}")

@app.post("/knowledge-base/add")
async def add_knowledge(
    documents_path: str,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Neue Dokumente zur Knowledge Base hinzufügen"""

    try:
        await chatbot.add_knowledge_base(documents_path)
        return {"message": "Dokumente erfolgreich hinzugefügt"}

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Knowledge Base Fehler: {str(e)}")

@app.get("/health")
async def health_check():
    """Health Check Endpoint"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# dsgvo_compliance.py
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
from cryptography.fernet import Fernet
import hashlib
import logging

class DSGVOComplianceManager:
    def __init__(self, key_vault_url: str):
        self.credential = DefaultAzureCredential()
        self.secret_client = SecretClient(key_vault_url, self.credential)
        self.encryption_key = self.secret_client.get_secret("encryption-key").value
        self.cipher = Fernet(self.encryption_key)

    def anonymize_pii(self, text: str) -> str:
        """Entfernt personenbezogene Daten aus Text"""
        # Implementieren Sie PII-Erkennung und -Entfernung
        # Verwenden Sie Azure AI Language für PII Detection
        pass

    def encrypt_conversation(self, conversation_data: Dict) -> bytes:
        """Verschlüsselt Gesprächsdaten"""
        data_str = json.dumps(conversation_data)
        return self.cipher.encrypt(data_str.encode())

    def decrypt_conversation(self, encrypted_data: bytes) -> Dict:
        """Entschlüsselt Gesprächsdaten"""
        decrypted = self.cipher.decrypt(encrypted_data)
        return json.loads(decrypted.decode())

    def log_access(self, user_id: str, action: str, resource: str):
        """Protokolliert Zugriffe für Audit-Zwecke"""
        hashed_user_id = hashlib.sha256(user_id.encode()).hexdigest()

        logging.info({
            "timestamp": datetime.now().isoformat(),
            "user_hash": hashed_user_id,
            "action": action,
            "resource": resource,
            "compliance": "DSGVO"
        })

    def retention_policy_check(self, conversation_age_days: int) -> bool:
        """Prüft Aufbewahrungsfristen"""
        max_retention_days = 2555  # 7 Jahre für Geschäftsunterlagen
        return conversation_age_days <= max_retention_days

# monitoring.py
from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import time

class ChatbotMonitoring:
    def __init__(self, connection_string: str):
        # Azure Application Insights konfigurieren
        configure_azure_monitor(connection_string=connection_string)

        self.tracer = trace.get_tracer(__name__)

    def track_chatbot_interaction(
        self,
        user_message: str,
        response: str,
        response_time: float,
        confidence: float,
        sources_count: int
    ):
        """Trackt Chatbot-Interaktionen"""

        with self.tracer.start_as_span("chatbot_interaction") as span:
            span.set_attribute("user.message_length", len(user_message))
            span.set_attribute("response.length", len(response))
            span.set_attribute("response.time", response_time)
            span.set_attribute("confidence.score", confidence)
            span.set_attribute("sources.count", sources_count)

            # Custom Metrics
            span.set_attribute("custom.metric.response_quality", self._calculate_quality_score(response))
            span.set_attribute("custom.metric.user_satisfaction", self._estimate_satisfaction(user_message, response))

    def _calculate_quality_score(self, response: str) -> float:
        """Berechnet Qualitäts-Score der Antwort"""
        # Implementieren Sie Qualitätsmetriken
        # Länge, Kohärenz, Relevanz, etc.
        return 0.85  # Placeholder

    def _estimate_satisfaction(self, question: str, answer: str) -> float:
        """Schätzt User-Zufriedenheit"""
        # Implementieren Sie Sentiment-Analyse
        return 0.82  # Placeholder

    @staticmethod
    def performance_timer(func):
        """Decorator für Performance-Messung"""
        def wrapper(*args, **kwargs):
            start_time = time.time()
            result = func(*args, **kwargs)
            execution_time = time.time() - start_time

            # Log Performance
            print(f"Function {func.__name__} executed in {execution_time:.4f} seconds")

            return result
        return wrapper

# Dockerfile
FROM python:3.11-slim

# System Dependencies
RUN apt-get update && apt-get install -y \
    curl \
    gnupg \
    && rm -rf /var/lib/apt/lists/*

# Azure CLI installieren
RUN curl -sL https://aka.ms/InstallAzureCLIDebian | bash

# Python Dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Application Code
COPY . /app
WORKDIR /app

# Non-root User
RUN useradd --create-home --shell /bin/bash chatbot
USER chatbot

# Health Check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000

CMD ["python", "chatbot_api.py"]

# kubernetes-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: enterprise-chatbot
spec:
  replicas: 3
  selector:
    matchLabels:
      app: chatbot
  template:
    metadata:
      labels:
        app: chatbot
    spec:
      containers:
      - name: chatbot
        image: your-registry/chatbot:latest
        ports:
        - containerPort: 8000
        env:
        - name: AZURE_OPENAI_ENDPOINT
          valueFrom:
            secretKeyRef:
              name: chatbot-secrets
              key: azure-openai-endpoint
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

# caching.py
from azure.storage.blob import BlobServiceClient
from cachetools import TTLCache
import json

class ChatbotCache:
    def __init__(self, connection_string: str, container_name: str):
        self.blob_service = BlobServiceClient.from_connection_string(connection_string)
        self.container_name = container_name
        self.memory_cache = TTLCache(maxsize=1000, ttl=3600)  # 1 Stunde TTL

    async def get_cached_response(self, query_hash: str) -> Optional[str]:
        """Holt gecachte Antwort aus Memory oder Blob Storage"""
        # Memory Cache prüfen
        if query_hash in self.memory_cache:
            return self.memory_cache[query_hash]

        # Blob Storage prüfen
        try:
            blob_client = self.blob_service.get_blob_client(
                container=self.container_name,
                blob=f"cache/{query_hash}.json"
            )
            data = await blob_client.download_blob()
            cached_data = json.loads(data.readall())

            # In Memory Cache speichern
            self.memory_cache[query_hash] = cached_data['response']
            return cached_data['response']

        except Exception:
            return None

    async def cache_response(self, query_hash: str, response: str, metadata: Dict):
        """Speichert Antwort im Cache"""
        cache_data = {
            'response': response,
            'metadata': metadata,
            'timestamp': datetime.now().isoformat()
        }

        # Memory Cache aktualisieren
        self.memory_cache[query_hash] = response

        # Blob Storage aktualisieren
        blob_client = self.blob_service.get_blob_client(
            container=self.container_name,
            blob=f"cache/{query_hash}.json"
        )

        await blob_client.upload_blob(
            json.dumps(cache_data),
            overwrite=True
        )