智能体试运行时怎样设置暂停条件

告警群里的 API 扣费通知在半小时内响了上千次。单次调用费用从几分钱突然飙升到单小时数百美金。

当时线上正在运行一个处理客服多模态工单的 AI Agent 系统。该 Agent 负责读取用户上传的故障截图与文本描述，通过 Tool Calling 调用内部查询接口并自动回复。然而某个边缘场景下，图像识别返回了一个格式微小的错误，Agent 在理解这个报错时陷入了自推导死循环：不断尝试调用查询工具、收到报错、再次修正参数、再次发起调用。

当系统脱离了明确的流程边界，非确定性的 Agent 就会像一个不知疲倦且不断吞噬 API 预算的盲目轮询程序。

1. 凌晨 3 点的账单预警：单小时调用费用暴涨 18倍

排查日志时，发现这起事故的起因非常琐碎。

一位用户提交了一张尺寸异常的透明 PNG 格式报错截图。多模态大模型在读取图片时，OCR 节点提取出了空字符串，随后将控制权交给了 ReAct（Reasoning + Acting）Agent 循环。

Agent 在思考步骤中给出了这样的推理：
"当前无法获取错误码，尝试调用 query_order_status(order_id='') 补充信息。"

后端 API 拦截到空参数后返回了标准的 400 Bad Request。Agent 收到 400 状态码后，没有选择放弃，而是判定“参数可能格式不对”，接着构造了 query_order_status(order_id='null') 重新发包。

这个死循环在 40 分钟内重复了 1400 次。因为多模态模型的输入包含高清图像 Token，单次 Tool Calling 携带的上下文历史层层叠加，Token 消耗呈立方级爆缩。

2. 追踪 Agent 轨迹日志：Tool Calling 陷入循环补全死结

分析 Agent 的推理轨迹（Execution Trace），发现传统的超时机制（Timeout）在这里失效了。

因为单次 HTTP API 调用都是在 200ms 内成功完成的，服务端认为请求完全健康。失效发生在 Agent 的“语义决策环”上。

Agent Trace 节点记录:
Step 1: Vision LLM 解析图片 ➔ 返回空文本 ➔ 选择 Tool [query_db]
Step 2: Tool [query_db] 报错 400 ➔ 拼接报错信息至 Context ➔ 选择 Tool [query_db]
Step 3: Tool [query_db] 报错 400 ➔ 拼接更长上下文 ➔ 选择 Tool [query_db]
...
Step 82: Context 长度增加至 112K Tokens ➔ 每次单步调用耗时 8 秒 ➔ 费用暴增

LLM 的补全倾向在这种场景下成了副作用。在缺乏显式限制的情况下，模型会不断试图通过调整参数来“讨好”返回报错的工具，直到达到上下文窗口极限或被外部强行杀掉掉。

3. 状态机分层熔断与 Token 预算控制链路

要在运营过程中及时止损，就不能把控制权全盘托管给 Agent 的内部思考逻辑。必须在框架层施加确定性的物理闸门。

我们构建了一套三层止损防御体系：

这套架构把止损拆为三个维度：
第一层是硬性的 Token 消耗与金额预算上限；
第二层是相同工具调用的重复失败阈值；
第三层是上下文历史的滑动窗口裁减，防止历史报错无休止地拉大上下文体积。

4. 面向生产环境的多模态 Agent 耗速限制与状态熔断器实现

下面是在 Agent 框架核心调度环中集成的熔断与止损控制逻辑。

import time
from typing import List, Dict, Any, Callable

class AgentBudgetExceededException(Exception):
    """单次会话预算超限异常"""
    pass

class ToolLoopDetectedException(Exception):
    """工具调用死循环异常"""
    pass

class AgentGuardrailController:
    def __init__(
        self,
        max_cost_usd: float = 0.50,
        max_tool_calls: int = 10,
        max_same_tool_failures: int = 3,
        input_token_cost_per_k: float = 0.003,
        output_token_cost_per_k: float = 0.015
    ):
        self.max_cost = max_cost_usd
        self.max_tool_calls = max_tool_calls
        self.max_same_tool_failures = max_same_tool_failures
        self.input_cost_k = input_token_cost_per_k
        self.output_cost_k = output_token_cost_per_k
        
        # 运行时状态变量
        self.accumulated_cost = 0.0
        self.total_tool_calls = 0
        self.tool_failure_tracker: Dict[str, int] = {}

    def track_token_usage(self, prompt_tokens: int, completion_tokens: int):
        """实时计算 Token 计费"""
        cost = (prompt_tokens / 1000.0) * self.input_cost_k + \
               (completion_tokens / 1000.0) * self.output_cost_k
        self.accumulated_cost += cost
        
        if self.accumulated_cost > self.max_cost:
            raise AgentBudgetExceededException(
                f"会话预算达到上限！已消耗 ${self.accumulated_cost:.4f}，超过阈值 ${self.max_cost}"
            )

    def validate_tool_execution(self, tool_name: str):
        """校验工具调用频次上限"""
        self.total_tool_calls += 1
        if self.total_tool_calls > self.max_tool_calls:
            raise ToolLoopDetectedException(f"单次 Agent 执行工具总次数超过上限: {self.max_tool_calls}")

    def record_tool_result(self, tool_name: str, is_success: bool):
        """记录工具执行状态，发现连续失败立即熔断"""
        if is_success:
            self.tool_failure_tracker[tool_name] = 0
        else:
            current_failures = self.tool_failure_tracker.get(tool_name, 0) + 1
            self.tool_failure_tracker[tool_name] = current_failures
            if current_failures >= self.max_same_tool_failures:
                raise ToolLoopDetectedException(
                    f"工具 [{tool_name}] 已连续失败 {current_failures} 次，触发链路熔断！"
                )

class SafeAgentRunner:
    def __init__(self, agent_engine, guardrail: AgentGuardrailController):
        self.agent = agent_engine
        self.guardrail = guardrail

    def run_multimodal_task(self, image_path: str, user_text: str) -> str:
        context = self._init_context(image_path, user_text)
        
        while True:
            try:
                # 调用大模型决策下一步
                step_output = self.agent.predict_next_step(context)
                self.guardrail.track_token_usage(
                    step_output.usage.prompt_tokens,
                    step_output.usage.completion_tokens
                )
                
                if step_output.is_final_answer:
                    return step_output.final_response
                
                # 尝试执行 Tool
                tool_name = step_output.tool_to_call
                tool_args = step_output.tool_args
                
                self.guardrail.validate_tool_execution(tool_name)
                
                tool_success, result_data = self._execute_tool_safely(tool_name, tool_args)
                self.guardrail.record_tool_result(tool_name, tool_success)
                
                # 裁减上下文长度，只保留最近 5 轮 Tool 对话
                context.append_tool_result(tool_name, result_data)
                context.truncate_history(max_rounds=5)
                
            except (AgentBudgetExceededException, ToolLoopDetectedException) as e:
                # 记录兜底日志并触发业务止损转接
                self._log_emergency(str(e))
                return "很抱歉，当前工单处理遇到异常，已自动为您转接人工客服处理。"

    def _execute_tool_safely(self, name: str, args: dict) -> tuple[bool, Any]:
        try:
            res = self.agent.call_tool(name, args)
            return True, res
        except Exception as err:
            return False, str(err)

    def _log_emergency(self, msg: str):
        print(f"[AGENT 熔断拦截点]: {msg}")
    def _init_context(self, img: str, txt: str):
        return self.agent.build_multimodal_context(img, txt)

这段实现确保了 Agent 在底层模型产生逻辑死循环时，能够在 3 次工具失败或消费达到 0.5 美元时终止执行，避免把资源浪费在无意义的尝试上。

5. 止损之后的防线：别把业务控制权交给模型

多模态 Agent 展现了极强的自主分析能力，但系统设计的核心永远是可控性。

在生产环境中落地 Agent 业务，必须确立三条安全红线。

第一，所有的外部写操作（如退款、下单、修改数据库）必须经过显式确认授权。Agent 可以决定读操作，但在执行有副作用的写操作时，必须生成确定性的 API 参数并提交给审批闸门。

第二，Token 消费必须做租户级与会话级双重 Rate Limiting。不设置上限的 Agent 调度器，本质上是一个开放的资金消耗漏洞。

第三，建立监控告警指标而非只看 Agent 成功率。平均工具调用次数、工具连续失败率、会话 Token 增长梯度，这些指标比单一的最终成功率更容易提前暴露系统的潜在风险。