Agent Plan × DeepSeek Harness — 自适应规划与自纠错回路
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 融合范式:自适应规划
自适应规划的核心洞察是:计划不应是静态蓝图,而应是动态文档。它应该像一个经验丰富的项目经理的工作计划——有全局框架,但允许根据实际情况进行局部调整、任务重排、条件分支和紧急回滚。
下表对比了三种范式的关键差异:
| 维度 | ReAct | Plan-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 的架构由六个核心组件构成,通过明确的数据流和控制流串联成一个有机整体。
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 闭环全貌
"思考-执行-反思-修正"闭环是整个框架的核心运转机制。它不是简单的循环,而是一个带有条件分支和状态管理的有控流程。
4.2 思考阶段
思考阶段是闭环的起点,也是 DeepSeek 推理能力发挥核心作用的地方。规划器接收任务描述、历史反思记忆(通过向量检索获取相关的过往经验)和领域知识,通过 Chain-of-Thought 推理完成以下工作:
- 任务分解:将复杂任务拆解为可独立执行的原子步骤,每个步骤有明确的输入、输出和验证标准。
- 依赖分析:识别步骤间的前置/后置关系,构建有向无环图(DAG)。
- 风险评估:为每个步骤标注风险等级(高/中/低),高风险步骤需要配备替代方案。
- 计划输出:生成结构化的计划对象。
以下是计划生成的伪代码:
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 瞬时重试 | ~1s | 0 | 68% | 45% |
| Tier-2 策略切换 | ~3.5s | ~800 | 82% | 30% |
| 局部重新规划 | ~6s | ~2000 | 76% | 15% |
| 降级到替代方案 | ~2s | 0 | 91% | 7% |
| 全局重新规划 | ~12s | ~5000 | 88% | 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 时,系统不直接执行修正,而是:
- 收集更多上下文(执行更详细的工具诊断)
- 生成多个归因假设,通过额外的探测步骤验证
- 若仍无法确定,选择"最安全"的修正策略(即影响范围最小的策略)
当多个反思记忆对同一模式给出矛盾的归因时,系统采用时间衰减加权——较新的记忆权重更高,因为环境可能已变化。同时保留所有矛盾的归因,在后续执行中通过实际结果验证哪种归因更准确。
6.4 归因准确性的量化评估
归因系统的可靠性可以通过两个核心指标衡量:归因精确率(Precision)和归因召回率(Recall)。精确率衡量"系统给出的归因中有多少是正确的",召回率衡量"实际存在的根因中有多少被系统识别"。
在内部测试集(120 个标注了真实根因的失败案例)上的评估结果:
| 归因类型 | 精确率 | 召回率 | F1 | 典型误判模式 |
|---|---|---|---|---|
| 计划错误 | 0.81 | 0.74 | 0.77 | 将"前提错误"误判为"执行错误" |
| 执行错误 | 0.88 | 0.82 | 0.85 | 将"工具内部 bug"误判为"参数错误" |
| 环境错误 | 0.92 | 0.90 | 0.91 | 表现最好,因错误信号明确 |
| 需求错误 | 0.69 | 0.58 | 0.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 策略状态模型
系统不仅在单次任务内进行修正,还能跨任务进化其策略。策略进化基于一个核心假设:相似的错误模式在不同任务中反复出现,对应的成功修正策略应该被固化和复用。
每个策略有四种状态,状态转换由反馈信号驱动:
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 推理链的因果敏感性分析
反事实推理(第三层应用)的输出不仅是修正建议,还包括一份因果敏感性图谱——标注哪些步骤的参数变化对后续步骤影响最大。这份图谱在后续规划中被用于:
- 敏感步骤加固:高敏感性步骤配备更严格的验证规则和更丰富的替代方案。
- 非敏感步骤加速:低敏感性步骤可以跳过部分验证检查以降低延迟。
- 依赖链优化:识别出可以解耦的步骤对,将其改为并行执行。
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 经验沉淀
整个任务执行过程中产生了两条反思记忆:
"execution_error:param_error:postgres_query(field=timestamp)"— PostgreSQL 中保留字作为字段名时需要加双引号。"execution_error:parse_error:csv_writer(type_mismatch)"— CSV writer 对整数类型的处理需要显式类型转换。
这两条记忆被存入长期记忆库。当未来遇到类似任务时,规划器在生成计划阶段就会检索到这些经验,并在 s3 的步骤中直接添加约束:“注意 PostgreSQL 保留字字段名需加引号”,在 s5 的步骤中添加约束:“确保数据类型与 CSV writer 模板匹配”。
这就是策略进化的具体体现——一次踩坑,永久受益。
10.6 完整执行轨迹的量化复盘
对整个任务执行过程进行量化复盘,可以直观展示框架的运作效率:
| 指标 | 值 | 说明 |
|---|---|---|
| 计划步骤总数 | 5 | 初始计划 5 步,未新增步骤 |
| 执行步骤次数 | 7 | 5 次成功 + 2 次失败重试 |
| 修正触发次数 | 2 | s4 SQL 语法 + s5 CSV 类型 |
| Tier-1 重试 | 0 | 无瞬时错误 |
| Tier-2 重试 | 2 | 两次均成功 |
| Tier-3 降级 | 0 | 未触发降级 |
| DeepSeek 推理调用 | 6 | 1 次规划 + 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_query → sql_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 协调等方面持续投入。我们期待这一方向能激发更多工程实践和研究探索。
更多推荐

所有评论(0)