链式思考（Chain-of-Thought, CoT）是让 AI 在回答问题前先"说出自己的思考过程"——就像学生做数学题要写"解"的过程。对 Agent 来说，这意味着在执行前先列出步骤。
❌ 没有规划：
用户: "帮我做一个市场调研报告"
Agent: 直接开始写 → 写到一半发现缺数据 → 乱写 → 质量差
✅ 有规划：
用户: "帮我做一个市场调研报告"
Agent 思考:
步骤1: 明确调研范围和目标
步骤2: 搜索行业数据和竞品信息
步骤3: 分析数据，提取关键洞察
步骤4: 组织报告结构（概述→市场分析→竞品→趋势→建议）
步骤5: 逐段撰写
步骤6: 检查完整性和一致性创建 planning_agent.py：
#!/usr/bin/env python3
"""
PlanningAgent — 具备任务规划能力的 AI Agent
=============================================
核心能力：
✅ 复杂任务自动拆分
✅ 分步骤有序执行
✅ 每步结果验证
✅ 异常步骤自动调整
✅ 最终结果汇总
运行：python3 planning_agent.py
"""
import json
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any
from dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
# ================================================================
# 第1部分：任务规划器
# ================================================================
class TaskPlanner:
"""任务规划器 — 负责把复杂任务拆成可执行的步骤"""
def __init__(self, client: Anthropic, model: str = "claude-sonnet-4-6"):
self.client = client
self.model = model
def decompose(self, task: str, context: dict = None) -> list[dict]:
"""
将复杂任务拆解为多个子步骤
返回格式: [{"step": 1, "action": "做什么", "tool": "用哪个工具(可选)",
"params": {}, "expected": "预期结果"}]
"""
context_text = f"\n当前上下文: {json.dumps(context, ensure_ascii=False)}" if context else ""
prompt = f"""你是一个任务规划专家。请将以下用户任务拆解为可执行的子步骤。
用户任务: {task}{context_text}
请以 JSON 数组格式返回步骤列表，每个步骤包含：
- step: 步骤编号
- action: 具体要做什么（中文描述）
- tool: 需要调用的工具名（可选，如 calculator/file_reader/web_search）
- params: 工具参数（如不需要工具则为空对象）
- expected: 完成后预期的结果描述
可用工具：calculator(expression), file_reader(filepath), web_search(query),
text_summarizer(text), time_getter()
规则：
1. 步骤要具体、可执行，不能模糊
2. 步骤之间逻辑连贯，前一步的输出是后一步的输入
3. 如果任务很简单（1步就能完成），也返回单元素数组
4. 每个步骤的结果需要能够验证是否成功
只返回 JSON 数组，不要加其他文字。"""
try:
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
plan_text = response.content[0].text.strip()
# 提取 JSON
if "```" in plan_text:
plan_text = plan_text.split("```")[1].split("```")[0]
if plan_text.startswith("```json"):
plan_text = plan_text[7:]
plan = json.loads(plan_text)
return plan if isinstance(plan, list) else [plan]
except json.JSONDecodeError as e:
print(f"⚠️ 规划 JSON 解析失败: {e}")
print(f"原始返回: {plan_text[:200]}")
# 降级：当作单步任务
return [{"step": 1, "action": task, "tool": None, "params": {}, "expected": "完成任务"}]
def adjust_plan(self, original_plan: list[dict], failed_step: int, error_info: str) -> list[dict]:
"""当某步骤失败时，调整后续计划"""
prompt = f"""原计划执行到第{failed_step}步时失败。
失败信息: {error_info}
原计划:
{json.dumps(original_plan, ensure_ascii=False, indent=2)}
请调整第{failed_step}步及之后的步骤，给出修正后的完整计划。
返回格式与原计划相同的 JSON 数组。"""
try:
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
text = response.content[0].text.strip()
if "```" in text:
text = text.split("```")[1].split("```")[0]
if text.startswith("json"):
text = text[4:]
new_plan = json.loads(text)
return new_plan if isinstance(new_plan, list) else original_plan
except Exception:
return original_plan # 调整失败，返回原计划
# ================================================================
# 第2部分：工具集
# ================================================================
class ToolExecutor:
"""工具执行器 — 执行规划步骤中指定的工具"""
def __init__(self):
self.execution_log: list[dict] = []
def execute(self, tool_name: str, params: dict) -> str:
"""执行指定工具"""
tools = {
"calculator": self._calc,
"file_reader": self._read_file,
"web_search": self._search,
"text_summarizer": self._summarize,
"time_getter": self._get_time,
}
func = tools.get(tool_name)
if not func:
return f"❌ 未知工具: {tool_name}"
try:
first_arg = list(params.values())[0] if params else ""
result = func(first_arg)
self.execution_log.append({
"tool": tool_name,
"params": params,
"result": result[:200],
"time": datetime.now().isoformat(),
})
return result
except Exception as e:
return f"❌ 执行失败: {e}"
def _calc(self, expr: str) -> str:
allowed = set("0123456789+-*/().% ")
if not all(c in allowed for c in expr):
return f"❌ 不支持的表达式: {expr}"
try:
return str(eval(expr, {"__builtins__": {}}, {}))
except Exception as e:
return f"❌ {e}"
def _read_file(self, path: str) -> str:
p = Path(path)
if not p.exists():
return f"❌ 文件不存在: {path}"
try:
content = p.read_text("utf-8")
return content[:2000]
except Exception as e:
return f"❌ {e}"
def _search(self, query: str) -> str:
return f"[搜索结果] 关于 '{query}' 的模拟搜索结果：这是搜索功能的占位实现。"
def _summarize(self, text: str) -> str:
return f"[摘要] 文本长度 {len(text)} 字符，主要内容概括..."
def _get_time(self, _=None) -> str:
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# ================================================================
# 第3部分：规划执行 Agent
# ================================================================
class PlanningAgent:
"""具备任务规划与执行能力的 Agent"""
def __init__(self, name: str = "PlanningAgent"):
self.name = name
self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.model = "claude-sonnet-4-6"
self.planner = TaskPlanner(self.client, self.model)
self.executor = ToolExecutor()
self.max_retries_per_step = 2
self.memory: list[dict] = []
def run(self, user_input: str) -> str:
"""主运行流程：规划 → 执行 → 汇总"""
print(f"\n{'='*60}")
print(f"👤 用户任务: {user_input}")
print(f"{'='*60}")
# 阶段1：任务规划
print("\n📋 [阶段1] 任务规划中...")
plan = self.planner.decompose(user_input)
print(f"📋 将任务拆分为 {len(plan)} 个步骤:")
for step in plan:
tool_info = f" [工具: {step['tool']}]" if step.get('tool') else ""
print(f" 步骤{step['step']}: {step['action']}{tool_info}")
# 阶段2：逐步执行
print(f"\n⚡ [阶段2] 开始执行 ({len(plan)} 步)...")
step_results = []
current_plan = plan.copy()
for i, step in enumerate(current_plan):
step_num = step["step"]
print(f"\n 📍 执行步骤{step_num}/{len(current_plan)}: {step['action']}")
# 执行步骤（带重试）
result = self._execute_step(step, i)
if result.startswith("❌"):
print(f" ⚠️ 步骤{step_num}失败: {result}")
# 尝试调整计划
print(f" 🔄 尝试调整计划...")
current_plan = self.planner.adjust_plan(current_plan, step_num, result)
# 重试当前步骤
if i < len(current_plan):
result = self._execute_step(current_plan[i], i)
if result.startswith("❌"):
print(f" ❌ 调整后仍然失败，跳过此步骤")
step_results.append(f"[步骤{step_num}失败: {result}]")
continue
step_results.append(result)
print(f" ✅ 步骤{step_num}完成: {result[:100]}")
# 阶段3：结果汇总
print(f"\n📊 [阶段3] 结果汇总...")
summary = self._summarize_results(user_input, plan, step_results)
# 保存到记忆
self.memory.append({
"task": user_input,
"plan": plan,
"results": step_results,
"summary": summary,
"time": datetime.now().isoformat(),
})
print(f"\n{'='*60}")
print(f"🤖 最终结果:\n{summary}")
print(f"{'='*60}")
return summary
def _execute_step(self, step: dict, index: int) -> str:
"""执行单个步骤，带重试"""
for attempt in range(self.max_retries_per_step + 1):
if step.get("tool"):
# 有工具，调用工具执行
result = self.executor.execute(step["tool"], step.get("params", {}))
else:
# 无工具，让 Claude 直接完成这一步
result = self._llm_step(step)
if not result.startswith("❌"):
return result
if attempt < self.max_retries_per_step:
print(f" 🔄 重试 {attempt+1}/{self.max_retries_per_step}...")
return result
def _llm_step(self, step: dict) -> str:
"""让 Claude 直接完成无工具的步骤"""
try:
response = self.client.messages.create(
model=self.model,
max_tokens=512,
messages=[{"role": "user", "content":
f"请完成以下步骤（只输出执行结果，不要解释）：\n{step['action']}\n预期结果: {step['expected']}"}],
temperature=0.3,
)
return response.content[0].text.strip()
except Exception as e:
return f"❌ {e}"
def _summarize_results(self, task: str, plan: list, results: list) -> str:
"""汇总所有步骤的结果"""
try:
steps_text = "\n".join([
f"步骤{s['step']}: {s['action']} → {r[:150]}"
for s, r in zip(plan, results)
])
response = self.client.messages.create(
model=self.model,
max_tokens=800,
messages=[{"role": "user", "content":
f"请将以下任务执行结果汇总为清晰的中文报告：\n\n"
f"任务: {task}\n\n执行步骤与结果:\n{steps_text}\n\n"
f"格式要求：先概述完成情况，再逐条列出关键结果。"}],
temperature=0.3,
)
return response.content[0].text.strip()
except Exception as e:
return "\n".join([f"{i+1}. {r}" for i, r in enumerate(results)])
# ================================================================
# 第4部分：交互入口
# ================================================================
def main():
print("=" * 60)
print("📋 PlanningAgent — 任务规划与执行 Agent")
print("=" * 60)
print("支持复杂任务，如：")
print(" • 帮我计算 (100+50)×0.8，然后把结果保存到result.txt")
print(" • 读取test.txt的内容，总结要点，搜索相关资料")
print(" • 先看现在几点，然后计算距离明天早上9点还有多少小时")
print(" • quit → 退出")
print("=" * 60)
if not os.getenv("ANTHROPIC_API_KEY"):
print("❌ 请先配置 ANTHROPIC_API_KEY！")
return
# 准备测试文件
Path("test.txt").write_text(
"AI Agent 是一种能够自主感知环境、制定计划、使用工具、"
"执行任务的智能程序。它不同于传统的聊天机器人，"
"能够独立思考和行动直到任务完成。\n",
encoding="utf-8"
)
agent = PlanningAgent()
while True:
try:
user_input = input("\n👤 任务: ").strip()
if not user_input:
continue
if user_input.lower() in ("quit", "exit", "q"):
break
agent.run(user_input)
except KeyboardInterrupt:
break
print(f"\n📊 本次会话完成 {len(agent.memory)} 个任务")
print(f"👋 {agent.name} 已退出。")
if __name__ == "__main__":
main()
| 问题 | 表现 | 优化方案 |
|---|---|---|
| 规划过于粗略 | 步骤太大，执行时频繁失败 | 提示词中加入"每个步骤必须原子化（不可再分）"的约束 |
| 规划过于细节 | 产生大量无用微步骤，效率低 | 设置步骤数上限（如5-8步），超出则合并 |
| 步骤遗漏 | 最终结果缺少关键环节 | 在规划后增加"验证清单"检查，如"是否包含输入→处理→输出→验证" |
| 依赖关系错误 | 步骤B在步骤A之前执行，但B依赖A | 增加步骤间依赖声明，规划时让 AI 输出 DAG（有向无环图）结构 |
| 概念 | 一句话 |
|---|---|
| 链式思考 CoT | 让 AI 在执行前先说/写出思考过程，大幅提升复杂任务的准确性 |
| 任务拆解 | 把大任务分解为多个原子化的子步骤，每步有明确输入、输出、验证标准 |
| 迭代规划 | 执行中某步失败时，动态调整后续计划，而非全部重新开始 |
| 结果汇总 | 将所有步骤的结果整合为完整、连贯的最终输出 |