Agent 终态判定：何时该停止思考、给出最终回复

一、你的 Agent 在"再想想"的循环里绕了 12 轮，用户已经关窗口了

Agent 与人最大的区别是：人知道什么时候该停下来给答案，Agent 会一直"想"下去。你给 Agent 接了搜索引擎 API，它第一轮搜了一下，觉得信息不够全，第二轮换了关键词再搜，第三轮觉得 "还有 3 篇相关文章没读"，第四轮读了之后发现"这篇文章引用了另一篇——我再搜一下"……十二轮后用户早就不在了。

Agent 的终止判定是一个比想象中更复杂的问题。不是因为怎么停止很难，而是"什么情况下应该停止"需要几个维度的判断：信息完备度够不够、用户是否有明确的时间预期、本轮回答的质量是否已经"够好"。

当前大多数 Agent 框架的终止逻辑极其简单。要么是固定步数限制（最多调用 5 步就停），要么是 LLM 自己说了算（它觉得该停了就停了）。固定步数太硬——复杂任务 5 步不够，简单任务 1 步就够了。LLM 自己判断太软——它可能永远不觉得"信息够了"。

二、底层机制与原理剖析

Agent 终态判定的多维度决策模型：

四个判定维度：

信息增益（Information Gain）：每一步工具调用都会带回新的信息。衡量这一步带来的信息增量——如果新信息与已收集信息的语义重复度超过 95%，说明再搜下去也不会获得新信息了（边际收益递减）。计算方法：新信息的嵌入向量 vs 已收集信息的嵌入向量之间的余弦相似度。

步数预算：不是固定上限（如 5 步），而是动态预算。简单问题（如"今天天气怎么样"）1-2 步足够，复杂问题（如"分析 A 公司财报并做竞品对比"）给 8-10 步。LLM 在初始阶段评估任务复杂度，生成预算值。

置信度阈值：每一步后让 LLM 自评当前答案的置信度（0-100%）。超过阈值（如 85%）就输出答案；未超过就继续。这比固定步数更弹性——有时 1 步就能高置信度回答，有时 8 步才能。

用户显式信号：最高优先级的停止信号。用户的任何"结束意图"的表达（"可以了""够了""先这样吧""明白了"）都应该立即停止推理并输出最终回答。

三、生产级代码实现

"""
Agent 终态判定器
四个维度：信息增益、步数预算、置信度阈值、用户显式信号
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
import numpy as np
from enum import Enum

class StopReason(Enum):
    INFORMATION_SATURATED = "info_saturated"   # 信息饱和
    BUDGET_EXCEEDED = "budget_exceeded"         # 步数预算耗尽
    CONFIDENCE_REACHED = "confidence_reached"   # 置信度达标
    USER_SIGNAL = "user_signal"                 # 用户显式终止
    MAX_STEPS_FORCED = "max_steps_forced"       # 硬上限强制终止

@dataclass
class StopDecision:
    """终态判定结果"""
    should_stop: bool
    reason: StopReason
    confidence: float             # 当前置信度
    information_gain: float       # 最后一步的信息增益
    steps_used: int               # 已用步数
    steps_budget: int             # 总步数预算
    detail: str                   # 人类可读的决策说明

@dataclass
class AgentContext:
    """Agent 当前步骤的上下文"""
    steps_taken: int
    steps_budget: int
    collected_info: List[str]     # 已收集的信息片段
    last_action_result: str       # 最后一步动作的结果
    current_confidence: float     # 当前置信度 (0-100)
    user_last_message: str        # 用户最后一条消息

class TerminationJudge:
    """Agent 终态判定器"""
    
    def __init__(
        self,
        embed_fn,
        confidence_threshold: float = 85.0,
        information_gain_threshold: float = 0.05,
        max_steps_fallback: int = 15,     # 绝对硬上限
    ):
        self.embed_fn = embed_fn
        self.confidence_threshold = confidence_threshold
        self.information_gain_threshold = information_gain_threshold
        self.max_steps_fallback = max_steps_fallback
        
        # 用户终止信号词
        self.stop_signals = [
            "可以了", "够了", "先这样", "明白了", "清楚了",
            "不用了", "够了谢谢", "就这样吧", "好的谢谢",
            "ok", "got it", "that's enough", "stop",
        ]
    
    def judge(self, context: AgentContext) -> StopDecision:
        """
        判定是否应该停止
        
        判定优先级：用户信号 > 信息饱和 > 置信度达标 > 步数预算 > 硬上限
        """
        
        # === 优先级 1: 用户显式信号 ===
        for signal in self.stop_signals:
            if signal in context.user_last_message.lower():
                return StopDecision(
                    should_stop=True,
                    reason=StopReason.USER_SIGNAL,
                    confidence=context.current_confidence,
                    information_gain=0,
                    steps_used=context.steps_taken,
                    steps_budget=context.steps_budget,
                    detail=f"用户发出了终止信号: '{signal}'",
                )
        
        # === 优先级 2: 信息饱和（最后一步的信息增益太低）===
        info_gain = self._calculate_information_gain(
            context.last_action_result,
            context.collected_info,
        )
        
        if context.steps_taken >= 2 and info_gain < self.information_gain_threshold:
            return StopDecision(
                should_stop=True,
                reason=StopReason.INFORMATION_SATURATED,
                confidence=context.current_confidence,
                information_gain=info_gain,
                steps_used=context.steps_taken,
                steps_budget=context.steps_budget,
                detail=f"信息增益 {info_gain:.3f} 低于阈值 {self.information_gain_threshold}，继续搜索收益递减",
            )
        
        # === 优先级 3: 置信度达标 ===
        if context.current_confidence >= self.confidence_threshold:
            return StopDecision(
                should_stop=True,
                reason=StopReason.CONFIDENCE_REACHED,
                confidence=context.current_confidence,
                information_gain=info_gain,
                steps_used=context.steps_taken,
                steps_budget=context.steps_budget,
                detail=f"置信度 {context.current_confidence}% >= 阈值 {self.confidence_threshold}%",
            )
        
        # === 优先级 4: 步数预算耗尽 ===
        if context.steps_taken >= context.steps_budget:
            return StopDecision(
                should_stop=True,
                reason=StopReason.BUDGET_EXCEEDED,
                confidence=context.current_confidence,
                information_gain=info_gain,
                steps_used=context.steps_taken,
                steps_budget=context.steps_budget,
                detail=f"已用 {context.steps_taken}/{context.steps_budget} 步，预算耗尽",
            )
        
        # === 优先级 5: 绝对硬上限 ===
        if context.steps_taken >= self.max_steps_fallback:
            return StopDecision(
                should_stop=True,
                reason=StopReason.MAX_STEPS_FORCED,
                confidence=context.current_confidence,
                information_gain=info_gain,
                steps_used=context.steps_taken,
                steps_budget=self.max_steps_fallback,
                detail=f"达到绝对上限 {self.max_steps_fallback} 步，强制终止",
            )
        
        # === 继续 ===
        return StopDecision(
            should_stop=False,
            reason=StopReason.INFORMATION_SATURATED,  # 实际未触发
            confidence=context.current_confidence,
            information_gain=info_gain,
            steps_used=context.steps_taken,
            steps_budget=context.steps_budget,
            detail="继续下一步推理",
        )
    
    def _calculate_information_gain(
        self, new_info: str, existing_info: List[str]
    ) -> float:
        """
        计算新信息相对于已有信息的信息增益
        
        方法：计算新信息向量与已有信息向量的最大余弦相似度。
        1 - 最大相似度 = 信息增益。
        新信息与已有信息越相似，增益越低。
        """
        if not existing_info or not new_info:
            return 1.0  # 没有旧信息，新信息的增益是 100%
        
        try:
            new_embed = self.embed_fn(new_info)
            
            # 计算与每条已有信息的相似度，取最大值
            max_similarity = 0.0
            for info in existing_info[-5:]:  # 只比较最近 5 条，降低计算量
                try:
                    info_embed = self.embed_fn(info)
                    sim = self._cosine_sim(new_embed, info_embed)
                    max_similarity = max(max_similarity, sim)
                except Exception:
                    continue
            
            # 增益 = 1 - 最大相似度
            gain = 1.0 - max_similarity
            return max(0.0, gain)
            
        except Exception:
            return 0.5  # 计算失败时返回中性值
    
    @staticmethod
    def _cosine_sim(a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8)
    
    def estimate_budget(self, user_query: str) -> int:
        """
        根据用户问题复杂度估算步数预算
        
        简单启发式（生产环境应用 LLM 估计复杂度）：
        - 短问题（< 20 字）：2-3 步
        - 中等问题（20-50 字）：4-6 步
        - 长问题（> 50 字）：7-10 步
        """
        length = len(user_query)
        if length < 20:
            return 3
        elif length < 50:
            return 6
        elif length < 100:
            return 8
        else:
            return 10
    
    def build_stop_message(self, decision: StopDecision) -> str:
        """根据终止原因构建给用户的说明"""
        if decision.reason == StopReason.BUDGET_EXCEEDED:
            return (
                f"（已进行 {decision.steps_used} 步分析，"
                f"当前置信度：{decision.confidence:.0f}%。"
                f"如需更深入的分析，可以提出更具体的问题。）"
            )
        elif decision.reason == StopReason.CONFIDENCE_REACHED:
            return ""
        elif decision.reason == StopReason.INFORMATION_SATURATED:
            return (
                f"（搜索到的信息开始重复，"
                f"当前置信度：{decision.confidence:.0f}%。"
                f"如需更多信息源，可以指定方向。）"
            )
        else:
            return ""

四、边界分析与架构权衡

信息增益的准确性：

用 embedding 向量计算语义相似度作为信息增益的代理变量，会产生误判。两段文字在语义上相似但包含了关键的差异信息（如两个不同城市的人口数据——结构相同但数据不同）——语义相似度高但信息增益应该高。这是当前方法的盲区。

置信度评分的可靠性：

让 LLM 自己评估自己的置信度，存在"过度自信"或"不自信"的偏差。模型的置信度评估本身就是不可靠的——它可能对错误答案给出 90% 的置信度。可以用 Self-Consistency 方法（让模型多次回答同一问题，统计答案的一致程度）来增加置信度评分的可信度。

适用边界：

最适合多步推理的 Agent（如研究助手、数据分析 Agent、代码调试 Agent）。任务步数在 3-15 步之间、需要多轮信息检索或分析的场景。

禁用场景：

不适合单轮问答的 Agent——不需要多步推理的场景不存在终止判定的问题。也不适合实时对话——用户应该在任意时刻都能打断 Agent。

五、总结

Agent 终态判定不是简单的最大步数限制。四个维度的综合决策——用户显式信号（优先级最高）、信息增益（边际收益判断）、置信度阈值（答案质量判断）、步数预算（资源边界）。当任一维度触发时，Agent 应在回答中附上置信度和停止理由，让用户知道"这个回答是经过几轮推理得出的，可信度如何"。关键是让 Agent 能在"过度推理"和"草率给出答案"之间找到平衡。