微软AutoGen介绍——Custom Agents创建自己的Agents网站首页 技术交流

微软AutoGen介绍——Custom Agents创建自己的Agents

梦丶晓羽 2025-07-31 00:01:04

简介微软AutoGen介绍——Custom Agents创建自己的Agents

介绍

大家好，这次给大家分享的内容是微软AutoGen框架一个重要的功能模块Custom Agents(自定义代理)。本次分享的内容需要大家具备一定的面向对象编程经验。那么本次分享的Custom Agents又是什么呢，我们直接进入正题。

Custom Agents

我们可能会遇到行为不符合预设的智能体。在这种情况下，我们可以构建自定义智能体。

AgentChat中的所有智能体都继承自BaseChatAgent类，并实现以下抽象方法和属性：

on_messages()：这是一个抽象方法，用于定义智能体在收到消息时的响应行为。当在run()中要求智能体提供响应时，会调用此方法。它返回一个Response对象。
on_reset()：此抽象方法用于将智能体重置为初始状态。当要求智能体自行重置时，会调用此方法。
produced_message_types：智能体在其响应中可能生成的ChatMessage消息类型列表。

我们也可以选择实现on_messages_stream()方法，以便在智能体生成消息时将其流式输出。如果未实现此方法，智能体将使用on_messages_stream()的默认实现，该默认实现会调用on_messages()方法，并逐一生成响应中的所有消息。

CountDownAgent

我们写个示例来演示。我们创建一个简单的智能体，它从给定数字倒数到零，并生成一系列包含当前计数的消息流。

完整代码

import asyncio
from typing import AsyncGenerator, List, Sequence, Tuple
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessage
from autogen_core import CancellationToken


class CountDownAgent(BaseChatAgent):
    def __init__(self, name: str, count: int = 3):
        super().__init__(name, "A simple agent that counts down.")
        self._count = count

    @property
    def produced_message_types(self) -> Sequence[type[ChatMessage]]:
        return (TextMessage,)

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # 调用on_messages_stream方法。
        response: Response | None = None
        async for message in self.on_messages_stream(messages, cancellation_token):
            if isinstance(message, Response):
                response = message
        assert response is not None
        return response

    async def on_messages_stream(
            self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken
    ) -> AsyncGenerator[AgentEvent | ChatMessage | Response, None]:
        inner_messages: List[AgentEvent | ChatMessage] = []
        for i in range(self._count, 0, -1):
            msg = TextMessage(content=f"{i}...", source=self.name)
            inner_messages.append(msg)
            yield msg
        # 响应会在流结束时返回。
        # 它包含最终消息以及所有内部消息。
        yield Response(chat_message=TextMessage(content="Done!", source=self.name), inner_messages=inner_messages)

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass


async def run_countdown_agent() -> None:
    # 创建一个倒计时智能体。
    countdown_agent = CountDownAgent("countdown")

    # 用给定任务运行该智能体，并将响应以流的形式输出。
    async for message in countdown_agent.on_messages_stream([], CancellationToken()):
        if isinstance(message, Response):
            print(message.chat_message.content)
        else:
            print(message.content)


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

asyncio.run(main())

运行结果

3...
2...
1...
Done!

进程已结束，退出代码为 0

ArithmeticAgent

我们再写个示例来演示，我们创建一个智能体类，它能够对给定的整数执行简单的算术运算。然后，我们会在SelectorGroupChat中使用这个智能体类的不同实例，通过应用一系列算术运算，将给定的整数转换为另一个整数。

ArithmeticAgent类接受一个operator_func，该函数接受一个整数，对该整数应用一种算术运算后，返回一个整数。在其on_messages方法中，它将operator_func应用于输入消息中的整数，并返回一个包含结果的响应。

from typing import Callable, Sequence, List
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import ChatMessage, TextMessage
from autogen_core import CancellationToken


class ArithmeticAgent(BaseChatAgent):
    def __init__(self, name: str, description: str, operator_func: Callable[[int], int]) -> None:
        super().__init__(name, description=description)
        self._operator_func = operator_func
        self._message_history: List[ChatMessage] = []

    @property
    def produced_message_types(self) -> Sequence[type[ChatMessage]]:
        return (TextMessage,)

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # 更新消息历史记录。
        # 注意：消息列表有可能为空，这意味着该智能体之前已被选中。
        self._message_history.extend(messages)

        # 解析最后一条消息中的数字。
        assert isinstance(self._message_history[-1], TextMessage)
        number = int(self._message_history[-1].content)

        # 将运算符函数应用于该数字。
        result = self._operator_func(number)

        # 用结果创建一条新消息。
        response_message = TextMessage(content=str(result), source=self.name)

        # 更新消息记录。
        self._message_history.append(response_message)

        # 返回响应。
        return Response(chat_message=response_message)

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass

博主笔记：on_messages方法可能会接收到一个空的消息列表，这种情况下，意味着该智能体之前已被调用过，现在是再次被调用，调用方并未提供任何新消息。因此，记录智能体之前接收的消息历史并利用该历史来生成响应，这一点至关重要。

现在，我们可以创建一个包含 5 个ArithmeticAgent实例的SelectorGroupChat：

一个实例将输入整数加1；
一个实例将输入整数减1；
一个实例将输入整数乘以2；
一个实例将输入整数除以2并向下取整到最接近的整数；
一个实例返回不变的输入整数。

然后，我们使用这些智能体创建一个SelectorGroupChat，并设置相应的选择器设置：

允许连续选择同一个智能体，以便进行重复操作；
定制选择器提示，使模型的响应更贴合特定任务。

代码演示

async def run_number_agents() -> None:
    # 创建用于数字运算的智能体。
    add_agent = ArithmeticAgent("add_agent", "Adds 1 to the number.", lambda x: x + 1)
    multiply_agent = ArithmeticAgent("multiply_agent", "Multiplies the number by 2.", lambda x: x * 2)
    subtract_agent = ArithmeticAgent("subtract_agent", "Subtracts 1 from the number.", lambda x: x - 1)
    divide_agent = ArithmeticAgent("divide_agent", "Divides the number by 2 and rounds down.", lambda x: x // 2)
    identity_agent = ArithmeticAgent("identity_agent", "Returns the number as is.", lambda x: x)

    # 终止条件是在生成10条消息后停止。
    termination_condition = MaxMessageTermination(10)

    # 创建一个选择器群组聊天。
    selector_group_chat = SelectorGroupChat(
        [add_agent, multiply_agent, subtract_agent, divide_agent, identity_agent],
        model_client=OpenAIChatCompletionClient(model="gpt-4o"),
        termination_condition=termination_condition,
        allow_repeated_speaker=True,  # 允许同一个智能体发言多次，这对于此任务而言是必要的。
        selector_prompt=(
            "Available roles:
{roles}
Their job descriptions:
{participants}
"
            "Current conversation history:
{history}
"
            "Please select the most appropriate role for the next message, and only return the role name."
        ),
    )

    # 使用给定任务运行选择器群组聊天，并以流的形式输出响应。
    task: List[ChatMessage] = [
        TextMessage(content="Apply the operations to turn the given number into 25.", source="user"),
        TextMessage(content="10", source="user"),
    ]
    stream = selector_group_chat.run_stream(task=task)
    await Console(stream, output_stats=True)

完整代码

import asyncio
from typing import Callable, Sequence, List
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.messages import ChatMessage, TextMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient


class ArithmeticAgent(BaseChatAgent):
    def __init__(self, name: str, description: str, operator_func: Callable[[int], int]) -> None:
        super().__init__(name, description=description)
        self._operator_func = operator_func
        self._message_history: List[ChatMessage] = []

    @property
    def produced_message_types(self) -> Sequence[type[ChatMessage]]:
        return (TextMessage,)

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # 更新消息历史记录。
        # 注意：消息列表有可能为空，这意味着该智能体之前已被选中。
        self._message_history.extend(messages)

        # 解析最后一条消息中的数字。
        assert isinstance(self._message_history[-1], TextMessage)
        number = int(self._message_history[-1].content)

        # 将运算符函数应用于该数字。
        result = self._operator_func(number)

        # 用结果创建一条新消息。
        response_message = TextMessage(content=str(result), source=self.name)

        # 更新消息记录。
        self._message_history.append(response_message)

        # 返回响应。
        return Response(chat_message=response_message)

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass


async def run_number_agents() -> None:
    # 创建用于数字运算的智能体。
    add_agent = ArithmeticAgent("add_agent", "Adds 1 to the number.", lambda x: x + 1)
    multiply_agent = ArithmeticAgent("multiply_agent", "Multiplies the number by 2.", lambda x: x * 2)
    subtract_agent = ArithmeticAgent("subtract_agent", "Subtracts 1 from the number.", lambda x: x - 1)
    divide_agent = ArithmeticAgent("divide_agent", "Divides the number by 2 and rounds down.", lambda x: x // 2)
    identity_agent = ArithmeticAgent("identity_agent", "Returns the number as is.", lambda x: x)

    # 终止条件是在生成 10 条消息后停止。
    termination_condition = MaxMessageTermination(10)

    # 创建一个选择器群组聊天。
    selector_group_chat = SelectorGroupChat(
        [add_agent, multiply_agent, subtract_agent, divide_agent, identity_agent],
        model_client=OpenAIChatCompletionClient(model="gpt-4o"),
        termination_condition=termination_condition,
        allow_repeated_speaker=True,  # 允许同一个智能体发言多次，这对于此任务而言是必要的。
        selector_prompt=(
            "Available roles:
{roles}
Their job descriptions:
{participants}
"
            "Current conversation history:
{history}
"
            "Please select the most appropriate role for the next message, and only return the role name."
        ),
    )

    # 使用给定任务运行选择器群组聊天，并以流的形式输出响应。
    task: List[ChatMessage] = [
        TextMessage(content="Apply the operations to turn the given number into 25.", source="user"),
        TextMessage(content="10", source="user"),
    ]
    stream = selector_group_chat.run_stream(task=task)
    await Console(stream, output_stats=True)


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

asyncio.run(main())

运行结果

---------- user ----------
Apply the operations to turn the given number into 25.
---------- user ----------
10
---------- add_agent ----------
11
---------- multiply_agent ----------
22
---------- add_agent ----------
23
---------- add_agent ----------
24
---------- add_agent ----------
25
---------- identity_agent ----------
25
---------- identity_agent ----------
25
---------- identity_agent ----------
25
---------- Summary ----------
Number of messages: 10
Finish reason: Maximum number of messages 10 reached, current message count: 10
Total prompt tokens: 0
Total completion tokens: 0
Duration: 97.18 seconds

进程已结束，退出代码为 0

从输出结果中我们可以看到，通过选择合适的智能体按顺序执行算术运算，这些智能体成功地将输入整数从10变换到了25。

说明

如果大家在运行上述代码的时候有AutoGen相关的提示或报错(例如：该参数不存在，没有此类方法等)，请尝试更新一下AutoGen，博主在分享这篇博文的时候，AutoGen的版本是0.4.5稳定版。

安装或更新命令

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

另外大家要根据业务需求，设置使用的LLM，不一定要按照我给大家分享代码中的设置来，如果只是为了测试并看运行结果可直接复制粘贴代码(完整代码)。

结束

好了，以上就是本次分享的全部内容，不知道大家是否理解了Custom Agents的功能以及在AutoGen框架中的作用。为了希望大家能够掌握本次分享的内容，博主再次帮大家对Custom Agents进行梳理：

在微软的AutoGen中，Custom Agents指的是自定义智能体，是用户根据自身特定需求和业务逻辑创建的具有独特功能和行为的智能体。

定义与功能

自主性：Custom Agents能够自主地进行信息处理和决策。它们可以根据接收到的信息以及自身预设的规则和算法，自行决定如何行动，而不需要每一步都由用户或外部程序进行明确的指令操作。
交互性：可以与其他智能体、用户或外部环境进行交互。在AutoGen的框架下，它们能够通过发送和接收消息来与其他组件进行通信，从而实现信息的共享和协同工作。
自定义逻辑：用户可以为Custom Agents定义特定的业务逻辑和行为规则。这意味着根据不同的应用场景和需求，开发者可以让智能体执行各种不同的任务，例如在特定条件下触发特定的操作，对特定类型的信息进行特殊处理等。

应用场景

个性化智能助手：在智能办公、智能客服等场景中，Custom Agents可以根据不同用户的需求和使用习惯，提供个性化的服务和支持。比如，为不同部门的员工定制专门的工作流程助手，或者为不同类型的客户提供针对性的问题解答和服务推荐。
复杂任务协同：在涉及多个环节和多种类型任务的复杂业务流程中，Custom Agents可以被设计用来负责不同的任务模块，并与其他智能体协同工作，以完成整个业务流程。例如在项目管理中，不同的Custom Agents可以分别负责任务分配、进度跟踪、资源协调等工作，它们之间相互配合，确保项目的顺利进行。
特定领域应用：在一些特定领域，如医疗、金融、法律等，Custom Agents可以根据该领域的专业知识和规则进行定制，用于处理专业问题和提供专业建议。比如在医疗领域，定制的智能体可以根据患者的症状和病历信息进行初步诊断和治疗建议；在金融领域，智能体可以进行风险评估和投资策略制定。

大家明白了吗？博主希望本次分享对目前正在使用AutoGen框架做大模型应用开发的同事，还有正在学习使用AutoGen框架的同学有所启发和帮助，对AutoGen框架中的各个功能模块有深入的理解及熟练的使用，这样博主也就达到分享的目的了。博主还是那句话，请大家多去大胆的尝试和使用，博主希望本次分享能够对大家有所帮助。如果大家对博主分享的内容感兴趣或有帮助，请点赞和关注，博主会持续更新关于微软AutoGen更多和更有趣的内容。

风语者！平时喜欢研究各种技术，目前在从事后端开发工作，热爱生活、热爱工作。