Ахова выклікаў інструментаў агента праз 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 працуе як незалежны керавальнік, які перэхопляе такія запіты пры ўсуненні іх і пераглядае іх па правілах, нормах безпекі і прагах грошовых выдач.

Спадні матэрыялы

• MAF 1.0, LangGraph чы CrewAI? Выбір фрэймворку агента ў 2026 годзе — Пораўнанне Microsoft Agent Framework 1.0, LangGraph і CrewAI па стану, памяці, налагоджэнні, протаколах, шляхах у Azure і вартасці, з короткім шляхам прыняцтва рашэння і мадэллю впроваджэння.

Ваш першы LangGraph-агент: асістэнт па футболу з інструментамі і маршрутазаваннем — пасляпасовы кансэпт LangGraph: спяльны стан, інструменты, вузел планавання, умовныя рэшткі і стык з Markdown, створаны на адпаведніце з асістэнтом па статыстыцы футболу.