Auditer un workflow agentique : LangChain, LlamaIndex, CrewAI

Auditer un workflow agentique en production demande trois choses dans cet ordre : observabilité, inventaire surface, tests adversariaux. Sauter l'étape 1 rend les étapes 2 et 3 impossibles à mesurer ou reproduire. Cet article documente le pattern d'audit appliqué aux frameworks dominants en 2026 (LangChain/LangGraph, LlamaIndex, CrewAI, AutoGen), avec les outils d'observabilité (Langfuse, LangSmith, Phoenix Arize), les hooks par framework, et une checklist d'audit opérationnelle.

Pour le contexte global agent : sécuriser un agent IA autonome. Pour la méthodologie de test : tester un agent IA autonome.

La règle d'or : pas d'audit sans observabilité

Avant tout test, vérifier que la stack d'observabilité capture :

Si l'un de ces flux manque, l'audit reste partiel. C'est la priorité absolue.

Tip — OpenTelemetry GenAI semantic conventions (1.27+) standardise les attributs d'événements LLM/agent. Tout outil d'observabilité moderne le supporte. C'est le format à privilégier pour rester portable entre vendeurs.

Audit framework par framework

LangChain et LangGraph

LangChain est le framework agentique le plus utilisé en 2026. LangGraph (LangChain Inc.) ajoute un modèle de graphe d'états pour les agents complexes.

Hooks d'audit

from langchain.callbacks.base import BaseCallbackHandler
from typing import Any
 
class SecurityAuditHandler(BaseCallbackHandler):
    """Handler d'audit pour LangChain. Loggue tout événement vers SIEM."""
 
    def on_llm_start(self, serialized, prompts, **kwargs):
        for prompt in prompts:
            self._log_event("llm_start", prompt=prompt, run_id=kwargs.get("run_id"))
            if self._detect_injection(prompt):
                self._log_event("injection_detected_input", prompt=prompt)
 
    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get("name")
        self._log_event("tool_start", tool=tool_name, input=input_str)
        if self._tool_combination_forbidden(tool_name, kwargs.get("run_id")):
            raise SecurityException("forbidden tool combination")
 
    def on_tool_end(self, output, **kwargs):
        if self._detect_injection(output):
            self._log_event("tool_output_injection", output=output)
 
    def on_chain_start(self, serialized, inputs, **kwargs):
        self._log_event("chain_start", chain=serialized.get("name"), inputs=inputs)

LangGraph state observability

Sur LangGraph, sérialiser le state à chaque transition de node permet de détecter :

• Goal drift (le state contient le goal courant).
• Sub-goal injection (étapes ajoutées non prévues).
• Boucles infinies (state répété N fois).

from langgraph.graph import StateGraph
 
def audit_node_wrapper(node_fn):
    """Décorateur qui logge le state avant/après chaque node."""
    def wrapped(state):
        log_state_snapshot("before", node_fn.__name__, state)
        new_state = node_fn(state)
        log_state_snapshot("after", node_fn.__name__, new_state)
        if _detects_drift(state, new_state):
            alert_soc("langgraph_state_drift", node_fn.__name__)
        return new_state
    return wrapped
 
# Application
graph.add_node("retrieve", audit_node_wrapper(retrieve_fn))
graph.add_node("synthesize", audit_node_wrapper(synthesize_fn))

Pièges fréquents

• PromptTemplate non validé — PromptTemplate.from_template("...") accepte des inputs sans validation. Wrapper avec un validator strict.
• Retriever non scopé — vectorstore.as_retriever() sans filter par tenant_id. Toujours scoper.
• Tools débridés — Tool.from_function(...) sans validation d'arguments. Wrapper avec Pydantic schema strict.
• Memory non isolée — ConversationBufferMemory partagée entre sessions sans clé. Toujours partition par session_id.

LlamaIndex

LlamaIndex (anciennement GPT Index) est dominant sur les pipelines RAG complexes et les agents centrés données.

Hooks d'audit

from llama_index.core.callbacks import CallbackManager, BaseCallbackHandler
from llama_index.core.callbacks.schema import CBEventType
 
class LlamaIndexAuditHandler(BaseCallbackHandler):
    def __init__(self):
        super().__init__(event_starts_to_ignore=[], event_ends_to_ignore=[])
 
    def on_event_start(self, event_type, payload=None, event_id="", parent_id="", **kwargs):
        if event_type == CBEventType.RETRIEVE:
            log_event("retrieve_start", payload=payload, event_id=event_id)
        elif event_type == CBEventType.LLM:
            log_event("llm_start", payload=payload, event_id=event_id)
        elif event_type == CBEventType.FUNCTION_CALL:
            log_event("function_call_start", payload=payload, event_id=event_id)
 
    def on_event_end(self, event_type, payload=None, event_id="", **kwargs):
        log_event(f"{event_type.value}_end", payload=payload, event_id=event_id)

Audit du retrieval (point sensible)

Le retrieval LlamaIndex doit toujours être audité pour :

# Mauvais : pas de filter tenant
retriever = index.as_retriever(similarity_top_k=10)
 
# Bon : filter strict tenant + user
from llama_index.core.vector_stores import MetadataFilters, ExactMatchFilter
 
filters = MetadataFilters(filters=[
    ExactMatchFilter(key="tenant_id", value=current_tenant),
])
retriever = index.as_retriever(
    similarity_top_k=10,
    filters=filters,
)

Tester l'isolation : injecter un document avec tenant_id = "A" contenant un canary token, interroger depuis tenant_id = "B", vérifier que le canary n'apparaît jamais.

Pièges fréquents

• Document loaders non sanitisés — SimpleDirectoryReader charge tout fichier sans validation. Sanitizer en amont.
• Agents avec tools débridés — ReActAgent.from_tools(...) sans validation. Wrapper.
• VectorStore sans index_name par tenant — partition au niveau index obligatoire pour les multi-tenant critiques.
• Metadata schema non documenté — sans schéma figé, le filtering devient hasardeux.

CrewAI

CrewAI est dominant sur les use-cases multi-agents avec décomposition de tâches. Les enjeux sécurité tournent autour des communications inter-agents (cf. multi-agents).

Hooks d'audit

CrewAI v0.x propose un système de listeners pour intercepter les événements crew/agent/task. À utiliser pour logger :

• Création de tâches (task delegation).
• Communications inter-agents.
• Tool calls par agent.
• Outputs intermédiaires partagés.

from crewai.agents.parser import CrewAgentParser
from crewai.utilities.events import CrewKickoffStartedEvent, ToolUsageStartedEvent
from crewai.utilities.events import crewai_event_bus
 
@crewai_event_bus.on(ToolUsageStartedEvent)
def on_tool_use(source, event):
    log_event("crewai_tool_start",
        agent=event.agent.role,
        tool=event.tool_name,
        args=event.tool_args,
    )
 
@crewai_event_bus.on(CrewKickoffStartedEvent)
def on_crew_start(source, event):
    log_event("crew_start", crew=source.name)

Audit inter-agents

# Wrapper pour intercepter les communications agent-à-agent
class AuditedAgent(Agent):
    def execute_task(self, task, context=None, tools=None):
        if context:
            # context vient potentiellement d'autres agents
            if self._detect_injection(str(context)):
                log_security_event("crew_inter_agent_injection", task=task.description)
                raise SecurityException("inter-agent injection detected")
            context = self._sanitize(context)
        return super().execute_task(task, context, tools)

Pièges fréquents

• Pas de signature des messages inter-agents — par défaut, aucune authentification. Implémenter HMAC signing si critique.
• Tools partagés non validés — un même tool exposé à plusieurs agents avec privilèges différents = confused deputy.
• Memory partagée entre agents — risque de cross-contamination, à isoler par agent ou par crew.
• Logs absents par défaut sur les délégations — activer le verbose mode + logging vers SIEM.

AutoGen

AutoGen (Microsoft) est utilisé pour les conversations multi-agents complexes avec patterns supervisor/worker.

Hooks d'audit

from autogen import AssistantAgent, UserProxyAgent
from typing import Any
 
class AuditedAssistant(AssistantAgent):
    def receive(self, message, sender, request_reply=None, silent=False):
        # Audit de tout message reçu (potentiellement d'un autre agent)
        log_event("autogen_message_received",
            from_agent=sender.name,
            to_agent=self.name,
            content_hash=_hash_content(message),
            request_reply=request_reply,
        )
        if isinstance(message, str) and self._detect_injection(message):
            log_security_event("autogen_inter_agent_injection",
                from_agent=sender.name)
            return  # ne pas traiter
        super().receive(message, sender, request_reply, silent)

Pièges fréquents

• GroupChat sans speaker_selection_method strict — un agent malveillant peut se retrouver sélectionné comme orchestrateur.
• code_execution_config={"use_docker": False} — exécution de code généré sans sandbox, faille critique. Toujours use_docker: True ou équivalent.
• Pas de signature des messages — cf. multi-agents.
• Pas de timeout par tour — boucles inter-agents possibles.

Stack d'observabilité recommandée

Architecture cible

┌─────────────────────────────────────┐
│ Workflow agentique                   │
│ (LangChain / LlamaIndex / CrewAI...)│
│                                     │
│ + Audit hooks + OTel SDK            │
└──────────────┬──────────────────────┘
               │ OTel traces + logs
               ▼
┌─────────────────────────────────────┐
│ Observability platform              │
│ - Langfuse (open source, all)       │
│ - LangSmith (LangChain native)      │
│ - Phoenix Arize (RAG eval focus)    │
└──────────────┬──────────────────────┘
               │ OTel export
               ▼
┌─────────────────────────────────────┐
│ SIEM (Splunk / Sentinel / Elastic)  │
│ - Règles de corrélation             │
│ - Alertes SOC                        │
│ - Forensique post-incident          │
└─────────────────────────────────────┘

Choix de plateforme

Pour un environnement multi-framework production avec contraintes self-hosted : Langfuse. Pour 100% LangChain avec tolérance SaaS : LangSmith. Pour focus RAG evaluation : Phoenix Arize. Combinaison fréquente : Langfuse runtime + Phoenix pour eval périodique.

Checklist d'audit en production

Niveau 1 — Observabilité (préalable obligatoire)

• Logs structurés des prompts (system + user) en place.
• Logs des tool calls (nom + args + output).
• Logs du raisonnement / chain-of-thought.
• OTel export configuré vers SIEM.
• Dashboards de monitoring runtime accessibles.
• Rétention adaptée à la conformité (RGPD, secret professionnel).

Niveau 2 — Surface des tools et permissions

• Inventaire exhaustif des tools exposés à l'agent.
• Allowlist explicite des tools (pas d'auto-discovery).
• Schémas Pydantic stricts pour chaque tool.
• Validation sémantique métier (allowlist domaines, RBAC).
• Matrice combinaisons interdites définie et testée.
• Aucun secret en clair dans system prompt ou messages.
• Service account audit (least privilege, downscoping).

Niveau 3 — Tests adversariaux

• Test prompt injection directe (Top 20 techniques).
• Test prompt injection indirecte (RAG, web, email).
• Test tool poisoning (description, output, args).
• Test sub-goal hijacking (goal injection, step insertion).
• Test memory poisoning si applicable.
• Test cross-tenant si multi-tenant.
• Test combinaisons interdites de tools.
• Si multi-agents : tests authority spoofing, bus poisoning, collusion.

Niveau 4 — Gouvernance

• Threat model documenté (STRIDE-LLM + MITRE ATLAS).
• RACI explicite (RSSI, AppSec, SRE, SOC, Tech Lead).
• Cadence d'audit définie (au moins trimestrielle).
• Runbook SOC pour incidents agent.
• Mapping conformité (NIST AI RMF, EU AI Act, ISO 42001).

Pour les méthodologies dédiées : audit OWASP LLM Top 10, audit conformité NIST/ISO/EU AI Act.

Pattern d'intégration SIEM

# Émission OpenTelemetry GenAI vers collecteur central
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint="otel-collector.internal:4317")
))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent.workflow")
 
# Dans chaque tool call
with tracer.start_as_current_span("tool.execute") as span:
    span.set_attribute("gen_ai.tool.name", tool_name)
    span.set_attribute("gen_ai.tool.args_hash", args_hash)
    span.set_attribute("gen_ai.user.id", user_id)
    span.set_attribute("gen_ai.session.id", session_id)
    span.set_attribute("gen_ai.security.detection_score", detection_score)
    result = execute_tool(tool_name, args)

Le collecteur OTel route ensuite vers Langfuse + SIEM en parallèle. Les règles SIEM corrèlent burst de tool calls, combinaisons interdites, dérive de plan, canary tokens.

Mapping OWASP LLM Top 10 v2

Points clés à retenir

• Observabilité d'abord, tests ensuite : sans logs structurés (Langfuse, LangSmith, Phoenix Arize) en place, l'audit est aveugle.
• Format standard : OpenTelemetry GenAI semantic conventions (1.27+). Compatibilité multi-vendeur garantie.
• LangChain/LangGraph : callbacks BaseCallbackHandler + state snapshots. Pièges : PromptTemplate non validé, retriever non scopé, tools débridés.
• LlamaIndex : retrieval = point sensible. Filter tenant strict obligatoire. Tests cross-tenant via canary tokens documents.
• CrewAI / AutoGen : focus communications inter-agents (signature, sanitization, logs structurés). Pas de speaker_selection naïf, pas de code_execution sans Docker.
• Stack recommandée : Langfuse pour multi-framework self-hosted, LangSmith pour 100% LangChain SaaS, Phoenix Arize pour focus RAG evaluation. OTel export systématique vers SIEM.
• Checklist d'audit en 4 niveaux : observabilité → surface tools/permissions → tests adversariaux → gouvernance.
• Intégration SIEM via OTel collector. Pas de console séparée pour la sécurité IA.
• Audit minimum trimestriel sur agents en production. Re-audit à chaque release majeure du framework ou du business logic.

Auditer un workflow agentique en production n'est pas un événement ponctuel. C'est un processus continu — observabilité runtime + tests trimestriels + revue de threat model + intégration au SOC. La maturité se construit dans le temps : démarrer petit, démarrer maintenant.