Agent 调用图出现循环但没人发现

多 Agent 系统中,Agent 之间的调用关系形成了有向环,导致任务无限循环、token 持续消耗,直到预算耗尽才暴露问题。本文分析循环产生的原因并给出静态检测和运行时防护方案。

你的 AutoGen 系统里,Planner Agent 把「验证」任务分发给 Validator Agent,Validator 发现问题后把「修复」任务交给 Fixer Agent,Fixer 修复后把「重新验证」任务交回给 Planner,Planner 再次交给 Validator——一个完美的闭环,没有任何退出条件。或者在动态生成任务图的 CrewAI 流水线里,两个 Agent 的目标互相依赖(Agent A 需要 B 的输出作为输入,Agent B 的任务目标包含了「基于 A 的输出改进」),形成隐式的循环调用,没有任何框架告警,只是 token 在静静地消耗。

常见原因

1. 条件退出检查缺失或条件永远不满足

循环存在「Planner → Validator → Fixer → Planner」,但退出条件是「验证 100% 通过」,而实际上的代码问题永远无法达到 100%(可能因为验证器本身有 bug,或者任务本身不可能完全满足)。循环永远不会退出。

怎么判断:找到循环路径的退出条件,用已知的「难以修复」的测试输入验证是否能触发退出。如果退出条件在实际数据上从未被满足,就是条件过于严格。

2. 动态任务生成没有深度限制

Agent A 在处理任务时决定生成子任务,子任务被分配给 Agent B,Agent B 在处理时也生成了新的子任务分给 Agent A,任务树无限增长。没有 max_depthmax_tasks 的限制,系统不会检测到这是循环。

怎么判断:追踪任务的「父任务链」,如果某个任务的祖先链里出现了与自身相同的 task_type,就是在循环。

3. Agent 的工具列表里包含了「调用另一个 Agent」的工具,但没有环路检测

在 OpenAI Swarm 或自定义框架里,Agent 的工具集里包含 call_agent_b(),Agent B 的工具集里包含 call_agent_a()。编排器没有对 Agent 调用关系做全局的拓扑分析,不知道这两个工具调用会形成环。

怎么判断:把所有「agent-to-agent 调用」的工具画成有向图,对图做拓扑排序,有环则说明存在潜在的循环调用路径。

4. 重试机制触发了无效循环

任务失败后被重新路由,路由逻辑把失败的任务发给了另一个 Agent,那个 Agent 失败后路由回来,形成「失败-路由-失败」的循环。这在每个 Agent 对该任务都「力不从心」时会无限循环下去。

怎么判断:统计同一个任务 ID 被多少个不同的 Agent 处理过,如果同一个任务经历了 3 个以上不同的 Agent,就可能是在兜圈子。

5. 循环在 Trace 里被折叠显示,看起来像正常的多次调用

Langfuse 或 LangSmith 对重复的 span 会做折叠显示,把「相同节点被调用了 15 次」显示为「调用了 1 次」。开发者在看 Trace 时没有注意到调用次数,误以为流程正常。

怎么判断:展开 Trace 的原始 JSON,统计同一个 Agent 的 span 数量,而不是依赖 UI 的折叠视图。

6. 版本升级后新增了 Agent 间的依赖,但没有重新检测拓扑

最初的 6 个 Agent 的依赖图是 DAG(无环),但功能迭代中给某个 Agent 加了一个新的工具(调用了另一个 Agent),引入了环。没有任何 CI 检查会在 PR 合并时重新验证 Agent 依赖图是否仍然是 DAG。

怎么判断:对比当前版本和上个发版的 Agent 依赖图,找出新增的 Agent-to-Agent 调用边,用拓扑排序验证是否引入了环。

最短修复路径

Step 1:在每次任务启动时做运行时调用链检测

from collections import defaultdict

class CallChainTracker:
    """运行时追踪 Agent 调用链,检测循环。"""
    
    def __init__(self, max_depth: int = 10):
        self.max_depth = max_depth
        self._chain: list[str] = []  # 当前调用链
    
    def enter_agent(self, agent_id: str):
        # 检测循环:当前 agent 是否已经在调用链里
        if agent_id in self._chain:
            cycle_start = self._chain.index(agent_id)
            cycle = self._chain[cycle_start:] + [agent_id]
            raise CycleDetected(
                f"检测到 Agent 调用循环:{'→'.join(cycle)}"
            )
        
        # 检测深度
        if len(self._chain) >= self.max_depth:
            raise DepthLimitExceeded(
                f"Agent 调用深度超过上限 {self.max_depth},当前链:{'→'.join(self._chain)}"
            )
        
        self._chain.append(agent_id)
    
    def exit_agent(self, agent_id: str):
        if self._chain and self._chain[-1] == agent_id:
            self._chain.pop()

# 每个任务一个 tracker 实例
tracker = CallChainTracker(max_depth=8)

def call_agent(agent_id: str, task: dict, tracker: CallChainTracker):
    tracker.enter_agent(agent_id)
    try:
        return AGENTS[agent_id].run(task, tracker=tracker)
    finally:
        tracker.exit_agent(agent_id)

Step 2:启动前对静态依赖图做拓扑排序

def build_agent_dependency_graph(agent_configs: list[dict]) -> dict[str, list[str]]:
    """从 Agent 的工具配置里提取 agent-to-agent 调用关系。"""
    graph = {a["id"]: [] for a in agent_configs}
    
    for agent in agent_configs:
        for tool in agent.get("tools", []):
            if tool.get("type") == "agent_call":
                graph[agent["id"]].append(tool["target_agent"])
    
    return graph

def assert_no_cycles(graph: dict[str, list[str]]):
    """Kahn 算法拓扑排序,有环时抛异常。"""
    from collections import deque
    in_degree = {n: 0 for n in graph}
    for node, neighbors in graph.items():
        for neighbor in neighbors:
            in_degree[neighbor] = in_degree.get(neighbor, 0) + 1
    
    queue = deque([n for n, d in in_degree.items() if d == 0])
    visited = 0
    
    while queue:
        node = queue.popleft()
        visited += 1
        for neighbor in graph.get(node, []):
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    
    if visited < len(graph):
        cycle_nodes = [n for n, d in in_degree.items() if d > 0]
        raise CyclicDependency(
            f"Agent 依赖图有环,涉及节点:{cycle_nodes}"
        )

# 在应用启动时运行
graph = build_agent_dependency_graph(AGENT_CONFIGS)
assert_no_cycles(graph)  # 有环则拒绝启动

Step 3:为任务循环设置计数器和硬性退出条件

class LoopGuard:
    """防止无限循环的任务计数器。"""
    
    def __init__(self, max_iterations: int = 20):
        self.max_iterations = max_iterations
        self._counts: dict[str, int] = defaultdict(int)
    
    def check_and_increment(self, task_type: str):
        self._counts[task_type] += 1
        if self._counts[task_type] > self.max_iterations:
            raise LoopLimitExceeded(
                f"任务类型 '{task_type}' 已执行 {self._counts[task_type]} 次,"
                f"超过上限 {self.max_iterations},强制终止以防无限循环"
            )
    
    def summary(self) -> dict:
        return dict(self._counts)

Step 4:在循环路径上加有效的退出条件

# 反例:退出条件永远不满足
# while not validator.is_perfect(output):  # "完美" 太难达到
#     output = fixer.fix(output)

# 正例:有界的迭代
MAX_FIX_ROUNDS = 3

async def validate_and_fix(output: str) -> str:
    for round_num in range(1, MAX_FIX_ROUNDS + 1):
        issues = await validator.check(output)
        
        if not issues:
            return output  # 无问题,退出循环
        
        critical_issues = [i for i in issues if i["severity"] == "critical"]
        if not critical_issues:
            # 只有非关键问题,接受当前输出
            return output
        
        if round_num == MAX_FIX_ROUNDS:
            # 达到最大轮数,返回最好的结果并记录未解决的问题
            logger.warning(f"达到最大修复轮数 {MAX_FIX_ROUNDS},仍有 {len(critical_issues)} 个关键问题未解决")
            return output
        
        output = await fixer.fix(output, issues=critical_issues)
    
    return output

Step 5:在 CI 里对 Agent 依赖图做回归测试

# .github/workflows/agent-dag-check.yml
- name: Check Agent dependency graph for cycles
  run: python scripts/check_agent_dag.py --fail-on-cycle

预防建议

  • 在应用启动时对所有 Agent 的调用依赖做静态拓扑排序,有环则拒绝启动
  • 为所有循环路径(有意为之的循环,如 validate-fix-validate)设置有界的最大迭代次数,而不是用「完美」这类无法达到的退出条件
  • 在运行时追踪每个任务的 Agent 调用链,调用链里出现重复 Agent ID 时立即中止
  • 每次新增 Agent 间的工具调用时,在 CI 里自动检测是否引入了新的有向环
  • 为整个 workflow 设置总计调用次数上限(如 100 次 LLM 调用),超过后强制终止
  • 在 Trace 视图里展示调用次数的原始数字而不是折叠视图,方便发现异常的高频调用
  • 对每个 Agent 的单一任务设置最大执行时间(如 10 分钟),超时后强制中止并告警

常见问答 (FAQ)

Q: LangGraph 的 conditional edge 能形成合法的循环(如 validate-fix-validate)吗? A: 能,LangGraph 支持有环的图(通过 conditional edge 实现循环)。但必须在 conditional edge 的路由函数里有明确的终止条件(如迭代计数达到上限,或质量分数超过阈值)。LangGraph 不会自动检测「终止条件永远不满足」的逻辑错误,这需要开发者自己保证。

Q: 如何区分「正常的迭代优化循环」和「意外的无限循环」? A: 关键区别是是否有「收敛指标」在每次迭代后改善。正常的 validate-fix 循环,每次迭代后问题数量应该单调减少(或者至少不增加)。如果迭代了 3 次后问题数量没有变化,这个循环就可能是无效的。可以在循环条件里加「本次迭代是否有进展」的检查,无进展时直接退出。

Q: 如果 Agent 依赖图在运行时动态生成(不是静态配置的),如何做循环检测? A: 静态检测无法覆盖动态生成的依赖关系。这种情况必须依赖运行时检测:在每次 Agent 调用前,检查当前的调用链里是否已经包含了目标 Agent。CallChainTracker(见 Step 1)是针对这种情况的运行时解决方案。

Q: 调用链深度上限设置为多少合理? A: 对于「人类可以追踪和理解」的系统,建议不超过 5 层。实践中,超过 8 层的调用链通常是设计问题(任务分解粒度太细或 Agent 职责不清晰),而不是正常的业务需求。从 8 层开始,可以在 warning 级别记录,超过 10 层时强制中止。

相关阅读

标签: #AI 编程 #Agents #排查