Реализация кода для создания иерархического планирующего агента ИИ с использованием открытых LLM, выполнения инструментов и структурированного мультиагентного рассуждения

В этом руководстве мы создаём иерархического агента-планировщика, используя модель с открытым исходным кодом. Мы разрабатываем структурированную мультиагентную архитектуру, включающую агента-планировщика, агента-исполнителя и агента-агрегатора, где каждый компонент играет специализированную роль в решении сложных задач.

Агент-планировщик

Используем агента-планировщика для разбиения высокоуровневых целей на выполнимые шаги.

Агент-исполнитель

Агент-исполнитель выполняет эти шаги, используя рассуждения или выполнение инструментов Python.

Агент-агрегатор

Агент-агрегатор синтезирует результаты в согласованный окончательный ответ.

Установка необходимых библиотек

«`python
!pip -q install -U transformers accelerate bitsandbytes sentencepiece
«`

«`python
import json
import re
import io
import contextlib
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
«`

Инициализация модели

«`python
MODEL_ID = «Qwen/Qwen2.5-1.5B-Instruct»
DEVICE = «cuda» if torch.cuda.is_available() else «cpu»

print(«Device:», DEVICE)

tokenizer = AutoTokenizer.frompretrained(MODELID, use_fast=True)

model = None
if DEVICE == «cuda»:
try:
model = AutoModelForCausalLM.from_pretrained(
MODEL_ID,
device_map=»auto»,
torch_dtype=»auto»,
loadin4bit=True,
)
print(«Loaded model in 4-bit.»)
except Exception as e:
print(«4-bit load failed, falling back to normal load:», str(e)[:200])
model = AutoModelForCausalLM.from_pretrained(
MODEL_ID,
device_map=»auto»,
torch_dtype=»auto»,
)
else:
model = AutoModelForCausalLM.from_pretrained(
MODEL_ID,
torch_dtype=torch.float32,
).to(DEVICE)

model.eval()
«`

Определение основной функции взаимодействия

«`python
def llmchat(system: str, user: str, maxnew_tokens: int = 500, temperature: float = 0.3) -> str:
messages = [
{«role»: «system», «content»: system.strip()},
{«role»: «user», «content»: user.strip()},
]
prompt = tokenizer.applychattemplate(messages, tokenize=False, addgenerationprompt=True)
inputs = tokenizer(prompt, return_tensors=»pt»).to(model.device)

with torch.no_grad():
out = model.generate(
inputs,
maxnewtokens=maxnewtokens,
do_sample=True if temperature > 0 else False,
temperature=temperature,
top_p=0.9,
repetition_penalty=1.05,
eostokenid=tokenizer.eostokenid,
)

text = tokenizer.decode(out[0], skipspecialtokens=True)
return text.split(user.strip())[-1].strip()
«`

Реализация механизма извлечения JSON

«`python
def extractjsonblock(text: str) -> Optional[Any]:
fenced = re.search(r»«`json\s(.?)\s*«`», text, flags=re.DOTALL | re.IGNORECASE)
if fenced:
cand = fenced.group(1).strip()
try:
return json.loads(cand)
except:
pass

start_obj = text.find(«{«)
start_arr = text.find(«[«)
starts = [i for i in [startobj, startarr] if i != -1]
if not starts:
return None
start = min(starts)

s = text[start:]
stack = []
end = None
for i, ch in enumerate(s):
if ch in «{[«:
stack.append(ch)
elif ch in «}]»:
if not stack:
continue
op = stack.pop()
if (op == «{» and ch != «}») or (op == «[» and ch != «]»):
return None
if not stack:
end = i + 1
break
if end is None:
return None

cand = s[:end].strip()
try:
return json.loads(cand)
except:
return None
«`

Определение системы для агента-планировщика

«`python
PLANNER_SYSTEM = «»»
You are a Hierarchical Planner Agent.
You break down the user’s task into 3-8 clear steps.
You MUST output ONLY valid JSON (no extra text).
Schema:
{
«goal»: «string»,
«assumptions»: [«string», …],
«steps»: [
{
«id»: 1,
«title»: «short title»,
«instruction»: «what to do»,
«tool»: «none|llm|python»,
«expected_output»: «what we should get»
}
]
}
Guidelines:

  • Use tool=»python» only if calculation / data processing / simulation helps.

  • Keep steps independent and executable.

«»»
«`

Определение системы для агента-исполнителя

«`python
EXECUTOR_SYSTEM = «»»
You are an Executor Agent.
Given a step and the current context, you produce the result for that step.
If the step tool is «python», output ONLY Python code (no backticks).
If the step tool is «llm» or «none», output a concise result, and reference any prior step outputs when relevant.
«»»
«`

Определение системы для агента-агрегатора

«`python
AGGREGATOR_SYSTEM = «»»
You are an Aggregator Agent.
You combine step outputs into a final, polished response to the original task.
Be structured, correct, and practical.
If the task asks for an actionable plan, include bullet points and clear next actions.
«`

Демонстрационная задача

«`python
demo_task = «»»
Create a practical checklist to launch a small multi-agent system in Python for coordinating logistics:

  • One planner agent that decomposes tasks

  • Two executor agents (routing + inventory)

  • A simple memory store for past decisions

Keep it lightweight and runnable in Colab.
«»»

= runhierarchicalagent(demotask, verbose=True)
«`

Мы реализовали иерархическую мультиагентную систему, которая демонстрирует структурированное планирование, выполнение и агрегацию в рамках единой структуры. Мы позволили агенту-планировщику разбивать сложные задачи, наделили агента-исполнителя возможностью рассуждать и выполнять динамический код Python и использовали агента-агрегатора для генерации уточнённых, действенных выходных данных. Объединив открытые языковые модели, структурированные запросы и выполнение инструментов, мы создали мощную и расширяемую основу для построения продвинутых автономных агентов.

1. Какие компоненты включает в себя структурированная мультиагентная архитектура, представленная в статье?

Ответ: в статье представлена структурированная мультиагентная архитектура, включающая агента-планировщика, агента-исполнителя и агента-агрегатора. Агент-планировщик разбивает высокоуровневые цели на выполнимые шаги, агент-исполнитель выполняет эти шаги, используя рассуждения или выполнение инструментов Python, а агент-агрегатор синтезирует результаты в согласованный окончательный ответ.

2. Какие инструменты и библиотеки используются для реализации иерархического планирующего агента ИИ в статье?

Ответ: для реализации иерархического планирующего агента ИИ в статье используются следующие инструменты и библиотеки: `transformers`, `accelerate`, `bitsandbytes`, `sentencepiece`, `torch`, `AutoTokenizer`, `AutoModelForCausalLM`.

3. Какие функции и механизмы используются для взаимодействия между компонентами иерархической мультиагентной системы?

Ответ: для взаимодействия между компонентами иерархической мультиагентной системы используются функции `llmchat`, `extractjson_block`, а также системы для агента-планировщика, агента-исполнителя и агента-агрегатора. Эти системы определяют правила и механизмы взаимодействия между компонентами, позволяя им эффективно решать сложные задачи.

4. Какие шаги включает в себя процесс создания иерархического планирующего агента ИИ с использованием открытых LLM и инструментов Python?

Ответ: процесс создания иерархического планирующего агента ИИ включает в себя следующие шаги: установка необходимых библиотек, инициализация модели, определение основной функции взаимодействия, реализация механизма извлечения JSON, определение системы для агента-планировщика, агента-исполнителя и агента-агрегатора, а также демонстрационная задача для проверки работы системы.

5. Какие преимущества даёт использование иерархической мультиагентной системы для решения сложных задач?

Ответ: использование иерархической мультиагентной системы позволяет разбить сложные задачи на более мелкие и выполнимые шаги, эффективно распределить роли между компонентами системы, а также обеспечить согласованность и точность выходных данных. Это делает систему более гибкой, масштабируемой и эффективной для решения сложных задач.

Источник