用状态机与执行预算治理智能体的工具调用死循环：一个可运行的 Python 实现-阿里云开发者社区

一、背景：智能体进生产，最先出问题的是工具调用

把一个基于大模型的智能体（Agent）从 Demo 推到生产环境，团队往往会遇到同一类问题：模型在单轮对话里表现得很聪明，一旦放开工具调用权限，就开始出现各种无法预期的行为——反复查询同一张订单、把一个本可以一步完成的检索拆成七八轮、在多轮失败后仍然不肯收敛，最终把响应时间拖到超时、把调用成本推高数倍。

来看一个典型例子：售后智能体收到用户问题「订单 A1001 到货了吗」，正常的执行路径应该是「调用订单查询工具 -> 拿到状态 -> 生成回答」。但实际运行时，由于模型在上下文中没有正确识别工具已经返回了结果，它会连续三次调用完全相同参数（order_id=A1001）的查询工具，直到上下文被重复的工具结果占满，最终返回一句「抱歉我没查到」。

这类问题的本质，是把一个需要工程约束的系统当成了一次模型推理。模型负责决策，但它没有「步数概念」，没有「这次调用是不是和上次一样」的判断能力，也没有成本意识。因此，治理手段不能只写在 Prompt 里（「请不要重复调用工具」这种指令的可靠性很有限），必须落在代码层的硬约束上。

本文要解决的问题很具体：如何让智能体的工具调用具备可收敛、可预期、可观测的行为。

二、方案设计：三层约束

整体思路是在模型与工具之间加一层执行控制，把「自由的 while 循环」拆成三部分受控逻辑：

约束层	解决的问题	具体手段
状态机	执行流程不透明、无法插入校验点	显式 PLAN → ACT → OBSERVE → PLAN 推进，终态只有 DONE / FAILED
执行预算	步数失控、时间失控、成本失控	最大决策轮数、最大工具调用次数、墙钟超时（可扩展 token 预算）
工具层治理	重复调用、非幂等副作用	工具注册表 + 参数哈希幂等缓存 + 重复命中计数触发拦截

执行时序如下：

         ┌──────────────────────────────┐
         │            PLAN              │  policy(question, history) 产出动作
         └──────────────┬───────────────┘
                        │ action.type == tool
                        ▼
         ┌──────────────────────────────┐
         │            ACT               │  ToolRegistry.call() 前先做
         │  预算检查 / 缓存命中 / 重复检测 │
         └──────────────┬───────────────┘
                        ▼
         ┌──────────────────────────────┐
         │          OBSERVE             │  观测结果写回 history
         └──────────────┬───────────────┘
                        │
                        └────► 回到 PLAN（受 max_steps 约束）

  任意阶段抛出 BudgetExceeded / LoopDetected ──► FAILED（记录 reason）

关键在于：模型只能产出「意图」，能不能真正执行由控制层决定。这样即使换了模型、换了 Prompt，行为边界依然稳定。

三、实现步骤

定义异常与预算对象：把「超预算」和「打转」定义成两类显式异常，便于上层分别处理（前者通常是扩容或降级问题，后者是 Prompt 或工具设计问题）。
实现工具注册表：工具以名称注册，携带幂等标记；调用前用「工具名 + 参数规范 JSON」的哈希作为缓存键。
加入重复检测：同一个缓存键被命中超过阈值，说明模型在原地打转，直接抛 LoopDetected 中断。
实现执行器：状态机主循环，每轮消耗一次步数预算，每次真实调用工具消耗一次工具预算。
离线复现：用一个「脚本化 policy」替代真实模型，把正常路径和死循环路径固化成可回归的用例。

四、代码实现

以下代码仅依赖 Python 标准库，可直接保存为 agent_guard.py 运行：

from __future__ import annotations

import hashlib
import json
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable


class State(str, Enum):
    PLAN = 'PLAN'
    ACT = 'ACT'
    OBSERVE = 'OBSERVE'
    DONE = 'DONE'
    FAILED = 'FAILED'


class BudgetExceeded(RuntimeError):
    '''步数 / 工具调用次数 / 墙钟时间超预算'''


class LoopDetected(RuntimeError):
    '''相同工具 + 相同参数被反复调用'''


@dataclass
class Budget:
    max_steps: int = 8
    max_tool_calls: int = 6
    deadline_sec: float = 30.0
    steps: int = 0
    tool_calls: int = 0
    started_at: float = field(default_factory=time.time)

    def tick(self) -> None:
        self.steps += 1
        if self.steps > self.max_steps:
            raise BudgetExceeded(f'steps>{self.max_steps}')
        if time.time() - self.started_at > self.deadline_sec:
            raise BudgetExceeded('deadline')

    def spend_tool(self) -> None:
        self.tool_calls += 1
        if self.tool_calls > self.max_tool_calls:
            raise BudgetExceeded(f'tool_calls>{self.max_tool_calls}')


@dataclass
class Tool:
    name: str
    fn: Callable[..., Any]
    idempotent: bool = True


class ToolRegistry:
    def __init__(self, max_cache_hits: int = 2) -> None:
        self._tools: dict[str, Tool] = {
   }
        self._cache: dict[str, Any] = {
   }
        self._hit: dict[str, int] = {
   }
        self.max_cache_hits = max_cache_hits
        self.call_log: list[dict] = []

    def register(self, name: str, fn: Callable[..., Any], idempotent: bool = True) -> None:
        self._tools[name] = Tool(name, fn, idempotent)

    @staticmethod
    def _key(name: str, args: dict) -> str:
        raw = name + '|' + json.dumps(args, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(raw.encode('utf-8')).hexdigest()[:16]

    def call(self, name: str, args: dict, budget: Budget) -> Any:
        tool = self._tools.get(name)
        if tool is None:
            raise KeyError(f'unregistered tool: {name}')
        key = self._key(name, args)

        # 幂等工具：命中缓存则不再真实执行
        if tool.idempotent and key in self._cache:
            self._hit[key] = self._hit.get(key, 0) + 1
            if self._hit[key] > self.max_cache_hits:
                raise LoopDetected(f'{name}({args}) 重复命中 {self._hit[key]} 次')
            self.call_log.append({
   'name': name, 'args': args, 'cache': 'hit'})
            return self._cache[key]

        budget.spend_tool()
        value = tool.fn(**args)
        if tool.idempotent:
            self._cache[key] = value
        self.call_log.append({
   'name': name, 'args': args, 'cache': 'miss'})
        return value


class AgentRunner:
    def __init__(self, registry: ToolRegistry, budget: Budget | None = None) -> None:
        self.registry = registry
        self.budget = budget or Budget()

    def run(self, policy: Callable[[str, list], dict], question: str) -> dict:
        state, history, trace = State.PLAN, [], []
        action: dict = {
   }
        try:
            while True:
                self.budget.tick()
                if state is State.PLAN:
                    action = policy(question, history)
                    if action.get('type') == 'final':
                        state = State.DONE
                        history.append({
   'role': 'assistant', 'content': action['content']})
                        trace.append({
   'state': state.value, 'content': action['content']})
                        break
                    state = State.ACT
                if state is State.ACT:
                    obs = self.registry.call(action['tool'], action['args'], self.budget)
                    state = State.OBSERVE
                if state is State.OBSERVE:
                    history.append({
   'role': 'tool', 'name': action['tool'], 'content': obs})
                    trace.append({
   'state': 'OBSERVE', 'tool': action['tool'], 'obs': obs})
                    state = State.PLAN
        except (BudgetExceeded, LoopDetected) as exc:
            state = State.FAILED
            trace.append({
   'state': state.value, 'reason': str(exc)})
        return {
   
            'state': state.value,
            'trace': trace,
            'call_log': self.registry.call_log,
            'steps': self.budget.steps,
            'tool_calls': self.budget.tool_calls,
        }


def build_registry() -> ToolRegistry:
    reg = ToolRegistry(max_cache_hits=2)
    db = {
   'A1001': {
   'status': '已签收', 'amount': 199.0},
          'B2002': {
   'status': '运输中', 'amount': 88.0}}
    reg.register('query_order', lambda order_id: db[order_id])
    return reg


def scripted_policy(actions: list[dict]) -> Callable[[str, list], dict]:
    '''把预设动作序列包装成 policy，用于离线复现模型行为'''
    it = iter(actions)

    def _policy(question: str, history: list) -> dict:
        try:
            return next(it)
        except StopIteration:
            return {
   'type': 'final', 'content': '（决策脚本已耗尽）'}

    return _policy


if __name__ == '__main__':
    # 场景一：正常收敛
    good_policy = scripted_policy([
        {
   'type': 'tool', 'tool': 'query_order', 'args': {
   'order_id': 'A1001'}},
        {
   'type': 'final', 'content': '订单 A1001 已签收，实付 199.0 元。'},
    ])
    print(json.dumps(AgentRunner(build_registry()).run(good_policy, 'A1001 到货了吗'),
                     ensure_ascii=False, indent=2))

    # 场景二：模型反复查询同一订单，被缓存与重复检测拦截
    loop_policy = scripted_policy([
        {
   'type': 'tool', 'tool': 'query_order', 'args': {
   'order_id': 'A1001'}}
        for _ in range(5)
    ])
    print(json.dumps(AgentRunner(build_registry()).run(loop_policy, 'A1001 到货了吗'),
                     ensure_ascii=False, indent=2))

关键步骤说明

Budget.tick() 在每轮决策前调用：它约束的是「模型思考了几轮」，而不是「工具调用了多少次」。两者分开计数，才能区分是模型话太多还是工具太慢。
缓存键由工具名与规范化参数共同决定：json.dumps(..., sort_keys=True) 保证参数顺序不影响哈希，避免 {'a':1,'b':2} 与 {'b':2,'a':1} 被当成两次不同调用。
缓存命中计数而不是调用计数做重复检测：第一次调用是正常行为，第二次命中缓存通常是为了确认结果，第三次之后基本可以判定为原地打转，此时抛出异常比继续消耗轮次更划算。
异常被收敛为 FAILED 终态：执行器不让异常穿透到调用方，而是回填 reason，上游可以据此决定重试、降级到人工客服或返回兜底话术。
call_log 是排查的主要依据：生产中把这份日志连同 traceId 一起写入日志服务，就能快速判断一次失败到底发生在模型决策侧还是工具执行侧。

接入真实模型时，只需把 scripted_policy 换成一个调用模型并解析 tool_calls 的函数，其余控制逻辑完全不用改。工具实现本身可以落在函数计算上，按需伸缩，避免为低频工具常驻资源。

五、注意事项

幂等缓存只适用于只读或天然幂等的工具。query_order 可以缓存，create_refund 不行。写操作建议改为「业务幂等键 + 结果查询」，而不是直接在框架层缓存返回值。
预算值要按 SLA 反推。如果首字返回要求 3 秒内，deadline_sec 设为 30 秒没有意义；max_steps 建议从实际流量分布的中位数上浮 2 到 3 倍。
重复检测阈值不宜过低。用户连续问两次同样的问题，模型查两次订单是合理的，阈值设成 1 会误伤；设为 2 到 3 比较稳妥。
缓存要考虑一致性与内存增长。生产环境应使用带 TTL 的缓存（如 Redis 兼容服务），并限制 key 数量，否则长会话场景下内存会持续上涨。
失败要可观测。FAILED 不应静默返回兜底话术，建议按 reason 打点，把 LoopDetected 和 BudgetExceeded 分开告警，前者对应 Prompt 或工具描述问题，后者对应容量问题。
不要把约束全押在 Prompt 上。提示词可以降低重复调用概率，但无法提供确定性保证，硬约束必须写在代码里。

六、总结

智能体的可控性，很大程度上取决于工具调用这一层是否被当作「基础设施」来设计，而不是当作「模型能力的延伸」来放任。本文给出的三层约束——显式状态机、执行预算、工具层幂等与重复检测——并不复杂，但能把「智能体偶尔抽风」这个模糊的问题，转化为若干个可度量、可告警、可回归的具体指标：步数、工具调用次数、缓存命中次数、失败原因分布。

先用脚本化 policy 把正常路径和死循环路径固化成回归用例，再接入真实模型，是一个成本很低但收益明显的落地顺序。

用状态机与执行预算治理智能体的工具调用死循环：一个可运行的 Python 实现