Как собрать ИИ-агента на LangGraph: state, nodes, tools, memory и human approval

Что получится в результате

Соберем ИИ-агента на LangGraph как управляемый граф состояний. Агент будет принимать запрос, хранить state, вызывать LLM, выбирать tools, выполнять RAG-поиск, требовать approval для рискованных действий, сохранять checkpoints, писать логи и проходить тестовый сценарий.

В первой версии агент не должен быть бесконечным циклом "модель решила сама". Нужно явно описать state schema, nodes, edges, условия переходов, лимиты шагов, tool policy, обработку ошибок, human review и критерии завершения.

В результате будет рабочий MVP:

1. проект лежит в `langgraph_agent_project`;
2. настройки лежат в `.env`;
3. состояние агента описано в `AgentState`;
4. узлы графа описаны в `graph_nodes`;
5. переходы описаны в `graph_edges`;
6. tools описаны в `tool_registry`;
7. права tools лежат в `tool_policy`;
8. память сессии работает через checkpointer;
9. долговременные факты описаны в `memory_store`;
10. RAG-контекст приходит из `retrieval_node`;
11. approval идет через `human_approval_node`;
12. ошибки идут через `error_handler_node`;
13. запуски пишутся в `agent_runs`;
14. вызовы tools пишутся в `tool_call_log`;
15. checkpoints пишутся в `checkpoint_store`;
16. тесты лежат в `langgraph_test_cases`.

Финальная проверка: агент получает вопрос, ищет в базе знаний, выбирает безопасный tool, останавливается на approval для рискованного действия и возвращает ответ с понятным trace.

Что понадобится

Подготовьте:

1. Python 3.11 или 3.12;
2. виртуальное окружение;
3. LangGraph;
4. LangChain OpenAI или другой provider;
5. LLM API key;
6. embedding/vector store, если нужен RAG;
7. список tools;
8. список запрещенных действий;
9. тестовые вопросы;
10. отдельный staging API key;
11. LangSmith или другой tracing, если нужен production-debug;
12. базу данных или checkpointer для persistence.

Для первого запуска можно использовать in-memory checkpointer и простые tools, но структуру проекта сразу делайте такой, чтобы потом заменить память и store на production-хранилища.

Шаг 1. Опишите сценарий агента

Для MVP возьмем агента поддержки по базе знаний.

Он должен уметь:

1. принять вопрос пользователя;
2. классифицировать запрос;
3. найти контекст в базе знаний;
4. ответить по источникам;
5. вызвать безопасный read-only tool;
6. остановиться, если нужен write-tool;
7. передать случай человеку;
8. сохранить summary сессии;
9. записать trace.

Не включайте в первый сценарий:

1. платежи;
2. удаление данных;
3. массовые рассылки;
4. изменение CRM-сделок;
5. произвольный web browsing;
6. shell-команды;
7. долгие автономные циклы.

Проверка: сценарий можно нарисовать как граф из 6-10 узлов.

Шаг 2. Создайте проект

Команды:

mkdir langgraph_agent_project
cd langgraph_agent_project
python -m venv .venv
source .venv/bin/activate
pip install -U langgraph langchain-openai langchain-core python-dotenv pydantic

Для Windows:

python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -U langgraph langchain-openai langchain-core python-dotenv pydantic

Структура:

langgraph_agent_project
  app
    graph.py
    state.py
    nodes.py
    tools.py
    policy.py
    memory.py
    prompts.py
    tests.py
  .env
  requirements.txt

Проверка: `python -c "import langgraph"` завершается без ошибки.

Шаг 3. Создайте `.env`

Файл `.env`:

OPENAI_API_KEY=replace_me
AGENT_MODEL=gpt-4.1-mini
AGENT_MAX_STEPS=8
AGENT_ENABLE_WRITE_TOOLS=false
AGENT_ENV=staging

Правила:

1. `.env` не попадает в Git;
2. API key не передается в prompt;
3. модель фиксируется в настройках;
4. write-tools выключены по умолчанию;
5. staging и production имеют разные ключи.

Проверка: `OPENAI_API_KEY` не печатается в логах.

Шаг 4. Опишите таблицу запусков

Даже если сначала работаете локально, сразу заведите модель данных.

`agent_runs`:

id
run_id
thread_id
user_id
status
input_text
final_answer
error_message
started_at
finished_at

Статусы:

1. `created`;
2. `running`;
3. `waiting_approval`;
4. `completed`;
5. `failed`;
6. `cancelled`;
7. `timeout`.

Проверка: каждый запуск агента имеет `run_id` и `thread_id`.

Шаг 5. Создайте state schema

Файл `app/state.py`:

from typing import Annotated, Literal, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user_id: str
    thread_id: str
    intent: str | None
    retrieved_context: list[dict]
    tool_results: list[dict]
    next_action: str | None
    needs_human: bool
    error: str | None
    step_count: int

Что хранить в state:

1. сообщения;
2. user id;
3. thread id;
4. intent;
5. RAG context;
6. tool results;
7. флаг human review;
8. ошибку;
9. счетчик шагов.

Что не хранить:

1. API keys;
2. пароли;
3. tokens;
4. большие документы целиком;
5. платежные данные.

Проверка: state содержит только данные, нужные для маршрутизации графа.

Шаг 6. Опишите реестр узлов

Создайте `graph_nodes`.

id
node_key
node_type
description
can_call_llm
can_call_tools
requires_approval

Узлы MVP:

1. `input_guard_node`;
2. `classify_intent_node`;
3. `retrieval_node`;
4. `llm_answer_node`;
5. `tool_router_node`;
6. `tool_executor_node`;
7. `human_approval_node`;
8. `memory_update_node`;
9. `finalize_node`;
10. `error_handler_node`.

Проверка: у каждого узла есть одна понятная ответственность.

Шаг 7. Создайте guard node

Файл `app/nodes.py`:

def input_guard_node(state: AgentState) -> dict:
    last_message = state["messages"][-1].content
    blocked_phrases = ["покажи api key", "игнорируй инструкции", "удали все"]

    if any(phrase in last_message.lower() for phrase in blocked_phrases):
        return {
            "needs_human": True,
            "next_action": "human_review",
            "error": "guardrail_triggered",
        }

    return {
        "step_count": state.get("step_count", 0) + 1,
    }

Проверяйте:

1. prompt injection;
2. просьбы раскрыть system prompt;
3. просьбы показать ключи;
4. опасные write-actions;
5. персональные данные;
6. слишком длинный ввод.

Проверка: опасный запрос не идет сразу в LLM node.

Шаг 8. Сделайте классификацию intent

Узел должен понять маршрут.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)


def classify_intent_node(state: AgentState) -> dict:
    user_text = state["messages"][-1].content
    prompt = f"""
Верни один intent:
kb_question, tool_request, lead_request, complaint, unsafe, other.

Сообщение: {user_text}
"""
    result = llm.invoke(prompt).content.strip()
    return {"intent": result}

Для production лучше вернуть structured output, но в MVP можно начать с короткого списка.

Проверка: вопрос по базе знаний получает `kb_question`, а запрос на удаление данных - `unsafe`.

Шаг 9. Подключите RAG node

`retrieval_node` должен вернуть только релевантные chunks.

def retrieval_node(state: AgentState) -> dict:
    question = state["messages"][-1].content

    chunks = [
        {
            "source": "kb:getting-started",
            "title": "Начало работы",
            "text": "ИИ-агент отвечает по базе знаний и передает сложные случаи оператору.",
            "score": 0.82,
        }
    ]

    return {"retrieved_context": chunks}

В реальном проекте замените заглушку на:

1. vector search;
2. keyword search;
3. metadata filters;
4. access rules;
5. rerank;
6. citations.

Проверка: если chunks не найдены, следующий узел должен уметь сказать `no_answer`.

Шаг 10. Напишите prompt ответа

Файл `app/prompts.py`:

SYSTEM_PROMPT = """
Ты ИИ-агент поддержки.
Отвечай только на основе переданного контекста.
Если ответа нет в контексте, скажи, что нужен человек.
Не выдумывай цены, сроки, условия и ссылки.
Не проси пароли, коды из SMS и токены.
Если нужен write-tool, поставь needs_human = true.
"""

Правила:

1. RAG context отделен от memory context;
2. sources передаются явно;
3. отсутствие источника не маскируется;
4. опасные темы уходят на handoff;
5. ответ короткий.

Проверка: модель не отвечает по внутренним знаниям, если context пустой.

Шаг 11. Сделайте LLM node

from langchain_core.messages import AIMessage, SystemMessage
from app.prompts import SYSTEM_PROMPT


def llm_answer_node(state: AgentState) -> dict:
    context = "\n\n".join(
        f"[{i + 1}] {chunk['title']}\n{chunk['text']}\nSource: {chunk['source']}"
        for i, chunk in enumerate(state.get("retrieved_context", []))
    )

    messages = [
        SystemMessage(content=SYSTEM_PROMPT),
        SystemMessage(content=f"Контекст:\n{context or 'Контекст не найден'}"),
        *state["messages"],
    ]

    response = llm.invoke(messages)
    return {"messages": [AIMessage(content=response.content)]}

Проверка: ответ добавляется в `messages`, а не заменяет историю.

Шаг 12. Создайте tool registry

Файл `app/tools.py`:

from langchain_core.tools import tool


@tool
def search_order_status(order_id: str) -> str:
    """Read-only проверка статуса заказа."""
    return f"Заказ {order_id}: тестовый статус processing"


@tool
def create_crm_task(summary: str) -> str:
    """Создает задачу менеджеру. Требует approval в MVP."""
    return f"CRM task draft: {summary}"


TOOLS = {
    "search_order_status": search_order_status,
    "create_crm_task": create_crm_task,
}

Создайте `tool_registry`:

id
tool_name
description
is_readonly
requires_approval
timeout_seconds
is_active

Проверка: write-tool явно помечен как `requires_approval`.

Шаг 13. Добавьте tool policy

Файл `app/policy.py`:

TOOL_POLICY = {
    "search_order_status": {
        "enabled": True,
        "requires_approval": False,
        "max_calls_per_run": 3,
    },
    "create_crm_task": {
        "enabled": True,
        "requires_approval": True,
        "max_calls_per_run": 1,
    },
}


def can_call_tool(tool_name: str) -> bool:
    return TOOL_POLICY.get(tool_name, {}).get("enabled", False)


def requires_approval(tool_name: str) -> bool:
    return TOOL_POLICY.get(tool_name, {}).get("requires_approval", True)

Проверяйте:

1. tool включен;
2. не превышен лимит вызовов;
3. write-tool требует approval;
4. аргументы валидны;
5. tool имеет timeout.

Проверка: модель не может вызвать tool, которого нет в registry.

Шаг 14. Сделайте router node для tools

Для MVP можно маршрутизировать по intent.

def tool_router_node(state: AgentState) -> dict:
    if state.get("intent") == "tool_request":
        return {"next_action": "call_tool"}

    if state.get("intent") in ["complaint", "unsafe"]:
        return {"next_action": "human_review", "needs_human": True}

    return {"next_action": "answer"}

Проверка: complaint не идет в tool executor.

Шаг 15. Сделайте tool executor node

from app.policy import can_call_tool, requires_approval
from app.tools import TOOLS


def tool_executor_node(state: AgentState) -> dict:
    tool_name = "search_order_status"

    if not can_call_tool(tool_name):
        return {"error": "tool_disabled", "needs_human": True}

    if requires_approval(tool_name):
        return {"next_action": "human_review", "needs_human": True}

    result = TOOLS[tool_name].invoke({"order_id": "123"})
    return {"tool_results": [{"tool": tool_name, "result": result}]}

В production:

1. tool name берется из structured tool call;
2. аргументы валидируются;
3. timeout обязателен;
4. результат логируется;
5. ошибка не ломает весь graph.

Проверка: tool result попадает в state и виден следующему узлу.

Шаг 16. Создайте human approval node

from langgraph.types import Command, interrupt


def human_approval_node(state: AgentState) -> dict:
    approval = interrupt({
        "type": "approval_request",
        "tool_name": state.get("tool_name"),
        "tool_args": state.get("tool_args"),
        "question": "Подтвердить выполнение write-tool?",
    })

    if approval.get("decision") != "approve":
        return {
            "needs_human": False,
            "next_action": "finalize",
            "messages": state["messages"],
            "errors": state["errors"] + ["Write-tool отклонен человеком"],
        }

    return {
        "needs_human": False,
        "next_action": "tool_executor",
        "messages": state["messages"],
    }

Чтобы продолжить graph после паузы, вызывайте его с тем же `thread_id` и `Command(resume=...)`:

config = {"configurable": {"thread_id": "thread-123"}}

paused = graph.invoke(initial_state, config=config)

resumed = graph.invoke(
    Command(resume={"decision": "approve"}),
    config=config,
)

В реальном проекте этот узел:

1. создает approval task;
2. отправляет оператору summary;
3. сохраняет proposed action;
4. ставит run в `waiting_approval`;
5. ждет resume;
6. продолжает graph после решения.

Проверка: write-action не выполняется до подтверждения человека.

Шаг 17. Опишите graph edges

Создайте `graph_edges`.

id
from_node
to_node
condition_key
description

Переходы:

1. `START -> input_guard_node`;
2. `input_guard_node -> classify_intent_node`;
3. `classify_intent_node -> retrieval_node`;
4. `retrieval_node -> llm_answer_node`;
5. `llm_answer_node -> tool_router_node`;
6. `tool_router_node -> tool_executor_node`;
7. `tool_router_node -> human_approval_node`;
8. `tool_executor_node -> llm_answer_node`;
9. `llm_answer_node -> memory_update_node`;
10. `memory_update_node -> finalize_node`;
11. `error_handler_node -> finalize_node`.

Проверка: в графе есть путь к завершению и нет бесконечного цикла без лимита.

Шаг 18. Соберите StateGraph

Файл `app/graph.py`:

from langgraph.graph import END, START, StateGraph
from app.state import AgentState
from app.nodes import (
    input_guard_node,
    classify_intent_node,
    retrieval_node,
    llm_answer_node,
    tool_router_node,
    tool_executor_node,
    human_approval_node,
)


def route_after_router(state: AgentState) -> str:
    action = state.get("next_action")
    if action == "call_tool":
        return "tool_executor"
    if action == "human_review":
        return "human_approval"
    return "finalize"


builder = StateGraph(AgentState)
builder.add_node("input_guard", input_guard_node)
builder.add_node("classify_intent", classify_intent_node)
builder.add_node("retrieval", retrieval_node)
builder.add_node("llm_answer", llm_answer_node)
builder.add_node("tool_router", tool_router_node)
builder.add_node("tool_executor", tool_executor_node)
builder.add_node("human_approval", human_approval_node)
builder.add_node("finalize", lambda state: state)

builder.add_edge(START, "input_guard")
builder.add_edge("input_guard", "classify_intent")
builder.add_edge("classify_intent", "retrieval")
builder.add_edge("retrieval", "llm_answer")
builder.add_edge("llm_answer", "tool_router")
builder.add_conditional_edges("tool_router", route_after_router)
builder.add_edge("tool_executor", "llm_answer")
builder.add_edge("human_approval", END)
builder.add_edge("finalize", END)

graph = builder.compile()

Проверка: `graph` компилируется без ошибки.

Шаг 19. Запустите первый invoke

Создайте `app/main.py`:

from langchain_core.messages import HumanMessage
from app.graph import graph

result = graph.invoke({
    "messages": [HumanMessage(content="Как работает агент поддержки?")],
    "user_id": "u_1",
    "thread_id": "thread_1",
    "intent": None,
    "retrieved_context": [],
    "tool_results": [],
    "next_action": None,
    "needs_human": False,
    "error": None,
    "step_count": 0,
})

print(result["messages"][-1].content)

Команда:

python -m app.main

Проверка: агент возвращает ответ, а не падает на state schema.

Шаг 20. Добавьте checkpointer

Для short-term memory используйте checkpointer.

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Запуск с thread:

config = {"configurable": {"thread_id": "thread_1"}}
graph.invoke(input_state, config=config)

В production замените in-memory вариант на постоянное хранилище.

Проверка: второй invoke с тем же `thread_id` видит историю checkpoint.

Шаг 21. Добавьте memory store

Для долговременной памяти используйте отдельный store, а не state.

`memory_store`:

id
user_id
memory_key
memory_value
memory_type
expires_at
created_at

Правила:

1. preferences сохраняются после consent;
2. секреты не сохраняются;
3. state не раздувается;
4. retrieval памяти ограничен top 3-5;
5. пользователь может удалить память.

Проверка: память пользователя не смешивается между thread_id.

Шаг 22. Добавьте лимит шагов

Цикл tool executor -> LLM может стать бесконечным.

Добавьте проверку:

def max_steps_guard(state: AgentState) -> dict:
    step_count = state.get("step_count", 0) + 1
    if step_count > 8:
        return {
            "step_count": step_count,
            "needs_human": True,
            "error": "max_steps_exceeded",
            "next_action": "human_review",
        }
    return {"step_count": step_count}

Проверяйте:

1. max steps;
2. max tool calls;
3. max tokens;
4. max cost;
5. timeout.

Проверка: ошибка tool не заставляет агента бесконечно повторять вызов.

Шаг 23. Добавьте обработку ошибок

Создайте `error_handler_node`.

def error_handler_node(state: AgentState) -> dict:
    return {
        "needs_human": True,
        "next_action": "human_review",
        "messages": state["messages"],
    }

Ошибки:

1. LLM timeout;
2. invalid JSON;
3. disabled tool;
4. tool timeout;
5. retrieval unavailable;
6. checkpoint error;
7. rate limit;
8. max steps.

Проверка: пользователь получает fallback, а trace содержит причину.

Шаг 24. Добавьте structured output

Для решений graph лучше использовать строгий формат.

from pydantic import BaseModel


class AgentDecision(BaseModel):
    answer: str
    next_action: str
    needs_human: bool
    confidence_score: float
    used_sources: list[str]

Проверяйте:

1. JSON валидный;
2. `confidence_score` от 0 до 1;
3. `next_action` из списка;
4. sources не пустые для RAG;
5. `needs_human` true для risky cases.

Проверка: свободный текст вместо JSON не управляет графом.

Шаг 25. Добавьте streaming

Streaming нужен для UI, но не должен ломать state.

Используйте streaming только для ответа пользователю:

1. graph state обновляется после завершения шага;
2. partial tokens не пишутся как final answer;
3. tool calls не показываются пользователю сырыми;
4. error закрывает stream fallback-сообщением;
5. trace сохраняет полный результат.

Проверка: при обрыве stream run получает понятный статус.

Шаг 26. Добавьте tracing

Создайте `agent_trace_log`.

id
run_id
node_key
input_hash
output_hash
duration_ms
status
created_at

Логируйте:

1. вход в узел;
2. выход из узла;
3. routing decision;
4. tool call;
5. retrieval;
6. approval wait;
7. errors;
8. final answer.

Можно подключить LangSmith, если нужен визуальный trace.

Проверка: по `run_id` видно, через какие nodes прошел агент.

Шаг 27. Добавьте eval cases

Создайте `langgraph_test_cases`.

id
case_key
input_text
expected_route_json
expected_answer_contains_json
forbidden_tools_json
is_critical

Минимальные тесты:

1. вопрос по базе знаний;
2. отсутствие ответа;
3. read-only tool;
4. write-tool требует approval;
5. prompt injection;
6. max steps;
7. tool timeout;
8. retrieval unavailable;
9. memory preference;
10. complaint handoff.

Проверка: critical tests проходят перед deploy.

Шаг 28. Прогоните smoke test

Сценарий:

1. запустить graph;
2. задать вопрос по базе знаний;
3. проверить `retrieval_node`;
4. проверить `llm_answer_node`;
5. запросить read-only tool;
6. проверить `tool_executor_node`;
7. запросить write-tool;
8. проверить `human_approval_node`;
9. проверить checkpoint;
10. проверить trace.

Проверка: smoke test проходит без ручной правки state.

Шаг 29. Подготовьте API wrapper

Для production заверните graph в API.

Endpoint:

POST /api/agent/run

Вход:

{
  "thread_id": "thread_1",
  "user_id": "u_1",
  "message": "Как подключить агента к CRM?"
}

Ответ:

{
  "run_id": "run_1",
  "status": "completed",
  "answer": "..."
}

Правила:

1. API проверяет auth;
2. thread_id валидируется;
3. graph вызывается с config;
4. errors не раскрывают stack trace;
5. run пишется в базу.

Проверка: API может вызвать graph несколько раз в одном thread.

Шаг 30. Подготовьте production persistence

Для production нужны постоянные хранилища.

Замените:

1. in-memory checkpointer на Postgres или другое хранилище;
2. in-memory store на базу;
3. локальные заглушки tools на backend services;
4. print logs на structured logs;
5. ручные tests на regression suite.

Создайте `checkpoint_store`:

id
thread_id
checkpoint_id
state_json
created_at

Проверка: после перезапуска процесса thread можно продолжить.

Шаг 31. Минимальный результат для запуска

MVP готов, если выполнены условия:

1. проект создан;
2. `.env` настроен;
3. `AgentState` описан;
4. nodes имеют одну ответственность;
5. edges компилируются;
6. graph запускается;
7. RAG node возвращает context;
8. LLM node отвечает по context;
9. tool registry есть;
10. tool policy работает;
11. write-tool требует approval;
12. checkpointer работает;
13. memory store не смешивает пользователей;
14. max steps защищает от циклов;
15. errors уходят в handler;
16. structured output валидируется;
17. trace пишется;
18. eval cases есть;
19. smoke test пройден;
20. API wrapper готов.

Финальная проверка: запустите три сценария: обычный вопрос, read-only tool и write-tool. Первый должен завершиться ответом, второй - tool result, третий - human approval без выполнения опасного действия.

Что нельзя автоматизировать в первой версии

В первой версии не автоматизируйте:

1. shell-команды;
2. произвольный web browsing;
3. платежи;
4. возвраты;
5. изменение CRM без approval;
6. удаление данных;
7. бесконечные циклы tools;
8. запись секретов в state;
9. хранение полного transcript без TTL;
10. автоматическое изменение tool policy;
11. auto-deploy graph без evals;
12. выполнение write-tools из prompt;
13. раскрытие trace пользователю;
14. запуск без max steps;
15. production без persistent checkpoint.

Сначала LangGraph должен дать контроль над маршрутом, состоянием и безопасными tools. Уже потом добавляйте multi-agent, параллельные ветки, сложную память и долгие автономные процессы.

Частые вопросы

LangGraph подходит новичкам?

LangGraph требует базового Python и понимания stateful workflow. Если нужен быстрый no-code прототип, проще начать с n8n или Flowise. Если нужен контроль над состоянием, ветвлениями и тестами, LangGraph подходит лучше.

Чем LangGraph отличается от LangChain?

LangChain дает компоненты для моделей, prompts, tools и интеграций. LangGraph фокусируется на orchestration: state, nodes, edges, циклы, persistence, human-in-the-loop и управляемые agent workflows.

Можно ли использовать LangGraph без RAG?

Да. RAG нужен только если агент должен отвечать по документам или базе знаний. LangGraph можно использовать и для tool workflows, классификации, multi-step обработки заявок и approval-сценариев.

Где хранить память LangGraph-агента?

Краткосрочный контекст храните через checkpointer по `thread_id`. Долговременную память храните отдельно в `memory_store` с consent, TTL, safety checks и удалением.

Когда нужен human approval?

Approval нужен для write-tools, CRM-изменений, платежей, удаления данных, жалоб, персональных данных и любых действий, где ошибка агента может повлиять на пользователя или бизнес.