Проектирование транзакционных агентных AI с LangGraph

Обзор

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

Безопасность и аудит

С помощью этого подхода мы показываем, как можно проектировать агентные системы с безопасностью, аудитом и контролируемостью в их основе, переходя от реактивных чат-агентов к структурированным, осознающим управление рабочим процессам AI, которые надежно работают в Google Colab с использованием моделей OpenAI.

Настройка окружения

Мы настраиваем окружение выполнения следующим образом:

!pip -q install -U langgraph langchain-openai
 
import os, json, uuid, copy, math, re, operator
from typing import Any, Dict, List, Optional
from typing_extensions import TypedDict, Annotated
 
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
 
def _set_env_openai():
    if os.environ.get("OPENAI_API_KEY"):
        return
    try:
        from google.colab import userdata
        k = userdata.get("OPENAI_API_KEY")
        if k:
            os.environ["OPENAI_API_KEY"] = k
            return
    except Exception:
        pass
    import getpass
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Введите OPENAI_API_KEY: ")
 
_set_env_openai()
 
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
llm = ChatOpenAI(model=MODEL, temperature=0)

Абстракция книги

Далее мы определяем основную абстракцию книги:

SAMPLE_LEDGER = [
    {"txn_id": "T001", "name": "Asha", "email": "ASHA@Example.com", "amount": "1,250.50", "date": "12/01/2025", "note": "Продление членства"},
    {"txn_id": "T002", "name": "Ravi", "email": "ravi@example.com", "amount": "-500", "date": "2025-12-02", "note": "Возврат?"},
    {"txn_id": "T003", "name": "Sara", "email": "sara@example.com", "amount": "700", "date": "02-12-2025", "note": "Штраф отменен"},
]
 
ALLOWED_OPS = {"замена", "удалить", "добавить"}

Моделирование рабочего процесса

Мы моделируем внутреннее состояние агента и определяем каждую ноду в рабочем процессе LangGraph:

class TxnState(TypedDict):
    messages: Annotated[List[AnyMessage], add_messages]
    raw_rows: List[Dict[str, Any]]
    sandbox_rows: List[Dict[str, Any]]
    patch: List[Dict[str, Any]]
    validation: Dict[str, Any]
    approved: Optional[bool]
 
# Определите ваши узлы
...

Запуск агента

Наконец, мы выполняем транзакционного агента:

def run():
    state = {
        "messages": [],
        "raw_rows": SAMPLE_LEDGER,
        "sandbox_rows": [],
        "patch": [],
        "validation": {},
        "approved": None,
    }
 
    out = app.invoke(state)
 
    print(out["messages"][-1].content)
run()

Таким образом, данное руководство демонстрирует, как LangGraph позволяет нам создавать агентов, которые рассуждают над состояниями, устанавливают контрольные точки валидации и сотрудничают с людьми в четко определенных контрольных точках. Мы воспринимаем агента не как оракула, а как координатора транзакций, способного планировать, проверять и отменять свои действия, сохраняя полный аудит.