fengqun/swarm_minimal/academic_evaluation.py

"""Academic-style evaluation metadata for the minimal swarm prototype."""

from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class MarkovProcessAssessment:
    """Result of evaluating whether the prototype satisfies Markov assumptions."""

    markov_style_state_machine: bool
    formal_markov_process: bool
    formal_markov_decision_process: bool
    sufficient_state: tuple[str, ...]
    limiting_factors: tuple[str, ...]
    conclusion: str


ACADEMIC_STANDARD_SOURCES = (
    {
        "id": "NIST-AI-RMF",
        "name": "NIST AI Risk Management Framework 1.0",
        "use": "govern, map, measure and manage risk framing for autonomous AI behavior",
        "url": "https://www.nist.gov/itl/ai-risk-management-framework",
    },
    {
        "id": "NIST-AI-600-1",
        "name": "NIST AI RMF Generative AI Profile",
        "use": "generative-AI risks such as confabulation, privacy, information security and component integration",
        "url": "https://doi.org/10.6028/NIST.AI.600-1",
    },
    {
        "id": "OWASP-LLM",
        "name": "OWASP Top 10 for Large Language Model Applications",
        "use": "sensitive information disclosure, excessive agency and tool-boundary checks",
        "url": "https://owasp.org/www-project-top-10-for-large-language-model-applications/",
    },
    {
        "id": "OWASP-AST10",
        "name": "OWASP Agentic Skills Top 10",
        "use": "agentic skill risk checks for autonomous tools and delegated execution boundaries",
        "url": "https://owasp.org/www-project-agentic-skills-top-10/",
    },
    {
        "id": "MITRE-ATLAS",
        "name": "MITRE ATLAS",
        "use": "adversarial-AI and agent misuse framing for failure, abuse and recovery scenarios",
        "url": "https://atlas.mitre.org/",
    },
    {
        "id": "OTEL",
        "name": "OpenTelemetry documentation",
        "use": "observable traces, metrics, logs and event evidence expectations",
        "url": "https://opentelemetry.io/docs/",
    },
    {
        "id": "LANGGRAPH-HANDOFF",
        "name": "LangGraph handoff reference",
        "use": "active-agent handoff and transfer_to_<agent> continuity reference",
        "url": "https://reference.langchain.com/python/langgraph-swarm/handoff/create_handoff_tool",
    },
)


ALGORITHMS_USED = (
    {
        "name": "capability-based task claiming",
        "location": "swarm_minimal.core.InMemorySwarmStore.claim_next",
        "description": "agents claim pending tasks matching their capability; ties are ordered by pheromone score",
    },
    {
        "name": "pheromone / score reinforcement",
        "location": "swarm_minimal.core.InMemorySwarmStore.complete_task and fail_task",
        "description": "successful task scores add positive feedback; failed tasks receive negative feedback",
    },
    {
        "name": "winner-take-highest-score convergence",
        "location": "swarm_minimal.core.InMemorySwarmStore.converge",
        "description": "the highest-scoring completed task becomes the accepted result after scenario-level quality gates",
    },
    {
        "name": "quality-aware output scoring",
        "location": "examples.run_continuous_reasoning_acceptance.assess_output_quality and score_output",
        "description": "refusal, role-boundary, off-target and broken-handoff outputs are penalized before convergence",
    },
    {
        "name": "retry and fallback model recovery",
        "location": "examples.run_continuous_reasoning_acceptance.chat_with_fallback",
        "description": "low-quality model output triggers a retry and then fallback model takeover for the same Agnet step",
    },
    {
        "name": "weighted multi-round consensus",
        "location": "swarm_minimal.core.ConsensusSwarm.run",
        "description": "role-weighted votes accumulate until leader share and margin thresholds are reached",
    },
    {
        "name": "score evaporation",
        "location": "swarm_minimal.core.ConsensusSwarm._evaporate_scores",
        "description": "candidate scores decay between rounds before new evidence is added",
    },
    {
        "name": "distinct model discovery and selection",
        "location": "swarm_minimal.newapi_agnet.discover_newapi_models and select_distinct_models",
        "description": "NewAPI models are discovered from compatible endpoints and de-duplicated for multi-agent tests",
    },
)


def assess_markov_process_fit() -> MarkovProcessAssessment:
    """Classify the prototype against Markov-process requirements.

    The local swarm can be interpreted as a Markov-style state machine if the
    complete environment state is treated as the state variable. It is not a
    formal Markov process or MDP because the implementation does not define a
    transition probability kernel, action/reward tuple, or stochastic model for
    external LLM/API behavior.
    """

    return MarkovProcessAssessment(
        markov_style_state_machine=True,
        formal_markov_process=False,
        formal_markov_decision_process=False,
        sufficient_state=(
            "tasks with status, owner, output, score and error",
            "pheromone score table",
            "shared_state key-value environment",
            "observations already emitted",
            "agent policy functions and current round index for consensus",
        ),
        limiting_factors=(
            "no transition probability kernel P(s_next | s_current)",
            "no formal action space, reward function or policy optimization objective",
            "uuid/time and external NewAPI/LLM calls are not modeled as stochastic variables",
            "some acceptance outputs deliberately preserve history as audit evidence",
        ),
        conclusion=(
            "满足工程意义上的马尔可夫式状态转移：给定完整当前状态和 agent policy，"
            "下一步 claim、score 更新和收敛选择由当前状态决定。"
            "但不满足严格数学意义的 Markov process / MDP 定义。"
        ),
    )