AI–智能体Plan-and-Solve架构
0.前言
在前面我们掌握了ReAct这种反应式,步进式决策的智能体范式后,接下来将介绍的是Plan-and-Solve,这种架构的核心是将任务分为了两个阶段,先规划,后执行。如果说ReAct是一位通过逐层盖楼边盖边规划,最终盖成的工程师,那 Plan-and-Solve则是在动工前绘制完整蓝图,最后严格按照图纸操作的工程师
1.Plan-and-Solve的工作原理
Plan-and-Solve Prompting 由 Lei Wang 在2023年提出,其核心动机就是将流程结构为两个核心,从而解决思维链在处理多出步骤,复杂问题时容易偏离轨道的问题
首先,智能体会接收用户的完整问题。它的第一个任务不是直接去解决问题或调用工具,而是将问题分解,并制定出一个清晰、分步骤的行动计划。这个计划本身就是一次大语言模型的调用产物。
在获得完整的计划后,智能体进入执行阶段。它会严格按照计划中的步骤,逐一执行。每一步的执行都可能是一次独立的 LLM 调用,或者是对上一步结果的加工处理,直到计划中的所有步骤都完成,最终得出答案。
从流程中可以看出,Plan-and-Solve 尤其适用于那些结构性强、可以被清晰分解的复杂任务,例如:
- 多步数学应用题:需要先列出计算步骤,再逐一求解。
- 需要整合多个信息源的报告撰写:需要先规划好报告结构(引言、数据来源A、数据来源B、总结),再逐一填充内容。
- 代码生成任务:需要先构思好函数、类和模块的结构,再逐一实现。
2.Plan-and-Solve逻辑实现
为了凸显 Plan-and-Solve 范式在结构化推理任务上的优势,我们将不使用工具的方式,而是通过提示词的设计,完成一个推理任务。,之类推理的特点是答案无法通过单次查询或计算得出,必须先将问题分解为一系列逻辑连贯的子步骤,然后按顺序求解。
依旧系统提示词这一块,为了让计划结构化,以便于我们的代码进行轻松解析和拆解执行,因此设计的提示词需要明确告诉模型角色以及任务,最好是能够提供一个输出格式的例子
1 2 3 4 5 6 7 8 9 10 11 12
| PLANNER_PROMPT_TEMPLATE = """ 你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。 请确保计划中的每个步骤都是一个独立的、可执行的子任务,并且严格按照逻辑顺序排列。 你的输出必须是一个Python列表,其中每个元素都是一个描述子任务的字符串。
问题: {question}
请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的: ```python ["步骤1", "步骤2", "步骤3", ...] ``` """
|
这样子通过角色设定,基于LLM模型更清晰的定位,任务描述则赋予了分解问题的目标和要求,最后的格式约束是为了让输出更加清晰
有了规划专家的提示词,接下来,我们定义我们的规划器类。
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
|
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}") try: plan_str = response_text.split("```python")[1].split("```")[0].strip() 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 []
|
在规划器生成了行动蓝图后,我们就需要一个执行器来逐一完成计划中的任务。执行器不仅负责调用大语言模型来解决每个子问题,还承担着一个至关重要的角色:状态管理。它必须记录每一步的执行结果,并将其作为上下文提供给后续步骤,确保信息在整个任务链条中顺畅流动
因此它的目标不是分解问题,而是在已有上下文的基础上,专注解决当前这一个步骤,需要的内容如下:
- 原始问题: 确保模型始终了解最终目标。
- 完整计划: 让模型了解当前步骤在整个任务中的位置。
- 历史步骤与结果: 提供至今为止已经完成的工作,作为当前步骤的直接输入。
- 当前步骤: 明确指示模型现在需要解决哪一个具体任务。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
| EXECUTOR_PROMPT_TEMPLATE = """ 你是一位顶级的AI执行专家。你的任务是严格按照给定的计划,一步步地解决问题。 你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。 请你专注于解决“当前步骤”,并仅输出该步骤的最终答案,不要输出任何额外的解释或对话。
# 原始问题: {question}
# 完整计划: {plan}
# 历史步骤与结果: {history}
# 当前步骤: {current_step}
请仅输出针对“当前步骤”的回答: """
|
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
| 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 方法来启动整个流程。
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
| 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}") plan = self.planner.plan(question) if not plan: print("\n--- 任务终止 --- \n无法生成有效的行动计划。") return
final_answer = self.executor.execute(question, plan) print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")
|
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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
| import re from LLM_Client import HelloAgentsLLM import ast
PLANNER_PROMPT_TEMPLATE = """ 你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。 请确保计划中的每个步骤都是一个独立的、可执行的子任务,并且严格按照逻辑顺序排列。 你的输出必须是一个Python列表,其中每个元素都是一个描述子任务的字符串。
问题: {question}
请严格按照以下格式输出你的计划,```python与```作为前后缀是必要的: ```python ["步骤1", "步骤2", "步骤3", ...] ``` """
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}") try: plan_str = response_text.split("```python")[1].split("```")[0].strip() 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}") plan = self.planner.plan(question) if not plan: print("\n--- 任务终止 --- \n无法生成有效的行动计划。") return
final_answer = self.executor.execute(question, plan) print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")
if __name__ == '__main__': try: llm_client = HelloAgentsLLM() agent = PlanAndSolveAgent(llm_client) question = "一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果?,请" agent.run(question) except ValueError as e: print(e)
|