Planning、Reflection、RAG 分别解决什么问题：分清 Agent 系统的职责边界

本文是「Agent 基础与工程」系列专栏的第 7 篇。专栏总览参见：《Agent 基础认知与工程架构全景》。

在构建较复杂的 Agent 系统时，Planning（任务规划）、Reflection（自省反思）与 RAG（检索增强生成）常被打包作为“高级智能体”的标准配置。然而，如果在系统设计初期没有明确这三者的职责边界与失效模式，极易在工程落地中产生典型的能力错配：

  • 外部事实缺失时，试图通过“让模型多思考几轮 Planning”来解决知识盲区，导致一本正经地编造幻觉；
  • 能够通过确定性单元测试或正则断言校验的输出，强行引入另一个大模型执行 Reflection，导致推理延迟倍增且校验者容易对生成者盲目附和；
  • 面对高度不确定的动态环境，试图在第一轮生成包含数十个子步骤的超长静态计划，第一步执行出错便导致整体链路崩溃。

明确三者的本质分工，是构建高效、确定性交付系统的关键前提。

核心职责解耦：三元协同架构

从系统工程角度看，三者分别向大模型注入了三种截然不同的确定性能力：

核心组件 核心解题域 输入 输出 典型失败模式
Planning（规划） 动作拓扑与调度 原始目标、当前状态、可用工具清单 结构化任务分解（DAG 或步骤序列） 过度规划（Over-planning）：假设前提过多，一旦环境扰动计划全部失效
RAG（检索） 外部事实与上下文注入 检索 Query、过滤条件 相关的文本切片（Chunks）与结构化元数据 上下文污染（Context Pollution）：召回过多低质量或矛盾切片，冲垮注意力
Reflection（自省） 输出断言与收敛保证 执行结果、验收标准、历史轨迹 评分、布尔断言、归因诊断（Critique） 盲目确认（Sycophancy）：反思模型对生成模型的错误保持附和与妥协

1. Planning：任务拓扑分解与动态编排

Planning 的本质是将复杂的高维目标降低到当前工具能够直接处理的粒度。

常见的三种 Planning 模式

  1. 先规划后执行（Plan-and-Solve / 一次性分解）：
    • 流程：在启动执行前，由 Planner 模型一次性将目标切分为子任务列表（Step 1 到 Step N），随后由 Executor 严格按序执行。
    • 适用场景：依赖关系固定、不确定性极低的批处理任务（例如“下载文件 解压 解析 CSV 导入 MySQL”）。
    • 缺点：对运行期异常极度脆弱。
  2. 交替式动态步进（Dynamic Step-by-step / ReAct）：
    • 流程：模型不预先锁定全量步骤，而是执行一次 Thought $\to$ Action $\to$ Observation，根据当前观察到的真实数据，现场决定下一步动作。
    • 适用场景：调试排障、网页探索、复杂信息调研。
    • 缺点：容易在局部陷入死循环，步数不可控。
  3. 分层式规划（Hierarchical Planning / Supervisor-Worker）：
    • 流程：顶层规划者（Supervisor）维护全局高阶里程碑，底层执行者（Worker）在受控的局部小循环内解决单点子任务，定期向上层汇报状态并触发重新规划（Replanning）。
    • 适用场景：软件工程开发、复杂业务研报撰写。

2. Reflection：质量断言与错误纠偏

很多系统在模型产出内容后不做任何检验便直接返回给用户，导致幻觉或语法错误直接暴露。Reflection 是在系统闭环中建立的质量防护栅栏。

生产级 Reflection 的三种实现层级

在工程实践中，永远优先使用代码驱动的确定性断言（L1），而非无节制地调用 LLM 进行模糊复盘（L2）。例如：

  • 校验 SQL 是否符合语法，直接调用 sqlparse 或在只读连接上执行 EXPLAIN，而不是让模型“检查一下这段 SQL 写得对不对”；
  • 校验代码修复是否生效，直接运行对应的测试用例（pytest），依据退出码判断；
  • 只有当评估对象属于主观自然语言质量（如“回复是否符合品牌语调”、“总结是否遗漏核心事实”）时，才下沉至专职的 Critic 模型。

3. RAG：事实与参数外知识注入

RAG 在 Agent 系统中的角色不是简单的“向量召回”，而是作为一种按需调用的感知工具（Agentic RAG）。

在普通 RAG 流水线中，检索动作是预先固化的：接收用户输入 检索向量库 拼装 Prompt 生成文本。
而在 Agent 架构中，RAG 表现为一个或多个具名工具：

  • search_product_manual(query="错误码 0x44")
  • query_customer_profile(user_id="U8901")

模型只有在自主判断“当前已知事实不足以回答问题”时，才会主动发起检索调用，并根据召回结果的质量决定是否需要改写关键词重新发起二次检索。

模块协同实战：代码级调度实现

以下通过 Python 演示一个清晰解耦的闭环处理框架，完整呈现 Planning、RAG 注入、Execution 与确定性 Reflection 的协同推进：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
import json
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field

# 1. 结构化定义规划输出
class SubTask(BaseModel):
    id: int
    action_type: str = Field(..., description="调用类型: rag / tool / compute")
    instruction: str
    tool_name: Optional[str] = None
    tool_args: Optional[Dict[str, Any]] = None

class ExecutionPlan(BaseModel):
    thought: str
    subtasks: List[SubTask]

# 2. 模拟 RAG 与外部工具能力
class ToolBox:
    @staticmethod
    def rag_search(query: str) -> str:
        # 模拟从知识库召回精准切片
        if "安全合规" in query:
            return "【企业安全制度第 4 条】：生产环境导出数据单次不得超过 1000 条，必须进行脱敏。"
        return "未检索到相关规章。"

    @staticmethod
    def export_data(table: str, limit: int) -> Dict[str, Any]:
        return {"status": "success", "exported_rows": limit, "table": table}

# 3. 确定性 Reflection 断言器
class DeterministicCritic:
    @staticmethod
    def evaluate_export_action(limit: int, rag_context: str) -> tuple[bool, str]:
        # 基于程序规则进行断言，不依赖模型主观臆断
        if "不得超过 1000 条" in rag_context and limit > 1000:
            return False, f"合规断言失败：请求导出 {limit} 条，超出制度上限 1000 条约束。"
        return True, "断言通过"

# 4. 驱动中枢
def run_orchestrated_task(user_goal: str):
    print(f"收到目标: {user_goal}")
    
    # 阶段 1: Planning 模块生成执行步骤
    # 模拟 Planner 模型生成的静态 DAG
    plan = ExecutionPlan(
        thought="用户需要导出生产数据，需要先检索合规上限，再执行导出操作。",
        subtasks=[
            SubTask(id=1, action_type="rag", instruction="检索数据导出的安全合规规则"),
            SubTask(id=2, action_type="tool", instruction="执行订单数据导出", 
                    tool_name="export_data", tool_args={"table": "orders", "limit": 2000})
        ]
    )
    
    context_memory: Dict[str, Any] = {}

    # 阶段 2: 按计划分发推进
    for task in plan.subtasks:
        print(f"\n--- 执行子任务 {task.id}: {task.instruction} ---")
        
        if task.action_type == "rag":
            # RAG 事实注入
            retrieved_chunk = ToolBox.rag_search(task.instruction)
            context_memory["compliance_rules"] = retrieved_chunk
            print(f"[RAG 注入成功]: {retrieved_chunk}")
            
        elif task.action_type == "tool":
            # 在执行高危动作前触发 Reflection 拦截
            args = task.tool_args or {}
            limit = args.get("limit", 0)
            rules = context_memory.get("compliance_rules", "")
            
            # 运行自省断言
            passed, critique = DeterministicCritic.evaluate_export_action(limit, rules)
            if not passed:
                print(f"[Reflection 拦截报错]: {critique}")
                print("[触发重规划]: 纠正参数 limit 为 1000 并重试执行...")
                args["limit"] = 1000
                
            # 执行真实工具调用
            res = ToolBox.export_data(**args)
            print(f"[工具执行完成]: {res}")
            context_memory["export_result"] = res

    print("\n任务全链路推进完毕，交付结果。")

if __name__ == "__main__":
    run_orchestrated_task("导出生产数据库 orders 表的最近 2000 条记录")

工程选型与避坑决策树

面对新业务需求时，通过以下决策矩阵定位所需组件，避免概念堆砌：

1
2
3
4
5
6
7
8
9
10
11
12
13
遇到系统瓶颈时：
├── 1. 模型不知道具体事实、私有字段、最新政策？
│   └── 接入 RAG：补充外部检索工具，提供确切文档依据。
│       （禁止：通过多轮让模型思考或编造计划来弥补知识缺失）
│
├── 2. 任务包含多步操作、顺序依赖、工具调用链条长？
│   └── 接入 Planning：拆分子任务状态机，明确先后拓扑。
│       （禁止：在高度不确定的动态未知场景使用长静态计划）
│
└── 3. 模型产出经常出现格式错误、字段缺失、逻辑漏洞或越权风险？
    └── 接入 Reflection：
        ├── 能够用代码规则/测试用例/Schema 检验的 ──> 用确定性程序断言
        └── 纯主观内容/语气风格评估 ──> 使用独立 Prompt 的专职 Critic 模型

系列导航与参考