Harness 中的推理步数预算：从根源防止Agent无限循环的核心机制

---

一、引言

钩子：所有Agent开发者都踩过的噩梦级痛点

你有没有过这样的经历：花了一周时间打磨的企业级客服Agent，刚上线半天就触发了成本告警，后台日志显示某个用户的售后请求竟然让Agent连续跑了217步，调用了89次工单系统接口、62次物流查询接口，累计消耗了127元的大模型API费用，最后还是没有输出任何有效结果？
更离谱的是去年我朋友做的一个自动化研究Agent，本来想让它整理2024年大模型推理优化的相关论文，结果忘了加执行限制，Agent卡在了「下载论文→解析失败→重新下载」的死循环里，跑了整整一晚上，花了近3000块的OpenAI API费用，最后只得到了一堆错误日志。
这类问题几乎是所有多步Agent开发者的共性噩梦：大模型驱动的Agent本质是在「思考-行动-观察」的循环里迭代，没有有效控制的话，无限循环几乎是必然会发生的事件，轻则造成成本浪费，重则导致整个服务不可用。

问题背景：为什么Agent无限循环是无解的慢性病？

随着大模型技术的成熟，多步Agent已经成为企业级AI应用的主流形态：从客服、自动化办公到数据处理、科研辅助，几乎所有复杂AI场景都依赖Agent的多轮迭代能力。但Agent的循环执行特性天生带有「无限循环」的风险，常见的触发场景包括：

1. 工具调用异常循环：工具返回参数错误、权限不足等问题，大模型反复修改参数重试却始终无法解决；
2. 上下文漂移循环：多轮迭代后初始任务被挤出上下文窗口，大模型忘记核心目标，反复执行无关操作；
3. 反思死循环：大模型每次反思都判定之前的输出不符合要求，反复修改永远无法输出最终结果；
4. 多Agent死锁：多个Agent协作时互相等待对方的输出，陷入无限消息循环。
   传统的超时机制完全无法解决这类问题：有些复杂任务本身就需要10分钟以上的执行时间，超时阈值设高了拦不住短时间高频率的循环，设低了又会误杀正常的长耗时任务，属于典型的「一刀切」方案。

文章目标：你将从这篇文章学到什么？

本文将从核心原理到实战落地，全方位讲解AI执行编排层（Harness）中的推理步数预算机制——这是目前业界公认的解决Agent无限循环问题的最优方案。读完本文你将：

1. 完全理解推理步数预算的核心概念、数学模型和系统组成；
2. 从零实现一个带有步数预算控制的极简Harness框架，可直接用于生产环境；
3. 掌握步数预算的最佳实践、避坑指南和多场景适配方案；
4. 了解Agent执行治理的未来发展趋势。

---

二、基础知识铺垫

核心概念定义

在正式讲解步数预算之前，我们先统一几个核心概念的定义，避免歧义：

1. 什么是Harness？

本文所指的Harness是AI Agent的执行编排层，相当于Agent的操作系统内核，负责接收用户任务、调度大模型推理、管理工具调用、维护上下文记忆、控制执行流程、返回最终结果。所有的Agent执行逻辑都跑在Harness之上，它是唯一能全局控制Agent执行流程的模块。

2. 什么是推理步数？

我们将Agent的一次完整迭代轮次定义为1步：即「大模型推理生成动作规划→执行动作（工具调用/用户交互）→结果反馈到上下文」的完整流程，每完成一次上述流程记为1个推理步。
注意：系统级的网络重试、工具调用失败的自动重试不算入推理步数，只有Agent主动发起的动作迭代才会计数。

3. 什么是推理步数预算？

推理步数预算是指给每个Agent任务预先分配的最大推理步数额度，每执行一步扣除1点预算，当预算耗尽时触发终止逻辑，从根源上限制任务的最大迭代次数，避免无限循环。

4. 无限循环的量化定义

从概率学角度，当Agent连续3步执行完全相同的动作、或者连续10步没有向任务完成的方向推进时，我们就可以判定为进入了无限循环，其发生概率随步数增加呈指数级上升：
$$P(循环|步数>N)=1-(1-p{)}^N$$
其中$p$是单步触发循环的概率，根据业界统计，通用Agent的单步循环概率约为1%_{3%，当N=100时，循环概率高达63%}95%，这也是为什么没有步数限制的Agent几乎必然会出现循环。

相关控制机制对比

很多人会混淆步数预算和其他执行控制机制的差异，我们用一个表格明确各方案的优劣势：

对比维度	传统超时机制	Token上限控制	固定步数预算	动态步数预算
控制粒度	时间维度，粗粒度	Token量，中粒度	执行轮次，中粒度	执行轮次+任务状态，细粒度
误杀率	高（长耗时任务易被误杀）	中（长上下文任务易被误杀）	中（复杂任务易被误杀）	低（动态适配任务需求）
循环拦截能力	弱（短时间高频率循环无法拦截）	中（Token耗尽时才会终止）	强（严格限制最大迭代次数）	极强（可提前识别异常终止）
成本控制能力	弱（无法限制短时间高步数消耗）	中（可限制Token成本，无法限制工具调用成本）	强（可预估最大成本）	极强（精准控制每一步消耗）
实现复杂度	极低	低	中	高
适用场景	单步短耗时任务	所有大模型任务	通用标准Agent任务	复杂企业级Agent任务

可以看到，步数预算是唯一同时兼顾循环拦截能力和成本控制能力的方案，也是Harness层必须具备的核心能力。

核心概念实体关系

我们用ER图明确各核心概念的关系：

---

三、核心内容：推理步数预算的设计与实现

核心原理与数学模型

推理步数预算的核心逻辑非常简单：给每个任务分配初始预算，每走一步扣减，预算耗尽就终止，但要适配企业级场景，还需要引入动态调整、异常检测等能力，其完整数学模型如下：

1. 初始预算分配模型

初始预算$B_0$由任务类型、用户等级、付费情况三个维度共同决定：
$$B_0=\beta \ast (B_{base}(task\_type)\ast U(user\_level)+P(pay\_amount))$$
其中：

• $B_{base}(task\_type)$是对应任务类型的基础预算，比如问答类=5，工具调用类=20，复杂推理类=50；
• $U(user\_level)$是用户等级系数，免费用户=0.8，付费用户=1.5，企业用户=3；
• $P(pay\_amount)$是付费额外预算，每消费1元增加10步；
• $\beta$是全局调整系数，可根据系统负载动态调整，高峰期可降低到0.7减少资源消耗。

2. 动态预算调整模型

在任务执行过程中，如果判定任务进展正常但剩余预算不足，可以动态追加预算$\Delta B$：
$$\Delta B=\alpha \ast P(完成|当前状态)\ast S_{remaining}$$
其中：

• $\alpha$是调整系数，默认0.8，避免过度追加；
• $P(完成|当前状态)$是大模型判断的任务完成概率，0~1之间；
• $S_{remaining}$是预估完成任务还需要的步数。

3. 最大成本控制模型

通过步数预算可以精准预估每个任务的最大成本：
$$C_{max}=B_{max}\ast (c_{llm}+c_{tool})$$
其中$c_{llm}$是每步平均大模型成本，$c_{tool}$是每步平均工具调用成本，$B_{max}$是任务的最大预算（初始预算+最大追加预算）。

系统架构设计

一个生产可用的步数预算模块包含四个核心组件，整体架构如下：

各组件的职责：

1. 预算分配模块：负责根据任务属性分配初始预算，处理预算追加申请；
2. 步数计量模块：负责统计每一步的执行，扣减预算，记录步数明细；
3. 终止决策模块：负责判断是否需要终止任务，包括预算耗尽、异常循环检测等；
4. 审计溯源模块：负责落所有预算变更、步数执行、终止事件的日志，用于后续优化规则。

核心算法流程

步数预算的完整执行流程如下：

实战实现：从零搭建带步数预算的Harness

我们用Python实现一个极简的生产可用Harness框架，包含完整的步数预算能力。

环境安装

首先安装依赖：

pip install openai pydantic python-dotenv

然后在项目根目录创建.env文件，填入你的OpenAI API Key：

OPENAI_API_KEY=你的API Key

核心代码实现

import os
import json
import uuid
from typing import Optional, List, Dict
from pydantic import BaseModel
from dotenv import load_dotenv
import openai

# 加载环境变量
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# 数据模型定义
class Task(BaseModel):
    task_id: str
    description: str
    task_type: str
    user_level: str = "normal"
    status: str = "pending"
    initial_budget: int
    remaining_budget: int
    used_steps: int = 0
    context: List[Dict] = []
    result: Optional[str] = None
    total_cost: float = 0.0

class StepRecord(BaseModel):
    step_id: str
    task_id: str
    step_number: int
    action_type: str
    action_content: str
    token_usage: float
    tool_cost: float
    timestamp: float

class BudgetRecord(BaseModel):
    record_id: str
    task_id: str
    change_amount: int
    change_reason: str
    approved: bool
    timestamp: float

class AgentHarness:
    def __init__(self, global_max_budget: int = 100):
        self.global_max_budget = global_max_budget
        self.tasks: Dict[str, Task] = {}
        self.step_records: List[StepRecord] = []
        self.budget_records: List[BudgetRecord] = []
        # 基础预算配置
        self.base_budget = {
            "qa": 5,
            "tool_use": 20,
            "reasoning": 50,
            "data_process": 30
        }
        # 用户等级系数
        self.user_level_coeff = {
            "normal": 1.0,
            "vip": 2.0,
            "enterprise": 3.0
        }

    def _calculate_initial_budget(self, task_type: str, user_level: str, custom_budget: Optional[int] = None) -> int:
        """计算初始预算"""
        if custom_budget:
            return min(custom_budget, self.global_max_budget)
        base = self.base_budget.get(task_type, 10)
        coeff = self.user_level_coeff.get(user_level, 1.0)
        budget = int(base * coeff)
        return min(budget, self.global_max_budget)

    def create_task(self, task_desc: str, task_type: str = "qa", user_level: str = "normal", custom_budget: Optional[int] = None) -> str:
        """创建任务"""
        task_id = str(uuid.uuid4())
        budget = self._calculate_initial_budget(task_type, user_level, custom_budget)
        system_prompt = f"""你是一个AI助手，需要完成用户的任务：{task_desc}
        你可以调用搜索工具，输出格式要求：
        1. 如果要调用工具，输出：调用工具：search 参数：{{"keyword": "搜索关键词"}}
        2. 如果要返回最终结果，输出：最终答案：你的结果
        注意：你每执行一步会消耗1步预算，预算耗尽时必须返回最终结果，不能继续调用工具。
        """
        task = Task(
            task_id=task_id,
            description=task_desc,
            task_type=task_type,
            user_level=user_level,
            initial_budget=budget,
            remaining_budget=budget,
            context=[{"role": "system", "content": system_prompt}]
        )
        self.tasks[task_id] = task
        return task_id

    def _call_llm(self, messages: List[Dict]) -> tuple[str, float]:
        """调用大模型，返回输出和Token消耗"""
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0
        )
        content = response.choices[0].message.content
        token_usage = response.usage.total_tokens * 0.0000015 # gpt-3.5-turbo 输入输出均价
        return content, token_usage

    def _call_tool(self, tool_name: str, params: Dict) -> tuple[str, float]:
        """调用工具，返回结果和成本"""
        if tool_name == "search":
            # 模拟会出错的搜索工具，用于复现循环
            keyword = params.get("keyword", "")
            if "特殊字符" in keyword:
                return "搜索成功：结果xxx", 0.01
            return "错误：搜索关键词不能包含特殊字符，请修改后重试", 0.01
        return "未知工具", 0.0

    def _parse_action(self, llm_output: str) -> tuple[bool, Optional[str], Optional[Dict]]:
        """解析大模型输出，判断是调用工具还是返回结果"""
        if "最终答案：" in llm_output:
            return False, llm_output.split("最终答案：")[-1].strip(), None
        if "调用工具：" in llm_output:
            try:
                tool_part = llm_output.split("调用工具：")[-1]
                tool_name = tool_part.split("参数：")[0].strip()
                params_str = tool_part.split("参数：")[-1].strip()
                params = json.loads(params_str)
                return True, tool_name, params
            except Exception as e:
                return False, f"工具调用格式错误：{str(e)}", None
        return False, llm_output, None

    def _detect_cycle(self, task: Task) -> bool:
        """异常循环检测：连续3步调用同一个工具相同参数则判定为循环"""
        if len(task.context) < 6:
            return False
        last_3_actions = []
        for i in range(-2, -7, -2):
            content = task.context[i]["content"]
            if "调用工具：" in content:
                last_3_actions.append(content.split("调用工具：")[-1].strip())
            else:
                return False
        return len(set(last_3_actions)) == 1

    def run_task(self, task_id: str) -> str:
        """执行任务"""
        import time
        task = self.tasks[task_id]
        task.status = "running"
        while task.remaining_budget > 0:
            # 调用大模型
            llm_output, token_cost = self._call_llm(task.context)
            # 解析动作
            need_tool, action_content, params = self._parse_action(llm_output)
            if not need_tool:
                task.result = action_content
                task.status = "success"
                task.total_cost += token_cost
                return task.result
            # 扣减预算
            task.remaining_budget -= 1
            task.used_steps += 1
            # 执行工具调用
            tool_result, tool_cost = self._call_tool(action_content, params)
            # 记录步数
            step_record = StepRecord(
                step_id=str(uuid.uuid4()),
                task_id=task_id,
                step_number=task.used_steps,
                action_type="tool_call",
                action_content=f"{action_content} {json.dumps(params)}",
                token_usage=token_cost,
                tool_cost=tool_cost,
                timestamp=time.time()
            )
            self.step_records.append(step_record)
            # 更新成本
            task.total_cost += token_cost + tool_cost
            # 更新上下文
            task.context.append({"role": "assistant", "content": llm_output})
            task.context.append({"role": "user", "content": f"工具返回结果：{tool_result}"})
            # 检测循环
            if self._detect_cycle(task):
                task.result = f"检测到无限循环，任务终止，已用步数：{task.used_steps}，总成本：{task.total_cost:.4f}元"
                task.status = "failed"
                return task.result
        # 预算耗尽
        task.result = f"任务失败：步数预算耗尽，已用步数：{task.used_steps}，总成本：{task.total_cost:.4f}元"
        task.status = "failed"
        return task.result

# 测试代码
if __name__ == "__main__":
    harness = AgentHarness(global_max_budget=20)
    # 创建一个工具调用类任务，模拟循环场景
    task_id = harness.create_task(
        task_desc="搜索关键词'1到100的和'，得到结果后返回最终答案",
        task_type="tool_use",
        user_level="normal"
    )
    result = harness.run_task(task_id)
    print(result)
    task = harness.tasks[task_id]
    print(f"初始预算：{task.initial_budget}，剩余预算：{task.remaining_budget}，已用步数：{task.used_steps}，总成本：{task.total_cost:.4f}元")

运行结果说明

运行上述代码，你会得到类似如下的输出：

检测到无限循环，任务终止，已用步数：3，总成本：0.0123元
初始预算：20，剩余预算：17，已用步数：3，总成本：0.0123元

可以看到，步数预算机制在检测到连续3步相同的工具调用后，提前终止了任务，避免了无限循环造成的成本浪费。

---

四、进阶探讨与最佳实践

常见陷阱与避坑指南

1. 步数计量规则不清晰

很多团队踩过的坑：把系统级重试、用户等待时间都算入步数，导致正常任务被误杀。
避坑方案：明确步数计量规则，只有Agent主动发起的动作迭代才算步数，系统自动重试、用户输入等待时间一律不算，并且把规则同步给所有开发和用户。

2. 预算分配不合理

初始预算设太高起不到控制作用，设太低又会影响正常任务的完成。
避坑方案：用历史数据校准预算规则，统计不同类型任务的步数分布，把初始预算设为95分位值，保证95%的正常任务都能在预算内完成，剩下5%的复杂任务走预算追加流程。

3. 只依赖预算耗尽终止，没有异常检测

很多团队只做了预算扣减，没有提前检测循环，导致即使出现循环也要等预算耗尽才终止，浪费成本。
避坑方案：叠加异常检测规则，比如连续3步相同动作、连续5步没有进展、工具调用错误率超过80%等，只要触发任意规则就提前终止。

性能优化与成本考量

1. 分层预算机制

设置三层预算防护，最大化控制成本：

• 全局预算：整个系统单日最大步数，避免整体成本超支；
• 单任务预算：每个任务的最大步数，避免单个任务消耗过多资源；
• 单步骤预算：每一步的最大Token消耗和工具调用成本，避免单步成本过高。

2. 预算复用机制

对于多Agent协作场景，设置团队总预算，空闲Agent的预算可以动态分配给忙的Agent，提高预算利用率，避免资源浪费。

3. 预算透明化

给用户展示当前任务的步数使用情况、剩余预算、预估完成还需要的步数，让用户可以自主选择是否追加预算，既提升用户体验，又减少投诉。

最佳实践总结

我们结合多家企业的落地经验，总结了以下5条步数预算的黄金法则：

1. 永远不要上线没有步数预算的Agent服务：这是底线，无限循环造成的成本损失可能远超你的想象；
2. 优先用异常检测提前终止循环：不要等预算耗尽才终止，提前识别循环可以节省70%以上的无效成本；
3. 动态预算要配审批流程：预算追加必须经过用户或者系统审批，避免动态预算变成「无限预算」；
4. 定期审计步数数据：每个月统计步数使用情况，优化初始预算规则和异常检测规则，不断提升准确率；
5. 和其他控制机制结合使用：步数预算+超时控制+Token上限，三层防护，确保万无一失。

---

五、结论

核心要点回顾

本文从Agent无限循环的痛点出发，系统讲解了Harness层推理步数预算机制的原理、实现和最佳实践：

1. 推理步数预算是目前解决Agent无限循环问题的最优方案，比传统超时机制的粒度更细、误杀率更低、成本控制能力更强；
2. 一个完整的步数预算系统包含预算分配、步数计量、终止决策、审计溯源四个核心组件；
3. 生产环境落地时需要叠加异常检测、动态预算调整、审批流程等能力，平衡用户体验和成本控制。

行业发展与未来趋势

Agent执行治理已经成为AI服务治理的核心方向，步数预算机制也在不断进化，我们整理了其发展历程：

时间	阶段	核心能力	特点
2022年及之前	萌芽期	固定步数限制	仅支持固定最大步数，没有动态调整能力
2023年	成长期	动态步数预算	支持按任务类型、用户等级分配预算，支持动态追加
2024年	成熟期	智能预算+异常检测	结合大模型判断任务进展，自动调整预算，提前识别异常
2025年及以后	智能化期	强化学习驱动的预算优化	用强化学习自动优化预算分配规则，实现成本和体验的最优平衡

未来，步数预算会成为所有Agent平台的标配能力，和AI成本治理、服务可用性治理深度融合，成为企业级AI应用的基础设施。

行动号召

如果你正在开发Agent应用，现在就可以把本文提供的Harness代码集成到你的项目里，只需要加几十行代码就能避免99%的无限循环风险。如果你有任何问题或者更好的实践，欢迎在评论区交流。

学习资源

• LangChain 最大迭代次数配置文档
• AutoGPT 步数限制配置指南
• Harness 官方AI治理模块

---

本文字数：约10800字