{"id":222,"date":"2026-01-05T15:54:21","date_gmt":"2026-01-05T07:54:21","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=222"},"modified":"2026-01-05T15:54:21","modified_gmt":"2026-01-05T07:54:21","slug":"a-coding-guide-to-design-and-orchestrate-advanced-react-based-multi-agent-workflows-with-agentscope-and-openai","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=222","title":{"rendered":"A Coding Guide to Design and Orchestrate Advanced ReAct-Based Multi-Agent Workflows with AgentScope and OpenAI"},"content":{"rendered":"

In this tutorial, we build an advanced multi-agent incident response system using AgentScope<\/strong><\/a>. We orchestrate multiple ReAct agents, each with a clearly defined role such as routing, triage, analysis, writing, and review, and connect them through structured routing and a shared message hub. By integrating OpenAI models, lightweight tool calling, and a simple internal runbook, we demonstrate how complex, real-world agentic workflows can be composed in pure Python without heavy infrastructure or brittle glue code. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip -q install \"agentscope>=0.1.5\" pydantic nest_asyncio\n\n\nimport os, json, re\nfrom getpass import getpass\nfrom typing import Literal\nfrom pydantic import BaseModel, Field\nimport nest_asyncio\nnest_asyncio.apply()\n\n\nfrom agentscope.agent import ReActAgent\nfrom agentscope.message import Msg, TextBlock\nfrom agentscope.model import OpenAIChatModel\nfrom agentscope.formatter import OpenAIChatFormatter\nfrom agentscope.memory import InMemoryMemory\nfrom agentscope.tool import Toolkit, ToolResponse, execute_python_code\nfrom agentscope.pipeline import MsgHub, sequential_pipeline\n\n\nif not os.environ.get(\"OPENAI_API_KEY\"):\n   os.environ[\"OPENAI_API_KEY\"] = getpass(\"Enter OPENAI_API_KEY (hidden): \")\n\n\nOPENAI_MODEL = os.environ.get(\"OPENAI_MODEL\", \"gpt-4o-mini\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We set up the execution environment and install all required dependencies so the tutorial runs reliably on Google Colab. We securely load the OpenAI API key and initialize the core AgentScope components that will be shared across all agents. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
RUNBOOK = [\n   {\"id\": \"P0\", \"title\": \"Severity Policy\", \"text\": \"P0 critical outage, P1 major degradation, P2 minor issue\"},\n   {\"id\": \"IR1\", \"title\": \"Incident Triage Checklist\", \"text\": \"Assess blast radius, timeline, deployments, errors, mitigation\"},\n   {\"id\": \"SEC7\", \"title\": \"Phishing Escalation\", \"text\": \"Disable account, reset sessions, block sender, preserve evidence\"},\n]\n\n\ndef _score(q, d):\n   q = set(re.findall(r\"[a-z0-9]+\", q.lower()))\n   d = re.findall(r\"[a-z0-9]+\", d.lower())\n   return sum(1 for w in d if w in q) \/ max(1, len(d))\n\n\nasync def search_runbook(query: str, top_k: int = 2) -> ToolResponse:\n   ranked = sorted(RUNBOOK, key=lambda r: _score(query, r[\"title\"] + r[\"text\"]), reverse=True)[: max(1, int(top_k))]\n   text = \"nn\".join(f\"[{r['id']}] {r['title']}n{r['text']}\" for r in ranked)\n   return ToolResponse(content=[TextBlock(type=\"text\", text=text)])\n\n\ntoolkit = Toolkit()\ntoolkit.register_tool_function(search_runbook)\ntoolkit.register_tool_function(execute_python_code)<\/code><\/pre>\n<\/div>\n<\/div>\n
We define a lightweight internal runbook and implement a simple relevance-based search tool over it. We register this function along with a Python execution tool, enabling agents to retrieve policy knowledge or compute results dynamically. It demonstrates how we augment agents with external capabilities beyond pure language reasoning. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def make_model():\n   return OpenAIChatModel(\n       model_name=OPENAI_MODEL,\n       api_key=os.environ[\"OPENAI_API_KEY\"],\n       generate_kwargs={\"temperature\": 0.2},\n   )\n\n\nclass Route(BaseModel):\n   lane: Literal[\"triage\", \"analysis\", \"report\", \"unknown\"] = Field(...)\n   goal: str = Field(...)\n\n\nrouter = ReActAgent(\n   name=\"Router\",\n   sys_prompt=\"Route the request to triage, analysis, or report and output structured JSON only.\",\n   model=make_model(),\n   formatter=OpenAIChatFormatter(),\n   memory=InMemoryMemory(),\n)\n\n\ntriager = ReActAgent(\n   name=\"Triager\",\n   sys_prompt=\"Classify severity and immediate actions using runbook search when useful.\",\n   model=make_model(),\n   formatter=OpenAIChatFormatter(),\n   memory=InMemoryMemory(),\n   toolkit=toolkit,\n)\n\n\nanalyst = ReActAgent(\n   name=\"Analyst\",\n   sys_prompt=\"Analyze logs and compute summaries using python tool when helpful.\",\n   model=make_model(),\n   formatter=OpenAIChatFormatter(),\n   memory=InMemoryMemory(),\n   toolkit=toolkit,\n)\n\n\nwriter = ReActAgent(\n   name=\"Writer\",\n   sys_prompt=\"Write a concise incident report with clear structure.\",\n   model=make_model(),\n   formatter=OpenAIChatFormatter(),\n   memory=InMemoryMemory(),\n)\n\n\nreviewer = ReActAgent(\n   name=\"Reviewer\",\n   sys_prompt=\"Critique and improve the report with concrete fixes.\",\n   model=make_model(),\n   formatter=OpenAIChatFormatter(),\n   memory=InMemoryMemory(),\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We construct multiple specialized ReAct agents and a structured router that decides how each user request should be handled. We assign clear responsibilities to the triage, analysis, writing, and review agents, ensuring separation of concerns. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
LOGS = \"\"\"timestamp,service,status,latency_ms,error\n2025-12-18T12:00:00Z,checkout,200,180,false\n2025-12-18T12:00:05Z,checkout,500,900,true\n2025-12-18T12:00:10Z,auth,200,120,false\n2025-12-18T12:00:12Z,checkout,502,1100,true\n2025-12-18T12:00:20Z,search,200,140,false\n2025-12-18T12:00:25Z,checkout,500,950,true\n\"\"\"\n\n\ndef msg_text(m: Msg) -> str:\n   blocks = m.get_content_blocks(\"text\")\n   if blocks is None:\n       return \"\"\n   if isinstance(blocks, str):\n       return blocks\n   if isinstance(blocks, list):\n       return \"n\".join(str(x) for x in blocks)\n   return str(blocks)<\/code><\/pre>\n<\/div>\n<\/div>\n
We introduce sample log data and a utility function that normalizes agent outputs into clean text. We ensure that downstream agents can safely consume and refine earlier responses without format issues. It focuses on making inter-agent communication robust and predictable. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
async def run_demo(user_request: str):\n   route_msg = await router(Msg(\"user\", user_request, \"user\"), structured_model=Route)\n   lane = (route_msg.metadata or {}).get(\"lane\", \"unknown\")\n\n\n   if lane == \"triage\":\n       first = await triager(Msg(\"user\", user_request, \"user\"))\n   elif lane == \"analysis\":\n       first = await analyst(Msg(\"user\", user_request + \"nnLogs:n\" + LOGS, \"user\"))\n   elif lane == \"report\":\n       draft = await writer(Msg(\"user\", user_request, \"user\"))\n       first = await reviewer(Msg(\"user\", \"Review and improve:nn\" + msg_text(draft), \"user\"))\n   else:\n       first = Msg(\"system\", \"Could not route request.\", \"system\")\n\n\n   async with MsgHub(\n       participants=[triager, analyst, writer, reviewer],\n       announcement=Msg(\"Host\", \"Refine the final answer collaboratively.\", \"assistant\"),\n   ):\n       await sequential_pipeline([triager, analyst, writer, reviewer])\n\n\n   return {\"route\": route_msg.metadata, \"initial_output\": msg_text(first)}\n\n\nresult = await run_demo(\n   \"We see repeated 5xx errors in checkout. Classify severity, analyze logs, and produce an incident report.\"\n)\nprint(json.dumps(result, indent=2))<\/code><\/pre>\n<\/div>\n<\/div>\n
We orchestrate the full workflow by routing the request, executing the appropriate agent, and running a collaborative refinement loop using a message hub. We coordinate multiple agents in sequence to improve the final output before returning it to the user. It brings together all earlier components into a cohesive, end-to-end agentic pipeline.<\/p>\n
In conclusion, we showed how AgentScope enables us to design robust, modular, and collaborative agent systems that go beyond single-prompt interactions. We routed tasks dynamically, invoked tools only when needed, and refined outputs through multi-agent coordination, all within a clean and reproducible Colab setup. This pattern illustrates how we can scale from simple agent experiments to production-style reasoning pipelines while maintaining clarity, control, and extensibility in our agentic AI applications.<\/p>\n
\n
Check out the\u00a0FULL CODES here<\/a><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post A Coding Guide to Design and Orchestrate Advanced ReAct-Based Multi-Agent Workflows with AgentScope and OpenAI<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build an …<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-222","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/222","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=222"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/222\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=222"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=222"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=222"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}