ReCAP实战：用递归思维让大模型像程序员一样思考（附LangGraph实现）

最近在折腾智能体系统时，我总在琢磨一个问题：为什么大模型在规划复杂任务时，总给人一种“想得挺好，一做就废”的感觉？比如让它写个自动化运维脚本，它能洋洋洒洒列出一堆步骤，但只要中间某个命令执行失败，整个计划就卡壳了，要么报错退出，要么开始胡言乱语。这和我们程序员写代码的思路完全不同——我们习惯用函数、循环和条件判断来构建程序，遇到错误会捕获异常、重试或者走备用逻辑。有没有可能让大模型也学会这种“递归式”的思考方式呢？

答案是肯定的。这正是ReCAP（递归上下文感知推理与规划）框架要解决的核心问题。它不是一个简单的提示词技巧，而是一套完整的执行范式，旨在赋予大模型类似程序执行栈的递归、回溯和自我修正能力。对于从事工具链编排、自动化运维或者任何需要动态调整计划的开发者来说，理解并实现ReCAP，意味着你能构建出真正健壮、可控的智能体系统。今天，我就结合自己在LangGraph框架下的具体实现，来聊聊如何把ReCAP的递归思想落地为可运行的代码，分享一些调试过程中的真实心得。

1. 为什么传统智能体框架在复杂任务中“力不从心”？

在深入ReCAP之前，我们得先搞清楚现有方案的瓶颈在哪里。目前主流的大模型任务执行范式，大致可以归为三类，它们各有各的“内伤”。

思维链是最基础的，它让模型“一步一步思考”，把推理过程写出来。这提升了单步推理的透明度，但整个思考是一次性完成的。就像你写文章不打草稿，一口气写完，发现中间逻辑有问题，也只能全盘重来。它缺乏执行和修正的机制。

ReAct范式前进了一大步，它结合了推理和行动，让模型可以根据环境反馈决定下一步做什么。这听起来很智能，但其执行本质上是线性的、顺序的。模型根据当前状态决定下一个动作，执行，再根据新状态决定下一个动作。这种模式有两个致命伤：第一，它没有层级结构，无法将一个大任务递归拆解为子任务并分别管理其上下文；第二，一旦某步偏离正轨，它缺乏系统性的回溯和计划重写能力，容易陷入死循环或无效动作。

规划器-执行器模式则采用了更工程化的思路：先让一个“规划器”模型生成完整的任务步骤列表，再由“执行器”模型或代码逐一执行。这比ReAct更有结构，但问题在于规划是“一次性”的。如果执行到第三步时发现第二步的结果与预期不符，整个初始计划就失效了。执行器要么硬着头皮执行错误计划，要么只能报错停止，无法动态调整后续步骤。

注意：这三种范式的共同短板，是缺乏一个类似程序运行时的“控制流”机制。程序之所以能处理复杂逻辑，靠的是函数调用栈、条件分支、循环和异常处理。而大模型之前的推理，更像是在一个平坦的、无状态的文本空间中线性推进。

下面的表格清晰地对比了这几种范式的核心能力差异：

能力维度	思维链	ReAct	规划器-执行器	ReCAP
分步骤推理	✔️	✔️	✔️	✔️
工具调用/执行	✖️	✔️	✔️	✔️
递归任务分解	✖️	✖️	✖️	✔️
执行中动态修正计划	✖️	部分（依赖模型临场反应）	✖️	✔️
上下文隔离与管理	弱（全部在同一个提示中）	弱（历史动作序列）	中（规划与执行分离）	强（每层任务独立上下文）
系统可控性与可调试性	低	中	中	高（清晰的栈结构）

可以看到，ReCAP在递归结构和自我修复能力上是独一无二的。它的目标不是让模型“想得更深”，而是为模型的“思考-行动”过程，套上一个健壮的、可编程的执行引擎。

2. 拆解ReCAP：三个核心组件与一个递归循环

ReCAP的论文将这套机制抽象为三个核心函数和一个执行循环。理解它们，是后续用代码实现的关键。

2.1 上下文感知规划函数 π(C)

这不是一个简单的“生成待办列表”的提示。π(C) 是一个规划函数，它接收当前的上下文C，输出两样东西：

1. 思考：一段关于当前任务如何解决的推理文本。这相当于程序员的“注释”，说明了为什么这么规划。
2. 子任务列表：一个有序的、待执行的子任务列表。每个子任务可以是一个基本动作，也可以是一个复合任务。

关键在于上下文C。它不仅仅包含最初的任务描述，还包括：

• 父任务传递下来的信息。
• 当前已执行步骤的历史和结果。
• 之前步骤产生的任何反馈或错误信息。

因此，π(C) 是“上下文感知”的。同一个任务在不同的执行阶段、收到不同的反馈后，生成的子任务列表可能完全不同。这模拟了程序员在编写函数时，会根据传入参数和之前代码的执行结果，来决定接下来的逻辑。

2.2 递归执行栈：任务即帧

这是ReCAP最具工程美学的部分。它将每个任务（无论是根任务还是子任务）视为一个栈帧。每个帧包含：

• task_description: 任务目标。
• think: 当前规划步骤的推理。
• subtasks: 由π(C)生成的子任务列表。
• cursor: 一个指针，指向subtasks中当前正在执行或下一个要执行的子任务索引。

当执行到一个复合子任务时，系统不是直接处理它，而是创建一个新的栈帧，将其压入执行栈顶，然后开始执行这个新任务。这完全模拟了编程语言中的函数调用。子任务完成后，其帧被弹出，结果返回给父任务帧，父任务帧的cursor前进，并根据结果更新上下文。

2.3 基于反馈的计划修正函数 ρ(C)

这是ReCAP实现“自我修复”的灵魂。每当一个基本动作执行完毕，都会得到一个环境反馈obs（可能是成功的结果，也可能是错误信息）。ρ(C) 函数就负责消化这个obs。

它的工作不是简单的记录，而是动态重写剩余的计划。具体可能包括：

• 删除无效或已完成的子任务。
• 新增修复性的子任务（例如，上一步失败了，新增一个“重试”或“采用备用方案”的任务）。
• 重新排序子任务列表。
• 更新当前的think推理，以反映最新的认知。

ρ(C) 让智能体具备了“吃一堑长一智”的能力。错误不再是终点，而是调整后续行动路线的宝贵信息。

2.4 核心执行循环

将以上三个组件组合起来，就得到了ReCAP的核心算法，它本质上是一个递归函数：

def solve_task(task_frame):
    # 1. 规划：基于当前帧的上下文，生成思考和子任务列表
    think, subtasks = plan(context=task_frame.context)

    while subtasks:  # 子任务列表不为空
        current_subtask = subtasks[0]

        if is_primitive(current_subtask):
            # 2. 执行：如果是基本动作，就执行它，获得反馈obs
            obs = execute(current_subtask)
            # 3. 修正：基于反馈，动态修正剩余的子任务列表
            subtasks = refine(think, subtasks[1:], obs)
        else:
            # 4. 递归：如果是复合任务，递归调用solve_task
            # 这会创建一个新的栈帧并压栈
            solve_task(current_subtask)
            # 递归返回后，当前子任务已完成，从列表中移除
            subtasks = subtasks[1:]

    # 当前任务所有子任务完成，返回（帧弹出）
    return task_frame.result

这个循环在每个栈帧中运行，实现了深度优先的递归任务执行。整个过程就像程序在运行一个由模型动态生成的“决策树”。

3. 用LangGraph构建ReCAP执行引擎

理论很优美，但如何用代码实现？LangGraph是一个绝佳的选择。它专为构建有状态、多环节的智能体工作流而设计，其“图”的概念和“状态”管理，能非常自然地映射ReCAP的栈帧和循环。

下面，我将分步展示如何用LangGraph搭建一个简易但完整的ReCAP框架。我们会模拟一个“编写项目周报”的复杂任务。

3.1 定义状态与栈帧

首先，我们需要定义系统全局状态和栈帧的数据结构。

from typing import TypedDict, List, Annotated, Union
from langgraph.graph import StateGraph, END
import operator

# 定义基本动作的结果反馈
class Observation(TypedDict):
    success: bool
    data: Union[str, dict, None]
    error: Union[str, None]

# 定义任务栈帧
class TaskFrame(TypedDict):
    task_id: str
    description: str  # 任务描述
    think: str  # 当前规划思考
    subtasks: List[str]  # 子任务列表（这里用字符串ID简化表示）
    cursor: int  # 当前执行到的子任务索引
    context: dict  # 该帧的上下文（历史、参数、结果等）
    result: Union[Observation, None]  # 该帧的最终结果

# 定义ReCAP智能体的全局状态
class ReCAPState(TypedDict):
    # 核心：任务执行栈
    task_stack: Annotated[List[TaskFrame], operator.add]
    # 记录所有任务帧的详细信息，便于查找
    frames: dict[str, TaskFrame]
    # 最新一次基本动作的执行反馈
    last_observation: Union[Observation, None]
    # 最终输出
    final_output: Union[str, None]

这里，task_stack 是我们的核心执行栈，frames 字典用于存储所有帧的详细信息。last_observation 用于在plan和refine节点间传递反馈。

3.2 实现关键节点：规划、执行与修正

接下来，我们实现图中的各个节点。首先是规划节点 plan_node。它查看栈顶帧，调用LLM生成新的子任务。

def plan_node(state: ReCAPState) -> dict:
    """规划节点：为栈顶任务生成子任务列表。"""
    if not state['task_stack']:
        return {'final_output': '任务栈为空，执行结束。'}

    current_frame = state['task_stack'][-1]
    frame_id = current_frame['task_id']

    # 构建规划提示词，融入当前上下文
    prompt = f"""
    你是一个任务规划专家。当前需要解决的任务是：{current_frame['description']}

    当前上下文信息：
    {current_frame['context']}

    请进行思考，并将该任务拆解为一个有序的子任务列表。
    输出格式必须严格遵循：
    THINK: [你的推理思考过程]
    SUBTASKS: [子任务1描述, 子任务2描述, ...]
    """
    # 这里模拟LLM调用，实际应接入OpenAI、Anthropic等API
    # 为简化，我们使用一个预定义的规划逻辑
    if current_frame['description'] == "编写项目周报":
        think = "需要先收集数据，然后分析整理，最后撰写报告。"
        subtasks = ["收集代码提交数据", "收集JIRA工单状态", "分析本周工作亮点与风险", "撰写周报正文", "发送周报邮件"]
    elif current_frame['description'] == "收集代码提交数据":
        think = "需要调用Git API获取本周提交记录，并统计主要贡献者。"
        subtasks = ["调用GitLab API", "解析提交数据", "生成统计摘要"]
    else:
        think = "这是一个可以直接执行的基本任务。"
        subtasks = []

    # 更新栈顶帧
    updated_frame = {
        **current_frame,
        'think': think,
        'subtasks': subtasks,
        'cursor': 0
    }
    state['frames'][frame_id] = updated_frame
    state['task_stack'][-1] = updated_frame

    return state

然后是执行与路由节点 execute_or_recurse_node。它决定当前子任务是基本动作还是复合任务。

def execute_or_recurse_node(state: ReCAPState) -> dict:
    """执行/递归节点：判断栈顶任务的下一个子任务类型并处理。"""
    current_frame = state['task_stack'][-1]
    frame_id = current_frame['task_id']

    # 如果当前帧没有子任务或已执行完，则弹出栈（任务完成）
    if current_frame['cursor'] >= len(current_frame['subtasks']):
        completed_frame = state['task_stack'].pop()
        # 如果有父任务，将本帧结果作为反馈传递给父任务
        if state['task_stack']:
            parent_frame = state['task_stack'][-1]
            # 简化处理：将完成的任务描述作为反馈
            state['last_observation'] = Observation(
                success=True,
                data=f"子任务 '{completed_frame['description']}' 已完成。",
                error=None
            )
        return state

    # 获取当前要执行的子任务
    subtask_desc = current_frame['subtasks'][current_frame['cursor']]
    subtask_id = f"{frame_id}_{current_frame['cursor']}"

    # 判断是否为基本动作（这里用简单规则：描述中包含“API”、“发送”、“生成”等词）
    is_primitive = any(word in subtask_desc for word in ["API", "发送", "生成", "解析", "调用"])

    if is_primitive:
        # 基本动作，进入执行分支
        # 更新cursor，因为即将执行当前任务
        new_cursor = current_frame['cursor'] + 1
        updated_frame = {**current_frame, 'cursor': new_cursor}
        state['frames'][frame_id] = updated_frame
        state['task_stack'][-1] = updated_frame
        # 设置一个待执行的基本动作标识
        state['pending_primitive'] = subtask_desc
        return state
    else:
        # 复合任务，创建新帧并压栈（递归）
        new_frame = TaskFrame(
            task_id=subtask_id,
            description=subtask_desc,
            think="",
            subtasks=[],
            cursor=0,
            context={**current_frame['context'], 'parent_task': current_frame['description']},
            result=None
        )
        state['task_stack'].append(new_frame)
        state['frames'][subtask_id] = new_frame
        return state

对于基本动作，我们需要一个执行节点 execute_primitive_node 和一个修正节点 refine_node。

def execute_primitive_node(state: ReCAPState) -> dict:
    """执行节点：执行一个基本动作，并产生反馈。"""
    task_desc = state.pop('pending_primitive', '未知任务')
    # 模拟执行结果，有时成功，有时失败
    import random
    if random.random() > 0.7:  # 30%概率模拟失败
        obs = Observation(success=False, data=None, error=f"执行 '{task_desc}' 时网络超时。")
    else:
        obs = Observation(success=True, data=f"成功执行: {task_desc}。获得模拟数据XYZ。", error=None)

    state['last_observation'] = obs
    return state

def refine_node(state: ReCAPState) -> dict:
    """修正节点：根据上一次执行的反馈，调整当前栈顶任务的剩余计划。"""
    current_frame = state['task_stack'][-1]
    obs = state['last_observation']
    frame_id = current_frame['task_id']

    remaining_subtasks = current_frame['subtasks'][current_frame['cursor']:]

    # 模拟 refine 逻辑：如果上次执行失败，可能插入一个重试或替代任务
    new_subtasks = remaining_subtasks
    new_think = current_frame['think']

    if obs and not obs['success']:
        error_msg = obs['error']
        # 例如，如果错误是网络超时，在剩余任务前插入一个“重试”任务
        if "超时" in error_msg:
            retry_task = f"重试: {remaining_subtasks[0] if remaining_subtasks else '上一个动作'}"
            # 将重试任务插入到剩余任务的最前面
            new_subtasks = [retry_task] + remaining_subtasks
            new_think = f"{current_frame['think']} 上一步遇到超时错误，将先尝试重试。"
            print(f"[Refine Log] 由于错误 '{error_msg}'，在任务 '{current_frame['description']}' 中插入了重试任务。")

    # 更新栈顶帧的剩余子任务和思考
    updated_frame = {
        **current_frame,
        'think': new_think,
        'subtasks': current_frame['subtasks'][:current_frame['cursor']] + new_subtasks
        # cursor 保持不变，指向下一个待处理的任务（现在可能是新插入的重试任务）
    }
    state['frames'][frame_id] = updated_frame
    state['task_stack'][-1] = updated_frame

    return state

3.3 组装LangGraph工作流

最后，我们将所有节点连接起来，形成一个完整的、有条件分支的工作流图。

def create_recap_graph():
    # 初始化图
    workflow = StateGraph(ReCAPState)

    # 添加节点
    workflow.add_node("plan", plan_node)
    workflow.add_node("execute_or_recurse", execute_or_recurse_node)
    workflow.add_node("execute_primitive", execute_primitive_node)
    workflow.add_node("refine", refine_node)

    # 设置入口点
    workflow.set_entry_point("plan")

    # 添加边，定义执行流程
    workflow.add_edge("plan", "execute_or_recurse")

    # 从 execute_or_recurse 出发，有三种可能：
    # 1. 当前任务完成（栈空或帧弹出后栈空） -> END
    # 2. 下一个是基本动作 -> execute_primitive
    # 3. 下一个是复合任务 -> plan (为新帧规划)
    def decide_after_execute_or_recurse(state: ReCAPState) -> str:
        # 检查任务栈是否为空（任务全部完成）
        if not state['task_stack']:
            return "end"
        # 检查是否有待执行的基本动作
        if 'pending_primitive' in state:
            return "execute_primitive"
        # 否则，继续为当前栈顶任务规划或执行下一个子任务
        # 这里简单返回 “plan”，实际上如果帧已有子任务且cursor有效，可以直接跳回execute_or_recurse
        # 我们设计为总是经过plan，让plan节点有机会根据最新上下文调整（即使已有子任务列表）
        return "plan"

    workflow.add_conditional_edges(
        "execute_or_recurse",
        decide_after_execute_or_recurse,
        {
            "end": END,
            "execute_primitive": "execute_primitive",
            "plan": "plan",
        }
    )

    # 执行完基本动作后，必须经过refine节点修正计划
    workflow.add_edge("execute_primitive", "refine")
    # 修正后，回到 execute_or_recurse 节点处理下一个子任务或处理递归返回
    workflow.add_edge("refine", "execute_or_recurse")

    # 编译图
    return workflow.compile()

# 初始化状态并运行
init_state = ReCAPState(
    task_stack=[
        TaskFrame(
            task_id="root",
            description="编写项目周报",
            think="",
            subtasks=[],
            cursor=0,
            context={"project": "LangGraph智能体项目", "week": "第25周"},
            result=None
        )
    ],
    frames={},
    last_observation=None,
    final_output=None
)
frames['root'] = init_state['task_stack'][0]

graph = create_recap_graph()
# 可以运行多步，观察状态变化
for step in range(20):
    result = graph.invoke(init_state)
    print(f"\n=== 步骤 {step+1} ===")
    print(f"栈深度: {len(result['task_stack'])}")
    if result['task_stack']:
        top = result['task_stack'][-1]
        print(f"栈顶任务: {top['description']}")
        print(f"剩余子任务: {top['subtasks'][top['cursor']:]}")
    if result.get('final_output'):
        print(f"最终输出: {result['final_output']}")
        break

运行这段代码，你会看到智能体如何递归地拆解“编写周报”任务，并在模拟执行“调用GitLab API”失败时，自动在计划中插入“重试”步骤。整个过程的状态变迁在 task_stack 中清晰可见，极大增强了系统的可观测性和可调试性。

4. 实战技巧与避坑指南

在真实项目中应用ReCAP，有几个关键点需要特别注意，这些都是在调试中踩过的坑。

第一，上下文的精心设计与管理。 这是ReCAP高效运行的生命线。不能简单地把所有历史信息都塞进每一层的上下文。我的经验是：

• 分层隔离：父任务帧的上下文应包含子任务需要知道的约束和输入，以及子任务完成后需要汇总的输出摘要。子任务无需知道兄弟任务的细节。
• 结果压缩：子任务完成后，不要将其完整的、可能冗长的执行历史全部上传给父任务。应该设计一个summarize_result函数，提取关键结论或状态变更。例如，子任务“分析本周代码提交”的结果，应该被压缩为“本周共有X次提交，主要贡献者为A、B、C，新增Y个功能模块”，而不是原始的API返回的JSON数组。
• 关键错误传递：虽然结果要压缩，但错误类型和关键失败原因必须清晰、结构化地向上传递。这是ρ(C)修正函数能有效工作的基础。

第二，为规划与修正函数设计稳健的提示词。 π(C)和ρ(C)本质上是两个特定的LLM调用。它们的提示词需要高度结构化，并强制输出可解析的格式（如JSON、YAML或严格的文本分隔格式）。对于ρ(C)，提示词必须引导模型基于“错误类型”做出不同的修正策略，而不是笼统地“重新思考”。例如：

给定任务目标、当前计划（思考与剩余步骤）以及上一步的执行结果，请决定如何调整计划。 如果上一步成功，则按原计划继续。 如果上一步失败原因为“[网络超时]”，则在剩余步骤前插入一个“等待2秒后重试[原动作]”的步骤。 如果失败原因为“[权限不足]”，则插入一个“向用户申请XXX权限”的步骤，并暂停后续依赖此权限的步骤。 请输出调整后的新计划。

第三，设置递归深度与超时限制。 和任何递归程序一样，必须防止无限递归或过深的调用栈。需要在全局状态中维护一个max_depth计数器，并在创建新栈帧时检查。同时，对于每个基本动作的执行，都应该有超时控制，避免因某个工具调用挂起而导致整个智能体僵死。

第四，利用栈状态进行可视化调试。 这是ReCAP相比黑盒智能体的巨大优势。你可以在每个循环后打印或记录当前的task_stack，清晰地看到：

• 智能体正在执行哪一层级的任务？
• 每一层任务的当前计划是什么？
• cursor指向哪里？历史执行反馈是什么？ 这种透明度使得定位问题变得极其简单。我通常会用一个简单的函数来可视化栈：

def visualize_stack(stack: List[TaskFrame]):
    for i, frame in enumerate(reversed(stack)):  # 从栈顶开始打印
        indent = "  " * (len(stack) - i - 1)
        print(f"{indent}└─ [{frame['task_id']}] {frame['description']}")
        if frame['think']:
            print(f"{indent}   Think: {frame['think'][:50]}...")
        if frame['cursor'] < len(frame['subtasks']):
            next_task = frame['subtasks'][frame['cursor']]
            print(f"{indent}   Next: -> {next_task}")

当你在控制台看到这个动态缩进的树状结构时，智能体的“思考过程”就变得一目了然了。

最后，从模拟环境开始迭代。 不要一开始就将ReCAP智能体接入真实的生产工具链（如真实的GitLab、JIRA）。先用模拟的执行器（就像我们上面的例子一样），让智能体在可控的、可注入故障的环境下跑通整个递归逻辑。验证其规划、递归、修正的链条是否稳固。然后再逐步替换模拟组件为真实的工具调用。这能节省大量调试时间和API成本。

实现ReCAP的过程，更像是在为LLM编写一个轻量级的“操作系统”或“运行时环境”。它没有改变模型本身的认知能力，但通过外部的递归执行框架，极大地提升了其在复杂、动态环境下的任务达成率和可靠性。当你看到智能体因为一个API错误而自动调整后续步骤，而不是崩溃或胡来时，你会觉得这一切的工程努力都是值得的。