Создание системы многопользовательского реагирования на инциденты с OpenAI

Обзор

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

Передача задач и интеграция инструментов

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

Настройка окружения

Мы настраиваем окружение и безопасно загружаем ключ API OpenAI, чтобы блокнот мог безопасно работать в Google Colab. Мы обеспечиваем извлечение ключа из секретов Colab, когда это возможно, и в противном случае обращаемся к скрытой подсказке. Это делает аутентификацию простой и многоразовой.

!pip -q install -U openai
!pip -q install -U "git+https://github.com/openai/swarm.git"
 
import os
 
def load_openai_key():
   try:
       from google.colab import userdata
       key = userdata.get("OPENAI_API_KEY")
   except Exception:
       key = None
   if not key:
       import getpass
       key = getpass.getpass("Введите OPENAI_API_KEY (скрыто): ").strip()
   if not key:
       raise RuntimeError("OPENAI_API_KEY не предоставлен")
   return key
 
os.environ["OPENAI_API_KEY"] = load_openai_key()

Инициализация клиента Swarm

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

import json
import re
from typing import List, Dict
from swarm import Swarm, Agent
 
client = Swarm()

Создание базы знаний

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

KB_DOCS = [
   {
       "id": "kb-incident-001",
       "title": "План действий по инцидентам с задержкой API",
       "text": "Если p95 задержки увеличивается, проверьте развертывания, зависимости и количеству ошибок. Откат, кеширование, ограничение частоты, увеличение масштабов. Сравните p50 с p99 и проверьте тайм-ауты вверх по потоку."
   },
   {
       "id": "kb-risk-001",
       "title": "Руководство по коммуникации риска",
       "text": "Обновления должны содержать влияние, масштаб, смягчение, владельца и следующее обновление. Избегайте обвинений и разделите внутренние и внешние сообщения."
   },
   {
       "id": "kb-ops-001",
       "title": "Шаблон передачи дежурного",
       "text": "Включите сводку, временную линию, текущий статус, смягчения, открытые вопросы, следующие действия и владельцев."
   },
]
 
def _normalize(s: str) -> List[str]:
   return re.sub(r"[^a-z0-9\s]", " ", s.lower()).split()
 
def search_kb(query: str, top_k: int = 3) -> str:
   q = set(_normalize(query))
   scored = []
   for d in KB_DOCS:
       score = len(q.intersection(set(_normalize(d["title"] + " " + d["text"]))))
       scored.append((score, d))
   scored.sort(key=lambda x: x[0], reverse=True)
   docs = [d for s, d in scored[:top_k] if s > 0] or [scored[0][1]]
   return json.dumps(docs, indent=2)

Оценка стратегий смягчения

Мы вводим структурированный инструмент, который оценивает и ранжирует стратегии смягчения на основе уверенности и риска.

def estimate_mitigation_impact(options_json: str) -> str:
   try:
       options = json.loads(options_json)
   except Exception as e:
       return json.dumps({"error": str(e)})
   ranking = []
   for o in options:
       conf = float(o.get("confidence", 0.5))
       risk = o.get("risk", "medium")
       penalty = {"low": 0.1, "medium": 0.25, "high": 0.45}.get(risk, 0.25)
       ranking.append({
           "option": o.get("option"),
           "confidence": conf,
           "risk": risk,
           "score": round(conf - penalty, 3)
       })
   ranking.sort(key=lambda x: x["score"], reverse=True)
   return json.dumps(ranking, indent=2)

Управление передачей задач агентами

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

def handoff_to_sre():
   return sre_agent
 
def handoff_to_comms():
   return comms_agent
 
def handoff_to_handoff_writer():
   return handoff_writer_agent
 
def handoff_to_critic():
   return critic_agent

Настройка специализированных агентов

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

triage_agent = Agent(
   name="Triage",
   model="gpt-4o-mini",
   instructions="""
Определите, какой агент должен обработать запрос.
Используйте SRE для реагирования на инциденты.
Используйте Comms для обновлений для клиентов или руководства.
Используйте HandoffWriter для заметок дежурного.
Используйте Critic для рецензирования или улучшения.
""",
   functions=[search_kb, handoff_to_sre, handoff_to_comms, handoff_to_handoff_writer, handoff_to_critic]
)
 
# Определения других агентов...

Запуск оркестровочного пайплайна

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

def run_pipeline(user_request: str):
   messages = [{"role": "user", "content": user_request}]
   r1 = client.run(agent=triage_agent, messages=messages, max_turns=8)
   messages2 = r1.messages + [{"role": "user", "content": "Просмотреть и улучшить последний ответ"}]
   r2 = client.run(agent=critic_agent, messages=messages2, max_turns=4)
   return r2.messages[-1]["content"]

Заключение

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