fengqun/examples/run_next_boundary_acceptance.py

from __future__ import annotations

import json
from pathlib import Path
import sys


ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from swarm_minimal.core import (  # noqa: E402
    Agent,
    CandidateOutput,
    ConsensusVote,
    InMemorySwarmStore,
    QuestioningAgent,
    QuestioningConsensusSwarm,
    SwarmCoordinator,
    Task,
    TaskStatus,
    fuse_candidate_outputs,
)


def main() -> None:
    checks = [
        scaled_autonomous_claim_acceptance(),
        candidate_fusion_acceptance(),
        questioning_consensus_acceptance(),
    ]
    report = {
        "standard": "next-boundary-minimal-acceptance-v1",
        "status": "PASS" if all(check["passed"] for check in checks) else "FAIL",
        "scope": {
            "sw_aqs_16": "3/5/7 Agent autonomous claim and scale stress",
            "candidate_fusion": "merge multiple valid candidates instead of selecting one winner only",
            "questioning_consensus": "challenge, revise, and re-vote before final acceptance",
        },
        "checks": checks,
    }
    print(json.dumps(report, ensure_ascii=False, indent=2))
    if report["status"] != "PASS":
        raise SystemExit(1)


def scaled_autonomous_claim_acceptance() -> dict[str, object]:
    cases = []
    for agent_count in (3, 5, 7):
        store = InMemorySwarmStore()
        run_id = f"s09-scaled-claim-{agent_count}"
        task_count = agent_count * 2
        store.shared_state[f"run:{run_id}:goal"] = "scaled autonomous claim"
        store.shared_state[f"run:{run_id}:status"] = "running"
        for index in range(task_count):
            store.add_task(Task(kind="stress", input=f"claim target {index}"))

        agents = [
            Agent(
                id=f"claim-agent-{index}",
                capability="stress",
                run=lambda task, shared_state, index=index: (
                    f"agent={index}; task={task.input}; autonomous_claim=ok",
                    0.7 + index / 100,
                ),
            )
            for index in range(agent_count)
        ]
        report = SwarmCoordinator(store=store, agents=agents).run_autonomous_until_converged(run_id)
        claimed_agents = {event.agent_id for event in report.claim_events}
        case = {
            "agent_count": agent_count,
            "task_count": task_count,
            "completed_tasks": report.completed_tasks,
            "failed_tasks": report.failed_tasks,
            "duplicate_claims": list(report.duplicate_claims),
            "participating_agents": len(claimed_agents),
            "converged": report.converged,
            "passed": (
                report.converged
                and report.completed_tasks == task_count
                and report.failed_tasks == 0
                and report.duplicate_claims == ()
                and len(claimed_agents) == agent_count
                and all(task.status == TaskStatus.DONE for task in store.tasks.values())
            ),
        }
        cases.append(case)

    return {
        "name": "sw_aqs_16_scaled_autonomous_claim_3_5_7",
        "passed": all(case["passed"] for case in cases),
        "evidence": cases,
    }


def candidate_fusion_acceptance() -> dict[str, object]:
    fused = fuse_candidate_outputs(
        [
            CandidateOutput("quality", "保留质量门\n保留 fallback 补救", 0.91, "quality"),
            CandidateOutput("coverage", "保留 fallback 补救\n加入候选融合输出", 0.86, "coverage"),
            CandidateOutput("noise", "低分噪声", 0.2, "below-threshold"),
        ],
        min_score=0.5,
    )
    expected_terms = ["保留质量门", "保留 fallback 补救", "加入候选融合输出"]
    passed = (
        fused.source_candidate_ids == ("quality", "coverage")
        and all(term in fused.text for term in expected_terms)
        and "低分噪声" not in fused.text
    )
    return {
        "name": "candidate_fusion_output",
        "passed": passed,
        "evidence": {
            "source_candidate_ids": list(fused.source_candidate_ids),
            "score": fused.score,
            "contains": expected_terms,
            "evidence": fused.evidence,
        },
    }


def questioning_consensus_acceptance() -> dict[str, object]:
    result = QuestioningConsensusSwarm(
        [
            make_questioning_agent("critic", "反驳", "revise_again"),
            make_questioning_agent("repairer", "修正", "revise_again"),
            make_questioning_agent("verifier", "验收", "approve_fused_candidate"),
        ],
        threshold=0.7,
        min_margin=0.2,
        max_rounds=3,
        evaporation=0.85,
    ).run("questioning consensus", "初始候选：质量门后的最高分输出")

    passed = (
        result.converged
        and result.accepted_candidate == "approve_fused_candidate"
        and len(result.rounds) >= 2
        and not result.rounds[0].consensus_round.converged
        and result.rounds[-1].consensus_round.converged
        and all(item.challenges for item in result.rounds)
        and all(item.revisions for item in result.rounds)
    )
    return {
        "name": "question_revise_revote_consensus",
        "passed": passed,
        "evidence": {
            "accepted_candidate": result.accepted_candidate,
            "round_count": len(result.rounds),
            "first_round_converged": result.rounds[0].consensus_round.converged,
            "last_round_converged": result.rounds[-1].consensus_round.converged,
        },
    }


def make_questioning_agent(agent_id: str, role: str, first_vote: str) -> QuestioningAgent:
    return QuestioningAgent(
        id=agent_id,
        role=role,
        weight=1.0,
        challenge=lambda candidate, state, round_index, role=role: f"{role} 质询 round={round_index}: 需要补充证据",
        revise=lambda candidate, challenges, state, round_index, role=role: (
            f"{candidate}\n{role} 修正 round={round_index}: 已回应 {len(challenges)} 个质询"
        ),
        vote=lambda candidate, state, round_index, agent_id=agent_id, role=role, first_vote=first_vote: ConsensusVote(
            agent_id=agent_id,
            role=role,
            candidate=first_vote if round_index == 1 else "approve_fused_candidate",
            confidence=0.55 if round_index == 1 else 0.78,
            evidence=f"{role} 已检查反驳、修正和再投票",
        ),
    )


if __name__ == "__main__":
    main()