一、框架概览

1. 框架概览

• 框架背景：AutoGen是由微软推出的开源框架，专为构建基于大型语言模型的复杂应用而设计。它改变了传统AI应用开发的方式，不再依赖单一的‘超级英雄’式智能体，而是通过多个智能体的协作来解决复杂任务。
• 核心思想：其核心思想是通过智能体间的对话与协作来动态驱动任务的解决。这种模式将任务分解为多个子任务，并分配给不同的智能体，每个智能体专注于特定的角色和任务，从而实现高效的协作。
• 架构优势：AutoGen框架的出现是对早期单一、巨型智能体设计局限性的一种回应。它通过任务分解和角色专业化，优化了问题解决的效率，提升了系统的灵活性和可扩展性。

2. 与LangChain和SK对比

• LangChain与Semantic Kernel：LangChain通过模块化和组件化编排任务逻辑，形成‘链’或‘图’；Semantic Kernel则侧重于将大型语言模型的能力集成到企业级应用中，通过‘内核’与‘插件’动态编排执行计划。
• AutoGen的独特性：AutoGen的核心抽象是‘对话’，通过智能体间的协作与对话隐式地驱动任务解决。开发者的工作重心从编排任务逻辑转向设计智能体团队，这种方式更加灵活，能够适应复杂的任务需求。

3. 设计理念深度解析

• 以对话为中心：AutoGen的设计理念是‘以多智能体对话为中心’，对话不仅是信息交换的形式，更是整个应用框架的计算核心。通过对话结构，开发者可以管理复杂的控制流，包括条件逻辑、循环和错误处理。
• 动态执行图：对话被抽象为动态的执行图，开发者可以通过设计智能体的角色、目标和沟通规则来间接引导应用程序的行为，这种方式更灵活、更具适应性。
• 决策与推理的转移：这种设计理念将应用程序的决策和推理部分从代码转移到对话本身，充分发挥了大型语言模型的自然语言理解和生成能力。
• 灵活性与适应性：开发者通过设计智能体的角色和沟通规则来引导应用程序的行为，这种方式不仅灵活，还能更好地适应不同的任务需求，契合了大型语言模型的特性。

4. 分层架构设计

• 核心API
  • 功能：核心API是AutoGen框架的基石，提供最基础、最通用的构建模块，关注底层消息传递机制、智能体的生命周期管理以及对本地与分布式运行时的支持。
  • 使用场景：当开发者需要构建全新的智能体交互模式或将其集成到已有的事件驱动系统中时，核心API提供了必要的灵活性和控制力。
• 智能体对话API
  • 优势：智能体对话API是一个面向应用开发者的高级API，封装了底层核心API的复杂性，提供直观的默认实现和预设模式，支持常见的协作模式，如双智能体对话和群聊。
  • 开发便利性：使用智能体对话API，开发者可以通过配置组件快速开发应用，利用预设的智能体类和成熟的团队协作模式，大大提高了开发效率。
  • 局限性：在需要高度定制化交互逻辑或与复杂外部系统深度集成时，智能体对话API可能会受到限制，此时需要使用更底层的核心API。
• 扩展API
  • 作用：扩展API是一个能力扩展库，提供具体的模型客户端、工具等可插拔功能。开发者可以从扩展库中导入和使用这些功能，以增强智能体的能力。
  • 保持轻量与拓展能力：扩展API的设计保障了AutoGen框架本身保持轻量和核心稳定，同时通过不断丰富的扩展库来拓展其能力边界，满足不同开发者的需求。

5. 生态系统

• 生态系统组件角色
  • AutoGen Studio：AutoGen Studio是一个无代码/低代码图形界面，用于快速原型设计和测试多智能体工作流，适合希望快速验证想法的开发者。
  • Core：Core是底层API，用于构建自定义、可扩展和高性能的智能体系统，适合需要高度定制化的开发者。
  • AgentChat：AgentChat是高级API，用于快速开发常见的多智能体对话应用，封装了底层复杂性，适合希望快速构建应用的开发者。
  • Extensions：Extensions是一个插件库，提供具体的模型客户端、工具等可插拔功能，适合需要扩展智能体能力的开发者。

组件	主要功能	目标用户	架构层次
AutoGen Studio	无代码/低代码图形界面，用于快速原型设计和测试多智能体工作流	产品经理、AI应用设计师、快速原型开发者	UI层(高层)
AgentChat	高级API，用于快速开发常见的、任务驱动的多智能体对话应用	AI应用开发者、大多数AutoGen用户	智能体对话API 层(高层)
Core	底层AP，事件驱动的框架，用于构建自定义、可扩展和高性能的智能体系统	系统架构师、框架开发者、需要极致控制力的专家	核心AP
Extensions	插件库，提供具体的模型客户端、工具等可插拔功能	所有开发者(作为库使用)	扩展API层(通用层)

• AgentChat与Core关系
  • 便利性与控制力的权衡：AgentChat通过其‘有偏见’的设计为开发者提供便利性，大部分编排逻辑已被封装在预置组件中，适合快速开发常见应用。
  • 底层控制能力：Core作为‘无偏见’的底层API，将编排逻辑的控制权完全交给开发者，允许实现任何交互模式或集成到外部事件驱动架构中。

6. 学习路径建议

1. 从AgentChat入手：从AgentChat组件入手，重点理解通过‘对话’组织和驱动工作流的方式，特别是群聊和顺序对话等高级交互模式。
2. 探索Extensions：探索Extensions组件，学习如何为智能体添加自定义工具和配置不同模型客户端，提升智能体的能力。
3. 深入Core：在必要时深入Core组件，当项目需求超出AgentChat的抽象范围时，学习核心API的底层控制能力，实现高度定制化。

7. 总结与展望

• 核心优势：AutoGen框架的核心优势在于以对话为中心的运作方式、灵活的分层架构和强大的动态多智能体协作模式，突破了传统单智能体设计的瓶颈。
• 应用场景：AutoGen为构建复杂AI应用提供了全新的思路和方法，适用于多种需要动态协作的场景。
• 未来发展：展望AutoGen在未来AI应用开发中的潜力，鼓励开发者深入学习和探索，利用其构建更多创新的AI应用，推动人工智能技术的发展。

二、AgentChat

1. 基础概念

• 可对话智能体的定义：可对话智能体是一个能够发送和接收消息并基于消息产生回应的实体，是状态化和消息驱动的。
• 状态化：状态化意味着智能体能够在多次交互中维持内部状态，如对话历史或任务理解，从而处理长周期的连贯任务。
• 消息驱动：消息驱动表示智能体的所有行为都是由接收到的消息触发的，它可以生成自然语言回复、调用外部工具或请求人类用户反馈。
• 可对话智能体:具备了参与结构化消息交换的能力

2. 智能体内部机制

• 内部机制：智能体接收到消息后，会处理传入消息列表，结合内部对话历史理解上下文，基于预设角色、工具集和当前状态决定下一步行动，并生成出站消息作为回应。
• 可对话的抽象：‘可对话’是一种强大的抽象，将智能体的内部业务逻辑与外部通信协议解耦，使得智能体只需专注于如何根据对话历史回应，而不需关心通信细节。

3. 预设智能体角色

• 助手智能体：助手智能体是执行任务的核心力量，由大语言模型驱动，其行为由系统提示和工具集定义。
• 用户代理智能体：用户代理智能体是人类用户的代理人，负责发起对话并将用户请求传递给其他智能体，同时作为人机交互的桥梁，能够代表用户执行代码并反馈结果。
• 协作模式：这两种智能体在一问一答中构成AutoGen中最经典的双智能体协作模式，是应用中的重要基础。

4. 双智能体对话模式

• 工作机制：双智能体对话模式模拟两个人之间的一对一交流，由一个智能体发起对话，将初始任务描述发送给另一个智能体，接收方根据配置生成回复并发送回发起方，形成轮流发言的循环。
• 终止条件：对话循环会在满足预设终止条件时结束，如达到最大轮次或包含特定终止关键词。
• 基本原理：这种模式完整地体现了AutoGen对话驱动的基本原理，即通过消息传递驱动任务进展，并通过终止条件控制流程结束。

5. 顺序对话模式

• 特点：顺序对话是一连串双智能体对话的链式组合，前一个对话的输出成为后一个对话的输入，通过‘接续信息’实现任务上下文的传递。
• 优势：这种模式适合步骤固定、依赖关系明确的复杂任务，如内容创作流程，能够确保最终产出物的连贯性和高质量。

6. 群聊对话模式

• 工作原理：群聊允许多个智能体在一个共享对话空间中交流，发言顺序动态决定，引入群聊管理器作为核心角色。
• 群聊管理器：群聊管理器集中控制对话流程，决定每次消息广播后由谁发言，通过智能选择发言者引导对话朝着解决问题的方向发展。
• 优势：群聊模式适用于需要集体智慧解决的复杂任务，能够实现动态和自适应协作，突出其与双智能体对话和顺序对话模式的区别和优势。

7. 快速入门示例

1. 准备环境：通过命令行安装，使用PyCharm，在终端窗口执行命令

   pip install -U "autogen-agentchat" "autogen-ext[openai]"
   

2. 项目创建与配置：通过具体代码示例介绍如何创建项目、配置环境、安装依赖、导入模块，并定义模型客户端和工具。

3. 智能体定义：详细说明如何实例化智能体，设置智能体的系统指令和工具列表，为后续任务做好准备。

4. 任务启动：展示如何通过 agent.run_stream 启动对话，并使用Console工具观察对话和工具调用过程。

5. 任务完成：通过运行脚本展示Agent如何接收任务、调用工具并生成最终答案，完成一个简单的天气查询任务。

import os
import dotenv
dotenv.load_dotenv()

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-latest",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,             # 通义千问不支持视觉
        "function_calling": True,    # 函数调用
        "json_output": True,
        "structured_output": True,
        "family": "qwen",            # 可以任意名字
    }
)

# 定义一个 Agent 可以使用的简单函数工具
async def get_weather(city: str) -> str:
    """获取给定城市的天气。"""
    print(f"--- 正在调用工具 get_weather(city='{city}') ---")
    return f"天气预报：{city} 现在是 23 摄氏度，晴天。☀️"

# 定义一个 AssistantAgent
agent = AssistantAgent(
    name="weather_agent",          # 代理的名字
    model_client=model_client,     # 代理使用的 LLM 客户端
    tools=[get_weather],           # 代理可以使用的工具列表
    system_message="你是一个乐于助人的AI助手。", # 给代理的系统指令
    reflect_on_tool_use=True,      # 允许代理在使用工具后进行反思
    model_client_stream=True,      # 启用流式输出
)

# 定义主执行函数
async def main() -> None:
    # 启动代理，给它一个任务，并将输出流式传输到控制台
    await Console(agent.run_stream(task="杭州现在天气怎么样？"))

    # 任务完成后，关闭与模型客户端的连接
    await model_client.close()

# 运行主函数
import asyncio

if __name__ == "__main__":
    asyncio.run(main())

8. 多智能体入门示例

1. 创建专家智能体：介绍如何创建专家智能体（如天气专家和开发专家），并强调 description 参数的重要性，它帮助其他智能体了解其能力并决定何时调用。
2. 主控智能体：展示如何将智能体包装成工具，创建主控智能体，并通过主控智能体调用专家工具来解决问题。
3. 协作流程：通过具体例子（如查询武汉天气）展示多智能体协作的流程，突出AutoGen框架如何通过智能体间的对话和协作完成复杂任务。

import os
import dotenv
dotenv.load_dotenv()

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.tools import AgentTool

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-latest",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "structured_output": True,
        "family": "qwen",
    }
)

# 定义一个 Agent 可以使用的简单函数工具
async def get_weather(city: str) -> str:
    """获取给定城市的天气。"""
    print(f"--- 正在调用工具 get_weather(city='{city}') ---")
    return f"天气预报：{city} 现在是 23 摄氏度，晴天。☀️"

weather_agent = AssistantAgent(
        name="weather_agent",
        model_client=model_client,
        system_message="你是一个天气预报员。",
        description="一个负责天气预报的专家",
        tools=[get_weather],
        model_client_stream=True,
    )

develop_agent = AssistantAgent(
    name="develop_agent",
    model_client=model_client,
    system_message="你是一个经验丰富的 Python 程序员，性格沉默寡言，说话总是很简洁。",
    description="一个擅长 Python 的软件开发专家",
    model_client_stream=True,
)

weather_agent_tool = AgentTool(weather_agent, return_value_as_last_message=True)
develop_agent_tool = AgentTool(develop_agent, return_value_as_last_message=True)

agent = AssistantAgent(
    name="assistant",
    system_message="你是一个通用助理，需要时使用专家工具。",
    model_client=model_client,
    model_client_stream=True,
    tools=[weather_agent_tool, develop_agent_tool],
    max_tool_iterations=10
)

# 定义主执行函数
async def main() -> None:
    # 启动代理，给它一个任务，并将输出流式传输到控制台
    await Console(agent.run_stream(task="Python异步方法怎么用？"))

    # 任务完成后，关闭与模型客户端的连接
    await model_client.close()

# 运行主函数
import asyncio

if __name__ == "__main__":
    asyncio.run(main())

三、团队模式

1. 团队的定义与优势

• 团队与多智能体协作：AgentChat 团队是一组智能体集合，它们为了达成共同目标而协同工作。这种团队模式通过多智能体协作，能够有效克服单一智能体的性能瓶颈，实现任务的高效分解和模块化，提升整体功能的高内聚性。
• 多智能体协作优势：多智能体协作具有显著优势，如灵活性和效率。通过任务分解，每个智能体专注于特定任务，避免了单一智能体在复杂任务中可能出现的性能瓶颈。这种协作方式能够更好地应对复杂多变的任务需求。
• AutoGen 团队模式：AutoGen 中的团队模式是一种结构化的、可通过编程控制的对话框架。它简化了复杂工作流程的编排和优化，使得开发者能够更高效地管理和调度多智能体协作，提升开发效率和系统性能。

2. 单一智能体的局限性

• 单一智能体的困境：单一智能体在面对复杂任务时存在明显局限性。例如，在规划、执行、观察和反思等多个阶段难以兼顾，导致系统提示臃肿和混乱。在编写代码、调试代码和验证执行结果时，单一智能体往往力不从心。
• 职责分离的重要性：将复杂任务中的不同职责分离给多个智能体是解决单一智能体局限性的关键。通过多智能体协作，每个智能体可以专注于特定任务，从而提高整体效率和准确性。实际案例表明，多智能体协作能够有效解决单一智能体在复杂任务中的表现不佳的问题。

3. 基础团队模式：轮询式小组对话

• 轮询式小组对话原理：轮询式小组对话的工作原理是所有成员共享同一个对话上下文，并按顺序轮流发言。这种模式的核心是‘对话即状态’，即共享的对话历史记录本身就是团队的共享状态。其优点是完全可预测，但缺点是缺乏灵活性。
• 适用场景：轮询式小组对话适用于流程固定、步骤明确的任务。例如，在调试复杂对话流程或需要严格控制流程的场景中，这种模式能够确保任务按预定顺序执行，便于管理和监控。
• 局限性与对比：然而，在动态任务中，轮询式小组对话的灵活性不足。与其他模式相比，它更适合简单、结构化的任务，而在复杂、动态的任务中可能需要更灵活的协作方式。
• 示例

1. 环境准备：在实现轮询式小组对话之前，需要进行环境准备，包括安装必要的库和配置运行环境。确保所有依赖项正确安装，为后续的团队创建和运行提供支持。
2. 创建团队与运行：通过代码创建包含两个助手 Agent 的团队，模拟内容创作和审核流程。定义团队成员的角色和职责，并设置终止条件。运行团队后，观察每个 Agent 按顺序发言的过程。
3. 代码实现与结果：展示代码片段，详细解释每一步的作用和意义。运行结果显示团队成员按预定顺序发言，完成内容创作和审核任务，验证了轮询式小组对话的有效性。

import asyncio
import os
import dotenv

dotenv.load_dotenv()

from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-max",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "structured_output": True,
        "family": "qwen",
    }
)

# 创建创作者智能体
# 它的任务是成为一个有帮助的人工智能助手
primary_agent = AssistantAgent(
    name="primary_agent",
    model_client=model_client,
    model_client_stream=True,
    system_message="你是一位创作者，负责根据要求创作内容。"
)

# 创建评论家智能体
# 它的任务是提供建设性反馈，并在反馈被采纳后，回复 "APPROVE" 来结束任务
critic_agent = AssistantAgent(
    name="critic_agent",
    model_client=model_client,
    model_client_stream=True,
    system_message="你的职责是提供有建设性的反馈意见。只有当你的反馈意见被采纳后，才会回复APPROVE，否则不要回复APPROVE。",
)

# 定义一个文本提及终止条件
# 当任何消息中出现 "APPROVE" 这个词时，任务就会停止
text_termination = TextMentionTermination("APPROVE")

# 将创作者和评论家智能体放入一个列表中
# 列表中的顺序决定了它们发言的顺序：创作者先说，然后是评论家
agents = [primary_agent, critic_agent]

# 创建一个轮询式小组对话团队
# 传入智能体列表和终止条件
team = RoundRobinGroupChat(
    agents,
    termination_condition=text_termination
)

# 定义一个异步主函数来运行团队
async def main():
    # 使用 Console 辅助类来格式化并打印流式输出
    # 我们给团队分配一个初始任务：“写一首关于秋天的短诗”
    await Console(team.run_stream(task="写一首关于秋天的唐诗。"))

# 运行异步主函数
# 在 Python 脚本中，我们通常使用 asyncio.run() 来启动顶层的异步函数
if __name__ == "__main__":
    asyncio.run(main())

4. 团队的生命周期管理

团队生命周期管理工具将 AutoGen 团队从黑盒化的多智能体交互过程转变为可观察、可干预、可持久化的计算状态机。开发者可以通过这些工具实现对团队的精细控制，并将其无缝集成到复杂的企业级应用中。

• 重置方法：团队的重置方法能够清空团队及智能体的内部状态，恢复到初始状态。这可以避免历史对话对新任务的干扰，确保每次运行都是独立的。

await Console(team.run_stream(task="写一首关于秋天的唐诗。"))
await team.reset()	# 通过reset方法，让状态恢复到初始，避免干扰后面的任务
await Console(team.run_stream(task="写一首关于夏天的故事。"))

• 外部停止：通过外部终止条件可以优雅地停止团队运行。这种功能在需要提前结束任务时非常有用，能够确保团队在合适的时间点停止。

# 导入外部终止条件类
from autogen_agentchat.conditions import ExternalTermination

# 定义一个文本提及终止条件
# 当任何消息中出现 "APPROVE" 这个词时，任务就会停止
text_termination = TextMentionTermination("APPROVE")

# 创建一个外部终止条件实例
external_termination = ExternalTermination()

# 传入智能体列表和终止条件
team = RoundRobinGroupChat(
    agents,
    termination_condition=text_termination | external_termination,
)

async def main():
    # 在一个单独的任务中运行团队
    run_task = asyncio.create_task(
        Console(team.run_stream(task="写一首秋天的唐诗"))
    )

    # 等待一小段时间，模拟一些外部事件的发生
    await asyncio.sleep(5)

    # 从外部代码中设置停止信号
    print("\n[INFO] 从外部发送停止信号...\n")
    external_termination.set()
    
    await run_task

• 恢复运行：一个已停止的团队可以从停止的地方继续运行。这一特性在人机协作场景中非常强大，能够确保任务在中断后可以无缝恢复。

• 中止方法：通过取消令牌可以实现任务的即时中止。这种机制在需要快速停止任务时非常有效，能够避免不必要的资源浪费。

# 导入取消令牌类
from autogen_core import CancellationToken

async def main():
    # 创建一个取消令牌
    cancellation_token = CancellationToken()

    # 在另一个任务中运行团队，并传入取消令牌
    run_task = asyncio.create_task(
        Console(team.run_stream(
            task="写一首秋天的唐诗",
            cancellation_token=cancellation_token)
        )
    )
    
    # 模拟在某个时刻决定中止任务
    await asyncio.sleep(5)
    print("\n[INFO] 发送取消信号，立即中止任务...\n")
    cancellation_token.cancel()

    # 捕获并处理取消错误
    try:
        await run_task
    except asyncio.CancelledError:
        print(" 任务已成功被中止。")

四、高级团队模式

1. 选择式小组对话

• 工作原理：选择式小组对话通过大语言模型的‘选择器’动态决定下一个发言的智能体。选择器的决策依据包括对话历史、候选智能体的名称和描述，能够根据上下文灵活选择合适的智能体。
• 灵活性与适应性：这种模式具有很强的灵活性和对未知情况的适应性。它可以根据任务的动态需求，实时调整发言顺序，确保任务高效推进。
• 示例：通过一个模拟软件开发的案例演示选择式小组对话的实现过程。创建包含产品经理、Python工程师和代码审查员的团队，定义终止条件并运行团队。

import os
import dotenv
import asyncio

dotenv.load_dotenv()

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelInfo
from autogen_agentchat.ui import Console

# 配置 Qwen 模型信息
model_info = ModelInfo(
    vision=False,
    function_calling=True,
    json_output=True,
    structured_output=True,
    family="qwen"
)

# 配置通用模型客户端
general_model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-latest",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info=model_info
)

# 配置代码模型客户端
code_model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen3-coder-plus",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info=model_info
)

# 创建产品经理智能体
# 它的描述清晰地说明了其职责是定义需求和编写规格
product_manager = AssistantAgent(
    name="product_manager",
    model_client=general_model_client,
    model_client_stream=True,
    system_message="""
    你是一名软件开发产品经理。你的任务是接收用户的需求，编写清晰、详细的功能说明文档。
    你的说明文档会被交给.net工程师进行开发，请注意专业性。
    """,
    description="产品经理：负责理解用户需求，并将其转化为清晰、具体的功能规格说明文档，分派给工程师。",
)

# 创建Python工程师智能体
# 它的描述指明了其专长是根据规格编写代码
python_engineer = AssistantAgent(
    name="python_engineer",
    model_client=code_model_client,
    model_client_stream=True,
    system_message="""
    你是一名资深.net工程师。你的任务是根据产品经理提供的规格说明文档，编写高质量、符合要求的.net代码。
    你只需要输出代码和注释，不需要输出任何其它内容。
    """,
    description=".net工程师：根据产品经理提供的规格说明文档编写.Net代码。",
)

# 创建代码审查员智能体
# 它的描述强调了其审查代码和结束任务的能力
code_reviewer = AssistantAgent(
    name="code_reviewer",
    model_client=code_model_client,
    model_client_stream=True,
    system_message="""
    你是一名代码审查员。你的任务是仔细审查.net工程师提交的代码，确保其完全符合规格、没有错误。
    你只有两个选择：
    - 代码没有问题，只需要回复 'TERMINATE' 来结束任务
    - 代码有问题，提出具体的修改意见。
    """,
    description="代码审查员：审查由工程师编写的代码，检查其质量和正确性。",
)

# 当任何消息中出现 "TERMINATE" 这个词时，任务就会停止
text_termination = TextMentionTermination("TERMINATE")

# 将所有智能体放入一个列表中
agents=[product_manager, python_engineer, code_reviewer]

# 创建一个选择器驱动的小组对话团队
team = SelectorGroupChat(
    agents,
    # 团队本身（即选择器）也需要要指定一个模型客户端
    model_client=general_model_client,
    termination_condition=text_termination
)

# 定义一个异步主函数来运行团队
async def main():
    # 运行团队，并分配一个研究任务
    await Console(team.run_stream(task="我有一个产品需求，开发一个用于计算斐波那契数列的第n项的程序。程序需要能正确处理n<=1的边界情况。"))

if __name__ == "__main__":
    # 运行异步主函数
    # 在 Python 脚本中，我们通常使用 asyncio.run() 来启动顶层的异步函数
    asyncio.run(main())

展示代码实现步骤，解释选择器如何根据上下文动态选择智能体。运行结果显示团队成员根据任务需求动态发言，验证了这种模式在复杂任务中的优势。

• 选择式小组对话的定制能力

定制能力：选择式小组对话支持通过自定义选择器提示来增加人类定义的规则，实现对对话流程的精细控制。

自定义提示模板：通过自定义提示，可以更好地引导智能体的行为。

# 自定义选择器提示
# 我们在这个提示中加入了一条强规则：在规划师分配任务之前，不要选择其他智能体
custom_selector_prompt = """
请仔细阅读以下对话历史和角色描述，然后从 {participants} 中选择一个智能体来执行下一步任务。

**重要规则**:
- 在产品经理明确地将任务分解并分配给工程师之前，你必须优先选择产品经理。
- 请只选择一个智能体。

**角色描述**:
{roles}

**对话历史**:
{history}

请根据以上信息，选择最合适的下一个发言者。
"""

• 轮询式小组对话与选择式小组对话对比

特性	轮询式小组对话	选择式小组对话
发言者选择机制	基于预设的列表顺序，固定轮流。	基于大语言模型对上下文的理解，动态选择。
对话流程	线性、可预测、僵化。	动态、自适应、灵活。
适用场景	流程固定的任务，如“生成-审查”循环。	流程不确定的复杂任务，需要动态规划和任务分配。
实现复杂度	极低，只需定义智能体顺序。	较高，需要精心设计智能体描述和可能的选择器提示。
控制方式	通过智能体在列表中的位置进行硬编码控制。	通过自然语言描述和提示进行软编码控制。

优缺点对比：轮询式更适合固定流程任务，而选择式在动态任务中更具灵活性和可控性。

应用价值：选择式小组对话的定制能力使其能够适应各种复杂场景，为开发者提供了强大的工具来优化多智能体协作流程。

2. 人机协助模式

• 必要性探讨：在某些任务中，人类的监督、决策和领域知识至关重要。人机协作能够充分发挥人类和智能体的优势，提升任务的准确性和可靠性。
• 应用场景：人机协作的主要原因包括监督与纠偏、提供外部信息、处理模糊性和最终审批。结合实际案例展示人机协作在复杂任务中的应用场景，尤其是在高风险任务中的重要性。
• 用户代理智能体的实现
  • 用户代理智能体的作用：用户代理智能体作为人类用户在团队中的‘代理人’，在需要时暂停团队流程并等待人类输入。这种机制确保了人类可以随时介入，提供关键决策。
  • 交互过程：通过实际运行结果展示用户代理智能体在人机协作中的交互过程。当团队需要人类输入时，用户代理智能体会暂停流程，等待人类反馈后再继续。
  • 示例

import os
import dotenv
import asyncio

dotenv.load_dotenv()

from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.ui import Console

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-latest",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "structured_output": True,
        "family": "qwen",
    }
)

# 创建创作者智能体
writer_agent = AssistantAgent(
    name="writer",
    model_client=model_client,
    model_client_stream=True,
    system_message="你是一位创作者，负责根据要求创作内容。"
)

# 创建代表人类用户的代理智能体
human_proxy_agent = UserProxyAgent(
    name="human",
    # 一个接受提示信息并返回用户输入字符串的函数
    input_func=lambda prompt: input("\n请给出建议，无建议请回复'同意': ")
)

text_termination = TextMentionTermination("同意")

# 组建团队，注意智能体的顺序
agents=[writer_agent, human_proxy_agent]
team = RoundRobinGroupChat(
    agents,
    termination_condition=text_termination
)

# 定义一个异步主函数来运行团队
async def main():
    # 运行团队，并分配一个研究任务
    await Console(team.run_stream(task="写一篇秋天的唐诗"))
    await model_client.close()

if __name__ == "__main__":
	# 运行异步主函数
	# 在 Python 脚本中，我们通常使用 asyncio.run() 来启动顶层的异步函数
    asyncio.run(main())

import os
import dotenv
import asyncio

dotenv.load_dotenv()

from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import TextMentionTermination, HandoffTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.ui import Console

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-latest",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "structured_output": True,
        "family": "qwen",
    }
)

# 创建一个总是转交给用户的懒惰智能体。
lazy_agent = AssistantAgent(
    name="lazy_agent",
    model_client=model_client,
    model_client_stream=True,
    handoffs=[Handoff(target="user", message="需要用户的帮助")],
    system_message="如果你无法完成任务，请转交给 user。否则，完成后，请回复 'TERMINATE'。"
)

# 转交消息的终止条件  transfer_to_user
handoff_termination = HandoffTermination(target="user")
# 特定文本的终止条件
text_termination = TextMentionTermination("TERMINATE")

# 创建团队
lazy_agent_team = RoundRobinGroupChat(
    [lazy_agent],
    termination_condition = handoff_termination | text_termination)

# 定义一个异步主函数来运行团队
async def main():
    await Console(lazy_agent_team.run_stream(task="今天武汉的天气怎么样？"))
    await Console(lazy_agent_team.run_stream(task="武汉是晴天"))

if __name__ == "__main__":
	# 运行异步主函数
	# 在 Python 脚本中，我们通常使用 asyncio.run() 来启动顶层的异步函数
    asyncio.run(main())

3. 多次运行之间的人机协作

在团队多次运行之间注入人类反馈的人机协作模式，允许团队在运行过程中暂停，收集人类反馈后再继续执行。这种模式通过设置最大对话轮数或使用转交终止条件实现，具有很强的灵活性和适应性。

import os
import dotenv
import asyncio

dotenv.load_dotenv()

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.ui import Console

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-latest",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "structured_output": True,
        "family": "qwen",
    }
)

# 创建创作者智能体
# 它的任务是成为一个有帮助的人工智能助手
writer_agent = AssistantAgent(
    name="writer",
    model_client=model_client,
    model_client_stream=True,
    system_message="你是一位创作者，负责根据要求创作内容。"
)

# 组建团队，设置max_turns最大轮数
team = RoundRobinGroupChat([writer_agent], max_turns=1)

# 定义一个异步主函数来运行团队
async def main():
    task = "写一篇秋天的唐诗"
    while True:
        # 运行团队，并分配一个研究任务
        await Console(team.run_stream(task=task))
        task = input("请输入回复 (输入 'exit' 退出): ")
        if task.lower().strip() == "exit":
            break

if __name__ == "__main__":
    # 运行异步主函数
	# 在 Python 脚本中，我们通常使用 asyncio.run() 来启动顶层的异步函数
    asyncio.run(main())

4. 去中心化群聊模式

• 去中心化任务传递：Swarm团队模式摒弃了集中式调度，采用去中心化的任务传递方式。每个智能体通过HandoffMessage主动将任务传递给下一个合适的协作者，使任务流转更加灵活。
• 流程透明与可追溯：与集中式调度相比，Swarm模式通过显式的移交消息，使任务传递过程更加透明和可追溯，便于监控和分析任务的流转路径。
• 对模型能力的要求：Swarm模式要求底层模型具备可靠的工具调用能力，以确保智能体能够准确生成和理解HandoffMessage，从而实现有效的任务传递。

示例

• 角色与任务：在旅行客服场景中，旅行客服负责接收用户初始请求，识别退款需求后将任务传递给退款处理员，退款处理员完成退款后反馈给客服。
• 任务闭环：退款处理员调用退款工具完成退款后，将结果反馈给旅行客服，客服总结并结束对话，形成完整的任务闭环。

import os
import dotenv
dotenv.load_dotenv()

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.messages import HandoffMessage
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 配置通义千问模型客户端
model_client = OpenAIChatCompletionClient(
    # 指定要使用的通义千问模型，例如 qwen-plus, qwen-turbo, qwen-max 等
    model="qwen-plus-2025-07-28",
    # 填入您在 DashScope 获得的 API Key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 填入通义千问的 OpenAI 兼容模式 API 地址
    base_url=os.getenv("DASHSCOPE_API_BASE"),
    # 为自定义模型提供信息
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": True,
        "structured_output": True,
        "family": "qwen",
    }
)

# 定义一个模拟的退款工具
def refund_flight(flight_number: str) -> str:
    """用于处理航班退款。"""
    print(f"[工具执行] 正在为航班 {flight_number} 办理退款...")
    return f"航班 {flight_number} 的退款已成功处理。"

# 创建旅行客服智能体
# 它可以通过 handoffs 参数将任务传递给 "flights_refunder" 或 "user"
travel_agent = AssistantAgent(
    name="travel_agent",
    model_client=model_client,
    model_client_stream=True,
    handoffs=["flights_refunder", "user"],
    system_message="""
    你是一名旅行客服，可以回答任何与旅行产品相关的问题，每当你回复完毕请将任务转交给 user。
    如果你需要用户回答，请将任务转交给 user。
    如果用户需要退款，请将任务移交给 flights_refunder 负责。
    如果用户的所有问题都解决，请输出 '对话完毕' 以结束对话。
    """
)

# 创建退款处理员智能体
# 它拥有退款工具，并能将任务交还给 "travel_agent" 或 "user"
flights_refunder = AssistantAgent(
    name="flights_refunder",
    model_client=model_client,
    model_client_stream=True,
    tools=[refund_flight],
    handoffs=["travel_agent", "user"],
    system_message="""
    你是一名专门处理航班退款的专员。
    退款需要航班号才能处理，如果你不知道航班号，必须将任务转交给 user 补充。
    退款完成后请将任务转交给 travel_agent。
    """,
)

# 定义终止条件
# 当任务被传递给 "user" 时，或者当任何智能体说了 "TERMINATE" 时，对话就暂停
termination_condition = HandoffTermination(target="user") | TextMentionTermination("对话完毕")

# 使用 Swarm 组建团队
# 列表中的第一个智能体，即 travel_agent，将是第一个发言者
agents=[travel_agent, flights_refunder]
team = Swarm(
    agents,
    termination_condition=termination_condition,
)

import asyncio
from autogen_agentchat.ui import Console

async def main():
    # 初始任务
    task = "你好，你是谁？"
    # 运行团队
    task_result = await Console(team.run_stream(task=task))

    # 检查是否需要人类输入
    last_message = task_result.messages[-1]
    # 如果最后一条消息是一个指向用户的 Handoff，则进入人机交互循环
    while isinstance(last_message, HandoffMessage) and last_message.target == "user":
        user_input = input("请输入您的回复: ")
        # 将用户的输入作为新的 HandoffMessage 传递回团队
        # 注意 target 是 last_message.source，这样可以确保回复给正确的智能体
        task_result = await Console(
            team.run_stream(
                task=HandoffMessage(source="user", target=last_message.source, content=user_input)
            )
        )
        last_message = task_result.messages[-1]

if __name__ == "__main__":
    asyncio.run(main())

Swarm代码实现要点：

• handoffs参数配置：在创建智能体时，通过handoffs参数指定可移交任务的目标智能体列表，明确智能体之间的协作关系。
• 系统消息定义：为智能体编写详细的系统消息，指导其何时以及如何传递任务，确保智能体能够准确理解任务要求。
• 终止条件设置：定义HandoffTermination和TextMentionTermination等终止条件，以控制任务的结束时机，避免无限循环。
• 人机交互处理：通过人机交互循环，处理用户输入，确保任务能够在用户参与下继续进行，实现智能体与人类的协同工作。