3.2 让Agent学会规划与执行

🚄 前言

Agent学会了使用工具，但面对复杂任务时，它容易犯两类错误：一是不验证自己的输出（幻觉），二是一口气做所有事导致后面越做越乱。这节课你将学会如何让Agent先反思、再规划、再执行，从"蛮力做事"进化到"有章法地做事"。

🍁 课程目标

你将学到：

• 理解Agent自我反思的两种模式：自我反馈和外部验证
• 掌握工作流的五种编排模式：流水线、分支、并行、MoA、人机协作
• Plan & Execute：Agent先规划再执行，代码即计划
• 判断何时用固定流程、何时用自主规划

# 加载百炼的 API Key 用于调用千问大模型
import os, sys
os.chdir(os.path.join(os.path.dirname(os.path.abspath('')), 'course_core'))
sys.path.insert(0, os.getcwd())

from openai import OpenAI
from config.load_key import load_key
load_key()

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
print(f"API Key 已加载：{os.environ['DASHSCOPE_API_KEY'][:5]}*****")

1 让 Agent 学会反思

1.1 课程被意外修改

有一位同事写了一篇 notebook 格式的交互式课程，里面有一些可执行的 Python 代码。为了让课程的语言更生动，你让机器人帮忙润色整篇文档的语言风格。

经过润色后，课程的语言更加流畅、结构更加清晰了。但当他尝试运行课程里的代码示例时，却发现了一个关键问题：

# 课程的原始代码
def get_user_data(usr_id: str):
  # ... some logic ...
  return f"Data for {usr_id}"

# 正常执行
print(get_user_data(usr_id="u-123"))

机器人在“优化”语言风格时，悄悄地“修正”了代码：

# Agent 润色后的代码
def get_user_data(user_id: str):
  # ... some logic ...
  return f"Data for {user_id}"

# 运行时报错！
# print(get_user_data(usr_id="u-123")) 
# TypeError: get_user_data() got an unexpected keyword argument 'usr_id'

机器人认为 user_id 比 usr_id 更规范，于是帮忙“优化”了函数定义，但这却导致调用函数的代码报错了，因为调用处的参数名没有一起改。

1.2 大模型的幻觉

你该如何解决这个问题呢？一个最直接的想法，就是告诉机器人要“小心一点”。试着在它的指令（System Prompt）里加上一句警告：

你是一位顶尖的课程作家，负责润色 Jupyter Notebook 格式的课程。
请优化文本，使其更具吸引力。
**重要：请绝对不要修改任何代码块中的内容，包括变量名、参数名和函数调用。**

但这样做效果不佳，因为你无法通过罗列禁止事项来覆盖所有可能的场景，即使你明确禁止了某些修改，大模型仍然有一定概率把原本正确的内容"顺手"改错。

其原因是，由于大模型存在幻觉，它的每一次生成都可能引入新的错误，而且缺乏验证和修正的内在机制。这种"优化"不仅限于变量名，还可能发生在

• 命令行参数（将 -p 8080 改为 --port 8080）
• API 版本号（将 v1/users 更新为 v2/users）
• 示例值（将演示用的 IP 地址 192.168.1.10 替换为 127.0.0.1）

等任何它认为"不够规范"的地方。最终，一篇经过"优化"却混杂了错误的课程，会让学员陷入无尽的调试循环。

因此，你自然会想到另一种思路：让大模型输出前自己检查一遍，就像在考试时，人类在交卷前会检查一遍答案。这正是目前业界解决此类问题的一种方法，我们称之为反思（Reflection）。

1.3 反思的两种模式

1.3.1 模式一：自我反馈 (Self Review)

要实现反思，最直接的思路是让模型自己审视输出结果。这里有两种实现方式，一种简单但效果有限，另一种更复杂但效果显著。

1.3.1.1 方法一：单步指令式反思

这是一种最容易想到的方法，你尝试在一次模型调用中，通过 Prompt 指示模型在生成答案的同时进行反思。

## 角色
你是一位顶尖的课程作家。

## 任务
1. 润色课程的语言表达，输出润色后的全文内容。

2. 反思润色后的内容：
    ⁃ 是否符合写作规范。
    ⁃ 除了语言表达，其他内容是否被意外修改。
输出反思的结果和修改建议。

3. 对照建议修改课程，输出修改后的全文内容。

## 课程初稿
{original_notebook_content}

这种方法有其优势：模型可以在生成过程中及时发现问题并纠错，同时上下文保持连贯，减少了多轮往返的开销。

但也存在明显的缺陷：反思和生成绑定在一起时，模型容易带着相同的思维偏差进行自我验证，陷入"自证正确"的怪圈。例如，如果它在第一步就把 usr_id 改成了 user_id，那么在第二步反思时，它很可能会认为这是一次"规范化改进"而非错误。

1.3.1.2 方法二：两步式"生成-反馈"

你也可以将反思过程独立出来，把任务拆分成两次模型调用，一次负责生成，另一次负责审查。

# 第一次调用 (写作 Agent)
你是一位顶尖的课程作家，负责润色 Jupyter Notebook 格式的课程。请优化文本，使其更具吸引力。

【课程初稿】:
{original_notebook_content}

# 第二次调用 (技术审查 Agent)
你是一个严苛的技术审查员。你的任务是审查润色后的课程内容，确保：
1. 润色后的课程符合写作规范。
2. 代码、配置、数据等技术内容没有被意外修改。

请比对【原始内容】和【润色后内容】：
- 如果只修改了文案表达，代码等技术内容完全一致（或只有注释、格式等非功能性变化），就回答"通过"。
- 如果发现任何技术内容被修改（如代码逻辑、变量名、参数、配置项等），就回答"不通过"，并指出具体是哪处内容被意外修改了。

【原始内容】:
{original_notebook_content}

【润色后内容】:
{draft_content}

由于审查 Agent 的视角与写作 Agent 不同，这种方法能有效避免大部分角色偏见。你甚至可以设置多个专项审查 Agent——事实审查、逻辑审查、风格审查、安全审查，这不仅提高了审查的全面性，还便于后期进行 A/B 测试和指标统计。

但这种方法也有明显的成本：每增加一个审查 Agent 就多一轮模型调用，token 消耗更高。

扩展阅读：成本与优化

这笔成本值得吗？

"生成-反馈"循环看似增加了调用次数，但我们不能简单地将"一次高质量输出"与"一次低质量输出"的成本进行比较。更公平的比较是：为获得一个"可用"的答案，两种路径的总成本是多少？

如果单次调用无法有效解决问题，那么它的成本无论多低，都是一种浪费。相反，通过"生成-反馈"的两步调用，让中端模型达到顶级模型 95% 的效果，就能用更低的成本实现几乎相同的业务价值。

如何进一步降低成本？

在"生成-反馈"循环中，原始课程文档会在多次模型调用中重复传递——写作 Agent 需要它，审查 Agent 也需要它，修正阶段可能还要再用一次。如果文档很长（比如包含几十个代码示例的完整课程），这种重复会带来显著的 token 消耗。

阿里云百炼提供了上下文缓存（Context Cache）机制来解决这个问题：首次调用时缓存共享内容（如原始文档），后续调用检测到相同前缀就直接复用，缓存命中部分按标准单价的 20% 计费。

你可以这样组织 Prompt 结构：

[缓存部分 - 所有调用都相同]
## 原始课程文档
{很长的原始notebook内容}

[变化部分 - 每次调用不同]
## 任务
{写作指令 / 审查指令 / 修正指令}

## 待处理内容
{润色后的草稿 / 审查反馈等}

通过这种方式，原始文档的 token 在首次调用时按标准单价计费，在后续审查、修正环节则享受缓存折扣，从而显著降低多轮调用的总成本。

需要注意的是，应该让审查 Agent 只负责"发现问题"而不是"直接重写"。如果让它直接修改内容，可能会破坏原文的表达意图或引入新的错误。

所以，你最好将审查结果 (review_result) 反馈给写作Agent，让它根据反馈继续修改回答。这就形成了一个“生成—反馈”循环，机器人的回答忠实度得到了显著提升。

1.3.1.3 自我反馈的缺陷

自我反馈更适合静态文本层面的比对（如检查代码是否被意外修改），但当你需要验证代码是否能正常运行时，这种模式的效果比较差。

这是因为大模型擅长判断代码"看起来是否合理"，但无法像编译器那样进行精确的语义验证。更关键的是，模型无法真正执行代码——它看不到运行日志、捕捉不到异常报错、也感知不到依赖库版本冲突或环境配置问题。

要解决这个问题，我们需要引入外部反馈机制。

1.3.2 模式二：外部反馈 (External Feedback)

1.3.2.1 使用工具验证结果

外部反馈的核心思路是：把大模型的生成结果放到真实环境中执行，再把执行结果（成功/失败、错误信息、测试报告等）反馈给模型，让它根据这些落地的事实进行迭代修正。

如果说自我反馈是“我认为我做得对不对”，那么外部反馈就是“事实证明我做得对不对”。

回到我们的 Jupyter Notebook 课程润色场景。假设课程中包含一些代码示例，我们希望在润色文案的同时，确保代码仍然能够正常运行。

这要求我们将上一章学习的工具使用 (Tool Use) 与反思机制结合起来。在这里，最直接的外部反馈来源，就是一个代码执行工具。AgentScope 内置了一个强大的 execute_python_code 工具，它为 Agent 提供了一个安全的代码解释器环境，使其能够真实地执行 Python 代码并获取结果或错误信息。

具体的工作流程如下：

1. 生成：写作 Agent 润色文档，输出了包含代码的 notebook 内容。
2. 与外部工具交互：系统自动提取出代码块，并调用 execute_python_code 工具尝试执行它。
3. 获取外部反馈：工具返回执行结果，假设遇到了错误：TypeError: get_user_data() got an unexpected keyword argument 'usr_id'。
4. 修正：Agent 接收到这份错误反馈，它现在明确地知道代码出现了什么问题。基于这份反馈，它会生成修正后的代码。

下面使用 AgentScope 实现了一个简单的例子。AgentScope 内置的 execute_python_code 工具无需手动创建工具函数，只需通过 toolkit.register_tool_function(execute_python_code) 注册即可。

"""
使用 AgentScope 实现外部反馈的反思机制（简化版）
场景：润色 Jupyter Notebook 课程，通过代码解释器验证代码正确性
"""

import asyncio
import os
from textwrap import dedent

from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit, execute_python_code


# ============================================================
# 创建写作 Agent（装备代码解释器工具）
# ============================================================

def create_writer_agent() -> ReActAgent:
    """创建负责润色课程内容的写作 Agent"""
    
    # 创建工具包并注册内置的代码执行工具
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)
    
    writer = ReActAgent(
        name="Writer",
        sys_prompt=dedent("""你是一位技术课程作家，负责润色 Jupyter Notebook 课程。

            任务要求：
            1. 优化文本表达，使其更流畅生动
            2. **绝对不要修改代码中的变量名、函数名、参数名**
            3. 润色后，使用 execute_python_code 工具验证所有代码块
            4. 如果代码执行失败，检查是否意外修改了代码并修正

            记住：只改文案，不改代码逻辑！
        """),
        model=DashScopeChatModel(
            model_name="qwen-plus",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
            stream=False,
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        memory=InMemoryMemory(),
        max_iters=10,  # 最多迭代 10轮（示例配置，实际应根据任务复杂度和成本预算调整）
    )
    
    return writer


# ============================================================
# 主流程
# ============================================================

async def main():
    """主函数 - 演示外部反馈的反思机制"""
    
    # 原始课程内容（包含"不规范"的变量名 usr_id）
    original_notebook = """
# Python 函数示例课程

本节学习如何定义和调用函数。

## 定义函数

def get_user_data(usr_id: str):
    return f"Data for {usr_id}"

## 调用函数

print(get_user_data(usr_id="u-123"))

## 小结

你学会了定义和调用 Python 函数。
"""
    
    print("="*60)
    print("原始课程内容：")
    print("="*60)
    print(original_notebook)
    
    # 创建 Agent
    writer = create_writer_agent()
    
    # 发送润色请求
    user_msg = Msg(
        name="user",
        content=dedent(f"""请润色以下课程内容，要求：

            {original_notebook}

            1. 让文案更生动易懂
            2. **不要修改代码中的变量名、函数名**
            3. 润色后用 execute_python_code 验证代码能否运行
            4. 如果报错，说明你可能改错了代码，请修正
        """),
        role="user"
    )
    
    print("\n" + "="*60)
    print("Agent 开始工作...")
    print("="*60)
    
    # Agent 自动进行：润色 → 验证 → 修正 → 再验证
    result = await writer(user_msg)
    
    print("\n" + "="*60)
    print("最终输出：")
    print("="*60)
    print(result.get_text_content())

try:
    await main()
except NameError:
    asyncio.run(main())

通过这种方式，Agent 可以使用外部工具验证输出结果，从而实现了高效、精准的自我修正。

1.3.2.2 其他验证方法

除了代码执行，这种模式还能应用于其他需要客观事实的场景：

• 场景：优化科研论文中的 Matplotlib 图表

  • 问题：你要求 Agent 为一篇科研论文生成数据可视化代码。Agent 生成了一段看起来"合理"的 Matplotlib 代码，包含了正确的数据处理逻辑和绘图函数调用。但代码能运行不代表图表效果好——实际渲染出来的图可能存在坐标轴标签重叠、图例遮挡数据点、字体过小难以辨认、配色不适合打印等问题。
  • 外部反馈解法：系统调用代码解释器工具执行 Matplotlib 代码，并返回生成的图片文件。Agent 通过视觉输入接收到实际渲染的图表，能够直观地发现视觉问题（例如"x 轴标签重叠了""图例挡住了关键数据点"），然后针对性地调整代码参数（如 plt.xticks(rotation=45) 旋转标签、bbox_to_anchor 调整图例位置、增大 fontsize 等）。这个"生成-渲染-调整"的循环，确保最终图表符合学术出版的质量要求。

• 场景：核对复杂计算题的答案

  • 问题：在一节量子力学课程中，你需要出一道练习题：“计算一个电子（质量 ≈ 9.11e-31 kg）被限制在长度为 1 纳米（1e-9 m）的一维无限深势阱中时，其基态能量是多少焦耳？（普朗克常数 h ≈ 6.626e-34 J·s）” Agent 在解题步骤中可能由于复杂的指数运算而出错。
  • 外部反馈解法：系统识别出需要计算的表达式 (1**2 * (6.626e-34)**2) / (8 * 9.11e-31 * (1e-9)**2)，并将其交给代码解释器或计算器工具执行。工具返回精确的数值结果。这份客观的计算结果会作为反馈，让 Agent 修正最终的答案。

• 场景：结构化输出校验

  • 问题：你要求 Agent 生成符合特定 JSON Schema 的配置文件，但它可能遗漏必填字段或使用错误的数据类型。
  • 外部反馈解法：系统使用 Pydantic 等库对生成的 JSON 进行校验。当输出不符合 Schema 时，校验器返回详细的错误报告（如"字段 'timeout' 应为整数而非字符串"）。这份客观、精准的反馈让 Agent 能够修正输出，直到完全符合预定义的结构。这个"生成-校验-反馈"的循环，是反思机制在实践中最常见、最基础的应用之一。

最后，如何选择这两种反馈模式呢？这取决于你的具体需求、预算和可接受的错误率。比如，如果只是润色一篇没有代码的博客文章，那么可能根本不需要反思。但如果是修改一份包含几十个代码示例的交互式课程，那么引入一个基于代码解释器的“外部反馈”循环，就是确保文档质量、避免发布事故的必要投资。

1.4 总结

让我们回顾一下你在本节学到的知识：

• 直接指令的局限性：直接在 Prompt 中要求大模型“更小心”或“不许改代码”通常效果不佳，因为模型以生成“更合理”的文本为目标，有时会“好心办坏事”。
• “反思”的核心思路：模仿人类的“元认知”，让模型有机会审视和评估自己已经生成的完整内容，从而发现并修正错误。这比简单的指令更可靠。
• 自我反馈 vs. 外部反馈：“自我反馈”是让另一个 Agent 检查初稿，适合主观评估；“外部反馈”是借助工具（如代码解释器）来验证结果，适合需要客观事实的场景。
• 工程实现：“生成-反馈”循环：实现“反思”的有效方法是采用多次调用：第一次生成初稿，后续调用则负责评估初稿、提供反馈，并基于反馈进行修改。

2 构建工作流

2.1 单个 Agent 处理复杂流程

周一早上，你收到了这样一条消息：

"上周五提交的《Python数据分析入门》课程需要尽快上线。能不能用你那个AI助手跑一遍标准审核流程？就是：1)所有代码示例要能运行 2)技术概念不能有硬伤 3)难度梯度适合零基础 4)语言风格符合我们的规范。下午3点前要结果。"

你心想，这个任务挺简单的。你的机器人已经能验证代码了，处理这个"组合任务"应该不难。于是你把所有要求塞进一个prompt里：

prompt = """请完成以下课程审核任务：
1. 验证所有Python代码能正确运行
2. 检查技术概念的准确性（特别是pandas和numpy的用法说明）
3. 评估难度曲线是否适合编程新手
4. 按照我们的风格指南调整用词（避免"很简单""超级"等口语）
请给出完整的审核报告。"""
response = agent.run(prompt + course_content)

几分钟后，你打开返回的结果：

• Agent确实验证了代码，但在"优化语言"时，它把代码里的super()也改成了parent()，因为它觉得"super太口语化"。
• 它发现第 3 章对DataFrame的解释有误，改正后却引入了更严重的错误：说"DataFrame是Python内置的数据结构"。
• 至于难度评估？它在处理到第 50 个问题时似乎已经忘了这回事。

你试图让它重来，这次它记住了评估难度，却漏掉了一半的代码验证。第三次，它所有任务都做了，但把原本正确的概念"改错了"。

问题很明显：让单个Agent同时接管多个复杂任务，就像让一个实习生同时接四个电话——总有一个会出问题。

2.2 失败原因分析

2.2.1 注意力机制的遗忘效应

大语言模型在生成回复时，对不同位置信息的"注意力"是不均等的。当Agent处理到第四个任务（润色语言）时，第一个任务（代码验证）的细节已经被大量中间信息稀释。

具体表现：模型在后期修改时"忘记"了前期验证的约束。比如你已经验证super()是正确的Python语法，但在语言润色阶段，模型可能因为"super"这个词看起来口语化而将其替换，导致代码出错。

这不是简单的"记忆力差"，而是Transformer架构中self-attention机制的固有限制——token之间的注意力权重会随着序列长度增加而被稀释。

2.2.2 错误的级联放大

在顺序执行多个任务时，早期的错误会成为后续处理的"事实基础"。假设Agent在事实检查阶段错误地"纠正"了DataFrame的定义（比如说它是Python内置的），这个错误会进入后续的上下文。

更糟的是，模型会基于这个错误的"事实"继续推理：

• 既然DataFrame是"内置的"，那么就不需要import pandas
• 教学案例中就会删除导入语句
• 代码示例全部失效

这种错误传播不是线性的，而是指数级的——一个小错误会触发一连串的错误决策。

2.2.3 扁平化理解 vs 结构化需求

你的复杂需求其实是一个有内在结构的任务图：

• 有些任务可以并行（代码检查和风格检查互不干扰）
• 有些任务有依赖（必须先理解概念才能评估难度）
• 有些任务需要全局视角（评估整体难度曲线）

但对模型来说，你的prompt只是一个扁平的token序列。它无法自动识别出这种结构关系，而是试图用一个线性的生成过程去解决一个本质上是图结构的问题。

就像让某人同时玩四个不同规则的游戏，还不告诉他哪些可以轮流玩，哪些必须同时进行。

最终，依赖单个 Agent 执行复杂审阅，就像在没有施工蓝图的情况下建造一栋房子，每砌一块砖都可能影响整栋建筑的结构稳定。

2.3 工作流的几种模式

既然让一个 Agent 包办所有事情行不通，那你自然会想到另一种思路：就像你在管理一个项目时，你会把项目拆分成几个子任务，分配给不同的人或在不同时间点完成。这种“分而治之”（Divide and Conquer）的思想，正是解决此类问题的核心。

我们将这种把复杂任务拆解成多个节点，并定义它们之间执行关系的模式，称为工作流 (Workflow)。

构建工作流的关键，是理解业务并做出最合适的任务拆解。下面，我们将从最简单的模式开始，一步步构建出能够处理复杂课程审阅的强大工作流。

2.3.1 模式一：流水线 (Pipeline)

这是最基础、最直观的工作流。它将一个任务分解为多个固定的、按顺序执行的步骤。前一个步骤的输出，严格作为后一个步骤的输入，整个过程像工厂的流水线一样，单向且不可变。你在本课程早期构建的 RAG 问答机器人，就是这种模式的完美体现。

"""
模式一：流水线（Pipeline） - 场景：课程快速检查流程
目标：用户提交一篇课程初稿，系统按顺序完成
1) 提取课程中的代码 -> 2) 验证代码可执行性 -> 3) 生成代码检查报告
"""
import asyncio
from agentscope.message import Msg
from agentscope.pipeline import sequential_pipeline
from chatbot.agent import create_agent, disable_console_output

async def run_pipeline() -> None:
    # 节点A：代码提取 Agent
    # multi_agent=True 是 AgentScope 中的一个配置，用于确保 Agent 之间的通信格式兼容，
    # 是构建工作流和多智能体系统的推荐设置。
    code_extractor = create_agent(
        name="代码提取器",
        sys_prompt=(
            "你是代码提取专家。请从用户提供的课程文本中，精确地提取出所有 Python 代码块。"
            "只输出代码，不要有任何其他解释。"
        ),
        model_name="qwen-flash",
        multi_agent=True,
    )

    # 节点B：代码验证 Agent (可调用外部工具)
    code_validator = create_agent(
        name="代码验证器",
        sys_prompt=(
            "你是代码执行与验证专家。你将接收到代码文本。请使用代码解释器来执行它。"
            "报告代码是否能成功运行，如果不能，请指出错误。"
        ),
        # 此处可以配置 code_validator 使用你在第二节学过的代码解释器工具
        model_name="qwen-plus",
        multi_agent=True,
    )

    # 节点C：报告生成 Agent
    report_generator = create_agent(
        name="报告生成器",
        sys_prompt=(
            "你是审阅报告撰写助理。根据上一步的代码验证结果，为课程设计师生成一份简洁明了的检查报告。"
        ),
        model_name="qwen-max",
        multi_agent=True,
    )

    agents = [code_extractor, code_validator, report_generator]
    disable_console_output(agents)

    course_draft = (
        "这是我们的新课程。第一部分是`print('Hello, World!')`。"
        "第二部分是一个有问题的代码`x = 1 / 0`。"
    )
    result = await sequential_pipeline(
        agents=agents,
        msg=Msg("user", course_draft, "user"),
    )

    print("=" * 50)
    print("流水线输出：")
    print(result.content)
    print("=" * 50)

async def main() -> None:
    await run_pipeline()

await main()

它的核心优势在于简单、可预测且易于调试。因为流程是固定的，所以当出现问题时，你可以很容易地定位到是哪个环节（提取、验证还是报告）出了问题。这种确定性在许多企业级应用中至关重要。

它适用于业务流程非常固定、逻辑单一的任务。例如：

• 代码初步检查：先提取代码，再运行验证。
• 文档翻译：先提取文本，再进行翻译，最后格式化输出。
• 新员工入职材料分发：先生成欢迎邮件，再附上公司文档，最后发送。

流水线的刚性是它最大的优点，也是最致命的缺点。它无法处理流程之外的任何变化。面对“帮我评估一下这篇课程的趣味性”这样的请求，这条为“代码检查”设计的流水线会完全不知所措，因为它没有处理这种意图的能力。它假设所有输入都应遵循同一套处理逻辑。

2.3.2 模式二：分支选择 (Branching)

为了克服流水线的僵化，你需要引入决策能力。分支选择模式的核心是在工作流的开始或关键节点设置一个“路由器”或“调度中心”。这个决策节点会分析输入（例如，用户的审阅要求），然后像一个交通警察一样，将任务引导到不同的、预设好的处理路径（即不同的流水线或专家）上去。

"""
模式二：分支选择（Branching） - 场景：课程审阅任务分发
路由 Agent 读取审阅请求，选择以下分支之一：
1) code_check: 仅快速检查代码
2) style_guide: 按风格指南润色语言
3) full_review: 进行全面的多维度评审
"""
import asyncio
from typing import Literal
from pydantic import BaseModel, Field
from agentscope.message import Msg
from chatbot.agent import create_agent, disable_console_output

class RouteChoice(BaseModel):
    choice: Literal["code_check", "style_guide", "full_review", None] = Field(
        description="根据用户意图选择分支：code_check/style_guide/full_review/None"
    )
    extra: str | None = Field(default=None, description="对任务的简要说明")

async def branch_code_check(user_msg: Msg) -> Msg:
    agent = create_agent(
        name="代码快检专家",
        sys_prompt="你是代码快检专家。根据用户需求，快速验证课程中的代码片段是否能运行。",
        model_name="qwen-plus",
        multi_agent=True,
    )
    disable_console_output([agent])
    return await agent(user_msg)

async def branch_style_guide(user_msg: Msg) -> Msg:
    agent = create_agent(
        name="语言润色专家",
        sys_prompt="你是语言润色专家。请根据公司风格指南，改写和润色用户提供的课程文本。",
        model_name="qwen-max",
        multi_agent=True,
    )
    disable_console_output([agent])
    return await agent(user_msg)

async def branch_full_review(user_msg: Msg) -> Msg:
    agent = create_agent(
        name="首席评审",
        sys_prompt="你是首席评审。告知用户，你将启动一个包含代码、事实和教学法在内的全面评审流程。",
        model_name="qwen-flash", # 使用轻量模型模拟启动流程的告知动作
        multi_agent=True,
    )
    disable_console_output([agent])
    return await agent(user_msg)


async def run_branching() -> None:
    router = create_agent(
        name="审阅任务分发员",
        sys_prompt=(
            "你是课程审阅任务的分发员，根据用户输入选择分支：\n"
            "- 如果只是想检查代码，输出 code_check\n"
            "- 如果是想润色文笔，输出 style_guide\n"
            "- 如果是需要完整、全面的评审，输出 full_review\n"
            "仅通过结构化输出来表达你的选择，不要正文回答。"
        ),
        model_name="qwen-plus",
        multi_agent=False,
    )

    user_text = "这篇课程写的差不多了，帮我全面检查一下，特别是代码和难度。"
    
    res = await router(
        Msg("user", user_text, "user"),
        structured_model=RouteChoice,
    )
    choice = res.metadata.get("choice")

    if choice == "code_check":
        out = await branch_code_check(Msg("user", user_text, "user"))
    elif choice == "style_guide":
        out = await branch_style_guide(Msg("user", user_text, "user"))
    elif choice == "full_review":
        out = await branch_full_review(Msg("user", user_text, "user"))
    else:
        # 默认走全面评审，保证示例可运行
        out = await branch_full_review(Msg("user", user_text, "user"))

    print("=" * 50)
    print(f"分支选择：{choice}")
    print(out.content)
    print("=" * 50)

async def main() -> None:
    await run_branching()

await main()

与单一流水线相比，分支选择让系统变得更灵活、更智能。它使得一个应用能够处理多种不同类型的任务，极大地扩展了其适用范围，提升了用户体验。

常见的应用场景有：

• 智能客服：根据用户问题类型（课程内容咨询、平台技术支持、购买建议）转接到不同的处理流程。
• 多工具 Agent：Agent 根据任务需求，决定是调用代码解释器、搜索引擎还是内部知识库。
• 内容处理系统：根据内容类型（视频、文本、交互式 Notebook）调用不同的审核流程。

分支选择本质上是“多选一”，它依然是串行的。它能处理“代码检查”或“语言润色”，但无法处理“一边检查代码，一边润色语言”的复合请求。对于我们最初那个包含四个审核维度的复杂请求，它一次只能走一条分支，需要用户与机器人进行多次独立的对话才能完成，效率极低。

2.3.3 模式三：并行执行 (Parallel Execution)

当一个请求可以被分解为多个互不依赖的子任务时，让它们排队等待是极大的浪费。并行执行模式的核心思想是“同时进行”。它首先将一个复杂任务拆解成多个子任务，然后将这些子任务分发到不同的执行单元（Agent或工具）同时处理，最后再将所有结果汇集起来，形成最终的输出。

"""
模式三：并行执行（Parallel Execution） - 场景：课程完整审阅
将独立的审阅子任务并行执行：代码检查、事实核对、教学法评估、语言风格检查。
示例使用 fanout_pipeline 并行收集报告，再由合成 Agent 汇总。
"""
import asyncio
from typing import List
from agentscope.message import Msg
from agentscope.pipeline import fanout_pipeline
from chatbot.agent import create_agent, disable_console_output

async def run_parallel() -> None:
    # 四个独立子任务的“专家”Agent
    code_checker = create_agent(
        name="代码检查员",
        sys_prompt="验证课程中的代码是否正确无误，并给出修复建议。",
        model_name="qwen-plus",
        multi_agent=True,
    )
    fact_checker = create_agent(
        name="事实核查员",
        sys_prompt="核对课程中的技术概念、函数解释是否准确，引用是否规范。",
        model_name="qwen-plus",
        multi_agent=True,
    )
    pedagogy_evaluator = create_agent(
        name="教学法评估师",
        sys_prompt="评估课程的难度曲线、案例趣味性和练习有效性。",
        model_name="qwen-flash",
        multi_agent=True,
    )
    style_editor = create_agent(
        name="风格编辑",
        sys_prompt="根据公司风格指南，检查并报告语言风格、术语一致性问题。",
        model_name="qwen-flash",
        multi_agent=True,
    )

    experts = [code_checker, fact_checker, pedagogy_evaluator, style_editor]
    disable_console_output(experts)

    course_content = "这是我们新开发的 Python 数据分析入门课..."
    msgs = await fanout_pipeline(
        agents=experts,
        msg=Msg("user", course_content, "user"),
        enable_gather=True,
    )

    # 汇总 Agent
    summarizer = create_agent(
        name="总编辑",
        sys_prompt="将来自多位专家的审阅意见汇总成一份结构清晰、条理分明的总审阅报告。",
        model_name="qwen-max",
        multi_agent=True,
    )
    disable_console_output([summarizer])

    merged_text: List[str] = [m.content[0]["text"] for m in msgs]
    prompt = "\n\n".join(merged_text)
    summary = await summarizer(Msg("user", prompt, "user"))

    print("=" * 50)
    print("并行执行输出：")
    print(summary.content)
    print("=" * 50)

async def main() -> None:
    await run_parallel()

await main()

最显著的优势是效率的大幅提升。工作流的总耗时不再是所有子任务耗时之和，而是取决于耗时最长的那个子任务。这使得 Agent 能够快速响应包含多个步骤的复杂请求。

常见的应用场景有：

• 处理复杂审阅请求：如我们的课程审阅场景，同时处理代码、事实、教学法、风格等多个维度。
• 生成综合报告：同时从不同数据源（用户反馈、市场趋势、竞品分析）拉取信息，并分别进行分析，最后汇总成一份新课程立项报告。
• 批量数据处理：同时对多个课程单元执行相同的格式化或检查操作。

此模式的前提是子任务之间相互独立。如果任务之间存在依赖关系（例如，必须先确认课程的核心知识点，才能评估其案例是否贴切），则无法简单地并行。此外，它假设每个执行单元都能给出“正确”的答案，不适用于需要多方比较、权衡才能得出最佳方案的创造性或决策性任务。

2.3.4 模式四：混合专家 (Mixture-of-Agents, MoA)

与并行执行旨在提升效率不同，混合专家模式的核心目标是追求极致的质量、鲁棒性和创造性。MoA 的核心理念基于一个关键发现：不同的大语言模型具有各自独特的优势和专长，而当一个模型能够参考其他模型的输出时，往往能生成质量更高的响应——这种现象称为模型的"协作性"(Collaborativeness)。MoA 的做法是，让多个不同的大语言模型同时处理同一个任务，然后由一个聚合模型对所有输出进行综合、分析和融合，从而产生一个远超任何单个模型水平的最终结果。

"""
模式四：混合专家（Mixture-of-Agents, MoA） - 场景：课程核心卖点提炼
使用三个不同的大语言模型并行处理同一任务，通过聚合模型融合它们的输出，
利用模型间的协作性(Collaborativeness)产生更高质量的结果。
"""
import asyncio
from agentscope.message import Msg
from agentscope.pipeline import fanout_pipeline
from chatbot.agent import create_agent, disable_console_output

async def run_moa() -> None:
    # 使用三个不同的模型作为提议者(Proposer)
    # 每个模型有其独特的优势，但处理相同的任务
    proposer1 = create_agent(
        name="Qwen3-Max",
        sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
        model_name="qwen3-max",
        multi_agent=True,
    )
    proposer2 = create_agent(
        name="DeepSeek-V3.2",
        sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
        model_name="deepseek-v3.2-exp",
        multi_agent=True,
    )
    proposer3 = create_agent(
        name="Kimi-K2",
        sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
        model_name="kimi-k2-thinking",
        multi_agent=True,
    )

    proposers = [proposer1, proposer2, proposer3]
    disable_console_output(proposers)

    task = (
        "这是一门新的'面向Web开发者的AI大模型应用'课程，请为其提炼核心卖点和宣传文案。"
    )
    msgs = await fanout_pipeline(
        agents=proposers,
        msg=Msg("user", task, "user"),
        enable_gather=True,
    )

    # 聚合器(Aggregator)接收所有模型的输出，综合产生最佳结果
    aggregator = create_agent(
        name="聚合器",
        sys_prompt=(
            "你的任务是综合多个大语言模型对同一问题的回答。"
            "这些回答来自不同的模型，各有优劣。请批判性地评估这些回答，"
            "识别其中的优点和不足，然后融合这些信息，生成一个高质量、准确、全面的最终回答。"
            "确保你的回答结构清晰、逻辑连贯，并达到最高的准确性和可靠性标准。"
        ),
        model_name="qwen3-max",
        multi_agent=True,
    )
    disable_console_output([aggregator])

    # 将所有提议者的输出合并，传递给聚合器
    merged = "\n\n".join([
        f"模型 {i+1} 的回答：\n{m.content}"
        for i, m in enumerate(msgs)
    ])
    final = await aggregator(Msg("user", merged, "user"))

    print("=" * 50)
    print("MoA 聚合输出：")
    print(final.content)
    print("=" * 50)

async def main() -> None:
    await run_moa()

await main()

Mixture-of-Agents(MoA) 工作原理：

MoA 将参与的模型分为两类角色：

1. 提议者(Proposers)：多个不同的模型并行处理同一任务，各自生成响应。这些模型可能在某些方面表现出色（如逻辑推理、创意表达、事实准确性等）。
2. 聚合器(Aggregator)：接收所有提议者的输出，通过批判性评估、比较和融合，生成一个质量更高的最终响应。

关键的是，聚合器并非简单地选择最好的答案，而是能够从多个响应中提取各自的优点，综合产生一个超越任何单一模型的结果。

MoA 的优势：

• 利用模型多样性：不同模型有不同的训练数据、架构和优化目标，导致它们在不同任务上表现各异。MoA 能够同时利用多个模型的长处。
• 增强鲁棒性：即使某个模型在特定输入上表现不佳，其他模型的高质量输出也能保证最终结果的质量下限。
• 质量的涌现效应：研究表明，即使提议者模型的单独输出质量较低，聚合后的结果仍可能超越任何单一模型——这是"模型协作性"的直接体现。

MOA 适用于那些没有唯一标准答案、对结果质量要求极高、价值巨大的开放性或创造性任务。

• 核心文案撰写：如课程Slogan、推广文案、品牌故事。
• 复杂决策分析：综合不同模型的分析报告，形成更全面的新课程方向决策建议。
• 代码生成与优化：让不同模型生成一段示例代码，再由评审 Agent 择优或进行融合重构，以达到最佳的教学效果。

不过，MOA 最主要的制约是成本。调用 N 个专家 Agent 会带来 N 倍的计算成本和相应的延迟。这是一种用资源换质量的策略，因此必须用在"好钢用在刀刃上"的关键环节，不适合用于常规的、对成本敏感的日常任务。

进阶：多层 MoA (Multi-Layer MoA)

前面展示的 MoA 是一个"2层结构"：第一层有多个专家并行处理任务，第二层由一个聚合器综合所有专家的输出。然而，MoA 也可以扩展到 3 层或更多层，通过多轮迭代不断提炼和优化结果，从而获得比单层 MoA 更卓越的输出质量。

多层 MoA 的核心思想是：将上一层的聚合输出作为下一层的输入，再次交给多个专家进行审视、批判和改进，然后由新的聚合器进行更高层次的综合。这种"迭代提炼"的过程，类似于人类团队中的多轮评审和打磨，每一轮都能发现前一轮遗漏的问题，激发新的创意，最终达到单轮难以企及的质量水平。

多层 MoA 的优势：

• 质量的进一步提升：第二层、第三层的专家可以站在第一层结果的基础上，进行更深入的分析和优化，就像编辑团队对初稿进行多轮润色一样。
• 纠错能力增强：即使第一层的某些专家犯了错误，后续层次的专家有机会发现并纠正这些错误，使最终结果更加可靠。
• 创意的涌现：多层交互可能产生"1+1>2"的效果，不同层次专家的思想碰撞可能激发出任何单层都无法产生的创新方案。

实现要点：

async def run_multi_layer_moa() -> None:
    """
    多层 MoA 示例：3层架构提炼课程营销方案
    Layer 1: 3个不同模型作为提议者并行生成方案
    Layer 2: 3个不同模型对第一层的聚合输出进行优化和改进
    Layer 3: 最终聚合器综合所有信息，输出最佳方案
    """
    # 通用的聚合提示词
    aggregate_prompt = (
        "你的任务是综合多个大语言模型对同一问题的回答。"
        "请批判性地评估这些回答，识别其中的优点和不足，"
        "然后融合这些信息，生成一个高质量、准确、全面的最终回答。"
    )

    # Layer 1: 初始提议者层(3个不同模型)
    layer1_proposers = [
        create_agent(
            name="Proposer-L1-1",
            sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
            model_name="qwen3-max",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L1-2",
            sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
            model_name="deepseek-v3.2-exp",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L1-3",
            sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
            model_name="kimi-k2-thinking",
            multi_agent=True
        ),
    ]

    task = "这是一门新的'面向Web开发者的AI大模型应用'课程，请为其提炼核心卖点和宣传文案。"
    layer1_outputs = await fanout_pipeline(
        agents=layer1_proposers,
        msg=Msg("user", task, "user"),
        enable_gather=True,
    )

    # Layer 1 聚合器
    layer1_aggregator = create_agent(
        name="Aggregator-L1",
        sys_prompt=aggregate_prompt,
        model_name="qwen3-max",
        multi_agent=True,
    )
    layer1_merged = "\n\n".join([f"模型 {i+1}:\n{m.content}" for i, m in enumerate(layer1_outputs)])
    layer1_result = await layer1_aggregator(Msg("user", layer1_merged, "user"))

    # Layer 2: 第二轮提议者层(3个不同模型，基于第一层的聚合结果进行优化)
    layer2_proposers = [
        create_agent(
            name="Proposer-L2-1",
            sys_prompt="你是一个专业的课程分析师。请审视给定的营销方案，并提出改进建议或优化版本。",
            model_name="qwen3-max",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L2-2",
            sys_prompt="你是一个专业的课程分析师。请审视给定的营销方案，并提出改进建议或优化版本。",
            model_name="deepseek-v3.2-exp",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L2-3",
            sys_prompt="你是一个专业的课程分析师。请审视给定的营销方案，并提出改进建议或优化版本。",
            model_name="kimi-k2-thinking",
            multi_agent=True
        ),
    ]

    layer2_prompt = f"以下是第一轮分析生成的营销方案：\n\n{layer1_result.content}\n\n请在此基础上提出改进建议或优化版本。"
    layer2_outputs = await fanout_pipeline(
        agents=layer2_proposers,
        msg=Msg("user", layer2_prompt, "user"),
        enable_gather=True,
    )

    # Layer 2: 最终聚合层
    final_aggregator = create_agent(
        name="Final-Aggregator",
        sys_prompt=aggregate_prompt,
        model_name="qwen3-max",
        multi_agent=True,
    )

    layer2_merged = "\n\n".join([f"模型 {i+1}:\n{m.content}" for i, m in enumerate(layer2_outputs)])
    final_output = await final_aggregator(Msg("user", layer2_merged, "user"))

    print("多层 MoA 最终输出：")
    print(final_output.content)

await run_multi_layer_moa()

何时使用多层 MoA：

• 高价值任务：如公司年度战略报告、重要产品发布文案、核心课程体系设计等，这些任务的成败可能直接影响业务结果，值得投入更多资源。
• 高创造性任务：如品牌故事创作、教学方法创新设计等，需要多轮思想碰撞才能激发出最佳创意的场景。
• 低容错性任务：如法律文书、技术白皮书等，任何错误都可能带来严重后果，需要多层审核来确保准确性。

成本权衡：

多层 MoA 的成本随层数线性增长。以上面的 3 层架构为例：每层 3 个提议者模型 + 1 个聚合器，意味着至少调用 (3+1) + (3+1) = 8 次模型，相比单层的 3+1=4 次，成本增加了一倍多。因此，只有在任务的价值明显高于成本时，才应考虑使用多层 MoA。

你可以通过以下策略来平衡成本和质量：

• 减少提议者数量：第一层使用 3-4 个不同模型以获得多样性，后续层可以减少到 2-3 个模型进行精炼。
• 减少层数：对于大多数任务，2 层 MoA（提议者层 + 聚合层）已经能带来显著的质量提升，3 层或更多层通常只在极高价值任务中才值得使用。
• 混合模型配置：在提议者中混合使用不同性能和成本的模型，而聚合器使用质量最高的模型以确保最终输出的质量。

扩展阅读：工作流的成本优化与资源管理

你已经意识到，像混合专家（MoA）这样的高级模式会带来显著的成本增加。这引出了一个在将任何工作流投入生产前都必须面对的关键问题：如何管理资源和优化成本？幸运的是，你可以通过一系列精细化的工程策略，在不牺牲过多质量的前提下，显著降低工作流的运行成本。

• 差异化模型分配：工作流中的不同节点，其任务复杂度和重要性天差地别。你可以为简单的任务（如意图识别、格式转换）分配轻量、廉价的模型，而只为最关键的核心任务（如最终决策、文案生成）保留昂贵的高级模型。研究表明，通过合理的优化策略，根据具体实施情况，企业可以节省 40-70% 的 Token 成本。
• 系统性缓存：仔细观察你的工作流，你会发现许多节点的计算是可重复的。例如，对同一份公司风格指南的检索、对同一篇课程的审阅请求解析等。通过为这些节点增加缓存机制，你可以存储中间结果。当下次遇到相同的输入时，系统可以直接返回缓存的结果，完全绕过模型调用，从而大幅降低成本和延迟。
• 智能批处理 (Batching)：并非所有任务都需要立即响应。对于课程质量报告生成、用户反馈分析等非实时性任务，你可以设计工作流来智能地聚合一批相似的请求，然后通过一次模型调用进行“批量处理”，而不是为每个请求都单独调用一次。这能在成本和响应时间之间找到一个更优的平衡点。

2.3.5 模式五：人机协作 (Human-in-the-Loop, HITL)

至此，你设计的所有工作流都是全自动的。然而，在现实世界中，将所有决策权完全交给 AI 存在风险，尤其是在处理模糊不清的教学概念或高价值的课程内容时。人机协作模式不再追求完全的自动化，而是有意地在工作流中设计一个或多个“暂停节点”，将控制权交还给人类，由人类进行决策、审批或质量把关后，再将任务交还给工作流继续执行。这是一种构建可信、安全 AI 系统的关键模式。

"""
模式五：人机协作（HITL） - 场景：课程疑难点审核（Human as a Tool）
AI 先对课程中的一个潜在疑难点给出修改建议；随后由具备“人类咨询”工具的改写 Agent 自主决定何时调用该工具向人类请示，最终完成修改。
"""
import asyncio
import os
from agentscope.agent import ReActAgent, UserAgent
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.tool import Toolkit, ToolResponse
from chatbot.agent import create_agent, disable_console_output


# 将人类介入封装为一个工具：ask_human_decision
async def ask_human_decision(question: str) -> ToolResponse:
    """向人类专家征求决策或意见。

    Args:
        question (str): 想要请人类确认或补充的具体问题。
    """
    human_expert = UserAgent(name="教学专家")
    reply = await human_expert(
        Msg(
            "assistant",
            question,
            "assistant",
        )
    )
    return ToolResponse(
        content=[
            TextBlock(type="text", text=reply.get_text_content()),
        ]
    )


async def run_hitl() -> None:
    # AI：给出修改建议
    suggester = create_agent(
        name="疑难点分析师",
        sys_prompt=(
            "你是一名资深教学设计师。请找出课程中对初学者可能最难理解的一个概念，"
            "并提供一个更通俗易懂的解释作为修改建议。"
        ),
        model_name="qwen-plus",
        multi_agent=True,
    )
    disable_console_output([suggester])

    course_content = "在Python中，装饰器本质上是一个接收函数作为参数并返回一个新函数的函数..."
    suggestion = await suggester(Msg("user", course_content, "user"))

    print("AI 建议如下：\n")
    print(suggestion.content)

    # 将“人类介入”作为工具交给改写 Agent，自主决定是否调用
    toolkit = Toolkit()
    toolkit.register_tool_function(ask_human_decision)

    rewriter = ReActAgent(
        name="内容改写器",
        sys_prompt=(
            "你是课程内容改写器。基于提供的 AI 建议完成最终修改。\n"
            "- 若你有把握，请直接完成修改并给出确认信息；\n"
            "- 若存在不确定、歧义或高风险，请调用工具 ask_human_decision 先向人类专家请示，"
            "再据此完成修改；\n"
            "- 在最终结果中简要说明是否咨询了人类及原因。"
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY", "your-api-key"),
            stream=False,
        ),
        formatter=DashScopeMultiAgentFormatter(),
        toolkit=toolkit,
    )
    disable_console_output([rewriter])

    # 将课程内容与 AI 建议一并提供给改写 Agent
    task = (
        "下面是课程摘录与 AI 的修改建议。根据系统提示完成最终修改：\n\n"
        f"[课程内容]\n{course_content}\n\n"
        f"[AI 建议]\n{suggestion.get_text_content()}\n"
    )
    final_action = await rewriter(Msg("user", task, "user"))

    print("=" * 50)
    print("HITL 最终输出：")
    print(final_action.content)
    print("=" * 50)


async def main() -> None:
    await run_hitl()

await main()

人类介入工作流的优势在于：

• 提升准确性：通过引入人类的常识和领域知识来处理 AI 难以判断的教学模糊性（如一个比喻是否恰当），确保最终内容的正确性。
• 增强安全性：对于直接发布课程、修改核心代码库等高风险操作，人类的最终审批是防止 AI 误操作导致严重后果的最后一道防线。
• 建立信任：让课程设计师参与到 AI 的审阅和修改过程中，使其对系统的行为有更强的控制感和信任感。

常见的应用场景有：

• 处理模糊需求：当需求不明确时（如“让课程更有趣”），由 AI 提供多个教学设计方案，人类做出选择。
• 高价值操作审批：在执行任何涉及课程内容发布、删除旧版本等操作前，必须由课程负责人进行审批。
• 关键产出质量审核：在“混合专家”生成一份重要的课程大纲初稿后，工作流的最后一步应是将其发送给教学总监进行最终审核，而不是直接投入开发。

引入人类会显著降低工作流的自动化程度和执行速度。因此，它不适用于追求高吞吐量和毫秒级响应的全自动化场景。HITL 节点的设计需要精心考虑，只在绝对必要的环节介入，避免过多的人工干预拖慢整个流程。

扩展阅读：生产级框架

你无需从零开始实现这些复杂的模式。业界已经有成熟的框架来帮助你构建和管理 Agent 工作流，也内置了异常处理和状态管理工具。

• 代码框架：AgentScope、LangGraph 等库允许你用 Python 代码灵活地定义节点和边，构建任意复杂的图结构工作流，提供了最高的定制化能力。
• 可视化编排平台：阿里云百炼、Dify 等低代码/无代码平台，允许你通过拖拽组件、连接线条的方式，像绘制流程图一样构建工作流。这极大地降低了开发门槛，适合快速原型验证和业务流程相对固定的场景。

2.3.6 选择合适的模式

你已经了解了五种功能各异的工作流模式。一个很自然的问题是：在面对一个具体的业务问题时，我应该如何选择，甚至组合这些模式呢？

记住一个核心原则：没有“最好”的模式，只有“最适合”的模式。你的选择应该由任务的内在属性决定，例如任务的复杂度、子任务间的依赖关系、对成本和效率的要求，以及对结果质量和风险的容忍度。

2.4 总结

让我们回顾一下你在本节学到的知识：

• 单一 Agent 的局限性：面对包含多个步骤的复杂任务，单个 Agent 难以维持稳定的执行计划，容易因“注意力涣散”或“错误累积”而导致任务失败。
• 工作流的核心思想：借鉴“分而治之”的理念，将复杂任务拆解为多个独立的、可管理的节点，并定义它们之间的执行关系，从而确保流程的可靠性。
• 五种核心编排模式：你学习了流水线、分支选择、并行执行、混合专家和人机协作这五种模式，它们分别用于处理固定流程、多选一决策、并行提效、追求质量和引入人工审核的场景。

3 从固定流程到自主规划

3.1 固定工作流的局限性

在上一章，你学习了如何为重复性工作构建固定的工作流。假设你为公司开发了一个“课程前期调研”机器人，它有一个固定的流程：当收到调研需求时，并行分析用户画像、竞品课程和行业需求。

现在，一位课程设计师向机器人发出了指令：“请帮我完成一门新的 Python 入门课程的前期调研。”

机器人忠实地启动了你预设的工作流：

1. 行业需求分析子任务：调用工具，成功。
2. 用户画像定义子任务：调用工具，成功。
3. 竞品课程分析子任务：调用analyze_competitor_course工具，却收到了一个错误：“错误：因竞品网站布局更新，无法解析课程大纲。”

这时，你的机器人将无法继续执行。因为它被设计的流程里，没有处理“竞品分析工具失效”这个意外情况的步骤。

3.2 朴素解法：增加新的分支

你会想，只要为工作流添加异常处理分支就可以了。你可以在原有的固定流程里增加一个分支：如果analyze_competitor_course失败，那就执行一个新的步骤，比如提醒课程设计师人工处理。

这种“打补丁”的方案看似有效，但如果下一次是行业需求分析的API接口临时维护失败了呢？你是不是又要加一个新的分支？如果定义用户画像时，需要的数据源格式变更了呢？还要再加一个分支吗？

你会发现，你永远无法预知所有可能的意外。试图为每一种异常都预设一个处理流程，会让你的工作流变得无比复杂和臃肿，难以维护。更重要的是，一旦出现你没有预料到的新问题，整个系统依然会“卡住”。

根本原因在于，这个 Agent 只是一个流程的忠实执行者，而不是一个问题的解决者。它并不“理解”用户的最终目标是“完成课程前期调研”，它只知道要严格按照你画好的流程图一步步往下走。当其中一条路被堵死时，它不知道如何像人一样，为了达成最终目标而主动寻找别的路。

扩展阅读：目标 (Goal) vs. 任务 (Task)

理解“目标”和“任务”的区别，是理解 Agent 自主规划能力的关键。

• 目标 (Goal)：是用户希望达成的最终状态。它是高层次的、有时甚至是模糊的。例如：“帮我完成Python入门课程的调研”。
• 任务 (Task)：是为了达成目标而需要执行的具体、明确的动作。例如：“调用 analyze_competitor_course 工具，参数为 {url: 'some-site.com' }”。

一个只有固定工作流的 Agent，它处理的是“任务”。当任务执行失败时，它无能为力。而一个更智能的 Agent，应该能聚焦于用户的“目标”。当一个具体的任务失败时，它应该能意识到这只是达成目标的一条路走不通了，然后主动规划出新的任务来继续逼近目标。

3.3 让 Agent 学会自主规划任务

这个过程启发了我们：能否将这种“规划”的权力也赋予 Agent，让它在面对未知挑战时，能自主地设计和调整自己的工作流？

这正是业界解决此类问题的核心理论：规划 (Planning)。

引入规划能力后，Agent 的工作模式发生了根本性转变。开发者的角色从“流程设计师”转变为“目标设定者”与“能力（工具）提供者”，而 Agent 则从“任务执行者”升级为“解决方案规划师”。

它的工作流程变成了这样：

1. 接收目标：Agent 接收一个高阶的用户目标（例如：“完成Python课程的前期调研”）。
2. 动态规划：Agent 的“大脑”（LLM）首先进行思考，将目标分解，动态地生成一份包含多个步骤的行动计划 (Plan)。
3. 执行计划：一个执行程序接收这份计划，然后像执行一份普通工作流一样，按顺序调用工具，完成每个步骤。

在这个新模式下，“计划”本身成了一种可被生成和执行的数据。大模型不再是流程中的一个固定环节，而是流程的创造者。

为了让你更直观地理解 Agent 的自主规划能力，我们提供了一个基于 AgentScope 框架的完整示例。这个示例模拟了文章开头的场景：当竞品分析工具失效时，Agent 如何自主规划并找到解决方案。

"""AgentScope - Agent自主规划与执行示例（简化版）"""
import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit, ToolResponse
from agentscope.plan import PlanNotebook


# 模拟业务工具
async def analyze_competitor_course(url: str) -> ToolResponse:
    """分析竞品课程页面的大纲"""
    # 模拟因网站改版导致解析失败
    return ToolResponse(content=[
        TextBlock(type="text", text=f"❌ 错误：因 {url} 网站布局更新，无法解析课程大纲。")
    ])

async def search_industry_demand(topic: str) -> ToolResponse:
    """查询行业的技能需求"""
    return ToolResponse(content=[
        TextBlock(type="text", text=f"✅ 报告：关于“{topic}”的行业需求分析已完成。")
    ])

async def google_search(query: str) -> ToolResponse:
    """谷歌网页搜索"""
    if "syllabus" in query:
        return ToolResponse(content=[
            TextBlock(type="text", text="搜索结果：找到了'Python入门课程'的大纲PDF，地址 a.com/syllabus.pdf")
        ])
    return ToolResponse(content=[TextBlock(type="text", text="未找到相关信息")])

async def extract_text_from_pdf(url: str) -> ToolResponse:
    """从PDF链接中提取文本"""
    return ToolResponse(content=[
        TextBlock(type="text", text=f"✅ 已从 {url} 提取大纲文本：1. 变量与数据类型... 2. ...")
    ])


# 用于监控计划变化的钩子函数
plan_snapshots = []

def capture_plan_snapshot(notebook, plan):
    """捕获计划快照"""
    if plan:
        plan_snapshots.append({
            "name": plan.name,
            "description": plan.description,
            "state": plan.state,
            "subtasks": [
                {
                    "name": st.name,
                    "state": st.state,
                    "outcome": st.outcome
                }
                for st in plan.subtasks
            ]
        })


async def main():
    
    print("=" * 60)
    print("🤖 Agent自主规划演示")
    print("=" * 60)
    
    # 创建PlanNotebook并注册钩子
    plan_notebook = PlanNotebook()
    plan_notebook.register_plan_change_hook("capture", capture_plan_snapshot)
    
    # 创建工具箱
    toolkit = Toolkit()
    toolkit.register_tool_function(analyze_competitor_course)
    toolkit.register_tool_function(search_industry_demand)
    toolkit.register_tool_function(google_search)
    toolkit.register_tool_function(extract_text_from_pdf)
    
    # 创建Agent
    agent = ReActAgent(
        name="CourseResearcherAgent",
        sys_prompt=(
            "你是课程调研助手。遇到复杂任务时：\n"
            "1. 用create_plan创建计划\n"
            "2. 逐步执行，用finish_subtask标记完成\n"
            "3. 遇到问题灵活调整，例如使用google_search寻找替代方案\n"
            "4. 完成后用finish_plan结束"
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        plan_notebook=plan_notebook,
    )
    
    # 用户请求
    print("\n💬 用户: 请帮我完成一门新的 Python 入门课程的前期调研。\n")
    print("-" * 60)
    
    await agent(Msg("user", "请帮我完成一门新的 Python 入门课程的前期调研，竞品是 some-site.com 的课程。", "user"))
    
    # 显示结果（从快照中获取最后的完整计划）
    print("\n" + "=" * 60)
    print("📊 执行结果")
    print("=" * 60)
    
    if plan_snapshots:
        final_plan = plan_snapshots[-1]
        finished = sum(1 for st in final_plan["subtasks"] if st["state"] == "finished")
        
        print(f"\n✅ 计划: {final_plan['name']}")
        print(f"📊 进度: {finished}/{len(final_plan['subtasks'])}")
        print(f"🎯 状态: {final_plan['state']}\n")
        
        print("子任务详情:")
        for i, subtask in enumerate(final_plan["subtasks"], 1):
            icon = "✅" if subtask["state"] == "finished" else "⏳"
            print(f"  {icon} {i}. {subtask['name']}")


await main()

通过这个示例，你可以看到：

1. 自主创建计划：Agent 使用 create_plan 工具自动规划调研任务。
2. 灵活执行：遇到竞品分析工具失效时，Agent 自动调整策略，转而使用 google_search。
3. 进度追踪：使用 finish_subtask 标记完成的任务。
4. 完整闭环：从规划创建到任务完成的全流程。

这正是 PlanNotebook 为 Agent 带来的核心能力：将其从"流程执行者"提升为"问题解决者"。

扩展阅读：生产级框架

像 AgentScope 和 LangChain 这样的开源框架，都提供了实现这种“规划-执行”循环的机制。它们允许你定义一系列工具，然后让大模型作为规划器 (Planner) 来决定在每一步应该调用哪个工具，并将工具返回的结果作为后续思考的输入，从而实现复杂的任务拆解和执行。在阿里云机器学习平台 PAI 上，你可以方便地部署和管理这些框架所需的大模型服务，为 Agent 提供强大的“大脑”。

3.4 执行 Agent 生成的规划

那么，如何让大模型生成一份机器可以理解和执行的“计划”呢？

最简单的方式，是让它生成自然语言的步骤列表。但这样做，下游的执行程序很难精确解析。你之前在让 Agent 调用工具时学过，可以使用结构化的 JSON 格式 输出工具调用参数。这里，你也可以把“执行计划”看作调用工具。每一个步骤都是一个定义清晰的对象，包含要调用的工具名和对应的参数。

{
  "plan": [
    {
      "step": 1,
      "thought": "我首先需要分析行业需求，这是课程定位的关键。",
      "tool_name": "search_industry_demand",
      "tool_params": {"topic": "Python 基础"}
    },
    {
      "step": 2,
      "thought": "接下来，我尝试分析竞品课程的大纲。",
      "tool_name": "analyze_competitor_course",
      "tool_params": {"url": "some-site.com/python-course"}
    }
  ]
}

这是一种有效的方法，但它的表达能力有限。如果计划中需要包含“如果竞品分析失败，则改用谷歌搜索”这样的条件逻辑，简单的 JSON 列表就难以胜任了。

为了表达更复杂的逻辑，你可以让大模型直接生成代码 (Code as Action) 来表达其计划，再通过调用“代码解释器”这个工具来执行代码。

# Plan generated by LLM
def execute_research_plan():
    # Step 1: Analyze industry demand
    demand_result = search_industry_demand(topic="Python 基础")
    print(demand_result)

    # Step 2: Analyze competitor course
    competitor_result = analyze_competitor_course(url="some-site.com/python-course")

    # Step 3: Handle analysis failure
    if not competitor_result.success and "无法解析" in competitor_result.message:
        print("竞品分析工具失效，正在寻找备选方案...")
        search_results = google_search(query="some-site.com python course syllabus")
        # Assume search_results gives a PDF link
        pdf_url = extract_pdf_link(search_results) 
        if pdf_url:
            syllabus_text = extract_text_from_pdf(url=pdf_url)
            print(syllabus_text)
    else:
        print(competitor_result)

execute_research_plan()

通过生成代码，大模型可以利用编程语言内置的丰富能力（如变量、条件判断、循环）和强大的第三方库（如 Pandas）来制定和执行极其复杂的计划。这使得 Agent 不仅能应对简单的线性流程，还能处理包含逻辑判断和数据处理的复杂场景。

3.5 进阶：让 Agent 创建新工具

你已经掌握了让大模型通过生成代码来制定计划的强大方法。这种方式赋予了 Agent 运用变量、条件判断和循环等复杂逻辑的能力。

但这里还有一个潜在的瓶颈：Agent 仍然受限于你预先提供给它的工具集。如果它在执行计划时，发现需要一个你并未提供的新工具，比如一个用于计算不同技术关键词在招聘网站上出现频率的函数，它该怎么办？

最直接的办法是让 Agent 停下来，请求你（开发者）为它编写这个新工具。但这中断了任务的自主流程。让我们更进一步思考：既然 Agent 已经能够生成用于规划的代码，它是否也能生成用于创造新能力的代码呢？

这便引出了一种更高级的规划能力：动态创造工具 (Dynamic Tool Creation)。

要实现这一点，你需要在提供具体业务工具（如 analyze_competitor_course）之外，再为 Agent 提供一个核心的“元工具”：代码执行环境 (Code Interpreter)。

当 Agent 识别到现有工具无法满足需求时，它的规划会包含一系列特殊的步骤：

1. 决策：大模型分析任务，识别出需要一个当前不存在的新工具。
2. 生成代码：在它的计划中，它会编写一段代码来定义、测试并封装一个新的工具函数。
3. 调用新工具：在新工具于代码执行环境中被成功创建后，Agent 可以在后续的计划步骤中直接调用它，就好像这个工具一开始就存在一样。
4. 扩展工具库：这个新生成的工具可以被加入到本次任务的临时工具库中，供后续步骤复用。

下面是使用 AgentScope 构建的一个简单示例，Agent 可以在规划过程中，基于已有的 add 工具，自主创建并注册一个 factorial（阶乘）工具，并在后续任务中调用它。

"""AgentScope Agent自主创建工具 - 精简版"""
import asyncio
import os
import sys
from io import StringIO

from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit, ToolResponse

# 全局工具箱
toolkit = None


async def code_exec(code: str) -> ToolResponse:
    """代码解释器 - 用于创建和注册新工具"""
    global toolkit
    
    namespace = {
        'ToolResponse': ToolResponse,
        'TextBlock': TextBlock,
        'asyncio': asyncio,
        'agent_toolkit': toolkit,
        'math': __import__('math'),
    }
    
    stdout, sys.stdout = sys.stdout, StringIO()
    
    try:
        exec(code, namespace)
        output = sys.stdout.getvalue()
        sys.stdout = stdout
        return ToolResponse(content=[TextBlock(
            type="text", 
            text=output or "✅ 执行成功"
        )])
    except Exception as e:
        sys.stdout = stdout
        return ToolResponse(content=[TextBlock(
            type="text",
            text=f"❌ 错误: {e}"
        )])


async def add(a: float, b: float) -> ToolResponse:
    """加法工具"""
    return ToolResponse(content=[TextBlock(
        type="text", 
        text=f"{a} + {b} = {a + b}"
    )])


async def main():
    if "DASHSCOPE_API_KEY" not in os.environ:
        print("❌ 请设置 DASHSCOPE_API_KEY")
        return
    
    global toolkit
    toolkit = Toolkit()
    toolkit.register_tool_function(add)
    toolkit.register_tool_function(code_exec)
    
    agent = ReActAgent(
        name="ToolMaker",
        sys_prompt=(
            "你可以通过 code_exec 创建新工具。\n"
            "模板:\n"
            "async def tool_name(param: type) -> ToolResponse:\n"
            "    '''描述'''\n"
            "    result = ...\n"
            "    return ToolResponse(content=[TextBlock(type='text', text=f'{result}')])\n"
            "agent_toolkit.register_tool_function(tool_name)\n"
            "print('✅ 已注册 tool_name')"
        ),
        model=DashScopeChatModel(
            model_name="qwen-plus",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        memory=InMemoryMemory(),
    )
    
    print("=" * 60)
    print("🚀 Agent 自主创建工具演示")
    print("=" * 60)
    
    # 使用现有工具
    print("\n▶️ 场景1: 使用现有工具")
    await agent(Msg("user", "计算 30 + 45", "user"))
    
    # 创建新工具
    print("\n▶️ 场景2: 创建阶乘工具")
    await agent(Msg("user", "创建 factorial 工具计算阶乘", "user"))
    
    # 使用新工具
    print("\n▶️ 场景3: 使用新工具")
    await agent(Msg("user", "用 factorial 计算 5 的阶乘", "user"))
    
    # 显示工具箱
    print("\n📦 最终工具箱:")
    for i, s in enumerate(toolkit.get_json_schemas(), 1):
        print(f"{i}. {s['function']['name']}")


if __name__ == "__main__":
    await main()

通过提供代码解释器，你将 Agent 从一个单纯的工具使用者，提升为了一个工具创造者。它的能力边界不再被你预先定义的工具集所束缚，从而具备了真正的创造性和问题解决的适应性。

3.6 何时选择自主规划？

你已经了解了“固定工作流”和“自主规划”两种模式，你可能会问：我是不是应该在所有场景下都使用更智能的自主规划，彻底放弃固定工作流？

这样想是不对的。自主性更高的模式不是银弹，自主性更低的模式也有广泛的应用场景。在生产实践中，一种非常有效的最佳实践是采用**“探索-固化”混合模式**。

这种模式将任务处理分为两个阶段：

1. 探索阶段：对于新出现的、流程不明确的任务（例如，你需要调研一个全新的、之前从未接触过的小众技术领域），你无法预先定义一个完美的流程。这时，就应该派出自主规划 Agent。它的任务是探索解决问题的不同路径，调用它认为合适的工具，即便过程中会犯错或走到死胡同，最终的目标是找到一条能稳定解决问题的方案。

2. 固化阶段：当自主规划 Agent 经过多次探索，验证并总结出一条稳定、高效的解决方案路径后（例如，它发现“先用A工具从特定网站爬取信息，再用B工具进行数据清洗，最后用C工具生成总结报告”的流程成功率最高），你就可以将这条被验证过的路径抽象并固化下来，封装成一个可靠的“固定工作流”，用于后续大规模、重复性的生产调用。

这样，你就建立了一个持续优化的正向循环。

预告：在第 7 章"Agent Skills"中，你将通过一个真实的教程审查任务， 亲身体验自主 Agent 的能力边界，并学习如何将专家知识封装为 Skill 来突破这些限制。

3.7 案例分析：让 Agent 操作网页

为了让你更具体地理解这种“感知-规划-行动”循环在实际产品中的应用，让我们来看一个高自主性网页操作 Agent 的案例，例如开源项目 Browser Use。

传统的网页自动化（RPA）工具需要为每个网站、每个任务编写固定的操作脚本。一旦网站界面稍有改动，脚本就会失效，维护成本极高。

一个具备规划能力的 Agent 则可以从根本上解决这个问题。它不依赖固定的脚本，而是像人一样理解用户的目标，并感知当前的网页状态，动态地规划出下一步操作。

执行流程拆解： 当用户给出指令 “在亚马逊上搜索关于 AI 的书籍” 时：

1. 理解与初步规划：LLM 将模糊的目标分解为一系列高阶步骤：“1. 打开亚马逊网站；2. 找到搜索框；3. 输入'AI书籍'；4. 点击搜索；5. 分析结果。”
2. 行动与感知：Agent 执行第一步（打开网站）。然后它“感知”新页面——这不仅是看 HTML 代码，还可能包括分析截图的视觉布局，来理解页面上有哪些元素。
3. 决策与再规划：基于感知到的信息，它决策下一步行动：找到那个看起来最像“搜索框”的输入区域。如果页面上有多个输入框，它会根据位置、标签等信息进行推理判断。
4. 循环执行：它持续这个“感知-规划-行动”的循环，直到完成所有步骤，并返回搜索到的书籍列表。
5. 异常处理：如果在任何一步遇到意外，比如点击搜索后弹出一个验证码，它不会卡住。它会感知到这个新情况，并将“处理验证码”作为一个新的障碍插入到当前计划中，尝试解决它或向用户求助。

这个案例完整地展示了规划型 Agent 的核心优势：它不再是脚本执行器，而是通过持续的“感知-规划-行动”循环，实现了对动态、未知网页环境的真正自适应操作。

3.8 总结

让我们回顾一下你在本节学到的知识：

• 固定工作流的局限性：面对如“工具失效”等预期外的障碍时，预先设定的固定流程会“卡住”，因为它缺乏适应性。简单地为每个意外增加分支会让流程变得复杂且难以维护。

• “规划”是核心解法：模仿人类解决问题的方式，我们让 Agent 从“任务执行者”升级为“解决方案规划师”。它不再被动执行固定步骤，而是围绕用户的最终“目标”，自主地动态生成和调整行动计划。

• 将计划转化为行动：你可以引导大模型生成结构化的计划（如JSON或代码），让机器精确执行。更高阶的方法是为 Agent 提供代码解释器，使其能在规划中动态创造并使用新工具，突破预设能力的限制。

• 在稳定与灵活间权衡：“自主规划”适用于探索未知、流程多变的创新任务，而“固定工作流”则保障核心业务的稳定和效率。在生产环境中，你需要结合人机协作、探索-固化等策略，根据具体业务需求，在这两种模式间做出明智的选择。

🔥 课后小测验

🔍 单选题 3.2.1

你开发了一个"课程前期调研"Agent，使用固定工作流并行调用三个工具：行业分析、用户画像、竞品分析。上线后竞品网站改版导致爬虫工具频繁失败，整个流程卡死。你该如何改进❓

• A. 为竞品分析工具增加重试机制，失败时自动重试三次后报错退出
• B. 在工作流中为每个工具失败场景都预设分支处理逻辑和降级方案
• C. 将固定工作流改为Plan & Execute模式，让Agent根据目标自主规划并动态应对异常
• D. 更换为更稳定的竞品数据API服务商，从数据源层面消除工具失败的可能

【点击查看答案】

✅ 参考答案：C 📝 解析： A只是延迟失败，网站改版后重试也无法成功。B方向对但手段弱，为每种异常预设分支会让工作流复杂度急剧膨胀，且无法穷举所有意外。D是理想化方案，现实中无法保证任何外部服务永远可用。C是正解：Plan & Execute模式让Agent从"执行者"变为"规划者"，面对工具失败时能自主调整计划（如改用其他信息源或跳过该步骤），从架构层面获得适应性。

✉️ 评价反馈

欢迎你参与阿里云大模型ACP课程问卷 (opens new window) 反馈学习体验和课程评价。 你的批评和鼓励都是我们前进的动力！