A Coding Guide to Design and Orchestrate Advanced ReAct-Based Multi-Agent Workflows with AgentScope and OpenAI
These articles are AI-generated summaries. Please check the original sources for full details.
A Coding Guide to Design and Orchestrate Advanced ReAct-Based Multi-Agent Workflows with AgentScope and OpenAI
This tutorial introduces a method for building an advanced multi-agent incident response system using AgentScope, a framework for orchestrating ReAct agents. The system connects agents – routing, triage, analysis, writing, and review – through structured routing and a shared message hub, leveraging OpenAI models and lightweight tool calling.
Current AI agent frameworks often require complex infrastructure and brittle code for multi-agent coordination; AgentScope provides a streamlined Python-centric approach for composing these workflows. Scaling agentic systems without increasing operational overhead is a significant challenge, and this approach aims to reduce that friction.
Key Insights
- AgentScope v0.1.5: Released in late 2025, offering a Python-native approach to multi-agent systems.
- ReAct Agents: Combining reasoning and acting allows agents to dynamically utilize tools and adapt to changing information, improving response quality.
- Stripe & Coinbase: Companies utilizing Temporal, a similar workflow orchestration tool, demonstrate the industry demand for robust agentic systems.
Working Example
!pip -q install "agentscope>=0.1.5" pydantic nest_asyncio
import os, json, re
from getpass import getpass
from typing import Literal
from pydantic import BaseModel, Field
import nest_asyncio
nest_asyncio.apply()
from agentscope.agent import ReActAgent
from agentscope.message import Msg, TextBlock
from agentscope.model import OpenAIChatModel
from agentscope.formatter import OpenAIChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.tool 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")
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()))
d = re.findall(r"[a-z0-9]+", d.lower())
return sum(1 for w in d if w in q) / max(1, len(d))
async def search_runbook(query: str, top_k: int = 2) -> ToolResponse:
ranked = sorted(RUNBOOK, key=lambda r: _score(query, r["title"] + r["text"]), reverse=True)[: max(1, int(top_k))]
text = "\n\n".join(f"[{r['id']}] {r['title']}\n{r['text']}" for r in ranked)
return ToolResponse(content=[TextBlock(type="text", text=text)])
toolkit = Toolkit()
toolkit.register_tool_function(search_runbook)
toolkit.register_tool_function(execute_python_code)
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", "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,
)
analyst = ReActAgent(
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(),
)
Practical Applications
- Incident Management: Automating initial triage and analysis of system alerts, as demonstrated in the example.
- Pitfall: Over-reliance on a single agent for complex tasks can lead to errors; a well-defined multi-agent workflow with clear roles is crucial.
References:
Continue reading
Next article
AI Interview Series #5: Prompt Caching
Related Content
Building Production-Ready Agentic Workflows with AgentScope and ReAct Agents
Learn to build production-ready AgentScope workflows using ReAct agents, custom toolkits, and Pydantic for structured outputs. This tutorial demonstrates how to orchestrate multi-agent debates and concurrent analysis pipelines using OpenAI models to achieve high-fidelity reasoning and automated tool execution for enterprise-grade AI applications.
Building Multi-Agent Data Analysis Pipelines with Google ADK
Learn to build a modular multi-agent system using Google ADK to automate data ingestion, statistical modeling, and visualization in Python. This tutorial demonstrates orchestrating five specialized agents to perform Shapiro-Wilk tests and ANOVA, significantly reducing manual analysis time in production-grade pipelines.
Building an Autonomous Wet-Lab Protocol Planner with Salesforce CodeGen for Agentic Experiment Design and Safety Optimization
A detailed tutorial on creating an AI-driven system for automating lab protocols, reagent validation, and safety checks using Salesforce CodeGen and Python.