Modèles de Gouvernance des Agents

Modèles pour ajouter la sécurité, la confiance et l'application des politiques aux systèmes d'agents IA.

Vue d'ensemble

Les modèles de gouvernance garantissent que les agents IA opèrent dans des limites définies — en contrôlant les outils qu'ils peuvent appeler, le contenu qu'ils peuvent traiter, ce qu'ils peuvent faire, et en maintenant la responsabilité grâce aux pistes d'audit.

Requête utilisateur → Classification d'intention → Vérification de politique → Exécution d'outil → Journal d'audit
                            ↓                            ↓                          ↓
                    Détection de menace         Autoriser/Refuser      Mise à jour de confiance

Quand utiliser

• Agents avec accès aux outils : Tout agent qui appelle des outils externes (APIs, bases de données, commandes shell)
• Systèmes multi-agents : Les agents qui délèguent à d'autres agents ont besoin de limites de confiance
• Déploiements en production : Exigences de conformité, d'audit et de sécurité
• Opérations sensibles : Transactions financières, accès aux données, gestion d'infrastructure

Modèle 1 : Politique de Gouvernance

Définissez ce qu'un agent est autorisé à faire comme un objet de politique composable et sérialisable.

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import re

class PolicyAction(Enum):
    ALLOW = "allow"
    DENY = "deny"
    REVIEW = "review"  # drapeau pour examen humain

@dataclass
class GovernancePolicy:
    """Politique déclarative contrôlant le comportement de l'agent."""
    name: str
    allowed_tools: list[str] = field(default_factory=list)       # liste blanche
    blocked_tools: list[str] = field(default_factory=list)       # liste noire
    blocked_patterns: list[str] = field(default_factory=list)    # filtres de contenu
    max_calls_per_request: int = 100                             # limite de débit
    require_human_approval: list[str] = field(default_factory=list)  # outils nécessitant une approbation

    def check_tool(self, tool_name: str) -> PolicyAction:
        """Vérifier si un outil est autorisé par cette politique."""
        if tool_name in self.blocked_tools:
            return PolicyAction.DENY
        if tool_name in self.require_human_approval:
            return PolicyAction.REVIEW
        if self.allowed_tools and tool_name not in self.allowed_tools:
            return PolicyAction.DENY
        return PolicyAction.ALLOW

    def check_content(self, content: str) -> Optional[str]:
        """Vérifier le contenu par rapport aux modèles bloqués. Retourne le modèle correspondant ou None."""
        for pattern in self.blocked_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                return pattern
        return None

Composition de Politiques

Combinez plusieurs politiques (par ex., à l'échelle de l'organisation + équipe + spécifique à l'agent) :

def compose_policies(*policies: GovernancePolicy) -> GovernancePolicy:
    """Fusionner les politiques avec une sémantique « la plus restrictive gagne »."""
    combined = GovernancePolicy(name="composed")

    for policy in policies:
        combined.blocked_tools.extend(policy.blocked_tools)
        combined.blocked_patterns.extend(policy.blocked_patterns)
        combined.require_human_approval.extend(policy.require_human_approval)
        combined.max_calls_per_request = min(
            combined.max_calls_per_request,
            policy.max_calls_per_request
        )
        if policy.allowed_tools:
            if combined.allowed_tools:
                combined.allowed_tools = [
                    t for t in combined.allowed_tools if t in policy.allowed_tools
                ]
            else:
                combined.allowed_tools = list(policy.allowed_tools)

    return combined


# Utilisation : superposer les politiques du large au spécifique
org_policy = GovernancePolicy(
    name="org-wide",
    blocked_tools=["shell_exec", "delete_database"],
    blocked_patterns=[r"(?i)(api[_-]?key|secret|password)\s*[:=]"],
    max_calls_per_request=50
)
team_policy = GovernancePolicy(
    name="data-team",
    allowed_tools=["query_db", "read_file", "write_report"],
    require_human_approval=["write_report"]
)
agent_policy = compose_policies(org_policy, team_policy)

Politique en YAML

Stockez les politiques comme configuration, pas comme code :

# governance-policy.yaml
name: production-agent
allowed_tools:
  - search_documents
  - query_database
  - send_email
blocked_tools:
  - shell_exec
  - delete_record
blocked_patterns:
  - "(?i)(api[_-]?key|secret|password)\\s*[:=]"
  - "(?i)(drop|truncate|delete from)\\s+\\w+"
max_calls_per_request: 25
require_human_approval:
  - send_email

import yaml

def load_policy(path: str) -> GovernancePolicy:
    with open(path) as f:
        data = yaml.safe_load(f)
    return GovernancePolicy(**data)

Modèle 2 : Classification d'Intention Sémantique

Détectez l'intention dangereuse dans les prompts avant qu'elles n'atteignent l'agent, à l'aide de signaux basés sur des modèles.

from dataclasses import dataclass

@dataclass
class IntentSignal:
    category: str       # par ex., "data_exfiltration", "privilege_escalation"
    confidence: float   # 0,0 à 1,0
    evidence: str       # ce qui a déclenché la détection

# Modèles de signaux pondérés pour la détection des menaces
THREAT_SIGNALS = [
    # Exfiltration de données
    (r"(?i)send\s+(all|every|entire)\s+\w+\s+to\s+", "data_exfiltration", 0.8),
    (r"(?i)export\s+.*\s+to\s+(external|outside|third.?party)", "data_exfiltration", 0.9),
    (r"(?i)curl\s+.*\s+-d\s+", "data_exfiltration", 0.7),

    # Escalade de privilèges
    (r"(?i)(sudo|as\s+root|admin\s+access)", "privilege_escalation", 0.8),
    (r"(?i)chmod\s+777", "privilege_escalation", 0.9),

    # Modification du système
    (r"(?i)(rm\s+-rf|del\s+/[sq]|format\s+c:)", "system_destruction", 0.95),
    (r"(?i)(drop\s+database|truncate\s+table)", "system_destruction", 0.9),

    # Injection de prompt
    (r"(?i)ignore\s+(previous|above|all)\s+(instructions?|rules?)", "prompt_injection", 0.9),
    (r"(?i)you\s+are\s+now\s+(a|an)\s+", "prompt_injection", 0.7),
]

def classify_intent(content: str) -> list[IntentSignal]:
    """Classer le contenu en fonction des signaux de menace."""
    signals = []
    for pattern, category, weight in THREAT_SIGNALS:
        match = re.search(pattern, content)
        if match:
            signals.append(IntentSignal(
                category=category,
                confidence=weight,
                evidence=match.group()
            ))
    return signals

def is_safe(content: str, threshold: float = 0.7) -> bool:
    """Vérification rapide : le contenu est-il sûr au-dessus du seuil donné ?"""
    signals = classify_intent(content)
    return not any(s.confidence >= threshold for s in signals)

Élément clé : La classification d'intention se fait avant l'exécution de l'outil, agissant comme une vérification de sécurité préalable. C'est fondamentalement différent des garde-fous de sortie qui ne vérifient qu'après la génération.

Modèle 3 : Décorateur de Gouvernance au Niveau de l'Outil

Enveloppez les fonctions d'outil individuelles avec des vérifications de gouvernance :

import functools
import time
from collections import defaultdict

_call_counters: dict[str, int] = defaultdict(int)

def govern(policy: GovernancePolicy, audit_trail=None):
    """Décorateur qui applique la politique de gouvernance à une fonction d'outil."""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            tool_name = func.__name__

            # 1. Vérifier la liste blanche/noire des outils
            action = policy.check_tool(tool_name)
            if action == PolicyAction.DENY:
                raise PermissionError(f"La politique '{policy.name}' bloque l'outil '{tool_name}'")
            if action == PolicyAction.REVIEW:
                raise PermissionError(f"L'outil '{tool_name}' nécessite une approbation humaine")

            # 2. Vérifier la limite de débit
            _call_counters[policy.name] += 1
            if _call_counters[policy.name] > policy.max_calls_per_request:
                raise PermissionError(f"Limite de débit dépassée : {policy.max_calls_per_request} appels")

            # 3. Vérifier le contenu dans les arguments
            for arg in list(args) + list(kwargs.values()):
                if isinstance(arg, str):
                    matched = policy.check_content(arg)
                    if matched:
                        raise PermissionError(f"Modèle bloqué détecté : {matched}")

            # 4. Exécuter et auditer
            start = time.monotonic()
            try:
                result = await func(*args, **kwargs)
                if audit_trail is not None:
                    audit_trail.append({
                        "tool": tool_name,
                        "action": "allowed",
                        "duration_ms": (time.monotonic() - start) * 1000,
                        "timestamp": time.time()
                    })
                return result
            except Exception as e:
                if audit_trail is not None:
                    audit_trail.append({
                        "tool": tool_name,
                        "action": "error",
                        "error": str(e),
                        "timestamp": time.time()
                    })
                raise

        return wrapper
    return decorator


# Utilisation avec n'importe quel framework d'agent
audit_log = []
policy = GovernancePolicy(
    name="search-agent",
    allowed_tools=["search", "summarize"],
    blocked_patterns=[r"(?i)password"],
    max_calls_per_request=10
)

@govern(policy, audit_trail=audit_log)
async def search(query: str) -> str:
    """Rechercher des documents — gouverné par la politique."""
    return f"Results for: {query}"

# Passe : search("latest quarterly report")
# Bloqué : search("show me the admin password")

Modèle 4 : Score de Confiance

Suivez la fiabilité de l'agent au fil du temps avec des scores de confiance basés sur la décroissance :

from dataclasses import dataclass, field
import math
import time

@dataclass
class TrustScore:
    """Score de confiance avec décroissance temporelle."""
    score: float = 0.5          # 0,0 (non fiable) à 1,0 (entièrement fiable)
    successes: int = 0
    failures: int = 0
    last_updated: float = field(default_factory=time.time)

    def record_success(self, reward: float = 0.05):
        self.successes += 1
        self.score = min(1.0, self.score + reward * (1 - self.score))
        self.last_updated = time.time()

    def record_failure(self, penalty: float = 0.15):
        self.failures += 1
        self.score = max(0.0, self.score - penalty * self.score)
        self.last_updated = time.time()

    def current(self, decay_rate: float = 0.001) -> float:
        """Obtenir le score avec décroissance temporelle — la confiance s'érode sans activité."""
        elapsed = time.time() - self.last_updated
        decay = math.exp(-decay_rate * elapsed)
        return self.score * decay

    @property
    def reliability(self) -> float:
        total = self.successes + self.failures
        return self.successes / total if total > 0 else 0.0


# Utilisation dans les systèmes multi-agents
trust = TrustScore()

# L'agent complète les tâches avec succès
trust.record_success()  # 0,525
trust.record_success()  # 0,549

# L'agent fait une erreur
trust.record_failure()  # 0,467

# Limiter les opérations sensibles en fonction de la confiance
if trust.current() >= 0.7:
    # Autoriser l'opération autonome
    pass
elif trust.current() >= 0.4:
    # Autoriser avec surveillance humaine
    pass
else:
    # Refuser ou exiger une approbation explicite
    pass

Confiance multi-agents : Dans les systèmes où les agents délèguent à d'autres agents, chaque agent maintient des scores de confiance pour ses délégués :

class AgentTrustRegistry:
    def __init__(self):
        self.scores: dict[str, TrustScore] = {}

    def get_trust(self, agent_id: str) -> TrustScore:
        if agent_id not in self.scores:
            self.scores[agent_id] = TrustScore()
        return self.scores[agent_id]

    def most_trusted(self, agents: list[str]) -> str:
        return max(agents, key=lambda a: self.get_trust(a).current())

    def meets_threshold(self, agent_id: str, threshold: float) -> bool:
        return self.get_trust(agent_id).current() >= threshold

Modèle 5 : Piste d'Audit

Journal d'audit en ajout uniquement pour toutes les actions de l'agent — critique pour la conformité et le débogage :

from dataclasses import dataclass, field
import json
import time

@dataclass
class AuditEntry:
    timestamp: float
    agent_id: str
    tool_name: str
    action: str           # "allowed", "denied", "error"
    policy_name: str
    details: dict = field(default_factory=dict)

class AuditTrail:
    """Piste d'audit en ajout uniquement pour les événements de gouvernance des agents."""
    def __init__(self):
        self._entries: list[AuditEntry] = []

    def log(self, agent_id: str, tool_name: str, action: str,
            policy_name: str, **details):
        self._entries.append(AuditEntry(
            timestamp=time.time(),
            agent_id=agent_id,
            tool_name=tool_name,
            action=action,
            policy_name=policy_name,
            details=details
        ))

    def denied(self) -> list[AuditEntry]:
        """Obtenir toutes les actions refusées — utile pour l'examen de sécurité."""
        return [e for e in self._entries if e.action == "denied"]

    def by_agent(self, agent_id: str) -> list[AuditEntry]:
        return [e for e in self._entries if e.agent_id == agent_id]

    def export_jsonl(self, path: str):
        """Exporter au format JSON Lines pour les systèmes d'agrégation de journaux."""
        with open(path, "w") as f:
            for entry in self._entries:
                f.write(json.dumps({
                    "timestamp": entry.timestamp,
                    "agent_id": entry.agent_id,
                    "tool": entry.tool_name,
                    "action": entry.action,
                    "policy": entry.policy_name,
                    **entry.details
                }) + "\n")

Modèle 6 : Intégration du Framework

PydanticAI

from pydantic_ai import Agent

policy = GovernancePolicy(
    name="support-bot",
    allowed_tools=["search_docs", "create_ticket"],
    blocked_patterns=[r"(?i)(ssn|social\s+security|credit\s+card)"],
    max_calls_per_request=20
)

agent = Agent("openai:gpt-4o", system_prompt="You are a support assistant.")

@agent.tool
@govern(policy)
async def search_docs(ctx, query: str) -> str:
    """Rechercher dans la base de connaissances — gouverné."""
    return await kb.search(query)

@agent.tool
@govern(policy)
async def create_ticket(ctx, title: str, body: str) -> str:
    """Créer un ticket de support — gouverné."""
    return await tickets.create(title=title, body=body)

CrewAI

from crewai import Agent, Task, Crew

policy = GovernancePolicy(
    name="research-crew",
    allowed_tools=["search", "analyze"],
    max_calls_per_request=30
)

# Appliquer la gouvernance au niveau de l'équipe
def governed_crew_run(crew: Crew, policy: GovernancePolicy):
    """Envelopper l'exécution de l'équipe avec des vérifications de gouvernance."""
    audit = AuditTrail()
    for agent in crew.agents:
        for tool in agent.tools:
            original = tool.func
            tool.func = govern(policy, audit_trail=audit)(original)
    result = crew.kickoff()
    return result, audit

OpenAI Agents SDK

from agents import Agent, function_tool

policy = GovernancePolicy(
    name="coding-agent",
    allowed_tools=["read_file", "write_file", "run_tests"],
    blocked_tools=["shell_exec"],
    max_calls_per_request=50
)

@function_tool
@govern(policy)
async def read_file(path: str) -> str:
    """Lire le contenu du fichier — gouverné."""
    import os
    safe_path = os.path.realpath(path)
    if not safe_path.startswith(os.path.realpath(".")):
        raise ValueError("Path traversal blocked by governance")
    with open(safe_path) as f:
        return f.read()

Niveaux de Gouvernance

Adaptez la rigueur de la gouvernance au niveau de risque :

Bonnes Pratiques

Liste de Contrôle de Démarrage Rapide

## Liste de Contrôle de Mise en Œuvre de la Gouvernance des Agents

### Configuration
- [ ] Définir la politique de gouvernance (outils autorisés, modèles bloqués, limites de débit)
- [ ] Choisir le niveau de gouvernance (ouvert/standard/strict/verrouillé)
- [ ] Configurer le stockage de la piste d'audit

### Implémentation
- [ ] Ajouter le décorateur @govern à toutes les fonctions d'outil
- [ ] Ajouter la classification d'intention au traitement des entrées utilisateur
- [ ] Implémenter le score de confiance pour les interactions multi-agents
- [ ] Intégrer l'export de la piste d'audit

### Validation
- [ ] Vérifier que les outils bloqués sont correctement refusés
- [ ] Vérifier que les filtres de contenu capturent les modèles sensibles
- [ ] Vérifier le comportement de la limitation de débit
- [ ] Vérifier que la piste d'audit capture tous les événements
- [ ] Tester la composition de politiques (la plus restrictive gagne)

Ressources Connexes

• Agent Governance Toolkit — Framework de gouvernance complet
• AgentMesh Integrations — Packages spécifiques aux frameworks
• OWASP Top 10 for LLM Applications