![[Trivia] Extracting only the final agent's response in Strands Agents Swarm](https://images.ctfassets.net/ct0aopd36mqt/4o8n2qvRpfnsx0yGmKgDZr/1b05f3211bdf64deb5322bf5be42b202/StrandsAgents.png?w=3840&fm=webp)
[Trivia] Extracting only the final agent's response in Strands Agents Swarm
This page has been translated by machine translation. View original
Introduction
Hello, I'm Jinno from the Consulting Division, currently practicing my driving.
I'd love to get confident at parking as soon as possible.
But enough about that — are you using Swarm in Strands Agents?
Some of you may be using this pattern to implement multi-agent systems.
Recently, when I used Swarm, the final answer ended up mixing in intermediate responses from all the agents. If you only want to show users the final report compiled by the last agent, having everything visible is a bit annoying.
This is a small tip for that situation — I'll show you how to extract only the final agent's output from Swarm stream events.
Prerequisites
- Python 3.12
- strands-agents 1.47.0
The Swarm configuration assumes 3 agents handing off in sequence: reception (front) → research (researcher) → report creation (synthesizer).
swarm = Swarm(
[front_agent, researcher, synthesizer],
entry_point=front_agent,
max_handoffs=10,
max_iterations=15,
)
For how to use Swarm itself and safety parameters (such as max_handoffs and execution_timeout), please refer to the official documentation.
The Cause and Solution Right Away
The cause is simple: the multiagent_node_stream events emitted by stream_async() output text from all agents, so concatenating them mixes together everyone's conversations.
The solution is to accumulate text into a node-keyed dict using the node_id held by multiagent_node_stream events, and then at the end adopt only the report creation (synthesizer) portion.
Please note this is only effective in cases like this one where you want to adopt only the final agent's response.
Let's look at the specific implementation.
The Challenge I Faced
First, here is what I initially implemented. Note that if your application wants to show the collaborative process directly to users, this form works perfectly fine.
The case I'm assuming here is specifically when you only want to show the final report.
final_text = ""
async for event in swarm.stream_async(user_input):
if event.get("type") == "multiagent_node_stream":
inner = event.get("event", {})
if isinstance(inner, dict) and "data" in inner:
text = inner["data"] if isinstance(inner["data"], str) else ""
if text:
final_text += text
This simply concatenates multiagent_node_stream text into final_text. The basics of streaming with stream_async() are covered in this official documentation.
It seems fine at first glance, but the response you get looks like this.
I have received the problem report from the user. I will structure it as follows.
1. Problem Summary
...
Now I will hand off to researcher for detailed investigation.
Starting investigation. I will collect information in parallel first.
...
（The actual report finally starts here）
The reception agent's structuring, the handoff remarks, the research commentary — it's all included. Quite immersive.
While this is correct behavior for the internal collaborative process of Swarm, there are cases where only the final report is sufficient as a response to the user, unless you specifically want to show the real-time progress.
Solution
Since multiagent_node_stream events hold node_id (which agent is speaking) as a key at the top level, we use this to accumulate text into a per-node dict. Additionally, by keeping track of the "last node that spoke" as last_text_node each time text is accumulated, it can be used for the fallback described later.
node_texts: dict[str, str] = {}
last_text_node = ""
async for event in swarm.stream_async(user_input):
if event.get("type") == "multiagent_node_stream":
node_id = event.get("node_id", "")
inner = event.get("event", {})
if isinstance(inner, dict) and "data" in inner:
text = inner["data"] if isinstance(inner["data"], str) else ""
if text:
node_texts[node_id] = node_texts.get(node_id, "") + text
last_text_node = node_id
# Adopt only the output from the final report agent (synthesizer)
final_text = node_texts.get("synthesizer", "").strip()
# Fallback in case execution ended before reaching synthesizer
if not final_text and last_text_node:
final_text = node_texts.get(last_text_node, "").strip()
Text is accumulated in node_texts with the agent name as the key, and at the end only the synthesizer's portion is extracted. Note that being able to extract using synthesizer as the key assumes a configuration where the final report agent is fixed, as in this case.
The fallback is also important. Swarm can end before reaching the final agent due to timeouts or exceeding max_handoffs, and in that case using the synthesizer key would result in an empty response. By substituting with the output from the last node that spoke, at least the partial investigation results up to that point can be returned.
Verification
When running the improved version with the query "The list screen of the web app suddenly became slow since yesterday. Please investigate and summarize the cause," the response contained only the synthesizer's report.
Note that Agent outputs speech to standard output by default (PrintingCallbackHandler), so during verification I specified callback_handler=None for each Agent. This only applies when checking via CLI and has nothing to do with the issue of mixed responses. For more details on callback_handler behavior, please see here.
nodes: {'front': 575, 'researcher': 3375, 'synthesizer': 1886}
============================================================
Final answer (synthesizer report only)
============================================================
# Investigation Results Summary
## Problem Overview
- **Symptoms**: Response delay of over 10 seconds occurring on the web application list screen display
- **Time of occurrence**: Response delay suddenly started yesterday
...
Looking at the logs, front and researcher are having their conversations, but only the synthesizer's portion is extracted in the response. The reception's structuring and handoff remarks are gone, and only the report you want to show users has been extracted!
By the way, if you want to show progress along the way, returning node_texts and handoff history in separate fields could be useful for displaying the process on the UI side.
The full code used for verification in this article is folded below.
Full code used for verification
Here is the Swarm definition.
"""Swarm definition (3-agent configuration: front → researcher → synthesizer)"""
from strands import Agent
from strands.multiagent import Swarm
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
def create_swarm() -> Swarm:
front_agent = Agent(
name="front",
model=MODEL_ID,
callback_handler=None, # Stop automatic output to standard output (default behavior)
system_prompt=(
"You are a reception agent."
"Structure the user's inquiry into the form of 'problem summary', 'related information', and 'what you want investigated',"
"output that content, and then hand off to researcher."
"Do not perform any investigation or provide answers yourself."
),
)
researcher = Agent(
name="researcher",
model=MODEL_ID,
callback_handler=None,
system_prompt=(
"You are a research agent."
"Based on the content you receive, identify possible causes and related information using your knowledge."
"You may also output your investigation process and thoughts."
"Once the investigation is complete, hand off to synthesizer."
),
)
synthesizer = Agent(
name="synthesizer",
model=MODEL_ID,
callback_handler=None,
system_prompt=(
"You are a report creation agent."
"Summarize the investigation results so far as a final report to show the user."
"Start with the heading 'Investigation Results Summary' and output only a concise, easy-to-read report."
"No handoff to other agents is necessary."
),
)
return Swarm(
[front_agent, researcher, synthesizer],
entry_point=front_agent,
max_handoffs=10,
max_iterations=15,
)
USER_INPUT = (
"The response of the web app suddenly became slow since yesterday."
"In particular, the list screen takes over 10 seconds to display. The database is RDS (PostgreSQL)."
"Please investigate and summarize the possible causes."
)
Here is an example of concatenating all agents' speech as-is.
"""Example of concatenating all agents' speech as-is
If you want to show the collaborative process (reception structuring,
handoff remarks, research commentary) to users, this form is fine.
If you only want to show the final report, use the form in final_only.py.
"""
import asyncio
from swarm_def import USER_INPUT, create_swarm
async def main() -> None:
swarm = create_swarm()
final_text = ""
async for event in swarm.stream_async(USER_INPUT):
if event.get("type") == "multiagent_node_stream":
inner = event.get("event", {})
if isinstance(inner, dict) and "data" in inner:
text = inner["data"] if isinstance(inner["data"], str) else ""
if text:
final_text += text
print("=" * 60)
print("Response (all agents' speech concatenated as-is)")
print("=" * 60)
print(final_text)
if __name__ == "__main__":
asyncio.run(main())
Here is the improved version that extracts only the final agent's output.
"""Improved version: accumulate text by node and adopt only the final agent's (synthesizer) portion
Since multiagent_node_stream events hold node_id at the top level,
use it as a key to accumulate into a per-node dict.
If execution ends without reaching synthesizer, substitute with the last node that spoke.
"""
import asyncio
import logging
from swarm_def import USER_INPUT, create_swarm
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)
FINAL_NODE = "synthesizer"
async def main() -> None:
swarm = create_swarm()
node_texts: dict[str, str] = {}
last_text_node = ""
async for event in swarm.stream_async(USER_INPUT):
if event.get("type") == "multiagent_node_stream":
node_id = event.get("node_id", "")
inner = event.get("event", {})
if isinstance(inner, dict) and "data" in inner:
text = inner["data"] if isinstance(inner["data"], str) else ""
if text:
node_texts[node_id] = node_texts.get(node_id, "") + text
last_text_node = node_id
# Adopt only the output from the final report agent (synthesizer)
final_text = node_texts.get(FINAL_NODE, "").strip()
# Fallback in case execution ended before reaching synthesizer
if not final_text and last_text_node:
final_text = node_texts.get(last_text_node, "").strip()
# For debugging: how many characters each agent spoke
logger.info(f"nodes: { {k: len(v) for k, v in node_texts.items()} }")
print("=" * 60)
print("Final answer (synthesizer report only)")
print("=" * 60)
print(final_text)
if __name__ == "__main__":
asyncio.run(main())
Run as follows (using Bedrock's Haiku 4.5).
uv init
uv add "strands-agents==1.47.0"
AWS_REGION=us-west-2 uv run stream_all.py
AWS_REGION=us-west-2 uv run final_only.py
Conclusion
Whether to display the entire Swarm collaborative process or show only the final answer depends on the application, but I tried this out because accumulating events by node allows you to handle either case.
I hope this article is helpful in some way.
Thank you for reading to the end!