Проектирование самоисправляющейся AI системы на основе Gemini

Обзор AI Оркестрации

В этом руководстве мы исследуем, как проектировать и запускать полипредставленную AI оркестрацию, используя семантическую маршрутизацию, символические охранные механизмы и петли самокоррекции на основе Gemini.

Структурирование агентов и распределение задач

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

Настройка основной среды

import os
import json
import time
import typing
from dataclasses import dataclass, asdict
from google import genai
from google.genai import types
 
API_KEY = os.environ.get("GEMINI_API_KEY", "API Key")
client = genai.Client(api_key=API_KEY)
 
@dataclass
class AgentMessage:
   source: str
   target: str
   content: str
   metadata: dict
   timestamp: float = time.time()

Мы настраиваем основную среду, импортируя необходимые библиотеки, определяя API-ключ и инициализируя клиент Gemini. Устанавливаем структуру AgentMessage, которая служит общим форматом общения между агентами.

Реализация когнитивного движка

class CognitiveEngine:
   @staticmethod
   def generate(prompt: str, system_instruction: str, json_mode: bool = False) -> str:
       config = types.GenerateContentConfig(
           temperature=0.1,
           response_mime_type="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}")

Мы строим когнитивный уровень, используя Gemini, что позволяет генерировать как текстовые, так и JSON-выводы в зависимости от инструкции.

Семантическая маршрутизация: анализ запросов

class SemanticRouter:
   def __init__(self, agents_registry: dict):
       self.registry = agents_registry
 
   def route(self, user_query: str) -> str:
       prompt = f"""
       Вы - главный диспетчер. Проанализируйте запрос пользователя и сопоставьте его с одним лучшим агентом.
       ДОСТУПНЫЕ АГЕНТЫ:
       {json.dumps(self.registry, indent=2)}
       ЗАПРОС ПОЛЬЗОВАТЕЛЯ: "{user_query}"
       Верните ТОЛЬКО JSON-объект: {{"selected_agent": "agent_name", "reasoning": "brief reason"}}
       """
       response_text = CognitiveEngine.generate(prompt, "Вы - система маршрутизации.", json_mode=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"

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

Создание рабочих агентов

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
       )

Мы создаем рабочих агентов и центрального оркестратора, определяя роли для каждого агента: аналитик, креативный или кодер.

Проверка ограничений и самокоррекция

def validate_constraint(self, content: str, constraint_type: str) -> tuple[bool, str]:
   if constraint_type == "json_only":
       try:
           json.loads(content)
           return True, "Valid JSON"
       except:
           return False, "Output was not valid JSON."
   if constraint_type == "no_markdown":
       if "```" in content:
           return False, "Вывод содержит блоки кода Markdown, что запрещено."
       return True, "Valid Text"
   return True, "Pass"

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

Исполнение сценариев

if __name__ == "__main__":
   orchestrator = Orchestrator()
   orchestrator.run_task(
       "Сравните ВВП Франции и Германии в 2023 году.",
       constraint="json_only"
   )
   orchestrator.run_task(
       "Напишите функцию Python для чисел Фибоначчи.",
       constraint="no_markdown"
   )

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

Заключение

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