SK-4-Filters : Filtres et Observabilite

Navigation : << 03-Agents | Index | 05-VectorStores >>

Objectifs d’apprentissage

A la fin de ce notebook, vous saurez : 1. Intercepter les appels de fonctions avec des filtres 2. Modifier les prompts avant envoi au LLM 3. Contrôler le Function Calling automatique 4. Configurer le logging pour le debugging 5. Comprendre l’integration OpenTelemetry pour le monitoring

Prerequis

  • Python 3.10+
  • Notebooks 01-03 completes
  • Cle API OpenAI configuree (.env)

Duree estimee : 45 minutes

Sommaire

Section Contenu Concepts cles
1 Introduction Pourquoi les filtres ?
2 Function Filters Avant/après invocation
3 Prompt Filters Modification du prompt
4 Auto-Invoke Filters Contrôle du function calling
5 Logging Configuration, niveaux
6 OpenTelemetry Tracing, metriques
7 Conclusion Resume, exercices

Pourquoi les filtres ? Les filtres permettent d’intercepter et modifier les appels a tous les niveaux de SK : avant/après les fonctions, avant/après les prompts, et lors du function calling automatique. C’est essentiel pour le logging, la securite, et le monitoring en production.

Le diagramme ci-dessous rend ce pipeline de filtres sous forme de graphe : chaque filtre s’intercale a un point précis du cycle requête -> reponse.

flowchart TD
    U(["User Request"]) --> PF["Prompt Filter<br/>(modifie le prompt)"]
    PF --> LLM["LLM Call"]
    LLM --> AIF["Auto-Invoke Filter<br/>(contrôle les appels de fonction)"]
    AIF --> FF["Function Filter<br/>(avant/après chaque fonction)"]
    FF --> R(["Response"])
    classDef filt fill:#fff3cd,stroke:#b8860b,color:#5c4400
    classDef io fill:#d1e7dd,stroke:#0f5132,color:#0a3622
    classDef core fill:#cfe2ff,stroke:#084298,color:#052c65
    class PF,AIF,FF filt
    class LLM core
    class U,R io

Lecture. Les filtres de Semantic Kernel sont des intercepteurs qui s’executent a des points de jonction du pipeline : le Prompt Filter peut reecrire le prompt avant l’appel au LLM ; l’Auto-Invoke Filter arbitre si/comment les appels de fonction proposes par le modèle sont executes ; le Function Filter entoure l’exécution de chaque fonction (logging, validation, court-circuit). C’est le point d’ancrage de l’observabilite et de la gouvernance sans modifier le code metier.

# Installation et imports

import os
import logging
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function, KernelArguments
from semantic_kernel.filters import FilterTypes
from semantic_kernel.exceptions import KernelInvokeException

# Chargement du fichier .env (cles API)
load_dotenv("../.env")

api_key = os.getenv("OPENAI_API_KEY")
print(f"Configuration: API Key {'OK' if api_key else 'MANQUANTE'}")
print("Imports OK")
Configuration: API Key OK
Imports OK

1. Introduction aux Filtres

Semantic Kernel propose un système de filtres inspire des middlewares web :

Type de Filtre Point d’interception Cas d’usage
Function Invocation Avant/après chaque fonction Logging, validation, timing
Prompt Rendering Avant envoi au LLM Injection de règles, anonymisation
Auto Function Invocation Lors du function calling Rate limiting, approbation

Architecture des filtres

User Request
    |
    v
[Prompt Filter] --> Modifie le prompt
    |
    v
LLM Call
    |
    v
[Auto-Invoke Filter] --> Contrôle les appels de fonction
    |
    v
[Function Filter] --> Avant/après chaque fonction
    |
    v
Response

2. Function Invocation Filters

Les filtres de fonction permettent d’intercepter chaque appel de fonction du kernel.

from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext
from typing import Callable, Coroutine, Any
import time

# Creation du kernel
kernel = Kernel()
kernel.add_service(OpenAIChatCompletion(service_id="default"))

# Plugin de demonstration
class MathPlugin:
    @kernel_function(description="Additionne deux nombres")
    def add(self, a: int, b: int) -> int:
        return a + b
    
    @kernel_function(description="Multiplie deux nombres")
    def multiply(self, a: int, b: int) -> int:
        return a * b

kernel.add_plugin(MathPlugin(), plugin_name="math")

# Filtre de logging avec timing
@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def logging_filter(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]]
):
    """Filtre qui log les appels de fonction avec leur duree."""
    func_name = f"{context.function.plugin_name}.{context.function.name}"
    print(f"[AVANT] Appel de {func_name}")
    print(f"  Arguments: {context.arguments}")
    
    start_time = time.time()
    
    # Appel de la fonction (ou du filtre suivant)
    await next(context)
    
    duration = time.time() - start_time
    print(f"[APRES] {func_name} termine en {duration:.4f}s")
    print(f"  Resultat: {context.result}")

# Test du filtre
result = await kernel.invoke(kernel.get_function("math", "add"), KernelArguments(a=5, b=3))
print(f"\nResultat final: {result}")
[AVANT] Appel de math.add
  Arguments: {'a': 5, 'b': 3}
[APRES] math.add termine en 0.0001s
  Resultat: 8

Resultat final: 8

Exercice 1 : Filtre de limitation de taux (rate limiter)

Objectif : Implementer un Function Invocation Filter qui limite le nombre d’appels a une fonction par fenêtre de temps (sliding window).

En production, il est essentiel de proteger les APIs externes contre les appels excessifs. Vous allez créer un filtre qui bloque les appels si le nombre d’invocations depasse un seuil configurable dans les N dernières secondes.

Indices : - # Étape 1 : Maintenir une liste call_timestamps = [] avec les horodatages de chaque appel - # Étape 2 : Avant chaque appel, filtrer les timestamps plus anciens que la fenêtre (ex: 60s) - # Étape 3 : Si le nombre de timestamps restants depasse la limite, lever une exception - # Indice : time.time() retourne le timestamp actuel en secondes

class RateLimiter:
    """
    Limiteur de taux avec fenetre glissante.
    
    # Etape 1 : Initialiser call_timestamps comme liste vide
    # Etape 2 : Implementer is_allowed() qui nettoie les vieux timestamps et verifie la limite
    # Etape 3 : Implementer le Function Filter SK qui utilise ce limiteur
    """
    def __init__(self, max_calls: int = 5, window_seconds: float = 60.0):
        self.max_calls = max_calls
        self.window_seconds = window_seconds
        self.call_timestamps = []  # TODO etudiant : sera mis a jour dans is_allowed()
    
    def is_allowed(self) -> bool:
        """
        Verifie si un nouvel appel est autorise.
        
        # Indice : Utilisez time.time() pour le timestamp actuel
        # Indice : Filtrez les timestamps > (maintenant - window_seconds)
        # Indice : Retournez True si len(timestamps_restants) < max_calls
        """
        # TODO etudiant : implementer la logique de rate limiting
        print("Exercice a completer : RateLimiter.is_allowed")
        return None  # TODO etudiant : remplacer par la vraie logique

# Test du rate limiter seul
limiter = RateLimiter(max_calls=3, window_seconds=60.0)
for i in range(5):
    allowed = limiter.is_allowed()
    verdict = "?" if allowed is None else ("autorise" if allowed else "BLOQUE")
    print(f"Appel {i+1}: {verdict}")
Exercice a completer : RateLimiter.is_allowed
Appel 1: ?
Exercice a completer : RateLimiter.is_allowed
Appel 2: ?
Exercice a completer : RateLimiter.is_allowed
Appel 3: ?
Exercice a completer : RateLimiter.is_allowed
Appel 4: ?
Exercice a completer : RateLimiter.is_allowed
Appel 5: ?

Interpretation : Function Invocation Filter

Le filtre ci-dessus illustre le pattern middleware :

  1. Avant l’appel : On log le nom de la fonction et ses arguments
  2. await next(context) : On execute la fonction (ou le filtre suivant)
  3. Après l’appel : On log le résultat et la duree

Points cles : - context.function : Metadata de la fonction (nom, plugin, description) - context.arguments : Arguments passes a la fonction - context.result : Résultat après exécution - next(context) : Appelle le filtre suivant ou la fonction elle-même

Attention : Oublier await next(context) bloquera l’exécution de la fonction !

# Exemple de filtre de validation
@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def validation_filter(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]]
):
    """Filtre qui valide les arguments avant execution."""
    # Validation specifique pour les fonctions math
    if context.function.plugin_name == "math":
        a = context.arguments.get("a", 0)
        b = context.arguments.get("b", 0)
        
        # Exemple : bloquer les nombres negatifs
        if a < 0 or b < 0:
            raise ValueError(f"Les nombres negatifs ne sont pas autorises: a={a}, b={b}")
    
    await next(context)

# Test avec nombres valides
try:
    result = await kernel.invoke(kernel.get_function("math", "multiply"), KernelArguments(a=4, b=5))
    print(f"Resultat: {result}")
except Exception as e:
    print(f"Erreur: {e}")

# Test avec nombre negatif (devrait echouer)
try:
    result = await kernel.invoke(kernel.get_function("math", "multiply"), KernelArguments(a=-3, b=5))
    print(f"Resultat: {result}")
except KernelInvokeException as e:
    # L'exception ValueError est encapsulee dans KernelInvokeException
    print(f"Erreur de validation (KernelInvokeException): nombres negatifs bloques")
except ValueError as e:
    print(f"Erreur de validation: {e}")
[AVANT] Appel de math.multiply
  Arguments: {'a': 4, 'b': 5}
[APRES] math.multiply termine en 0.0000s
  Resultat: 20
Resultat: 20
[AVANT] Appel de math.multiply
  Arguments: {'a': -3, 'b': 5}
Erreur de validation (KernelInvokeException): nombres negatifs bloques

Exercice 2 : Filtre de cache pour les appels de fonction

Objectif : Implementer un Function Invocation Filter qui met en cache les résultats des fonctions pour eviter des appels repetes couteux.

Le caching est un pattern fondamental en production : si une fonction pure (comme une addition) est appelee plusieurs fois avec les mêmes arguments, on retourne le résultat en cache au lieu de re-executer. Ce principe s’applique aussi aux appels LLM couteux.

Indices : - # Étape 1 : Créer un dictionnaire global cache = {} pour stocker les résultats - # Étape 2 : Construire une cle de cache unique a partir du nom de fonction et des arguments - # Étape 3 : Si la cle existe dans le cache, définir context.result et retourner (sans appeler next) - # Indice : f"{context.function.plugin_name}.{context.function.name}:{sorted(context.arguments.items())}" pour la cle

# Cache global pour stocker les resultats
function_cache = {}

async def caching_filter(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]]
):
    """
    Filtre de cache pour les appels de fonction.
    
    # Etape 1 : Construire la cle de cache (nom_fonction + arguments tries)
    # Etape 2 : Si la cle existe, definir context.result et retourner (cache HIT)
    # Etape 3 : Sinon, appeler await next(context), puis stocker le resultat (cache MISS)
    
    # Indice : Utilisez print("[CACHE HIT]") / print("[CACHE MISS]") pour debugger
    """
    # TODO etudiant : implementer le filtre de cache
    await next(context)

# Test du filtre de cache (decommentez apres implementation)
# kernel_cache = Kernel()
# kernel_cache.add_service(OpenAIChatCompletion(service_id="default"))
# kernel_cache.add_plugin(MathPlugin(), plugin_name="math_cache")
# kernel_cache.filter(FilterTypes.FUNCTION_INVOCATION)(caching_filter)
# 
# # Premier appel (MISS)
# r1 = await kernel_cache.invoke(kernel_cache.get_function("math_cache", "add"), KernelArguments(a=10, b=20))
# print(f"Resultat 1: {r1}")
# # Deuxieme appel (HIT)
# r2 = await kernel_cache.invoke(kernel_cache.get_function("math_cache", "add"), KernelArguments(a=10, b=20))
# print(f"Resultat 2: {r2}")
# print(f"Taille du cache: {len(function_cache)}")

print("Exercice a completer")
Exercice a completer

Interprétation : Validation Filter avec Exception Handling

Sortie obtenue : - Premier appel (4 × 5) : Succès avec résultat 20 - Second appel (-3 × 5) : Blocage avec KernelInvokeException

Aspect Comportement Explication
Exception wrapping ValueError → KernelInvokeException SK encapsule les exceptions des filtres
Ordre d’exécution Validation → Logging → Fonction Les filtres sont chaînés dans l’ordre d’enregistrement
Blocage précoce Avant calcul Le filtre de validation empêche l’exécution de la fonction

Points clés : 1. Chaînage des filtres : Le filtre de validation s’exécute AVANT le filtre de logging (voir les logs) 2. Exception handling : Les exceptions levées dans les filtres sont automatiquement wrappées dans KernelInvokeException 3. Validation métier : Pattern idéal pour valider les inputs avant traitement coûteux (LLM, DB, etc.) 4. Court-circuit : Si on lève une exception, next(context) n’est jamais appelé → fonction non exécutée

Note technique : Pour capturer l’exception originale, utiliser try/except KernelInvokeException as e et inspecter e.__cause__.

3. Prompt Rendering Filters

Les filtres de prompt permettent de modifier le prompt avant son envoi au LLM.

from semantic_kernel.filters.prompts.prompt_render_context import PromptRenderContext
from semantic_kernel.prompt_template import PromptTemplateConfig

# Nouveau kernel pour les filtres de prompt
kernel_prompt = Kernel()
kernel_prompt.add_service(OpenAIChatCompletion(service_id="default"))

# Filtre qui ajoute des instructions de securite
@kernel_prompt.filter(FilterTypes.PROMPT_RENDERING)
async def security_prompt_filter(
    context: PromptRenderContext,
    next: Callable[[PromptRenderContext], Coroutine[Any, Any, None]]
):
    """Ajoute des regles de securite au prompt."""
    # Executer le rendu du template d'abord
    await next(context)
    
    # Ajouter des instructions de securite apres le rendu
    security_rules = """

REGLES DE SECURITE:
- Ne jamais reveler d'informations personnelles
- Ne pas generer de contenu offensant
- Refuser les demandes de code malveillant
"""
    context.rendered_prompt = context.rendered_prompt + security_rules
    print(f"[Prompt Filter] Regles de securite ajoutees")

# Creation d'une fonction avec template
prompt_config = PromptTemplateConfig(
    template="Tu es un assistant. Reponds a: {{$input}}",
    name="secure_chat",
    template_format="semantic-kernel"
)

chat_function = kernel_prompt.add_function(
    function_name="chat",
    plugin_name="demo",
    prompt_template_config=prompt_config
)

# Test
response = await kernel_prompt.invoke(chat_function, KernelArguments(input="Bonjour, comment ca va?"))
print(f"\nReponse: {response}")
[Prompt Filter] Regles de securite ajoutees

Reponse: Bonjour ! Ça va bien, merci. Et toi, comment tu vas aujourd’hui ?

Exercice 3 : Filtre d’anonymisation de données sensibles

Objectif : Créer un Prompt Filter qui detecte et masque automatiquement les adresses email et numéros de telephone dans les prompts avant envoi au LLM.

En production, il est crucial d’empêcher les données personnelles (PII) d’atteindre les services LLM externes. Vous allez implementer un filtre qui remplace les patterns sensibles par des masques avant l’appel API.

Indices : - # Étape 1 : Utiliser des expressions regulieres (module re) pour detecter emails et telephones - # Étape 2 : Remplacer les patterns detectes par [EMAIL_CACHE] et [TEL_CACHE] - # Étape 3 : Appliquer le filtre sur context.rendered_prompt après le rendu du template - # Indice : re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+', '[EMAIL_CACHE]', text) pour les emails

import re

def anonymize_text(text: str) -> tuple[str, dict[str, str]]:
    """
    Detecte et masque les PII dans un texte.
    
    # Etape 1 : Trouver tous les emails avec re.findall()
    # Etape 2 : Remplacer chaque email par [EMAIL_1], [EMAIL_2], etc.
    # Etape 3 : Faire de meme pour les numeros de telephone (pattern : 10 chiffres ou +33...)
    # Indice : Stocker les correspondances masque -> valeur originale dans un dict
    
    Returns:
        tuple: (texte_anonymise, dictionnaire_des_correspondances)
    """
    # TODO etudiant : implementer l'anonymisation
    return text, {}

# Test de la fonction d'anonymisation
test_text = "Contactez moi a jean.dupont@example.com ou au 0612345678 pour plus d'infos."
anonymized, mapping = anonymize_text(test_text)
print(f"Original    : {test_text}")
print(f"Anonymise   : {anonymized}")
print(f"Correspondances : {mapping}")

# TODO etudiant : creer le Prompt Filter SK qui utilise cette fonction
# @kernel_prompt.filter(FilterTypes.PROMPT_RENDERING)
# async def pii_filter(context, next):
#     await next(context)
#     anonymized, _ = anonymize_text(context.rendered_prompt)
#     context.rendered_prompt = anonymized
print("Exercice a completer")
Original    : Contactez moi a jean.dupont@example.com ou au 0612345678 pour plus d'infos.
Anonymise   : Contactez moi a jean.dupont@example.com ou au 0612345678 pour plus d'infos.
Correspondances : {}
Exercice a completer

Interprétation : Prompt Filter avec Injection de Règles

Sortie obtenue : Le message “[Prompt Filter] Règles de securite ajoutees” confirme que le filtre s’est exécuté, et le LLM a bien reçu le prompt modifié.

Étape Contenu Rôle
Template original “Tu es un assistant. Reponds a: {$input}” Défini par le développeur
Rendu “Tu es un assistant. Reponds a: Bonjour…” Variables remplacées
Après filtre Prompt + RÈGLES DE SECURITE Instructions système ajoutées
Envoyé au LLM Prompt complet avec règles Le LLM “voit” les règles

Points clés : 1. Ordre d’exécution : D’abord await next(context) (rendu du template), PUIS modification du rendered_prompt 2. Injection transparente : Le développeur qui appelle la fonction ne sait pas que des règles sont ajoutées 3. Sécurité systémique : Toutes les fonctions du kernel héritent automatiquement des règles 4. Visibilité : Le LLM “voit” les règles dans son prompt (contrairement aux system messages cachés)

Cas d’usage production : - Ajout de contraintes légales (RGPD, conformité) - Injection de contexte d’entreprise (brand guidelines) - Rate limiting explicite (“Tu as droit à 3 appels de fonction max”) - Instructions de formatage (JSON, Markdown, etc.)

Note : Pour voir le prompt complet envoyé au LLM, activer le logging en mode DEBUG (voir section 5).

4. Auto Function Invocation Filters

Ces filtres controlent le comportement du function calling automatique.

from semantic_kernel.filters.auto_function_invocation.auto_function_invocation_context import AutoFunctionInvocationContext
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings
from semantic_kernel.connectors.ai import FunctionChoiceBehavior

# Nouveau kernel pour les filtres auto-invoke
kernel_auto = Kernel()
kernel_auto.add_service(OpenAIChatCompletion(service_id="default"))

# Plugin sensible
class DatabasePlugin:
    @kernel_function(description="Execute une requete SQL")
    def execute_sql(self, query: str) -> str:
        print(f"  [DB] Execution: {query}")
        return f"Resultats pour: {query}"
    
    @kernel_function(description="Supprime des donnees")
    def delete_data(self, table: str) -> str:
        print(f"  [DB] DELETE FROM {table}")
        return f"Donnees supprimees de {table}"

kernel_auto.add_plugin(DatabasePlugin(), plugin_name="database")

# Compteur d'appels pour rate limiting
call_count = {"total": 0}

@kernel_auto.filter(FilterTypes.AUTO_FUNCTION_INVOCATION)
async def rate_limit_filter(
    context: AutoFunctionInvocationContext,
    next: Callable[[AutoFunctionInvocationContext], Coroutine[Any, Any, None]]
):
    """Limite le nombre d'appels de fonction automatiques."""
    MAX_CALLS = 3
    
    call_count["total"] += 1
    
    if call_count["total"] > MAX_CALLS:
        print(f"[Rate Limit] Limite atteinte ({MAX_CALLS} appels max)")
        context.terminate = True  # Stoppe le function calling
        return
    
    func_name = context.function.name
    print(f"[Auto-Invoke] Appel #{call_count['total']}: {func_name}")
    
    # Bloquer les operations dangereuses
    if "delete" in func_name.lower():
        print(f"[Auto-Invoke] BLOQUE: Operation de suppression non autorisee")
        context.terminate = True
        return
    
    await next(context)

print("Filtre auto-invoke configure. Voir section 5 pour un exemple complet.")
Filtre auto-invoke configure. Voir section 5 pour un exemple complet.

Interprétation : Auto Function Invocation Filter avec Rate Limiting

Architecture du contrôle : Ce filtre intercepte les décisions du LLM AVANT l’exécution des fonctions.

Mécanisme Implémentation Effet
Rate limiting Compteur global call_count Maximum 3 appels de fonction
Blacklist Check sur "delete" in func_name Bloque les opérations dangereuses
Terminaison context.terminate = True Stoppe le function calling loop
Court-circuit Return sans await next(context) Fonction non exécutée

Points clés : 1. Function calling loop : Quand le LLM appelle une fonction, SK peut automatiquement exécuter la fonction et renvoyer le résultat au LLM (qui peut ensuite appeler une autre fonction, etc.) 2. terminate = True : Indique à SK d’arrêter le loop et de retourner la réponse actuelle du LLM 3. Sécurité critique : Sans ce filtre, un LLM pourrait décider d’appeler delete_data() sans supervision 4. Audit trail : Chaque tentative d’appel est loggée, même si bloquée

Workflow complet :

User: "Supprime les anciennes données"
    ↓
LLM decide: database.delete_data(table="old_data")
    ↓
[Auto-Invoke Filter] BLOQUE: Opération de suppression
    ↓
LLM reçoit: "Opération refusée"
    ↓
Response finale

Note technique : Ce filtre est complémentaire aux Function Filters (section 2). L’Auto-Invoke Filter contrôle les appels automatiques du LLM, tandis que les Function Filters interceptent tous les appels (manuels ou auto).

5. Logging et Debugging

SK utilise le module logging standard de Python.

import logging
import sys

# Configuration du logging pour SK
def configure_sk_logging(level=logging.INFO):
    """Configure le logging pour Semantic Kernel."""
    # Handler pour la console
    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(level)
    
    # Format detaille
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    
    # Configurer le logger SK
    sk_logger = logging.getLogger("semantic_kernel")
    sk_logger.setLevel(level)
    sk_logger.addHandler(handler)
    
    return sk_logger

# Niveaux de logging disponibles
print("Niveaux de logging disponibles:")
print("| Niveau   | Valeur | Description                              |")
print("|----------|--------|------------------------------------------|")
print("| DEBUG    | 10     | Details tres fins (prompts complets)     |")
print("| INFO     | 20     | Informations generales                   |")
print("| WARNING  | 30     | Avertissements                           |")
print("| ERROR    | 40     | Erreurs                                  |")
print("| CRITICAL | 50     | Erreurs critiques                        |")

# Activer le logging DEBUG pour voir les details
logger = configure_sk_logging(logging.DEBUG)
print("\nLogging configure en mode DEBUG")
Niveaux de logging disponibles:
| Niveau   | Valeur | Description                              |
|----------|--------|------------------------------------------|
| DEBUG    | 10     | Details tres fins (prompts complets)     |
| INFO     | 20     | Informations generales                   |
| WARNING  | 30     | Avertissements                           |
| ERROR    | 40     | Erreurs                                  |
| CRITICAL | 50     | Erreurs critiques                        |

Logging configure en mode DEBUG
# Exemple avec logging actif
from semantic_kernel.prompt_template import PromptTemplateConfig

kernel_log = Kernel()
kernel_log.add_service(OpenAIChatCompletion(service_id="default"))

# Fonction simple
simple_config = PromptTemplateConfig(
    template="Dis bonjour a {{$name}} en francais.",
    name="greet"
)

greet_function = kernel_log.add_function(
    function_name="greet",
    plugin_name="demo",
    prompt_template_config=simple_config
)

# Execution avec logging
print("=" * 50)
print("Execution avec logging actif:")
print("=" * 50)
response = await kernel_log.invoke(greet_function, KernelArguments(name="Alice"))
print(f"\nReponse finale: {response}")
2026-09-10 22:50:05,733 - semantic_kernel.prompt_template.kernel_prompt_template - DEBUG - Extracting blocks from template: Dis bonjour a {{$name}} en francais.
==================================================
Execution avec logging actif:
==================================================
2026-09-10 22:50:05,735 - semantic_kernel.functions.kernel_function - INFO - Function demo-greet invoking.
2026-09-10 22:50:05,735 - semantic_kernel.functions.kernel_function - DEBUG - Function arguments: {'name': 'Alice'}
2026-09-10 22:50:05,736 - semantic_kernel.prompt_template.kernel_prompt_template - DEBUG - Rendering list of 3 blocks
2026-09-10 22:50:05,736 - semantic_kernel.prompt_template.kernel_prompt_template - DEBUG - Rendered prompt: Dis bonjour a Alice en francais.
2026-09-10 22:50:06,934 - semantic_kernel.connectors.ai.open_ai.services.open_ai_handler - INFO - OpenAI usage: CompletionUsage(completion_tokens=6, prompt_tokens=15, total_tokens=21, completion_tokens_details=CompletionTokensDetails(accepted_prediction_tokens=0, audio_tokens=0, reasoning_tokens=0, rejected_prediction_tokens=0), prompt_tokens_details=PromptTokensDetails(audio_tokens=0, cached_tokens=0))
2026-09-10 22:50:06,935 - semantic_kernel.functions.kernel_function - INFO - Function demo-greet succeeded.
2026-09-10 22:50:06,935 - semantic_kernel.functions.kernel_function - DEBUG - Function result: Bonjour Alice.
2026-09-10 22:50:06,936 - semantic_kernel.functions.kernel_function - INFO - Function completed. Duration: 1.200640s

Reponse finale: Bonjour Alice.

Interprétation : Logging DEBUG - Anatomie d’une Invocation

Sortie obtenue : Les logs DEBUG révèlent chaque étape interne de SK lors de l’exécution d’une fonction.

Timestamp Logger Message clé Signification
08:07:55.813 kernel_prompt_template “Extracting blocks from template” Parsing du template Jinja
08:07:55.814 kernel_function “Function demo-greet invoking” Début de l’invocation
08:07:55.815 kernel_function “Function arguments: {…}” Arguments reçus
08:07:55.815 kernel_prompt_template “Rendering list of 3 blocks” Rendu du template (3 blocs détectés)
08:07:55.816 kernel_prompt_template “Rendered prompt: Dis bonjour…” Prompt final avant envoi
08:07:56.214 open_ai_handler “OpenAI usage: CompletionUsage(…)” Tokens consommés (16 prompt + 3 completion)
08:07:56.215 kernel_function “Function demo-greet succeeded” Succès
08:07:56.216 kernel_function “Function completed. Duration: 0.400880s” Durée totale

Points clés : 1. 3 blocs dans le template : SK découpe le template en blocs (texte statique + variables) 2. Tokens consommés : 16 tokens prompt + 3 tokens completion = 19 tokens total (~0.00038$ avec GPT-4o) 3. Latence : 400ms totale dont ~398ms pour l’appel OpenAI (réseau + génération) 4. Pas de cache : cached_tokens=0 → prompt non mis en cache par OpenAI

Utilisation en production : - Debugging : Identifier où une fonction échoue (template, LLM, post-processing) - Optimisation : Mesurer la latence de chaque composant - Cost tracking : Tracker les tokens consommés par fonction - Audit : Logger tous les prompts envoyés (compliance RGPD)

Niveaux de logging recommandés : - Development : DEBUG (tout voir) - Staging : INFO (invocations + erreurs) - Production : WARNING (erreurs seulement) + OpenTelemetry pour les métriques

6. OpenTelemetry : observer un appel au LLM span par span

La section 5 utilisait logging : un canal texte, lisible mais non structuré. OpenTelemetry (OTel) standardise l’observabilité en trois signaux — traces, metrics, logs — et représente chaque opération par un span : une unité datée, nommée, porteuse d’attributs clé-valeur.

Pour le GenAI, OTel définit des conventions sémantiques (gen_ai.*) qui normalisent ce qu’on veut lire sur un appel modèle :

Attribut Sens
gen_ai.system Fournisseur (openai, aws.bedrock, …)
gen_ai.request.model Modèle invoqué
gen_ai.usage.input_tokens / gen_ai.usage.output_tokens Tokens consommés (prompt + génération)
gen_ai.operation.name Nature de l’opération (chat, execute_tool, …)

Cette section exécute un appel réel et capture le span produit — l’attribut de modèle et les attributs d’usage de tokens y sont lisibles, avec leurs noms exacts tels qu’émis par les conventions.

Backend OTLP : le dashboard Aspire (.NET) observe la chaîne Python

Un exporter OTLP pousse les spans vers un endpoint standard (http://localhost:4317 en gRPC). Côté .NET, le dossier aspire-otel/ embarque le backend : un AppHost Aspire minimal (modèle de code) et la commande de démarrage du dashboard Aspire en mode standalone — le mode documenté pour recevoir la télémétrie d’applications externes, endpoint OTLP ouvert sur localhost:4317. Le notebook Python envoie ses spans, le dashboard .NET les reçoit, les stocke et les expose via son API de télémétrie — qu’une cellule ci-dessous relit pour fermer la boucle. C’est une forme de parité forte : un outil .NET qui rend service à une chaîne Python, sans changer de langage, la frontière étant le protocole OTLP.

# Activation des diagnostics GenAI de SK + configuration OpenTelemetry.
#
# Subtilité : SK lit ses flags de diagnostics à l'import (cellule 2). Pour activer
# les spans gen_ai SANS modifier les sections 1 à 5 (déjà exécutées sans diagnostics),
# on ré-instancie le singleton de settings après avoir positionné les variables d'env.
# C'est self-contained à la section 6 : §1-§5 ne sont pas affectés.
import os
os.environ["SEMANTICKERNEL_EXPERIMENTAL_GENAI_ENABLE_OTEL_DIAGNOSTICS"] = "true"
os.environ["SEMANTICKERNEL_EXPERIMENTAL_GENAI_ENABLE_OTEL_DIAGNOSTICS_SENSITIVE"] = "true"

import semantic_kernel.utils.telemetry.model_diagnostics.decorators as _sk_diag
from semantic_kernel.utils.telemetry.model_diagnostics.model_diagnostics_settings import ModelDiagnosticSettings
_sk_diag.MODEL_DIAGNOSTICS_SETTINGS = ModelDiagnosticSettings()  # relit les env (désormais True)

# Deux exporters : Console (preuve reproductible dans la cellule) + OTLP (dashboard Aspire).
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, BatchSpanProcessor
from opentelemetry.sdk.resources import Resource

provider = TracerProvider(resource=Resource.create({"service.name": "sk-filters-observability"}))

# Exporter personnalisé : imprime un résumé propre des spans gen_ai (attributs + noms exacts)
# au moment où ils se ferment. Plus lisible qu'un dump Console brut, et capture bien les
# attributs normalisés par les conventions sémantiques GenAI d'OTel.
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult

# Champs gen_ai.* imprimés : allowlist PEDAGOGIQUE uniquement (opération,
# système, modèle, compteurs, finish reason, outil). Les identifiants de
# télémétrie opérationnelle (gen_ai.response.id et assimilés) sont des
# captures de trafic, pas des preuves pédagogiques — ils ne vont pas dans
# les outputs publics (review #15436).
_ATTRS_PEDAGOGIQUES = frozenset({
    "gen_ai.operation.name", "gen_ai.system", "gen_ai.request.model",
    "gen_ai.tool.name", "gen_ai.usage.input_tokens",
    "gen_ai.usage.output_tokens", "gen_ai.response.finish_reason",
})

class GenAiSpanPrinter:
    """Imprime les spans portant des attributs gen_ai.* (preuve reproductible dans la cellule)."""
    def export(self, spans):
        for sp in spans:
            attrs = dict(sp.attributes or {})
            gen = {k: v for k, v in attrs.items() if k in _ATTRS_PEDAGOGIQUES}
            if gen:
                print(f"[span capturé] {sp.name}")
                for k in sorted(gen):
                    val = repr(gen[k])
                    if len(val) > 100:
                        val = val[:100] + "…(tronqué)"
                    print(f"    {k} = {val}")
        return SpanExportResult.SUCCESS
    def shutdown(self): return True
    def force_flush(self, timeout_millis=30000): return True

provider.add_span_processor(SimpleSpanProcessor(GenAiSpanPrinter()))

# Exporter OTLP vers le dashboard Aspire (localhost:4317) — activé seulement si l'AppHost
# (aspire-otel/, mode standalone) tourne. Sinon le BatchSpanProcessor retryerait en boucle (bruit) :
# on port-check avant d'activer.
import socket
_tmp_s = socket.socket(); _tmp_s.settimeout(0.5)
try:
    _otlp_up = _tmp_s.connect_ex(("localhost", 4317)) == 0
finally:
    _tmp_s.close()
if _otlp_up:
    try:
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
        provider.add_span_processor(BatchSpanProcessor(
            OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)))
        print("Exporter OTLP branché -> http://localhost:4317 (dashboard Aspire)")
    except Exception as exc:
        print(f"OTLP non branché : {exc}")
else:
    print("Dashboard Aspire non joint (localhost:4317 fermé) — seul GenAiSpanPrinter est actif. Démarrez le backend aspire-otel/ (cf. son README) pour le pont OTLP.")

trace.set_tracer_provider(provider)
print("OpenTelemetry configuré : Console + OTLP")
Exporter OTLP branché -> http://localhost:4317 (dashboard Aspire)
OpenTelemetry configuré : Console + OTLP
# Un appel de complétion réel. On cible un endpoint OpenAI-compatible auto-hébergé
# (le proxy claudish models.myia.io, modèle Qwen3.6) dont les credentials sont lus
# depuis ../.env : c'est le span, pas le fournisseur, qui est l'objet de la section.
# Un endpoint OpenAI marchand ferait l'affaire de la même manière.
from dotenv import load_dotenv
from openai import AsyncOpenAI
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

load_dotenv("../.env")
# Clé DEDIÉE à la façade proxy exclusivement : aucun repli sur OPENAI_API_KEY — un
# fallback enverrait le credential OpenAI au mauvais hôte (credential
# cross-provider interdit, review #15436). C'est la clé de la façade
# (CLAUDISH_PROXY_KEY), distincte de VLLM_API_KEY qui authentifie le sidecar
# vLLM du cluster : le recâblage de l'endpoint impose celui de la clé.
_proxy_key = os.getenv("CLAUDISH_PROXY_KEY")
if not _proxy_key:
    raise ValueError(
        "CLAUDISH_PROXY_KEY absente : clé dédiée à la façade proxy "
        "models.myia.io requise (pas de repli sur OPENAI_API_KEY).")
# Endpoint auto-hébergé joignable depuis toute machine worker (issue #15431) ;
# substituable par tout endpoint OpenAI-compatible.
_client = AsyncOpenAI(api_key=_proxy_key, base_url="https://models.myia.io/v1")

kernel_otlp = Kernel()
svc_otlp = OpenAIChatCompletion(
    service_id="vllm-otel", ai_model_id="qwen3.6-35b-a3b", async_client=_client)
kernel_otlp.add_service(svc_otlp)
_settings = svc_otlp.instantiate_prompt_execution_settings(service_id="vllm-otel")

# Marqueur temporel pris AVANT l'appel : la relecture Aspire (cellule suivante)
# n'affichera que les spans démarrés après lui — jamais ceux d'une exécution
# précédente encore stockés dans le dashboard.
import time as _time
_DEBUT_APPEL = _time.time()

# L'appel réel : le span gen_ai émerge à la fermeture (ConsoleSpanExporter l'imprime ci-dessous).
# function_name stable (review #15436) : sans lui, Semantic Kernel genere un
# jeton alphabetique aleatoire a chaque execution, publie dans execute_tool /
# gen_ai.tool.name / la relecture Aspire — un identifiant operationnel, pas un
# libelle pedagogique. Nommer a la source, jamais scrubber la sortie.
_result = await kernel_otlp.invoke_prompt(
    "Donne une phrase courte expliquant ce qu'est l'observabilite.",
    settings=_settings,
    function_name="definition_observabilite")
print("Reponse du modele :", str(_result).strip()[:200])
2026-09-10 22:50:07,351 - semantic_kernel.prompt_template.kernel_prompt_template - DEBUG - Extracting blocks from template: Donne une phrase courte expliquant ce qu'est l'observabilite.
2026-09-10 22:50:07,352 - semantic_kernel.functions.kernel_function - INFO - Function definition_observabilite invoking.
2026-09-10 22:50:07,352 - semantic_kernel.functions.kernel_function - DEBUG - Function arguments: {}
2026-09-10 22:50:07,353 - semantic_kernel.prompt_template.kernel_prompt_template - DEBUG - Rendering list of 1 blocks
2026-09-10 22:50:07,354 - semantic_kernel.prompt_template.kernel_prompt_template - DEBUG - Rendered prompt: Donne une phrase courte expliquant ce qu'est l'observabilite.
2026-09-10 22:50:07,354 - semantic_kernel.utils.telemetry.model_diagnostics.decorators - INFO - {"role": "user", "content": "Donne une phrase courte expliquant ce qu'est l'observabilite."}
2026-09-10 22:50:13,833 - semantic_kernel.connectors.ai.open_ai.services.open_ai_handler - INFO - OpenAI usage: CompletionUsage(completion_tokens=693, prompt_tokens=26, total_tokens=719, completion_tokens_details=None, prompt_tokens_details=None)
2026-09-10 22:50:13,835 - semantic_kernel.utils.telemetry.model_diagnostics.decorators - INFO - {"message": {"role": "assistant", "content": "\n\nL'observabilit\u00e9 est la capacit\u00e9 de comprendre l'\u00e9tat interne d'un syst\u00e8me \u00e0 partir de ses donn\u00e9es externes (logs, m\u00e9triques, traces) pour en diagnostiquer et optimiser le comportement."}, "finish_reason": "stop"}
[span capturé] chat qwen3.6-35b-a3b
    gen_ai.operation.name = 'chat'
    gen_ai.request.model = 'qwen3.6-35b-a3b'
    gen_ai.response.finish_reason = 'FinishReason.STOP'
    gen_ai.system = 'openai'
    gen_ai.usage.input_tokens = 26
    gen_ai.usage.output_tokens = 693
2026-09-10 22:50:13,836 - semantic_kernel.functions.kernel_function - INFO - Function definition_observabilite succeeded.
2026-09-10 22:50:13,837 - semantic_kernel.functions.kernel_function - DEBUG - Function result: 

L'observabilité est la capacité de comprendre l'état interne d'un système à partir de ses données externes (logs, métriques, traces) pour en diagnostiquer et optimiser le comportement.
2026-09-10 22:50:13,837 - semantic_kernel.functions.kernel_function - INFO - Function completed. Duration: 6.484786s
[span capturé] execute_tool definition_observabilite
    gen_ai.operation.name = 'execute_tool'
    gen_ai.tool.name = 'definition_observabilite'
Reponse du modele : L'observabilité est la capacité de comprendre l'état interne d'un système à partir de ses données externes (logs, métriques, traces) pour en diagnostiquer et optimiser le comportement.
# Relecture du span DEPUIS le dashboard Aspire (.NET) : la boucle est bouclée.
#
# La cellule précédente a poussé ses spans via OTLP ; ici on interroge l'API de
# télémétrie du dashboard (JSON sur /api/telemetry/spans) et on retrouve le span
# gen_ai de CE notebook avec ses attributs. Subtilité : le BatchSpanProcessor
# d'OTel exporte par lots (flush ~5 s) — on sonde donc l'API jusqu'à le voir
# arriver, avec une attente bornée.
import json as _json
import time as _time
import urllib.request as _urlreq

_DASH = "http://localhost:18888"
_SERVICE = "sk-filters-observability"

def _spans_genai_depuis_dashboard(depuis_unix_s=0.0):
    """Spans gen_ai du service de ce notebook, stockés côté dashboard Aspire (.NET),
    démarrés après ``depuis_unix_s`` (marqueur de la présente exécution)."""
    with _urlreq.urlopen(f"{_DASH}/api/telemetry/spans", timeout=10) as _resp:
        _payload = _json.loads(_resp.read().decode())
    _capturés = []
    for _rs in _payload.get("data", {}).get("resourceSpans", []):
        _res = {a["key"]: next(iter(a["value"].values()), "?")
                for a in _rs.get("resource", {}).get("attributes", [])}
        if _res.get("service.name") != _SERVICE:
            continue
        for _ss in _rs.get("scopeSpans", []):
            for _sp in _ss.get("spans", []):
                _attrs = {a["key"]: next(iter(a["value"].values()), "?")
                          for a in _sp.get("attributes", [])}
                if any(k.startswith("gen_ai.") for k in _attrs):
                    _debut_ns = int(_sp.get("startTimeUnixNano", "0"))
                    if _debut_ns >= (depuis_unix_s - 2.0) * 1e9:  # 2 s de marge d'horloge
                        _capturés.append((_sp.get("name"), _attrs))
    return _capturés

try:
    _DEBUT_APPEL = globals().get("_DEBUT_APPEL", 0.0)  # 0.0 = relecture seule (cellule ré-exécutée)
    _capturés = []
    for _essai in range(15):  # attente bornée du flush par lots (~15 s max)
        _capturés = _spans_genai_depuis_dashboard(_DEBUT_APPEL)
        if _capturés:
            break
        _time.sleep(1)
    if _capturés:
        print(f"{len(_capturés)} span(s) gen_ai du service '{_SERVICE}' relu(s) depuis le dashboard Aspire ({_DASH}) :")
        for _name, _attrs in _capturés[-3:]:
            print(f"  {_name}")
            # Même allowlist pédagogique que le printer de la cellule de setup :
            # pas d'identifiant de télémétrie opérationnelle dans les outputs
            # publics (review #15436).
            for _k in sorted(_attrs):
                if _k in _ATTRS_PEDAGOGIQUES:
                    _v = str(_attrs[_k])
                    if len(_v) > 120:
                        _v = _v[:120] + "…(tronqué)"
                    print(f"      {_k} = {_v}")
        print("Pont Python -> .NET vérifié : le span émis par SK est stocké dans le dashboard Aspire.")
    else:
        print("Dashboard joignable mais aucun span gen_ai DE CET appel stocké après 15 s — re-exécutez la cellule d'appel au LLM.")
except Exception as _exc:
    print(f"Dashboard Aspire non joignable ({_exc}) — la relecture requiert le backend aspire-otel/ démarré (mode standalone, cf. README).")
2 span(s) gen_ai du service 'sk-filters-observability' relu(s) depuis le dashboard Aspire (http://localhost:18888) :
  execute_tool definition_observabilite
      gen_ai.operation.name = execute_tool
      gen_ai.tool.name = definition_observabilite
  chat qwen3.6-35b-a3b
      gen_ai.operation.name = chat
      gen_ai.request.model = qwen3.6-35b-a3b
      gen_ai.response.finish_reason = FinishReason.STOP
      gen_ai.system = openai
      gen_ai.usage.input_tokens = 26
      gen_ai.usage.output_tokens = 693
Pont Python -> .NET vérifié : le span émis par SK est stocké dans le dashboard Aspire.

Lecture du span capturé

La sortie de la cellule précédente contient le span JSON produit par SK. On y lit, avec leurs noms exacts tels que normalisés par les conventions sémantiques GenAI d’OTel :

  • gen_ai.system : openai (le connecteur utilisé, indépendamment du moteur réel derrière l’endpoint compatible) ;
  • gen_ai.request.model : le modèle invoqué ;
  • gen_ai.usage.input_tokens et gen_ai.usage.output_tokens : les tokens consommés (prompt + génération) — la grandeur facturable, capturée automatiquement ;
  • gen_ai.operation.name : chat pour l’appel LLM, et un span frère execute_tool pour l’invocation de fonction SK.

Note de parité :

Côté Python, le collecteur OTEL (ou un SaaS comme LangSmith / Langfuse) joue le même rôle d’agrégation que le dashboard Aspire côté .NET. La différence : Aspire embarque le dashboard + l’endpoint OTLP out-of-the-box, là où l’écosystème Python assemble ces briques séparément. Le protocole (OTLP) est commun — un backend .NET peut donc observer une chaîne Python, et réciproquement.

Voir aspire-otel/README.md pour démarrer le backend (dashboard Aspire en mode standalone) et visualiser ces spans.

La cellule de relecture ci-dessus interroge l’API de télémétrie du dashboard (/api/telemetry/spans) : le span n’est pas seulement affiché dans une UI, il est relu programmatiquement depuis le côté .NET — la preuve enregistrée que le pont OTLP fonctionne dans les deux sens. En ligne de commande, aspire otel spans --dashboard-url http://localhost:18888 --search gen_ai donne la même lecture.

Exercice 4 : Instrumenter un filtre avec un span personnalisé

Objectif : Ajouter un span personnalisé autour d’un filtre (par ex. le filtre de cache de l’Exercice 2) pour mesurer son effet. Créez un span manuellement avec tracer.start_as_current_span("cache_lookup"), ajoutez-lui un attribut cache.hit (booléen), et comparez la latence cache-hit vs cache-miss sur 5 appels.

Indice : trace.get_tracer(__name__) récupère le tracer configuré plus haut ; un bloc with tracer.start_as_current_span("cache_lookup") as span: pose le span, et span.set_attribute("cache.hit", True) l’annote.

# Exercice 4 (à compléter) : span personnalisé sur un filtre.
# Squelette — instrumentez le filtre de cache (Exercice 2) avec un span "cache_lookup".
from opentelemetry import trace

_tracer = trace.get_tracer(__name__)

def cache_lookup_with_span(key, cache):
    """Mesure le hit/miss d'un cache sous un span personnalisé."""
    # TODO étudiant : ouvrez un span "cache_lookup" avec start_as_current_span,
    # positionnez l'attribut "cache.hit" (True/False) selon le résultat,
    # et retournez (valeur_ou_None, hit: bool).
    pass

# Test : 5 appels (le 1er miss, les suivants hit) puis lecture des latences côté dashboard.
resultat = None  # TODO : boucle sur 5 appels et collecte des latences
print("Exercice à compléter : instrumentez cache_lookup_with_span puis mesurez 5 appels.")
Exercice à compléter : instrumentez cache_lookup_with_span puis mesurez 5 appels.

Conclusion

Resume des concepts

Concept Description Code cle
Function Filter Intercepte avant/après fonctions @kernel.filter(FilterTypes.FUNCTION_INVOCATION)
Prompt Filter Modifie le prompt @kernel.filter(FilterTypes.PROMPT_RENDERING)
Auto-Invoke Filter Contrôle function calling @kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION)
Logging Debug avec logging standard logging.getLogger("semantic_kernel")
OpenTelemetry Monitoring production Traces, metriques, logs

Points cles a retenir

  1. Les filtres suivent le pattern middleware - await next(context) passe au suivant
  2. Plusieurs filtres peuvent etre chaînes - Ordre d’enregistrement = ordre d’exécution
  3. Le contexte est mutable - On peut modifier arguments, résultats, prompts
  4. terminate = True - Stoppe l’exécution (utile pour le rate limiting)
  5. OpenTelemetry pour la production - Traces et metriques automatiques

Exercices suggeres

  1. Filtre de cache : Créer un filtre qui cache les résultats de fonctions
  2. Filtre d’anonymisation : Masquer les emails/numéros dans les prompts
  3. Filtre d’approbation : Demander confirmation avant les opérations sensibles

Pour aller plus loin

Notebook Contenu
05-VectorStores RAG avec Qdrant
06-ProcessFramework Workflows orchestres

Navigation : << 03-Agents | Index | 05-VectorStores >>

Retour au sommet