В этом руководстве мы создадим продвинутую, но практическую мультиагентную систему с использованием OpenAI Swarm, которая работает в Colab. Мы покажем, как можно организовать работу специализированных агентов, таких как агент сортировки (triage agent), агент SRE, агент коммуникаций (communications agent) и критик (critic), для совместной работы над реальным сценарием производственного инцидента.
Настройка среды
Мы настроим среду и безопасно загрузим ключ API OpenAI, чтобы блокнот мог безопасно работать в Google Colab. Ключ будет извлечён из секретов Colab, когда это возможно, а в противном случае будет использоваться скрытый запрос. Это упрощает аутентификацию и позволяет использовать её повторно в разных сессиях.
Импортирование основных утилит Python и инициализация клиента Swarm
«`python
import os
def loadopenaikey():
try:
from google.colab import userdata
key = userdata.get(«OPENAIAPIKEY»)
except Exception:
key = None
if not key:
import getpass
key = getpass.getpass(«Enter OPENAIAPIKEY (hidden): «).strip()
if not key:
raise RuntimeError(«OPENAIAPIKEY not provided»)
return key
os.environ[«OPENAIAPIKEY»] = loadopenaikey()
«`
«`python
import json
import re
from typing import List, Dict
from swarm import Swarm, Agent
client = Swarm()
«`
Определение внутренней базы знаний и реализация функции поиска
«`python
KB_DOCS = [
{
«id»: «kb-incident-001»,
«title»: «API Latency Incident Playbook»,
«text»: «If p95 latency spikes, validate deploys, dependencies, and error rates. Rollback, cache, rate-limit, scale. Compare p50 vs p99 and inspect upstream timeouts.»
},
{
«id»: «kb-risk-001»,
«title»: «Risk Communication Guidelines»,
«text»: «Updates must include impact, scope, mitigation, owner, and next update. Avoid blame and separate internal vs external messaging.»
},
{
«id»: «kb-ops-001»,
«title»: «On-call Handoff Template»,
«text»: «Include summary, timeline, current status, mitigations, open questions, next actions, and owners.»
},
]
def _normalize(s: str) -> List[str]:
return re.sub(r»[^a-z0-9\s]», » «, s.lower()).split()
def searchkb(query: str, topk: 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)
«`
Введение структурированного инструмента для оценки и ранжирования стратегий смягчения последствий
«`python
def estimatemitigationimpact(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)
«`
Определение функций передачи управления
«`python
def handofftosre():
return sre_agent
def handofftocomms():
return comms_agent
def handofftohandoff_writer():
return handoffwriteragent
def handofftocritic():
return critic_agent
«`
Конфигурация специализированных агентов
«`python
triage_agent = Agent(
name=»Triage»,
model=»gpt-4o-mini»,
instructions=»»»
Decide which agent should handle the request.
Use SRE for incident response.
Use Comms for customer or executive messaging.
Use HandoffWriter for on-call notes.
Use Critic for review or improvement.
«»»,
functions=[searchkb, handofftosre, handofftocomms, handofftohandoffwriter, handofftocritic]
)
sre_agent = Agent(
name=»SRE»,
model=»gpt-4o-mini»,
instructions=»»»
Produce a structured incident response with triage steps,
ranked mitigations, ranked hypotheses, and a 30-minute plan.
«»»,
functions=[searchkb, estimatemitigation_impact]
)
comms_agent = Agent(
name=»Comms»,
model=»gpt-4o-mini»,
instructions=»»»
Produce an external customer update and an internal technical update.
«»»,
functions=[search_kb]
)
handoffwriteragent = Agent(
name=»HandoffWriter»,
model=»gpt-4o-mini»,
instructions=»»»
Produce a clean on-call handoff document with standard headings.
«»»,
functions=[search_kb]
)
critic_agent = Agent(
name=»Critic»,
model=»gpt-4o-mini»,
instructions=»»»
Critique the previous answer, then produce a refined final version and a checklist.
«»»
)
«`
Запуск конвейера
«`python
def runpipeline(userrequest: str):
messages = [{«role»: «user», «content»: user_request}]
r1 = client.run(agent=triageagent, messages=messages, maxturns=8)
messages2 = r1.messages + [{«role»: «user», «content»: «Review and improve the last answer»}]
r2 = client.run(agent=criticagent, messages=messages2, maxturns=4)
return r2.messages[-1][«content»]
request = «»»
Production p95 latency jumped from 250ms to 2.5s after a deploy.
Errors slightly increased, DB CPU stable, upstream timeouts rising.
Provide a 30-minute action plan and a customer update.
«»»
print(run_pipeline(request))
«`
Мы собрали полный конвейер оркестрации, который последовательно выполняет сортировку, специализированные рассуждения и критическую доработку. Этот фрагмент показывает, как мы запускаем сквозной рабочий процесс с помощью одного вызова функции. Он связывает всех агентов и инструменты в согласованную систему, готовую к работе в производственных условиях.
В заключение мы установили чёткий шаблон для проектирования агентно-ориентированных систем с помощью OpenAI Swarm, который подчёркивает ясность, разделение обязанностей и итеративное совершенствование. Мы показали, как можно разумно распределять задачи, обогащать рассуждения агентов с помощью локальных инструментов и улучшать качество выходных данных с помощью цикла критики, сохраняя при этом простую настройку, удобную для Colab. Этот подход позволяет нам перейти от экспериментов к реальным операционным сценариям использования, делая Swarm мощной основой для построения надёжных агентских ИИ-рабочих процессов, готовых к работе в производственных условиях.
1. Какие специализированные агенты используются в мультиагентной системе реагирования на инциденты и какие функции они выполняют?
*В мультиагентной системе используются следующие агенты:
Triage Agent* — сортировка запросов и определение, какой агент должен обработать запрос.
SRE* — создание структурированного ответа на инцидент с шагами сортировки, ранжированными смягчениями, гипотезами и 30-минутным планом.
Comms* — создание внешних обновлений для клиентов и внутренних технических обновлений.
HandoffWriter* — создание документа передачи смены с стандартными заголовками.
Critic — критика предыдущего ответа, создание уточнённой финальной версии и чек-листа.
2. Как организована работа агентов в системе и как они взаимодействуют друг с другом?
Агенты взаимодействуют через общий клиент Swarm. Triage Agent определяет, какой агент должен обработать запрос, и передаёт управление соответствующему агенту. Например, если запрос требует ответа от SRE, Triage Agent передаёт управление SRE. SRE выполняет свою задачу и передаёт управление обратно Triage Agent или другому агенту при необходимости.
3. Какие инструменты используются для поиска информации в базе знаний и оценки стратегий смягчения последствий?
Для поиска информации в базе знаний используется функция searchkb, которая ищет документы по запросу пользователя. Для оценки стратегий смягчения последствий используется функция estimatemitigation_impact, которая ранжирует стратегии по их эффективности.
4. Как обеспечивается безопасность при работе с ключами API в Google Colab?
Ключ API OpenAI извлекается из секретов Colab, когда это возможно. В противном случае используется скрытый запрос. Это упрощает аутентификацию и позволяет использовать её повторно в разных сессиях.
5. Какие преимущества даёт использование мультиагентной системы реагирования на инциденты по сравнению с традиционными подходами?
Мультиагентная система позволяет организовать работу специализированных агентов для совместной работы над реальным сценарием производственного инцидента. Это обеспечивает более эффективное и быстрое реагирование на инциденты, а также улучшает качество выходных данных за счёт итеративного совершенствования.