Простой AI-агент — это вызов LLM с промптом. Он ответил, сессия закончилась, всё забыто. Реальные агенты устроены иначе: они помнят, что было сказано раньше, могут вызывать инструменты в несколько шагов и принимать решения на основе промежуточных результатов.
LangGraph — это фреймворк для таких агентов. Не просто обёртка над LLM, а система построения агентов как графа состояний. Я разберу, как это работает, и соберу рабочего агента с нуля.
Статья для разработчиков, которые уже работали с LLM через API и хотят построить агента сложнее, чем «промпт → ответ». Базовый Python нужен. Если вы только начинаете — сначала прочитайте про ReAct-паттерн, это фундамент.
Почему именно LangGraph, а не просто LangChain
LangChain — популярная библиотека для работы с LLM. Цепочки (chains) отлично работают для линейных сценариев: загрузить документ → разбить на чанки → поместить в векторную БД → найти похожие → дать контекст модели.
Но когда агент должен:
- Принять решение на основе ответа инструмента
- Повторить шаг, если результат неудовлетворительный
- Перейти к другой ветке логики в зависимости от ситуации
— линейная цепочка ломается. Нужен граф.
LangGraph строит агентов как ориентированный граф (directed graph): узлы — это действия, рёбра — переходы между ними. Можно делать циклы, условные переходы, разветвления.
Инфо
Граф состояний (state graph) — архитектура, где программа движется между состояниями по определённым правилам. Банкомат: ожидание → ввод PIN → выбор операции → выдача денег → завершение. LangGraph использует ту же идею для AI-агентов.
Установка
pip install langgraph langchain-anthropic python-dotenvСоздаём .env:
ANTHROPIC_API_KEY=sk-ant-...
Базовая концепция: State + Nodes + Edges
Три ключевых понятия:
State — словарь с текущим состоянием агента. Всё, что он помнит между шагами. Например, история сообщений, результат поиска, счётчик попыток.
Nodes — функции-обработчики. Принимают State на вход, возвращают обновлённый State. Каждый узел делает одно действие: вызывает LLM, выполняет инструмент, обрабатывает результат.
Edges — правила переходов между узлами. Обычное ребро — всегда идти туда. Условное ребро — выбрать путь на основе состояния.
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict, List
from langchain_core.messages import HumanMessage, AIMessage
# Определяем State
class AgentState(TypedDict):
messages: List[dict]
tool_calls: int
# Инициализируем модель
model = ChatAnthropic(model="claude-sonnet-4-6")Собираем агента с памятью: пошагово
Построим агента-ассистента, который:
- Принимает вопрос пользователя
- Решает, нужно ли использовать инструменты
- Формирует ответ с учётом контекста
Шаг 1: Определяем инструменты
from langchain_core.tools import tool
import ast
import operator
SAFE_OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
}
def safe_calculate(node):
"""Безопасное вычисление простых арифметических выражений."""
if isinstance(node, ast.Constant):
return node.value
if isinstance(node, ast.BinOp) and type(node.op) in SAFE_OPERATORS:
return SAFE_OPERATORS[type(node.op)](
safe_calculate(node.left),
safe_calculate(node.right)
)
raise ValueError("Unsupported operation")
@tool
def search_web(query: str) -> str:
"""Поиск информации в интернете. Используй когда нужны актуальные данные."""
# В реальном агенте здесь был бы Tavily или другой поиск
return f"Результаты поиска по запросу '{query}': [здесь данные из поиска]"
@tool
def calculate(expression: str) -> str:
"""Вычисляет простое арифметическое выражение (+, -, *, /)."""
try:
tree = ast.parse(expression, mode='eval')
result = safe_calculate(tree.body)
return f"Результат: {result}"
except Exception as e:
return f"Ошибка: {e}"
tools = [search_web, calculate]
model_with_tools = model.bind_tools(tools)Шаг 2: Узлы графа
from langchain_core.messages import ToolMessage
def call_model(state: AgentState) -> AgentState:
"""Основной узел: вызываем LLM с историей сообщений."""
response = model_with_tools.invoke(state["messages"])
return {
"messages": state["messages"] + [response],
"tool_calls": state["tool_calls"]
}
def call_tools(state: AgentState) -> AgentState:
"""Узел выполнения инструментов."""
last_message = state["messages"][-1]
tool_results = []
tool_map = {t.name: t for t in tools}
for tool_call in last_message.tool_calls:
tool_fn = tool_map[tool_call["name"]]
result = tool_fn.invoke(tool_call["args"])
tool_results.append(
ToolMessage(content=str(result), tool_call_id=tool_call["id"])
)
return {
"messages": state["messages"] + tool_results,
"tool_calls": state["tool_calls"] + 1
}Шаг 3: Условные рёбра
def should_use_tools(state: AgentState) -> str:
"""Решаем, нужно ли вызывать инструменты или завершить."""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "call_tools"
if state["tool_calls"] >= 5:
return END
return ENDШаг 4: Собираем граф
workflow = StateGraph(AgentState)
workflow.add_node("call_model", call_model)
workflow.add_node("call_tools", call_tools)
workflow.set_entry_point("call_model")
workflow.add_conditional_edges(
"call_model",
should_use_tools,
{
"call_tools": "call_tools",
END: END
}
)
# После инструментов — снова к модели
workflow.add_edge("call_tools", "call_model")
app = workflow.compile()Шаг 5: Запускаем с памятью
initial_state = {
"messages": [
HumanMessage(content="Сколько будет 1337 * 42? И найди информацию о LangGraph.")
],
"tool_calls": 0
}
result = app.invoke(initial_state)
print(result["messages"][-1].content)Совет
Для отладки используйте app.get_graph().print_ascii() — выводит граф в текстовом виде. Видно все узлы и рёбра. Удобно, чтобы убедиться, что структура именно такая, как задумывали.
Добавляем персистентную память
Пока агент помнит только в рамках одного запуска. Чтобы память сохранялась между сессиями — нужен checkpointer:
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# thread_id привязывает историю к конкретной сессии
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(
{"messages": [HumanMessage(content="Привет, как тебя зовут?")], "tool_calls": 0},
config=config
)
# Второй запрос — агент помнит первый
result2 = app.invoke(
{"messages": [HumanMessage(content="Что я спросил тебя только что?")], "tool_calls": 0},
config=config
)
print(result2["messages"][-1].content)
# Агент ответит: "Вы спросили, как меня зовут"Для продакшна используйте SqliteSaver вместо MemorySaver:
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("agent_memory.db")
app = workflow.compile(checkpointer=memory)Внимание
MemorySaver хранит данные в оперативной памяти — при перезапуске всё теряется. Для реального использования нужен внешний storage: SQLite, PostgreSQL, Redis.
Реальный кейс: агент для планирования задач
Вот сценарий, где LangGraph реально незаменим: агент получает описание задачи, разбивает её на шаги и выполняет каждый по очереди.
class TaskState(TypedDict):
task: str
steps: List[str]
completed_steps: List[str]
current_step: int
def plan_steps(state: TaskState) -> TaskState:
"""Разбиваем задачу на шаги."""
prompt = f"Разбей задачу на 3-5 конкретных шагов: {state['task']}"
response = model.invoke(prompt)
steps = [line.strip() for line in response.content.split('\n') if line.strip()]
return {**state, "steps": steps, "current_step": 0}
def execute_step(state: TaskState) -> TaskState:
"""Выполняем текущий шаг."""
step = state["steps"][state["current_step"]]
prompt = f"Выполни шаг: {step}. Контекст задачи: {state['task']}"
response = model.invoke(prompt)
completed = state["completed_steps"] + [f"✓ {step}: {response.content[:100]}"]
return {**state, "completed_steps": completed, "current_step": state["current_step"] + 1}
def is_done(state: TaskState) -> str:
if state["current_step"] >= len(state["steps"]):
return "finish"
return "execute_step"LangGraph vs просто цикл на Python
Честный вопрос: зачем граф, если можно написать while True и if/else?
Можно. Для простого агента на 2-3 шага — цикл проще. LangGraph даёт ценность когда:
- Параллельные ветки — несколько узлов работают одновременно (встроенная поддержка)
- Наблюдаемость — каждый шаг логируется автоматически, можно смотреть в LangSmith
- Персистентность — встроенная поддержка checkpointing
- Сложные условия — визуально видно структуру агента
Для продакшн-агентов с 5+ узлами и памятью — LangGraph значительно упрощает жизнь. Для скрипта на 50 строк — оверхед.
Следующий шаг
Если хотите пойти дальше — изучите multi-agent системы: несколько специализированных агентов, которые передают задачи друг другу. LangGraph поддерживает это через subgraphs. Это следующий уровень после одиночного агента.
Про паттерны агентов на более высоком уровне — читайте про ReAct, там объясняю базовую архитектуру «думать → действовать → наблюдать», на которой строится любой агент.
