Find out how to Construct a Manufacturing-Prepared Multi-Agent Incident Response System Utilizing OpenAI Swarm and Instrument-Augmented Brokers

!pip -q set up -U openai
!pip -q set up -U "git+https://github.com/openai/swarm.git"


import os


def load_openai_key():
   attempt:
       from google.colab import userdata
       key = userdata.get("OPENAI_API_KEY")
   besides Exception:
       key = None
   if not key:
       import getpass
       key = getpass.getpass("Enter OPENAI_API_KEY (hidden): ").strip()
   if not key:
       elevate RuntimeError("OPENAI_API_KEY not offered")
   return key


os.environ["OPENAI_API_KEY"] = load_openai_key()

import json
import re
from typing import Checklist, Dict
from swarm import Swarm, Agent


shopper = Swarm()

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) -> Checklist[str]:
   return re.sub(r"[^a-z0-9s]", " ", s.decrease()).break up()


def search_kb(question: str, top_k: int = 3) -> str:
   q = set(_normalize(question))
   scored = []
   for d in KB_DOCS:
       rating = len(q.intersection(set(_normalize(d["title"] + " " + d["text"]))))
       scored.append((rating, d))
   scored.type(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:
   attempt:
       choices = json.masses(options_json)
   besides Exception as e:
       return json.dumps({"error": str(e)})
   rating = []
   for o in choices:
       conf = float(o.get("confidence", 0.5))
       danger = o.get("danger", "medium")
       penalty = {"low": 0.1, "medium": 0.25, "excessive": 0.45}.get(danger, 0.25)
       rating.append({
           "choice": o.get("choice"),
           "confidence": conf,
           "danger": danger,
           "rating": spherical(conf - penalty, 3)
       })
   rating.type(key=lambda x: x["score"], reverse=True)
   return json.dumps(rating, 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(
   title="Triage",
   mannequin="gpt-4o-mini",
   directions="""
Resolve which agent ought to deal with the request.
Use SRE for incident response.
Use Comms for buyer or govt messaging.
Use HandoffWriter for on-call notes.
Use Critic for assessment or enchancment.
""",
   capabilities=[search_kb, handoff_to_sre, handoff_to_comms, handoff_to_handoff_writer, handoff_to_critic]
)


sre_agent = Agent(
   title="SRE",
   mannequin="gpt-4o-mini",
   directions="""
Produce a structured incident response with triage steps,
ranked mitigations, ranked hypotheses, and a 30-minute plan.
""",
   capabilities=[search_kb, estimate_mitigation_impact]
)


comms_agent = Agent(
   title="Comms",
   mannequin="gpt-4o-mini",
   directions="""
Produce an exterior buyer replace and an inside technical replace.
""",
   capabilities=[search_kb]
)


handoff_writer_agent = Agent(
   title="HandoffWriter",
   mannequin="gpt-4o-mini",
   directions="""
Produce a clear on-call handoff doc with customary headings.
""",
   capabilities=[search_kb]
)


critic_agent = Agent(
   title="Critic",
   mannequin="gpt-4o-mini",
   directions="""
Critique the earlier reply, then produce a refined closing model and a guidelines.
"""
)

def run_pipeline(user_request: str):
   messages = [{"role": "user", "content": user_request}]
   r1 = shopper.run(agent=triage_agent, messages=messages, max_turns=8)
   messages2 = r1.messages + [{"role": "user", "content": "Review and improve the last answer"}]
   r2 = shopper.run(agent=critic_agent, messages=messages2, max_turns=4)
   return r2.messages[-1]["content"]


request = """
Manufacturing p95 latency jumped from 250ms to 2.5s after a deploy.
Errors barely elevated, DB CPU steady, upstream timeouts rising.
Present a 30-minute motion plan and a buyer replace.
"""


print(run_pipeline(request))