Update aggregation examples

Author: pamelafox

…mples/workflow_aggregator_llm_summary.py examples/workflow_aggregator_summary.pyexamples/workflow_aggregator_llm_summary.py renamed to examples/workflow_aggregator_summary.py examples/workflow_aggregator_llm_summary.py renamed to examples/workflow_aggregator_summary.py

@@ -7,14 +7,16 @@
 Aggregation technique: LLM synthesis (Agent as post-processor).
 
 Run:
-    uv run examples/workflow_aggregator_llm_summary.py
-    uv run examples/workflow_aggregator_llm_summary.py --devui  (opens DevUI at http://localhost:8101)
+    uv run examples/workflow_aggregator_summary.py
+    uv run examples/workflow_aggregator_summary.py --devui  (opens DevUI at http://localhost:8101)
 """
 
 import asyncio
 import os
 import sys
 
+from typing import Never
+
 from agent_framework import Agent, AgentExecutorResponse, Executor, WorkflowBuilder, WorkflowContext, handler
 from agent_framework.openai import OpenAIChatClient
 from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
@@ -53,20 +55,32 @@ async def dispatch(self, prompt: str, ctx: WorkflowContext[str]) -> None:
         await ctx.send_message(prompt)
 
 
-class FormatBranchResults(Executor):
-    """Fan-in collector that formats branch outputs into a single prompt."""
+class SummarizerExecutor(Executor):
+    """Fan-in aggregator that synthesizes expert outputs via a wrapped Agent."""
+
+    agent: Agent
+
+    def __init__(self, client: OpenAIChatClient, id: str = "Summarizer"):
+        super().__init__(id=id)
+        self.agent = Agent(
+            client=client,
+            name=id,
+            instructions=(
+                "You receive analysis from three domain experts (researcher, marketer, legal). "
+                "Synthesize their combined insights into a concise 3-sentence executive brief "
+                "that a CEO could read in 30 seconds. Do not repeat the raw analysis."
+            ),
+        )
 
     @handler
-    async def format(
-        self,
-        results: list[AgentExecutorResponse],
-        ctx: WorkflowContext[str],
-    ) -> None:
-        """Combine expert outputs into labeled sections for the summarizer."""
+    async def run(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
+        """Format branch outputs and feed them to the summarizer Agent."""
         sections = []
         for result in results:
             sections.append(f"[{result.executor_id}]\n{result.agent_response.text}")
-        await ctx.send_message("\n\n---\n\n".join(sections))
+        combined = "\n\n---\n\n".join(sections)
+        response = await self.agent.run(combined)
+        await ctx.yield_output(response.text)
 
 
 dispatcher = DispatchPrompt(id="dispatcher")
@@ -101,19 +115,9 @@ async def format(
     ),
 )
 
-formatter = FormatBranchResults(id="formatter")
-
-# The summarizer Agent is the final node — it receives the formatted expert
-# outputs and synthesizes them into a concise executive brief.
-summarizer = Agent(
-    client=client,
-    name="Summarizer",
-    instructions=(
-        "You receive analysis from three domain experts (researcher, marketer, legal). "
-        "Synthesize their combined insights into a concise 3-sentence executive brief "
-        "that a CEO could read in 30 seconds. Do not repeat the raw analysis."
-    ),
-)
+# The summarizer Executor wraps an Agent to handle fan-in directly —
+# it formats the collected expert outputs and synthesizes a brief.
+summarizer = SummarizerExecutor(client=client)
 
 workflow = (
     WorkflowBuilder(
@@ -123,8 +127,7 @@ async def format(
         output_executors=[summarizer],
     )
     .add_fan_out_edges(dispatcher, [researcher, marketer, legal])
-    .add_fan_in_edges([researcher, marketer, legal], formatter)
-    .add_edge(formatter, summarizer)
+    .add_fan_in_edges([researcher, marketer, legal], summarizer)
     .build()
 )
 

examples/workflow_rag_ingest_parallel.py

@@ -8,7 +8,6 @@ This plan is organized by proposed example file. For each file: (1) source examp
 3. `workflow_aggregator_structured.py` (Pydantic structured extraction)
 4. `workflow_aggregator_voting.py` (ensemble classification — majority vote)
 5. `workflow_aggregator_ranked.py` (generate N candidates — score & rank)
-6. `workflow_rag_ingest_parallel.py` (parallelized version of `workflow_rag_ingest.py`)
 7. `workflow_multi_selection_edge_group.py`
 8. `workflow_agents_concurrent.py`
 9. `workflow_concurrent_custom_aggregator.py`
@@ -63,16 +62,6 @@ This plan is organized by proposed example file. For each file: (1) source examp
   - Aggregator (Agent or Executor) scores each proposal on criteria (creativity, memorability, brand fit) and yields a ranked list.
   - Teaching point: aggregation can be evaluative — the fan-in step judges via LLM, not just collects. Pattern applies to "generate N candidates, pick the best."
 
-### 6) `/workspace/examples/workflow_rag_ingest_parallel.py`
-- **Pull from**
-  - Local: `/workspace/examples/workflow_rag_ingest.py` (extract/chunk/embed pipeline and provider bootstrap)
-  - Upstream: `python/samples/03-workflows/parallelism/fan_out_fan_in_edges.py`, `python/samples/03-workflows/parallelism/map_reduce_and_visualization.py`
-  - Docs: Edges and parallelism concepts (`add_fan_out_edges`, `add_fan_in_edges`)
-- **Workflow accomplishes**
-  - Parallelizes embedding generation by fan-out over chunk batches and fan-in aggregation of embedded chunks.
-  - Keeps the same business goal as the current ingest demo (RAG ingestion) while changing the execution model.
-  - Provides a direct comparison against `workflow_agents_concurrent.py` to show domain-level parallelism (specialist perspectives) vs data-level parallelism (batch processing).
-
 ### 7) `/workspace/examples/workflow_multi_selection_edge_group.py`
 - **Pull from**
   - Local: `/workspace/examples/workflow_switch_case.py` (edge-group style, structured routing)
@@ -141,6 +130,17 @@ This plan is organized by proposed example file. For each file: (1) source examp
 - **Purpose**: contrasts language-level concurrency (`asyncio.gather`) with framework-level orchestration.
 - **Why appendix**: can distract from the core built-in workflow builder narrative for audiences new to async Python.
 
+### `/workspace/examples/workflow_rag_ingest_parallel.py`
+- **Pull from**
+  - Local: `/workspace/examples/workflow_rag_ingest.py` (extract/chunk/embed pipeline and provider bootstrap)
+  - Upstream: `python/samples/03-workflows/parallelism/fan_out_fan_in_edges.py`, `python/samples/03-workflows/parallelism/map_reduce_and_visualization.py`
+  - Docs: Edges and parallelism concepts (`add_fan_out_edges`, `add_fan_in_edges`)
+- **Workflow accomplishes**
+  - Parallelizes embedding generation by fan-out over chunk batches and fan-in aggregation of embedded chunks.
+  - Keeps the same business goal as the current ingest demo (RAG ingestion) while changing the execution model.
+  - Provides a direct comparison against `workflow_agents_concurrent.py` to show domain-level parallelism (specialist perspectives) vs data-level parallelism (batch processing).
+
+
 ## Scope Boundaries
 - Include only built-in workflow builder content for Session 5.
 - Exclude HITL-focused demos (next session).
@@ -150,7 +150,7 @@ This plan is organized by proposed example file. For each file: (1) source examp
 ## Verification Plan
 1. Smoke run each new script with standard provider env pattern used by current examples.
 2. Verify each script’s terminal output demonstrates the intended teaching point above.
-3. Run an explicit A/B comparison between `workflow_agents_concurrent.py` and `workflow_rag_ingest_parallel.py` and capture whether the audience sees a clear difference in orchestration pattern (specialist-agent parallelism vs data-pipeline parallelism).
 4. Rehearse in talk order with one fallback per segment (if a demo fails, which demo replaces it).
+
 ## Open Questions for MAF Team
 - **When is `output_executors` needed/recommended?** By default, `WorkflowBuilder` surfaces all outputs from all executors as events. If you only want outputs from certain executors, use `output_executors`. In our fan-out/fan-in demos, without `output_executors=[<aggregator>]`, the intermediate `Agent` nodes' `AgentResponse` objects leak into `get_outputs()` alongside the aggregator's output. We added `output_executors` to every fan-out/fan-in demo. Worth a slide callout — possibly on the same slide as the aggregation patterns table.