Ochrona wywołań narzędzi agenta za pomocą Jev i LangGraph przed ich wykonaniem

Agenty, które mogą wydawać pieniądze lub usuwać dane, potrzebują punktu kontrolnego przed uruchomieniem narzędzia oraz kolejnego po zakończeniu pracy. W tym przykładzie oba te punkty kontrolne są połączone z agentem e-commerce LangGraph, przy czym jako mechanizm sterujący użyto modelu decyzyjnego od TypeSafe AI o nazwie Jev.

Co zwraca Jev

Jev to model decyzyjny typu „System One”. Podczas gdy zwykłe modele językowe takie jak GPT-4 odpowiadają w formie otwartych tekstów, Jev przyjmuje dane w formie strukturyzowanej lub niestrukturyzowanej i zwraca sformułowane decyzje, wartości pewności oraz rozkłady prawdopodobieństwa. Oferuje trzy podstawowe funkcje.

• Choice wybiera jeden etykietę z ustalonej listy i zwraca ją wraz z wartością pewności oraz pełnym rozkładem prawdopodobieństwa. Nadaje się do podejmowania decyzji dotyczących dostępu do narzędzi (allow, deny, confirm, review) oraz do klasyfikacji intencji.
• Score umieszcza wartość na skali porządkowej, ponownie wraz z wartością pewności. Nadaje się do oceny ryzyka w finansach lub polityce (Niskie, Umiarkowane, Wysokie, Krytyczne) oraz do oceny stopnia powagi.
• Noul podaje prawdopodobieństwo w zakresie od 0.0 do 1.0, że dana warunek jest spełniony. Służy do sprawdzania wartości logicznych, przestrzegania zasad oraz określania pewności ukończenia zadania.

Ustawienia projektu

Agent działa w nowym wirtualnym środowisku Python z zainstalowanymi następującymi zależnościami.

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

Warstwa bezpieczeństwa: jev_utils.py

Ten plik pełni rolę middleware do walidacji, stanowiąc zaporę pomiędzy agentem a jego narzędziami. Zajmuje się dwoma odrębnymi problemami.

Przed wykonaniem: guard_tool_call

Autonomiczne agenty mogą uruchamiać delikatne operacje, takie jak płatności, usuwanie danych lub zwroty, bez żadnej nadzoru. Ta funkcja znajduje się przed każdym wywołaniem narzędzia. Nazwa narzędzia, jego argumenty oraz odpowiednie zasady bezpieczeństwa trafiają do uporządkowanego klasyfikatora, który wydaje decyzję (allow, confirm, review lub deny) wraz z oceną ryzyka.

Po wykonaniu: review_completion

LLM często nie potrafi stwierdzić, czy cel wieloetapowy został rzeczywiście osiągnięty, czy też praca jest nadal niewykonana. Ta funkcja porównuje ostatnią historię stanów z pierwotnym celem użytkownika i oznacza zadanie jako complete lub incomplete, podając przy tym wartość pewności.

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

Orkiestrator: agent_graph.py

Ten plik tworzy cykliczny, stanowy workflow LangGraph i pełni trzy funkcje.

• Decydowanie o dalszych krokach odbywa się w agent_node. Plik ten czyta historię wiadomości oraz aktualny cel, a następnie wykorzystuje API do wywoływania narzędzi od OpenAI w celu wyboru odpowiedzi lub wywołania narzędzia.
• Weryfikacja wykonywanych działań odbywa się w tools_node. Funkcja ta przerywa wywołania narzędzi proponowane przez model i sprawdza je za pomocą guard_tool_call. Działania o wysokim ryzyku, takie jak zwroty pieniędzy przekraczające 500 dolarów, są blokowane. Zatwierdzone narzędzia są uruchamiane, a ich wyniki zapisywane są w tool_outputs, dzięki czemu ostateczna odpowiedź zawiera ustrukturyzowany zapis sytuacji.

Życiowy cykl i routowanie należą do funkcji build_agent_graph. Funkcja ta ustala warunkowe rozgałęzianie się od węzła początkowego do agenta, następnie albo do narzędzi, albo do sprawdzenia ukończenia zadania, a na końcu do węzła końcowego. Pętla kontynuuje naprzemienną pracę nad rozumowaniem i używaniem narzędzi aż do zakończenia zadania.

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

main.py

Ostatni plik udostępnia agenta przez HTTP za pomocą FastAPI. Jego składniki to:

• Endpoint REST /chat, który przyjmuje historię wiadomości, bieżącą wiadomość oraz opcjonalny cel.
• Przekształcenie otrzymanej historii w formacie JSON na obiekty LangChain HumanMessage i AIMessage przeznaczone do zastosowania w skompilowanym grafie.
• Wezwanie do grafu z przygotowanym stanem: wiadomościami, celem oraz pustym списkiem tool_outputs.
• Typizowana odpowiedź w formacie Pydantic, ChatResponse, zawierająca ostateczną reply agenta oraz state_snapshot z zawartymi w nim tool_outputs i 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")
        )
    )

Rozpocznij serwer za pomocą uvicorn:

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

Następnie wyślij prośbę o dużą zwrot pieniędzy:

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

Odpowiedź pokazuje, że ochroniarz wykonywa swoje obowiązki. Pracownik odmawia przeprowadzenia zwrotu w wysokości 750 dolarów i wspomina, że zwroty o dużej wartości mogą wymagać dodatkowej weryfikacji lub zatwierdzenia.

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

Dlaczego zwrot został zablokowany

Wewnątrz tools_node w pliku agent_graph.py istnieje wyraźna warunek, który określa, czy dane wezwanie zostanie zablokowane:

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

Proszona kwota 750 dolarów przekracza limit 500 dolarów, dlatego is_blocked jest natychmiast ustawiany na true. Za tym wynikiem kryją się dwa czynniki.

• Konfiguracja polityki: TOOL_POLICIES stanowi, że issue_refund wymaga zatwierdzenia dla kwot przekraczających 500 dolarów.
• Ocena zabezpieczeń: przed wykonaniem issue_refund funkcja guard_tool_call w pliku jev_utils.py ocenia poziom ryzyka. Duży transfer środków zazwyczaj otrzymuje ocenę High lub Critical, albo jakąkolwiek inną niż allow, w wyniku czego proces jest oznaczany jako zablokowany.

Lekcja z projektowania

Jev pełni rolę warstwy bezpieczeństwa, a nie generatora. Model OpenAI (gpt-4.1) zajmuje się rozumowaniem, zwykłą rozmową oraz wyborem narzędzi do wykorzystania. Jev funkcjonuje jako niezależny regulator, który przerywa te żądania przed ich wykonaniem i sprawdza je pod kątem zasad, reguł bezpieczeństwa oraz progów finansowych.