AI Agent 进入生产环境后，系统工程需要重做哪些功课

agent.run
├── model.plan
├── policy.check
├── tool.get_order
├── model.replan
├── policy.check
└── tool.create_refund

import hashlib
import json
import sys
import time
import uuid

ORDERS = {
    "ORD-100": {"status": "paid", "amount": 39.90}
}
ALLOWED_TOOLS = {"get_order", "create_refund"}
MAX_STEPS = 4
PROCESSED_KEYS = set()

SYSTEM_PROMPT = """
You are an order-support planner.
Return structured tool calls only.
Never invent an order ID or refund amount.
Refunds require a paid order and must not exceed 50 USD.
Stop when the objective is complete or when a tool returns an error.
""".strip()


def audit(run_id, event, **fields):
    record = {
        "timestamp": round(time.time(), 3),
        "run_id": run_id,
        "event": event,
        **fields,
    }
    print(json.dumps(record, ensure_ascii=False))


def mock_plan(objective):
    # 生产环境可在这里调用 LLM，并使用 JSON Schema 校验返回值。
    # 本示例只支持一个固定订单，避免把模拟器误认为真实智能体。
    if "ORD-100" not in objective:
        return []
    return [
        {"tool": "get_order", "arguments": {"order_id": "ORD-100"}},
        {
            "tool": "create_refund",
            "arguments": {"order_id": "ORD-100", "amount": 39.90},
        },
    ]


def get_order(order_id):
    order = ORDERS.get(order_id)
    if not order:
        raise ValueError("order not found")
    return {"order_id": order_id, **order}


def create_refund(order_id, amount, idempotency_key, dry_run=True):
    order = ORDERS.get(order_id)
    if not order or order["status"] != "paid":
        raise ValueError("only paid orders can be refunded")
    if amount <= 0 or amount > order["amount"] or amount > 50:
        raise ValueError("refund rejected by policy")
    if idempotency_key in PROCESSED_KEYS:
        return {"status": "duplicate_ignored", "order_id": order_id}
    if dry_run:
        return {"status": "would_refund", "order_id": order_id, "amount": amount}
    PROCESSED_KEYS.add(idempotency_key)
    return {"status": "refunded", "order_id": order_id, "amount": amount}


def run_agent(objective, dry_run=True):
    run_id = str(uuid.uuid4())
    audit(run_id, "run_started", objective=objective, dry_run=dry_run)
    plan = mock_plan(objective)

    if not plan:
        audit(run_id, "run_stopped", reason="unsupported_objective")
        return

    for index, step in enumerate(plan, start=1):
        if index > MAX_STEPS:
            audit(run_id, "run_stopped", reason="step_budget_exceeded")
            return

        tool = step["tool"]
        arguments = step["arguments"]
        if tool not in ALLOWED_TOOLS:
            audit(run_id, "tool_blocked", tool=tool)
            return

        key_source = json.dumps(
            {"run_id": run_id, "tool": tool, "arguments": arguments},
            sort_keys=True,
        )
        idempotency_key = hashlib.sha256(key_source.encode()).hexdigest()
        audit(run_id, "tool_started", step=index, tool=tool, arguments=arguments)

        try:
            if tool == "get_order":
                result = get_order(**arguments)
            else:
                result = create_refund(
                    **arguments,
                    idempotency_key=idempotency_key,
                    dry_run=dry_run,
                )
            audit(run_id, "tool_finished", step=index, tool=tool, result=result)
        except Exception as exc:
            audit(run_id, "tool_failed", step=index, tool=tool, error=str(exc))
            return

    audit(run_id, "run_finished", status="completed")


if __name__ == "__main__":
    objective = " ".join(sys.argv[1:]) or "Check ORD-100 and refund it"
    run_agent(objective, dry_run=True)

python agent_demo.py "Check ORD-100 and refund it"