第5章：Planning — 让 Agent 学会"思考"
📌 本章目标
- 理解 ReAct 模式的完整实现
- 掌握 Chain-of-Thought（思维链）提示技巧
- 学会任务分解：把大任务拆成小步骤
- 实现带自我反思的 Agent
5.1 为什么 Tool Calling 还不够？
第3章的 Tool Calling Agent 能工作，但它有一个问题：它不怎么"想"，直接"做"。
看一个例子：
用户: "帮我规划一个周末北京两日游" Tool Calling Agent 的行为: 🔧 调 get_weather("北京") → 直接回复答案 问题: 它没有"思考过程"—— - 没有先想"我需要哪些信息？天气、景点、交通..." - 没有规划"先查天气 → 再找景点 → 最后排路线" - 没有验证"我拿到的信息够不够完整？"ReAct 模式的核心就是让 Agent先把思考说出来，再行动。
5.2 ReAct 模式完整实现
ReAct =Reasoning +Acting，要求 Agent 按固定格式输出：
Thought: 我现在需要做什么？为什么？ Action: 工具名称 Action Input: 工具参数（JSON） Observation: （系统填入工具执行结果） ...（重复上述步骤） Thought: 我现在有足够信息了 Final Answer: 最终回答完整代码实现
""" ReAct Agent —— 会"思考"的 Agent 使用前请先: pip install openai python-dotenv 并在 .env 文件中设置: DEEPSEEK_API_KEY=sk-你的密钥 """importjsonimportreimportosfromopenaiimportOpenAIfromdotenvimportload_dotenv load_dotenv()# DeepSeek 客户端client=OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),base_url="https://api.deepseek.com/v1")defchat(messages:list,model:str="deepseek-chat",temperature:float=0.7)->str:"""封装 LLM 调用"""response=client.chat.completions.create(model=model,messages=messages,temperature=temperature,)returnresponse.choices[0].message.content# ── 工具定义（复用第3章的） ──defget_weather(city:str)->str:weather_db={"北京":"晴，25°C，轻度雾霾","上海":"多云，28°C，空气质量优","深圳":"阵雨，30°C，湿度大","杭州":"小雨，22°C，适合室内活动",}returnweather_db.get(city,f"未找到{city}天气")defweb_search(query:str)->str:"""模拟搜索（实际接搜索API）"""search_db={"北京景点":"故宫(5A)、长城(5A)、颐和园(5A)、798艺术区、南锣鼓巷","北京美食":"烤鸭(全聚德/大董)、炸酱面、豆汁、卤煮、涮羊肉","北京交通":"地铁覆盖主城区，建议办交通卡；高峰期(7-9点,17-19点)打车困难",}forkey,valueinsearch_db.items():ifkeyinquery:returnvaluereturnf"搜索「{query}」未找到相关信息"TOOLS={"get_weather":get_weather,"web_search":web_search,}# ── 构建 System Prompt ──REACT_SYSTEM_PROMPT="""你是一个能使用工具的 AI Agent。你必须严格按照以下格式回复： 当你需要调用工具时： Thought: 分析当前情况，解释为什么需要调用这个工具 Action: 工具名称 Action Input: {"参数名": "参数值"} 当工具执行完后，你会看到： Observation: 工具返回的结果 然后你应该再次思考，直到得到最终答案： Thought: 我现在有了足够的信息 Final Answer: 给用户的完整回答 重要规则： 1. 每次只调用一个工具 2. Thought 必须解释你的推理过程 3. Action Input 必须是合法的 JSON 4. 信息足够时，直接给 Final Answer 5. 工具失败时，思考替代方案 可用工具： - get_weather: 查询城市天气，参数 city (字符串) - web_search: 搜索网络信息，参数 query (字符串) """defparse_action(text:str)->tuple[str,dict]|None:"""从 LLM 输出中解析 Action 和 Action Input"""action_match=re.search(r"Action:\s*(\w+)",text)input_match=re.search(r"Action Input:\s*(\{.*?\})",text,re.DOTALL)ifaction_matchandinput_match:try:action=action_match.group(1)action_input=json.loads(input_match.group(1))returnaction,action_inputexceptjson.JSONDecodeError:passreturnNonedefparse_final_answer(text:str)->str|None:"""从 LLM 输出中解析 Final Answer"""match=re.search(r"Final Answer:\s*(.*)",text,re.DOTALL)ifmatch:returnmatch.group(1).strip()returnNone# ── ReAct Agent 主循环 ──defreact_agent(task:str,max_steps:int=8,verbose:bool=True)->str:""" ReAct Agent：先思考，再行动，观察结果，继续思考... """messages=[{"role":"system","content":REACT_SYSTEM_PROMPT},{"role":"user","content":task}]forstepinrange(max_steps):ifverbose:print(f"\n{'='*60}")print(f"🟢 第{step+1}步")print(f"{'='*60}")# 调用 LLMresponse=chat(messages,temperature=0.3)# 低温度保证格式稳定messages.append({"role":"assistant","content":response})ifverbose:print(f"💭 LLM 输出:\n{response}")# 检查是否是最终回答final_answer=parse_final_answer(response)iffinal_answer:ifverbose:print(f"\n✅ 任务完成！共{step+1}步")returnfinal_answer# 检查是否需要执行工具action_data=parse_action(response)ifaction_data:action_name,action_input=action_dataifaction_nameinTOOLS:# 执行工具func=TOOLS[action_name]result=func(**action_input)ifverbose:print(f"🔧 执行:{action_name}({action_input})")print(f"👀 结果:{result}")# 把结果作为 Observation 加入messages.append({"role":"user","content":f"Observation:{result}"})else:messages.append({"role":"user","content":f"Observation: 错误 - 未知工具 '{action_name}'"})else:# 既不是 Action 也不是 Final Answer，提示模型遵循格式messages.append({"role":"user","content":"请按照格式回复：先 Thought，然后选择 Action 或 Final Answer。"})return"任务在最大步数内未完成，请简化需求。"# ── 测试 ──if__name__=="__main__":# 简单对比：同一个任务，看 ReAct Agent 的思考过程task="我想周末去北京玩两天，帮我查一下天气，推荐几个景点和美食，再给个出行建议"result=react_agent(task)print(f"\n{'='*60}")print(f"📝 最终回答:\n{result}")运行效果
🟢 第 1 步 💭 LLM 输出: Thought: 用户要规划北京两日游，我需要先知道天气来确定户外还是室内活动 Action: get_weather Action Input: {"city": "北京"} 🔧 执行: get_weather({'city': '北京'}) 👀 结果: 晴，25°C，轻度雾霾 🟢 第 2 步 💭 LLM 输出: Thought: 天气适合户外(晴天25°C)，但有雾霾需要注意。现在查景点和美食 Action: web_search Action Input: {"query": "北京景点"} 🔧 执行: web_search({'query': '北京景点'}) 👀 结果: 故宫(5A)、长城(5A)、颐和园(5A)、798艺术区、南锣鼓巷 🟢 第 3 步 💭 LLM 输出: Thought: 有了景点列表，现在查美食 Action: web_search Action Input: {"query": "北京美食"} 🔧 执行: web_search({'query': '北京美食'}) 👀 结果: 烤鸭(全聚德/大董)、炸酱面、豆汁、卤煮、涮羊肉 🟢 第 4 步 💭 LLM 输出: Thought: 天气、景点、美食都有了。现在可以给出完整的规划 Final Answer: 北京周末天气晴好，25°C，但有轻度雾霾建议戴口罩... ✅ 任务完成！共 4 步5.3 Chain-of-Thought：让 Agent 先想再做的简单方法
如果你不想用 ReAct 这种严格的格式，一个更轻量的方法是思维链提示（Chain-of-Thought）：
""" 思维链 Agent —— 将本节代码追加到5.2节完整代码后面即可运行 依赖5.2节定义的 client 和 chat 函数 """defagent_with_cot(task:str)->str:"""使用思维链提示的 Agent — 更简单，但效果也不错"""system_prompt="""你是一个会仔细思考的助手。回答每个问题之前，先： 1. 分析问题：用户真正想要什么？ 2. 列出步骤：需要做哪些事？ 3. 逐步执行：一步一步来 4. 总结回答：给用户清晰的结论 在 {{思考}} 中写下你的分析过程（这部分不会给用户看到）， 然后用自然语言回复用户。"""response=chat([{"role":"system","content":system_prompt},{"role":"user","content":task}])returnresponse这个方法不需要解析 Action/Action Input，但对于简单任务效果很好。
5.4 任务分解：把大任务拆小
复杂任务一次性交给 Agent，它容易"迷路"。更好的做法是先分解，再逐个击破。
""" 任务分解 —— 将本节代码追加到5.2节完整代码后面即可运行 依赖5.2节定义的 client、chat、react_agent 函数 """defdecompose_task(complex_task:str)->list[str]:"""用 LLM 把复杂任务分解为子任务列表"""prompt=f"""请把以下复杂任务分解成3-5个简单的子任务，每个子任务一句话。 用序号列出，不要说其他内容。 复杂任务:{complex_task}子任务列表:"""result=chat([{"role":"system","content":"你是任务分解专家，只输出子任务列表。"},{"role":"user","content":prompt}])# 解析子任务subtasks=[]forlineinresult.strip().split("\n"):line=line.strip()iflineand(line[0].isdigit()orline.startswith("-")):# 去掉序号和符号task=line.lstrip("0123456789.-) ").strip()iftask:subtasks.append(task)returnsubtasksdefagent_with_decomposition(complex_task:str)->str:""" 先分解任务，再逐个执行，最后汇总 """# 第1步：分解任务print("📋 正在分解任务...")subtasks=decompose_task(complex_task)print(f"分解为{len(subtasks)}个子任务:")fori,tinenumerate(subtasks,1):print(f"{i}.{t}")# 第2步：逐个执行results=[]fori,subtaskinenumerate(subtasks,1):print(f"\n🔨 执行子任务{i}/{len(subtasks)}:{subtask}")result=react_agent(subtask,max_steps=3,verbose=False)results.append({"task":subtask,"result":result})print(f" 结果:{result[:100]}...")# 第3步：汇总print("\n📊 正在汇总结果...")summary_prompt=f"""请基于以下子任务的结果，给出一份完整的回答： 原始任务:{complex_task}子任务结果:{json.dumps(results,ensure_ascii=False,indent=2)}请给出完整、连贯、有用的回答："""final_answer=chat([{"role":"system","content":"你是信息整合专家"},{"role":"user","content":summary_prompt}])returnfinal_answer5.5 自我反思：Agent 检查自己的输出
好的 Agent 应该能自己检查自己。这是一个简单的反思模式：
""" 自我反思 Agent —— 将本节代码追加到5.2节完整代码后面即可运行 依赖5.2节定义的 client、chat、react_agent 函数 """defagent_with_reflection(task:str)->str:"""Agent 先回答问题 → 再反思自己的回答 → 改进后输出"""# 第1轮：生成初始回答print("✍️ 生成初始回答...")initial_answer=react_agent(task,verbose=False)# 第2轮：自我反思print("🔍 自我反思中...")reflection_prompt=f"""请检查以下回答的质量，指出问题： 原始问题:{task}待检查的回答:{initial_answer}请检查： 1. 信息是否准确完整？ 2. 逻辑是否清晰？ 3. 有没有遗漏用户的需求？ 4. 有没有更好的建议？ 如果回答已经很好，说"无需改进"。 如果有问题，具体指出。"""reflection=chat([{"role":"system","content":"你是严格的质量审核员"},{"role":"user","content":reflection_prompt}])print(f"反思结果:{reflection[:200]}...")# 第3轮：改进if"无需改进"notinreflection:print("🔧 改进回答中...")improve_prompt=f"""请基于以下反馈改进你的回答： 原始回答:{initial_answer}改进建议:{reflection}请给出改进后的完整回答："""improved_answer=chat([{"role":"system","content":"你追求卓越，善于接受反馈"},{"role":"user","content":improve_prompt}])returnimproved_answerreturninitial_answer5.6 三种 Planning 模式对比
| 模式 | 复杂度 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|---|
| Tool Calling | ⭐ | 简单查询、单步骤任务 | 快速、可靠 | 没有显式推理 |
| ReAct | ⭐⭐ | 多步骤推理任务 | 思考过程可见、易调试 | 格式可能不稳定 |
| 分解+汇总 | ⭐⭐⭐ | 复杂、开放式任务 | 复杂任务结构化 | 需要多次调用、成本高 |
| 反思改进 | ⭐⭐⭐ | 需要高质量输出的场景 | 输出质量更高 | 多一次调用 |
📝 本章小结
- ReAct= 让 Agent 在每一步先说 Thought（思考），再做 Action（行动）
- 思维链= 要求 LLM “让我们一步一步思考”
- 任务分解= 大任务拆小，逐个击破，最后汇总
- 自我反思= Agent 检查自己的输出，发现不足并改进
✏️ 练习题
基础题：给 ReAct Agent 增加一个
save_note(content)工具，让它能把用户的行程规划保存下来。进阶题：改造 React Agent，让它支持一次输出多个 Action（并行工具调用），提高效率。
挑战题：实现一个"Plan-and-Execute" Agent——先用 LLM 制定完整计划（不执行），用户确认后再逐步执行。
下一章预告：第6章：Multi-Agent — 多个 Agent 协作 —— 一个 Agent 不够？我们来组建一个 Agent 团队！
