Створення агента планування в LangGraph: планування, виконання, перепланування, звітування
Чотиризв’язковий агент планування LangGraph для фінансових досліджень: структуровані плани, реєстр інструментів із урахуванням простору імен, обмеження на перепланування та запис виконання прикладу з трасуванням.
Джерела: Your AI Agent Isn’t Smart. Here’s How to Build One That Actually Thinks
Більшість агентів — це прикриті ланцюги запитів
Ось неприємний факт про AI-агентів: багато з них — це просто урізноманітнені ланцюги запитів, загорнуті у „парадний“ вигляд. Схема знайома: системний запит, кілька інструментів та, можливо, цикл ReAct, і результат називається агентом. Це працює під час демонстрацій та у сприятливих умовах. Але якщо інструмент повертає помилку, дані виглядають дивно чи запит вимагає більше ніж трьох кроків, вся система руйнується.
Альтернативою є агент планування. Він розробляє повний план дослідження перед тим, як використовувати будь-які інструменти, виконує цей план крок за кроком, а коли проміжні результати суперечать очікуванням, переписує те, що залишилося. Демо-версія — це система фінансових досліджень. Надайте їй запит простою англійською мовою, наприклад: «Порівняйте оцінки NVDA та AAPL та скажіть, яка з них зараз виглядає краще», — і вона створить багатокроковий план, виконає кожен крок за допомогою інструментів обробки даних, виправить себе, якщо щось піде не так, та напише звіт у стилі аналітика. Вона базується на LangGraph, працює з хмарними моделями через Groq або локальними через Ollama та поміщається приблизно у чотири файли.
Чому звичайний ReAct не підходить
ReAct (Reason-Act-Observe) — це шаблон, з якого починають більшість агентів. Штучний інтелект обмірковує свій наступний крок, викликає інструмент, читає отриману інформацію, знову обмірковує ситуацію та викликає наступний інструмент. Цей підхід популярний, простий та підходить для невеликих завдань.
Основною слабкістю є жадібність, адже агент ReAct планує лише один крок уперед. Якщо попросити дані про п’ять акцій за трьома показниками разом із останніми новинами, він почне, скажімо, з коефіцієнта P/E акції NVDA, і перший результат відволікає його увагу. Один з прототипів витратив сім викликів інструментів на кожен показник Tesla, перш ніж згадати про завдання порівняти кілька акцій; його кінцевий звіт складався зі зведення про прибутки Tesla та одного речення про інші акції.
Проблема не у можливостях моделі, а у її архітектурі. Якщо під час однієї процедури обробки даних необхідно вирішувати, що досліджувати, як це зробити та як інтерпретувати результати, жодне з цих рішень не отримує належної уваги. Агент планування розділяє мислення та дію: спочатку планує, потім виконує, а потім перевіряє, чи план все ще актуальний. Кожен вузол обробляє одне завдання, і вони виконуються по черзі.
Чотири вузли та один цикл
Агент планування — це орієнтований граф із чотирма вузлами, схожий на діаграму потоків, де кожен блок обробляє одне завдання та передає стан далі.
START --> Planner --> Executor <--> Replanner --> Reporter --> END
- Планувальник бере запит користувача та створює структурований, крок за кроком план дослідження. Кожен крок вказує інструмент, який потрібно використати, аргументи та причину.
- Виконавець бере наступний крок, викликає вказаний інструмент та записує результат у тимчасовий блок нотаток.
- Перепланувальник переглядає все, що було зібрано до цього моменту після кожного виконання. Якщо інструмент дав помилку або дані показують щось несподіване, він переписує залишкові кроки. Інакше він залишає план без змін.
- Звітувач перетворює всі записи з тимчасового блоку нотаток на остаточний звіт аналітика, який містить конкретні цифри та рекомендації.
Те, що керує процесом проектування, — це цикл виконавця-планувальника. Два вузли по черзі виконують завдання до завершення плану, і саме ця зміна призводить до самокоригуючої поведінки, адже після кожної дії агент може переглянути свої рішення.
Стан: спільна пам’ять
У 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 додавати нові елементи та залишати існуючі на своїх місцях. Коли виконавець повертає {"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,“ вказується „Compare P/E ratios to identify relative valuation,“ 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" if result.startswith("Error:") else "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
Графік майже схожий на англійську мову. Планувальник передає завдання Виконавцю, Виконавець — Перепланувальнику, а Перепланувальник, якщо залишилися кроки, передає їх знову Виконавцю, а коли все виконано — до Звітника. Звітник передає сигнал КІНЕЦЬ.
Енумерація 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 з використанням моделі 4B.
Найважливіші оновлення
Текуща версія використовує штучні дані, але архітектура готова до експлуатації. Найважливішими є три оновлення:
- Справжні фінансові дані. Замініть пошук у словнику на виклики сервісів Yahoo Finance або Alpha Vantage. Підписи залишаються незмінними, а змінюються лише тіла функцій завдяки шаблону декоратора
@tool; тому Planner та Executor ніколи не дізнаються, чи походять числа з фіктивного JSON-файлу, чи з REST API. - Постійна пам’ять. Зараз тимчасові дані зберігаються лише протягом одного запуску. За допомогою векторних баз даних, таких як ChromaDB чи Pinecone, агент може звертатися до попередніх аналізів, щоб відповісти на запитання на кшталт „Як NVDA порівнюється з тим, що ми бачили минулого місяця?“, стаючи еволюціонуючим аналітиком, а не інструментом без стану.
Що структура не може забезпечити
Цикл планування-виконання-перепланування ефективно виконує свою функцію. Структурований вихід допомагає агенту триматися курсу, реєстри інструментів запобігають хаосу, а механізми безпеки стримують неконтрольовані витрати. Дизайн залишається стабільним, а код — охайним.
Навіть так, інтелект системи в цілому залежить від LLM, який знаходиться під цією структурою. Сильніший модель створює кращі плани, помічає більше помилок під час перепланування та пише більш детальні звіти, тоді як слабкіший модель слідує структурі сумлінно та надає посередній аналіз. Граф не може зробити модель розумнішою; він лише надає їй кращу основу для використання вже наявного інтелекту.
Рівень управління проектом у більшій мірі вирішений: стан, оркестрація інструментів, потік виконання та самокорекція всі слідують чітким шаблонам. Однак питання судження залишаються відкритими. Коли саме Replanner має перепланувати, а коли слід залишатися без дій? Коли число є аномальним, а коли просто дивним? Наразі ці рішення формулюються у запитах, сподіваючись, що LLM їх вирішить — це працює частіше, ніж очікується, але рідше, ніж хотілося б.
Пов’язана література
- DS-STAR: пояснення, агент з науки про дані від Google та причини його успіху — детальний огляд агента з науки про дані DS-STAR від Google: його сім модулів, цикл звітності DS-STAR+, опубліковані запити, результати тестів та існуючі обмеження.
- Агент AI для DevOps на Python: LangChain, Claude та інструменти Bash — створення агента, який виконує команди Bash, читає журнали та здійснює запити до Kubernetes: спочатку правила безпеки, потім інструменти, цикл ReAct, пам’ять, обгортка FastAPI та способи усунення поширених проблем.