Plan-and-Solve

提示:本章代码需基于上一节的client和tools文件

Plan-and-Solve。顾名思义,这种范式将任务处理明确地分为两个阶段:先规划 (Plan),后执行 (Solve)。

1、Plan-and-Solve 的工作原理

Plan-and-Solve Prompting 由 Lei Wang 在2023年提出[2]。其核心动机是为了解决思维链在处理多步骤、复杂问题时容易“偏离轨道”的问题。

与 ReAct 将思考和行动融合在每一步不同,Plan-and-Solve 将整个流程解耦为两个核心阶段,如图4.2所示:

  1. 规划阶段 (Planning Phase): 首先,智能体会接收用户的完整问题。它的第一个任务不是直接去解决问题或调用工具,而是将问题分解,并制定出一个清晰、分步骤的行动计划。这个计划本身就是一次大语言模型的调用产物。
  2. 执行阶段 (Solving Phase): 在获得完整的计划后,智能体进入执行阶段。它会严格按照计划中的步骤,逐一执行。每一步的执行都可能是一次独立的 LLM 调用,或者是对上一步结果的加工处理,直到计划中的所有步骤都完成,最终得出答案。

这种“先谋后动”的策略,使得智能体在处理需要长远规划的复杂任务时,能够保持更高的目标一致性,避免在中间步骤中迷失方向。

我们可以将这个两阶段过程进行形式化表达。首先,规划模型 $\pi_{\text{plan}}$ 根据原始问题 $q$ 生成一个包含 $n$ 个步骤的计划 $P = (p_1, p_2, \dots, p_n)$:

$$ P = \pi_{\text{plan}}(q) $$

随后,在执行阶段,执行模型 $\pi_{\text{solve}}$ 会逐一完成计划中的步骤。对于第 $i$ 个步骤,其解决方案 $s_i$ 的生成会同时依赖于原始问题 $q$、完整计划 $P$ 以及之前所有步骤的执行结果 $(s_1, \dots, s_{i-1})$:

$$ s_i = \pi_{\text{solve}}(q, P, (s_1, \dots, s_{i-1})) $$

最终的答案就是最后一个步骤的执行结果 $s_n$。

Plan-and-Solve 尤其适用于那些结构性强、可以被清晰分解的复杂任务,例如:

2、 规划阶段

为了凸显 Plan-and-Solve 范式在结构化推理任务上的优势,我们将不使用工具的方式,而是通过提示词的设计,完成一个推理任务。

这类任务的特点是,答案无法通过单次查询或计算得出,必须先将问题分解为一系列逻辑连贯的子步骤,然后按顺序求解。这恰好能发挥 Plan-and-Solve “先规划,后执行”的核心能力。

我们的目标问题是:“一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果?”

这个问题对于大语言模型来说并不算特别困难,但它包含了一个清晰的逻辑链条可供参考。在某些实际的逻辑难题上,如果大模型不能高质量的推理出准确的答案,可以参考这个设计模式来设计自己的Agent完成任务。智能体需要:

  1. 规划阶段:首先,将问题分解为三个独立的计算步骤(计算周二销量、计算周三销量、计算总销量)。
  2. 执行阶段:然后,严格按照计划,一步步执行计算,并将每一步的结果作为下一步的输入,最终得出总和。

规划阶段的目标是让大语言模型接收原始问题,并输出一个清晰、分步骤的行动计划。这个计划必须是结构化的,以便我们的代码可以轻松解析并逐一执行。因此,我们设计的提示词需要明确地告诉模型它的角色和任务,并给出一个输出格式的范例。

PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务,并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表,其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""

这个提示词通过以下几点确保了输出的质量和稳定性:

接下来,我们将这个提示词逻辑封装成一个 Planner 类,这个类也是我们的规划器。

# 假定 llm_client.py 中的 HelloAgentsLLM 类已经定义好
# from llm_client import HelloAgentsLLM

class Planner:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def plan(self, question: str) -> list[str]:
        """
        根据用户问题生成一个行动计划。
        """
        prompt = PLANNER_PROMPT_TEMPLATE.format(question=question)
        
        # 为了生成计划,我们构建一个简单的消息列表
        messages = [{"role": "user", "content": prompt}]
        
        print("--- 正在生成计划 ---")
        # 使用流式输出来获取完整的计划
        response_text = self.llm_client.think(messages=messages) or ""
        
        print(f"✅ 计划已生成:\n{response_text}")
        
        # 解析LLM输出的列表字符串
        try:
            # 找到```python和```之间的内容
            plan_str = response_text.split("```python")[1].split("```")[0].strip()
            # 使用ast.literal_eval来安全地执行字符串,将其转换为Python列表
            plan = ast.literal_eval(plan_str)
            return plan if isinstance(plan, list) else []
        except (ValueError, SyntaxError, IndexError) as e:
            print(f"❌ 解析计划时出错: {e}")
            print(f"原始响应: {response_text}")
            return []
        except Exception as e:
            print(f"❌ 解析计划时发生未知错误: {e}")
            return []

3、 执行器与状态管理

在规划器 (Planner) 生成了清晰的行动蓝图后,我们就需要一个执行器 (Executor) 来逐一完成计划中的任务。执行器不仅负责调用大语言模型来解决每个子问题,还承担着一个至关重要的角色:状态管理。它必须记录每一步的执行结果,并将其作为上下文提供给后续步骤,确保信息在整个任务链条中顺畅流动

执行器的提示词与规划器不同。它的目标不是分解问题,而是在已有上下文的基础上,专注解决当前这一个步骤。因此,提示词需要包含以下关键信息:

EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划,一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”,并仅输出该步骤的最终答案,不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

# 历史步骤与结果:
{history}

# 当前步骤:
{current_step}

请仅输出针对“当前步骤”的回答:
"""

我们将执行逻辑封装到 Executor 类中。这个类将循环遍历计划,调用 LLM,并维护一个历史记录(状态)。

class Executor:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def execute(self, question: str, plan: list[str]) -> str:
        """
        根据计划,逐步执行并解决问题。
        """
        history = "" # 用于存储历史步骤和结果的字符串
        
        print("\n--- 正在执行计划 ---")
        
        for i, step in enumerate(plan):
            print(f"\n-> 正在执行步骤 {i+1}/{len(plan)}: {step}")
            
            prompt = EXECUTOR_PROMPT_TEMPLATE.format(
                question=question,
                plan=plan,
                history=history if history else "无", # 如果是第一步,则历史为空
                current_step=step
            )
            
            messages = [{"role": "user", "content": prompt}]
            
            response_text = self.llm_client.think(messages=messages) or ""
            
            # 更新历史记录,为下一步做准备
            history += f"步骤 {i+1}: {step}\n结果: {response_text}\n\n"
            
            print(f"✅ 步骤 {i+1} 已完成,结果: {response_text}")

        # 循环结束后,最后一步的响应就是最终答案
        final_answer = response_text
        return final_answer

现在已经分别构建了负责“规划”的 Planner 和负责“执行”的 Executor。最后一步是将这两个组件整合到一个统一的智能体 PlanAndSolveAgent 中,并赋予它解决问题的完整能力。我们将创建一个主类 PlanAndSolveAgent,它的职责非常清晰:接收一个 LLM 客户端,初始化内部的规划器和执行器,并提供一个简单的 run 方法来启动整个流程。

class PlanAndSolveAgent:
    def __init__(self, llm_client):
        """
        初始化智能体,同时创建规划器和执行器实例。
        """
        self.llm_client = llm_client
        self.planner = Planner(self.llm_client)
        self.executor = Executor(self.llm_client)

    def run(self, question: str):
        """
        运行智能体的完整流程:先规划,后执行。
        """
        print(f"\n--- 开始处理问题 ---\n问题: {question}")
        
        # 1. 调用规划器生成计划
        plan = self.planner.plan(question)
        
        # 检查计划是否成功生成
        if not plan:
            print("\n--- 任务终止 --- \n无法生成有效的行动计划。")
            return

        # 2. 调用执行器执行计划
        final_answer = self.executor.execute(question, plan)
        
        print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")

这个 PlanAndSolveAgent 类的设计体现了“组合优于继承”的原则。它本身不包含复杂的逻辑,而是作为一个协调者 (Orchestrator),清晰地调用其内部组件来完成任务。

完整代码

import os
import ast
from llm_client import HelloAgentsLLM
from dotenv import load_dotenv
from typing import List, Dict

# 加载 .env 文件中的环境变量,处理文件不存在异常
try:
    load_dotenv()
except FileNotFoundError:
    print("警告:未找到 .env 文件,将使用系统环境变量。")
except Exception as e:
    print(f"警告:加载 .env 文件时出错: {e}")

# 规划阶段

"""
规划阶段的目标是让大语言模型接收原始问题,并输出一个清晰、分步骤的行动计划。
这个计划必须是结构化的,以便我们的代码可以轻松解析并逐一执行。
因此,我们设计的提示词需要明确地告诉模型它的角色和任务,并给出一个输出格式的范例。
"""

PLANNER_PROMPT_TEMPLATE = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务,并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表,其中每个元素都是一个描述子任务的字符串。

问题: {question}

请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的:
```python
["步骤1", "步骤2", "步骤3", ...]

"""

# 这个提示词通过以下几点确保了输出的质量和稳定性:
# 角色设定: “顶级的AI规划专家”,激发模型的专业能力。
# 任务描述: 清晰地定义了“分解问题”的目标。
# 格式约束: 强制要求输出为一个 Python 列表格式的字符串,这极大地简化了后续代码的解析工作,使其比解析自然语言更稳定、更可靠。

from llm_client import HelloAgentsLLM


class Planner:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def plan(self, question: str) -> list[str]:
        """
        根据用户问题生成一个行动计划。
        """
        prompt = PLANNER_PROMPT_TEMPLATE.format(question=question)

        # 为了生成计划,我们构建一个简单的消息列表
        messages = [{"role": "user", "content": prompt}]

        print("--- 正在生成计划 ---")
        # 使用流式输出来获取完整的计划
        response_text = self.llm_client.think(messages=messages) or ""

        print(f"✅ 计划已生成:\n{response_text}")

        # 解析LLM输出的列表字符串
        try:
            # 找到```python和```之间的内容
            plan_str = response_text.split("```python")[1].split("```")[0].strip()
            # 使用ast.literal_eval来安全地执行字符串,将其转换为Python列表
            plan = ast.literal_eval(plan_str)
            return plan if isinstance(plan, list) else []
        except (ValueError, SyntaxError, IndexError) as e:
            print(f"❌ 解析计划时出错: {e}")
            print(f"原始响应: {response_text}")
            return []
        except Exception as e:
            print(f"❌ 解析计划时发生未知错误: {e}")
            return []


EXECUTOR_PROMPT_TEMPLATE = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划,一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决“当前步骤”,并仅输出该步骤的最终答案,不要输出任何额外的解释或对话。

# 原始问题:
{question}

# 完整计划:
{plan}

# 历史步骤与结果:
{history}

# 当前步骤:
{current_step}

请仅输出针对“当前步骤”的回答:
"""


class Executor:
    def __init__(self, llm_client):
        self.llm_client = llm_client

    def execute(self, question: str, plan: list[str]) -> str:
        """
        根据计划,逐步执行并解决问题。
        """
        history = ""  # 用于存储历史步骤和结果的字符串

        print("\n--- 正在执行计划 ---")

        for i, step in enumerate(plan):
            print(f"\n-> 正在执行步骤 {i+1}/{len(plan)}: {step}")

            prompt = EXECUTOR_PROMPT_TEMPLATE.format(
                question=question,
                plan=plan,
                history=history if history else "无",
                current_step=step,
            )

            messages = [{"role": "user", "content": prompt}]

            response_text = self.llm_client.think(messages=messages) or ""

            # 更新历史记录,为下一步做准备
            history += f"步骤 {i+1} :{step}\n结果:{response_text}\n\n"

            print(f"✅ 步骤 {i+1} 已完成,结果: {response_text}")

        # 循环结束后,最后一步的响应就是最终答案
        final_answer = response_text
        return final_answer


class PlanAndSolveAgent:
    def __init__(self, llm_client):
        """
        初始化智能体,同时创建规划器和执行器实例。
        """
        self.llm_client = llm_client
        self.planner = Planner(self.llm_client)
        self.executor = Executor(self.llm_client)

    def run(self, question: str):
        """
        运行智能体的完整流程:先规划,后执行。
        """
        print(f"\n--- 开始处理问题 ---\n问题: {question}")

        # 1. 调用规划器生成计划
        plan = self.planner.plan(question)

        # 检查计划是否成功生成
        if not plan:
            print("\n--- 任务终止 --- \n无法生成有效的行动计划。")
            return

        # 2. 调用执行器执行计划
        final_answer = self.executor.execute(question, plan)

        print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")


# --- 5. 主函数入口 ---
if __name__ == "__main__":
    try:
        llm_client = HelloAgentsLLM()
        agent = PlanAndSolveAgent(llm_client)
        question = "一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果?"
        agent.run(question)
    except ValueError as e:
        print(e)


>>>
--- 开始处理问题 ---
问题: 一个水果店周一卖出了15个苹果周二卖出的苹果数量是周一的两倍周三卖出的数量比周二少了5个请问这三天总共卖出了多少个苹果--- 正在生成计划 ---
🧠 正在调用 coding-glm-4.7-free 模型...
✅ 大语言模型响应成功:
["确定周一卖出的苹果数量为15个", "根据周一的销量计算周二卖出的苹果数量(周一数量的2倍)", "根据周二的销量计算周三卖出的苹果数量(周二数量减去5个)", "将周一、周二和周三的销量相加,计算出这三天的总销量"]
✅ 计划已生成:
["确定周一卖出的苹果数量为15个", "根据周一的销量计算周二卖出的苹果数量(周一数量的2倍)", "根据周二的销量计算周三卖出的苹果数量(周二数量减去5个)", "将周一、周二和周三的销量相加,计算出这三天的总销量"]

--- 正在执行计划 ---

-> 正在执行步骤 1/4: 确定周一卖出的苹果数量为15个
🧠 正在调用 coding-glm-4.7-free 模型...
✅ 大语言模型响应成功:
15步骤 1 已完成结果: 15

-> 正在执行步骤 2/4: 根据周一的销量计算周二卖出的苹果数量周一数量的2倍)
🧠 正在调用 coding-glm-4.7-free 模型...
✅ 大语言模型响应成功:
30步骤 2 已完成结果: 30

-> 正在执行步骤 3/4: 根据周二的销量计算周三卖出的苹果数量周二数量减去5个)
🧠 正在调用 coding-glm-4.7-free 模型...
✅ 大语言模型响应成功:
25步骤 3 已完成结果: 25

-> 正在执行步骤 4/4: 将周一周二和周三的销量相加计算出这三天的总销量
🧠 正在调用 coding-glm-4.7-free 模型...
✅ 大语言模型响应成功:
70步骤 4 已完成结果: 70

--- 任务完成 ---
最终答案: 70
感谢你的访问,欢迎交流。