Schützen Sie Agenten-Tool-Aufrufe mit Jev und LangGraph, bevor sie ausgeführt werden.

Agenten, die Geld ausgeben oder Daten löschen können, benötigen einen Kontrollpunkt vor dem Ausführen eines Tools sowie einen weiteren, sobald die Arbeit angeblich abgeschlossen ist. In diesem Beispiel werden beide Kontrollpunkte in einen LangGraph-E-Commerce-Agent integriert, wobei ein Entscheidungsmodell von TypeSafe AI namens Jev als Steuermechanismus verwendet wird.

Was Jev zurückgibt

Jev ist ein „System One“-Entscheidungsmodell. Während allgemeine LLMs wie GPT-4 in offener Prosa antworten, nimmt Jev strukturierte oder unstrukturierte Eingaben entgegen und gibt typisierte Entscheidungen, Zuverlässigkeitswerte sowie Wahrscheinlichkeitsverteilungen zurück. Es bietet drei Grundfunktionen.

• Choice wählt ein Label aus einer festgelegten Menge aus und gibt es zusammen mit einem Zuverlässigkeitswert sowie der vollständigen Wahrscheinlichkeitsverteilung zurück. Es eignet sich für Entscheidungen zur Tool-Steuerung (allow, deny, confirm, review) sowie zur Intent-Klassifizierung.
• Score ordnet etwas auf einer ordinalen Skala ein, wobei erneut ein Vertrauenswert angegeben wird. Er eignet sich für Risikobewertungen in der Finanzbranche oder bei Politikmaßnahmen (Niedrig, Mittel, Hoch, Kritisch) sowie für Schweregradbewertungen.
• Noul gibt eine Wahrscheinlichkeit zwischen 0.0 und 1.0 an, dass eine Bedingung zutrifft. Er wird für boolesche Überprüfungen, Konformitätsprüfungen sowie zur Bestimmung des Vertrauens darin verwendet, dass eine Aufgabe abgeschlossen ist.

Projekteinrichtung

Der Agent läuft in einer neuen Python-Virtuellen Umgebung mit diesen installierten Abhängigkeiten.

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

Die Sicherheitsschicht: jev_utils.py

Diese Datei fungiert als Validierungs-Middleware, eine Schutzmaßnahme zwischen dem Agenten und seinen Tools. Sie kümmert sich um zwei separate Probleme.

Vor der Ausführung: guard_tool_call

Autonome Agenten können sensible Aktionen wie Zahlungen, Löschungen oder Rückerstattungen ausführen, ohne dass jemand zuschaut. Diese Funktion steht vor jedem Toolaufruf als Kontrollpunkt. Der Toolname, seine Argumente sowie die relevanten Sicherheitsrichtlinien werden in einen strukturierten Klassifikator eingegeben, der eine Entscheidung (allow, confirm, review oder deny) zusammen mit einer Risikobewertung ergibt.

Nach der Ausführung: review_completion

LLMs können oft nicht erkennen, ob ein mehrstufiges Ziel tatsächlich erreicht wurde oder ob noch Arbeit aussteht. Diese Funktion vergleicht den aktuellen Zustandsverlauf mit dem ursprünglichen Ziel des Benutzers und kennzeichnet die Aufgabe als complete oder incomplete, wobei ein Zuverlässigkeitswert angegeben wird.

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

Der Orchestrator: agent_graph.py

Diese Datei erstellt den zustandsbasierten, zyklischen LangGraph-Ablauf und übernimmt drei Aufgaben.

• Die Entscheidung darüber, was als Nächstes getan werden soll findet in agent_node statt. Dort wird die Nachrichtenhistorie sowie das aktuelle Ziel gelesen, anschließend wird mithilfe von OpenAI’s Tool-Calling-API eine Antwort oder ein Toolaufruf ausgewählt.
• Geschützte Ausführung erfolgt in tools_node. Hier werden die vom Modell vorgeschlagenen Toolaufrufe abgefangen und über guard_tool_call überprüft. Hochriskante Aktionen, wie Rückerstattungen über 500 Dollar, werden blockiert. Genehmigte Tools werden ausgeführt, und ihre Ergebnisse werden in tool_outputs protokolliert, sodass die endgültige Antwort eine strukturierte Übersicht enthält.

Lebenszyklus und Routing gehören zu build_agent_graph. Dabei wird eine bedingte Verzweigung vom Startknoten zum Agenten eingerichtet, anschließend entweder zu den Tools oder zur Abschlussprüfung und schließlich zum Ende. Der Loop wechselt weiterhin zwischen Denken und Toolnutzung, bis die Aufgabe erledigt ist.

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()

Die Web-Schicht: main.py

Die letzte Datei stellt den Agenten über HTTP mit FastAPI zur Verfügung. Ihre Bestandteile sind:

• Ein /chat REST-Endpunkt, der die Nachrichtenhistorie, die aktuelle Nachricht sowie ein optionaler Zielwert entgegennimmt.
• Die Umwandlung der eingehenden JSON-Historie in LangChain HumanMessage und AIMessage-Objekte für das kompilierte Graphen.
• Ein Aufruf des Graphen mit dem vorbereiteten Zustand: den Nachrichten, dem Zielwert sowie einer leeren tool_outputs-Liste.
• Eine typisierte Pydantic-Antwort, ChatResponse, die die endgültige reply-Antwort des Agenten sowie einen state_snapshot enthält, der tool_outputs und goal beinhaltet.

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")
        )
    )

Starten Sie den Server mit uvicorn:

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

Dann senden Sie eine Anfrage mit der Forderung nach einer großen Rückerstattung:

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

Die Antwort zeigt, dass der Ansprechpartner seine Aufgabe erfüllt. Der Mitarbeiter lehnt eine Rückerstattung von 750 Dollar ab und weist darauf hin, dass Rückerstattungen in hoher Höhe möglicherweise eine zusätzliche Überprüfung oder Genehmigung erfordern.

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

Warum die Rückerstattung blockiert wurde

In tools_node innerhalb von agent_graph.py entscheidet eine explizite Bedingung darüber, ob ein Aufruf blockiert wird:

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

Die angeforderte Summe von 750 $ liegt über der Obergrenze von 500 $, weshalb is_blocked sofort auf true gesetzt wird. Hinter diesem Ergebnis stehen zwei Faktoren.

• Politikkonfiguration: In TOOL_POLICIES ist festgelegt, dass issue_refund für Beträge über 500 $ eine Genehmigung erfordert.
• Bewertung durch Schutzmechanismen: Bevor issue_refund ausgeführt wird, bewertet guard_tool_call in jev_utils.py das Risiko. Große Geldüberweisungen werden in der Regel als High oder Critical eingestuft – oder als etwas anderes als allow – wodurch der Vorgang als blockiert markiert wird.

Die Lektion aus der Gestaltung

Jev fungiert als Sicherheitsschicht und nicht als Generator. Das OpenAI-Modell (gpt-4.1) übernimmt das Denken, normale Gespräche sowie die Auswahl der zu verwendenden Tools. Jev arbeitet als unabhängiger Kontrollmechanismus, der diese Anfragen vor ihrer Ausführung abfängt und sie anhand von Richtlinien, Sicherheitsregeln sowie Geldschwellenwerten prüft.