refactor: format dyanmic fan out fan in sample better.

boyangsvl · boyangsvl · commit c1bdd9bc3bc1 · 2026-04-10T11:33:08.000-07:00
Change-Id: Ifbab67f85122a1400f67df189b0bad24a3eb848b
diff --git a/contributing/workflow_samples/dynamic_fan_out_fan_in/agent.py b/contributing/workflow_samples/dynamic_fan_out_fan_in/agent.py
@@ -13,44 +13,56 @@
 # limitations under the License.
 
 from __future__ import annotations
-from google.adk import Agent, Context, Event, Workflow
-from google.adk.workflow import node
+
 import asyncio
 
+from google.adk import Agent
+from google.adk import Context
+from google.adk import Event
+from google.adk import Workflow
+from google.adk.workflow import node
+
 # Worker agent to generate a headline for a single topic
 generator = Agent(
     name="generator",
     model="gemini-2.5-flash",
-    instruction="Write a catchy headline about the topic provided in the user message.",
+    instruction=(
+        "Write a catchy one-line headline about the topic provided in the user"
+        " message."
+    ),
 )
 
+
 @node(rerun_on_resume=True)
 async def orchestrator(ctx: Context, node_input: str) -> str:
-    """Orchestrator node that performs dynamic fan-out and fan-in."""
-    # Split input comma-separated string into topics
-    topics = [t.strip() for t in node_input.split(",") if t.strip()]
-    yield Event(message=f"Processing {len(topics)} topics in parallel.")
-
-    # Fan-out: Schedule a dynamic node for each topic
-    tasks = []
-    for i, topic in enumerate(topics):
-        tasks.append(
-            ctx.run_node(
-                generator,
-                node_input=topic,
-                sub_branch=f"branch_{i}"  # Isolate events to prevent context mess-up
-            )
+  """Orchestrator node that performs dynamic fan-out and fan-in."""
+  # Split input comma-separated string into topics
+  topics = [t.strip() for t in node_input.split(",") if t.strip()]
+  yield Event(message=f"Processing {len(topics)} topics in parallel.")
+
+  # Fan-out: Schedule a dynamic node for each topic
+  tasks = []
+  for i, topic in enumerate(topics):
+    tasks.append(
+        ctx.run_node(
+            generator,
+            node_input=topic,
+            is_parallel=True,
         )
+    )
+
+  # Wait for all tasks to complete
+  results = await asyncio.gather(*tasks)
 
-    # Wait for all tasks to complete
-    results = await asyncio.gather(*tasks)
+  # Fan-in: Aggregate results
+  aggregated = "### Aggregated Headlines\n\n"
+  aggregated += "| Topic | Headline |\n"
+  aggregated += "| :--- | :--- |\n"
+  for topic, headline in zip(topics, results):
+    aggregated += f"| {topic} | {headline} |\n"
 
-    # Fan-in: Aggregate results
-    aggregated = "Aggregated Headlines:\n"
-    for topic, headline in zip(topics, results):
-        aggregated += f"- Topic [{topic}]: {headline}\n"
+  yield Event(message=aggregated)
 
-    yield Event(output=aggregated)
 
 root_agent = Workflow(
     name="dynamic_fan_out_fan_in",
