第五章 Planning — 让 Agent 学会“思考“

慕鱼而网

1569人浏览 · 2026-06-11 09:41:16

慕鱼而网 · 2026-06-11 09:41:16 发布

第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-你的密钥
"""
import json
import re
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# DeepSeek 客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"
)

def chat(messages: list, model: str = "deepseek-chat", temperature: float = 0.7) -> str:
    """封装 LLM 调用"""
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
    )
    return response.choices[0].message.content

# ── 工具定义（复用第3章的） ──
def get_weather(city: str) -> str:
    weather_db = {
        "北京": "晴，25°C，轻度雾霾",
        "上海": "多云，28°C，空气质量优",
        "深圳": "阵雨，30°C，湿度大",
        "杭州": "小雨，22°C，适合室内活动",
    }
    return weather_db.get(city, f"未找到{city}天气")

def web_search(query: str) -> str:
    """模拟搜索（实际接搜索API）"""
    search_db = {
        "北京景点": "故宫(5A)、长城(5A)、颐和园(5A)、798艺术区、南锣鼓巷",
        "北京美食": "烤鸭(全聚德/大董)、炸酱面、豆汁、卤煮、涮羊肉",
        "北京交通": "地铁覆盖主城区，建议办交通卡；高峰期(7-9点,17-19点)打车困难",
    }
    for key, value in search_db.items():
        if key in query:
            return value
    return f"搜索「{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 (字符串)
"""

def parse_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)
    
    if action_match and input_match:
        try:
            action = action_match.group(1)
            action_input = json.loads(input_match.group(1))
            return action, action_input
        except json.JSONDecodeError:
            pass
    return None

def parse_final_answer(text: str) -> str | None:
    """从 LLM 输出中解析 Final Answer"""
    match = re.search(r"Final Answer:\s*(.*)", text, re.DOTALL)
    if match:
        return match.group(1).strip()
    return None

# ── ReAct Agent 主循环 ──
def react_agent(task: str, max_steps: int = 8, verbose: bool = True) -> str:
    """
    ReAct Agent：先思考，再行动，观察结果，继续思考...
    """
    messages = [
        {"role": "system", "content": REACT_SYSTEM_PROMPT},
        {"role": "user", "content": task}
    ]
    
    for step in range(max_steps):
        if verbose:
            print(f"\n{'='*60}")
            print(f"🟢 第 {step+1} 步")
            print(f"{'='*60}")
        
        # 调用 LLM
        response = chat(messages, temperature=0.3)  # 低温度保证格式稳定
        messages.append({"role": "assistant", "content": response})
        
        if verbose:
            print(f"💭 LLM 输出:\n{response}")
        
        # 检查是否是最终回答
        final_answer = parse_final_answer(response)
        if final_answer:
            if verbose:
                print(f"\n✅ 任务完成！共 {step+1} 步")
            return final_answer
        
        # 检查是否需要执行工具
        action_data = parse_action(response)
        if action_data:
            action_name, action_input = action_data
            
            if action_name in TOOLS:
                # 执行工具
                func = TOOLS[action_name]
                result = func(**action_input)
                
                if verbose:
                    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 函数
"""
def agent_with_cot(task: str) -> str:
    """使用思维链提示的 Agent — 更简单，但效果也不错"""
    
    system_prompt = """你是一个会仔细思考的助手。回答每个问题之前，先：
1. 分析问题：用户真正想要什么？
2. 列出步骤：需要做哪些事？
3. 逐步执行：一步一步来
4. 总结回答：给用户清晰的结论

在 {{思考}} 中写下你的分析过程（这部分不会给用户看到），
然后用自然语言回复用户。"""

    response = chat([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": task}
    ])
    return response

这个方法不需要解析 Action/Action Input，但对于简单任务效果很好。

5.4 任务分解：把大任务拆小

复杂任务一次性交给 Agent，它容易"迷路"。更好的做法是先分解，再逐个击破。

"""
任务分解 —— 将本节代码追加到5.2节完整代码后面即可运行
依赖5.2节定义的 client、chat、react_agent 函数
"""
def decompose_task(complex_task: str) -> list[str]:
    """用 LLM 把复杂任务分解为子任务列表"""
    
    prompt = f"""请把以下复杂任务分解成3-5个简单的子任务，每个子任务一句话。
用序号列出，不要说其他内容。

复杂任务: {complex_task}

子任务列表:"""

    result = chat([
        {"role": "system", "content": "你是任务分解专家，只输出子任务列表。"},
        {"role": "user", "content": prompt}
    ])
    
    # 解析子任务
    subtasks = []
    for line in result.strip().split("\n"):
        line = line.strip()
        if line and (line[0].isdigit() or line.startswith("-")):
            # 去掉序号和符号
            task = line.lstrip("0123456789.-) ").strip()
            if task:
                subtasks.append(task)
    
    return subtasks


def agent_with_decomposition(complex_task: str) -> str:
    """
    先分解任务，再逐个执行，最后汇总
    """
    # 第1步：分解任务
    print("📋 正在分解任务...")
    subtasks = decompose_task(complex_task)
    print(f"分解为 {len(subtasks)} 个子任务:")
    for i, t in enumerate(subtasks, 1):
        print(f"  {i}. {t}")
    
    # 第2步：逐个执行
    results = []
    for i, subtask in enumerate(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}
    ])
    
    return final_answer

5.5 自我反思：Agent 检查自己的输出

好的 Agent 应该能自己检查自己。这是一个简单的反思模式：

"""
自我反思 Agent —— 将本节代码追加到5.2节完整代码后面即可运行
依赖5.2节定义的 client、chat、react_agent 函数
"""
def agent_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 "无需改进" not in reflection:
        print("🔧 改进回答中...")
        improve_prompt = f"""请基于以下反馈改进你的回答：

原始回答: {initial_answer}
改进建议: {reflection}

请给出改进后的完整回答："""
        
        improved_answer = chat([
            {"role": "system", "content": "你追求卓越，善于接受反馈"},
            {"role": "user", "content": improve_prompt}
        ])
        return improved_answer
    
    return initial_answer

5.6 三种 Planning 模式对比

模式	复杂度	适用场景	优点	缺点
Tool Calling	⭐	简单查询、单步骤任务	快速、可靠	没有显式推理
ReAct	⭐⭐	多步骤推理任务	思考过程可见、易调试	格式可能不稳定
分解+汇总	⭐⭐⭐	复杂、开放式任务	复杂任务结构化	需要多次调用、成本高
反思改进	⭐⭐⭐	需要高质量输出的场景	输出质量更高	多一次调用