Ideas On How To Assemble A Manufacturing-Ready Multi-Agent Incident Response System Using OpenAI Swarm And Software program-Augmented Brokers

!pip -q arrange -U openai
!pip -q arrange -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")
   in addition to Exception:
       key = None
   if not key:
       import getpass
       key = getpass.getpass("Enter OPENAI_API_KEY (hidden): ").strip()
   if not key:
       improve RuntimeError("OPENAI_API_KEY not supplied")
   return key


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

import json
import re
from typing import Document, 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) -> Document[str]:
   return re.sub(r"[^a-z0-9s]", " ", s.lower()).break up()


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.kind(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:
       decisions = json.lots(options_json)
   in addition to Exception as e:
       return json.dumps({"error": str(e)})
   score = []
   for o in decisions:
       conf = float(o.get("confidence", 0.5))
       hazard = o.get("hazard", "medium")
       penalty = {"low": 0.1, "medium": 0.25, "extreme": 0.45}.get(hazard, 0.25)
       score.append({
           "chance": o.get("chance"),
           "confidence": conf,
           "hazard": hazard,
           "score": spherical(conf - penalty, 3)
       })
   score.kind(key=lambda x: x["score"], reverse=True)
   return json.dumps(score, 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(
   determine="Triage",
   model="gpt-4o-mini",
   instructions="""
Resolve which agent must take care of the request.
Use SRE for incident response.
Use Comms for purchaser or authorities messaging.
Use HandoffWriter for on-call notes.
Use Critic for evaluation or enchancment.
""",
   options=[search_kb, handoff_to_sre, handoff_to_comms, handoff_to_handoff_writer, handoff_to_critic]
)


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


comms_agent = Agent(
   determine="Comms",
   model="gpt-4o-mini",
   instructions="""
Produce an exterior purchaser change and an interior technical change.
""",
   options=[search_kb]
)


handoff_writer_agent = Agent(
   determine="HandoffWriter",
   model="gpt-4o-mini",
   instructions="""
Produce a transparent on-call handoff doc with regular headings.
""",
   options=[search_kb]
)


critic_agent = Agent(
   determine="Critic",
   model="gpt-4o-mini",
   instructions="""
Critique the sooner reply, then produce a refined final mannequin and a tips.
"""
)

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 safe, upstream timeouts rising.
Current a 30-minute movement plan and a purchaser change.
"""


print(run_pipeline(request))