Создание агента планирования в LangGraph: планирование, выполнение, перепланирование, отчетность
Четырехузловой агент планирования LangGraph для финансовых исследований: структурированные планы, реестр инструментов с учетом пространства имён, ограничения на перепланирование и отслеживаемая запись выполнения примера.
Источники: Your AI Agent Isn’t Smart. Here’s How to Build One That Actually Thinks
Большинство агентов — это в действительности скрытые цепочки промптов
Вот неприятный факт о ИИ-агентах: многие из них представляют собой улучшенные цепочки промптов, облечённые в «парадную одежду». Схема знакома: системный промпт, несколько инструментов и, возможно, цикл ReAct, и результат называют агентом. В демо-версии и при благоприятных условиях всё работает. Но как только инструмент возвращает ошибку, данные оказываются необычными или запрос требует более трёх шагов, вся конструкция рушится.
Альтернативой является агент планирования. Он составляет полный план исследования перед использованием любых инструментов, выполняет его шаг за шагом, и когда промежуточные результаты противоречат ожиданиям, он переписывает оставшуюся часть плана. Демо-версия — это система финансовых исследований. Предоставьте ей запрос на простом английском, например: «Сравните оценки NVDA и AAPL и скажите, какая из них сейчас выглядит лучше», и она составит многоэтапный план, выполнит каждый шаг с помощью инструментов обработки данных, сама исправит ошибки и составит отчет в стиле аналитика. Она работает на LangGraph, может использовать облачные модели через Groq или локальные — через Ollama, и занимает примерно четыре файла.
Почему простой ReAct недостаточен
ReAct (Reason-Act-Observe) — это шаблон, с которого начинают работу большинство агентов. Языковая модель рассуждает о своем следующем действии, вызывает инструмент, читает полученный результат, снова рассуждает и вызывает следующий инструмент. Этот подход популярен, прост в использовании и подходит для небольших задач.
Основная слабость заключается в чрезмерной ориентации на ближайшую цель, поскольку агент, работающий по принципу ReAct, планирует только одно действие вперед. Если попросить данные по пяти акциям по трём показателям плюс последние новости, он начнёт с, скажем, коэффициента P/E акции NVDA, и первый полученный результат отвлечёт его внимание. Один из прототипов потратил семь вызовов инструментов на анализ каждого показателя Tesla, прежде чем вспомнить о задаче сравнения нескольких акций; его итоговый отчёт состоял из краткого обзора результатов деятельности Tesla и одного предложения об остальных акциях.
Проблема здесь не в возможностях модели, а в архитектуре. Если при каждом процессе инференции необходимо решать, что исследовать, как это сделать и как интерпретировать полученные результаты, ни одно из этих решений не получает должного внимания. Агент планирования разделяет мышление и действие: сначала планирует, затем выполняет действия, а после проверяет, сохраняется ли план актуальным. Каждый узел обрабатывает одну задачу, и они выполняются одна за другой.
Четыре узла и один цикл
Агент планирования представляет собой ориентированную графу с четырьмя узлами, похожую на диаграмму потоков, где каждый блок обрабатывает одну задачу и передает состояние дальше.
START --> Planner --> Executor <--> Replanner --> Reporter --> END
- Планировщик принимает запрос пользователя и создаёт структурированный, пошаговый план исследования. В каждом шаге указывается инструмент, который необходимо использовать, аргументы и причина их применения.
- Выполняющий элемент берёт следующий шаг, вызывает указанный инструмент и записывает результат в временную записную книжку.
- Перепланировщик анализирует всю собранную до этого информацию после каждой выполнения. Если инструмент дал ошибку или данные показали нечто неожиданное, он переписывает оставшиеся шаги. В противном случае план остаётся без изменений.
- Отчётчик преобразует все записи из временной записной книжки в окончательный аналитический отчёт, содержащий конкретные цифры и рекомендации.
Основой дизайна является цикл Executor-Replanner. Два узла поочередно выполняют действия до завершения плана, и именно эта смена обеспечивает самокорректирующееся поведение, поскольку после каждого действия агент может пересмотреть свои решения.
Состояние: общая память
В LangGraph состояние — это всё, а именно общая память, передаваемая между узлами. Полное определение:
import operator
from typing import Annotated, TypedDict
from pydantic import BaseModel, Field
class StrategyState(TypedDict, total=False):
"""Global state that flows through the LangGraph nodes."""
query: str
plan: list[dict]
scratchpad: Annotated[list[dict], operator.add]
current_step: int
final_report: str
replan_count: int
В этом состоянии query хранит исходный запрос пользователя без изменений, plan — упорядоченный список шагов, scratchpad собирает результаты работы всех инструментов, current_step указывает на соответствующий шаг плана, а replan_count обеспечивает соблюдение лимита MAX_REPLANS.
Важна аннотация Annotated[list[dict], operator.add] в scratchpad. Её функция сведения, operator.add, указывает LangGraph на необходимость добавления новых элементов в список, оставляя существующие на месте. Когда Executor возвращает {"scratchpad": [new_entry]}, LangGraph добавляет его в текущий список, благодаря чему узлы никогда не обрабатывают списки вручную, и не возникает ошибок из-за изменений.
Структурированные планы и решения
Planner никогда не возвращает текст в свободном формате. Его результат — это модель Pydantic, поэтому большая языковая модель должна генерировать планы в формате, который можно разобрать:
AVAILABLE_TOOLS_TEXT = """
- get_metrics(ticker, metric?): Return stock metrics. 'metric' is optional
(P/E, EPS, Revenue, Market Cap, Sector).
- search_news(ticker): Return recent news headlines for a ticker.
- compare_metrics(tickers: list, metric): Compare one metric across
multiple tickers.
"""
class PlanStep(BaseModel):
"""A single executable step inside an analysis plan."""
step_id: int = Field(description="Sequential step number")
tool: str = Field(
description=f"Tool to use. Must be one of:\n{AVAILABLE_TOOLS_TEXT}"
)
args: dict = Field(description="Arguments for the tool call")
purpose: str = Field(description="Why this step is needed")
class Plan(BaseModel):
"""The full plan generated by the planner node."""
goal: str = Field(description="The overall analysis goal")
steps: list[PlanStep] = Field(
description="Ordered list of steps to execute"
)
Поле purpose оказывается более полезным, чем ожидалось. Оно объясняет Replannerу, зачем существует та или иная операция, а не просто что она делает. Вместо простого вызова "call compare_metrics," указание "Сравнивать коэффициенты P/E для определения относительной оценки" позволяет Replannerу гораздо лучше решить, следует ли сохранить, изменить или удалить данную операцию.
У Replannerа есть собственная схема:
class ReplanDecision(BaseModel):
"""Output of the replanner node."""
reasoning: str = Field(
description="Analysis of current progress and findings"
)
should_replan: bool = Field(
description="Whether the plan needs modification"
)
updated_steps: list[PlanStep] = Field(
default_factory=list,
description="Remaining steps if replan is needed. Empty if no changes.",
)
Если сделать should_replan булевым значением, это заставляет ЯИИ принять одно из решений: план меняется или нет? Если оставить в структурированном выводе возможность для маневра у ЯИИ, оно будет её использовать каждый раз.
Система инструментов
Три инструмента для финансовых исследований работают с реалистичными фиктивными данными. В реальных условиях использовались бы настоящие API-запросы, но фиктивные данные имеют значительное преимущество во время разработки: детерминистичная отладка. Когда агент ведёт себя странно, сразу становится ясно, заключается ли проблема в его логике или в API.
Дизайн-паттерн, заслуживающий большего внимания, — это реестр инструментов с учётом пространств имён, который используется вместо того, чтобы предоставлять каждому узлу каждый инструмент:
from langchain_core.tools import BaseTool
from typing import Iterable, Mapping, Any
class ToolRegistry:
"""Namespace-aware container for LangChain tools."""
def __init__(self) -> None:
self._tools_by_toolset: dict[str, dict[str, BaseTool]] = {}
def add_tools(self, toolset: str, tools: Iterable[BaseTool]) -> None:
bucket = self._tools_by_toolset.setdefault(toolset, {})
bucket.update({t.name: t for t in tools})
def get_tools(self, toolset: str) -> tuple[BaseTool, ...]:
return tuple(self._tools_by_toolset.get(toolset, {}).values())
def invoke(
self, toolset: str, tool_name: str, tool_args: Mapping[str, Any]
) -> Any:
t = self._tools_by_toolset.get(toolset, {}).get(tool_name)
if t is None:
raise ValueError(
f"Unknown tool '{tool_name}' in toolset '{toolset}'"
)
return t.invoke(dict(tool_args))
Регистрация происходит в пространствах имён "planner", "executor" и "replanner". Все инструменты видны Plannerу, чтобы он мог планировать с их помощью, однако только Executor может их вызывать. Даже если ошибка в интерфейсе Replannerа каким-то образом приведёт к вызову инструмента, реестр откажет в этом, поскольку у данного пространства имён нет прав на вызов. Это и есть контроль доступа на основе ролей для агентов, причём разделение функций также служит механизмом безопасности, препятствующим нежелательным изменениям состояния во время планирования.
Узел Planner
def planner_node(state: StrategyState) -> dict:
"""Create a step-by-step research plan using structured output."""
planner = model.with_structured_output(Plan)
prompt = PLAN_PROMPT.format(
available_tools=AVAILABLE_TOOLS_TEXT,
ticker_choices=ticker_choices_text(),
metric_choices=metric_choices_text(),
query=state["query"],
)
plan: Plan = planner.invoke(prompt)
steps = [s.model_dump() for s in plan.steps]
return {"plan": steps, "current_step": 0}
model.with_structured_output(Plan) заставляет ЯИИ отвечать в виде объекта JSON, совместимого с Pydantic, а некорректный ответ просто приводит к повторной попытке. Это значительно превосходит обработку свободного текста с помощью регулярных выражений.
Ограничения в промпте плана
PLAN_PROMPT = """\
You are a financial research planner. Given a user's analysis request,
create a step-by-step research plan using the available tools.
Available tools:
{available_tools}
Rules:
- Use only the tools listed above.
- Every plan step must be executable with one of those tools.
- When a tool accepts 'ticker' or 'tickers', use only these exact values:
{ticker_choices}
- When a tool accepts 'metric', use one of these exact values:
{metric_choices}
- There are no other tools available. Final synthesis is handled separately.
Create an efficient plan. Group related lookups. Aim for 4-8 steps.
User request: {query}"""
Выделяются два важных момента. В запросе указаны допустимые значения параметров, поэтому большой языковой модели не нужно догадываться, какие тикеры или показатели существуют. Это, возможно, самый эффективный способ борьбы с галлюцинациями у агентов, выполняющих вызовы инструментов. Без этого модель уверенно планирует действия вроде get_metrics(ticker="BITCOIN", metric="vibes"), а затем удивляется, когда они терпят неудачу.
Второй момент — строка «Других инструментов нет». Когда её нет, большие языковые модели выдумывают действия вроде analyze_sentiment(ticker="NVDA") или generate_report(findings=...) — инструменты, которых на самом деле не существует.
Инструкция «стремиться к 4–8 шагам» сознательно сформулирована мягко. Жесткие ограничения вроде «ровно 5 шагов» приводили к обратному эффекту: простые запросы заполнялись избыточными шагами, а сложные — загружались в уже перегруженные процессы. Диапазон позволяет модели адаптироваться, и при наличии рекомендаций вместо обязательных правил она обычно принимает правильное решение.
Узел Executor
Узел Executor преднамеренно сделан самым простым: он берет следующий шаг, запускает инструмент через реестр и сохраняет результат.
MAX_STEPS = 12 # Safety limit on total steps
def executor_node(state: StrategyState) -> dict:
"""Execute the next pending step from the plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
if current_step >= len(plan):
return {}
if current_step >= MAX_STEPS:
return {"current_step": len(plan)}
step = plan[current_step]
tool_name = step["tool"]
tool_args = step["args"]
try:
result = str(
TOOL_REGISTRY.invoke(
AgentName.EXECUTOR.value, tool_name, tool_args
)
)
status = "Error" if result.startswith("Error:") else "Success"
except Exception as exc:
result = f"Error: {exc}"
status = "Error"
entry = {
"step": current_step + 1,
"tool": tool_name,
"args": tool_args,
"result": result,
"status": status,
}
return {
"scratchpad": [entry],
"current_step": current_step + 1,
}
MAX_STEPS = 12 служит жестким лимитом. Если Replanner будет продолжать добавлять шаги или застрянет, агент не сможет бесконечно тратить API-кредиты. Как только достигнут лимит, Executor устанавливает current_step в значение len(plan), и поток переходит к Reporter с собранными на данный момент данными.
Ошибки определяются по установленному правилу: status = "Error" если результат начинается с "Error:" в противном случае "Success". Каждый из трех инструментов добавляет в начало своих сообщений об ошибках строку "Error:", что позволяет Executor отмечать сбой без необходимости парсинга JSON или обработки исключений. Replanner считывает этот статус при принятии решения о повторной попытке или корректировке плана.
Узел Replanner
Этот узел показывает разницу между агентом планирования и простым оркестратором. После каждой работы инструмента он анализирует все имеющиеся данные и определяет, имеет ли оставшийся план ещё смысл:
MAX_REPLANS = 2 # Prevent infinite replanning
def replanner_node(state: StrategyState) -> dict:
"""Review progress and optionally modify the remaining plan."""
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
replan_count = state.get("replan_count", 0)
scratchpad = state.get("scratchpad", [])
remaining = plan[current_step:]
if len(remaining) = MAX_REPLANS:
return {}
scratchpad_text = "\n".join(
f"Step {e['step']}: {format_tool_call(e['tool'], e['args'])} "
f"-> [{e['status']}] {e['result'][:150]}..."
for e in scratchpad
)
remaining_text = "\n".join(
f"Step {s['step_id']}: {format_tool_call(s['tool'], s['args'])} "
f"- {s['purpose']}"
for s in remaining
)
replanner = model.with_structured_output(ReplanDecision)
prompt = REPLAN_PROMPT.format(
goal=state["query"],
scratchpad=scratchpad_text,
remaining_steps=remaining_text,
)
decision: ReplanDecision = replanner.invoke(prompt)
if decision.should_replan and decision.updated_steps:
new_steps = plan[:current_step] + [
s.model_dump() for s in decision.updated_steps
]
return {"plan": new_steps, "replan_count": replan_count + 1}
return {"replan_count": replan_count + 1}
Посмотрите на e['result'][:150]. Система Replanner обрабатывает результаты каждого инструмента, сокращая их до 150 символов, поскольку ей нужна суть, а не полный JSON-данный пакет, тогда как система Reporter позже получает несокращённые данные. Токены сохраняются там, где детали не имеют значения, и тратятся там, где они важны.
При повторном планировании plan[:current_step] + [s.model_dump() for s in decision.updated_steps] сохраняет завершённые шаги и заменяет только незавершённые, оставляя историю без изменений.
Запрос на повторное планирование
REPLAN_PROMPT = """\
You are a financial research planner reviewing progress on a research task.
Original goal: {goal}
Completed steps and findings so far:
{scratchpad}
Remaining steps in the plan:
{remaining_steps}
Based on the findings so far, should the remaining plan change?
If an expected tool failed or revealed something unexpected, add a step
to investigate.
If a step is now redundant, remove it.
Use only the available executable tools already shown in the plan.
Do not add recommendation, summary, or report-writing steps."""
Последняя строка выполняет ключевую функцию. Если убрать фразу «Не добавлять шаги с рекомендациями, краткими описаниями или составлением отчетов», то инструмент Replanner добавляет шаг синтеза к каждому плану, хотя эту задачу должен выполнять Reporter. Подсказки необходимы для четкого разграничения между узлами плана.
Значение MAX_REPLANS = 2 оказалось оптимальным. В одной из ранних версий для задачи из трех шагов требовалось шесть раундов перепланирования, причем ИИ находил так называемые «улучшения», которые на самом деле ничего не улучшали. Два раунда достаточны для реальной самокоррекции, например, при работе с несрабатывающим инструментом или удалении избыточных шагов, без риска аналитической паралича. В какой-то момент агенту необходимо указать прекратить размышления и завершить работу.
Репортер получает полную, неразрезанную запись и создаёт структурированный отчёт в формате Markdown. Он вызывает метод model.invoke(prompt).content, что возвращает простой текст, а не структурированный результат, позволяя большой языковой модели самостоятельно представить анализ в удобном для неё формате. Самая эффективная инструкция звучит так: «Будьте конкретны, приводите цифры из данных». Без неё отчёты становятся расплывчатыми, в них говорится что-то вроде «У NVDA сильные фундаментальные показатели», при этом не указываются реальные значения коэффициента P/E и доходов.
Связывание компонентов в LangGraph
Здесь встречаются четыре узла:
from langgraph.graph import StateGraph, END
from enum import Enum
class AgentName(Enum):
PLANNER = "planner"
EXECUTOR = "executor"
REPLANNER = "replanner"
REPORT = "report"
def build_graph():
"""Build and compile the LangGraph planning-agent workflow."""
workflow = StateGraph(StrategyState)
workflow.add_node(AgentName.PLANNER.value, planner_node)
workflow.add_node(AgentName.EXECUTOR.value, executor_node)
workflow.add_node(AgentName.REPLANNER.value, replanner_node)
workflow.add_node(AgentName.REPORT.value, report_node)
workflow.set_entry_point(AgentName.PLANNER.value)
workflow.add_edge(AgentName.PLANNER.value, AgentName.EXECUTOR.value)
workflow.add_edge(AgentName.EXECUTOR.value, AgentName.REPLANNER.value)
workflow.add_conditional_edges(
AgentName.REPLANNER.value,
should_continue_execution,
{
AgentName.EXECUTOR.value: AgentName.EXECUTOR.value,
AgentName.REPORT.value: AgentName.REPORT.value,
},
)
workflow.add_edge(AgentName.REPORT.value, END)
return workflow.compile()
def should_continue_execution(state: StrategyState) -> str:
"""Return the next node name after re-planning."""
if state.get("current_step", 0) >= len(state.get("plan", [])):
return AgentName.REPORT.value
return AgentName.EXECUTOR.value
График почти похож на английский текст. Планировщик передаёт задачу Выполняющему, Выполняющий — Перепланировщику; если шаги ещё остались, Перепланировщик направляет задачу обратно к Выполняющему, а когда всё выполнено — к Отчётчику. Отчётчик переходит в состояние END.
Энумерация AgentName кажется незначительной, но она полезна. Простые строковые литералы вроде "planner", разбросанные по коду, способствуют появлению опечаток, а "plannr" может тайно создать несвязанный узел. Благодаря энумерации ошибка проявляется уже при импорте.
Механизм маршрутизации проверяет одно условие: выполнены ли все шаги? Когда current_step >= len(plan), управление переходит к Отчётчику, в противном случае — обратно к Выполняющему. Цикл заканчивается, поскольку current_step увеличивается при каждой итерации, а значение MAX_STEPS = 12 ограничивает общее количество шагов, даже если Перепланировщик добавляет новые шаги.
Один запрос, отслеживаемый
В примерном запросе просится, чтобы агент рассмотрел компании NVDA и AMD, вывел их коэффициенты P/E и свежие новости рядом друг с другом, а затем выбрал одну из них.
Планирование. Инструмент планирования генерирует пять шагов: полные показатели для NVDA, полные показатели для AMD, прямое сравнение коэффициентов P/E, поиск новостей о NVDA и поиск новостей о AMD. Сначала предоставляются отдельные профили для полного обзора, затем сравнение двух компаний, а новости идут в конце, поскольку являются дополнительным контекстом, а не основными данными для оценки.
Выполнение. На шагах 1 и 2 возвращается чистый JSON с показателями, и каждый раз инструмент Replanner проверяет, что данные корректны и план не изменился. На шаге 3 возвращается сравнение значений коэффициентов P/E:
{
"metric": "P/E",
"values": {
"NVDA": 58.3,
"AMD": 102.5
}
}
Оба значения присутствуют и являются действительными, поэтому ничего не требуется перепланировать. На этапах 4 и 5 по три новости возвращаются соответственно. После завершения этапа 5 current_step будет равен len(plan), и управление переходит к Репортеру.
Отчет. Получив пять полных записей в блокноте, Репортер указывает конкретные цифры: например, коэффициент P/E NVDA в 58,3 против 102,5 у AMD указывает на то, что NVDA имеет лучшую относительную стоимость, несмотря на более высокую рыночную капитализацию в размере около 2,8 триллиона долларов. Полный процесс занимает примерно 8–10 секунд на Groq и 30–40 секунд на локальном Ollama с использованием модели объемом 4 миллиарда параметров.
Самые важные обновления
Текущая версия использует фиктивные данные, но архитектура готова к использованию в производственных условиях. Самыми важными считаются три обновления:
- Реальные финансовые данные. Вместо поиска в словаре используются запросы к Yahoo Finance или Alpha Vantage. Подписи остаются прежними, а меняются только тела функций благодаря шаблону декоратора
@tool; поэтому Planner и Executor никогда не узнают, происходят ли числа из фиктивного JSON-файла или из REST API. - Постоянная память. В настоящее время временная память существует только в течение одной работы. С помощью векторных хранилищ, таких как ChromaDB или Pinecone, агент может обращаться к предыдущим анализам, отвечая на вопросы вроде «Как NVDA сравнивается с тем, что мы видели в прошлом месяце?», тем самым превращаясь в развивающегося аналитика, а не в инструмент без состояния.
Чего не может обеспечить структура
Цикл планирование-выполнение-перепланирование эффективно справляется со своей задачей. Структурированный вывод помогает агенту держаться курса, реестры инструментов предотвращают хаос, а механизмы безопасности ограничивают чрезмерные расходы. Такая структура остаётся надёжной, а код — аккуратным.
Тем не менее, интеллект всей системы основан на ЯЗЫКЕ БОЛЬШОЙ МОДЕЛИ, находящемся в её основе. Более мощная модель создаёт лучшие планы, замечает больше ошибок в процессе их перепланирования и составляет более точные отчёты, тогда как более слабая модель слепо следует структуре и предоставляет посредственный анализ. Граф не может сделать модель умнее; он лишь предоставляет ей более эффективную структуру для использования уже имеющегося у неё интеллекта.
Проблема управления проектами в значительной степени решена: состояние, оркестрация инструментов, поток выполнения и самокоррекция следуют чётким шаблонам. Однако вопросы принятия решений остаются открытыми. Когда именно Replanner должен перепланировать действия, а когда следует молчать? Когда число считается аномальным, а когда просто удивительным? В настоящее время такие решения принимаются на основе промптов, в надежде, что ЯЗЫК БОЛЬШОЙ МОДЕЛИ сможет их разрешить, что происходит чаще, чем ожидается, но реже, чем хотелось бы.
Связанные материалы
- Объяснение DS-STAR: агент Data Science от Google и причины его успеха — Подробный обзор агента Data Science DS-STAR от Google: его семь модулей, цикл отчетности DS-STAR+, опубликованные инструкции, результаты тестов на эффективность и существующие ограничения.
- Агент AI для DevOps на Python: LangChain, Claude и инструменты Guarded Bash — Создание агента, способного запускать команды bash, читать логи и выполнять запросы к Kubernetes: сначала правила безопасности, затем инструменты, цикл ReAct, работа с памятью, обертка FastAPI и способы устранения распространенных сбоев.