Agent Loop 让 Agent 能自主执行。但"自主"不等于"乱来"。今天，我们让 Agent 先想清楚再动手。

---

今天的目标

实现 Plan-and-Execute 模式：

1. 规划阶段：AI 先制定完整的执行计划
2. 确认阶段：用户确认计划是否合理
3. 执行阶段：按计划逐步执行
4. 重规划：遇到问题时，AI 可以调整计划

---

为什么需要"先规划"

Day 5 的 Agent Loop 是"走一步看一步"：

思考 → 行动 → 思考 → 行动 → 思考 → 行动 ...

这种方式在很多场景下很好，但有些任务需要"先想清楚"：

• 复杂任务：用户说"帮我写一份调研报告"，Agent 应该先列出调研的几个方面，而不是想到什么搜什么
• 多步骤任务：步骤之间有依赖关系，顺序搞错了会出问题
• 资源消耗：每一步都要调用 AI，调用多了费用也多。先规划可以减少不必要的调用

---

Plan-and-Execute 的架构

用户输入
    ↓
┌─────────────┐
│   规划器     │ → 生成执行计划
│  (Planner)  │
└──────┬──────┘
       ↓
┌─────────────┐
│   用户确认   │ → 用户检查计划是否合理
└──────┬──────┘
       ↓
┌─────────────┐
│   执行器     │ → 逐步执行计划
│  (Executor) │
└──────┬──────┘
       ↓
┌─────────────┐
│  重规划？    │ → 如果执行失败，调整计划
│(Re-Planner) │
└──────┬──────┘
       ↓
    最终结果

---

完整代码

工具实现参考 Day 2。这里只展示规划器和执行器。

# day6_planner.py

import json
import inspect
from openai import OpenAI

client = OpenAI(
    api_key="你的API Key",
    base_url="https://api.mimo.ai/v1",
)


# ========== 规划器 ==========

def create_plan(user_input: str) -> dict:
    """AI 生成执行计划"""
    tools_prompt = build_tools_prompt()  # 从 Day 2 导入

    system_prompt = f"""你是一个任务规划专家。根据用户需求，制定详细的执行计划。

{tools_prompt}

规划要求：
1. 将复杂任务分解为具体的步骤
2. 每个步骤只调用一个工具
3. 步骤之间有明确的逻辑顺序
4. 考虑步骤之间的依赖关系

输出 JSON 格式：
{{
    "goal": "任务目标的一句话描述",
    "steps": [
        {{
            "id": 1,
            "description": "步骤描述",
            "tool": "工具名",
            "params": {{"参数名": "参数值"}},
            "depends_on": []
        }}
    ],
    "output": "最终输出的形式描述"
}}

注意：
- depends_on 是这个步骤依赖的步骤 ID 列表
- 如果没有依赖，写空列表 []
- 只输出 JSON"""

    response = client.chat.completions.create(
        model="mimo-v2-flash",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input},
        ],
        temperature=0,
    )

    reply = response.choices[0].message.content.strip()
    try:
        if reply.startswith("```"):
            reply = reply.split("\n", 1)[1]
            reply = reply.rsplit("```", 1)[0]
        return json.loads(reply)
    except json.JSONDecodeError:
        return {
            "goal": user_input,
            "steps": [{"id": 1, "description": "执行任务", "tool": "chat", "params": {}, "depends_on": []}],
            "output": "文字回复",
        }


def display_plan(plan: dict):
    """展示计划"""
    print(f"\n{'='*50}")
    print(f"执行计划")
    print(f"{'='*50}")
    print(f"目标：{plan.get('goal', '未知')}")
    print(f"共 {len(plan.get('steps', []))} 步：")
    print()

    for step in plan.get("steps", []):
        step_id = step.get("id", "?")
        desc = step.get("description", "")
        tool = step.get("tool", "")
        deps = step.get("depends_on", [])

        deps_str = f"（依赖步骤 {', '.join(map(str, deps))}）" if deps else ""
        print(f"  步骤 {step_id}：{desc}")
        print(f"    工具：{tool}{deps_str}")

    print(f"\n输出形式：{plan.get('output', '未知')}")
    print(f"{'='*50}")


# ========== 执行器 ==========

def execute_plan(plan: dict):
    """执行计划"""
    steps = plan.get("steps", [])
    results = {}  # step_id -> result
    completed = set()

    print(f"\n{'='*50}")
    print(f"开始执行")
    print(f"{'='*50}")

    max_iterations = len(steps) * 2  # 安全阀
    iteration = 0

    while len(completed) < len(steps) and iteration < max_iterations:
        iteration += 1

        # 找到下一步可以执行的步骤
        next_step = None
        for step in steps:
            step_id = step.get("id")
            if step_id in completed:
                continue

            # 检查依赖是否都满足
            deps = step.get("depends_on", [])
            if all(d in completed for d in deps):
                next_step = step
                break

        if next_step is None:
            print("\n[错误]: 无法找到下一步（可能存在循环依赖）")
            break

        step_id = next_step.get("id")
        desc = next_step.get("description", "")
        tool_name = next_step.get("tool", "")
        params = next_step.get("params", {})

        print(f"\n[步骤 {step_id}] {desc}")

        # 替换参数中的占位符
        for key, value in params.items():
            if isinstance(value, str):
                for dep_id in next_step.get("depends_on", []):
                    placeholder = f"{{step_{dep_id}_result}}"
                    if placeholder in value:
                        params[key] = value.replace(placeholder, results.get(dep_id, ""))

        # 执行工具
        if tool_name and tool_name != "chat":
            print(f"[调用工具]: {tool_name}")
            result = execute_tool(tool_name, params)  # 从 Day 2 导入
        else:
            result = params.get("response", "步骤完成")

        results[step_id] = result
        completed.add(step_id)

        print(f"[结果]: {result[:150]}{'...' if len(result) > 150 else ''}")

    # 生成最终结果
    print(f"\n{'='*50}")
    print(f"执行完成")
    print(f"{'='*50}")

    if results:
        summary = summarize_results(plan, results)
        print(f"\nAgent：{summary}")
    else:
        print("\nAgent：计划执行完毕，但没有产生任何结果。")


def summarize_results(plan: dict, results: dict) -> str:
    """让 AI 根据执行结果生成总结"""
    results_text = "\n".join(
        f"步骤 {step_id}: {result[:300]}"
        for step_id, result in results.items()
    )

    response = client.chat.completions.create(
        model="mimo-v2-flash",
        messages=[
            {"role": "system", "content": "根据任务计划和执行结果，用中文简洁地总结最终输出。"},
            {"role": "user", "content": f"任务目标：{plan.get('goal', '')}\n\n执行结果：\n{results_text}"},
        ],
        temperature=0,
    )
    return response.choices[0].message.content.strip()

---

运行测试

你：帮我做一个简单的项目统计报告

正在制定计划...

==================================================
执行计划
==================================================
目标：创建一个简单的项目统计报告

共 4 步：

  步骤 1：列出当前目录的所有文件
    工具：list_files
  步骤 2：筛选出 Python 文件
    工具：chat（AI 处理）
    依赖步骤 1
  步骤 3：统计每个 Python 文件的行数
    工具：count_lines
    依赖步骤 2
  步骤 4：将统计结果写入报告文件
    工具：write_file
    依赖步骤 3

输出形式：一份包含文件统计信息的报告文件
==================================================

确认执行？(y/n/修改): y

==================================================
开始执行
==================================================

[步骤 1] 列出当前目录的所有文件
[调用工具]: list_files
[结果]: 📄 day1_basic.py (1234 bytes)
📄 day2_tools.py (2345 bytes)
📄 notes.txt (100 bytes)

[步骤 2] 筛选出 Python 文件
[结果]: Python 文件有：day1_basic.py, day2_tools.py

[步骤 3] 统计每个 Python 文件的行数
[调用工具]: count_lines
[结果]: 文件 day1_basic.py：共 50 行，其中代码 45 行，空行 5 行

[步骤 4] 将统计结果写入报告文件
[调用工具]: write_file
[结果]: 成功写入文件 'report.txt'

==================================================
执行完成
==================================================

Agent：项目统计报告已生成并保存到 report.txt。报告显示：
- 当前目录有 2 个 Python 文件
- day1_basic.py：50 行（代码 45 行，空行 5 行）
- day2_tools.py：80 行（代码 72 行，空行 8 行）
- 总计：130 行（代码 117 行，空行 13 行）

---

核心概念

规划 vs 执行 分离

# 规划阶段
plan = create_plan(user_input)

# 执行阶段
execute_plan(plan)

规划和执行是两个独立的过程。这带来几个好处：

1. 用户可以审核计划：在执行前确认计划是否合理
2. 计划可以复用：保存计划，下次可以执行同样的任务
3. 错误更容易定位：是规划错了还是执行错了？

依赖管理

"depends_on": [1, 2]  # 这个步骤依赖步骤 1 和 2

执行器会检查依赖是否满足，确保步骤按正确顺序执行。

---

常见错误

错误 1：AI 规划的步骤之间没有依赖关系

# 错误：步骤 2 依赖步骤 1 的结果，但没有声明
"steps": [
    {"id": 1, "tool": "list_files", ...},
    {"id": 2, "tool": "count_lines", "params": {"file_path": "..."}, "depends_on": []},  # 缺少依赖！
]

# 正确：声明依赖关系
"depends_on": [1]  # 步骤 2 依赖步骤 1

错误 2：占位符格式不匹配

AI 返回的占位符有时格式不对：

# AI 返回的
"params": {"file_path": "{step_1_result}/test.py"}

# 但程序期望的是
"params": {"file_path": "{step_1_result}"}

解决方法：在 system_prompt 中明确说明占位符格式。

错误 3：没有用户确认就直接执行

# 错误：直接执行，用户无法审核
plan = create_plan(user_input)
execute_plan(plan)  # 用户看不到计划！

# 正确：展示计划，让用户确认
plan = create_plan(user_input)
display_plan(plan)
confirm = input("确认执行？(y/n): ")
if confirm == "y":
    execute_plan(plan)

---

三种执行模式的对比

模式	特点	适用场景
Day 3: 预规划	AI 在执行前规划所有步骤	步骤数固定、可预测
Day 5: Agent Loop	每一步都重新评估	复杂、不确定的任务
Day 6: Plan-and-Execute	先规划，用户确认，再执行	需要用户审核的重要任务

在实际应用中，这三种模式往往混合使用：

• 大方向用 Plan-and-Execute
• 每个步骤内部用 Agent Loop
• 遇到意外时触发 Re-Planning

---

动手实验

1. 把执行过程中故意制造一个错误，观察 Agent 的反应
2. 加一个"修改计划"功能，让用户可以手动调整步骤
3. 保存计划到文件，下次可以直接执行

---

明天预告

今天是系列的最后一天。我们将所有组件整合成一个完整的项目，并展望 Agent 技术的未来发展方向。

→ Day 7：完整项目 + 下一步去哪里