用 Amazon Bedrock AgentCore 构建面向 SAP 长周期项目的多智能体平台

{
  "task_id": "TEST-2025-0412",
  "project_id": "S4-MIGRATION-01",
  "assigned_agent": "test_designer",
  "objective": "为订单到收款流程生成回归测试草稿",
  "artifact_refs": ["process://order-to-cash/v7"],
  "approval_required": true
}

from __future__ import annotations

import json
import sys
import uuid
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Callable

STATE_FILE = Path("agent_state.json")


@dataclass
class Task:
    task_id: str
    agent: str
    objective: str
    status: str = "pending"
    approval_granted: bool = False
    result: dict[str, Any] | None = None


def search_project_docs(query: str) -> dict[str, Any]:
    return {
        "query": query,
        "matches": ["process://order-to-cash/v7"],
    }


def create_test_draft(process_ref: str) -> dict[str, Any]:
    return {
        "draft_id": f"DRAFT-{uuid.uuid4().hex[:8]}",
        "process_ref": process_ref,
        "cases": [
            "Create a sales order",
            "Post goods issue",
            "Generate an invoice",
        ],
    }


TOOLS: dict[str, Callable[..., dict[str, Any]]] = {
    "search_project_docs": search_project_docs,
    "create_test_draft": create_test_draft,
}

AGENT_POLICIES = {
    "process_analyst": {"search_project_docs"},
    "test_designer": {"search_project_docs", "create_test_draft"},
}

WRITE_TOOLS = {"create_test_draft"}


def load_tasks() -> dict[str, Task]:
    if not STATE_FILE.exists():
        return {}
    records = json.loads(STATE_FILE.read_text(encoding="utf-8"))
    return {task_id: Task(**data) for task_id, data in records.items()}


def save_tasks(tasks: dict[str, Task]) -> None:
    payload = {task_id: asdict(task) for task_id, task in tasks.items()}
    STATE_FILE.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def invoke_tool(task: Task, tool_name: str, **arguments: str) -> dict[str, Any]:
    allowed_tools = AGENT_POLICIES.get(task.agent, set())
    if tool_name not in allowed_tools:
        raise PermissionError(f"{task.agent} cannot call {tool_name}")
    if tool_name in WRITE_TOOLS and not task.approval_granted:
        raise PermissionError(f"{tool_name} requires human approval")
    return TOOLS[tool_name](**arguments)


def run(task: Task) -> None:
    task.status = "running"
    search = invoke_tool(task, "search_project_docs", query=task.objective)
    process_ref = search["matches"][0]

    if not task.approval_granted:
        task.status = "waiting_for_approval"
        task.result = {"proposed_process_ref": process_ref}
        return

    task.result = invoke_tool(
        task,
        "create_test_draft",
        process_ref=process_ref,
    )
    task.status = "completed"


def main() -> None:
    tasks = load_tasks()

    if len(sys.argv) == 1:
        task = Task(
            task_id=str(uuid.uuid4()),
            agent="test_designer",
            objective="Create regression tests for order to cash",
        )
        tasks[task.task_id] = task
        run(task)
        save_tasks(tasks)
        print(json.dumps(asdict(task), indent=2))
        print(f"Approve with: python {sys.argv[0]} {task.task_id}")
        return

    task_id = sys.argv[1]
    task = tasks[task_id]
    task.approval_granted = True
    run(task)
    save_tasks(tasks)
    print(json.dumps(asdict(task), indent=2))


if __name__ == "__main__":
    main()

python agent_workflow.py

python agent_workflow.py YOUR_TASK_ID