Add new workflow examples

Author: pamelafox

examples/spanish/workflow_conditional_state.py

@@ -22,7 +22,7 @@
 from agent_framework.openai import OpenAIChatClient
 from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
 from dotenv import load_dotenv
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel
 from typing_extensions import Never
 
 load_dotenv(override=True)
@@ -62,10 +62,7 @@ def parse_review_decision(message: Any) -> ReviewDecision | None:
     if not isinstance(message, AgentExecutorResponse):
         return None
 
-    try:
-        return ReviewDecision.model_validate_json(message.agent_response.text)
-    except ValidationError:
-        return None
+    return message.agent_response.value
 
 
 # Funciones de condición — reciben el mensaje del ejecutor anterior.
@@ -103,7 +100,7 @@ def needs_revision(message: Any) -> bool:
         "Define decision=REVISION_NEEDED si necesita mejoras.\n"
         "En feedback, explica brevemente tu razonamiento y da cambios accionables cuando aplique."
     ),
-    response_format=ReviewDecision,
+    default_options={"response_format": ReviewDecision},
 )
 
 editor = Agent(

examples/spanish/workflow_conditional_state_isolated.py

@@ -27,7 +27,7 @@
 from agent_framework.openai import OpenAIChatClient
 from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
 from dotenv import load_dotenv
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel
 from typing_extensions import Never
 
 load_dotenv(override=True)
@@ -67,10 +67,7 @@ def parse_review_decision(message: Any) -> ReviewDecision | None:
     if not isinstance(message, AgentExecutorResponse):
         return None
 
-    try:
-        return ReviewDecision.model_validate_json(message.agent_response.text)
-    except ValidationError:
-        return None
+    return message.agent_response.value
 
 
 # Funciones de condición — reciben el mensaje del ejecutor anterior.
@@ -109,7 +106,7 @@ def create_workflow(model_client: OpenAIChatClient):
             "Define decision=REVISION_NEEDED si necesita mejoras.\n"
             "En feedback, explica brevemente tu razonamiento y da cambios accionables cuando aplique."
         ),
-        response_format=ReviewDecision,
+        default_options={"response_format": ReviewDecision},
     )
 
     editor = Agent(

examples/spanish/workflow_conditional_structured.py

@@ -24,7 +24,7 @@
 from agent_framework.openai import OpenAIChatClient
 from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
 from dotenv import load_dotenv
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel
 from typing_extensions import Never
 
 load_dotenv(override=True)
@@ -66,10 +66,7 @@ def parse_review_decision(message: Any) -> ReviewDecision | None:
     if not isinstance(message, AgentExecutorResponse):
         return None
 
-    try:
-        return ReviewDecision.model_validate_json(message.agent_response.text)
-    except ValidationError:
-        return None
+    return message.agent_response.value
 
 
 def is_approved(message: Any) -> bool:
@@ -101,7 +98,7 @@ def needs_revision(message: Any) -> bool:
         "Si el borrador está listo, define decision=APPROVED e incluye la publicación lista para publicar en post_text. "
         "Si necesita cambios, define decision=REVISION_NEEDED y entrega feedback accionable."
     ),
-    response_format=ReviewDecision,
+    default_options={"response_format": ReviewDecision},
 )
 
 editor = Agent(

examples/spanish/workflow_converge.py

@@ -60,10 +60,7 @@ def parse_review_result(message: Any) -> ReviewResult | None:
     if not isinstance(message, AgentExecutorResponse):
         return None
 
-    try:
-        return ReviewResult.model_validate_json(message.agent_response.text)
-    except Exception:
-        return None
+    return message.agent_response.value
 
 
 def esta_aprobado(message: Any) -> bool:
@@ -104,7 +101,7 @@ def necesita_edicion(message: Any) -> bool:
         "- feedback: retroalimentación concisa y accionable\n"
         "- clarity, completeness, accuracy, structure: puntajes individuales (0-100)"
     ),
-    response_format=ReviewResult,
+    default_options={"response_format": ReviewResult},
 )
 
 

examples/spanish/workflow_switch_case.py

@@ -78,7 +78,7 @@ class ClassifyResult(BaseModel):
         "Question, Complaint o Feedback. "
         "Devuelve un objeto JSON con category, original_message y reasoning."
     ),
-    response_format=ClassifyResult,
+    default_options={"response_format": ClassifyResult},
 )
 
 
@@ -87,7 +87,7 @@ class ClassifyResult(BaseModel):
 @executor(id="extract_category")
 async def extract_category(response: AgentExecutorResponse, ctx: WorkflowContext[ClassifyResult]) -> None:
     """Analiza la salida JSON estructurada del clasificador y la envía aguas abajo."""
-    result = ClassifyResult.model_validate_json(response.agent_response.text)
+    result: ClassifyResult = response.agent_response.value
     print(f"→ Clasificado como: {result.category} — {result.reasoning}")
     await ctx.send_message(result)
 

examples/workflow_aggregator_llm_summary.py

@@ -0,0 +1,152 @@
+"""Fan-out/fan-in with LLM summarization aggregation.
+
+Same 3 expert branches as workflow_fan_out_fan_in_edges.py, but instead
+of a hand-coded template, a summarizer Agent synthesizes all branch
+outputs into a concise executive brief.
+
+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)
+"""
+
+import asyncio
+import os
+import sys
+
+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
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+API_HOST = os.getenv("API_HOST", "github")
+
+# Configure the chat client based on the API host
+async_credential = None
+if API_HOST == "azure":
+    async_credential = DefaultAzureCredential()
+    token_provider = get_bearer_token_provider(async_credential, "https://cognitiveservices.azure.com/.default")
+    client = OpenAIChatClient(
+        base_url=f"{os.environ['AZURE_OPENAI_ENDPOINT']}/openai/v1/",
+        api_key=token_provider,
+        model_id=os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT"],
+    )
+elif API_HOST == "github":
+    client = OpenAIChatClient(
+        base_url="https://models.github.ai/inference",
+        api_key=os.environ["GITHUB_TOKEN"],
+        model_id=os.getenv("GITHUB_MODEL", "openai/gpt-5-mini"),
+    )
+else:
+    client = OpenAIChatClient(
+        api_key=os.environ["OPENAI_API_KEY"], model_id=os.environ.get("OPENAI_MODEL", "gpt-5-mini")
+    )
+
+
+class DispatchPrompt(Executor):
+    """Emit the same prompt downstream so fan-out edges can broadcast it."""
+
+    @handler
+    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."""
+
+    @handler
+    async def format(
+        self,
+        results: list[AgentExecutorResponse],
+        ctx: WorkflowContext[str],
+    ) -> None:
+        """Combine expert outputs into labeled sections for the summarizer."""
+        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))
+
+
+dispatcher = DispatchPrompt(id="dispatcher")
+
+researcher = Agent(
+    client=client,
+    name="Researcher",
+    instructions=(
+        "You are an expert market researcher. "
+        "Given the prompt, provide concise factual insights, opportunities, and risks. "
+        "Use short bullet points."
+    ),
+)
+
+marketer = Agent(
+    client=client,
+    name="Marketer",
+    instructions=(
+        "You are a marketing strategist. "
+        "Given the prompt, propose clear value proposition and audience messaging. "
+        "Use short bullet points."
+    ),
+)
+
+legal = Agent(
+    client=client,
+    name="Legal",
+    instructions=(
+        "You are a legal and compliance reviewer. "
+        "Given the prompt, list constraints, disclaimers, and policy concerns. "
+        "Use short bullet points."
+    ),
+)
+
+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."
+    ),
+)
+
+workflow = (
+    WorkflowBuilder(
+        name="FanOutFanInLLMSummary",
+        description="Fan-out/fan-in with LLM summarization aggregation.",
+        start_executor=dispatcher,
+        output_executors=[summarizer],
+    )
+    .add_fan_out_edges(dispatcher, [researcher, marketer, legal])
+    .add_fan_in_edges([researcher, marketer, legal], formatter)
+    .add_edge(formatter, summarizer)
+    .build()
+)
+
+
+async def main() -> None:
+    """Run the sample and print the LLM-synthesized brief."""
+    prompt = "We are launching a budget-friendly electric bike for urban commuters."
+    print(f"Prompt: {prompt}\n")
+
+    events = await workflow.run(prompt)
+    for output in events.get_outputs():
+        print("=== Executive Brief (LLM-synthesized) ===")
+        print(output)
+
+    if async_credential:
+        await async_credential.close()
+
+
+if __name__ == "__main__":
+    if "--devui" in sys.argv:
+        from agent_framework.devui import serve
+
+        serve(entities=[workflow], port=8101, auto_open=True)
+    else:
+        asyncio.run(main())