【实战智能体】《大模型应用开发_动手做AI_Agent》_133.[第7章 Agent实战之智能调度] Plan-and-Execute Agent的计划阶段深度剖析
为什么你的Agent总是"想一步做一步"翻车？Plan-and-Execute的"计划阶段"才是隐藏BOSS！深度拆解133课核心机制，手把手教你让AI学会"谋定而后动"，告别执行混乱、任务烂尾的尴尬局面。
目录
- 核心概念篇：为什么Plan-and-Execute需要独立的计划阶段
- 计划生成机制：LLM如何变身"战略指挥官"
- 任务拆解艺术：原子任务的黄金分割法则
- 依赖关系构建：DAG是你的任务调度地图
- 动态调整策略：计划赶不上变化怎么办
- 实战避坑指南：那些让你半夜debug的诡异bug
- 性能优化心法：让计划阶段又快又省
嗨，大家好呀，我是你的老朋友精通代码大仙。接下来我们一起学习 《大模型应用开发_动手做AI_Agent》，震撼你的学习轨迹！
“磨刀不误砍柴工”——这句老话咱们从小听到大，但真到了写Agent的时候，多少人恨不得让AI拿到任务立刻就开干？
我见过太多这样的场景：新手开发者兴致勃勃地搭了个ReAct Agent，结果发现AI像个没头苍蝇，想到哪做到哪，做着做着就忘了最初要干啥。任务稍微复杂一点，直接原地爆炸，输出一堆前后矛盾的中间结果。
这就是Plan-and-Execute架构要解决的问题。而其中的计划阶段，才是整个架构的灵魂所在。今天咱们就把第7章133课的内容掰开了、揉碎了，好好聊聊怎么让AI学会"谋定而后动"。
一、核心概念篇：为什么Plan-and-Execute需要独立的计划阶段
点题
Plan-and-Execute（计划-执行）是一种将"思考规划"与"具体执行"显式分离的Agent架构。与ReAct的"边想边做"不同，它要求AI在动手之前，先完整地制定一份可执行的计划书。
痛点分析
误区一：“计划就是浪费时间，直接干更快”
很多新手觉得，让LLM先写一堆计划再执行，既耗Token又拖慢响应。于是他们直接把ReAct的prompt改改就用，结果遇到复杂任务就翻车。
举个真实案例：有个朋友要做"查询某城市天气，如果下雨就推荐室内景点，否则推荐户外景点，最后生成一份一日游攻略"。他用ReAct，结果AI查完天气直接开始写攻略，完全忘了要先判断室内外这个分支。最后输出的攻略和天气对不上，用户骂骂咧咧。
误区二：计划做得太粗，等于没做
还有人虽然用了Plan-and-Execute，但计划阶段只输出一句话：“先查天气，再推荐景点，最后生成攻略”。这种"计划"没有任何信息量，执行阶段该懵还是懵。
误区三：计划做得太细，把自己绕进去
另一个极端是计划事无巨细，连"打开浏览器"都要写进计划里。结果计划本身生成就要好几秒，执行时还被冗余步骤拖累。
解决方案/正确做法
理解Plan-and-Execute的核心优势：计划阶段买的是"全局一致性"，卖的是"局部灵活性"。
正确的认知框架：
| 维度 | ReAct | Plan-and-Execute |
|---|---|---|
| 适用场景 | 简单任务、工具少、路径明确 | 复杂任务、多步骤、需要条件分支 |
| 思考深度 | 单步局部最优 | 全局最优规划 |
| 容错能力 | 低，一步走错可能越偏越远 | 高，可在重规划时纠正 |
| Token成本 | 低（但可能重复尝试） | 计划阶段较高，执行阶段可控 |
关键设计原则：计划阶段输出的是任务依赖图（DAG），而非线性步骤列表。每个节点包含：任务ID、任务描述、所需输入、预期输出、依赖的前置任务。
# 好的计划输出示例
plan = {
"tasks": [
{
"id": "t1",
"description": "查询北京市今日天气",
"tool": "weather_api",
"inputs": {"city": "北京"},
"outputs": ["weather_condition", "temperature"],
"dependencies": []
},
{
"id": "t2",
"description": "根据天气选择景点类型",
"tool": "condition_router",
"inputs": {"weather": "$t1.weather_condition"},
"outputs": ["venue_type"],
"dependencies": ["t1"]
},
{
"id": "t3a",
"description": "推荐室内景点",
"tool": "venue_search",
"inputs": {"type": "indoor", "city": "北京"},
"outputs": ["venues"],
"dependencies": ["t2"]
},
{
"id": "t3b",
"description": "推荐户外景点",
"tool": "venue_search",
"inputs": {"type": "outdoor", "city": "北京"},
"outputs": ["venues"],
"dependencies": ["t2"]
},
{
"id": "t4",
"description": "生成一日游攻略",
"tool": "itinerary_generator",
"inputs": {"venues": "$t3a.venues or $t3b.venues"},
"outputs": ["itinerary"],
"dependencies": ["t3a", "t3b"] # 或依赖关系
}
]
}
小结
计划阶段不是可有可无的装饰，而是复杂任务可控执行的基石。投入Token买一份好计划，换来的是执行阶段的顺畅和结果的可预期。
二、计划生成机制：LLM如何变身"战略指挥官"
点题
计划阶段的核心是规划器（Planner），通常由一个LLM担任。它的工作是将用户的高层意图，转化为结构化的、可执行的任务图。这要求LLM具备：意图理解能力、任务分解能力、依赖推理能力。
痛点分析
痛点一：LLM"想象力过剩"，编造不存在的工具
新手常遇到的诡异情况：规划器输出了一个看起来很完美的计划，但执行时发现某个任务指定的工具根本不存在，或者参数格式完全不对。
# 翻车示例：LLM编造的"智能"工具
{
"id": "t2",
"description": "分析用户情绪并调整回复策略",
"tool": "emotion_analyzer_with_sentiment_scoring", # 不存在！
"inputs": {"text": "$t1.output", "depth": "deep"},
"outputs": ["emotion_score", "suggested_tone"]
}
痛点二：输出格式不稳定，JSON解析天天报错
让LLM输出结构化数据，结果它一会儿加markdown代码块，一会儿漏个引号，一会儿用单引号代替双引号。你的代码json.loads()天天抛异常，不得不写一堆容错补丁。
痛点三：对工具能力理解偏差，分配错误
LLM以为某个工具能做A，实际上它只能做B。比如把"生成图片"的任务分配给纯文本模型，或者让只能查当前天气的API去预测未来一周。
解决方案/正确做法
方案一：严格的工具描述Schema
给规划器的system prompt里，必须包含完整、准确、无歧义的工具描述。每个工具要说明：功能边界、输入参数（类型、必填/选填、示例）、输出格式、典型使用场景。
PLANNER_SYSTEM_PROMPT = """你是一个任务规划专家。可用工具如下：
【工具名】weather_current
【功能】查询指定城市当前实时天气
【输入】{"city": "string, 城市名，如'北京'", "lang": "string, 可选，语言"}
【输出】{"condition": "天气状况", "temp": "温度数值", "unit": "C/F"}
【限制】只能查当前时刻，不能预测未来
【示例】输入{"city": "上海"} → 输出{"condition": "多云", "temp": 22, "unit": "C"}
【工具名】image_generate
【功能】根据文本描述生成图片
【输入】{"prompt": "string, 英文描述", "size": "enum[256x256,512x512,1024x1024]"}
【输出】{"url": "图片URL", "seed": "生成种子"}
【限制】仅支持英文prompt，最长500字符
...（其他工具）
【输出格式】
你必须输出纯JSON，格式如下：
{
"reasoning": "规划思路说明",
"tasks": [
{
"id": "t1",
"description": "任务描述",
"tool": "工具名（必须从上面列表选）",
"inputs": {...},
"outputs": ["输出字段名"],
"dependencies": ["依赖的任务id列表，可为空"]
}
]
}
【重要规则】
1. 禁止编造工具名，必须从列表中选择
2. 引用前置任务输出使用 $任务id.字段名 格式
3. 确保无循环依赖
4. 纯JSON输出，不要markdown代码块
"""
方案二：输出解析与自修正机制
别指望LLM一次输出完美JSON，设计容错流程：
import json
from typing import Optional, Dict, Any
def robust_json_parse(text: str, max_retry: int = 2) -> Optional[Dict]:
"""带自修正的JSON解析"""
# 第一步：清理常见污染
cleaned = text.strip()
if cleaned.startswith("```json"):
cleaned = cleaned[7:]
if cleaned.startswith("```"):
cleaned = cleaned[3:]
if cleaned.endswith("```"):
cleaned = cleaned[:-3]
cleaned = cleaned.strip()
# 第二步：尝试解析
try:
return json.loads(cleaned)
except json.JSONDecodeError as e:
if max_retry <= 0:
return None
# 第三步：让LLM自己修
fix_prompt = f"""以下JSON解析出错：{str(e)}
请修正JSON格式错误，输出正确的JSON：
{cleaned}
只输出修正后的JSON，不要其他内容："""
# 调用LLM修正...
fixed_text = call_llm(fix_prompt)
return robust_json_parse(fixed_text, max_retry - 1)
# 更稳妥：用Pydantic做结构化验证
from pydantic import BaseModel, Field, validator
class TaskNode(BaseModel):
id: str = Field(..., pattern=r"^t\d+$")
description: str = Field(..., min_length=5)
tool: str
inputs: Dict[str, Any]
outputs: List[str]
dependencies: List[str] = []
@validator('tool')
def tool_must_exist(cls, v, values):
valid_tools = ["weather_current", "image_generate", "search", "calculator"]
if v not in valid_tools:
raise ValueError(f"未知工具: {v}")
return v
@validator('dependencies')
def no_self_dependency(cls, v, values):
task_id = values.get('id')
if task_id in v:
raise ValueError("任务不能依赖自己")
return v
class Plan(BaseModel):
reasoning: str
tasks: List[TaskNode]
方案三：工具能力示例增强（Few-shot）
在prompt中加入2-3个典型任务的规划示例，让LLM"照葫芦画瓢"：
EXAMPLES = """
【示例1：简单顺序任务】
用户：查北京天气，然后生成一张北京地标的图片
规划输出：
{
"reasoning": "需要先获取天气信息，再生成图片。两个任务有数据依赖。",
"tasks": [
{"id": "t1", "description": "查询北京当前天气", "tool": "weather_current",
"inputs": {"city": "北京"}, "outputs": ["condition"], "dependencies": []},
{"id": "t2", "description": "生成北京地标图片", "tool": "image_generate",
"inputs": {"prompt": "Beijing landmark, $t1.condition weather", "size": "512x512"},
"outputs": ["url"], "dependencies": ["t1"]}
]
}
【示例2：含条件分支】
用户：如果上海下雨就推荐博物馆，否则推荐公园
规划输出：
{
"reasoning": "这是一个条件分支任务，需要先用条件路由工具判断，再并行查询两类场所",
"tasks": [
{"id": "t1", "description": "查询上海天气", "tool": "weather_current",
"inputs": {"city": "上海"}, "outputs": ["condition"], "dependencies": []},
{"id": "t2", "description": "判断场所类型", "tool": "condition_router",
"inputs": {"condition": "$t1.condition", "if_rain": "museum", "else": "park"},
"outputs": ["venue_type"], "dependencies": ["t1"]},
{"id": "t3", "description": "查询推荐场所", "tool": "venue_search",
"inputs": {"type": "$t2.venue_type", "city": "上海"},
"outputs": ["venues"], "dependencies": ["t2"]}
]
}
"""
小结
规划器的可靠性取决于：工具描述的清晰度、输出格式的约束力、以及失败时的自修正能力。别跟LLM赌运气，用Schema和验证机制给它套上缰绳。
三、任务拆解艺术：原子任务的黄金分割法则
点题
任务拆解是计划阶段的核心技术。拆得太粗，执行时还是一团乱麻；拆得太细，计划本身就成了负担。原子任务的定义标准是：不可再分、工具可执行、结果可验证。
痛点分析
痛点一：“万能任务”——一个任务干太多事
新手常写这样的任务描述：“分析用户评论的情感、提取关键词、总结要点并生成回复”。一个任务塞了四个子功能，结果找不到匹配的工具，或者强行塞给LLM导致输出质量差。
痛点二：“碎片化灾难”——拆得太细失去意义
另一个极端：把"查询天气"拆成"构建HTTP请求"、“发送请求”、“解析响应”、“提取温度字段”。这些步骤对业务无意义，还增加了失败点和延迟。
痛点三：忽视任务的"可验证性"
有些任务输出模糊，比如"优化这段代码"，怎么算优化成功？没有明确验收标准，执行阶段无法判断是否完成，也无法触发重规划。
解决方案/正确做法
黄金法则：MECE + 工具边界 + 状态可观测
MECE原则：Mutually Exclusive, Collectively Exhaustive
- 任务之间不重叠（避免重复执行）
- 合起来覆盖完整目标（无遗漏）
工具边界：一个原子任务对应一次工具调用
- 外部API调用一次
- LLM生成一次（有明确prompt）
- 代码执行一段（有明确输入输出）
状态可观测：任务完成有明确信号
- 输出字段非空
- 状态码成功
- 通过验证规则
实战拆解示例
复杂任务：“帮我规划从北京到杭州的周末旅行，要便宜、有特色、适合拍照”
错误拆解（太粗）：
t1: 规划整个旅行（输入：需求，输出：完整攻略）
错误拆解（太细）：
t1: 打开航班查询网站
t2: 输入出发地北京
t3: 输入目的地杭州
t4: 选择日期
t5: 点击搜索按钮
...
正确拆解（原子任务）：
plan = {
"tasks": [
# 信息收集层（可并行）
{"id": "t1", "description": "查询北京-杭州往返低价航班",
"tool": "flight_search", "inputs": {...}, "outputs": ["flights"], "dependencies": []},
{"id": "t2", "description": "搜索杭州网红拍照景点",
"tool": "venue_search", "inputs": {"tags": ["photography", "instagrammable"]},
"outputs": ["spots"], "dependencies": []},
{"id": "t3", "description": "查询杭州特色民宿/酒店",
"tool": "hotel_search", "inputs": {"style": "boutique", "budget": "medium"},
"outputs": ["hotels"], "dependencies": []},
# 分析决策层（依赖收集结果）
{"id": "t4", "description": "综合航班、景点、酒店生成3个备选方案",
"tool": "itinerary_planner",
"inputs": {
"flights": "$t1.flights",
"spots": "$t2.spots",
"hotels": "$t3.hotels",
"constraints": {"budget": "low", "theme": "photography"}
},
"outputs": ["options"],
"dependencies": ["t1", "t2", "t3"]},
# 输出优化层
{"id": "t5", "description": "为每个方案生成拍照攻略和预估花费",
"tool": "detail_enhancer",
"inputs": {"options": "$t4.options"},
"outputs": ["final_plans"],
"dependencies": ["t4"]}
]
}
任务描述的最佳实践
| 要素 | 反例 | 正例 |
|---|---|---|
| 动作明确 | “处理图片” | “将图片压缩至宽度800px，保持比例，输出JPEG格式” |
| 输入清晰 | “需要的数据” | “输入：用户ID（string），日期范围（YYYY-MM-DD格式）” |
| 输出具体 | “返回结果” | “输出：订单列表（List[Order]），每个包含id、金额、状态” |
| 成功标准 | “完成分析” | “成功标准：输出至少3个关键发现，每个有数据支撑” |
小结
好的任务拆解像乐高积木——每个块足够简单、接口标准、组合灵活。记住：计划阶段的复杂度应该体现在结构关系上，而非单个任务的描述上。
四、依赖关系构建：DAG是你的任务调度地图
点题
依赖关系决定了任务的执行顺序和并行可能性。Plan-and-Execute使用**有向无环图（DAG）**建模依赖，既保证执行顺序正确，又能最大化并行效率。
痛点分析
痛点一：隐式依赖没发现，执行时数据缺失
规划器漏掉了数据依赖，比如任务B要用任务A的输出，但dependencies里没写。执行时B拿到空值或错误值，整个计划崩盘。
痛点二：过度串行化，浪费并行机会
新手写的计划所有任务都依赖前一个，形成一条长链。实际上很多任务可以并行，比如同时查询多个独立的数据源。
痛点三：循环依赖导致死锁
复杂规划中可能出现A依赖B、B依赖C、C又依赖A的情况。执行调度器陷入死循环，或者拓扑排序报错。
解决方案/正确做法
方案一：显式数据流分析
规划时强制检查：每个任务的输入，是否来自前置任务的输出？
def validate_data_flow(plan: Dict) -> List[str]:
"""验证数据流完整性"""
errors = []
task_outputs = {} # 累积所有可用输出
# 拓扑排序保证顺序
sorted_tasks = topological_sort(plan["tasks"])
for task in sorted_tasks:
# 检查每个输入
for input_key, input_val in task["inputs"].items():
# 检测引用格式 $task_id.field
if isinstance(input_val, str) and input_val.startswith("$"):
ref_parts = input_val[1:].split(".")
if len(ref_parts) != 2:
errors.append(f"{task['id']}: 输入引用格式错误 {input_val}")
continue
ref_task, ref_field = ref_parts
if ref_task not in task_outputs:
errors.append(f"{task['id']}: 引用了未完成的任务 {ref_task}")
elif ref_field not in task_outputs[ref_task]:
errors.append(f"{task['id']}: 引用了不存在的字段 {ref_field}")
# 注册当前任务的输出
task_outputs[task["id"]] = set(task["outputs"])
return errors
# 在规划器输出后立刻验证
raw_plan = call_planner(user_query)
plan = robust_json_parse(raw_plan)
errors = validate_data_flow(plan)
if errors:
# 让规划器修正
fix_prompt = f"计划有以下数据流错误，请修正：\n" + "\n".join(errors)
# ...
方案二：并行调度优化
识别可并行任务组，减少总执行时间：
from collections import defaultdict, deque
def build_execution_batches(tasks: List[Dict]) -> List[List[Dict]]:
"""将DAG分层，每层内任务可并行执行"""
# 构建邻接表和入度表
graph = defaultdict(list)
in_degree = {t["id"]: 0 for t in tasks}
task_map = {t["id"]: t for t in tasks}
for task in tasks:
for dep in task.get("dependencies", []):
graph[dep].append(task["id"])
in_degree[task["id"]] += 1
# Kahn算法分层
batches = []
queue = deque([tid for tid, deg in in_degree.items() if deg == 0])
while queue:
current_batch = list(queue)
batches.append([task_map[tid] for tid in current_batch])
queue.clear()
for tid in current_batch:
for neighbor in graph[tid]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
return batches
# 执行时使用异步并行
async def execute_batch(batch: List[Dict]):
"""并行执行一批无依赖任务"""
tasks = [execute_single_task(t) for t in batch]
results = await asyncio.gather(*tasks, return_exceptions=True)
return dict(zip([t["id"] for t in batch], results))
方案三：循环依赖检测
def detect_cycle(tasks: List[Dict]) -> Optional[List[str]]:
"""检测循环依赖，返回环路上的任务ID"""
WHITE, GRAY, BLACK = 0, 1, 2
color = {t["id"]: WHITE for t in tasks}
parent = {}
def dfs(node_id, path):
color[node_id] = GRAY
task = task_map[node_id]
for dep in task.get("dependencies", []):
if color[dep] == GRAY: # 发现回边
# 提取环路
cycle_start = path.index(dep)
return path[cycle_start:] + [dep]
if color[dep] == WHITE:
result = dfs(dep, path + [dep])
if result:
return result
color[node_id] = BLACK
return None
task_map = {t["id"]: t for t in tasks}
for task in tasks:
if color[task["id"]] == WHITE:
cycle = dfs(task["id"], [task["id"]])
if cycle:
return cycle
return None
小结
DAG是Plan-and-Execute的骨架。花时间在规划阶段建好依赖关系，执行阶段就能既正确又高效。记住：依赖要显式、并行要挖掘、循环要杜绝。
五、动态调整策略：计划赶不上变化怎么办
点题
再完美的计划，执行时也可能遇到意外：工具报错、结果不符合预期、用户中途改需求。**重规划（Replanning）**机制让Agent能根据执行反馈，动态调整后续计划。
痛点分析
痛点一：一错就崩，没有容错机制
很多新手实现Plan-and-Execute时，某个任务一失败就直接返回错误，不会尝试恢复或调整。用户体验极差。
痛点二：重规划太频繁，陷入震荡
稍有异常就全盘推翻重来，结果新计划执行不久又遇到问题，再重规划……Agent在原地打转，Token烧了一堆，事情没办成。
痛点三：重规划时"失忆"，重复已完成的任务
新计划里又包含了已经执行成功的任务，造成重复执行、数据不一致，甚至副作用（比如重复扣款）。
解决方案/正确做法
方案一：分层容错策略
class TaskExecutor:
def __init__(self):
self.max_retries = 2
self.retryable_errors = ["timeout", "rate_limit", "temporary_failure"]
async def execute_with_resilience(self, task: Dict, context: Dict) -> TaskResult:
# 第一层：重试
for attempt in range(self.max_retries + 1):
try:
result = await self.call_tool(task["tool"], task["inputs"])
return TaskResult(success=True, data=result)
except ToolError as e:
if e.code in self.retryable_errors and attempt < self.max_retries:
await asyncio.sleep(2 ** attempt) # 指数退避
continue
break # 不可重试，进入下一层
# 第二层：替代方案
alternative = self.find_alternative_tool(task)
if alternative:
try:
result = await self.call_tool(alternative, task["inputs"])
return TaskResult(success=True, data=result, used_alternative=True)
except:
pass
# 第三层：触发重规划
return TaskResult(success=False, error=e, needs_replanning=True)
方案二：智能重规划触发条件
不是所有失败都需要重规划，设定合理阈值：
| 场景 | 处理方式 | 示例 |
|---|---|---|
| 临时性错误 | 重试即可 | API超时、限流 |
| 工具替代可行 | 换工具执行 | 天气API A不可用，换B |
| 单任务失败但目标可达 | 局部调整 | 某景点查不到，换另一个 |
| 基础假设错误 | 全局重规划 | 用户说"便宜"实际要"豪华" |
| 目标本身不可行 | 终止并解释 | “明天去火星” |
方案三：状态继承的新计划生成
重规划时必须携带已执行状态，避免重复：
def replan(failed_task: Dict, execution_history: List[Dict],
original_plan: Dict, user_query: str) -> Dict:
# 构建状态摘要
completed_tasks = [h for h in execution_history if h["success"]]
failed_info = {
"task_id": failed_task["id"],
"error": failed_task.get("error"),
"partial_result": failed_task.get("partial_result")
}
replan_prompt = f"""原任务：{user_query}
已完成的步骤（**禁止重复执行**）：
{format_completed(completed_tasks)}
当前失败：
{json.dumps(failed_info, indent=2, ensure_ascii=False)}
请生成**剩余任务**的新计划，要求：
1. 不要包含已完成的任务
2. 基于已完成的结果继续
3. 针对失败提供替代方案或调整策略
4. 保持输出格式与原计划一致
输出："""
new_plan_raw = call_llm(replan_prompt)
new_plan = robust_json_parse(new_plan_raw)
# 验证：新计划的任务ID不能与已完成任务重复
completed_ids = {h["task_id"] for h in completed_tasks}
for task in new_plan["tasks"]:
if task["id"] in completed_ids:
raise ReplanError(f"新计划包含已完成的任务: {task['id']}")
return new_plan
方案四：用户介入的优雅降级
当自动重规划也解决不了时，优雅地请求用户帮助：
async def handle_unrecoverable(failure: Dict, context: Dict):
# 生成用户友好的解释
explanation = await generate_explanation(failure, context)
# 提供选项
options = [
{"id": "modify", "label": "修改需求", "example": "比如换个日期或地点"},
{"id": "simplify", "label": "简化任务", "example": "只做核心部分"},
{"id": "manual", "label": "手动提供信息", "example": "直接告诉我结果"}
]
return {
"status": "need_user_input",
"message": explanation,
"options": options,
"current_progress": calculate_progress(context)
}
小结
动态调整是Plan-and-Execute的"保险丝"。设计好分层容错、智能触发、状态继承这三层机制，你的Agent就能像老司机一样，遇到路况变化从容变道，而不是一脚刹车停死。
六、实战避坑指南：那些让你半夜debug的诡异bug
点题
理论再完美，落地总有坑。这一节汇总Plan-and-Execute计划阶段最常见的实战陷阱，以及排查思路。
痛点分析
坑点一：工具描述的"冰山效应"
你只描述了工具浮在水面上的功能，LLM不知道水下的限制。结果规划时看起来很合理，执行时各种报错。
真实案例：某搜索工具描述写"搜索网络信息"，没说明"每次最多返回10条"。LLM规划了一个"搜索并分析前50条结果"的任务，执行时只拿到10条，后续分析基于不完整数据，结论全错。
坑点二：异步执行的数据竞争
并行任务同时修改共享状态，导致数据混乱。
# 错误示例：共享状态无保护
shared_context = {}
async def task_a():
shared_context["temp"] = await fetch_data()
result = process(shared_context["temp"]) # 可能被task_b覆盖！
async def task_b():
shared_context["temp"] = await fetch_other()
# ...
坑点三：重规划的"记忆碎片"
多次重规划后，执行历史冗长，LLM处理时丢失关键信息，或者混淆不同版本的计划。
坑点四：计划与执行的"语义漂移"
计划阶段理解的工具语义，和执行阶段实际行为不一致。比如计划时以为"send_email"会返回发送状态，实际它是个fire-and-forget的异步接口。
解决方案/正确做法
排查工具描述完整性
制作工具描述检查清单：
□ 功能边界：能做什么，明确不能做什么
□ 输入约束：必填/选填、类型、格式、长度限制
□ 输出保证：成功时返回什么，失败时返回什么
□ 性能特征：典型延迟、并发限制、配额
□ 副作用：是否修改数据、是否可重入
□ 版本信息：API版本、弃用计划
异步执行的状态隔离
from dataclasses import dataclass
from typing import Dict, Any
@dataclass(frozen=True)
class TaskContext:
"""不可变的任务上下文，避免竞争"""
task_id: str
inputs: Dict[str, Any]
parent_outputs: Dict[str, Any] # 只读引用前置结果
def with_result(self, result: Any) -> "TaskContext":
"""创建新上下文，不修改原对象"""
return TaskContext(
task_id=self.task_id,
inputs=self.inputs,
parent_outputs={**self.parent_outputs, self.task_id: result}
)
# 每个任务独立上下文
async def execute_task(task: Dict, parent_ctx: TaskContext) -> TaskResult:
ctx = TaskContext(
task_id=task["id"],
inputs=resolve_inputs(task["inputs"], parent_ctx),
parent_outputs=parent_ctx.parent_outputs
)
# 执行时只读ctx，结果通过返回值传递
result = await tool_call(ctx.inputs)
return TaskResult(context=ctx.with_result(result), data=result)
执行历史的智能压缩
def compress_history(full_history: List[Dict], max_tokens: int = 4000) -> str:
"""压缩历史，保留关键信息"""
# 第一层：只保留关键节点
key_events = [
h for h in full_history
if h["type"] in ["plan_created", "task_completed", "replan_triggered", "user_input"]
]
# 第二层：摘要已完成任务
completed_summary = summarize_tasks(
[h for h in key_events if h["type"] == "task_completed"]
)
# 第三层：详细保留最近3个事件
recent_detail = format_detail(key_events[-3:])
return f"""
【执行摘要】
{completed_summary}
【最近事件】
{recent_detail}
"""
计划-执行一致性验证
class ToolContract:
"""工具契约，计划与执行的桥梁"""
def __init__(self, name: str, schema: Dict):
self.name = name
self.input_schema = schema["input"]
self.output_schema = schema["output"]
self.mock_behavior = schema.get("mock_for_planning")
def validate_plan_usage(self, planned_task: Dict) -> List[str]:
"""验证计划中的使用是否符合契约"""
errors = []
# 检查输入是否符合schema
# 检查输出引用是否合理
return errors
def mock_for_planning(self, inputs: Dict) -> Dict:
"""规划阶段用mock验证数据流"""
return self.mock_behavior(inputs)
小结
坑是踩出来的，经验是debug堆出来的。建立系统性的检查清单、保持状态隔离、做好契约验证，能让你少熬很多夜。记住：诡异bug背后，往往是某个假设 silently failed。
七、性能优化心法：让计划阶段又快又省
点题
计划阶段调用LLM生成完整计划，是主要的Token消耗点和延迟来源。优化目标：在保证质量的前提下，减少Token、降低延迟、提升可缓存性。
痛点分析
痛点一：相似任务重复规划，Token白白浪费
用户连续问"北京明天天气"、“上海明天天气”、“广州明天天气”，每次都要重新生成几乎一样的计划结构。
痛点二：规划模型"杀鸡用牛刀"
用GPT-4做简单任务的规划，延迟2秒+，成本高。其实很多结构化任务用更小模型也能做好。
痛点三：复杂计划的生成时间过长
任务多、依赖复杂时，LLM生成JSON的时间线性增长，用户等待焦虑。
解决方案/正确做法
方案一：计划模板与缓存
from functools import lru_cache
import hashlib
class PlanCache:
def __init__(self):
self.template_store = {} # 意图模板
self.instance_cache = {} # 具体实例
def get_plan_template(self, intent_signature: str) -> Optional[Dict]:
"""获取意图对应的计划模板"""
# 意图签名：从query提取的结构化表示
# "查{城市}天气" → "weather_query_template"
return self.template_store.get(intent_signature)
def instantiate_template(self, template: Dict, params: Dict) -> Dict:
"""用具体参数实例化模板"""
# 替换模板中的占位符
plan_json = json.dumps(template)
for key, val in params.items():
plan_json = plan_json.replace(f"{{{key}}}", json.dumps(val))
return json.loads(plan_json)
@lru_cache(maxsize=1000)
def cached_plan(self, query_hash: str) -> Optional[Dict]:
"""缓存具体查询的计划"""
return self.instance_cache.get(query_hash)
# 使用示例
def plan_with_cache(user_query: str):
# 1. 提取意图签名
intent = extract_intent(user_query) # e.g., "weather:city"
# 2. 检查模板
template = cache.get_plan_template(intent)
if template:
params = extract_params(user_query)
return cache.instantiate_template(template, params)
# 3. 回退到LLM规划
return llm_plan(user_query)
方案二：分层模型策略
class TieredPlanner:
def __init__(self):
self.tiers = {
"simple": {
"model": "gpt-3.5-turbo",
"max_tasks": 3,
"max_deps": 2,
"cost_per_1k": 0.0015
},
"standard": {
"model": "gpt-4-turbo-preview",
"max_tasks": 10,
"max_deps": 5,
"cost_per_1k": 0.01
},
"complex": {
"model": "gpt-4",
"max_tasks": 50,
"max_deps": 20,
"cost_per_1k": 0.03
}
}
def select_tier(self, query: str, estimated_complexity: float) -> str:
"""根据查询特征选择规划层级"""
# 快速启发式判断
if estimated_complexity < 0.3:
return "simple"
elif "如果" in query or "根据" in query: # 条件逻辑
return "complex"
elif estimated_complexity > 0.7:
return "complex"
else:
return "standard"
async def plan(self, query: str) -> Dict:
complexity = estimate_complexity(query) # 轻量级分类器
tier = self.select_tier(query, complexity)
config = self.tiers[tier]
return await call_llm(
model=config["model"],
prompt=build_prompt(query, config),
response_format={"type": "json_object"}
)
方案三：流式规划与渐进展示
async def streaming_plan(user_query: str):
"""边生成边验证，提前返回部分结果"""
# 第一阶段：快速生成骨架（高温度，快）
skeleton = await quick_skeleton_plan(user_query)
yield {"type": "skeleton", "data": skeleton}
# 第二阶段：并行填充细节
detail_tasks = [
expand_task_details(task)
for task in skeleton["tasks"]
]
for completed in asyncio.as_completed(detail_tasks):
detail = await completed
yield {"type": "task_detail", "data": detail}
# 已完成的可以开始执行了！
if can_start_early(detail):
yield {"type": "executable", "data": detail}
# 第三阶段：最终整合
final_plan = await consolidate(skeleton, detail_tasks)
yield {"type": "complete", "data": final_plan}
方案四：Prompt压缩技术
def compress_tool_descriptions(tools: List[Dict], query: str) -> List[Dict]:
"""根据查询相关性筛选工具描述"""
# 用embedding计算相关性
query_vec = embed(query)
scored_tools = []
for tool in tools:
tool_vec = embed(tool["description"])
similarity = cosine_similarity(query_vec, tool_vec)
scored_tools.append((similarity, tool))
# 取Top-K，其余用摘要
scored_tools.sort(reverse=True)
essential = scored_tools[:5] # 详细描述
others = scored_tools[5:10] # 简要描述
result = []
for sim, tool in essential:
result.append(tool) # 完整描述
for sim, tool in others:
result.append({
"name": tool["name"],
"one_liner": tool["description"][:100] + "..."
})
return result
小结
性能优化不是一味省钱，而是在用户体验和成本之间找平衡。模板缓存省的是重复劳动，分层模型省的是过度配置，流式规划省的是用户等待。记住：先度量，再优化，别凭感觉。
写在最后
聊到这里，Plan-and-Execute的计划阶段应该不再是个黑盒了。咱们从为什么要分离计划与执行，到规划器怎么设计、任务怎么拆解、依赖怎么管理、出了问题怎么调整、常见坑怎么避、性能怎么优化——这七个维度，构成了一个相对完整的知识体系。
但我最想说的是：别被架构的复杂度吓住。Plan-and-Execute不是银弹，ReAct也不是过时货。简单任务用简单方案，复杂任务才值得上复杂架构。很多新手的问题不是"学得太少"，而是"想得太复杂"，一个天气查询都要上DAG，反而把系统搞臃肿。
编程之路就是这样，每一步都是在权衡。权衡正确性与成本，权衡灵活性与可控性，权衡当下的交付与长期的维护。Plan-and-Execute教给我们的，不只是技术方案，更是一种思维方式：面对复杂问题，先停下来想清楚了再动手，往往比急着开干更高效。
这本书的第7章，133课，值得你反复读、动手试、踩坑再爬出来。Agent开发这个领域变化快，但底层的设计思想是稳定的。把基础打扎实了，新框架出来你也能快速上手。
保持好奇，持续学习，你也能成为代码高手。咱们下节课见！
关注私信备注：“资料代找获取”，全网计算机学习资料代找：例如:
《课程：2026 年多模态大模型实战训练营》
《课程：AI 大模型工程师系统课程 (22 章完整版 持续更新)》
《课程：AI 大模型系统实战课第四期 (2026 年开课 持续更新)》
《课程：2026 年 AGI 大模型系统课 23 期》
《课程：2026 年 AGI 大模型系统课 21 期》
《课程：AI 大模型实战课 8 期 (2026 年 2 月最新完结版)》
《课程：AI 大模型系统实战课三期》
《课程：AI 大模型系统课程 (2026 年 2 月开课 持续更新)》
《课程：AI 大模型全阶课程 (2025 年 12 月开课 2026 年 6 月结课)》
《课程：AI 大模型工程师全阶课程 (2025 年 10 月开课 2026 年 4 月结课)》
《课程：2026 年最新大模型 Agent 开发系统课 (持续更新)》
《课程：LLM 多模态视觉大模型系统课》
《课程：大模型 AI 应用开发企业级项目实战课 (2026 年 1 月开课)》
《课程：大模型智能体线上速成班 V2.0》
《课程：Java+AI 大模型智能应用开发全阶课》
《课程：Python+AI 大模型实战视频教程》
《书籍：软件工程 3.0: 大模型驱动的研发新范式.pdf》
《课程：人工智能大模型系统课 (2026 年 1 月底完结版)》
《课程：AI 大模型零基础到商业实战全栈课第五期》
《课程：Vue3.5+Electron + 大模型跨平台 AI 桌面聊天应用实战 (2025)》
《课程：AI 大模型实战训练营 从入门到实战轻松上手》
《课程：2026 年 AI 大模型 RAG 与 Agent 智能体项目实战开发课》
《课程：大模型训练营配套补充资料》
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
8
0- 0
已为社区贡献193条内容
所有评论(0)