从基础模型到可治理智能体：Google 企业 AI 一体化栈的关键价值

from __future__ import annotations

import json
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path

AUDIT_FILE = Path("agent-audit.jsonl")
ALLOWED_TOOLS = {"lookup_order", "refund_order"}
REFUND_LIMIT_WITHOUT_APPROVAL = 100.0


@dataclass
class Identity:
    user_id: str
    roles: list[str]


@dataclass
class ToolCall:
    name: str
    arguments: dict


def plan_action(request: str) -> ToolCall:
    """Local stand-in for a model-generated structured tool call."""
    if "refund" in request.lower():
        return ToolCall(
            name="refund_order",
            arguments={"order_id": "ORD-1042", "amount": 149.0},
        )
    return ToolCall(name="lookup_order", arguments={"order_id": "ORD-1042"})


def authorize(identity: Identity, call: ToolCall) -> tuple[bool, str]:
    if call.name not in ALLOWED_TOOLS:
        return False, "tool is not allowlisted"
    if call.name == "refund_order" and "support-agent" not in identity.roles:
        return False, "role cannot issue refunds"
    if (
        call.name == "refund_order"
        and float(call.arguments["amount"]) > REFUND_LIMIT_WITHOUT_APPROVAL
    ):
        return False, "human approval required"
    return True, "authorized"


def execute(call: ToolCall) -> dict:
    if call.name == "lookup_order":
        return {"order_id": call.arguments["order_id"], "status": "shipped"}
    if call.name == "refund_order":
        return {
            "order_id": call.arguments["order_id"],
            "refunded": call.arguments["amount"],
        }
    raise ValueError(f"Unsupported tool: {call.name}")


def audit(identity: Identity, call: ToolCall, decision: str, result: dict) -> None:
    event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "identity": asdict(identity),
        "tool_call": asdict(call),
        "decision": decision,
        "result": result,
    }
    with AUDIT_FILE.open("a", encoding="utf-8") as file:
        file.write(json.dumps(event, ensure_ascii=False) + "\n")


def main() -> None:
    identity = Identity(user_id="user-42", roles=["support-agent"])
    call = plan_action("Refund order ORD-1042 because the item arrived damaged")
    allowed, reason = authorize(identity, call)

    result = execute(call) if allowed else {"status": "blocked", "reason": reason}
    audit(identity, call, reason, result)
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

python governed_agent.py
cat agent-audit.jsonl