生产级 Agent 框架：用工程边界驯服概率性智能

原创

IT大佬 jzit-top

发布于 2026-10-26 09:42:28

600

Agent 的能力上限由模型决定，可用性下限由框架决定。一个演示用的 Agent 只需要一个while循环；一个能托管真实业务、7×24 运行的 Agent，必须在模型外面套上一层确定性工程：执行可中断可恢复、工具调用有硬边界、运行成本可熔断。本文从这三个设计点展开，并给出可落地的代码骨架。

一、执行内核：把“跑到哪了”变成框架事实

自由 ReAct 循环最致命的缺陷是不可恢复。进程重启后，while not finished 的中间状态随之消失，任务只能从头再来。生产级框架的做法是把执行进度显式建模为可序列化状态，每次迭代落盘，恢复时从快照继续，而不是重新推理：

@dataclass
class RunState:
    goal: str
    step: int = 0
    history: list = field(default_factory=list)
    pending: dict | None = None
    status: str = "running"   # running | interrupted | done | failed

class DurableLoop:
    def __init__(self, store):
        self.store = store    # 生产环境替换为 Postgres / 事件日志

    async def step(self, task_id: str, plan, act):
        s = self.store.load(task_id)
        decision = await plan(s)              # LLM 只负责决策
        if decision.kind == "tool":
            result = await act(decision)      # 框架负责执行
            s.history.append(result)
        elif decision.kind == "approval":
            s.pending, s.status = decision.request, "interrupted"
        else:
            s.status = "done"
        s.step += 1
        self.store.save(task_id, s)           # 每步持久化
        return s

要点在于 status 和 pending 是框架层的事实，不是模型“以为”自己该做什么。暂停后由人工审批，再以同一 task_id 恢复，从精确断点续跑。模型仍然自由规划，但“执行到哪、能不能继续”由状态机说了算——确定性的骨架，包住概率性的智能。

二、工具层：权限、幂等、补偿三道闸门

工具调用是最大的事故来源，而事故往往不是模型“想作恶”，是重试变成了重复副作用：支付工具超时后盲目重发，用户被扣两次款。执行器必须同时施加三道硬约束：

async def execute_tool(name: str, args: dict, ctx: dict):
    # 闸门一：权限分级。POLICY 由代码定义，模型无法用提示词绕过
    if POLICY[name] == "DANGEROUS" and not ctx["approved"]:
        raise PermissionError(f"{name} requires human approval")

    # 闸门二：幂等键。同一动作+参数的重试命中缓存，跳过真实副作用
    key = sha256(f"{name}:{json.dumps(args, sort_keys=True)}".encode()).hexdigest()
    if cached := await idem.get(key):
        return cached

    # 闸门三：执行 + 补偿登记。写操作必须伴随可回滚的逆操作
    result = await sandbox_execute(name, args)
    if POLICY[name] == "WRITE":
        await saga.push(tool=COMPENSATIONS[name], context=result)
    await idem.set(key, result)
    return result

三道闸门对应三种失败模式：权限分级防越权，幂等键防重复写入，补偿栈防多步写操作的中间失败留下脏数据——任一步失败，逆序执行已执行步骤的补偿，补偿本身失败则进死信队列转人工。三者叠加，“工具会出错”才不等于“系统会坏掉”。

三、运行时护栏：预算与循环检测必须在“还在跑”时生效

Agent 的失败是静默的：幻觉返回 HTTP 200，绕了十四圈的无效循环在日志上与正常执行毫无区别。因此成本与循环控制不能是事后报表，必须做成可插拔的停止条件，实时阻断：

class StopCondition(Protocol):
    def check(self, s: RunState) -> str | None: ...

class BudgetGuard:
    def __init__(self, max_usd: float, max_calls: int, max_repeat: int = 3):
        self.max_usd, self.max_calls, self.max_repeat = max_usd, max_calls, max_repeat

    def check(self, s: RunState) -> str | None:
        if s.cost_usd >= self.max_usd:
            return f"budget exhausted: ${s.cost_usd:.2f}"
        if s.tool_calls >= self.max_calls:
            return f"tool call cap: {s.tool_calls}"
        tail = s.history[-6:]
        if len(tail) == 6 and len({str(h) for h in tail}) <= self.max_repeat:
            return "loop detected: no new evidence"
        return None

# 主循环对“有哪些刹车”零感知
for guard in (BudgetGuard(2.0, 40), NoProgressGuard()):
    if reason := guard.check(state):
        state.status, state.stop_reason = "failed", reason
        break

协议化的价值在于：新增一种熔断策略只需实现 check，主循环不动。与之并列的是可观测性——每次 LLM 与工具调用都应在 span 中携带 cost_usd、tokens、tool_name，并在 span 离开进程前完成评估与拦截，而不是等数据落仓后才“发现”异常。

结语

三个设计点指向同一原则：框架不替模型做决策，而是让决策的边界不可逾越、状态不可丢失、代价不可静默累积。执行内核守住连续性，工具闸门守住副作用安全性，运行时护栏守住经济性与可诊断性。缺任一层，Agent 都只是“能演示”，而非“敢托管”。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

agent