langgraph agents guardrails fastapi llm safety

Proteger las llamadas a herramientas del agente con Jev y LangGraph antes de que se ejecuten.

Añadir verificaciones de seguridad antes de la ejecución y revisiones de finalización después de ella a un agente LangGraph, utilizando decisiones de tipo Jev para bloquear acciones riesgosas como reembolsos grandes.

Editor: REACTAPP.TOPAñadido a este sitio 28 oct 20268 min de lectura1817 palabras

Fuentes: JEV AI: Adding Pre-Execution Safety and Post-Execution Verification to LangGraph Agentic Framework.

Los agentes que pueden gastar dinero o eliminar datos necesitan un punto de control antes de que se ejecute una herramienta y otro una vez que se supone que el trabajo está completado. Este ejemplo conecta ambos en un agente de comercio electrónico de LangGraph, utilizando como regulador un modelo de toma de decisiones de TypeSafe AI llamado Jev.

Qué devuelve Jev

Jev es un modelo de toma de decisiones de tipo “System One”. Mientras que un LLM general como GPT-4 responde en prosa abierta, Jev recibe información en formato estructurado o no estructurado y devuelve decisiones en formato tipado, además de puntuaciones de confianza y distribuciones de probabilidad. Ofrece tres primitivas.

Choice elige una etiqueta de un conjunto fijo y la devuelve junto con un valor de confianza y toda la gama de probabilidades. Es adecuado para decisiones de control de herramientas (allow, deny, confirm, review) y para la clasificación de intenciones.
Puntuación sitúa algo en una escala ordinal, nuevamente con un valor de confianza. Se adapta a la evaluación de riesgos en finanzas o políticas (Bajo, Moderado, Alto, Crítico) y a las calificaciones de gravedad.
Noul proporciona una probabilidad entre 0.0 y 1.0 de que una condición sea verdadera. Funciona para verificaciones booleanas, el cumplimiento de políticas y la confianza de que una tarea está completada.

Configuración del proyecto

El agente funciona en un entorno virtual de Python nuevo con estas dependencias instaladas.

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 capa de seguridad: `jev_utils.py`

Este archivo funciona como middleware de validación, una barrera de protección entre el agente y sus herramientas. Se ocupa de dos problemas distintos.

Antes de la ejecución: `guard_tool_call`

Los agentes autónomos pueden iniciar acciones sensibles, como pagos, eliminaciones o reembolsos, sin que nadie los supervise. Esta función se encuentra en la entrada antes de cada llamada a una herramienta. El nombre de la herramienta, sus argumentos y las políticas de seguridad relevantes se introducen en un clasificador estructurado, el cual genera una decisión (allow, confirm, review o deny) junto con una puntuación de riesgo.

Después de la ejecución: `review_completion`

Los LLM a menudo no pueden determinar si se ha cumplido realmente un objetivo de múltiples pasos o si aún queda trabajo pendiente. Esta función compara el historial reciente de estados con el objetivo original del usuario y etiqueta la tarea como completa o incompleta, junto con un valor de confianza asociado.

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,
    }

El orquestador: `agent_graph.py`

Este archivo crea el flujo de trabajo cíclico y con estado de LangGraph, y tiene tres responsabilidades.

Decidir qué hacer a continuación ocurre en agent_node. Lee el historial de mensajes y la meta actual, y luego utiliza la API de llamadas a herramientas de OpenAI para elegir una respuesta o una llamada a herramienta.
Ejecución controlada tiene lugar en tools_node. Intercepta las llamadas a herramientas que propone el modelo y las verifica mediante guard_tool_call. Las acciones de alto riesgo, como reembolsos superiores a $500, se bloquean. Las herramientas aprobadas se ejecutan y sus resultados se registran en tool_outputs para que la respuesta final contenga un resumen estructurado.

Ciclo de vida y enrutamiento forman parte de build_agent_graph. Este proceso establece ramificaciones condicionales desde el nodo inicial hacia el agente, luego ya sea hacia las herramientas o hacia la verificación de finalización, y finalmente hacia el final. El bucle sigue alternando entre el razonamiento y el uso de herramientas hasta que se completa la tarea.

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 capa web: `main.py`

El último archivo expone el agente a través de HTTP con FastAPI. Sus componentes son:

Un endpoint REST /chat que acepta el historial de mensajes, el mensaje actual y un objetivo opcional.
La conversión del historial JSON recibido en objetos LangChain HumanMessage y AIMessage para el grafo compilado.
Una llamada al grafo con el estado preparado: los mensajes, el objetivo y una lista vacía tool_outputs.
Una respuesta tipada de Pydantic, ChatResponse, que contiene la reply final del agente y un state_snapshot que incluye tool_outputs y 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")
        )
    )

Inicie el servidor con uvicorn:

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

Luego envíe una solicitud en la que pida un reembolso completo:

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 respuesta muestra que el guardia está cumpliendo con su tarea. El agente se niega a procesar un reembolso de 750 dólares y menciona que los reembolsos de alto valor pueden requerir verificación o aprobación adicional.

{
  "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."
  }
}

Por qué se bloqueó el reembolso

Dentro de tools_node en agent_graph.py, una condición explícita determina si una llamada se bloquea:

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

La cantidad solicitada de $750 supera el límite de $500, por lo que is_blocked se establece en true de inmediato. Hay dos factores que explican este resultado.

Configuración de políticas: TOOL_POLICIES establece que issue_refund requiere aprobación para cantidades superiores a $500.
Evaluación de protecciones: antes de que se ejecute issue_refund, guard_tool_call en jev_utils.py evalúa el riesgo. Una transferencia de fondos importante suele clasificarse como Alto o Crítico, o como cualquier cosa que no sea permitir, y la operación se marca como bloqueada.

La lección de diseño

Jev funciona como capa de seguridad y no como generador. El modelo de OpenAI (gpt-4.1) se encarga del razonamiento, de las conversaciones habituales y de elegir qué herramientas utilizar. Jev actúa como un regulador independiente que intercepta esas solicitudes antes de que se ejecuten y las verifica en función de las políticas, las reglas de seguridad y los umbrales monetarios.

Leer más

MAF 1.0, LangGraph o CrewAI? Elegir un framework de agentes en 2026 — Compare el Microsoft Agent Framework 1.0, LangGraph y CrewAI en cuanto a estado, memoria, depuración, protocolos, rutas en Azure y costos, con una breve guía de decisión y un patrón de implementación.
Tu primer agente LangGraph: un asistente de fútbol con herramientas y enrutamiento — Tutorial paso a paso de LangGraph: estado compartido, herramientas, un nodo de planificador, aristas condicionales y un resumen en markdown, desarrollado en torno a un asistente ficticio de estadísticas de fútbol.

← Todos los artículos Enviar el brief

Qué devuelve Jev

Configuración del proyecto

La capa de seguridad: jev_utils.py

Antes de la ejecución: guard_tool_call

Después de la ejecución: review_completion

El orquestador: agent_graph.py

La capa web: main.py