Accueil / Articles / Après le tutoriel LangGraph : arêtes rigides, schémas et couches de sécurité

langgraph safety structured-output agents sql production

Après le tutoriel LangGraph : arêtes rigides, schémas et couches de sécurité

Transformez un agent SQL fonctionnel en un agent sécurisé grâce à des contrôles de type, à la mise à jour du schéma, au routage basé sur les coûts et à des vérifications en plusieurs niveaux.

2742 mots

Après le tutoriel, la validation verte

Les tutoriels de LangGraph permettent de faire fonctionner un graphe. L’évaluation commence lorsque quelqu’un demande pourquoi chaque décision est sûre. Cette réécriture décrit une semaine de travail visant à transformer un agent fonctionnel de style analyste SQL en quelque chose défendable : des architectures qui ne peuvent pas ignorer la sécurité, des sorties structurées, une fraîcheur du schéma, une conscience des coûts, ainsi que des erreurs détectées lors de leur mise par écrit.

Situation et tâche

Les tutoriels fournissent des chemins de codes cadeau, pas des invariants. La tâche consistait à conserver un projet fonctionnel tout en rendant chaque branche explicable lors de l’évaluation — en particulier les branches qui exécutent du SQL.

Action 1 — deux architectures afin qu’aucune ne puisse contourner la protection

Une barrière de sécurité doit être une contrainte stricte, et non une suggestion issue d’une instruction. Les jugements structurés doivent figurer dans des schémas typés :

class JudgeAgentSchema(BaseModel):
    answer: Literal["yes", "no"] = Field(
        description="Return 'yes' if the SQL query ONLY retrieves data (like SELECT). "
                    "Return 'no' if it modifies data (like INSERT, UPDATE, DELETE, DROP)."
    )
    comments: str = Field(default="", description="Reasoning behind the verdict")


llm_judge = llm.with_structured_output(schema=JudgeAgentSchema)

Itinéraire avec une condition explicite :

def is_safe_sql_condition(state: AgentSchema) -> str:
    if state.is_safe.lower() == "yes":
        return "Execute_SQL"
    return "Cancel_SQL_if_Not_Safe"

Les erreurs de validation apparaissent lorsque le modèle s’éloigne du vocabulaire littéral :

ValidationError: Input should be 'yes' or 'no'
[type=literal_error, input_value='No', input_type=str]

@field_validator('answer', mode='before')
@classmethod
def normalize_answer(cls, v):
    if isinstance(v, str):
        v = v.strip().lower()
    return v if v in ("yes", "no") else "no"   # fail closed

ValidationError: Input should be 'yes' or 'no'
[type=literal_error, input_value='', input_type=str]

Fixez soigneusement les valeurs par défaut et les énumérations :

is_safe: Literal["yes", "no"] = Field(default="no")      # fail closed
generated_sql_query: str = Field(default="")
messages: Annotated[list, add] = Field(default_factory=list)  # not default=[]

PydanticJsonSchemaWarning: Default value (...) is not JSON serializable

comments: str = (
    Field(..., description="..."),   # ← trailing comma makes this a tuple
)

def prompt_query_context(state: AgentSchema) -> AgentSchema:
    database_object = Database(connection_details)
    schema_info = database_object.get_schema_details("public")

Action 2 — sortie structurée en tant que composant programmable

Lorsque les réponses de sécurité sont littérales, les chemins deviennent du code. Le modèle est un composant doté d’un contrat, et non un narrateur du flux de contrôle.

Action 3 — récupérer à nouveau le schéma à chaque exécution

Un schéma obsolète dans les prompts entraîne des requêtes SQL erronées. La mise à jour coûte des tokens ; l’obsolescence provoque des incidents. Préférez la récupération à chaque exécution pour les bases de données en évolution, sauf si une version spécifique est définie explicitement.

Action 4 — coût de l’agent ≠ coût du pipeline

Les agents tournent en boucle. Budgetez en tenant compte des limites de récursivité et d’utiliser des modèles moins coûteux pour le routage :

def pick_llm(model_level: str) -> ChatAnthropic:
    if normalized_level == "basic":
        return ChatAnthropic(model="claude-haiku-4-5", temperature=0)
    elif normalized_level == "advanced":
        return ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
    elif normalized_level == "premium":
        return ChatAnthropic(model="claude-opus-4-6", temperature=0)

Détection des erreurs de codage

Le réducteur et le nœud ajoutent tous deux des éléments

messages: Annotated[list, add] = Field(default_factory=list)

state.messages = state.messages + [response]   # full list: old + new
return state                                    # reducer adds it AGAIN

def sql_node(state: DataAgentSchema):
    response = sql_analyst.invoke({...})
    return {"messages": [AIMessage(content=response["final_answer"])]}

Choisissez un seul écrivain : réducteur ou nœud, mais pas les deux.

Transmission de l’état complet du sous-agent vers le haut

response = sql_analyst.invoke({...})   # returns the full AgentSchema dict
state.messages = state.messages + [response]

Ne transmettez que les champs nécessaires au parent.

Sécurité probabiliste à un seul niveau

Les identifiants de base de données et l’analyse SQL doivent servir de soutien aux jugements du modèle :

POSTGRES_USER: agent_user
POSTGRES_PASSWORD: agent_pass

import sqlparse

def is_read_only(sql: str) -> bool:
    statements = sqlparse.parse(sql)
    if len(statements) != 1:          # blocks stacked queries
        return False
    return statements[0].get_type() == "SELECT"

CREATE ROLE agent_readonly LOGIN PASSWORD '...';
GRANT CONNECT ON DATABASE agent_db TO agent_readonly;
GRANT USAGE ON SCHEMA public TO agent_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_readonly;

Défense en profondeur : porte du modèle + analyseur + utilisateur de base de données aux droits limités.

Résultat et limite maximale

Le graphe est désormais susceptible d’être examiné : les arêtes assurent la sécurité, les schémas fonctionnent de manière fermée, les coûts sont intentionnels et les couches se chevauchent. La limite maximale correspond à des évaluations continues et à des tests de chaos — pas à un autre chapitre de tutoriel.

Pratiques à maintenir

Rédigez des ADR pour les forks d’architecture. Testez les chemins non éligibles. Enregistrez les versions du schéma. Isolez l’IAM des outils. Relisez les réducteurs après chaque changement d’état. Les tutoriels s’arrêtent au code exécutable ; la production commence avec des décisions mûrement réfléchies.

Intuition numérique éprouvée

Supposons qu’une étape cible coûte 10 ms et qu’un projet propose 5 tokens avec un taux d’acceptation moyen de 60 % pour 3 tokens. Le coût effectif par token accepté est inférieur à celui des étapes classiques à un seul token, même après les frais liés au projet, tant que le taux d’acceptation reste élevé. Si ce taux tombe à environ 1 token, le schéma échoue. C’est cette sensibilité qui fait que les tableaux de bord surpassent les anecdotes.

Protocole de régression de qualité

Au préalable de l’activation globale, exécutez les prompts corrigés sur des ensembles de questions factuelles, de codage et de refus. Comparez les taux de token identiques lorsque le mode spéculatif est configuré pour une correspondance exacte des distributions. Étudiez toute dérive systémique. Pour un arrêt précoce, suivez le taux de réussite sur les tâches notées ainsi que la préférence humaine lorsque cela est possible.

Emplacement du matériel

Colocalisez autant que possible le modèle de brouillon et celui cible sur le même nœud. Utiliser des brouillons sur différents hôtes ajoute du jitter réseau qui peut annuler les gains obtenus. Faites attention à la mémoire : deux modèles plus le cache KV peuvent épuiser les ressources d’un système qui pouvait aisément en accueillir un seul.

Planification des interactions

Les serveurs de batch continu doivent tenir compte des expansions spéculatives variables. Des planificateurs inefficaces fragmentent les batches et nuisent à l’utilisation des ressources. Coordonnez-vous avec les responsables du service ; ne modifiez pas les paramètres uniquement dans le code de l’application.

Honnêteté concernant les goulets d’étranglement restants

Même après une amélioration du décodage, les utilisateurs peuvent encore subir des retards liés aux appels vers l’outil, au chargement des données ou au traitement du Markdown côté client. Il est nécessaire de tracer le processus de bout en bout. Une optimisation inappropriée gaspille du temps d’ingénierie.

Résumé

Le décodage séquentiel représente la contrainte structurelle inhérente à l’auto-régression. Le décodage spéculatif et la sortie anticipée permettent de réduire cette contrainte dans des conditions mesurables. Il convient de les intégrer avec la même rigueur que n’importe quelle fonctionnalité en production : métriques, indicateurs, capacités de rollback et responsables clairement définis au sein de l’équipe de la plateforme de diffusion.

Intuition numérique éprouvée

Supposons qu’une étape cible coûte 10 ms et qu’un projet préliminaire propose 5 tokens avec un taux d’acceptation moyen de 60 % pour 3 tokens. Le coût effectif par token accepté est inférieur à celui des étapes classiques utilisant un seul token, même après prise en compte des coûts supplémentaires liés au projet préliminaire, tant que le taux d’acceptation reste élevé. Si ce taux tombe à environ 1 token, le système devient moins efficace. C’est cette sensibilité qui explique pourquoi les tableaux de bord sont supérieurs aux anecdotes.

Protocole de régression de qualité

Au préalable de l’activation globale, exécutez les prompts corrigés sur des ensembles de tests factuels, de codage et de refus. Comparez les taux de tokens identiques lorsque le mode spéculatif est configuré pour une correspondance exacte des distributions. Étudiez toute dérive systémique. Pour un arrêt précoce, suivez le taux de réussite sur les tâches notées ainsi que la préférence humaine lorsque cela est possible.

Emplacement du matériel

Colocalisez autant que possible le modèle de brouillon et celui cible sur le même nœud. Les brouillons hébergés sur des serveurs différents ajoutent du jitter réseau qui peut annuler les gains obtenus. Faites attention à la mémoire : deux modèles plus le cache KV peuvent épuiser les ressources d’un serveur qui pouvait aisément en accueillir un seul.

Planification des interactions

Les serveurs de batch continu doivent tenir compte des expansions spéculatives variables. Des planificateurs inefficaces fragmentent les lots et nuisent à l’utilisation des ressources. Coordonnez-vous avec les responsables du service ; ne modifiez pas les paramètres uniquement dans le code de l’application.

Honnêteté concernant les goulets d’étranglement restants

Même après une amélioration du décodage, les utilisateurs peuvent encore subir des retards liés aux appels vers l’outil, au récupération des données ou au traitement du Markdown côté client. Il est nécessaire de tracer le processus de bout en bout. Une optimisation inappropriée gaspille du temps d’ingénierie.

Résumé

Intuition numérique éprouvée

Supposons qu’une étape cible coûte 10 ms et qu’un projet préliminaire propose 5 tokens avec un taux d’acceptation moyen de 60 % pour 3 tokens. Le coût effectif par token accepté est inférieur à celui des étapes classiques utilisant un seul token, même après prise en compte des coûts supplémentaires liés au projet préliminaire, tant que le taux d’acceptation reste élevé. Si ce taux tombe à environ 1 token, le système perd en efficacité. C’est cette sensibilité qui explique pourquoi les tableaux de bord sont supérieurs aux anecdotes.

Protocole de régression de qualité

Emplacement du matériel

Colocalisez autant que possible le modèle de brouillon et celui cible sur le même nœud. Utiliser des brouillons sur des hôtes différents ajoute du jitter réseau qui peut annuler les gains obtenus. Faites attention à la mémoire : deux modèles plus le cache KV peuvent épuiser les ressources d’un système qui pouvait aisément en accueillir un seul.

Planification des interactions

Les serveurs de lotage continu doivent tenir compte des expansions spéculatives variables. Des planificateurs inefficaces fragmentent les lots et nuisent à l’utilisation des ressources. Coordonnez-vous avec les responsables du service ; ne modifiez pas les paramètres uniquement dans le code de l’application.

Honnêteté concernant les goulets d’étranglement restants

Même après une amélioration du décodage, les utilisateurs peuvent encore subir des retards liés aux appels vers l’outil, au récupération des données ou au traitement du Markdown côté client. Suivez le parcours de bout en bout. Une optimisation inappropriée gaspille du temps d’ingénierie.

Résumé

Le décodage séquentiel représente la contrainte structurelle inhérente à l’auto-régression. Le décodage spéculatif et la sortie anticipée réduisent cette contrainte dans des conditions mesurables. Intégrez-les avec la même rigueur que n’importe quelle fonctionnalité de production : métriques, indicateurs, capacités de rollback et responsables clairement désignés au sein de l’équipe de la plateforme de diffusion.

Explications pour les reviewers

Expliquez dans la PR pourquoi deux architectures existent : un chemin prouve qu’un mécanisme de protection sauté est impossible en l’absence d’une condition critique. Ce diagramme correspond exactement à ce que les auditeurs recherchent. Associez-le à des tests échoués qui tentent d’appeler le nœud SQL sans la valeur de sécurité requise.

Résumé

Le décodage séquentiel représente la contrainte structurelle inhérente à l’auto-régression. Le décodage spéculatif et la sortie anticipée permettent de réduire cette contrainte dans des conditions mesurables. Il convient de les intégrer avec la même rigueur que n’importe quelle fonctionnalité en production : métriques, indicateurs, capacités de rollback et responsables clairement désignés au sein de l’équipe de la plateforme de service.

Tous les articles Envoyer le brief