This tutorial shows you how to build a multi-agent system for incident response. AgentScope. We connect multiple ReAct Agents, with each agent having a role clearly defined, such as triage, routing, analysis, review and writing. This is done through a structured messaging hub and routed routing. Through the use of OpenAI models and lightweight tool-calling, as well as a simple runbook within our own company, we show how real agentic workflows are composed using pure Python, without a lot of infrastructure. See the FULL CODES here.
!pip -q install "agentscope>=0.1.5" pydantic nest_asyncio
import os, json, re
From getpass import Getpass
Literal import is the opposite of typing.
BaseModel and Field
import nest_asyncio
nest_asyncio.apply()
ReActAgent is imported by agentscope.agent
TextBlock imports Msg from Agentscope.message
OpenAIChat Model can be imported from Agentscope.model
OpenAIChatFormatter is imported by agentscope.formatter.
InMemoryMemory can be imported from agentscope.memory
From agentscope.tool, import Toolkit ToolResponse execute_python_code
Import MsgHub and sequential_pipeline from agentscope.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")Install all dependencies and set up an execution environment to ensure the tutorial will run on Google Colab. The OpenAI API is loaded securely and the AgentScope core components are initialized. Take a look at the FULL CODES here.
RUNBOOK [
{"id": "P0", "title": "Severity Policy", "text": "P0 critical outage, P1 major degradation, P2 minor issue"},
{"id": "IR1", "title": "Incident Triage Checklist", "text": "Assess blast radius, timeline, deployments, errors, mitigation"},
{"id": "SEC7", "title": "Phishing Escalation", "text": "Disable account, reset sessions, block sender, preserve evidence"},
]
Def _score (q, d).
q = set(re.findall(r"[a-z0-9]+", q.lower()))
Findall(r) = d"[a-z0-9]+", d.lower())
The return sum is 1 for each w if it's in d. Max (1, len(d), d)
async def search_runbook(query: str, top_k: int = 2) -> ToolResponse:
Ranking = sort(RUNBOOK key=lambda, r: score(query,r)["title"] +["text"]), reverse=True)[: max(1, int(top_k))]
Text = "nn".join(f"[{r['id']}] {r['title']}n{r['text']}" For r, see ranked
return ToolResponse(content=[TextBlock(type="text", text=text)])
Toolkit()
toolkit.register_tool_function(search_runbook)
toolkit.register_tool_function(execute_python_code)A lightweight internal runbook will be defined and a relevance-based tool implemented over top. This function is registered along with an execution tool in Python, which allows agents to dynamically retrieve policy information or calculate results. This demonstrates the ability to augment agents beyond simple language reasoning. Visit the FULL CODES here.
Def Make_Model():
OpenAIChatModel (
model_name=OPENAI_MODEL,
api_key=os.environ["OPENAI_API_KEY"],
generate_kwargs={"temperature": 0.2},
)
Class Route (BaseModel:
lane["triage", "analysis", "report", "unknown"] = Field(...)
Goal: str = Field (...)
Router = ReactAgent
name="Router",
sys_prompt="Route the request to triage, analysis, or report and output structured JSON only.",
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
)
triager = ReActAgent(
name="Triager",
sys_prompt="Classify severity and immediate actions using runbook search when useful.",
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
toolkit=toolkit,
)
ReActAgent (analyst) =
name="Analyst",
sys_prompt="Analyze logs and compute summaries using python tool when helpful.",
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
toolkit=toolkit,
)
"writer = ReActAgent"
name="Writer",
sys_prompt="Write a concise incident report with clear structure.",
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
)
Reviewer = ReActAgent
name="Reviewer",
sys_prompt="Critique and improve the report with concrete fixes.",
model=make_model(),
formatter=OpenAIChatFormatter(),
memory=InMemoryMemory(),
)
Multiple ReAct agents are created, and we use a structured routing system to decide how every user’s request is handled. To ensure separation of concerns, we assign clear roles to each agent, including the writing, reviewing, triaging, and analysis. Visit the FULL CODES here.
LOGS """timestamp,service,status,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("text")
If blocks is None
You can return to your original language by clicking here. ""
if isinstance()(blocks str)
Return blocks
If the block or list is instance:
You can return to your original language by clicking here. "n"Joining blocks together using str(x).
Return str (blocks)Sample log data is introduced, as well as a utility that converts the output of agents into clean text. We make sure that agents downstream can consume earlier responses and refine them without any format problems. This focuses on making agent communication predictable and robust. Visit the FULL CODES here.
async def run_demo(user_request: str):
route_msg = await router(Msg("user", user_request, "user"), structured_model=Route)
lane = (route_msg.metadata or {}).get("lane", "unknown")
if lane == "triage":
first = await triager(Msg("user", user_request, "user"))
elif lane == "analysis":
first = await analyst(Msg("user", user_request + "nnLogs:n" + LOGS, "user"))
elif lane == "report":
draft = await writer(Msg("user", user_request, "user"))
first = await reviewer(Msg("user", "Review and improve:nn" + msg_text(draft), "user"))
else:
First = Msg"system", "Could not route request.", "system")
MsgHub is in sync with MsgHubReturn
participants=[triager, analyst, writer, reviewer],
announcement=Msg("Host", "Refine the final answer collaboratively.", "assistant"),
):
await sequential_pipeline([triager, analyst, writer, reviewer])
return {"route": route_msg.metadata, "initial_output": msg_text(first)}
result = await run_demo(
"We see repeated 5xx errors in checkout. Classify severity, analyze logs, and produce an incident report."
)
print(json.dumps(result, indent=2))By routing requests, running the agent and executing a refinement cycle collaboratively using a hub, we orchestrate a full workflow. Before returning the output to the user, we coordinate the execution of multiple agents. The pipeline brings all components together into one cohesive end-to-end system.
We concluded by showing how AgentScope allows us to create robust and modular agent systems, which go beyond simple prompt interactions. The Colab system allowed us to route tasks in a dynamic way, use tools when necessary, and refine outputs using multi-agent coordinaton. This pattern shows how to scale agent experiments up to reasoning pipelines of production style while maintaining control, clarity and extensibility.
Click here to find out more FULL CODES here. Also, feel free to follow us on Twitter Don’t forget about our 100k+ ML SubReddit Subscribe now our Newsletter. Wait! What? now you can join us on telegram as well.
Asif Razzaq serves as the CEO at Marktechpost Media Inc. As an entrepreneur, Asif has a passion for harnessing Artificial Intelligence to benefit society. Marktechpost was his most recent venture. This platform, a Media Platform for Artificial Intelligence, is distinguished by its technical soundness and ease of understanding, while also providing in-depth reporting on machine learning and deeplearning news. This platform has over 2,000,000 monthly views which shows its popularity.
