4.3 动态任务生成与自适应工作流

4.3 动态任务生成与自适应工作流

本节约处在第四章第三站。前面所有编排都是"静态图"——任务在 kickoff 前就写死。但真实任务常有不确定性：用户给的输入越模糊，越需要系统在运行时自己拆解步骤。CrewAI 提供 planning 模式，让一个"规划者"在动手前先把任务图补全，这就是动态任务生成的落点。

先把"静态图 vs 动态规划"画

先把"静态图 vs 动态规划"画出来对比：

开启 planning 模式，需要给 Crew 配 planning=True 和一个规划模型。运行前，框架会用规划模型读取你给的任务和整体目标，产出一份更细的执行计划：

from crewai import LLM, Agent, Task, Crew, Process

planner_llm = LLM(model="openai/gpt-4o", temperature=0.2)

researcher = Agent(role="研究员", goal="找事实", backstory="严谨", verbose=True)
writer = Agent(role="写手", goal="成文", backstory="连贯", verbose=True)

# 只给粗粒度任务，细节交给规划者补全
goal_task = Task(description="产出 {topic} 的调研报告", expected_output="报告", agent=writer)

crew = Crew(
    agents=[researcher, writer],
    tasks=[goal_task],
    process=Process.sequential,
    planning=True,            # 开启规划
    planning_llm=planner_llm, # 规划专用模型
    verbose=True,
)
# crew.kickoff(inputs={"topic": "低空经济"})
​

开启后你会看到运行日志先打印"计

开启后你会看到运行日志先打印"计划"，再按计划执行。规划者可能把"产出报告"拆成"搜资料→列大纲→写初稿→自查"，这些子任务由它生成、不必你手写。适合输入模糊、步骤不确定的场景。

但动态不是免费午餐。我们给三条使用纪律：

• 规划者用独立模型、且温度压低，避免计划本身飘。把策划和执行的智能分开，互不污染。
• 动态生成的任务不进你的版本库，无法单测。所以只在对结果"形式要求宽松"的环节开，关键合规链路仍用静态图钉死。
• 动态模式拉长总耗时、加一次模型调用。能静态就静态，动态是给"真不确定"留的弹性。

另一种"轻量动态"不靠 planning，而是用 Task 的回调在跑中追加任务。CrewAI 提供 after_task 之类钩子，可在某步完成后根据产出决定是否补一步：

def maybe_add_check(tasks_output):
    # 若上一步产出标记'高风险'，返回需补的审核任务描述
    if "高风险" in str(tasks_output):
        return "请对高风险项做合规复核"
    return None

# crew = Crew(..., after_task=maybe_add_check)
​

这种"按产出分支"比全量 planning 更可控，因为它只在特定条件触发，大部分路径仍是你写死的静态图。我们偏爱这种局部弹性，胜过一上来全动态。

还有一个相关能力是 CrewAI 的 Flow（见第六章生态会提），它用代码显式描述"哪步完成后触发哪步"，比 planning 更确定、比静态图更灵活——属于"半动态"。但这里先记住主结论：动态任务生成解决"步骤未知"，而 4.2 的人在环解决"关键节点要人确认"，两者正交。

收尾提醒：自适应工作流的本质是"

收尾提醒：自适应工作流的本质是"把不确定性推迟到运行时再决定"。能推迟说明系统更健壮，但也更不可测。我们建议默认静态、局部钩子动态、全量 planning 仅在模糊输入场景开启，并全程留日志——因为你必须能回答"这次它到底自己加了哪几步"。

让 Crew 自己长出新任务

固定的 Task 列表应对不了"做到一半才发现还差一步"的情况。CrewAI 支持在运行中基于中间结果动态追加任务，实现自适应工作流。常见做法是先跑一个"规划任务"产出子任务清单，再实例化执行。

⚠️ 常见坑：动态生成的任务描述若含未定义占位符，后续 kickoff 会因 inputs 不全而崩。我们让规划任务的产出本身就是结构化的子任务描述，且子任务不引用外部占位符。

from crewai import Agent, Task, Crew, Process

planner = Agent(role="规划器", goal="拆出子任务", backstory="善于分解", verbose=True)
executor = Agent(role="执行者", goal="完成子任务", backstory="靠谱", verbose=True)

t_plan = Task(description="为 {goal} 列出三个子任务",
              expected_output="三个子任务标题", agent=planner)

# 假设规划产出后，我们据其动态建任务
sub_titles = ["调研现状", "设计方案", "写实施清单"]
dynamic_tasks = [
    Task(description=f"完成：{title}", expected_output="产出", agent=executor)
    for title in sub_titles
]

crew = Crew(agents=[planner, executor],
            tasks=[t_plan] + dynamic_tasks,
            process=Process.sequential)
print(len(crew.tasks), "个任务已组装")
​

💡 关键直觉：静态工作流像按图纸施工，动态工作流像边探边修。前者可预测、便宜；后者能应对未知，但你要约束"动态"的边界——只允许在既定范围内生成子任务，别让模型凭空造出脱离域的任务，否则链路会发散。

# 用白名单约束动态任务的命名空间，防止发散
ALLOWED = {"调研现状", "设计方案", "写实施清单"}
assert all(t in ALLOWED for t in sub_titles)
​

动态任务的收敛护栏

再强调一次收敛：动态生成必须落在白名单或模板内，不能让模型凭空造任务。我们常用的做法是"规划任务只产出结构化子任务标题"，执行器再据标题从预注册任务工厂里取实现，未知标题直接丢弃并告警。这样既保留了灵活，又杜绝了链路发散。

TASK_FACTORY = {
    "调研现状": lambda: Task(description="调研现状", expected_output="现状", agent=executor),
    "设计方案": lambda: Task(description="设计方案", expected_output="方案", agent=executor),
}

def build(title):
    return TASK_FACTORY.get(title)  # 未知标题返回 None，触发告警

assert build("调研现状") is not None and build("瞎写的") is None
​