In diesem Tutorial erstellen wir ein fortschrittliches Multi-Agent-Incident-Response-System mit AgentScope. Wir orchestrieren mehrere ReAct-Agenten mit jeweils klar definierten Rollen wie Routing, Triage, Analyse, Schreiben und Überprüfung und verbinden sie durch strukturiertes Routing und einen gemeinsamen Nachrichten-Hub. Durch die Integration von OpenAI-Modellen, leichtgewichtigen Instrument-Aufrufen und einem einfachen internen Runbook demonstrieren wir, wie komplexe, reale Agenten-Workflows in reinem Python ohne umfangreiche Infrastruktur oder spröden Klebercode erstellt werden können. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
!pip -q set up "agentscope>=0.1.5" pydantic nest_asyncio
import os, json, re
from getpass import getpass
from typing import Literal
from pydantic import BaseModel, Area
import nest_asyncio
nest_asyncio.apply()
from agentscope.agent import ReActAgent
from agentscope.message import Msg, TextBlock
from agentscope.mannequin import OpenAIChatModel
from agentscope.formatter import OpenAIChatFormatter
from agentscope.reminiscence import InMemoryMemory
from agentscope.device import Toolkit, ToolResponse, execute_python_code
from agentscope.pipeline import MsgHub, sequential_pipeline
if not os.environ.get("OPENAI_API_KEY"):
os.environ("OPENAI_API_KEY") = getpass("Enter OPENAI_API_KEY (hidden): ")
OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")Wir richten die Ausführungsumgebung ein und installieren alle erforderlichen Abhängigkeiten, damit das Tutorial zuverlässig auf Google Colab läuft. Wir laden den OpenAI-API-Schlüssel sicher und initialisieren die zentralen AgentScope-Komponenten, die von allen Agenten gemeinsam genutzt werden. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
RUNBOOK = (
{"id": "P0", "title": "Severity Coverage", "textual content": "P0 vital outage, P1 main degradation, P2 minor concern"},
{"id": "IR1", "title": "Incident Triage Guidelines", "textual content": "Assess blast radius, timeline, deployments, errors, mitigation"},
{"id": "SEC7", "title": "Phishing Escalation", "textual content": "Disable account, reset classes, block sender, protect proof"},
)
def _score(q, d):
q = set(re.findall(r"(a-z0-9)+", q.decrease()))
d = re.findall(r"(a-z0-9)+", d.decrease())
return sum(1 for w in d if w in q) / max(1, len(d))
async def search_runbook(question: str, top_k: int = 2) -> ToolResponse:
ranked = sorted(RUNBOOK, key=lambda r: _score(question, r("title") + r("textual content")), reverse=True)(: max(1, int(top_k)))
textual content = "nn".be part of(f"({r('id')}) {r('title')}n{r('textual content')}" for r in ranked)
return ToolResponse(content material=(TextBlock(kind="textual content", textual content=textual content)))
toolkit = Toolkit()
toolkit.register_tool_function(search_runbook)
toolkit.register_tool_function(execute_python_code)Wir definieren ein leichtes internes Runbook und implementieren darüber ein einfaches relevanzbasiertes Suchtool. Wir registrieren diese Funktion zusammen mit einem Python-Ausführungstool, das es Agenten ermöglicht, Richtlinienwissen abzurufen oder Ergebnisse dynamisch zu berechnen. Es zeigt, wie wir Agenten mit externen Fähigkeiten erweitern, die über das reine sprachliche Denken hinausgehen. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
def make_model():
return OpenAIChatModel(
model_name=OPENAI_MODEL,
api_key=os.environ("OPENAI_API_KEY"),
generate_kwargs={"temperature": 0.2},
)
class Route(BaseModel):
lane: Literal("triage", "evaluation", "report", "unknown") = Area(...)
purpose: str = Area(...)
router = ReActAgent(
title="Router",
sys_prompt="Route the request to triage, evaluation, or report and output structured JSON solely.",
mannequin=make_model(),
formatter=OpenAIChatFormatter(),
reminiscence=InMemoryMemory(),
)
triager = ReActAgent(
title="Triager",
sys_prompt="Classify severity and rapid actions utilizing runbook search when helpful.",
mannequin=make_model(),
formatter=OpenAIChatFormatter(),
reminiscence=InMemoryMemory(),
toolkit=toolkit,
)
analyst = ReActAgent(
title="Analyst",
sys_prompt="Analyze logs and compute summaries utilizing python device when useful.",
mannequin=make_model(),
formatter=OpenAIChatFormatter(),
reminiscence=InMemoryMemory(),
toolkit=toolkit,
)
author = ReActAgent(
title="Author",
sys_prompt="Write a concise incident report with clear construction.",
mannequin=make_model(),
formatter=OpenAIChatFormatter(),
reminiscence=InMemoryMemory(),
)
reviewer = ReActAgent(
title="Reviewer",
sys_prompt="Critique and enhance the report with concrete fixes.",
mannequin=make_model(),
formatter=OpenAIChatFormatter(),
reminiscence=InMemoryMemory(),
)
Wir erstellen mehrere spezialisierte ReAct-Agenten und einen strukturierten Router, der entscheidet, wie jede Benutzeranfrage behandelt werden soll. Wir weisen den Triage-, Analyse-, Schreib- und Überprüfungsagenten klare Verantwortlichkeiten zu und sorgen so für eine Trennung der Anliegen. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
LOGS = """timestamp,service,standing,latency_ms,error
2025-12-18T12:00:00Z,checkout,200,180,false
2025-12-18T12:00:05Z,checkout,500,900,true
2025-12-18T12:00:10Z,auth,200,120,false
2025-12-18T12:00:12Z,checkout,502,1100,true
2025-12-18T12:00:20Z,search,200,140,false
2025-12-18T12:00:25Z,checkout,500,950,true
"""
def msg_text(m: Msg) -> str:
blocks = m.get_content_blocks("textual content")
if blocks is None:
return ""
if isinstance(blocks, str):
return blocks
if isinstance(blocks, record):
return "n".be part of(str(x) for x in blocks)
return str(blocks)Wir stellen Beispielprotokolldaten und eine Hilfsfunktion vor, die Agentenausgaben in sauberen Textual content normalisiert. Wir stellen sicher, dass nachgelagerte Agenten frühere Antworten ohne Formatprobleme sicher nutzen und verfeinern können. Der Schwerpunkt liegt darauf, die Kommunikation zwischen Agenten strong und vorhersehbar zu machen. Schauen Sie sich das an VOLLSTÄNDIGE CODES hier.
async def run_demo(user_request: str):
route_msg = await router(Msg("consumer", user_request, "consumer"), structured_model=Route)
lane = (route_msg.metadata or {}).get("lane", "unknown")
if lane == "triage":
first = await triager(Msg("consumer", user_request, "consumer"))
elif lane == "evaluation":
first = await analyst(Msg("consumer", user_request + "nnLogs:n" + LOGS, "consumer"))
elif lane == "report":
draft = await author(Msg("consumer", user_request, "consumer"))
first = await reviewer(Msg("consumer", "Assessment and enhance:nn" + msg_text(draft), "consumer"))
else:
first = Msg("system", "Couldn't route request.", "system")
async with MsgHub(
individuals=(triager, analyst, author, reviewer),
announcement=Msg("Host", "Refine the ultimate reply collaboratively.", "assistant"),
):
await sequential_pipeline((triager, analyst, author, reviewer))
return {"route": route_msg.metadata, "initial_output": msg_text(first)}
consequence = await run_demo(
"We see repeated 5xx errors in checkout. Classify severity, analyze logs, and produce an incident report."
)
print(json.dumps(consequence, indent=2))Wir orchestrieren den gesamten Workflow, indem wir die Anfrage weiterleiten, den entsprechenden Agenten ausführen und mithilfe eines Nachrichten-Hubs eine kollaborative Verfeinerungsschleife ausführen. Wir koordinieren mehrere Agenten nacheinander, um die Endausgabe zu verbessern, bevor wir sie an den Benutzer zurückgeben. Es vereint alle früheren Komponenten in einer zusammenhängenden, durchgängigen Agenten-Pipeline.
Abschließend haben wir gezeigt, wie AgentScope es uns ermöglicht, robuste, modulare und kollaborative Agentensysteme zu entwerfen, die über Interaktionen mit nur einer Eingabeaufforderung hinausgehen. Wir haben Aufgaben dynamisch weitergeleitet, Instruments nur bei Bedarf aufgerufen und die Ergebnisse durch die Koordination mehrerer Agenten verfeinert – und das alles in einem sauberen und reproduzierbaren Colab-Setup. Dieses Muster veranschaulicht, wie wir von einfachen Agentenexperimenten zu produktionsähnlichen Argumentationspipelines skalieren und gleichzeitig Klarheit, Kontrolle und Erweiterbarkeit in unseren Agenten-KI-Anwendungen bewahren können.
Schauen Sie sich das an VOLLSTÄNDIGE CODES hier. Sie können uns auch gerne weiter folgen Twitter und vergessen Sie nicht, bei uns mitzumachen 100.000+ ML SubReddit und Abonnieren Unser E-newsletter. Warten! Bist du im Telegram? Jetzt können Sie uns auch per Telegram kontaktieren.
Asif Razzaq ist CEO von Marktechpost Media Inc.. Als visionärer Unternehmer und Ingenieur setzt sich Asif dafür ein, das Potenzial der künstlichen Intelligenz für das soziale Wohl zu nutzen. Sein jüngstes Unterfangen ist die Einführung einer Medienplattform für künstliche Intelligenz, Marktechpost, die sich durch eine ausführliche Berichterstattung über maschinelles Lernen und Deep-Studying-Nachrichten auszeichnet, die sowohl technisch fundiert als auch für ein breites Publikum leicht verständlich ist. Die Plattform verfügt über mehr als 2 Millionen monatliche Aufrufe, was ihre Beliebtheit beim Publikum verdeutlicht.