Захистіть виклики інструментів агента за допомогою Jev та LangGraph перед їх виконанням.

Агенти, які можуть витрачати гроші чи видаляти дані, потребують контрольної точки перед запуском інструменту та ще однієї після того, як робота нібито завершена. У цьому прикладі обидві контрольні точки підключені до електронного комерційного агента LangGraph, причому як керуючий механізм використовується модель прийняття рішень від TypeSafe AI під назвою Jev.

Що повертає Jev

Jev — це модель прийняття рішень типу „System One“. Тоді як звичайні LLM, такі як GPT-4, відповідають у відкритому форматі, Jev приймає структуровані чи неструктуровані дані та повертає типові рішення, показники впевненості та розподіли ймовірностей. Він пропонує три примітиви.

• Choice обирає одну мітку з фіксованого набору та повертає її разом із значенням впевненості та повним розподілом ймовірностей. Він підходить для прийняття рішень щодо доступу до інструментів (allow, deny, confirm, review) та класифікації намірів.
• Оцінка розміщує певний елемент у порядковій шкалі, знову ж таки з значенням довіри. Вона підходить для оцінки ризиків у фінансах чи політиці (Низький, Середній, Високий, Критичний) та для класифікації ступеня серйозності.
• Ймовірність вказує на ймовірність у межах від 0.0 до 1.0, що певна умова є істинною. Вона використовується для булевих перевірок, контролю дотримання правил та оцінки довіри до завершення завдання.

Налаштування проекту

Агент працює у новому віртуальному середовищі Python із встановленими цими залежностями.

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

Шар безпеки: jev_utils.py

Цей файл виконує функцію проміжного елемента для верифікації, що слугує захисним бар’єром між агентом та його інструментами. Він вирішує дві окремі проблеми.

Перед виконанням: guard_tool_call

Автономні агенти можуть ініціювати конфіденційні дії, такі як оплати, видалення чи повернення коштів, без нагляду. Ця функція розташована на вході перед кожним викликом інструменту. Назва інструменту, його аргументи та відповідні правила безпеки потрапляють у структурований класифікатор, який формує рішення (allow, confirm, review або deny) разом із оцінкою ризику.

Після виконання: review_completion

Штучні інтелекти часто не можуть визначити, чи справді була досягнута багатоетапна мета, чи ще є незавершена робота. Ця функція порівнює нещодавню історію стану з початковою метою користувача та позначає завдання як complete або incomplete, додаючи до цього значення довіри.

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

Оркестратор: agent_graph.py

Цей файл створює циклічний, зберігаючий стани робочий процес LangGraph та виконує три основні функції.

• Вирішення того, що робити далі відбувається у agent_node. Він читає історію повідомлень та поточну мету, а потім використовує API для виклику інструментів від OpenAI, щоб обрати відповідь або виклик інструменту.
• Контрольоване виконання здійснюється у tools_node. Він перехоплює виклики інструментів, запропоновані моделлю, та перевіряє їх за допомогою guard_tool_call. Дії з високим ризиком, такі як повернення грошей у розмірі понад 500 доларів, блокуються. Схвалені інструменти виконуються, а їхні результати записуються у tool_outputs, щоб остаточна відповідь містила структурований огляд.

Життєвий цикл та маршрутизація належать до функції build_agent_graph. Вона створює умовне розгалуження від початкового вузла до агента, потім або до інструментів, або до перевірки завершення, і нарешті до кінця. Цикл постійно чергує між міркуваннями та використанням інструментів, поки завдання не буде виконане.

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

Останній файл забезпечує доступ до агента через HTTP за допомогою FastAPI. Його складові:

• REST-кінцева точка /chat, яка приймає історію повідомлень, поточне повідомлення та необов’язкову мету.
• Перетворення надісланої JSON-історії на об’єкти LangChain HumanMessage та AIMessage для скомпільованої графики.
• Виклик графики з підготовленим станом: повідомленнями, метою та порожнім списком tool_outputs.
• Типізована відповідь Pydantic — ChatResponse — яка містить остаточну reply агента та state_snapshot, що містить tool_outputs та 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")
        )
    )

Запустіть сервер за допомогою uvicorn:

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

Потім надішліть запит із проханням про велику компенсацію:

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

У відповіді показано, що охоронець виконує свої обов’язки. Представник відмовляється обробити повернення суми у 750 доларів та зазначає, що для повернень великої суми може знадобитися додаткова перевірка чи схвалення.

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

Чому було заблоковано повернення грошей

Усередині tools_node у файлі agent_graph.py існує чітка умова, яка визначає, чи буде виклик заблокований:

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

Запитана сума у розмірі 750 доларів перевищує ліміт у 500 доларів, тому значення is_blocked одразу встановлюється як true. Цьому результату передують два фактори.

• Конфігурація політики: у TOOL_POLICIES зазначено, що для сум понад 500 доларів необхідна затвердження для операції issue_refund.
• Оцінка ризиків: перед виконанням issue_refund функція guard_tool_call у файлі jev_utils.py оцінює ризик. Великі перекази коштів зазвичай отримують оцінку High або Critical, або будь-яку іншу, крім allow, і така операція позначається як заблокована.

Урок з проектування

Jev виконує функцію захисного шару, а не генератора. Модель OpenAI (gpt-4.1) відповідає за міркування, звичайні розмови та вибір інструментів для використання. Jev працює як незалежний контролер, який перехоплює ці запити перед їх виконанням та перевіряє їх згідно з правилами політики, правилами безпеки та пороговими значеннями коштів.