为什么你的AI Agent总是跑偏？问题可能出在"不会做计划"

还记得上一篇结尾提到的那个任务吗？“帮我调研一下今年新能源汽车电池技术的最新进展，写一份两千字的报告。”

如果我们拿第三篇写的那个 ReAct 智能体去跑了一遍这个任务，它第一步“思考”了一下，调用了一次搜索工具，查到了“固态电池”这个热词，然后它就来劲了，接下来三四轮全在深挖固态电池一个话题，查了固态电池的原理、查了固态电池的几家代表公司、又查了固态电池的量产时间，查得挺起劲，可你会发现，报告要求的是 “新能源电池技术的最新进展”，是个全景式的调研 ，而不是 “固态电池深度专题” 。

「它把一个应该均匀覆盖多个技术方向的任务，活活走成了“逮着一个话题一路走到黑”」

这不是模型笨，而是 ReAct 这套“走一步看一步”的机制，天生就有这个毛病： 它没有一个全局蓝图，每一步只根据“当前手头这点信息”决定下一步干嘛，很容易在某个局部越钻越深，却忘了任务原本的边界和全貌 。

这就是今天要解决的问题——复杂任务，得先有规划，不能上来就莽

本文看点

• • 01、装修房子的比喻
  • 02、四步实现规划智能体
  • 03、ReAct 还是 Plan-and-Execute

01、装修房子的故事：为什么有些活儿必须先想清楚

我拿装修房子打个比方，你一下就能明白 ReAct 和“规划”之间的本质区别。

ReAct 式装修

工人今天进场，看了眼客厅，“这墙皮有点旧，先铲了吧”，铲完墙皮往下一看，“哎呀这里线路好像不对，得先改电路”，改完电路又发现，“这个插座位置得先问问业主想怎么摆家具”……每一步的判断都合理，但整个流程磕磕绊绊，非常容易出现“这活儿本该先干、结果最后才想起来干”的返工情况：比如刷完墙才发现电路没走完，得把刚刷好的墙重新凿开。

有规划的装修

正经的装修公司开工前，会先出一份完整的施工计划：拆改 → 水电改造 → 泥瓦 → 木工 → 油漆 → 安装，每一步先后顺序清清楚楚，因为老师傅都知道“电路必须在刷墙之前走完”这种硬性依赖关系。有了这份计划，才不会出现“刷完墙又要凿开重新走线”这种低级返工。

当然，有计划不代表计划一成不变。装修过程中真发现了原计划没预料到的问题：比如凿开墙才发现里面有根承重梁位置和图纸不符，这时候确实需要 暂停，根据新情况调整后续计划 ，而不是硬着头皮按原计划蛮干。

「这就是我们今天要讲的第二个关键概念：动态重规划」

好了，比喻讲完，翻译成智能体的语言：

1、走一步看一步的 ReAct，适合“哪一步该干嘛，走到那一步才能知道”的探索型任务；

2、先规划再执行的 Plan-and-Execute，适合那种“步骤大致可以提前预判、有明确先后依赖”的复杂任务；

3、动态重规划，是在执行过程中，一旦发现新情况让原计划不再合理，就主动修改计划，而不是死守着一份过时的蓝图硬跑到底。

02、Plan-and-Execute：先出图纸，再照图施工

这套模式说白了就是把智能体拆成两个角色分工合作：

规划者（Planner）

接到任务后，不着急动手，先在“脑子里”把整个任务拆解成一份有序的子任务清单，就像装修公司出的施工图纸。

执行者（Executor）

拿到这份清单，一条一条地去执行每个子任务：每个子任务内部，其实完全可以复用我们第三篇写的那套 ReAct 循环（想—做—看），只不过这次它的“眼界”被限定在当前这一个子任务范围内，不会像之前那样漫无边际地跑偏。

执行完一个子任务后，别急着往下走，还要经过一道 复盘（Replan） 关卡：拿当前已经执行完的结果，问一句“基于目前查到的东西，原计划还合理吗？要不要调整后面的步骤？”如果合理就继续按计划走，如果不合理，就更新计划，再继续执行。

画成流程图就是：

制定初始计划 → 执行子任务1 → 复盘/调整计划 → 执行子任务2 → 复盘/调整计划 → …… → 所有子任务完成 → 汇总产出最终结果

03、动手实现一个会“先出图纸”的调研智能体

我们直接拿新能源电池调研这个任务练手，用 deepseek-v4-flash 实现。为了聚焦在“规划”这个核心逻辑上，工具部分我们用一个搜索引擎，填上API即可。

STEP 01写规划者，把任务拆解成子任务清单

import json
import re
import time
from typing import Any

import requests
from openai import OpenAI

# ---------------------------------------------------------------- 配置

client = OpenAI(
api_key= 'sk-xxxx',
base_url= "https://api.deepseek.com",
)

BOCHA_API_KEY = 'sk-xxxx'
BOCHA_URL = "https://api.bochaai.com/v1/web-search"

MODEL = "deepseek-v4-flash"

MAX_PLAN_STEPS = 8        # 计划长度硬上限，防止重规划无限膨胀
MAX_REACT_ROUNDS = 3      # 单个子任务的 ReAct 轮数
MAX_SEARCH_RESULTS = 5    # 每次搜索返回的网页条数


# ---------------------------------------------------------------- 基础设施

def chat(messages: list, *, tools=None, thinking=False, max_tokens=4096, retries=3):
"""统一的模型调用入口：带重试 + 思考模式开关。"""

kwargs: dict[str, Any] = {
"model": MODEL,
"messages": messages,
"max_tokens": max_tokens,
    }
if tools:
        kwargs["tools"] = tools
if not thinking:
# 规划 / 抽取类任务不需要思考模式，省钱提速
kwargs["extra_body"] = {"thinking": {"type": "disabled"}}

    last_err = None
    for attempt in range(retries):
try:
return client.chat.completions.create(**kwargs)
except Exception as e:
            last_err = e
            wait = 2 ** attempt
print(f"调用失败（第 {attempt + 1} 次）：{e}，{wait}s 后重试")
            time.sleep(wait)

raise RuntimeError(f"模型调用连续失败 {retries} 次") from last_err


def parse_json_array(text: str) -> list | None:
"""健壮地从模型输出里抠出 JSON 数组。

    模型即使被要求「只输出 JSON」，也常常包上 ```json 围栏或加一句废话。
    原代码直接 json.loads 会 100% 在这里翻车。
    """
if not text:
return None

# 1. 剥离 markdown 围栏
text = re.sub(r"^\s*```(?:json)?\s*|\s*```\s*$", "", text.strip())

# 2. 直接试
try:
        data = json.loads(text)
return data if isinstance(data, list) else None
    except json.JSONDecodeError:
pass

# 3. 正则抠出第一个 [...] 块
match = re.search(r"\[.*\]", text, re.DOTALL)
if match:
try:
            data = json.loads(match.group())
return data if isinstance(data, list) else None
        except json.JSONDecodeError:
pass

    return None


def normalize_plan(raw: list | None) -> list[str]:
"""把计划统一归一化成 list[str]。

    模型可能返回 ["调研三元锂"] 也可能返回 [{"subtask": "调研三元锂"}]，
    在入口处一次性抹平，下游就不用到处写 isinstance 了。
    """
if not raw:
return []

    plan: list[str] = []
for item in raw:
if isinstance(item, str):
            text = item.strip()
elif isinstance(item, dict):
            text = str(
                item.get("subtask")
or item.get("task")
or item.get("description")
or ""
).strip()
else:
            text = str(item).strip()

if text:
            plan.append(text)

return plan


# ---------------------------------------------------------------- 真实搜索

def web_search(query: str) -> str:
"""博查 Web Search API —— DeepSeek 官方的联网搜索供应方，国内直连。"""

if not query or not query.strip():
return "搜索失败：查询词为空。"

payload = {
"query": query,
"freshness": "oneYear",   # 调研类任务限定近一年，"noLimit" 覆盖更广
"summary": True,          # 返回长摘要而非一句话 snippet
"count": MAX_SEARCH_RESULTS,
    }
    headers = {
"Authorization": f"Bearer {BOCHA_API_KEY}",
"Content-Type": "application/json",
    }

try:
        resp = requests.post(BOCHA_URL, headers=headers, json=payload, timeout=30)
        resp.raise_for_status()
        data = resp.json()
except requests.RequestException as e:
return f"搜索失败：{e}"
except json.JSONDecodeError:
return "搜索失败：返回内容不是合法 JSON。"

pages = (
        data.get("data", {})
        .get("webPages", {})
        .get("value", [])
    )

if not pages:
return f"未搜索到与「{query}」相关的信息。"

lines = []
for idx, page in enumerate(pages, start=1):
        title = page.get("name", "")
        url = page.get("url", "")
# summary 是长摘要，snippet 是短摘要，优先长的
content = (page.get("summary") or page.get("snippet") or "").strip()
        date = page.get("datePublished") or page.get("dateLastCrawled") or ""

# 截断，防止单条结果撑爆上下文
if len(content) > 500:
            content = content[:500] + "..."

lines.append(
f"[{idx}] {title}\n"
            f"    时间：{date[:10]}\n"
            f"    来源：{url}\n"
            f"    摘要：{content}"
)

return "\n\n".join(lines)


SEARCH_TOOL_SCHEMA = [
    {
"type": "function",
"function": {
"name": "web_search",
"description": "联网搜索实时信息。适用于查询最新进展、行业数据、企业动态等。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，简洁精准，5-15 字为宜。",
                    }
                },
"required": ["query"],
            },
        },
    }
]


# ---------------------------------------------------------------- 规划者

def make_plan(task: str) -> list[str]:
"""把复杂任务拆解成一份有序的子任务清单。"""

prompt = f"""你是一个经验丰富的任务规划专家。请把下面这个复杂任务，
拆解成 3-5 个具体的、有明确先后顺序的调研子任务，确保覆盖任务要求的各个方面，
不要在某个局部话题上过度深挖。

每个子任务必须是可以通过联网搜索完成的具体调研动作。

请只输出一个 JSON 字符串数组，例如：
["调研 A 的现状", "调研 B 的技术路线", "对比 A 与 B"]

不要输出任何多余的文字，不要用 markdown 代码块包裹。

任务：{task}"""

response = chat([{"role": "user", "content": prompt}])
    raw = parse_json_array(response.choices[0].message.content)
    plan = normalize_plan(raw)

if not plan:
# 兜底：解析失败时至少还能跑，但要让用户知道降级了
print("计划解析失败，降级为单步执行")
return [task]

return plan[:MAX_PLAN_STEPS]

跑一下 make_plan("调研今年新能源汽车电池技术的最新进展，写一份两千字的报告")，大概率会得到类似这样的清单：

...json

[

   调研今年最新发布的新能源汽车电池能量密度提升数据与主流车企量产车型的搭载情况

   调研固态电池、磷酸锰铁锂、钠离子电池等新技术的研发突破与商业化进展

   调研今年国内主要电池厂商（如宁德时代、比亚迪、中创新航）的技术路线与产品发布

   调研今年电池快充技术（如800V高压平台）的普及程度与配套充电基础设施发展

   调研今年电池安全技术（如热失控防护、无热蔓延设计）的改进与相关标准更新

]

看，有了这份清单，就不会出现“逮着固态电池一个话题查到天荒地老”的情况了：每个方向都被明确地分配了独立的名额 。

STEP 02写执行者，逐条完成子任务（复用第三篇的 ReAct 循环）

def execute_subtask(subtask: str) -> str:
"""针对单个子任务，用简化版 ReAct 循环完成它。"""

messages = [
        {
"role": "system",
"content": (
"你是一个调研助手。针对当前子任务进行联网搜索并总结要点。\n"
                "要求：\n"
                "1. 优先使用 web_search 工具获取真实信息，不要凭记忆编造；\n"
                "2. 总结时保留关键数据、时间和来源；\n"
                "3. 信息足够后直接给出总结，不要过度搜索。"
),
        },
        {"role": "user", "content": subtask},
    ]

for round_idx in range(MAX_REACT_ROUNDS):
        response = chat(messages, tools=SEARCH_TOOL_SCHEMA, thinking=True)
        reply = response.choices[0].message

# 没有调用工具 = 已经产出最终答案
if not reply.tool_calls:
return reply.content or "（模型未返回内容）"

messages.append(
            {
"role": "assistant",
"content": reply.content or "",
"tool_calls": [
                    {
"id": call.id,
"type": "function",
"function": {
"name": call.function.name,
"arguments": call.function.arguments,
                        },
                    }
for call in reply.tool_calls
                ],
            }
        )

for call in reply.tool_calls:
try:
                args = json.loads(call.function.arguments)
except json.JSONDecodeError:
                args = {}

# 用 .get 而不是 **args：模型多传一个键就不会把程序搞崩
query = args.get("query", "")
print(f"   🔍 搜索：{query}")
            result = web_search(query)

            messages.append(
                {
"role": "tool",
"tool_call_id": call.id,
"content": result,
                }
            )

# 轮数耗尽 —— 关键修复：不要丢弃已收集的信息，
# 而是去掉 tools 再调一次，逼模型基于已有上下文强制总结。
messages.append(
        {
"role": "user",
"content": "请停止搜索，直接基于以上已获得的信息，总结本子任务的调研要点。",
        }
    )
    final = chat(messages, thinking=True)
return final.choices[0].message.content or "（未能完成该子任务）"

STEP 03写复盘逻辑，支持动态重规划

def replan_if_needed(
    original_task: str,
    completed: list[dict],
    remaining: list[str],
    budget: int,
) -> list[str]:
"""根据最新执行结果，判断是否需要调整剩余计划。"""

if not remaining or budget <= 0:
return remaining

# 只喂结果摘要，避免上下文随轮数二次方膨胀
completed_brief = [
        {"subtask": c["subtask"], "result": c["result"][:300]}
for c in completed
    ]

    prompt = f"""原始任务：
{original_task}

已完成的子任务及结果摘要：
{json.dumps(completed_brief, ensure_ascii=False, indent=2)}

剩余待执行的子任务：
{json.dumps(remaining, ensure_ascii=False)}

请判断：基于已获得的信息，剩余计划是否仍然合理？

调整原则（保守优先 —— 没有充分理由就不要改）：
- 剩余任务已被前面的结果覆盖 → 删除；
- 发现了原始任务必需但计划遗漏的方向 → 补充；
- 先后顺序明显不合理 → 调整。

约束：剩余子任务最多 {budget} 条。

只输出一个 JSON 字符串数组，不要输出任何多余文字，不要用代码块包裹。
如果无需调整，原样输出剩余子任务数组。"""

response = chat([{"role": "user", "content": prompt}])
    raw = parse_json_array(response.choices[0].message.content)
    new_remaining = normalize_plan(raw)

if not new_remaining:
return remaining  # 解析失败 → 沿用原计划

return new_remaining[:budget]


def plans_differ(a: list[str], b: list[str]) -> bool:
"""语义级比较，而不是原代码那种逐字 != 。

    LLM 重述时标点、空格、措辞都会变，逐字比较等于每轮都判定"计划已调整"。
    """
if len(a) != len(b):
return True

    def norm(s: str) -> str:
return re.sub(r"[\s，。、,.\-—:：]", "", s).lower()

return [norm(x) for x in a] != [norm(x) for x in b]

STEP 04把三件套串起来，跑通整个 Plan-and-Execute 流程

def run_plan_and_execute(task: str) -> str:
    plan = make_plan(task)

print("初始计划：")
for index, item in enumerate(plan, start=1):
print(f"   {index}. {item}")

    completed: list[dict] = []
    i = 0

while i < len(plan) and i < MAX_PLAN_STEPS:
        subtask = plan[i]

print(f"\n正在执行第 {i + 1}/{len(plan)} 步：{subtask}")

        result = execute_subtask(subtask)
print(f"结果：{result[:200]}{'...' if len(result) > 200 else ''}")

        completed.append({"subtask": subtask, "result": result})

# 最后一步之后不需要重规划 —— 原代码这里在白烧 token
is_last = (i == len(plan) - 1)
if not is_last:
            old_remaining = plan[i + 1:]
            budget = MAX_PLAN_STEPS - (i + 1)   # 硬预算，杜绝无限膨胀

new_remaining = replan_if_needed(task, completed, old_remaining, budget)

if plans_differ(new_remaining, old_remaining):
print("计划已根据新信息调整：")
for idx, item in enumerate(new_remaining, start=i + 2):
print(f"   {idx}. {item}")
                plan = plan[:i + 1] + new_remaining

        i += 1

# 生成最终报告
print("\n正在撰写最终报告...")

    summary_prompt = f"""任务：
{task}

以下是各子任务的调研结果：

{json.dumps(completed, ensure_ascii=False, indent=2)}

请基于以上信息，撰写一份结构完整、逻辑清晰的报告。要求：
1. 有清晰的章节结构；
2. 保留调研结果中的关键数据和时间；
3. 只使用上述调研结果中出现的信息，不要编造；
4. 篇幅约 2000 字。"""

final = chat(
        [{"role": "user", "content": summary_prompt}],
thinking=True,
max_tokens=8192,   # 2000 字中文约 3000+ token，默认值会截断
)

return final.choices[0].message.content


if __name__ == "__main__":
    report = run_plan_and_execute(
"调研今年新能源汽车电池技术的最新进展，写一份两千字的报告"
)

print("\n最终报告：\n")
print(report)

跑起来后，你会清楚地看到智能体先亮出一份完整的调研提纲，然后一条一条按部就班地去查资料，每查完一条还会停下来“回头看看”计划还合不合理。如果中途发现某个方向信息特别丰富值得展开、或者某个方向查了发现没什么可挖的，它会主动调整后面的步骤，而不是死板地照本宣科。这比 ReAct 那种 “一路莽到底”的方式，明显更适合这种结构清晰、覆盖面要求高的任务 。

04、到底该用 ReAct，还是该用 Plan-and-Execute？

写到这儿，你可能会问：那以后是不是所有任务都该先规划一下？倒也不必，这里给你一个简单的判断标准：

直接用 ReAct

任务步骤数不确定、高度依赖中间结果、探索性强。比如“查一下报错原因，网络问题就重试，权限问题就提醒我”——下一步该干嘛完全取决于上一步查到了什么。

用 Plan-and-Execute

任务范围明确、有多个并列或有依赖关系的子目标。比如调研报告、多步骤数据处理、需要覆盖多个维度的分析任务——提前想清楚“要覆盖哪几块”，能显著提升产出的完整度。

实际项目里，这两者经常是嵌套使用的，就像我们上面的代码一样，外层是 Plan-and-Execute 把控大局，内层每个子任务的具体执行，又是一个小型的 ReAct 循环。

「该有全局规划的地方讲规矩，该灵活应变的细节处保留弹性」

下一步：一个人干不完，得叫帮手了

到目前为止，我们的智能体一直是“单打独斗”，一个模型，身兼规划者、执行者、复盘者数职。但现实中会遇到更复杂的场景：比如一个任务需要“一个角色专门负责调研，一个角色专门负责审核事实准确性，一个角色专门负责润色文字”，这种时候， 让一个模型分饰多角，很容易出现“既当裁判又当运动员”的自我认知混乱 。

下一篇，我们就来聊聊多智能体协作

怎么把一个大任务拆给多个各司其职的智能体，让它们像一个真正的团队一样分工配合、互相传递信息。下一篇见。

END

我是写代码的中年人，长期专注于 AI 算法、智能体、以及 AI 工程化落地。