董宏凯
1aeee86984
P0-1 migration: detect the same-named stage-3 execution_stage CHECK by sqltext (must contain rework + manager_review), SQLite rebuild / PG-MySQL drop+add, PG boolean default TRUE, validate_schema checks constraint content + seat_tasks status CHECK/unique/FKs; tests start from a real stage-3 schema. P0-2: a missing/invalid manager review (timeout/exception/empty/invalid JSON/missing per-seat decision) is never an acceptance - system rework under the cap, contract fallback (reassign/controlled_gap/ask_user) afterwards, source=system events. P0-3: unaccepted deliveries are isolated from the fact ledger, cross-review and final report (deferred merge, delivery_json envelope, accepted_task_results in checkpoint, accepted-only synthesis, no turn_results fallback). P0-4: seat-task init is atomic get-or-create over the full expected set; recovery fills missing tasks without overwriting frozen contracts; quality gate blocks missing required tasks. P1: zero config values honoured; duplicate detection persists SHA-256 conclusion digests in history (survives restarts); findings never satisfy evidence, evidence_refs must resolve to trusted ledger refs (unverified_evidence_ref); conversation SSE projection tests added and manager/finalizer messages are no longer mislabelled as seat.delivery.submitted.
321 lines
12 KiB
Python
321 lines
12 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,
|
|
quality_gate: Any | None = None,
|
|
waiting_input_terminal: Any | None = None,
|
|
) -> Any:
|
|
"""Build a structured (fixed-topology) MAF Workflow.
|
|
|
|
Topology (Stage 3): planner → parallel seat executors → cross_review →
|
|
approval gate → finalizer | waiting_terminal.
|
|
|
|
Topology (Stage 4 quality gate): quality seat-tasks loop → quality gate →
|
|
(waiting_input | cross_review) → approval gate → finalizer | waiting_terminal.
|
|
When ``quality_gate`` is provided, the seat fan-out is replaced by the
|
|
single quality loop executor and ``waiting_input_terminal`` becomes an
|
|
additional workflow output for the ``ask_user`` pause path.
|
|
"""
|
|
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)
|
|
if waiting_input_terminal is not None:
|
|
outputs.append(waiting_input_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 quality_gate is not None:
|
|
# Stage 4: single quality-loop node replaces the seat fan-out; the
|
|
# quality gate routes ask_user pauses to waiting_input_terminal and
|
|
# everything else to cross-review.
|
|
builder.add_edge(seat_start, quality_gate)
|
|
gate_cases = []
|
|
if waiting_input_terminal is not None:
|
|
gate_cases.append(Case(
|
|
condition=lambda out: isinstance(out, dict) and out.get("decision") == "waiting_input",
|
|
target=waiting_input_terminal,
|
|
))
|
|
gate_cases.append(Default(target=cross_review))
|
|
builder.add_switch_case_edge_group(quality_gate, gate_cases)
|
|
elif 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,
|
|
quality_gate: Any | None = None,
|
|
waiting_input_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,
|
|
quality_gate=quality_gate,
|
|
waiting_input_terminal=waiting_input_terminal,
|
|
)
|