用会话追踪与成本护栏定位 AI Agent 失控问题

2026-09-11 34 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:9 分钟

AI Agent 的故障往往不是一次明确的异常,而是一段逐渐偏离预期的执行过程:模型反复调用同一个工具、在多个步骤之间循环,或者持续消耗 Token 和外部 API 配额。单看最终错误信息,很难还原问题是从哪一步开始的。

会话追踪(session trace)和成本控制正在成为两项互补的可观测性手段:前者保留足够的执行上下文,帮助事故后复盘;后者在循环演变成高额账单之前主动终止任务。

为什么普通日志看不清 Agent 故障

传统服务通常围绕一次请求记录日志,而 Agent 的一次用户请求可能包含多轮模型推理、工具调用和状态更新。只记录最终响应,会遗漏最关键的中间过程。

一条实用的会话追踪至少应该能回答这些问题:

  • 这次执行属于哪个 session_id
  • 模型按什么顺序调用了哪些工具?
  • 相同工具和参数是否连续出现?
  • 每一步产生了多少估算成本,累计成本是多少?
  • 是工具报错、预算超限,还是循环检测器终止了任务?
  • 失败前最后一个有效状态是什么?

追踪事件应带有统一的会话标识和递增序号。这样,即使事件被发送到异步日志系统,也能重新组成完整时间线。

需要注意,追踪不等于把提示词、工具参数和返回值全部写入日志。原始内容可能包含个人信息、访问令牌或业务数据。更稳妥的做法是记录参数结构、内容长度、哈希指纹和受控的快照引用;只有经过授权的调试流程才能读取原文。

两道护栏:循环检测与会话预算

工具调用循环常见的形态是:Agent 得到一个未满足预期的结果,却没有改变计划,随后使用完全相同的参数再次调用工具。一个简单但有效的保护措施,是计算“工具名 + 规范化参数”的签名,并统计连续重复次数。

成本护栏则应在调用模型或付费工具之前检查预计支出。常见层级包括:

  • 单次调用最大成本;
  • 单个 Agent 会话累计预算;
  • 单个用户或租户的小时、日预算;
  • 工具调用次数和模型推理轮数上限。

硬预算适合阻止明显失控,软预算则可以触发告警、降低模型规格或要求人工确认。两者不应只返回一个笼统的“任务失败”,而要向追踪系统写入明确的终止原因,例如 tool_loopbudget_exceeded

可以这样实践:一个可运行的最小守卫

下面是一个基于 Python 标准库的简化示例。它不是特定 Agent 框架的官方接口,而是一种可以移植到工具执行层的实现方式。代码会把事件写入 agent-trace.jsonl,阻止连续三次相同调用,并限制单会话估算成本。

将代码保存为 agent_trace_demo.py,使用 Python 3.10 或更高版本运行:

import hashlib
import json
import os
import time
import uuid
from pathlib import Path

TRACE_FILE = Path('agent-trace.jsonl')
LOOP_LIMIT = int(os.getenv('LOOP_LIMIT', '3'))
BUDGET_USD = float(os.getenv('BUDGET_USD', '0.01'))
TOOL_PRICES = {'search': 0.002}


class AgentStopped(RuntimeError):
    pass


class GuardedSession:
    def __init__(self):
        self.session_id = str(uuid.uuid4())
        self.sequence = 0
        self.spend = 0.0
        self.last_signature = None
        self.repeat_count = 0
        self.emit('session_started', budget_usd=BUDGET_USD)

    def emit(self, event, **fields):
        self.sequence += 1
        record = {
            'timestamp': time.time(),
            'session_id': self.session_id,
            'sequence': self.sequence,
            'event': event,
            **fields,
        }
        with TRACE_FILE.open('a', encoding='utf-8') as handle:
            handle.write(json.dumps(record, ensure_ascii=False) + '\n')

    @staticmethod
    def fingerprint(tool, arguments):
        normalized = json.dumps(arguments, sort_keys=True, ensure_ascii=False)
        digest = hashlib.sha256(normalized.encode()).hexdigest()[:16]
        return f'{tool}:{digest}'

    def call_tool(self, tool, arguments, function):
        signature = self.fingerprint(tool, arguments)
        self.repeat_count = (
            self.repeat_count + 1 if signature == self.last_signature else 1
        )
        self.last_signature = signature

        estimated_cost = TOOL_PRICES.get(tool, 0.0)
        projected_spend = self.spend + estimated_cost

        self.emit(
            'tool_call_planned',
            tool=tool,
            argument_keys=sorted(arguments),
            argument_fingerprint=signature,
            repeat_count=self.repeat_count,
            projected_spend_usd=projected_spend,
        )

        if self.repeat_count >= LOOP_LIMIT:
            self.emit('guard_blocked', reason='tool_loop', tool=tool)
            raise AgentStopped('Repeated tool-call loop detected')

        if projected_spend > BUDGET_USD:
            self.emit('guard_blocked', reason='budget_exceeded', tool=tool)
            raise AgentStopped('Session budget exceeded')

        self.spend = projected_spend
        result = function(**arguments)
        self.emit(
            'tool_call_completed',
            tool=tool,
            result_type=type(result).__name__,
            estimated_spend_usd=self.spend,
        )
        return result


def fake_search(query):
    return {'items': [], 'query_length': len(query)}


session = GuardedSession()
try:
    for _ in range(10):
        session.call_tool('search', {'query': 'agent tracing'}, fake_search)
except AgentStopped as error:
    print(f'Session stopped: {error}')
    print(f'Trace written to: {TRACE_FILE}')

运行并查看追踪文件:

rm -f agent-trace.jsonl
python agent_trace_demo.py
cat agent-trace.jsonl

默认配置会在第三次相同工具调用之前触发循环保护。若要单独观察预算控制,可以临时提高循环阈值并降低预算:

rm -f agent-trace.jsonl
LOOP_LIMIT=99 BUDGET_USD=0.005 python agent_trace_demo.py
cat agent-trace.jsonl

这个示例只存储参数键名和哈希指纹,没有把查询内容直接写入追踪文件。生产环境还可以把模型版本、提示词模板版本、工具延迟、重试次数和父子步骤 ID 加入事件,但应避免无限制记录模型输入输出。

从演示代码走向生产系统

简单的签名比较只能识别参数完全相同的连续调用。Agent 也可能通过修改时间戳、分页参数或措辞形成“语义循环”,绕过这种规则。生产系统通常还需要结合滑动时间窗口、工具调用图、无进展状态计数,以及任务级最大步骤数。

成本数据也有边界。执行期间计算的通常是估算值,不能替代供应商最终账单。模型价格可能变化,缓存命中、批处理和工具侧计费也会造成偏差,因此应定期对账并更新价格表。

落地时可以按以下清单推进:

  • 为每次 Agent 执行生成稳定的会话 ID,并让模型调用、工具调用和状态更新共享该 ID;
  • 在真正发起外部调用前执行循环、次数和预算检查;
  • 用结构化事件记录“计划、完成、失败、被护栏阻止”四类结果;
  • 对提示词、工具参数和返回值实施脱敏、采样、加密与保留期限;
  • 同时配置会话级硬上限和租户级告警,避免单次事故扩散;
  • 定期用历史失败会话回放检测规则,检查误报和漏报。

好的 Agent 可观测性并不是记录得越多越好,而是在成本、隐私和可调试性之间做出明确选择。会话追踪负责留下证据,成本护栏负责缩小事故半径;只有把两者放进同一条执行链路,团队才能既看清失败,也及时阻止失控。


相关推荐