Protéger les appels d’outils de l’agent avec Jev et LangGraph avant qu’ils ne soient exécutés.

Les agents capables de dépenser de l’argent ou de supprimer des données ont besoin d’un point de contrôle avant que l’outil ne s’exécute, ainsi que d’un autre une fois que le travail est censé être terminé. Cet exemple relie ces deux points de contrôle à un agent de e-commerce LangGraph, en utilisant un modèle de décision de TypeSafe AI nommé Jev comme mécanisme de gouvernance.

Ce que renvoie Jev

Jev est un modèle de décision de type « System One ». Alors qu’un LLM général comme GPT-4 répond par du texte ouvert, Jev prend en entrée des états structurés ou non structurés et restitue des décisions sous forme textuelle, des scores de confiance et des distributions de probabilité. Il propose trois primitives.

• Choice sélectionne une étiquette parmi un ensemble fixe et la renvoie accompagnée d’une valeur de confiance ainsi que de la totalité de la distribution de probabilité. Il convient aux décisions de contrôle des outils (allow, deny, confirm, review) et à la classification des intentions.
• Score classe une valeur sur une échelle ordinale, accompagnée d’une valeur de confiance. Il est utile pour l’évaluation des risques en finance ou en politique (Bas, Moyen, Élevé, Critique) ainsi que pour les notations de gravité.
• Noul indique une probabilité comprise entre 0,0 et 1,0 que une condition soit vraie. Il sert aux vérifications booléennes, au respect des politiques et à évaluer la confiance quant à l’achèvement d’une tâche.

Configuration du projet

L’agent fonctionne dans un environnement virtuel Python frais, doté de ces dépendances installées.

fastapi==0.115.6
uvicorn==0.32.1
pydantic==2.10.4
langgraph==0.2.61
langchain==0.3.13
langchain-openai==0.3.1
langchain-typesafe==0.1.0
httpx==0.27.2
python-dotenv==1.0.1

La couche de sécurité : jev_utils.py

Celui-ci fonctionne comme un middleware de validation, une barrière de protection entre l’agent et ses outils. Il traite deux problèmes distincts.

Avant l’exécution : guard_tool_call

Les agents autonomes peuvent déclencher des actions sensibles, telles que des paiements, des suppressions ou des remboursements, sans surveillance. Cette fonction se trouve en amont de chaque appel à un outil. Le nom de l’outil, ses arguments ainsi que les politiques de sécurité pertinentes sont introduits dans un classificateur structuré, qui produit une décision (autoriser, confirmer, revoir ou refuser) accompagnée d’un score de risque.

Après l’exécution : review_completion

Les LLM ne parviennent souvent pas à déterminer si un objectif à plusieurs étapes a réellement été atteint ou s’il reste des tâches inachevées. Cette fonction compare l’historique des états récents avec l’objectif initial de l’utilisateur et qualifie la tâche de complétée ou incomplétée, en y associant une valeur de confiance.

import os
from typing import Any, Dict, List
from langchain_typesafe import Choice, Score, TypeSafeClassifier

if not os.getenv("TYPESAFE_API_KEY"):
    raise RuntimeError("TYPESAFE_API_KEY not set")

classifier = TypeSafeClassifier()

def guard_tool_call(tool: str, args: Dict[str, Any], policy: List[str], side_effects: List[str]) -> Dict[str, Any]:
    questions = {
        "decision": Choice(
            instructions="Choose the safest next action for this tool call.",
            criteria={
                "allow": "Execute immediately.",
                "confirm": "Proceed after user confirmation.",
                "review": "Require human review.",
                "deny": "Do not execute.",
            },
        ),
        "risk": Score(
            instructions="Score the financial and policy risk.",
            criteria=["Low", "Moderate", "High", "Critical"],
        ),
    }

    res = classifier.invoke({
        "state": {"tool": tool, "arguments": args, "policy": policy, "side_effects": side_effects},
        "questions": questions,
    })

    return {
        "decision": res.choices["decision"].choice,
        "confidence": res.choices["decision"].confidence,
        "risk_score": res.scores["risk"].score,
    }

def review_completion(state_summary: str, goal: str) -> Dict[str, Any]:
    questions = {
        "complete": Choice(
            instructions="Is the stated goal fully achieved?",
            criteria={
                "complete": "Goal fully achieved.",
                "incomplete": "Goal not achieved.",
            },
        )
    }

    res = classifier.invoke({
        "state": {"state_summary": state_summary, "goal": goal},
        "questions": questions,
    })

    return {
        "status": res.choices["complete"].choice,
        "confidence": res.choices["complete"].confidence,
    }

L’orchestrateur : agent_graph.py

Ce fichier crée le flux de travail cyclique et à état LangGraph, et il assume trois responsabilités.

• Décider de la prochaine action a lieu dans agent_node. Il lit l’historique des messages ainsi que l’objectif actuel, puis utilise l’API de appel d’outils d’OpenAI pour choisir une réponse ou un appel d’outil.
• Exécution contrôlée a lieu dans tools_node. Il intercepte les appels d’outils proposés par le modèle et les vérifie à l’aide de guard_tool_call. Les actions à haut risque, comme les remboursements supérieurs à 500 $, sont bloquées. Les outils approuvés s’exécutent, et leurs résultats sont enregistrés dans tool_outputs afin que la réponse finale contienne un aperçu structuré.

Cycle de vie et routage font partie de build_agent_graph. Ce dernier configure des branches conditionnelles allant du nœud de départ vers l’agent, puis soit vers les outils, soit vers la vérification de complétion, et enfin vers la fin. La boucle alterne continuellement entre raisonnement et utilisation des outils jusqu’à ce que la tâche soit terminée.

from typing import Annotated, Any, Dict, List, Optional
from typing_extensions import TypedDict
from langgraph.graph import START, END, StateGraph
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, BaseMessage, SystemMessage, ToolMessage
from langchain_openai import ChatOpenAI
from jev_utils import guard_tool_call, review_completion

# --- Typed State ---

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    goal: Optional[str]
    tool_outputs: List[Dict[str, Any]]

# --- Tools Definitions & Logic ---

TOOL_POLICIES = {
    "checkout": {
        "policy": ["Payment captured upon order creation."],
        "side_effects": ["Charges customer"],
    },
    "issue_refund": {
        "policy": ["Refunds above $500 require approval."],
        "side_effects": ["Moves funds"],
    },
}

def get_catalog_info(query: str) -> Dict[str, Any]:
    return {"result": f"Catalog info for: {query}"}

def add_to_cart(user_id: str, product_id: str, qty: int) -> Dict[str, Any]:
    return {"status": "ok", "user_id": user_id, "product_id": product_id, "qty": qty}

def checkout(user_id: str, cart_id: str, address: Dict[str, str], **_) -> Dict[str, Any]:
    return {
        "status": "ok",
        "order_id": "ord_12345",
        "user_id": cart_id,
        "items": [{"product_id": "prod_777", "qty": 2}],
        "ship_to": address,
    }

def issue_refund(order_id: str, amount_usd: float, reason: str) -> Dict[str, Any]:
    return {"status": "ok", "refund_id": "ref_999", "order_id": order_id, "amount_usd": amount_usd}

TOOLS = {
    "get_catalog_info": get_catalog_info,
    "add_to_cart": add_to_cart,
    "checkout": checkout,
    "issue_refund": issue_refund,
}

TOOL_SCHEMAS = [
    {
        "name": "get_catalog_info",
        "description": "Get product/catalog information.",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
    {
        "name": "add_to_cart",
        "description": "Add product to cart.",
        "parameters": {
            "type": "object",
            "properties": {"user_id": {"type": "string"}, "product_id": {"type": "string"}, "qty": {"type": "integer"}},
            "required": ["user_id", "product_id", "qty"],
        },
    },
    {
        "name": "checkout",
        "description": "Place an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "user_id": {"type": "string"},
                "cart_id": {"type": "string"},
                "address": {
                    "type": "object",
                    "properties": {"line1": {"type": "string"}, "city": {"type": "string"}, "state": {"type": "string"}, "country": {"type": "string"}, "pin": {"type": "string"}},
                    "required": ["line1", "city", "state", "country", "pin"],
                },
            },
            "required": ["user_id", "cart_id", "address"],
        },
    },
    {
        "name": "issue_refund",
        "description": "Issue order refund.",
        "parameters": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}, "amount_usd": {"type": "number"}, "reason": {"type": "string"}},
            "required": ["order_id", "amount_usd", "reason"],
        },
    },
]

llm = ChatOpenAI(model="gpt-4.1", temperature=0).bind_tools(TOOL_SCHEMAS)

# --- Graph Nodes ---

def agent_node(state: AgentState) -> Dict[str, Any]:
    sys_msg = SystemMessage(
        content=f"You are an e-commerce assistant. Target Goal: {state.get('goal') or 'N/A'}"
    )
    response = llm.invoke([sys_msg] + state["messages"])
    return {"messages": [response]}

def tools_node(state: AgentState) -> Dict[str, Any]:
    last_msg = state["messages"][-1]
    tool_messages = []
    tool_outputs = list(state.get("tool_outputs", []))

    for tc in getattr(last_msg, "tool_calls", []):
        name, args, call_id = tc["name"], tc["args"], tc["id"]

        # Guard check
        if name in TOOL_POLICIES:
            config = TOOL_POLICIES[name]
            guard = guard_tool_call(name, args, config["policy"], config["side_effects"])

            is_blocked = (
                guard["decision"] == "deny"
                or (name == "issue_refund" and args.get("amount_usd", 0) > 500)
            )

            if is_blocked:
                msg = f"Action `{name}` blocked by policy or high-risk evaluation."
                tool_messages.append(ToolMessage(content=msg, tool_call_id=call_id))
                tool_outputs.append({
                    "status": "blocked",
                    "tool": name,
                    "result": {"error": msg}
                })
                continue

        # Execute safe tool
        res = TOOLS[name](**args)
        tool_messages.append(ToolMessage(content=f"Result: {res}", tool_call_id=call_id))

        # Append structured tool execution record
        tool_outputs.append({
            "status": res.get("status", "ok"),
            "tool": name,
            "result": res
        })

    return {"messages": tool_messages, "tool_outputs": tool_outputs}

def completion_check_node(state: AgentState) -> Dict[str, Any]:
    # Pass-through node; does not pollute message history
    return {}

# --- Builder ---

def build_agent_graph():
    builder = StateGraph(AgentState)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", tools_node)
    builder.add_node("completion_check", completion_check_node)

    builder.add_edge(START, "agent")
    builder.add_conditional_edges(
        "agent",
        lambda state: "tools" if getattr(state["messages"][-1], "tool_calls", None) else "completion_check",
    )
    builder.add_edge("tools", "agent")
    builder.add_edge("completion_check", END)

    return builder.compile()

La couche web : main.py

Le dernier fichier expose l’agent via HTTP à l’aide de FastAPI. Ses composants sont :

• Un point de terminaison REST /chat qui accepte l’historique des messages, le message actuel et un objectif facultatif.
• La conversion de l’historique JSON reçu en objets LangChain HumanMessage et AIMessage destinés au graphe compilé.
• Une appel au graphe avec l’état préparé : les messages, l’objectif et une liste tool_outputs vide.
• Une réponse typée Pydantic, ChatResponse, contenant la reply finale de l’agent ainsi qu’un state_snapshot incluant tool_outputs et goal.

from typing import Any, Dict, List, Optional
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from langchain_core.messages import AIMessage, HumanMessage
from pydantic import BaseModel
from agent_graph import build_agent_graph

load_dotenv(override=True)

app = FastAPI()
agent = build_agent_graph()

class ChatRequest(BaseModel):
    user_id: str
    message: str
    goal: Optional[str] = None
    history: Optional[List[Dict[str, str]]] = None

class StateSnapshot(BaseModel):
    tool_outputs: List[Dict[str, Any]]
    goal: Optional[str] = None

class ChatResponse(BaseModel):
    reply: str
    state_snapshot: StateSnapshot

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    messages = []

    if req.history:
        for m in req.history:
            role, content = m.get("role"), m.get("content", "")
            if role == "user":
                messages.append(HumanMessage(content=content))
            elif role == "assistant":
                messages.append(AIMessage(content=content))

    messages.append(HumanMessage(content=req.message))

    try:
        final_state = agent.invoke({
            "messages": messages,
            "goal": req.goal,
            "tool_outputs": [],
        })
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    # Get the last AI message with non-empty content
    reply_msg = next(
        (m.content for m in reversed(final_state["messages"]) if isinstance(m, AIMessage) and m.content),
        ""
    )

    return ChatResponse(
        reply=str(reply_msg),
        state_snapshot=StateSnapshot(
            tool_outputs=final_state.get("tool_outputs", []),
            goal=final_state.get("goal")
        )
    )

Démarrer le serveur avec uvicorn :

#Run uvicorn server
$uvicorn main:app --reload --port 8000 --host 0.0.0.0

Ensuite, envoyez une demande demandant un remboursement important :

curl -X 'POST' \
  'http://127.0.0.1:8000/chat' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "user_id": "alice_123",
  "message": "Please issue a $750 refund for order ord_999 because the item arrived broken.",
  "goal": "Issue a $750 refund for order ord_999 to user alice_123."
}'

La réponse montre que le service de sécurité accomplit correctement sa tâche. L’agent refuse de traiter le remboursement de 750 $ et indique que les remboursements de grande valeur peuvent nécessiter une vérification ou une approbation supplémentaires.

{
  "reply": "I'm unable to process the $750 refund for order ord_999 due to a policy restriction or high-risk evaluation. For high-value refunds, additional verification or approval may be required.\n\nWould you like me to connect you with customer support or provide further instructions on how to proceed?",
  "state_snapshot": {
    "tool_outputs": [
      {
        "status": "blocked",
        "tool": "issue_refund",
        "result": {
          "error": "Action `issue_refund` blocked by policy or high-risk evaluation."
        }
      }
    ],
    "goal": "Issue a $750 refund for order ord_999 to user alice_123."
  }
}

Pourquoi le remboursement a été bloqué

Dans tools_node de agent_graph.py, une condition explicite détermine si une demande est bloquée :

is_blocked = (
    guard["decision"] == "deny"
    or (name == "issue_refund" and args.get("amount_usd", 0) > 500)
)

Le montant demandé de 750 $ dépasse le plafond de 500 $, ce qui fait que is_blocked est immédiatement mis à true. Deux facteurs expliquent ce résultat.

• Configuration de la politique : TOOL_POLICIES stipule que issue_refund nécessite une approbation pour les montants supérieurs à 500 $.
• Évaluation de la protection : avant que issue_refund ne s’exécute, guard_tool_call dans jev_utils.py évalue le risque. Un virement important a généralement pour résultat un niveau de risque Élevé ou Critique, ou tout autre résultat que autoriser, ce qui entraîne le blocage de l’opération.

La leçon tirée de cette conception

Jev agit comme couche de sécurité et non comme générateur. Le modèle OpenAI (gpt-4.1) s’occupe du raisonnement, des conversations ordinaires ainsi que du choix des outils à utiliser. Jev fonctionne comme un régulateur indépendant qui intercepte ces demandes avant qu’elles ne soient exécutées et les teste en fonction des politiques, des règles de sécurité et des seuils financiers.