294 lines
10 KiB
Python
294 lines
10 KiB
Python
"""MAF WorkflowBuilder graph construction for the roundtable experiment.
|
|
|
|
Creates real MAF Workflow graphs using the official SDK's WorkflowBuilder.
|
|
The SDK is imported lazily so the Gateway starts without it when the feature
|
|
flag is off.
|
|
|
|
Two topologies:
|
|
- ``structured`` (Phase 3): planner → parallel experts → cross-review →
|
|
approval gate → finalizer
|
|
- ``group_chat``: coordinator loop (dynamic speaker selection) → finalizer
|
|
|
|
Each node is a real MAF ``Executor`` subclass instance (from
|
|
``workflow_executors.py``). The builder uses the SDK's actual API:
|
|
``WorkflowBuilder(start_executor=...)``, ``add_edge``, ``add_fan_out_edges``,
|
|
``add_fan_in_edges``, ``add_switch_case_edge_group``, ``build()``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def build_manager_dispatch_workflow(*, manager_dispatch: Any, terminal: Any) -> Any:
|
|
"""Run the constrained manager decision before building a work topology.
|
|
|
|
The resulting checkpoint is the sole input to the second workflow. This
|
|
prevents a route's configured default topology from pre-empting a manager
|
|
decision such as manager-only, chain, or group chat.
|
|
"""
|
|
from agent_framework import WorkflowBuilder
|
|
|
|
builder = WorkflowBuilder(
|
|
start_executor=manager_dispatch,
|
|
name="maf-manager-dispatch",
|
|
output_from=[terminal],
|
|
)
|
|
builder.add_edge(manager_dispatch, terminal)
|
|
return builder.build()
|
|
|
|
|
|
def build_structured_workflow(
|
|
*,
|
|
planner: Any,
|
|
seat_executors: list[Any],
|
|
cross_review: Any,
|
|
approval_gate: Any,
|
|
finalizer: Any,
|
|
waiting_terminal: Any,
|
|
execution_mode: str = "parallel",
|
|
manager_dispatch: Any | None = None,
|
|
dispatch_router: Any | None = None,
|
|
dispatch_seat_entry: Any | None = None,
|
|
manager_answer: Any | None = None,
|
|
clarification_terminal: Any | None = None,
|
|
) -> Any:
|
|
"""Build a structured (fixed-topology) MAF Workflow.
|
|
|
|
Topology: planner → parallel seat executors → cross_review → approval gate
|
|
→ finalizer | waiting_terminal.
|
|
"""
|
|
from agent_framework import Case, Default, WorkflowBuilder
|
|
|
|
start = manager_dispatch or planner
|
|
if start is None:
|
|
raise ValueError("structured mode requires manager_dispatch or planner")
|
|
outputs = [finalizer, waiting_terminal]
|
|
if manager_answer is not None:
|
|
outputs.append(manager_answer)
|
|
if clarification_terminal is not None:
|
|
outputs.append(clarification_terminal)
|
|
builder = WorkflowBuilder(
|
|
start_executor=start,
|
|
name="maf-structured-roundtable",
|
|
output_from=outputs,
|
|
)
|
|
|
|
seat_start = planner
|
|
if manager_dispatch is not None:
|
|
if dispatch_router is None or dispatch_seat_entry is None:
|
|
raise ValueError("manager_dispatch requires structured dispatch route executors")
|
|
builder.add_edge(manager_dispatch, dispatch_router)
|
|
cases = []
|
|
if manager_answer is not None:
|
|
cases.append(Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "manager_only",
|
|
target=manager_answer,
|
|
))
|
|
if clarification_terminal is not None:
|
|
cases.append(Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "clarification",
|
|
target=clarification_terminal,
|
|
))
|
|
cases.append(Default(target=dispatch_seat_entry))
|
|
builder.add_switch_case_edge_group(dispatch_router, cases)
|
|
seat_start = dispatch_seat_entry
|
|
|
|
if execution_mode == "chain" and seat_executors:
|
|
builder.add_edge(seat_start, seat_executors[0])
|
|
for previous, current in zip(seat_executors, seat_executors[1:]):
|
|
builder.add_edge(previous, current)
|
|
builder.add_edge(seat_executors[-1], cross_review)
|
|
elif len(seat_executors) >= 2:
|
|
builder.add_fan_out_edges(seat_start, seat_executors)
|
|
builder.add_fan_in_edges(seat_executors, cross_review)
|
|
elif len(seat_executors) == 1:
|
|
builder.add_edge(seat_start, seat_executors[0])
|
|
builder.add_edge(seat_executors[0], cross_review)
|
|
else:
|
|
builder.add_edge(seat_start, cross_review)
|
|
|
|
builder.add_edge(cross_review, approval_gate)
|
|
|
|
builder.add_switch_case_edge_group(approval_gate, [
|
|
Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "waiting_approval",
|
|
target=waiting_terminal,
|
|
),
|
|
Default(target=finalizer),
|
|
])
|
|
|
|
return builder.build()
|
|
|
|
|
|
def build_manager_only_workflow(
|
|
*,
|
|
manager_answer: Any,
|
|
manager_dispatch: Any | None = None,
|
|
dispatch_router: Any | None = None,
|
|
clarification_terminal: Any | None = None,
|
|
) -> Any:
|
|
"""Build the direct-answer topology with no expert seat nodes."""
|
|
from agent_framework import Case, Default, WorkflowBuilder
|
|
|
|
if manager_dispatch is None:
|
|
builder = WorkflowBuilder(
|
|
start_executor=manager_answer,
|
|
name="maf-manager-only",
|
|
output_from=[manager_answer],
|
|
)
|
|
return builder.build()
|
|
if dispatch_router is None or clarification_terminal is None:
|
|
raise ValueError("manager-only dispatch requires route and clarification executors")
|
|
builder = WorkflowBuilder(
|
|
start_executor=manager_dispatch,
|
|
name="maf-manager-only",
|
|
output_from=[manager_answer, clarification_terminal],
|
|
)
|
|
builder.add_edge(manager_dispatch, dispatch_router)
|
|
builder.add_switch_case_edge_group(dispatch_router, [
|
|
Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "manager_only",
|
|
target=manager_answer,
|
|
),
|
|
Default(target=clarification_terminal),
|
|
])
|
|
return builder.build()
|
|
|
|
|
|
def build_group_chat_workflow(
|
|
*,
|
|
coordinator: Any,
|
|
seat_executors: list[Any],
|
|
approval_gate: Any,
|
|
finalizer: Any,
|
|
waiting_terminal: Any,
|
|
seat_agent_ids: list[str],
|
|
manager_dispatch: Any | None = None,
|
|
dispatch_router: Any | None = None,
|
|
manager_answer: Any | None = None,
|
|
clarification_terminal: Any | None = None,
|
|
) -> Any:
|
|
"""Build a dynamic group-chat MAF Workflow.
|
|
|
|
The coordinator selects the next speaker each round via switch-case edges.
|
|
Loop terminates on: coordinator signals completion, max_rounds reached.
|
|
"""
|
|
from agent_framework import Case, Default, WorkflowBuilder
|
|
|
|
start = manager_dispatch or coordinator
|
|
outputs = [finalizer, waiting_terminal]
|
|
if manager_answer is not None:
|
|
outputs.append(manager_answer)
|
|
if clarification_terminal is not None:
|
|
outputs.append(clarification_terminal)
|
|
builder = WorkflowBuilder(
|
|
start_executor=start,
|
|
name="maf-groupchat-roundtable",
|
|
output_from=outputs,
|
|
)
|
|
|
|
if manager_dispatch is not None:
|
|
if dispatch_router is None:
|
|
raise ValueError("group_chat manager_dispatch requires a route executor")
|
|
builder.add_edge(manager_dispatch, dispatch_router)
|
|
cases = []
|
|
if manager_answer is not None:
|
|
cases.append(Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "manager_only",
|
|
target=manager_answer,
|
|
))
|
|
if clarification_terminal is not None:
|
|
cases.append(Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "clarification",
|
|
target=clarification_terminal,
|
|
))
|
|
cases.append(Default(target=coordinator))
|
|
builder.add_switch_case_edge_group(dispatch_router, cases)
|
|
|
|
cases = []
|
|
for i, agent_id in enumerate(seat_agent_ids):
|
|
cases.append(Case(
|
|
condition=lambda out, aid=agent_id: isinstance(out, dict) and out.get("agent_id") == aid,
|
|
target=seat_executors[i],
|
|
))
|
|
cases.append(Default(target=approval_gate))
|
|
builder.add_switch_case_edge_group(coordinator, cases)
|
|
|
|
for seat_exec in seat_executors:
|
|
builder.add_edge(seat_exec, coordinator)
|
|
|
|
builder.add_switch_case_edge_group(approval_gate, [
|
|
Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "waiting_approval",
|
|
target=waiting_terminal,
|
|
),
|
|
Default(target=finalizer),
|
|
])
|
|
|
|
return builder.build()
|
|
|
|
|
|
def create_workflow(
|
|
*,
|
|
mode: str,
|
|
seat_executors: list[Any],
|
|
approval_gate: Any,
|
|
finalizer: Any,
|
|
waiting_terminal: Any,
|
|
seat_agent_ids: list[str],
|
|
coordinator: Any | None = None,
|
|
planner: Any | None = None,
|
|
cross_review: Any | None = None,
|
|
manager_answer: Any | None = None,
|
|
execution_mode: str = "parallel",
|
|
manager_dispatch: Any | None = None,
|
|
dispatch_router: Any | None = None,
|
|
dispatch_seat_entry: Any | None = None,
|
|
clarification_terminal: Any | None = None,
|
|
) -> Any:
|
|
"""Factory: create the appropriate workflow for the given mode."""
|
|
if mode == "manager_only":
|
|
if manager_answer is None:
|
|
raise ValueError("manager_only mode requires manager_answer")
|
|
return build_manager_only_workflow(
|
|
manager_answer=manager_answer,
|
|
manager_dispatch=manager_dispatch,
|
|
dispatch_router=dispatch_router,
|
|
clarification_terminal=clarification_terminal,
|
|
)
|
|
if mode == "group_chat":
|
|
if coordinator is None:
|
|
raise ValueError("group_chat mode requires coordinator")
|
|
return build_group_chat_workflow(
|
|
coordinator=coordinator,
|
|
seat_executors=seat_executors,
|
|
approval_gate=approval_gate,
|
|
finalizer=finalizer,
|
|
waiting_terminal=waiting_terminal,
|
|
seat_agent_ids=seat_agent_ids,
|
|
manager_dispatch=manager_dispatch,
|
|
dispatch_router=dispatch_router,
|
|
manager_answer=manager_answer,
|
|
clarification_terminal=clarification_terminal,
|
|
)
|
|
if planner is None or cross_review is None:
|
|
raise ValueError("structured mode requires planner and cross_review")
|
|
return build_structured_workflow(
|
|
planner=planner,
|
|
seat_executors=seat_executors,
|
|
cross_review=cross_review,
|
|
approval_gate=approval_gate,
|
|
finalizer=finalizer,
|
|
waiting_terminal=waiting_terminal,
|
|
execution_mode=execution_mode,
|
|
manager_dispatch=manager_dispatch,
|
|
dispatch_router=dispatch_router,
|
|
dispatch_seat_entry=dispatch_seat_entry,
|
|
manager_answer=manager_answer,
|
|
clarification_terminal=clarification_terminal,
|
|
)
|