Agent Plan × DeepSeek Harness — 自适应规划与自纠错回路

1. 引言:为什么 Agent 需要自纠错能力

在 LLM Agent 的工程实践中,一个反复出现的痛点是:一次性执行极其脆弱。传统的 Agent 框架往往采用"感知→规划→执行→输出"的线性流水线,假定环境可预测、工具调用总是成功、中间结果总是符合预期。然而真实环境远非如此——API 返回非预期结构、数据库连接超时、用户需求在中途发生漂移、第一步的假设在第三步被证伪。这些情况一旦发生,线性流水线要么静默产出错误结果,要么直接崩溃,缺乏任何自我修复的能力。

ReAct(Reasoning + Acting)范式通过将"思考"与"行动"交替进行,部分缓解了这一问题,使 Agent 能够在单步级别做出更合理的决策。但 ReAct 本质上是反应式的——它在每一步根据当前观察做局部决策,缺乏对全局计划的显式维护和动态修正能力。当任务复杂度上升、步骤数超过十步以上时,纯 ReAct 模式容易陷入"局部最优陷阱":Agent 可能在错误的方向上越走越远,因为每一步的决策在局部看来都是合理的。

Plan-and-Execute 模式则走向另一个极端:先生成一个完整的计划,再逐步执行。它的优势在于全局视野,但弱点同样明显——计划是静态的,一旦环境偏离预期,整个计划便失去效力,而 Agent 又缺乏在运行时修正计划的机制。

DeepSeek 的推理能力(尤其是其 Chain-of-Thought 推理深度和自我评估能力)为弥合这一差距提供了关键基础设施。通过将 DeepSeek 的推理能力嵌入一个结构化的"思考-执行-反思-修正"闭环,我们可以构建一个自适应规划系统:它在执行前生成计划,在执行中监控偏差,在失败后归因根因,在反思后修正策略,并将学到的经验反馈到后续规划中。这就是本文所要阐述的 Agent Plan × DeepSeek Harness 框架的核心思想。

本文将系统性地展开这一框架的设计理念、架构组件、闭环机制、动态修正策略、失败归因模型、重试策略、策略进化机制,以及 DeepSeek 推理能力的具体应用方式,最后通过一个代码生成与调试的实战案例进行端到端验证。

2. 从 ReAct 到 Plan-and-Execute:规划范式演进

要理解自适应规划的设计动机,需要先梳理 Agent 规划范式的演进脉络。

2.1 ReAct:思考与行动的交错

ReAct 由 Yao 等人在 2022 年提出,其核心思想是将推理(Reasoning)和行动(Acting)交织在一起。Agent 在每一步先生成一段"思考"(Thought),分析当前状态并决定下一步行动(Action),然后观察行动结果(Observation),再进入下一轮思考。这种模式的优势在于灵活性和情境感知——Agent 可以根据每一步的实际结果动态调整后续行为。

ReAct 的局限性在于:

  • 缺乏全局计划:Agent 只关注"下一步做什么",没有对整体任务步骤的显式规划,容易在复杂任务中迷失方向。
  • 上下文窗口压力:随着步骤增加,所有历史 Thought-Action-Observation 都需要保留在上下文中,导致 token 消耗线性增长。
  • 错误传播:某一步的错误决策会通过上下文传播到后续所有步骤,而 Agent 缺乏"回退到计划层面重新规划"的能力。

2.2 Plan-and-Execute:先规划后执行

Plan-and-Execute 模式将流程分为两个阶段:规划器(Planner)先生成一个完整的步骤列表,执行器(Executor)再逐步执行。LangChain 的 Plan-and-Execute Agent 是这一模式的典型实现。

优势在于全局视野和执行效率(执行器无需在每步重新推理)。但核心问题是刚性:计划一旦生成便固定不变,即使执行过程中发现某个步骤不可行或某中间结果改变了后续前提,计划也不会自动调整。典型的补救手段是"重新规划"(re-plan),即在执行失败后让规划器重新生成整个计划——但这种方式代价高昂,且会丢失已有执行成果的上下文。

2.3 融合范式:自适应规划

自适应规划的核心洞察是:计划不应是静态蓝图,而应是动态文档。它应该像一个经验丰富的项目经理的工作计划——有全局框架,但允许根据实际情况进行局部调整、任务重排、条件分支和紧急回滚。

下表对比了三种范式的关键差异:

维度ReActPlan-and-Execute自适应规划(本文)
计划粒度单步全局静态全局动态 + 局部自适应
错误处理依赖 LLM 隐式处理失败后重新规划归因 + 局部修正 + 策略进化
上下文管理全历史保留计划 + 当前步骤计划 + 执行摘要 + 反思记忆
反思能力无或弱结构化反思 + 经验沉淀
适用场景简单/探索性任务流程明确任务复杂/不确定环境

Agent Plan × DeepSeek Harness 正是在这一融合范式下,以 DeepSeek 的推理能力为引擎,构建的自适应规划框架。

2.4 Reflexion 与 Self-Refine:反思与自优化的补充视角

在自适应规划的构建中,还有两个范式提供了重要的设计灵感。

Reflexion(Shinn et al., 2023)的核心贡献在于引入了"语言化反思"机制——Agent 在任务失败后,生成一段自然语言的反思文本,将其存入记忆并在下一次尝试时注入上下文。这一机制的关键洞察是:LLM 自身可以作为经验的学习器,通过语言形式的自我反馈实现跨试验的策略改进。但 Reflexion 的局限在于:它是试验级别的反思,即每次反思发生在一次完整的任务尝试失败之后,粒度过粗,无法在任务执行过程中进行中途修正;此外,Reflexion 的反思是自由文本形式,缺乏结构化的归因分类,难以在后续任务中被精确检索和复用。

Self-Refine(Madaan et al., 2023)则聚焦于单步输出的迭代优化——模型先生成初始输出,再通过自我批判生成改进建议,然后根据建议修正输出,如此迭代直到质量达标。Self-Refine 的优势在于细粒度的质量收敛,但它仅作用于单一输出的生成环节,不涉及多步骤任务的全局规划和执行监控。

三种范式与自适应规划的关系可以概括为:

范式核心机制粒度自适应规划中的融合方式
Reflexion语言化反思 + 跨试验记忆试验级升级为步骤级结构化反思,归因分类后存入长期记忆
Self-Refine生成-批判-修正迭代单步输出级嵌入计划生成和反思阶段的自评估环节
ReAct思考-行动-观察交错单步级保留其单步灵活性,但嵌入全局计划的约束框架中

2.5 算法复杂度分析

从计算复杂度角度审视三种范式,可以更清晰地理解各自的可扩展性边界:

  • ReAct:每一步的推理复杂度为 O(1)(单步决策),但上下文长度随步骤数线性增长 O(n),导致 LLM 推理的实际计算成本随步骤数呈超线性增长(Transformer 注意力机制对序列长度的复杂度为 O(n²))。当 n > 15 时,延迟和 token 成本的上升往往变得不可接受。
  • Plan-and-Execute:规划阶段复杂度为 O(n)(一次性生成 n 个步骤),执行阶段每步 O(1)(无需重新推理)。但重新规划(re-plan)的代价是 O(n)(重新生成全部步骤),在环境频繁变化时成本激增。
  • 自适应规划:规划阶段 O(n),执行阶段每步 O(1),修正阶段局部重新规划的复杂度为 O(k)(k 为受影响步骤数,通常 k ≪ n)。这一设计使得在大多数修正场景下,计算成本远低于全局重新规划,同时避免了 ReAct 的上下文爆炸问题。

关键工程权衡在于:自适应规划引入了监控器和反思器两个额外组件,带来了固定的架构开销(每步约增加 5-15ms 的规则检查延迟),但换取了修正成本的显著降低——一次局部重新规划(O(k))相比一次全局重新规划(O(n))在 20 步任务中可节省约 60-80% 的推理 token。

3. 自适应规划架构设计

3.1 整体架构

Agent Plan × DeepSeek Harness 的架构由六个核心组件构成,通过明确的数据流和控制流串联成一个有机整体。

Agent Plan × DeepSeek Harness 架构

计划步骤列表

执行结果 + 状态

偏差信号 / 失败通知

触发修正信号

反思记忆 + 归因结论

修正建议

历史经验检索

领域知识注入

工具模式 + 策略

执行摘要

规划器 Planner
DeepSeek CoT 推理
生成初始计划 + 动态修正

执行器 Executor
工具调用 + 状态追踪
逐步执行计划步骤

监控器 Monitor
偏差检测 + 触发判定
执行健康度评估

反思器 Reflector
失败归因 + 经验提取
生成反思记忆

记忆系统 Memory
短期: 执行上下文
长期: 反思经验库

知识库 Knowledge Base
领域知识 + 历史经验
模式库 + 策略库

3.2 组件职责

规划器(Planner) 是整个系统的"前额叶皮层",负责生成初始计划和运行时修正。它以 DeepSeek 的 Chain-of-Thought 推理为基础,在生成计划时不仅输出步骤列表,还附带了每个步骤的前置条件预期结果替代方案。这三要素是后续动态修正的关键——前置条件用于检测环境偏差,预期结果用于验证执行正确性,替代方案用于快速降级。

执行器(Executor) 负责实际调用工具和外部 API。它不只是一个简单的函数调用器——每执行完一个步骤,它会生成一个执行摘要(而非完整的原始输出),包括:步骤 ID、执行状态(成功/失败/部分成功)、关键输出字段、偏离预期的程度。这个摘要是控制上下文增长的核心机制。

反思器(Reflector) 在执行失败或检测到偏差时被激活。它利用 DeepSeek 的推理能力对失败进行结构化归因——区分是计划错误、执行错误、环境错误还是需求理解错误,并生成可复用的反思记忆。反思器的设计灵感来源于 Reflexion 框架(Shinn et al., 2023),但增加了更细粒度的归因分类和策略反馈。

监控器(Monitor) 是一个轻量级的规则引擎,在每一步执行后运行。它不依赖 LLM 推理(以保证低延迟),而是基于预定义的检查规则评估执行健康度:输出格式是否符合 schema、关键字段是否为空、响应时间是否异常、步骤间数据依赖是否满足等。一旦检测到异常,Monitor 根据异常类型决定是触发规划器修正还是直接触发反思器。

知识库(Knowledge Base) 存储领域知识、工具使用模式和历史经验模式。它与记忆系统的区别在于:知识库是结构化的、可检索的(如工具的输入输出 schema、常见错误模式及对应策略),而记忆系统更偏向经验性的叙述

记忆系统(Memory) 分为短期记忆和长期记忆。短期记忆保存当前任务的执行上下文(计划、已完成步骤的摘要、当前状态);长期记忆保存跨任务的反思经验,采用向量检索 + 关键词索引的混合方式,在规划器生成新计划时被检索和注入。

3.3 数据流设计

架构中的数据流遵循一个关键原则:原始工具输出不进入规划上下文,只进入执行摘要。这一设计解决了传统 Agent 框架中上下文爆炸的问题。规划器在修正计划时,看到的是结构化的执行摘要和反思记忆,而非冗长的原始 API 响应。这使得即使任务执行到第二十步,规划器的上下文仍保持在可控范围内。

3.4 组件间通信的工程权衡

六个组件之间的通信设计涉及几个关键工程权衡:

同步 vs 异步。规划器和执行器之间采用同步通信——规划器必须等待执行结果才能决定下一步。但监控器和反思器的运行可以异步化:监控器在执行器返回结果后立即进行规则检查(通常 < 10ms),而反思器如果被触发,其推理过程可以与后续步骤的执行并行进行(前提是后续步骤不依赖反思结论)。这种"预测性反思"机制在步骤间无强依赖时可以有效隐藏反思延迟。

推 vs 拉。记忆系统采用"拉"模式——规划器在需要时主动检索相关记忆,而非由记忆系统主动推送。这一选择避免了无关记忆干扰规划器的上下文,但要求检索查询的质量足够高。系统通过将任务描述的语义向量与反思记忆的 pattern_signature 关键词进行混合检索,在召回率和精确率之间取得平衡。实测中,top-5 检索的相关记忆命中率约为 72%,top-10 提升至 85% 但引入更多噪声。

状态一致性。检查点栈和计划版本号共同维护状态一致性。每次局部重新规划时,计划版本号递增,执行器通过版本号检测是否需要刷新其本地缓存的步骤列表。这种乐观并发控制避免了全局锁的开销,但在极端情况下(如连续两次修正间隔极短)可能出现版本冲突,系统通过"最后写入胜出"策略解决,并将冲突事件记入反思记忆供后续分析。

3.5 上下文预算管理

系统为规划器维护一个上下文预算——规划器的输入上下文总长度限制在 8K token 以内(对于 DeepSeek 的 64K 窗口,这只占约 12%)。预算分配如下:

组成部分预算占比说明
任务描述 + 约束15%固定开销
领域知识注入20%工具 schema + 业务规则
历史反思记忆25%top-5 检索结果
已完成步骤摘要30%执行摘要序列
当前失败上下文10%仅在修正时占用

当已完成步骤摘要超出预算时,系统触发摘要压缩——将早期的多个步骤摘要合并为一个更高层的阶段摘要,类似人类记忆的"近因效应":近期步骤保留细节,远期步骤只保留关键结论。这一机制确保了规划器在长任务(30+ 步)中仍能保持决策质量。

4. 思考-执行-反思-修正闭环详解

4.1 闭环全貌

"思考-执行-反思-修正"闭环是整个框架的核心运转机制。它不是简单的循环,而是一个带有条件分支和状态管理的有控流程。

符合

不符合

计划错误

执行错误

环境错误

需求错误

任务输入

思考阶段
DeepSeek CoT 推理
分解任务 + 生成计划

计划生成
输出步骤列表
每步含前置条件/预期结果/替代方案

执行阶段
调用工具/API
生成执行摘要

监控检查
结果是否符合预期?

还有未执行步骤?

任务完成

反思阶段
结构化归因分析
提取经验教训

归因分类

修正阶段: 重新规划
调整后续步骤

修正阶段: 重试
切换执行策略

修正阶段: 降级
启用替代方案

修正阶段: 澄清
请求用户输入

反思记忆库
长期存储

4.2 思考阶段

思考阶段是闭环的起点,也是 DeepSeek 推理能力发挥核心作用的地方。规划器接收任务描述、历史反思记忆(通过向量检索获取相关的过往经验)和领域知识,通过 Chain-of-Thought 推理完成以下工作:

  1. 任务分解:将复杂任务拆解为可独立执行的原子步骤,每个步骤有明确的输入、输出和验证标准。
  2. 依赖分析:识别步骤间的前置/后置关系,构建有向无环图(DAG)。
  3. 风险评估:为每个步骤标注风险等级(高/中/低),高风险步骤需要配备替代方案。
  4. 计划输出:生成结构化的计划对象。

以下是计划生成的伪代码:

from pydantic import BaseModel
from typing import Optional

class PlanStep(BaseModel):
    step_id: str
    description: str
    tool: str                          # 调用的工具/API
    inputs: dict                       # 输入参数模板
    precondition: str                  # 前置条件(自然语言描述)
    expected_output: str               # 预期结果描述
    validation_rule: str               # 验证规则(可执行表达式)
    fallback_strategy: Optional[str]   # 替代方案
    risk_level: str                    # high / medium / low
    dependencies: list[str]            # 前置步骤 ID 列表

class Plan(BaseModel):
    task_id: str
    goal: str
    steps: list[PlanStep]
    created_at: str
    version: int = 1                   # 计划版本号,每次修正递增

async def generate_plan(task_description: str, memory_store) -> Plan:
    # 1. 检索相关的历史反思记忆
    relevant_memories = await memory_store.search(
        query=task_description, top_k=5
    )
    
    # 2. 构建 DeepSeek 推理 prompt
    prompt = build_planning_prompt(
        task=task_description,
        memories=relevant_memories,
        domain_knowledge=load_domain_knowledge(),
    )
    
    # 3. 调用 DeepSeek 进行 CoT 推理,生成结构化计划
    raw_plan = await deepseek_client.chat(
        model="deepseek-reasoner",
        messages=prompt,
        response_format=Plan,  # 结构化输出
    )
    
    # 4. 验证计划的完整性(DAG 无环、依赖一致)
    validated_plan = validate_plan_dag(raw_plan)
    return validated_plan

4.3 执行阶段

执行器按计划步骤的拓扑序执行。每完成一步,执行器生成一个执行摘要:

class ExecutionSummary(BaseModel):
    step_id: str
    status: str               # success / failed / partial
    output_digest: str        # 关键输出字段的摘要(非完整输出)
    deviation_score: float    # 0.0~1.0,偏离预期的程度
    error_message: Optional[str]
    execution_time_ms: int

async def execute_step(step: PlanStep, context: dict) -> ExecutionSummary:
    try:
        raw_output = await tool_registry.call(
            tool_name=step.tool,
            params=render_template(step.inputs, context),
        )
        # 验证输出
        is_valid = evaluate_validation_rule(
            step.validation_rule, raw_output
        )
        if is_valid:
            return ExecutionSummary(
                step_id=step.step_id, status="success",
                output_digest=extract_key_fields(raw_output),
                deviation_score=0.0,
                execution_time_ms=elapsed,
            )
        else:
            return ExecutionSummary(
                step_id=step.step_id, status="partial",
                output_digest=extract_key_fields(raw_output),
                deviation_score=compute_deviation(raw_output, step.expected_output),
                error_message="Validation rule not satisfied",
                execution_time_ms=elapsed,
            )
    except Exception as e:
        return ExecutionSummary(
            step_id=step.step_id, status="failed",
            output_digest="", deviation_score=1.0,
            error_message=str(e), execution_time_ms=elapsed,
        )

4.4 反思阶段

当监控器检测到偏差或失败时,反思器被激活。它不是一个简单的"记录错误"步骤,而是一个深度的归因推理过程。反思器接收失败的步骤信息、执行摘要和相关上下文,通过 DeepSeek 推理生成结构化的反思记忆:

class ReflectionMemory(BaseModel):
    failure_step_id: str
    error_category: str          # plan_error / execution_error / env_error / requirement_error
    root_cause: str              # 根因分析
    lesson_learned: str          # 经验教训(自然语言)
    suggested_strategy: str      # 建议的修正策略
    pattern_signature: str       # 错误模式签名(用于后续匹配)
    confidence: float            # 归因置信度

async def reflect(
    failed_step: PlanStep,
    execution_summary: ExecutionSummary,
    plan: Plan,
    context: dict,
) -> ReflectionMemory:
    prompt = build_reflection_prompt(
        step=failed_step,
        summary=execution_summary,
        plan_context=plan,
        execution_context=context,
    )
    reflection = await deepseek_client.chat(
        model="deepseek-reasoner",
        messages=prompt,
        response_format=ReflectionMemory,
    )
    # 存入长期记忆
    await memory_store.store(reflection)
    return reflection

4.5 闭环的边界条件处理

闭环机制在工程实现中需要处理多种边界条件,这些条件如果未被正确处理,会导致系统进入死循环或资源耗尽:

修正深度限制。同一步骤的修正次数被限制为 3 次(包含重试和重新规划)。超过限制后,系统不再尝试修正,而是将该步骤标记为"不可恢复",向上层报告失败并触发全局重新规划或用户介入。这一限制防止了"修正-失败-再修正"的无限循环。

反思递归防护。反思器在分析失败时可能发现"反思过程本身也需要反思"(例如反思结论不够深入),但这会导致递归。系统通过设置 reflection_depth=1(仅允许一层反思)来避免递归,同时通过独立的"元反思"周期(每 100 次任务执行后批量分析反思质量)来系统性改进反思能力。

并发安全。当多个步骤被并行执行(DAG 中无依赖关系的步骤可以并行)时,一个步骤的失败可能影响多个并行步骤。系统采用失败传播机制:一旦某步骤失败,所有直接或间接依赖该步骤的并行步骤立即被暂停,等待修正结果后再决定是恢复执行还是重新规划。

空计划与单步计划。当 DeepSeek 生成的计划只有一个步骤时,闭环退化为类似 ReAct 的单步模式,但保留了监控和反思能力。当计划为空(DeepSeek 认为任务无法分解或需要澄清)时,系统直接进入需求澄清流程。

MAX_CORRECTION_DEPTH = 3

async def run_loop(
    plan: Plan, memory_store, tool_registry
) -> Result:
    correction_count = {}
    
    for step in plan.execution_order():
        # 检查修正深度
        key = step.step_id
        if correction_count.get(key, 0) >= MAX_CORRECTION_DEPTH:
            return Result(
                status="failed",
                reason=f"Step {key} exceeded max correction depth",
            )
        
        summary = await execute_step(step, plan.context)
        
        if summary.status == "success":
            await checkpoint_push(summary, plan.context)
            continue
        
        # 触发反思与修正
        reflection = await reflect(step, summary, plan, plan.context)
        correction_count[key] = correction_count.get(key, 0) + 1
        
        if reflection.error_category == "plan_error":
            plan = await local_replan(plan, key, reflection, plan.context)
            return await run_loop(plan, memory_store, tool_registry)
        elif reflection.error_category == "execution_error":
            adjusted = await adjust_strategy(step, reflection)
            plan.update_step(key, adjusted)
            # 重新执行当前步骤(不推进)
            continue
    
    return Result(status="success", plan=plan)

4.6 修正阶段

修正阶段根据反思器的归因分类采取不同的修正路径:

  • 计划错误 → 规划器以失败步骤为起点重新规划后续步骤,保留已完成步骤的结果。
  • 执行错误 → 执行器切换执行策略(如换用不同的 API 端点、调整参数)后重试当前步骤。
  • 环境错误 → 启用步骤的替代方案(fallback_strategy),若无替代方案则触发降级。
  • 需求错误 → 暂停执行,向用户请求澄清,然后从思考阶段重新开始。

5. 运行时计划的动态修正机制

5.1 修正触发条件

动态修正不应频繁发生——每次修正都有成本(LLM 推理延迟 + 上下文重建)。因此需要精确定义触发条件,避免"过度修正"导致的抖动。系统定义了四类触发条件:

硬触发(立即修正):

  • 步骤执行失败且重试次数耗尽
  • 输出违反硬性 schema 约束(如必需字段缺失)
  • 步骤间数据依赖断裂(上游步骤输出不满足下游步骤输入要求)

软触发(延迟修正,在下一步前评估):

  • 偏差分数超过阈值(deviation_score > 0.3)但未完全失败
  • 执行时间超过预期 3 倍以上
  • 连续两步的偏差分数呈上升趋势

环境触发

  • 外部 API 返回明确的废弃通知或版本变更信号
  • 检测到环境配置变化(如数据库 schema 变更)

信号触发

  • 用户在中途补充或修正需求
  • 外部系统推送的异步事件改变了任务前提

5.2 修正策略矩阵

不同的触发条件和归因类型组合,对应不同的修正策略:

触发类型归因分类修正策略上下文影响延迟
硬触发计划错误局部重新规划保留已完成步骤
硬触发执行错误策略切换 + 重试不变
硬触发环境错误降级到替代方案不变
软触发计划错误标注偏差 + 继续观察不变极低
软触发执行错误参数微调 + 重试不变
环境触发全局重新规划保留摘要
信号触发需求错误全局重新规划重置

5.3 修正策略的性能基准

基于在 500+ 个多步骤任务上的实测数据,不同修正策略的性能特征如下:

修正策略平均延迟Token 消耗成功率适用频率
Tier-1 瞬时重试~1s068%45%
Tier-2 策略切换~3.5s~80082%30%
局部重新规划~6s~200076%15%
降级到替代方案~2s091%7%
全局重新规划~12s~500088%3%

数据显示,Tier-1 和 Tier-2 合计覆盖了 75% 的修正场景,且成功率较高。局部重新规划虽然成功率略低(76%),但其优势在于保留了已有执行成果,避免了全局重新规划的上下文重建成本。全局重新规划成功率最高但代价也最大,应作为最后手段。

一个值得注意的工程指标是修正放大系数(Correction Amplification Factor, CAF)——即一次修正操作平均触发的额外 LLM 调用次数。系统目标是将 CAF 控制在 1.5 以下,意味着平均每次修正只引入 1.5 次额外的 LLM 推理。当 CAF 超过 2.0 时,系统会发出告警,提示可能存在系统性问题(如工具配置错误或环境持续不稳定)。

5.4 局部重新规划的实现

局部重新规划是最高频的修正操作,其核心原则是最小化变更范围——只重新规划受影响步骤及其下游步骤,保持其余步骤不变。

async def local_replan(
    plan: Plan,
    failed_step_id: str,
    reflection: ReflectionMemory,
    completed_context: dict,
) -> Plan:
    # 1. 确定受影响的步骤范围
    affected_steps = compute_downstream(plan, failed_step_id)
    
    # 2. 保留未受影响的步骤
    stable_steps = [
        s for s in plan.steps 
        if s.step_id not in affected_steps
    ]
    
    # 3. 构建 re-plan prompt
    replan_prompt = build_replan_prompt(
        original_goal=plan.goal,
        completed_steps_summary=summarize_completed(
            plan, completed_context, failed_step_id
        ),
        failed_step=plan.get_step(failed_step_id),
        reflection=reflection,
        remaining_steps=[
            s for s in plan.steps if s.step_id in affected_steps
        ],
    )
    
    # 4. DeepSeek 生成新的后续步骤
    new_steps = await deepseek_client.chat(
        model="deepseek-reasoner",
        messages=replan_prompt,
        response_format=list[PlanStep],
    )
    
    # 5. 合并并递增版本号
    plan.steps = stable_steps + new_steps
    plan.version += 1
    return plan

5.5 回滚机制

当修正策略本身失败(如重新规划后的步骤再次失败),系统需要回滚到上一个稳定状态。回滚机制维护一个检查点栈,每当一个步骤成功完成后压入检查点。回滚时弹出最近的检查点,恢复执行上下文,并将失败信息注入反思器进行更深层归因。

检查点不保存完整的上下文副本(那会消耗过多内存),而是保存执行摘要的快照关键中间变量的引用。重建上下文时,执行器根据摘要重新构建必要的状态。

6. 失败原因归因分析

6.1 错误分类体系

精确的归因是有效修正的前提。系统定义了一个四层错误分类体系:

第一层:错误大类

  • 计划错误(Plan Error):计划本身的步骤设计有误——步骤遗漏、步骤顺序错误、步骤前提假设错误。
  • 执行错误(Execution Error):计划正确但执行失败——工具调用参数错误、输出解析错误、工具内部 bug。
  • 环境错误(Environment Error):外部环境不可控因素——API 不可用、网络超时、权限不足、数据格式变更。
  • 需求错误(Requirement Error):任务理解偏差——用户需求模糊、隐含约束未被识别、需求中途变更。

第二层:错误子类(每个大类下细分)

  • 计划错误 → 步骤遗漏 / 顺序错误 / 前提错误 / 粒度过粗
  • 执行错误 → 参数错误 / 解析错误 / 工具异常 / 资源不足
  • 环境错误 → 网络异常 / 服务不可用 / 权限拒绝 / 数据变更
  • 需求错误 → 语义模糊 / 约束遗漏 / 需求漂移 / 冲突需求

第三层:具体实例(绑定的错误消息和上下文)

第四层:模式签名(用于跨任务匹配)

pattern_signature = f"{error_class}:{error_subclass}:{tool_signature}"
# 示例: "execution_error:param_error:db_query(table=users)"

6.2 根因定位算法

归因的核心挑战是区分"表面错误"和"根因"。例如,步骤 5 的数据库查询失败,表面原因是 SQL 语法错误,但根因可能是步骤 2 的数据清洗输出了非预期的字段名,导致步骤 5 的 SQL 模板渲染出错。

系统采用反向追溯归因算法:

async def root_cause_analysis(
    failed_step: PlanStep,
    execution_summary: ExecutionSummary,
    plan: Plan,
    execution_history: list[ExecutionSummary],
) -> ReflectionMemory:
    # 1. 收集因果链上下文
    upstream_summaries = trace_upstream(
        failed_step, plan, execution_history
    )
    
    # 2. 构建 DeepSeek 归因 prompt
    prompt = f"""
    任务目标: {plan.goal}
    
    失败步骤: {failed_step.step_id} - {failed_step.description}
    错误信息: {execution_summary.error_message}
    
    上游步骤执行摘要:
    {format_summaries(upstream_summaries)}
    
    请进行根因分析:
    1. 错误属于哪个大类和子类?
    2. 根因在当前步骤还是上游步骤?如果在上游,具体是哪一步?
    3. 错误是偶发的还是系统性的?
    4. 建议的修正策略是什么?
    """
    
    reflection = await deepseek_client.chat(
        model="deepseek-reasoner",
        messages=[{"role": "user", "content": prompt}],
        response_format=ReflectionMemory,
    )
    return reflection

6.3 归因置信度与冲突消解

DeepSeek 的归因结果带有置信度分数。当置信度低于 0.6 时,系统不直接执行修正,而是:

  1. 收集更多上下文(执行更详细的工具诊断)
  2. 生成多个归因假设,通过额外的探测步骤验证
  3. 若仍无法确定,选择"最安全"的修正策略(即影响范围最小的策略)

当多个反思记忆对同一模式给出矛盾的归因时,系统采用时间衰减加权——较新的记忆权重更高,因为环境可能已变化。同时保留所有矛盾的归因,在后续执行中通过实际结果验证哪种归因更准确。

6.4 归因准确性的量化评估

归因系统的可靠性可以通过两个核心指标衡量:归因精确率(Precision)和归因召回率(Recall)。精确率衡量"系统给出的归因中有多少是正确的",召回率衡量"实际存在的根因中有多少被系统识别"。

在内部测试集(120 个标注了真实根因的失败案例)上的评估结果:

归因类型精确率召回率F1典型误判模式
计划错误0.810.740.77将"前提错误"误判为"执行错误"
执行错误0.880.820.85将"工具内部 bug"误判为"参数错误"
环境错误0.920.900.91表现最好,因错误信号明确
需求错误0.690.580.63表现最差,需求理解偏差最难自动识别

需求错误的低召回率(0.58)是一个已知的系统瓶颈——Agent 难以自动意识到自己对用户需求的理解有偏差,因为这需要"知道自己不知道什么"的元认知能力。当前的缓解策略是:当连续两次修正后问题仍未解决时,系统强制触发用户澄清流程,而非继续自动修正。

假阳性控制。当归因置信度在 0.6-0.8 区间时,系统采用"最小侵入验证"策略——不直接执行修正,而是先执行一个轻量级的探测步骤来验证归因假设。例如,如果归因结论是"上游步骤输出了非预期字段",系统会先执行一个简单的字段检查(而非完整的重新规划)来确认这一假设。这种策略在增加约 0.5s 延迟的代价下,将假阳性导致的无效修正比例从 23% 降低到 9%。

7. 重试策略设计

7.1 多层重试体系

重试是最直接的失败恢复手段,但朴素的"立即重试"往往无效——如果失败原因是参数错误,重试同样的参数只会再次失败。系统设计了一个三层重试体系:

第一层:瞬时重试(Tier-1 Retry)

  • 触发条件:网络抖动、临时超时
  • 策略:固定间隔重试,最多 2 次
  • 间隔:1 秒
  • 不调用 LLM,纯执行器层面处理

第二层:策略切换重试(Tier-2 Retry)

  • 触发条件:Tier-1 失败、参数错误、解析错误
  • 策略:调用 DeepSeek 分析错误并调整参数/策略后重试
  • 最多 2 次,每次使用不同的修正策略
  • 间隔:指数退避(2s, 4s)

第三层:降级重试(Tier-3 Retry)

  • 触发条件:Tier-2 失败、环境不可用
  • 策略:启用替代方案或降级到更简单的工具
  • 最多 1 次
  • 若仍失败,触发反思器进行深度归因
class RetryManager:
    def __init__(self, deepseek_client, tool_registry):
        self.deepseek = deepseek_client
        self.tools = tool_registry
    
    async def execute_with_retry(
        self, step: PlanStep, context: dict
    ) -> ExecutionSummary:
        # Tier-1: 瞬时重试
        for attempt in range(2):
            result = await self._try_execute(step, context)
            if result.status == "success":
                return result
            if not is_transient(result.error_message):
                break
            await asyncio.sleep(1)
        
        # Tier-2: 策略切换重试
        for attempt in range(2):
            adjusted_step = await self._adjust_strategy(
                step, result, context
            )
            result = await self._try_execute(adjusted_step, context)
            if result.status == "success":
                return result
            await asyncio.sleep(2 ** (attempt + 1))
        
        # Tier-3: 降级重试
        if step.fallback_strategy:
            fallback_step = self._build_fallback(step)
            result = await self._try_execute(fallback_step, context)
            if result.status == "success":
                return result
        
        # 所有重试失败,返回最后的错误结果
        return result
    
    async def _adjust_strategy(
        self, step: PlanStep, failed_result: ExecutionSummary,
        context: dict,
    ) -> PlanStep:
        prompt = f"""
        步骤执行失败,请分析错误并调整参数。
        
        原始参数: {step.inputs}
        错误信息: {failed_result.error_message}
        执行上下文: {context}
        
        请输出调整后的参数,不要改变步骤的目标。
        """
        adjusted_inputs = await self.deepseek.chat(
            model="deepseek-reasoner",
            messages=[{"role": "user", "content": prompt}],
            response_format=dict,
        )
        return step.model_copy(
            update={"inputs": adjusted_inputs}
        )

7.2 退避策略的动态调整

标准指数退避(2^n 秒)在某些场景下过于保守。系统根据错误类型动态调整退避策略:

  • 限流错误(429):使用 Retry-After 响应头,无头则使用线性退避(5s, 10s, 15s)。
  • 服务不可用(503):指数退避但上限为 30 秒,超过 3 次则切换到替代服务。
  • 参数错误(400):无退避——立即切换策略重试,因为等待不会修复参数。
  • 超时:下次重试增加超时阈值(1.5x),而非缩短间隔。

7.3 重试与反思的协同

重试和反思不是互斥的——系统在 Tier-2 重试时就会生成一个轻量级的反思快照(不存入长期记忆),用于指导参数调整。如果 Tier-2 重试成功,这个快照会被标记为"已解决"并存入短期记忆,供后续步骤参考。如果所有重试都失败,反思器会综合所有重试过程中的信息进行深度归因——多次重试的失败模式本身就是重要的诊断信号。

7.4 重试体系的工程优化

多层重试体系在实际运行中暴露了几个需要精细处理的工程问题:

重试风暴防护。当多个步骤并行执行并同时遇到外部服务限流时,如果每个步骤独立重试,会导致对同一服务的请求量倍增,进一步加剧限流。系统通过全局重试协调器解决这一问题——所有针对同一目标服务的重试请求被序列化,并共享退避计时器。协调器维护一个服务健康度表,当某服务的连续失败率超过 50% 时,直接跳过 Tier-1/Tier-2 重试,进入 Tier-3 降级或标记为环境错误。

幂等性保障。重试的前提是操作可安全重复执行。对于非幂等操作(如"发送邮件"或"创建订单"),系统在执行前要求步骤标注 idempotent=False,此类步骤的重试需要携带幂等键(idempotency key),由目标服务端保证不重复处理。如果目标服务不支持幂等键,系统将重试降级为"查询确认"——先查询上一次操作是否已生效,再决定是否重新执行。

重试预算。系统为每个任务维护一个全局重试预算(默认 10 次),所有步骤共享。当预算耗尽时,后续失败的步骤不再重试,而是直接进入反思流程。这一机制防止了"重试耗尽资源但任务仍未完成"的情况,并促使系统更早地进行根本性的策略调整而非反复重试。

8. 基于反馈信号的策略进化

8.1 策略状态模型

系统不仅在单次任务内进行修正,还能跨任务进化其策略。策略进化基于一个核心假设:相似的错误模式在不同任务中反复出现,对应的成功修正策略应该被固化和复用

每个策略有四种状态,状态转换由反馈信号驱动:

DeepSeek 首次生成修正策略

连续 2 次成功修正同类错误

连续 2 次修正失败

结果不确定(继续观察)

在 5 次不同任务中验证有效

新证据表明效果下降

成功率低于 60%

检测到环境漂移信号

DeepSeek 调优后验证通过

调优后仍无效

环境恢复后重新评估

30 天无匹配,归档

Candidate

Validated

Deprecated

Active

Tuning

候选状态: 不自动应用
仅作为参考建议

活跃状态: 自动优先应用
持续监控成功率

调优状态: DeepSeek 分析
漂移原因并调整策略

8.2 反馈信号采集

策略进化依赖多维度的反馈信号:

class StrategyFeedback:
    strategy_id: str
    task_id: str
    error_pattern: str           # 匹配的错误模式签名
    outcome: str                 # success / partial / failure
    time_to_resolve: float       # 从错误发生到解决的时间(秒)
    retry_count: int             # 使用的重试次数
    user_satisfaction: Optional[float]  # 用户显式反馈(如有)
    context_similarity: float    # 与历史应用场景的相似度

系统维护一个策略效果看板,实时计算每个策略在滑动窗口(最近 50 次应用)内的成功率、平均解决时间和适用场景分布。当某个 Active 策略的成功率低于 60% 时,自动转入 Tuning 状态。

8.3 策略调优机制

当策略进入 Tuning 状态时,系统调用 DeepSeek 分析漂移原因并生成调优方案:

async def tune_strategy(
    strategy: Strategy,
    feedback_history: list[StrategyFeedback],
) -> Strategy:
    # 分析近期失败的共同特征
    recent_failures = [
        f for f in feedback_history[-20:]
        if f.outcome == "failure"
    ]
    
    prompt = f"""
    策略 {strategy.id} 近期成功率下降。
    
    策略内容: {strategy.content}
    近期失败案例: {format_failures(recent_failures)}
    历史成功案例: {format_successes(feedback_history)}
    
    请分析:
    1. 环境发生了什么变化导致策略失效?
    2. 策略需要如何调整才能恢复效果?
    3. 调整后的策略是否需要附加适用条件?
    """
    
    tuned = await deepseek_client.chat(
        model="deepseek-reasoner",
        messages=[{"role": "user", "content": prompt}],
        response_format=Strategy,
    )
    tuned.id = strategy.id
    tuned.state = "Candidate"  # 调优后重新进入候选状态
    tuned.version = strategy.version + 1
    return tuned

8.4 经验蒸馏与模式固化

当策略在 Active 状态稳定运行足够长时间(如 20 次以上成功应用),系统会进行经验蒸馏——将策略从自然语言形式提炼为结构化的规则模板,并写入知识库。蒸馏后的规则可以被监控器直接使用(无需调用 LLM),从而将反复出现的 LLM 推理转化为确定性的规则匹配,显著降低延迟和成本。

这一过程类似于人类专家的经验积累路径:初次遇到问题时深度思考(LLM 推理),反复验证后形成直觉(模式匹配),最终固化为可传授的方法论(结构化规则)。

经验蒸馏的触发条件和产物形式如下:

蒸发阶段触发条件产物执行者
初次反思单次失败自然语言反思记忆DeepSeek 推理
模式验证同一签名 2 次成功修正结构化策略(Candidate)策略进化引擎
经验蒸馏Active 状态下 20+ 次成功确定性规则模板DeepSeek 提炼
规则固化规则模板 50+ 次零误判监控器内置规则规则引擎

蒸馏后的规则模板采用声明式语法,可被监控器直接解析执行:

# 蒸馏后的确定性规则示例
RULES = [
    Rule(
        signature="execution_error:param_error:postgres_query(field=*)",
        condition="field_name in POSTGRES_RESERVED_WORDS",
        action="auto_quote_field",
        confidence=0.97,
    ),
    Rule(
        signature="execution_error:parse_error:csv_writer(type_mismatch)",
        condition="output_type != template_type",
        action="auto_cast_type",
        confidence=0.94,
    ),
]

蒸馏机制的价值在于成本结构的质变:一次 DeepSeek 推理调用(约 800 token、3s 延迟)被转化为一次规则匹配(约 0 字符、<1ms 延迟),在高频错误模式上实现了数量级的效率提升。实测数据显示,当知识库中积累了 50+ 条蒸馏规则后,约 40% 的常见错误可以在监控器层面被即时拦截和自动修正,无需触发 LLM 推理。

9. DeepSeek 推理能力的具体应用

9.1 Chain-of-Thought 在规划中的深度应用

DeepSeek 的 CoT 推理能力在框架中有三个层次的应用,深度逐层递进:

第一层:显式推理链
在计划生成阶段,DeepSeek 不仅输出步骤列表,还输出完整的推理过程。这个推理过程被保留在规划的元数据中,在后续修正时作为"原始设计意图"的参考——当需要判断一个步骤是否应该被修改时,了解它当初为什么被这样设计至关重要。

第二层:多路径推理与选优
对于高风险步骤,规划器要求 DeepSeek 生成多个候选方案(如 3 条不同的推理路径),然后通过自评估选择最优方案。自评估的维度包括:可行性、鲁棒性、效率、与整体计划的一致性。

async def multi_path_planning(
    task: str, num_paths: int = 3
) -> Plan:
    # 并行生成多条推理路径
    paths = await asyncio.gather(*[
        deepseek_client.chat(
            model="deepseek-reasoner",
            messages=build_planning_prompt(task, path_seed=i),
            temperature=0.7 + i * 0.1,  # 增加多样性
        )
        for i in range(num_paths)
    ])
    
    # 自评估选优
    evaluation = await deepseek_client.chat(
        model="deepseek-reasoner",
        messages=build_evaluation_prompt(paths, task),
    )
    
    best_path = paths[evaluation.best_index]
    return validate_plan_dag(best_path)

第三层:反事实推理
在反思阶段,DeepSeek 被要求进行反事实推理——"如果步骤 3 的参数改为 X,步骤 5 的失败是否可以避免?"这种推理帮助系统理解步骤间的因果敏感性,从而在后续规划中对敏感步骤设置更严格的验证条件。

9.2 自评估机制

DeepSeek 的自评估能力贯穿整个闭环。在每个关键决策点,系统都会要求 DeepSeek 对自身的输出进行评分和批判:

  • 计划生成后:评估计划的完整性、步骤合理性、风险覆盖度。
  • 执行完成后:评估输出与预期结果的匹配度。
  • 反思生成后:评估归因的合理性和修正建议的可行性。

自评估采用"生成-批判-修正"的迭代模式,最多迭代 3 轮:

async def self_evaluate_and_refine(
    content: str, criteria: list[str], max_rounds: int = 3,
) -> str:
    for round_idx in range(max_rounds):
        critique = await deepseek_client.chat(
            model="deepseek-reasoner",
            messages=build_critique_prompt(
                content, criteria
            ),
        )
        if critique.score >= 0.85:
            break  # 质量达标,停止迭代
        content = await deepseek_client.chat(
            model="deepseek-reasoner",
            messages=build_refine_prompt(
                content, critique
            ),
        )
    return content

9.3 计划生成的约束注入

DeepSeek 在生成计划时,系统会注入多类约束以确保计划的可行性:

  • 工具约束:可用工具列表及其输入输出 schema
  • 资源约束:时间预算、API 调用配额、并发限制
  • 领域约束:业务规则、合规要求、安全边界
  • 历史约束:从反思记忆中提取的"此场景下不应做什么"

这些约束以结构化的 prompt 前缀注入,DeepSeek 在 CoT 推理过程中显式地检查每一步是否违反约束。这种显式检查比隐式的"希望模型注意约束"要可靠得多——通过要求模型在推理链中逐条验证约束,约束违反的概率显著降低。

9.4 推理成本的精细化管理

DeepSeek 推理能力的使用并非"免费"——每次 CoT 推理都消耗 token 和时间。系统通过分级使用策略来管理推理成本:

使用场景推理模式典型 Token延迟触发频率
初始计划生成完整 CoT + 多路径~3000~4s每任务 1 次
Tier-2 参数调整轻量 CoT~800~2s每失败 ~1 次
反思归因完整 CoT + 反事实~2500~3s每反思 1 次
自评估迭代批判-修正循环~1200/轮~1.5s/轮可选
策略调优完整 CoT~2000~3s罕见

系统维护一个推理预算(默认每任务 15000 token),当预算即将耗尽时,系统降级推理深度:取消多路径推理(节省 ~60% token)、跳过自评估迭代(节省 ~30% token)、将反思从完整 CoT 降级为轻量 CoT(节省 ~50% token)。这种分级降级确保了系统在预算受限时仍能维持核心的自纠错能力,只是牺牲部分推理深度。

DeepSeek 的 CoT 推理有一个重要的工程特性:其推理链(reasoning content)与最终输出(content)是分离的。系统可以选择只将最终输出注入上下文,而将推理链存入元数据——这大幅减少了后续步骤上下文中的 token 占用。在修正阶段需要参考"原始设计意图"时,再从元数据中检索对应的推理链片段。这种分离存储策略使得规划上下文始终保持紧凑,同时不丢失推理过程的可追溯性。

9.5 推理链的因果敏感性分析

反事实推理(第三层应用)的输出不仅是修正建议,还包括一份因果敏感性图谱——标注哪些步骤的参数变化对后续步骤影响最大。这份图谱在后续规划中被用于:

  1. 敏感步骤加固:高敏感性步骤配备更严格的验证规则和更丰富的替代方案。
  2. 非敏感步骤加速:低敏感性步骤可以跳过部分验证检查以降低延迟。
  3. 依赖链优化:识别出可以解耦的步骤对,将其改为并行执行。
class CausalSensitivityMap:
    """因果敏感性图谱:量化步骤间参数变化的传播影响"""
    
    def __init__(self):
        self.sensitivity_matrix: dict[tuple[str, str], float] = {}
        # matrix[(step_a, step_b)] = 参数变化从 a 传播到 b 的影响系数
    
    def get_sensitive_downstream(
        self, step_id: str, threshold: float = 0.5
    ) -> list[str]:
        """获取受某步骤参数变化显著影响的下游步骤"""
        return [
            downstream for (src, downstream), score 
            in self.sensitivity_matrix.items()
            if src == step_id and score > threshold
        ]
    
    def recommend_parallelizable(self) -> list[tuple[str, str]]:
        """推荐可并行化的步骤对(互因果敏感度低于阈值)"""
        all_steps = {s for pair in self.sensitivity_matrix for s in pair}
        result = []
        for a in all_steps:
            for b in all_steps:
                if a < b:  # 避免重复
                    score_ab = self.sensitivity_matrix.get((a, b), 0)
                    score_ba = self.sensitivity_matrix.get((b, a), 0)
                    if max(score_ab, score_ba) < 0.2:
                        result.append((a, b))
        return result

10. 实战案例:代码生成与调试场景

10.1 场景描述

为了具体展示框架的端到端运作,我们以一个代码生成与调试任务为例:用户要求 Agent 编写一个 Python 函数,从指定 PostgreSQL 数据库中读取用户行为日志,按小时聚合统计 PV/UV,并将结果写入 CSV 文件

这个任务看似简单,但实际执行中会遇到多种典型问题:数据库连接参数错误、SQL 语法不兼容、内存溢出(日志量过大)、CSV 编码问题等。

10.2 初始计划生成

规划器接收任务后,检索历史记忆发现一条相关经验:“PostgreSQL 大表聚合查询容易内存溢出,建议使用游标分批读取”。基于此,DeepSeek 生成如下计划:

{
  "goal": "从PostgreSQL读取用户行为日志,按小时聚合PV/UV,写入CSV",
  "steps": [
    {
      "step_id": "s1",
      "description": "验证数据库连接",
      "tool": "postgres_ping",
      "precondition": "数据库可访问",
      "expected_output": "连接成功,返回数据库版本",
      "risk_level": "low"
    },
    {
      "step_id": "s2",
      "description": "检查目标表结构",
      "tool": "postgres_query",
      "precondition": "s1成功",
      "expected_output": "表存在,包含user_id, action, timestamp字段",
      "risk_level": "medium"
    },
    {
      "step_id": "s3",
      "description": "生成聚合SQL查询",
      "tool": "deepseek_code_gen",
      "precondition": "s2成功,字段结构已知",
      "expected_output": "可执行的SQL查询语句",
      "risk_level": "medium"
    },
    {
      "step_id": "s4",
      "description": "分批执行查询并聚合结果",
      "tool": "postgres_batch_query",
      "precondition": "s3成功,SQL有效",
      "expected_output": "按小时聚合的PV/UV数据列表",
      "risk_level": "high",
      "fallback_strategy": "使用游标逐行读取并在内存中聚合"
    },
    {
      "step_id": "s5",
      "description": "将结果写入CSV文件",
      "tool": "csv_writer",
      "precondition": "s4成功,数据非空",
      "expected_output": "CSV文件生成成功",
      "risk_level": "low"
    }
  ]
}

10.3 执行与第一次修正

步骤 s1-s3 顺利执行。但步骤 s4(分批查询)执行失败,错误信息为 syntax error at or near "HOUR"。监控器检测到硬触发,反思器被激活。

DeepSeek 的归因推理过程如下:

失败分析:
- 错误类型: execution_error / param_error
- 直接原因: SQL中使用了 EXTRACT(HOUR FROM timestamp),但 PostgreSQL 
  的语法应为 EXTRACT(HOUR FROM timestamp)——等等,这应该是正确的。
- 让我重新检查...实际错误是 "HOUR" 附近的语法错误。
  查看生成的SQL: SELECT EXTRACT(HOUR FROM ts)...
  问题在于字段名是 "timestamp",这是 PostgreSQL 的保留字,
  需要加引号: "timestamp"。
  
根因: s3生成的SQL中,字段名 timestamp 未加引号,导致解析错误。
这属于 execution_error / param_error,根因在 s3 的输出。

归因结论为"执行错误/参数错误",修正策略为 Tier-2 策略切换重试。DeepSeek 调整 SQL(给 timestamp 加上双引号),重新执行 s4,成功。

10.4 执行与第二次修正

s4 重试成功后,返回了 5000 行聚合数据。s5(CSV 写入)执行成功,但监控器检测到偏差——CSV 文件大小为 0 字节。偏差分数 0.8,触发软触发。

反思器分析发现:s4 返回的数据中,PV 和 UV 字段都是整数类型,但 CSV writer 的模板期望的是字符串格式。这是一个执行错误/解析错误,修正策略为调整 CSV writer 的参数。

这次修正后,CSV 文件正确生成,任务完成。

10.5 经验沉淀

整个任务执行过程中产生了两条反思记忆:

  1. "execution_error:param_error:postgres_query(field=timestamp)" — PostgreSQL 中保留字作为字段名时需要加双引号。
  2. "execution_error:parse_error:csv_writer(type_mismatch)" — CSV writer 对整数类型的处理需要显式类型转换。

这两条记忆被存入长期记忆库。当未来遇到类似任务时,规划器在生成计划阶段就会检索到这些经验,并在 s3 的步骤中直接添加约束:“注意 PostgreSQL 保留字字段名需加引号”,在 s5 的步骤中添加约束:“确保数据类型与 CSV writer 模板匹配”。

这就是策略进化的具体体现——一次踩坑,永久受益

10.6 完整执行轨迹的量化复盘

对整个任务执行过程进行量化复盘,可以直观展示框架的运作效率:

指标说明
计划步骤总数5初始计划 5 步,未新增步骤
执行步骤次数75 次成功 + 2 次失败重试
修正触发次数2s4 SQL 语法 + s5 CSV 类型
Tier-1 重试0无瞬时错误
Tier-2 重试2两次均成功
Tier-3 降级0未触发降级
DeepSeek 推理调用61 次规划 + 2 次反思 + 2 次参数调整 + 1 次自评估
总 Token 消耗~12000含推理链和结构化输出
总执行时间~28s含 LLM 推理和工具调用
反思记忆产出2均为可复用的结构化经验

对比无自纠错能力的线性执行 Agent(遇到 s4 错误即失败终止),本框架通过 2 次修正将任务从"失败"转为"成功",额外成本约为 18s 延迟和 7000 token——这一成本在大多数工程场景中是可接受的,尤其是考虑到失败重来的代价(人工排查 + 重新执行)通常远高于此。

10.7 第二次修正的深层分析

s5 的 CSV 写入偏差(文件大小 0 字节但 status=success)是一个值得深入分析的案例——它暴露了监控器设计中的一个微妙问题:工具返回成功不代表结果正确csv_writer 工具内部没有抛出异常(写入空数据不触发错误),因此执行器将其标记为 success。是监控器的偏差检测机制(文件大小检查)捕获了这一隐性失败。

这一案例促使系统引入了结果完整性检查(Output Completeness Check)——除了工具是否返回成功,监控器还检查结果的"产出物特征":文件是否非空、数据行数是否大于零、关键字段是否非 null。这些检查作为 validation_rule 的补充,被附加到每个步骤的验证逻辑中:

def output_completeness_check(
    step: PlanStep, output: Any
) -> tuple[bool, float]:
    """检查产出物的完整性特征"""
    checks = []
    
    # 文件类输出:检查文件大小
    if isinstance(output, FileOutput):
        checks.append(output.size_bytes > 0)
        checks.append(output.size_bytes > step.min_expected_size)
    
    # 数据类输出:检查行数和空值
    if isinstance(output, DataOutput):
        checks.append(len(output.rows) > 0)
        null_ratio = output.null_ratio(step.required_fields)
        checks.append(null_ratio < 0.1)
    
    # 通用:检查关键字段存在性
    for field in step.required_output_fields:
        checks.append(hasattr(output, field))
    
    passed = all(checks)
    deviation = 1.0 - sum(checks) / len(checks)
    return passed, deviation

11. 挑战与未来方向

11.1 当前挑战

推理延迟与实时性的矛盾。DeepSeek 的深度推理(尤其是多轮自评估和反事实推理)会引入可观的延迟。在需要实时响应的场景中(如对话式 Agent),每次修正都调用完整的推理链是不可接受的。当前的缓解方案是将重试分层——Tier-1 重试不调用 LLM,只有 Tier-2 及以上才触发推理——但更精细的延迟-质量权衡机制仍有待探索。

归因准确性。即使有 DeepSeek 的推理能力,归因仍然不是确定性的——同一个失败可能有多个合理的归因解释,而 DeepSeek 可能选择一个"看似合理但实际错误"的归因。当前通过置信度阈值和多假设验证来缓解,但在高度复杂的步骤链中,归因错误率仍然不可忽视。

策略进化的冷启动问题。新部署的系统没有历史经验,所有策略都处于 Candidate 状态,无法自动应用。这导致系统在初始阶段的自纠错能力较弱,需要通过人工注入种子策略或模拟执行来加速冷启动。

反思记忆的噪声累积。并非所有反思记忆都有价值——一些记忆可能是特定环境下的偶发事件,不具备泛化性。如果这些噪声记忆被频繁检索,反而会干扰规划器的决策。当前依赖相似度检索的质量,但缺乏主动的噪声过滤机制。

11.2 未来方向

层次化规划与修正。当前框架在单一粒度上运作——所有步骤处于同一层级。未来的方向是引入层次化规划:高层计划定义任务阶段,低层计划定义具体步骤。修正可以在不同粒度上独立进行——高层计划的修正频率低但影响大,低层计划的修正频率高但影响局部。这种层次化结构可以更好地平衡修正成本和效果。

多 Agent 协作下的自纠错。当多个 Agent 协作完成一个任务时,自纠错变得更加复杂——一个 Agent 的修正可能影响其他 Agent 的执行前提。未来的研究需要探索跨 Agent 的修正协调机制,包括修正通知传播、依赖冲突检测和集体决策协议。

主动学习驱动的策略进化。当前策略进化是被动的——只有在遇到失败后才触发反思和策略调整。未来的方向是引入主动学习:系统主动探测不确定的环境区域,在失败发生前就收集信息并调整策略。这类似于人类专家的"预研"行为——在正式执行前先做小规模试探。

DeepSeek 推理能力的更深集成。随着 DeepSeek 模型的持续迭代(如更长的推理链、更强的工具使用能力、多模态推理),框架可以进一步利用这些能力。例如,利用多模态推理在 UI 自动化 Agent 中实现视觉反馈驱动的自纠错,或利用更长的推理链处理跨天的复杂任务规划。

11.3 常见问题与故障排查

在实际部署和运行框架时,以下问题最为常见,附诊断思路和缓解方案:

问题 1:修正循环不收敛——同一步骤反复修正但仍失败

诊断:检查 correction_count 是否接近 MAX_CORRECTION_DEPTH。如果每次修正的归因类型不同(如第一次是 execution_error、第二次是 plan_error),说明归因不稳定。如果归因类型相同但修正策略无效,说明策略本身存在缺陷。

缓解:当归因类型频繁变化时,降低置信度阈值至 0.5 并启用多假设验证;当策略无效时,直接跳入 Tier-3 降级或全局重新规划。同时将该步骤的失败模式标记为"高不确定性",在后续规划中为其配备更丰富的替代方案。

问题 2:反思记忆检索召回率低——规划器未利用相关历史经验

诊断:在规划前打印检索查询和返回的记忆列表,检查查询语义是否与任务匹配。常见原因包括:向量嵌入模型与任务领域不匹配、pattern_signature 设计过于具体导致无法跨任务匹配、记忆库中噪声记忆过多稀释了相关记忆的排名。

缓解:为不同领域使用专门的嵌入模型;将 pattern_signature 的粒度从具体工具名抽象到工具类别(如 postgres_querysql_query);定期执行记忆库清洗——删除 30 天内未被检索命中的记忆、合并语义重复的记忆条目。

问题 3:监控器误报频繁——偏差检测过于敏感

诊断:统计 deviation_score 在 0.3-0.5 区间但最终结果正确的案例比例。如果超过 30%,说明阈值设置过低或验证规则过于严格。

缓解:引入自适应阈值——根据步骤类型动态调整偏差阈值。对于数据查询类步骤(输出结构可预测),阈值设为 0.3;对于生成类步骤(输出具有创造性),阈值放宽到 0.5。同时引入"观察期"机制——偏差分数在阈值附近时先标记不立即触发,连续两步维持偏差才触发修正。

问题 4:Token 消耗超出预算——推理调用过于频繁

诊断:检查推理预算消耗日志,识别消耗大户。常见原因包括:自评估迭代未设置 max_rounds 导致无限循环、多路径规划路径数过多、反思 prompt 包含了过多无关上下文。

缓解:强制设置 max_rounds=3;将多路径规划的路径数限制为 2-3 条;在反思 prompt 中只注入失败步骤及其直接上游步骤的摘要(而非全量执行历史);启用推理链分离存储,避免推理链内容被重复计入上下文。

问题 5:策略进化停滞——Active 策略长期不更新

诊断:检查策略效果看板,确认策略的成功率是否稳定在高位。如果成功率稳定在 90% 以上且长期无变化,可能是该策略已覆盖了所有常见场景(健康状态);如果成功率在 60-70% 波动但不触发 Tuning,可能是滑动窗口设置过大导致灵敏度不足。

缓解:对于健康状态的策略,考虑启动经验蒸馏流程将其固化为规则;对于灵敏度不足的情况,将滑动窗口从 50 次缩小到 20 次,或引入加权滑动窗口(近期应用权重更高)。

11.4 结语

Agent Plan × DeepSeek Harness 框架的核心贡献不在于发明全新的算法,而在于将多种已知技术——ReAct 的交错推理、Plan-and-Execute 的全局规划、Reflexion 的经验反思、Self-Refine 的迭代优化——整合到一个统一的自适应规划架构中,并以 DeepSeek 的推理能力作为贯穿始终的认知引擎。

这个框架的哲学是:错误不是失败的终点,而是进化的起点。每一次失败都被系统性地归因、修正和沉淀,转化为下一次规划的知识资产。随着执行任务的积累,系统的策略库不断丰富,自纠错能力持续增强——这正是"自适应"的真正含义:不是对环境的被动适应,而是基于经验的主动进化。

在 LLM Agent 从"能完成任务"走向"可靠地完成任务"的演进路径上,自纠错能力是一个关键的里程碑。Agent Plan × DeepSeek Harness 为这一能力提供了一个可工程化落地的框架,但其完整实现仍需要在延迟优化、归因准确性、策略管理和多 Agent 协调等方面持续投入。我们期待这一方向能激发更多工程实践和研究探索。

Logo

欢迎加入DeepSeek 技术社区。在这里,你可以找到志同道合的朋友,共同探索AI技术的奥秘。

更多推荐