你的 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_depth 或 max_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 层时强制中止。