「规划-行动-反思」架构：Agent 是如何进行反思的？

我们习惯了在 Coze 或 Dify 里拖拽节点，觉得这一切理所当然。但如果想深入 AI 领域，不把这个黑盒打开，好像永远只能在别人定义好的框架里搭积木，而无法理解真正的系统边界。

当我脱离图形化界面，尝试阅读一个真实 Agent 项目的源码时，看到的是一堆 prompt、一堆工具函数。想要从 0 搭建一个 Agent，关键问题依然在那儿：

1. Plan（规划）到底写在哪里？
2. Act（行动）是程序在跑，但怎么让它合理且有秩序地执行？
3. Reflect（反思）是模型的自省吗？它怎么触发下一轮的行动？

下面是我这几天实践摸索过程中整理出来的一些知识点与心得，希望也可以帮助到你。（如有理解错误的地方，恳请指正）

先把黑盒打开：Plan-Act-Reflect 闭环

在我拆解的这个项目中，它的深度研究（Deep Research）模块就是一个最经典的 Plan → Act → Reflect → Answer 闭环。

（这里的 Deep Research 只是这个项目中相对于 web 搜索稍微复杂一点的搜索，并不是我们平时使用的 Deep Research 架构。）

可以抽象成一段伪代码：

def deep_research_loop(user_query):
    # 1. 初始化记忆
    memory_global = []

    while True:
        # Step 1: Plan（规划）
        # 模型不是直接干活，而是给出一份「行动清单」
        actions = model_plan(user_query, memory_global)

        # Step 2: Act（行动）
        # 纯 Python 代码负责跑腿，执行清单上的任务
        new_info = process_actions(actions)
        memory_global.append(new_info)

        # Step 3: Reflect（反思）
        # 模型像个门控器：信息够了吗？
        if model_reflect(user_query, memory_global) == "Enough":
            break

    # Step 4: Final Answer（输出）
    return model_answer(user_query, memory_global)

如果用一段话总结就是：

1. 规划模块先产出一个「动作列表」（结构化 JSON）。
2. 执行模块把这个列表扁平化后，依次串行调用对应工具。
3. 所有工具结果被汇总进一个全局记忆池 memory_global。
4. 反思模块读取用户的「初始问题 + 全局记忆」，判断信息是否够用。
5. 如果不够，再触发一轮搜索执行，把新结果继续追加进全局记忆。
6. 最后把「问题 + 全部搜索证据」拼成最终 Prompt，让模型生成答案。

接下来，我们把这个循环里的三个零件一个个拆开看。

一、Plan：规划模块到底「规划」了什么？

当把代码拆开看，它其实做的就是两件很朴素的事情：

• 判断要不要用工具
• 把大问题拆成可执行的子问题

Plan 本质上是模型写的一张工具调用剧本。 在代码中，agent_plan 函数会构造一个 Prompt，告诉大模型：「你是一个规划模块……请输出 JSON 格式的动作列表」。它不负责推理答案，它只负责输出结构化的数据：

[
  {
    "action_name": "网络搜索",
    "prompts": [
      "特斯拉2025年销量",
      "Model Y 真实车主续航评测"
    ]
  }
]

Agent 架构的核心在于：模型也是系统的一部分，负责产生决策数据。

二、Act：谁在真正动手？

这是最容易被误解的一点。Act 不是模型去调用 API，模型只负责出主意。真正干活的是 Python 程序。

（我之前的文章深度解析 Memory 本质：Agent 和 Chatbot 的差异中有详细写关于 function call 的工作流程，对工具调用模块不是很清楚的话可以去看一下那篇文章的第一个章节。）

表面上看 Act 就是调工具，但真正跑起来之后，很容易乱：同一个工具可能要被调用很多次，不同工具的结果还要汇总进 Memory，如果没有一套规则，程序就会变成一团浆糊。

在这个项目里，Act 做的其实不只是执行，更需要立规矩。它通过三层逻辑，让行动既合理，又有秩序。

1、先把「行动清单」标准化

规划模块给出的原始 JSON，往往是按照人类逻辑组织的：一个动作下挂着好几个问题。但计算机执行时喜欢简单粗暴：一个指令对应一个动作。

所以在真正执行前，代码先执行了一步 adjust_format(original_data)，把复杂的嵌套结构「拍扁」：

[
  {"action_name": "网络搜索", "prompt": "特斯拉2025年销量"},
  {"action_name": "网络搜索", "prompt": "Model Y 真实车主续航评测"}
]

这一步看起来只是改格式，但它其实是在帮 Act 阶段定下一个很重要的约束：所有动作都长一个样，执行逻辑才能统一。

2、按顺序一个一个执行

真正的干活现场在 process_actions 函数里。它像一个严谨的流水线工人：

def process_actions(actions):
    memory = []
    for action in actions:
        action_name = action['action_name']
        prompt = action['prompt']

        # 1. 统一的分发路由
        try:
            if action_name == '本地文档搜索':
                result = rag(prompt)                # 调本地 RAG
            elif action_name == '网络搜索':
                result = web_search_answer(prompt)  # 调 Serper
            else:
                result = f"未知的动作类型: {action_name}"

            # 2. 统一的记账格式
            memory_item = {"提问": prompt, "结果": result}
            memory.append(memory_item)

        # 3. 最简单的容错
        except Exception:
            print(f"执行出错，跳过: {prompt}")

    return memory

这里有几个细节，正是它保持「秩序感」的关键：

• 严格顺序执行：for 循环保证了执行顺序与 plan 的规划顺序完全一致。
• 统一的分发路由：通过 action_name 做一个简易的开关，不认识的动作直接报错，不会把系统搞崩。
• 统一的记账格式：无论调用的工具是大是小，回来的结果都被包装成标准的 {"提问": ..., "结果": ...}。这样后续的 Reflect 模块读取 Memory 时，不需要关心数据来源。
• 必要的容错：整个调用包在 try/except 里。当某个工具挂了，只跳过这一条，而不是让整个 Agent 躺平崩溃。

3、一边跑一边报进度：流式消息 + Act

当用某些 Agent 时，可能它转圈转了半天都没反应，这时会怀疑它是不是死机了。为了解决这个问题，代码在外层加了一个小巧思：通过 SSE 实时报幕。

在执行每个动作前，代码会先给前端发一条消息：

正在执行网络搜索：特斯拉2025年销量……

这样对用户来说，能实时看到 Agent 在干活，把黑箱变成了透明化，体验感极佳。

三、Reflect：反思不是「自省」，是一个门控器

反思这个词听起来很玄学，好像模型在自省。但在代码层面，它其实只是一个很清晰的控制逻辑：要不要再搜一轮？

函数 reflection(user_query, memory_global) 做的事情非常简单：

• 把用户的问题和目前所有工具调用的信息（memory_global）一起发给大模型。
• 问模型：「这些资料够回答问题了吗？如果不够，还需要搜什么？」
• 如果模型觉得不够，它会再次输出一个新的搜索 JSON。
• 程序捕获到这个 JSON，再来一轮 process_actions。

这应该算是 Anthropic 技术博客中提到的 Evaluator-Optimizer（评估-优化） 工作流的一个雏形。它让系统从「搜索一次就回答」升级为「证据驱动的迭代收敛」。

总结一下这个三步循环：其实所谓的 Agent 并不是把一个大模型扔在那里让它自己悟。它全过程都是由我们写好的 Python 代码（人工逻辑），在不断地重复调用不同的大模型。这一整体形成的 Agent，其实更像是一场接力赛（以下是一个比较理想化的比喻）：

• 第一棒（规划阶段）：代码调用模型 A（比如擅长推理的 DeepSeek-R1），问它「怎么解决这个问题？」模型 A 递出一张纸条（JSON），上面写着行动计划。
• 第二棒（执行阶段）：这是唯一没有模型参与的阶段。代码接过纸条，充当「跑腿小弟」，勤勤恳恳地去调 API、查数据库，把所有结果跑一遍，然后整理好塞进背包（Memory）。这个过程是一个小的循环：执行一个动作 → 更新状态 → 记录结果 → 再执行下一个。
• 第三棒（反思阶段）：代码调用模型 B（比如擅长审查的 Qwen），把背包里的东西倒出来给它看，问「这些资料够了吗？合规吗？」如果模型 B 摇头，代码就勒令「跑腿小弟」再去跑一圈。
• 最后一棒（输出阶段）：代码调用模型 C（比如文笔好的 Gemini），把所有经过验证的信息喂给它，生成最终漂亮的回答。

再看产品化的工程架构

拆解完一个 Agent 的大脑，此时可能会有一个疑惑：「我也可以写一个脚本跑通这个流程，但这离一个能给用户用的产品，好像感觉还差很远呢？」

确实，一个在终端里跑的 print(agent.run())，和一个能登录、能保存历史、能流式输出的 Web 应用，中间隔着厚厚的一层工程壁垒。

1、从脚本到产品，到底多了什么？

刚打开项目源码时，我也有些摸不着头脑。明明核心逻辑就是那几百行 Agent 代码，为什么项目里会有几十个文件呢？router、service、schemas、utils……刚开始学习可以先试着看懂以下三层：

API 层（Router）：前台接待。负责处理 HTTP 请求，鉴权（你是谁？），解析参数。它不思考业务，只负责收发信件。

Service 层：业务经理。负责串联逻辑。比如「先查查这个用户有没有知识库？有的话走 A 流程，没的话走 B 流程」。它负责调度资源。

Core 层：核心大脑。这里才是第一章拆解的 Plan-Act-Reflect 逻辑。它不关心你是通过网页访问还是 API 访问，只负责思考。

2、拆解骨架：一个真实项目的路由图

找到这个项目的 router 文件夹里这三个路由文件，刚好对应了产品的三个核心功能区：

• ai_search_rt：管脑子（核心 AI 接口）
• user_rt：管人（登录、注册）
• history_rt：管记忆（历史记录增删改查）

下面分享一下学习 ai_search_rt 里核心接口的一点心得。这个接口，就是第一章中 Plan-Act-Reflect 的宿主。它的路由代码其实非常简单：

@router.post("/deep_research/")
async def deep_research(request: ChatRequest):
    return StreamingResponse(
        # 直接调用那个复杂的 Agent 闭环函数
        final_answer(question),
        # 返回流式响应
        media_type="text/event-stream"
    )

这就是产品化的意义：把复杂的后台逻辑，封装成一个简单的按钮。

此时就可以以这个为起点，「Ctrl + 左键」点击进入 final_answer 这个函数，顺着找到这个函数的位置以及它的意义，以此类推。比如这里，点击进入 reflection 函数，去查看这个函数的位置和逻辑，再一步步顺着找它里面包含的函数。

一开始我也被这几十个文件绕晕了。后来，我把这个项目跑通，从用户的前端视角体验，再结合后端日志的输出，以及上面梳理出的函数流程，重复交叉着验证，渐渐地我就知道如果我想改「搜索逻辑」该去哪里找，想改「回答格式」该去哪里找了。

最后

看到这里，其实就可以发现，Agent 的入门门槛并不是要懂多少高深的概念，而是能不能把模型的决策和程序的执行这条边界真正搞清楚。

这是我摸索着一个已有的项目，对搭建一个 Agent 的理解。希望也可以帮助到你。