Ваши менеджеры тратят по 3-4 часа в день на то, чтобы собрать информацию по конкурентам, найти прецеденты для юриста или подготовить дайджест упоминаний бренда. Половина запросов теряется, ответы приходят с опозданием. А если попробовать нанять человека на эту работу - это ещё одна зарплата и никакой гарантии, что не ошибётся.
Выход - ИИ-агент. Не просто чат-бот, который отвечает на вопросы, а система, которая сама планирует шаги, ищет данные, принимает решения и выполняет действия. Разберём на примере LangGraph - инструмента, который позволяет собрать такого агента без программиста за пару часов.
Что такое агент простыми словами
Обычный чат-бот: вы спросили - он ответил. Агент: вы дали задачу - он сам решает, что делать, какие инструменты вызвать, в каком порядке. Например, агент для отдела продаж: получил запрос «подготовь коммерческое предложение для Иванова» - нашёл в CRM данные клиента, проверил остатки на складе, сформировал документ и отправил на согласование менеджеру.
LangGraph - это каркас для таких агентов. Он работает как блок-схема: есть узлы (шаги) и рёбра (переходы между ними). Вы описываете логику: «если клиент написал жалобу - перевести на менеджера, иначе - ответить автоматически». Всё наглядно и предсказуемо.
Почему граф, а не цепочка
Представьте конвейер на заводе: деталь проходит станцию А, потом Б, потом В строго по очереди. Это цепочка. Она хороша, когда шаги фиксированы. Но если нужно принять решение (например, «если брак - вернуть на доработку»), цепочка ломается.
Граф состояний (StateGraph) в LangGraph решает это. Каждый узел (шаг) может вести к разным следующим шагам в зависимости от ситуации. Это как блок-схема: «если клиент написал жалобу - перевести на менеджера, иначе - ответить автоматически».
Разберём на примере стройфирмы. У вас есть менеджер, который принимает заявки с сайта. Он должен классифицировать: ремонт квартиры, коттедж или коммерческий объект. Если ремонт - отправить прорабу, если коттедж - архитектору, если коммерческий - руководителю отдела. В цепочке это сделать сложно. В графе - просто: один узел «классификация», три условных ребра «куда идти дальше».
Преимущества перед старым способом (AgentExecutor):
- Вы видите, что агент делает на каждом шаге - полная прозрачность.
- Можно поставить выполнение на паузу в любой момент - например, чтобы менеджер одобрил действие.
- Параллельная работа нескольких веток - одновременно ищете в трёх источниках, а не последовательно.
- Чёткий контроль: какой агент имеет доступ к каким инструментам (инструмент - это функция, которую агент вызывает: поиск, отправка письма, запись в базу).
- Надёжное сохранение состояния: агент не «забывает» контекст при перезапуске сервиса.
Основные элементы LangGraph
Прежде чем строить агента, разберём четыре ключевых понятия. Не пугайтесь - всё просто.
- Узлы (nodes) - отдельные шаги: «спросить модель», «выполнить поиск», «записать результат».
- Рёбра (edges) - переходы между шагами.
- Условные рёбра (conditional edges) - переходы, которые зависят от результата предыдущего шага. Если поиск дал ответ - перейти к написанию отчёта. Если не дал - повторить запрос.
- Состояние (state) - что агент уже сделал и знает. Каждый узел получает состояние и возвращает обновление к нему. Состояние - это память агента внутри одного выполнения задачи.
Ниже - минимальный рабочий агент. Он получает вопрос, решает, нужен ли поиск в интернете, и либо отвечает сразу, либо сначала ищет информацию. Это базовый паттерн, на котором строится большинство агентных систем.
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator
# Схема состояния -- что агент помнит между шагами
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
llm = ChatOpenAI(model="gpt-4o-mini")
tools = [web_search, get_weather]
llm_with_tools = llm.bind_tools(tools)
# Узел-агент: спрашивает модель и получает решение
def agent_node(state: AgentState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
tool_node = ToolNode(tools)
# Условие: продолжать поиск или завершить работу
def should_continue(state: AgentState):
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")
app = graph.compile()
result = app.invoke({"messages": [HumanMessage("Какая погода в Москве?")]})
Управление состоянием и сохранение прогресса
Состояние - это то, что агент накапливает по ходу работы: история сообщений, результаты поиска, промежуточные выводы. Каждое поле состояния знает, как именно обновляться. Список сообщений пополняется. Счётчик суммируется. Финальный текст просто заменяется.
Для бизнеса это значит: правильно описанное состояние гарантирует, что агент не потеряет контекст задачи на полпути. Если агент исследует рынок конкурентов и нашёл 10 источников - они останутся в состоянии до финального шага. Без явного управления состоянием агент «забывает» данные между шагами - так же, как сотрудник без блокнота теряет детали задания по дороге от одного кабинета к другому.
Следующий блок показывает, как описать состояние для агента-исследователя. Каждое поле - с явным правилом накопления.
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator
class ResearchState(TypedDict):
messages: Annotated[list, add_messages] # история диалога
search_results: Annotated[list, operator.add] # накапливаем источники
final_report: str # итоговый отчёт
iteration_count: Annotated[int, lambda a, b: a + b] # счётчик шагов
Контрольная точка (checkpointer) - механизм, который сохраняет состояние агента после каждого шага. Если сервер перезапустится или пользователь вернётся через час, агент продолжит с того места, где остановился. Для компании с тысячами обращений это означает: ни одна задача не потеряется.
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(checkpointer=checkpointer)
# thread_id -- идентификатор сессии. Один клиент = один thread_id
config = {"configurable": {"thread_id": "session-42"}}
result = app.invoke(input_data, config=config)
Для боевых систем используют PostgreSQL вместо SQLite. SQLite удобен при разработке и тестировании.
Практика: агент-исследователь
Рассмотрим агента, который собирает информацию по заданной теме: ищет в интернете, при необходимости уточняет запросы, в конце пишет структурированный отчёт.
Разберём на примере контент-агентства. Редактору нужно собрать дайджест по конкурентам за неделю. Раньше он тратил 2 часа на ручной поиск. Теперь он пишет агенту: «Собери упоминания бренда X в новостях, соцсетях и блогах за последнюю неделю». Агент за 30 секунд находит 15 источников, фильтрует нерелевантное и выдаёт готовую сводку с ссылками. Редактор только проверяет и отправляет клиенту.
from tavily import TavilyClient
from langchain.tools import tool
@tool
def search_web(query: str) -> str:
client = TavilyClient(api_key="...")
results = client.search(query, max_results=3)
return "\n".join([r["content"] for r in results["results"]])
class ResearchState(TypedDict):
topic: str
messages: Annotated[list, add_messages]
sources: Annotated[list[str], operator.add]
report: str
iterations: int
def researcher(state: ResearchState):
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response], "iterations": 1}
def should_continue_research(state: ResearchState):
if state["iterations"] >= 5:
return "finalize"
last = state["messages"][-1]
return "tools" if last.tool_calls else "finalize"
def finalize(state: ResearchState):
report_prompt = f"На основе материалов напиши отчёт по теме: {state['topic']}"
report = llm.invoke(report_prompt + str(state["sources"]))
return {"report": report.content}
Типичный прогон: 4-8 поисковых запросов, 15-45 секунд, $0.05-0.15 при использовании GPT-4o-mini.
Человек в процессе: одобрение действий агента
Для бизнеса критично контролировать автоматические действия. Агент не должен самостоятельно удалять данные, отправлять контракты или списывать деньги. LangGraph поддерживает точки паузы: агент останавливается, показывает человеку то, что собирается сделать, и ждёт одобрения.
Разберём на примере отдела продаж. Агент подготовил коммерческое предложение. Прежде чем отправить его клиенту, система показывает менеджеру итоговый текст и ждёт подтверждения. Только после нажатия «Одобрить» письмо уходит. Это позволяет автоматизировать 80% работы, оставив человеку финальный контроль.
from langgraph.types import interrupt
def review_results(state: ResearchState):
# Агент останавливается и ждёт решения человека
human_review = interrupt({
"question": "Результаты поиска выглядят релевантными?",
"sources": state["sources"][-3:]
})
if human_review["approved"]:
return state
else:
return {"sources": [], "messages": [HumanMessage(human_review["feedback"])]}
# Граф с точкой паузы перед шагом review_results
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["review_results"]
)
После того как человек нажал «Одобрить», выполнение возобновляется одной командой. Состояние не потеряно - агент продолжает с того места, где остановился.
app.invoke(Command(resume={"approved": True}), config=config)
Параллельная работа нескольких веток
Когда задачи независимы, нет смысла выполнять их по очереди. LangGraph умеет запускать несколько узлов одновременно. Три параллельных поиска заканчиваются за 2 секунды вместо 6 последовательных.
Разберём на примере отдела маркетинга. Агент одновременно ищет упоминания бренда в новостях, социальных сетях и отраслевых блогах. Затем собирает всё в единую сводку. Вместо трёх инструментов с тремя отдельными запусками - один агент, параллельная работа, единый результат за треть времени.
# Запускаем поиск по трём источникам параллельно
graph.add_conditional_edges(
"router",
lambda state: ["search_arxiv", "search_web", "search_github"],
["search_arxiv", "search_web", "search_github"]
)
# Собираем результаты в единый узел-агрегатор
graph.add_edge("search_arxiv", "aggregate")
graph.add_edge("search_web", "aggregate")
graph.add_edge("search_github", "aggregate")
При параллельной работе важно, чтобы каждое поле состояния умело корректно принимать несколько результатов одновременно - иначе данные перезапишут друг друга.
Наблюдение за работой агентов: LangSmith
Для бизнеса важно понимать, почему агент принял то или иное решение. Без этого любой сбой превращается в «агент дал неправильный ответ» - и непонятно, что именно пошло не так. LangSmith - сервис от создателей LangGraph, который записывает каждый шаг выполнения: какой узел сработал, что получил на входе, что вернул, сколько времени занял, сколько стоил. Это как логи для обычного программного обеспечения, только читаемые и визуальные.
Подключается через переменные среды:
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="ls-..."
export LANGCHAIN_PROJECT="my-agent"
После этого каждый запуск автоматически попадает в панель LangSmith. Без такого наблюдения отладка агента в боевых условиях превращается в угадывание.
LangSmith также позволяет сохранять реальные запросы клиентов как тестовые примеры и проверять, что после доработки агент продолжает давать правильные ответы. Это важно при любых изменениях: обновили инструкцию агента - запустили набор тестов - убедились, что ничего не сломалось.
Типичные ошибки
Большинство проблем с LangGraph-агентами в боевых условиях приходят не из-за неправильного кода, а из-за неправильного проектирования. Вот что чаще всего идёт не так.
Забыть про контрольную точку. Без checkpointer каждый новый запрос начинается с нуля. Агент не помнит предыдущий разговор. Для клиентского сервиса это катастрофа: клиент объясняет проблему второй раз.
Не поставить ограничение на число итераций. Агент может зациклиться: ищет, не находит подходящего ответа, снова ищет. Всегда задавайте максимум шагов. Иначе один неудачный запрос съест месячный бюджет на API.
Параллельные ветки без правил слияния. Когда несколько узлов одновременно пишут в одно поле состояния без правил слияния, данные теряются. Используйте operator.add или свою функцию объединения.
Отдавать агенту все инструменты подряд. Агент с доступом к удалению данных и отправке писем опасен. Давайте каждому агенту только то, что нужно для его задачи. Принцип минимальных привилегий работает для агентов так же, как для сотрудников.
Запускать в продакшн без трассировки. Без LangSmith или аналога вы не знаете, что происходит внутри. Это не экономия - это потеря контроля.
FAQ
LangGraph или устаревший AgentExecutor?
Команда LangChain сама рекомендует LangGraph для новых проектов. AgentExecutor оставлен только для обратной совместимости со старым кодом. Если строите что-то новое - выбирайте LangGraph. Разница в том, что AgentExecutor скрывает логику внутри, а LangGraph делает её явной и управляемой.
Как агент помнит разговор между сессиями?
Через checkpointer - внешнее хранилище состояния. При разработке удобен SQLite. В боевых системах - PostgreSQL. Один идентификатор сессии (thread_id) привязан к одному контексту. Технически это означает: вернётся ли клиент через час или через день - агент продолжит с того же места.
Можно ли использовать LangGraph без LangChain?
Да. Узлы графа - обычные Python-функции. Можно подключить любую языковую модель, не устанавливая всю экосистему LangChain. LangGraph - самостоятельная библиотека для управления графом состояний.
Что делать, если инструмент вернул ошибку?
ToolNode автоматически передаёт сообщение об ошибке обратно агенту. Агент видит, что что-то пошло не так, и может попробовать другой способ. Для сложной обработки - добавьте условное ребро, которое проверяет тип последнего сообщения.
Что такое subgraph и зачем он нужен?
Subgraph - отдельный граф, встроенный как узел в большой граф. Это как модуль в программировании: изолирует сложную логику и позволяет переиспользовать её в разных местах. Например, весь процесс поиска и суммаризации можно оформить как subgraph и подключать к разным агентам. Один раз написали - используете везде.
Что дальше
LangGraph даёт полный контроль над агентом: видно каждый шаг, легко поставить паузу для проверки, просто добавить параллельные ветки. За эту гибкость платят более высоким порогом входа по сравнению с CrewAI или AutoGen. Выбор зависит от задачи: если нужна максимальная прозрачность, контроль и надёжное сохранение состояния - LangGraph. Если нужна скорость прототипирования с понятными ролями - CrewAI. Обе технологии не исключают друг друга: можно начать с CrewAI и перейти на LangGraph, когда задача вырастет.
Ваш следующий шаг: если у вас есть задача, которая требует сбора информации из нескольких источников с контролем человека - попробуйте собрать агента-исследователя по шаблону выше. Вам понадобится только аккаунт OpenAI (или другой модели) и 2-3 часа времени менеджера, который умеет копировать код. Бесплатный план LangSmith хватит для тестирования.
AI Компас (t.me/kosmoslab_ai) - канал для предпринимателей в РФ и СНГ, которые применяют AI в своём бизнесе без программиста. Разбираем инструменты и схемы - без курсов и теории.