📚 编程进阶教程站

第 13 章 · 规划 Planning、迭代与错误处理 ​

本章目标：掌握 planning=True 的"先计划后执行"机制，理解 max_iter 与 max_retry_limit 对 Agent 行为的约束，学会用 step_callback 和 usage_metrics 定位失败，并建立对工具循环、上下文超限等常见故障的处置直觉。

13.1 为什么需要 Planning ​

前面章节里，Crew 的执行顺序完全由任务列表决定，但每个任务内部怎么做是 Agent 临场发挥的。对于多步骤复杂任务（比如"调研→对比→写报告"），临场发挥容易漏步骤或顺序混乱。

CrewAI 的规划（Planning）功能把"想清楚再干"变成显式阶段：开启后，每次 kickoff 之前，所有 Crew 信息会先交给一个 AgentPlanner，它产出一份逐步计划（Step-by-Step Plan），并把这些计划内容追加到每个任务的描述中。

python
# planning_demo.py —— 最小规划示例
import os
from crewai import Agent, Task, Crew, Process, LLM

llm = LLM(
    model="openai/deepseek-chat",              # 三方 OpenAI 兼容模型
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7,
)

researcher = Agent(
    role="资深研究员",
    goal="深入调研大语言模型的最新进展",
    backstory="你在 AI 领域有十年研究经验，擅长信息检索与归纳。",
    llm=llm,
)

writer = Agent(
    role="报告撰写人",
    goal="基于研究素材撰写结构化报告",
    backstory="你是技术写作专家，输出条理清晰的中文报告。",
    llm=llm,
)

t1 = Task(
    description="全面调研 AI 大语言模型近一年的进展。",
    expected_output="10 条要点列表，每条一句话概括一项进展。",
    agent=researcher,
)
t2 = Task(
    description="基于上一任务的结果扩写成完整报告。",
    expected_output="一篇带小节标题的 Markdown 报告，不使用代码围栏。",
    agent=writer,
)

my_crew = Crew(
    agents=[researcher, writer],
    tasks=[t1, t2],
    process=Process.sequential,
    planning=True,          # 关键：开启规划
)
result = my_crew.kickoff()
print(result.raw)

运行时日志会出现 Planning the crew execution，随后打印出类似 "Task Number 1: ... Step-by-Step Plan:" 的完整计划——这份计划会被注入对应任务的 description。

注意默认规划模型

官方文档明确提示：开启 planning 后，CrewAI 默认使用 gpt-4o-mini 作为规划 LLM，这要求环境里有有效的 OpenAI API Key。如果你的团队统一走三方模型，务必显式指定 planning_llm，否则会在规划阶段报鉴权错误。

13.2 指定 planning_llm：让规划也走三方模型 ​

planning_llm 可以直接传字符串（走 LiteLLM provider 规则），也可以传一个 LLM 实例。按本课程约定，我们让它也指向 DeepSeek：

python
# 方式一：字符串形式（provider/model）
my_crew = Crew(
    agents=[researcher, writer],
    tasks=[t1, t2],
    process=Process.sequential,
    planning=True,
    planning_llm="openai/deepseek-chat",   # 走 openai/ 前缀 + 环境变量里的 key
)

# 方式二：显式 LLM 实例，可以精确控制 base_url 与温度
plan_llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.2,        # 计划要稳定，温度调低
)
my_crew2 = Crew(
    agents=[researcher, writer],
    tasks=[t1, t2],
    planning=True,
    planning_llm=plan_llm,
)

两个实践建议：

  • 规划模型温度调低（0~0.3）：计划需要的是确定性而非创造性；
  • 不必用最强模型：规划只做任务拆解，通常比执行用的模型便宜一档即可。

13.3 迭代上限 max_iter 与重试上限 max_retry_limit ​

Agent 执行任务是"思考→行动→观察"的循环。两个参数控制这个循环的失控边界（均为 Agent 构造参数）：

参数默认值含义
max_iter20单个任务内最大的思考-行动迭代次数；达到上限后 Agent 必须给出它当前的最好答案
max_retry_limit2任务执行出错时的最大重试次数；超过即彻底失败
python
# 为不同性质的 Agent 配置不同的失控边界
analyst = Agent(
    role="数据分析师",
    goal="对销售数据做多维分析",
    backstory="你精通数据分析，习惯反复验证结论。",
    llm=llm,
    max_iter=30,          # 复杂分析允许更多迭代
    max_retry_limit=3,    # 分析类任务多一次重试机会
)

coder = Agent(
    role="SQL 生成器",
    goal="把自然语言转成 SQL",
    backstory="你只输出 SQL。",
    llm=llm,
    max_iter=5,           # 简单确定性任务收紧上限，防止死循环烧 token
    max_retry_limit=1,    # 快速失败，暴露问题而不是掩盖
)

经验法则：

  • 工具多的探索型 Agent 放宽 max_iter（25~40），否则还没搜到结果就被强制收尾；
  • 格式转换型 Agent 收紧 max_iter（3~8），这类任务不该需要多次迭代；
  • 达到 max_iter 不算错误——Agent 会交出"尽力而为"的半成品答案，所以必须配合 expected_output 校验来判断结果是否可用；
  • max_retry_limit 只在真正抛异常时生效，"答得烂但没报错"不会触发重试。

13.4 用 step_callback 抓中间过程 ​

调试复杂 Crew 时，光看最终输出不够，你需要看到 Agent 每一步干了什么。step_callback 在每个 Agent 的每个步骤结束后被调用（Agent 级设置会覆盖 Crew 级设置）：

python
# 用回调把中间步骤记录成 JSONL 文件，便于事后分析
import json
import os
from datetime import datetime
from crewai import Agent, Task, Crew, LLM

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

def step_logger(step):
    """每个 Agent 步骤结束时触发；step 是该步的输出对象"""
    record = {
        "ts": datetime.now().isoformat(timespec="seconds"),
        "step": str(step)[:500],       # 截断防止超长行
    }
    with open("steps.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

researcher = Agent(role="研究员", goal="调研", backstory="专家", llm=llm)
t1 = Task(description="列出 2024 年三个值得关注的 AI 趋势。",
          expected_output="三条要点。", agent=researcher)

crew = Crew(agents=[researcher], tasks=[t1], step_callback=step_logger)
crew.kickoff()

与它对应的还有 Task(task_callback=...)：在整个任务完成时触发，接收任务输出对象，适合做"任务级"审计（第 14 章会系统对比两者）。调试期推荐组合：Crew 级 step_callback 看过程 + Task 级 callback 看产物。

另一个轻量手段是 verbose=True：直接在控制台打印执行日志。开发期开 verbose=True，生产期关掉改用回调/事件落盘。

13.5 usage_metrics：成本与规模监控 ​

每次 kickoff() 的返回值上带有 usage_metrics 属性，汇总本次执行的 token 消耗与请求次数：

python
result = crew.kickoff()
m = result.usage_metrics
print(f"总 tokens: {m.total_tokens}")
print(f"输入 tokens: {m.prompt_tokens}, 输出 tokens: {m.completion_tokens}")
print(f"成功请求数: {m.successful_requests}")

把它接入你的监控体系非常简单：

python
# 把每次执行的用量写入 CSV，供后续画趋势图
import csv

with open("usage.csv", "a", newline="") as f:
    csv.writer(f).writerow([
        datetime.now().isoformat(),
        m.total_tokens, m.prompt_tokens,
        m.completion_tokens, m.successful_requests,
    ])

如果某天发现 total_tokens 翻了几倍，先查三件事：是不是某个 Agent 在 max_iter 内疯狂调用工具？是不是上下文越滚越长？是不是规划/记忆功能引入了额外 LLM 调用？

13.6 常见失败模式与处置 ​

模式一：工具循环（Tool Loop）。Agent 反复调用同一个搜索工具，永远觉得"信息还不够"。处置：

  • 收紧 max_iter，逼它在限额内收敛；
  • 在任务 description 里写明"最多检索 3 次"，LLM 会尊重这类指令；
  • 检查工具返回内容是否为空或报错——空结果会让 Agent 不断换关键词重试。

模式二：上下文超限（Context Overflow）。前序任务输出了 8000 字报告，全部塞进后续任务的上下文，最终超过模型窗口报错。处置：

  • 让上游任务的 expected_output 明确限制长度（如"不超过 300 字摘要"）;
  • 用结构化输出（output_pydantic，见第 12 章）代替长文本传递；
  • 选择长上下文的三方模型，或在 Crew 中开启记忆做摘要压缩（第 10 章）。

模式三：静默降质。达到 max_iter 后不报错但答案残缺。处置：给关键任务加 guardrail 或在回调里做输出校验（长度、必需字段），不合格就抛异常触发重试。

python
# 一个简单的输出质量闸门
def quality_gate(task_output):
    if len(task_output.raw) < 100:
        raise ValueError("输出过短，疑似降质，需要重试")
    return task_output

t1 = Task(
    description="撰写产品分析短文。",
    expected_output="至少 200 字的分析短文。",
    agent=writer,
    callback=quality_gate,     # 任务完成时校验
)

13.7 本章小结 ​

  • planning=True 让 AgentPlanner 在执行前生成逐步计划并注入任务描述；默认规划模型是 gpt-4o-mini，三方模型场景必须显式配 planning_llm；
  • max_iter（默认 20）约束单任务内的迭代次数，触顶后交出"尽力而为"的答案；max_retry_limit（默认 2）只在抛异常时生效；
  • step_callback 抓每步过程，Task(callback=...) 校验产物，usage_metrics 监控 token 成本；
  • 工具循环靠收紧迭代数和明确检索预算治理，上下文超限靠控制上游输出长度与结构化输出治理。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 开启 planning=True 后，若不指定 planning_llm，CrewAI 会用什么模型做规划？

2. 关于 max_iter 与 max_retry_limit 的区别，正确的是？

3. step_callback 在什么时机被调用？

4. 下游任务因上游输出过长而频繁上下文超限，最对症的组合处置是？

🛠️ 动手实践 ​

  1. 给第 12 章的结构化输出示例加上 planning=True 与 DeepSeek 的 planning_llm，观察日志中的 Step-by-Step Plan 与不加规划时的差异。
  2. 写一个故意会陷入工具循环的任务（如"无限搜索直到找到完美答案"），分别用 max_iter=3 与 max_iter=20 跑一遍，用 usage_metrics 对比两者的 total_tokens。
  3. 实现 quality_gate 升级版：除了长度校验，再用正则检查输出是否包含指定的必备小节标题，不满足则抛异常并观察 max_retry_limit 生效的过程。

下一章我们把"黑盒回调"升级为完整的第 14 章 · 回调、日志与事件监听。