В этом руководстве мы рассмотрим, как спроектировать и запустить полный конвейер оркестровки агентского ИИ на базе семантической маршрутизации, символических ограничений и циклов самокоррекции с использованием Gemini. Мы разберём, как структурировать агентов, распределять задачи, обеспечивать соблюдение ограничений и совершенствовать выходные данные с помощью чистой модульной архитектуры.
Настройка окружения
Мы настраиваем нашу основную среду, импортируя необходимые библиотеки, определяя ключ API и инициализируя клиента Gemini. Также мы устанавливаем структуру AgentMessage, которая действует как общий формат коммуникации между агентами.
«`python
import os
import json
import time
import typing
from dataclasses import dataclass, asdict
from google import genai
from google.genai import types
APIKEY = os.environ.get(«GEMINIAPI_KEY», «API Key»)
client = genai.Client(apikey=APIKEY)
@dataclass
class AgentMessage:
source: str
target: str
content: str
metadata: dict
timestamp: float = time.time()
«`
Когнитивный движок
Мы создаём когнитивный слой с помощью Gemini, что позволяет нам генерировать как текст, так и выходные данные в формате JSON в зависимости от инструкции. Также мы реализуем семантический маршрутизатор, который анализирует запросы и выбирает наиболее подходящего агента.
«`python
class CognitiveEngine:
@staticmethod
def generate(prompt: str, systeminstruction: str, jsonmode: bool = False) -> str:
config = types.GenerateContentConfig(
temperature=0.1,
responsemimetype=»application/json» if json_mode else «text/plain»
)
try:
response = client.models.generate_content(
model=»gemini-2.0-flash»,
contents=prompt,
config=config
)
return response.text
except Exception as e:
raise ConnectionError(f»Gemini API Error: {e}»)
class SemanticRouter:
def init(self, agents_registry: dict):
self.registry = agents_registry
def route(self, user_query: str) -> str:
prompt = f»»»
You are a Master Dispatcher. Analyze the user request and map it to the ONE best agent.
AVAILABLE AGENTS:
{json.dumps(self.registry, indent=2)}
USER REQUEST: «{user_query}»
Return ONLY a JSON object: {{«selectedagent»: «agentname», «reasoning»: «brief reason»}}
«»»
responsetext = CognitiveEngine.generate(prompt, «You are a routing system.», jsonmode=True)
try:
decision = json.loads(response_text)
print(f» [Router] Selected: {decision[‘selected_agent’]} (Reason: {decision[‘reasoning’]})»)
return decision[‘selected_agent’]
except:
return «general_agent»
«`
Агенты и оркестратор
Мы конструируем агентов-исполнителей и центрального оркестратора. Каждому агенту назначается чёткая роль: аналитика, творчества или кодирования, и мы настраиваем оркестратор для управления ими.
«`python
class Agent:
def init(self, name: str, instruction: str):
self.name = name
self.instruction = instruction
def execute(self, message: AgentMessage) -> str:
return CognitiveEngine.generate(
prompt=f»Input: {message.content}»,
system_instruction=self.instruction
)
class Orchestrator:
def init(self):
self.agents_info = {
«analyst_bot»: «Analyzes data, logic, and math. Returns structured JSON summaries.»,
«creative_bot»: «Writes poems, stories, and creative text. Returns plain text.»,
«coder_bot»: «Writes Python code snippets.»
}
self.workers = {
«analystbot»: Agent(«analystbot», «You are a Data Analyst. output strict JSON.»),
«creativebot»: Agent(«creativebot», «You are a Creative Writer.»),
«coderbot»: Agent(«coderbot», «You are a Python Expert. Return only code.»)
}
self.router = SemanticRouter(self.agents_info)
«`
Валидация и выполнение задач
Мы реализуем символические ограничения и цикл самокоррекции для обеспечения соблюдения таких ограничений, как строгий формат JSON или отсутствие Markdown. Мы запускаем итеративное усовершенствование всякий раз, когда выходные данные нарушают требования, позволяя нашим агентам исправлять собственные ошибки.
«`python
def validateconstraint(self, content: str, constrainttype: str) -> tuple[bool, str]:
if constrainttype == «jsononly»:
try:
json.loads(content)
return True, «Valid JSON»
except:
return False, «Output was not valid JSON.»
if constrainttype == «nomarkdown»:
if ««`» in content:
return False, «Output contains Markdown code blocks, which are forbidden.»
return True, «Valid Text»
return True, «Pass»
def runtask(self, userinput: str, constraint: str = None, max_retries: int = 2):
print(f»\n— New Task: {user_input} —«)
targetname = self.router.route(userinput)
worker = self.workers.get(target_name)
currentinput = userinput
history = []
for attempt in range(max_retries + 1):
try:
msg = AgentMessage(source=»User», target=targetname, content=currentinput, metadata={})
print(f» [Exec] {worker.name} working… (Attempt {attempt+1})»)
result = worker.execute(msg)
if constraint:
isvalid, errormsg = self.validate_constraint(result, constraint)
if not is_valid:
print(f» [Guardrail] VIOLATION: {error_msg}»)
currentinput = f»Your previous answer failed a check.\nOriginal Request: {userinput}\nYour Answer: {result}\nError: {error_msg}\nFIX IT immediately.»
continue
print(f» [Success] Final Output:\n{result[:100]}…»)
return result
except Exception as e:
print(f» [System Error] {e}»)
time.sleep(1)
print(» [Failed] Max retries reached or self-correction failed.»)
return None
«`
Запуск задач
Мы выполняем два полных сценария, демонстрируя маршрутизацию, выполнение агентов и проверку ограничений в действии. Мы запускаем аналитическую задачу с ограничением JSON и задачу по кодированию с ограничениями Markdown, чтобы наблюдать за рефлексивным поведением.
«`python
if name == «main«:
orchestrator = Orchestrator()
orchestrator.run_task(
«Compare the GDP of France and Germany in 2023.»,
constraint=»json_only»
)
orchestrator.run_task(
«Write a Python function for Fibonacci numbers.»,
constraint=»no_markdown»
)
«`
В заключение мы видим, как несколько компонентов — маршрутизация, агенты-исполнители, ограничения и самокоррекция — объединяются для создания надёжной и интеллектуальной мультиагентной системы. Мы наблюдаем, как каждая часть способствует надёжному выполнению задач, обеспечивая точность и соответствие выходных данных.
1. Какие ключевые компоненты включает в себя мультиагентная система искусственного интеллекта, описанная в статье?
В статье описаны следующие ключевые компоненты:
* когнитивный слой с использованием Gemini для генерации текста и выходных данных в формате JSON;
* семантический маршрутизатор, который анализирует запросы и выбирает наиболее подходящего агента;
* агенты-исполнители, каждому из которых назначается чёткая роль (аналитика, творчества или кодирования);
* центральный оркестратор для управления агентами;
* символические ограничения и цикл самокоррекции для обеспечения соблюдения требований к формату выходных данных.
2. Какие роли назначены агентам в системе и как они взаимодействуют с оркестратором?
Агентам назначены чёткие роли: аналитика, творчества или кодирования. Оркестратор управляет ими, направляя запросы к соответствующим агентам и контролируя выполнение задач.
3. Какие ограничения реализованы в системе для обеспечения качества выходных данных?
В системе реализованы символические ограничения, такие как строгий формат JSON или отсутствие Markdown. Эти ограничения проверяются с помощью функции validate_constraint, которая возвращает True или False в зависимости от соответствия выходных данных требованиям.
4. Как система обеспечивает самокоррекцию при нарушении ограничений?
При нарушении ограничений система запускает итеративное усовершенствование. Если выходные данные не соответствуют требованиям, система исправляет ошибки и повторяет попытку. Это позволяет агентам исправлять собственные ошибки и обеспечивать соответствие выходных данных требованиям.
5. Какие примеры задач приведены в статье для демонстрации работы системы?
В статье приведены два примера задач:
* аналитическая задача с ограничением JSON (сравнение ВВП Франции и Германии в 2023 году);
* задача по кодированию с ограничениями Markdown (написание функции для вычисления чисел Фибоначчи на Python).