AI 智能体 + 大模型产品经理：从"答得对"到"走得对"的产品化设计

from enum import Enum
from pydantic import BaseModel, Field

class Risk(str, Enum):
    READ = "read"            # 直接执行，无感知
    WRITE = "write"          # 执行后告知，可撤销
    IRREVERSIBLE = "irreversible"  # 必须前置确认

class ToolSpec(BaseModel):
    name: str
    desc: str                # 面向模型的说明，PM 撰写
    risk: Risk
    max_amount: int | None = None
    approval_copy: str = ""  # 面向用户的确认文案，PM 撰写

TOOLS = [
    ToolSpec(name="query_order", desc="按订单号查询状态", risk=Risk.READ),
    ToolSpec(name="refund", desc="发起退款，金额单位分",
             risk=Risk.IRREVERSIBLE, max_amount=50000,
             approval_copy="即将为你退款 {amount} 元，确认后不可撤回"),
]

@dataclass
class GoldTrajectory:
    query: str
    expect_tools: list[str]      # 期望调用序列，顺序敏感
    forbidden: list[str]         # 绝不该出现的工具
    expect_final: str            # 终态：done / need_human / refused

def eval_trajectory(agent, golds: list[GoldTrajectory]) -> dict:
    ok_final = ok_path = ok_safe = 0
    for g in golds:
        run = agent.run(g.query)                     # 记录实际工具序列与终态
        ok_final += run.final == g.expect_final
        ok_path  += run.tools == g.expect_tools
        ok_safe  += not (set(run.tools) & set(g.forbidden))
    n = len(golds)
    return {"final_acc": ok_final / n, "path_acc": ok_path / n, "safety": ok_safe / n}

from enum import Enum

class TaskState(str, Enum):
    RUNNING = "running"; WAITING = "waiting_approval"
    DONE = "done"; EXPIRED = "expired"; FAILED = "failed"

# 产品规则：等待审批的超时与兜底，由 PM 定义
APPROVAL_TTL = 30 * 60          # 30 分钟
ON_TIMEOUT = "cancel_and_notify"  # 超时：撤销并通知，而非静默丢弃

def on_approval(state: TaskState, decision: str | None) -> TaskState:
    if state is not TaskState.WAITING:
        return state
    if decision == "approve":
        return TaskState.RUNNING
    if decision == "reject":
        return TaskState.FAILED
    return TaskState.EXPIRED          # 超时走明确终态

BUDGET = {"max_steps": 12, "max_usd": 0.15, "max_seconds": 60}

def guard(used_steps: int, used_usd: float, elapsed: float) -> str | None:
    if used_steps >= BUDGET["max_steps"]:
        return "任务步骤过多，已暂停，请补充信息后重试"
    if used_usd >= BUDGET["max_usd"]:
        return "该任务处理成本已达上限，建议转人工"
    if elapsed >= BUDGET["max_seconds"]:
        return "处理超时，已保存进度"
    return None    # None 表示继续