Защита вызовов инструментов агента с помощью Jev и LangGraph до их выполнения

Агентам, которые могут тратить деньги или удалять данные, необходимы точки контроля: одна перед запуском инструмента и другая после завершения работы. В этом примере обе точки контроля интегрированы в агента для электронной коммерции LangGraph, причем в качестве регулятора используется модель принятия решений от TypeSafe AI под названием Jev.

Что возвращает Jev

Jev — это модель принятия решений типа «System One». В то время как обычные большие языковые модели, такие как 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

Автономные агенты могут запускать чувствительные действия, такие как оплаты, удаление данных или возврат средств, без наблюдения кого-либо. Эта функция располагается у входа перед каждым вызовом инструмента. Название инструмента, его аргументы и соответствующие правила безопасности поступают в структурированный классификатор, который формирует решение (разрешить, подтвердить, пересмотреть или отклонить) вместе с оценкой риска.

После выполнения: 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 действует как независимый регулятор, перехватывающий эти запросы до их выполнения и проверяющий их согласно правилам политики, правилам безопасности и пороговым значениям, связанным с финансами.