用 Amazon Bedrock AgentCore 构建多智能体调度器：Reactiv 如何加速 Shopify 移动电商自动化

{
  "merchant_id": "shop_123",
  "scheduled_for": "2025-03-08T09:00:00Z",
  "timezone": "America/Toronto",
  "refresh_scope": ["catalog", "home_feed"],
  "publish_mode": "review_required",
  "correlation_id": "shop_123-20250308T090000Z"
}

cat > scheduler.py <<'PY'
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
import hashlib
import json

STATE_FILE = Path(".scheduler-state.json")
ALLOWED_ACTIONS = {"sync_catalog", "rebuild_home_feed", "publish_if_valid"}


@dataclass(frozen=True)
class RefreshJob:
    merchant_id: str
    scheduled_for: str
    refresh_scope: tuple[str, ...]
    publish_mode: str = "review_required"

    @property
    def idempotency_key(self) -> str:
        payload = f"{self.merchant_id}:{self.scheduled_for}"
        return hashlib.sha256(payload.encode()).hexdigest()[:20]


class PlannerAgent:
    def plan(self, job: RefreshJob) -> dict:
        actions = []
        if "catalog" in job.refresh_scope:
            actions.append("sync_catalog")
        if "home_feed" in job.refresh_scope:
            actions.append("rebuild_home_feed")
        actions.append("publish_if_valid")
        return {"job": asdict(job), "actions": actions}


class PolicyAgent:
    def approve(self, plan: dict) -> None:
        unknown = set(plan["actions"]) - ALLOWED_ACTIONS
        if unknown:
            raise ValueError(f"Blocked actions: {sorted(unknown)}")
        if plan["job"]["publish_mode"] not in {"review_required", "automatic"}:
            raise ValueError("Unsupported publish mode")


class ExecutionAgent:
    def execute(self, plan: dict) -> dict:
        # Production: replace this block with an authenticated backend call.
        print("Executing:", json.dumps(plan, indent=2))
        return {
            "status": "completed",
            "actions_completed": plan["actions"],
            "version": "preview-001"
        }


class VerifierAgent:
    def verify(self, plan: dict, result: dict) -> None:
        expected = set(plan["actions"])
        completed = set(result.get("actions_completed", []))
        if result.get("status") != "completed" or expected != completed:
            raise RuntimeError("Refresh verification failed")


def load_completed() -> set[str]:
    if not STATE_FILE.exists():
        return set()
    return set(json.loads(STATE_FILE.read_text()))


def save_completed(keys: set[str]) -> None:
    STATE_FILE.write_text(json.dumps(sorted(keys), indent=2))


def run(job: RefreshJob) -> None:
    completed = load_completed()
    if job.idempotency_key in completed:
        print(f"Skipped duplicate job: {job.idempotency_key}")
        return

    plan = PlannerAgent().plan(job)
    PolicyAgent().approve(plan)
    result = ExecutionAgent().execute(plan)
    VerifierAgent().verify(plan, result)

    completed.add(job.idempotency_key)
    save_completed(completed)
    print(f"Verified job: {job.idempotency_key}")


if __name__ == "__main__":
    scheduled_for = datetime.now(timezone.utc).replace(
        minute=0, second=0, microsecond=0
    ).isoformat()
    run(RefreshJob(
        merchant_id="shop_123",
        scheduled_for=scheduled_for,
        refresh_scope=("catalog", "home_feed")
    ))
PY

python scheduler.py
python scheduler.py