
🧠 记忆锚点:先拆可执行步骤,标依赖;结果偏离就重规划。
💡 答案要点
规划 = 把大任务分解成可执行的小步骤
规划方法对比
| 方法 | 原理 | 优点 | 缺点 | 适用 |
|---|---|---|---|---|
| Chain of Thought | 逐步推理 | 简单直观 | 不可回退 | 简单任务 |
| Tree of Thoughts | 树状搜索 | 可探索多路径 | 计算量大 | 需要试错 |
| Plan-and-Execute | 先整体规划再执行 | 结构清晰 | 计划可能过时 | 明确任务 |
| ReWOO | 预规划+批量执行 | 高效并行 | 灵活性差 | 工具调用多 |
方案1: Plan-and-Execute (推荐⭐)
流程:
Step 1: Planning - 制定完整计划
Step 2: Execution - 逐步执行
Step 3: Replanning - 根据结果调整计划实现:
展开 Python 代码示例(110 行)
python
class PlanAndExecuteAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
def run(self, goal):
# Step 1: 制定计划
plan = self.make_plan(goal)
print(f"计划: {plan}")
# Step 2: 执行每个步骤
results = []
for step in plan:
result = self.execute_step(step, results)
results.append(result)
# Step 3: 检查是否需要重新规划
if self.should_replan(step, result):
plan = self.replan(goal, results)
print(f"重新规划: {plan}")
# Step 4: 综合结果
final_answer = self.synthesize(goal, results)
return final_answer
def make_plan(self, goal):
"""制定计划"""
prompt = f"""
任务: {goal}
请制定详细的执行计划,每个步骤要具体可执行。
输出格式(JSON):
[
{{"step": 1, "action": "搜索最新的AI新闻", "tool": "search"}},
{{"step": 2, "action": "总结新闻要点", "tool": "llm"}},
{{"step": 3, "action": "生成周报", "tool": "llm"}}
]
"""
plan_json = self.llm.generate(prompt)
return json.loads(plan_json)
def execute_step(self, step, previous_results):
"""执行单个步骤"""
tool_name = step["tool"]
action = step["action"]
# 构造上下文(之前步骤的结果)
context = "\n".join([
f"步骤{i+1}结果: {r}"
for i, r in enumerate(previous_results)
])
# 执行工具
if tool_name == "search":
result = self.tools["search"].run(action)
elif tool_name == "llm":
result = self.llm.generate(f"{context}\n\n{action}")
else:
result = self.tools[tool_name].run(action)
print(f"步骤{step['step']}: {action} → {result[:100]}...")
return result
def should_replan(self, step, result):
"""判断是否需要重新规划"""
# 检查执行失败
if "错误" in result or "失败" in result:
return True
# 让LLM判断
prompt = f"""
步骤: {step['action']}
结果: {result}
这个结果是否符合预期? 是否需要调整后续计划?
回答: 是/否
"""
decision = self.llm.generate(prompt).strip()
return decision == "是"
def replan(self, goal, results):
"""重新规划"""
context = "\n".join([f"已完成{i+1}: {r[:50]}..." for i, r in enumerate(results)])
prompt = f"""
原始任务: {goal}
已完成步骤:
{context}
请根据当前进展,重新规划剩余步骤。
"""
new_plan = self.llm.generate(prompt)
return json.loads(new_plan)
# 使用示例
agent = PlanAndExecuteAgent(llm, tools)
result = agent.run("写一份本周AI行业的周报")
# 输出:
# 计划: [
# {"step": 1, "action": "搜索本周AI新闻", "tool": "search"},
# {"step": 2, "action": "总结新闻", "tool": "llm"},
# {"step": 3, "action": "撰写周报", "tool": "llm"}
# ]
# 步骤1: 搜索本周AI新闻 → 找到10篇新闻...
# 步骤2: 总结新闻 → OpenAI发布GPT-5, Google推出Gemini Ultra...
# 步骤3: 撰写周报 → 本周AI行业动态:...方案2: Tree of Thoughts (思维树)
适用: 需要探索多种可能性的任务(如写作、创意)
展开 Python 代码示例(78 行)
python
class TreeOfThoughts:
def __init__(self, llm, depth=3, breadth=3):
self.llm = llm
self.depth = depth # 思考深度
self.breadth = breadth # 每层生成几个候选
def solve(self, problem):
# 根节点
root = TreeNode(problem, score=0)
# 逐层扩展
for level in range(self.depth):
# 对当前层每个节点
for node in self.get_layer_nodes(root, level):
# 生成多个候选下一步
candidates = self.generate_candidates(node, self.breadth)
# 评估每个候选
for candidate in candidates:
score = self.evaluate(candidate)
child = TreeNode(candidate, score=score)
node.add_child(child)
# 找到最优路径
best_path = self.find_best_path(root)
return best_path
def generate_candidates(self, node, k):
"""生成k个候选思路"""
prompt = f"""
当前思路: {node.content}
请生成{k}个不同的后续思路。
"""
responses = []
for _ in range(k):
response = self.llm.generate(prompt, temperature=0.9)
responses.append(response)
return responses
def evaluate(self, thought):
"""评估思路质量"""
prompt = f"""
评估以下思路的质量(0-10分):
{thought}
评分:
"""
score = float(self.llm.generate(prompt).strip())
return score
def find_best_path(self, root):
"""找到最高分路径"""
def dfs(node, path, score):
if not node.children:
return path, score
best = (path, score)
for child in node.children:
candidate_path, candidate_score = dfs(
child,
path + [child.content],
score + child.score
)
if candidate_score > best[1]:
best = (candidate_path, candidate_score)
return best
path, score = dfs(root, [root.content], root.score)
return path
# 使用
tot = TreeOfThoughts(llm, depth=3, breadth=3)
best_solution = tot.solve("写一篇关于AI伦理的文章")
# 会探索 3^3=27 种可能路径,选最优方案3: ReWOO (预规划+批量执行)
优势: 一次性规划所有工具调用,批量并行执行
展开 Python 代码示例(70 行)
python
class ReWOO:
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
def run(self, task):
# Step 1: 一次性规划所有步骤
plan = self.plan_all_steps(task)
# Step 2: 识别可并行的步骤
parallel_groups = self.identify_parallel_groups(plan)
# Step 3: 批量执行
results = {}
for group in parallel_groups:
# 并行执行同组步骤
group_results = self.execute_parallel(group)
results.update(group_results)
# Step 4: 综合结果
return self.synthesize(task, results)
def plan_all_steps(self, task):
prompt = f"""
任务: {task}
请规划完整步骤,标注依赖关系:
格式:
#E1 = Search[最新AI新闻]
#E2 = LLM[总结 #E1]
#E3 = Search[AI政策]
#E4 = LLM[综合 #E2 和 #E3]
"""
plan = self.llm.generate(prompt)
return self.parse_plan(plan)
def identify_parallel_groups(self, plan):
"""识别可并行步骤"""
# #E1 和 #E3 无依赖,可并行
# #E2 依赖 #E1
# #E4 依赖 #E2 和 #E3
groups = [
[plan["E1"], plan["E3"]], # 第1组:并行
[plan["E2"]], # 第2组:等E1完成
[plan["E4"]] # 第3组:等E2,E3完成
]
return groups
def execute_parallel(self, steps):
"""并行执行步骤"""
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = {
executor.submit(self.execute_step, step): step
for step in steps
}
results = {}
for future in concurrent.futures.as_completed(futures):
step = futures[future]
results[step["id"]] = future.result()
return results
# 效果:
# 传统ReAct: 4个步骤串行,耗时20秒
# ReWOO: 步骤1,3并行,耗时12秒 (节省40%)面试话术:
"Agent规划我用Plan-and-Execute,先用LLM制定完整计划,再逐步执行并根据结果调整。复杂任务用Tree of Thoughts探索多路径,每层生成3个候选思路,评分选最优。工具调用多时用ReWOO预规划+批量并行,比ReAct快40%。关键是要能动态调整计划,而不是死板执行。"
📚 参考:CoALA:Cognitive Architectures for Language Agents(规划综述)