多智能体系统如何学会可靠委派：从任务契约到动态质疑

from dataclasses import dataclass
from statistics import median
from typing import Any, Callable


@dataclass(frozen=True)
class TaskContract:
    name: str
    required_fields: tuple[str, ...]
    allowed_fields: tuple[str, ...]
    risk: str
    max_cost_units: int
    verifier: Callable[[dict[str, Any]], bool]


def verify_department_summary(result: dict[str, Any]) -> bool:
    return (
        isinstance(result.get("department"), str)
        and isinstance(result.get("employee_count"), int)
        and result["employee_count"] >= 3
        and isinstance(result.get("median_salary"), (int, float))
        and result["median_salary"] > 0
    )


def cheap_worker(payload: dict[str, Any]) -> dict[str, Any]:
    salaries = payload["salaries"]
    return {
        "department": payload["department"],
        "employee_count": len(salaries),
        "median_salary": median(salaries),
    }


def reasoning_worker(payload: dict[str, Any]) -> dict[str, Any]:
    result = cheap_worker(payload)
    result["review_note"] = "High-risk result requires compensation-team approval."
    return result


def delegate(contract: TaskContract, source: dict[str, Any]) -> dict[str, Any]:
    missing = set(contract.required_fields) - source.keys()
    if missing:
        raise ValueError(f"Human input required; missing fields: {sorted(missing)}")

    # Build a new payload instead of forwarding the full source object.
    payload = {key: source[key] for key in contract.allowed_fields if key in source}

    if len(payload.get("salaries", [])) < 3:
        raise PermissionError("Human review required: anonymity threshold not met")

    worker = reasoning_worker if contract.risk == "high" else cheap_worker
    estimated_cost = 5 if worker is reasoning_worker else 1
    if estimated_cost > contract.max_cost_units:
        raise RuntimeError("Budget exceeded; renegotiate the task contract")

    result = worker(payload)
    if not contract.verifier(result):
        raise RuntimeError("Verification failed; do not pass result downstream")

    return {
        "status": "verified",
        "worker": worker.__name__,
        "estimated_cost_units": estimated_cost,
        "result": result,
    }


if __name__ == "__main__":
    contract = TaskContract(
        name="department_salary_summary",
        required_fields=("department", "salaries"),
        allowed_fields=("department", "salaries"),
        risk="low",
        max_cost_units=2,
        verifier=verify_department_summary,
    )

    payroll_record = {
        "department": "Platform",
        "salaries": [92000, 101000, 97000, 110000],
        "employee_names": ["A", "B", "C", "D"],
        "bank_accounts": ["REDACTED"] * 4,
    }

    print(delegate(contract, payroll_record))