AI 智能体开发实战：把"会调用工具"升级为"可交付的受控系统"

from enum import Enum
from pydantic import BaseModel, Field

class Risk(str, Enum):
    READ = "read"; WRITE = "write"; IRREVERSIBLE = "irreversible"

class Tool(BaseModel):
    name: str
    args_schema: type[BaseModel]
    risk: Risk
    fn: callable

class RefundArgs(BaseModel):
    order_id: str
    amount: int = Field(gt=0, le=500_00)   # 单位:分，硬性上限

TOOLS = {
    "get_order": Tool(name="get_order", args_schema=OrderArgs, risk=Risk.READ, fn=query_order),
    "refund":    Tool(name="refund",    args_schema=RefundArgs, risk=Risk.IRREVERSIBLE, fn=do_refund),
}

def invoke(name: str, raw_args: dict, approved: set[str]) -> dict:
    tool = TOOLS[name]
    args = tool.args_schema(**raw_args)          # 类型与范围校验，失败即抛错
    if tool.risk is Risk.IRREVERSIBLE and name not in approved:
        raise PermissionError(f"{name} 需人工审批")
    return tool.fn(**args.model_dump())

from enum import Enum

class State(str, Enum):
    COLLECT = "collect"; CONFIRM = "confirm"; EXECUTE = "execute"; DONE = "done"; FAILED = "failed"

TRANSITIONS = {
    State.COLLECT: {State.CONFIRM, State.FAILED},
    State.CONFIRM: {State.EXECUTE, State.COLLECT, State.FAILED},  # 用户可改单
    State.EXECUTE: {State.DONE, State.FAILED},
}

class Agent:
    def __init__(self, llm): self.llm, self.state, self.slots = llm, State.COLLECT, {}

    def step(self, user_msg: str) -> str:
        nxt, reply = self._decide(user_msg)          # LLM 只产出"建议迁移 + 回复"
        if nxt not in TRANSITIONS[self.state]:
            raise RuntimeError(f"非法迁移 {self.state} -> {nxt}")
        self.state = nxt
        return reply

    def _decide(self, msg: str) -> tuple[State, str]:
        out = self.llm.json(
            f"当前状态:{self.state} 槽位:{self.slots} 用户:{msg}\n"
            f"输出 {{next_state, reply, slots_patch}}，next_state 只能是 {list(State)}")
        self.slots.update(out.get("slots_patch", {}))
        return State(out["next_state"]), out["reply"]

import hashlib, time

def execute_idempotent(tool_name: str, args: dict, store) -> dict:
    key = hashlib.sha256(f"{tool_name}:{sorted(args.items())}".encode()).hexdigest()
    if (cached := store.get(key)) is not None:
        return cached                              # 命中即返回，不重复执行
    store.set(key, {"status": "running"}, ttl=3600)  # 先占位，防并发
    try:
        res = invoke(tool_name, args, approved=store.approved)
        store.set(key, {"status": "ok", "res": res}, ttl=86400)
        return res
    except TimeoutError:
        store.set(key, {"status": "unknown"}, ttl=86400)
        raise RuntimeError("结果未知，需对账，不得盲目重试")  # 关键:区分未知与失败

import json, time

def trace(step_no: int, state: str, payload: dict, log="agent.jsonl"):
    with open(log, "a") as f:
        f.write(json.dumps({"ts": time.time(), "step": step_no,
                            "state": state, **payload}, ensure_ascii=False) + "\n")