1.基本用法的相关代码

注意:生成最新的高于0.4版本代码,可以引入相关MCP: context7的 Autogen:

class AssistantAgent(BaseChatAgent, Component[AssistantAgentConfig]):
    """一个支持工具使用的助手代理。
    :meth:`on_messages` 方法返回一个 :class:`~autogen_agentchat.base.Response`,
    其中 :attr:`~autogen_agentchat.base.Response.chat_message` 是最终的响应消息。

    :meth:`on_messages_stream` 方法创建一个异步生成器,在消息生成时逐条产出内部消息,
    并在生成器关闭前的最后一步产出 :class:`~autogen_agentchat.base.Response` 对象。

    :meth:`BaseChatAgent.run` 方法返回一个 :class:`~autogen_agentchat.base.TaskResult`,
    其中包含代理生成的所有消息。在消息列表 :attr:`~autogen_agentchat.base.TaskResult.messages` 中,
    最后一条消息是最终的响应消息。

    :meth:`BaseChatAgent.run_stream` 方法创建一个异步生成器,在消息生成时逐条产出内部消息,
    并在生成器关闭前的最后一步产出 :class:`~autogen_agentchat.base.TaskResult` 对象。

    .. attention::

        调用者在每次调用 :meth:`on_messages`、:meth:`on_messages_stream`、
        :meth:`BaseChatAgent.run` 或 :meth:`BaseChatAgent.run_stream` 方法时,
        **必须仅传递新增的消息**给代理。
        代理会在这些方法调用之间维护自身状态。
        **切勿**在每次调用时将完整的对话历史传递给代理。

    .. warning::
        助手代理**不是线程安全或协程安全**的。
        不应在多个任务或协程之间共享该代理,也不应并发调用其方法。

    下图展示了助手代理的工作原理:

    .. image:: ../../images/assistant-agent.svg

    **结构化输出:**

    如果设置了 `output_content_type`,代理默认会以 :class:`~autogen_agentchat.messages.StructuredMessage`
    形式而非 :class:`~autogen_agentchat.messages.TextMessage` 返回最终响应。

    .. note::

        目前,设置 `output_content_type` 会阻止代理调用 `load_component` 和 `dump_component` 方法进行可序列化配置。
        此问题将在未来版本中修复。

    **工具调用行为:**

    * 如果模型未返回任何工具调用,则立即以 :class:`~autogen_agentchat.messages.TextMessage` 或
      :class:`~autogen_agentchat.messages.StructuredMessage`(使用结构化输出时)形式返回响应,
      并结束工具调用迭代循环,无论 `max_tool_iterations` 如何设置。
    * 当模型返回工具调用时,将立即执行:
        - 当 `reflect_on_tool_use` 为 False 时,工具调用结果将以 :class:`~autogen_agentchat.messages.ToolCallSummaryMessage`
          形式返回至 :attr:`~autogen_agentchat.base.Response.chat_message`。您可以通过静态格式字符串
          (`tool_call_summary_format`) **或** 可调用对象 (`tool_call_summary_formatter`) 自定义摘要;
          可调用对象对每个工具调用评估一次。
        - 当 `reflect_on_tool_use` 为 True 时,将基于工具调用及结果进行另一次模型推理,并以
          :class:`~autogen_agentchat.messages.TextMessage` 或 :class:`~autogen_agentchat.messages.StructuredMessage`
          (使用结构化输出时)形式返回最终响应至 :attr:`~autogen_agentchat.base.Response.chat_message`。
        - 当设置 `output_content_type` 时,默认 `reflect_on_tool_use` 为 `True`。
        - 当未设置 `output_content_type` 时,默认 `reflect_on_tool_use` 为 `False`。
    * 若模型返回多个工具调用,将并发执行。要禁用并行工具调用,需配置模型客户端。
      例如,对 :class:`~autogen_ext.models.openai.OpenAIChatCompletionClient` 和
      :class:`~autogen_ext.models.openai.AzureOpenAIChatCompletionClient` 设置 `parallel_tool_calls=False`。
    * `max_tool_iterations` 参数控制代理在单次运行中可执行的连续工具调用迭代次数。
      默认值为 1,表示代理仅执行一次模型生成的工具调用并返回结果。
      若设置更高值,当模型持续请求工具调用时,代理可进行额外模型调用以执行更多工具调用,实现多步骤工具工作流。
      代理会在模型返回文本响应(而非工具调用)或达到最大迭代次数时停止。

    .. tip::

        默认情况下,当执行工具调用时,工具调用结果将作为响应返回,
        因此请特别注意工具返回值的格式——尤其是当其他代理期望特定模式时。

        * 使用 **`tool_call_summary_format`** 进行简单的静态模板格式化。
        * 使用 **`tool_call_summary_formatter`** 实现完全的程序化控制
          (例如:“成功时隐藏大负载,错误时显示完整详情”)。

        *注意*:`tool_call_summary_formatter` **不可序列化**,
        在从 YAML/JSON 配置文件加载或导出代理时将被忽略。

    **移交行为:**

    * 若触发移交,将在 :attr:`~autogen_agentchat.base.Response.chat_message` 中返回
      :class:`~autogen_agentchat.messages.HandoffMessage`。
    * 若存在工具调用,将在返回移交前立即执行。
    * 工具调用及其结果将通过 :attr:`~autogen_agentchat.messages.HandoffMessage.context` 传递给目标代理。

    .. note::
        若检测到多个移交,仅执行第一个移交。
        为避免此情况,请在模型客户端配置中禁用并行工具调用。

    **限制发送给模型的上下文大小:**

    可通过将 `model_context` 参数设置为 :class:`~autogen_core.model_context.BufferedChatCompletionContext`
    来限制发送给模型的消息数量。这将限制发送给模型的最近消息数量,在模型有令牌处理限制时非常有用。
    另一种选择是使用 :class:`~autogen_core.model_context.TokenLimitedChatCompletionContext`,
    它将限制发送给模型的令牌数量。
    您也可以通过继承 :class:`~autogen_core.model_context.ChatCompletionContext`
    创建自定义模型上下文。

    **流式模式:**

    通过设置 `model_client_stream=True`,助手代理可启用流式模式。
    在此模式下,:meth:`on_messages_stream` 和 :meth:`BaseChatAgent.run_stream` 方法
    在模型客户端生成响应块时,还会产出 :class:`~autogen_agentchat.messages.ModelClientStreamingChunkEvent` 消息。
    这些块消息不会包含在最终响应的内部消息中。

    Args:
        name (str): 代理名称。
        model_client (ChatCompletionClient): 用于推理的模型客户端。
        tools (List[BaseTool[Any, Any] | Callable[..., Any] | Callable[..., Awaitable[Any]]] | None, optional): 向代理注册的工具。
        workbench (Workbench | Sequence[Workbench] | None, optional): 代理使用的工坊或工坊列表。
            设置 workbench 时不能使用工具,反之亦然。
        handoffs (List[HandoffBase | str] | None, optional): 代理的移交配置,
            允许其通过返回 :class:`HandoffMessage` 转移到其他代理。
            仅当团队处于 :class:`~autogen_agentchat.teams.Swarm` 时才会执行转移。
            若移交为字符串,应表示目标代理的名称。
        model_context (ChatCompletionContext | None, optional): 用于存储和检索 :class:`~autogen_core.models.LLMMessage` 的模型上下文。
            可预加载初始消息。重置代理时初始消息将被清除。
        description (str, optional): 代理描述。
        system_message (str, optional): 模型的系统消息。若提供,将在推理时前置到模型上下文的消息中。设为 `None` 以禁用。
        model_client_stream (bool, optional): 若为 `True`,模型客户端将以流式模式使用。
            :meth:`on_messages_stream` 和 :meth:`BaseChatAgent.run_stream` 方法在模型客户端生成响应块时,
            还会产出 :class:`~autogen_agentchat.messages.ModelClientStreamingChunkEvent` 消息。默认为 `False`。
        reflect_on_tool_use (bool, optional): 若为 `True`,代理将基于工具调用及结果进行另一次模型推理以生成响应。
            若为 `False`,工具调用结果将作为响应返回。默认情况下,若设置 `output_content_type`,此值为 `True`;
            若未设置 `output_content_type`,此值为 `False`。
        output_content_type (type[BaseModel] | None, optional): 作为 Pydantic 模型的 :class:`~autogen_agentchat.messages.StructuredMessage` 响应的输出内容类型。
            将与模型客户端配合生成结构化输出。
            若设置此项,代理将以 :class:`~autogen_agentchat.messages.StructuredMessage` 而非
            :class:`~autogen_agentchat.messages.TextMessage` 返回最终响应,
            除非 `reflect_on_tool_use` 为 `False` 且执行了工具调用。
        output_content_type_format (str | None, optional): (实验性)用于 :class:`~autogen_agentchat.messages.StructuredMessage` 响应内容的格式字符串。
        max_tool_iterations (int, optional): 执行工具调用的最大迭代次数,直到模型停止发出工具调用。默认为 `1`,
            表示代理仅执行一次模型生成的工具调用,并以 :class:`~autogen_agentchat.messages.ToolCallSummaryMessage`,
            或 :class:`~autogen_agentchat.messages.TextMessage` 或 :class:`~autogen_agentchat.messages.StructuredMessage`
            (使用结构化输出时)形式在 :attr:`~autogen_agentchat.base.Response.chat_message` 中返回结果作为最终响应。
            一旦模型停止发出工具调用,代理将停止执行工具调用并返回结果作为最终响应。
            该值必须大于或等于 1。
        tool_call_summary_format (str, optional): 应用于每个工具调用结果的静态格式字符串,用于构建 :class:`~autogen_agentchat.messages.ToolCallSummaryMessage`。
            默认为 ``"{result}"``。若提供 `tool_call_summary_formatter` 则忽略此项。当 `reflect_on_tool_use` 为 ``False`` 时,
            所有工具调用的摘要将用换行符 ('\\n') 连接并作为响应返回。模板中可用的占位符:
            `{tool_name}`, `{arguments}`, `{result}`, `{is_error}`。
        tool_call_summary_formatter (Callable[[FunctionCall, FunctionExecutionResult], str] | None, optional):
            接收 ``FunctionCall`` 及其 ``FunctionExecutionResult`` 并返回摘要字符串的可调用对象。
            提供时将覆盖 `tool_call_summary_format`,允许条件逻辑 — 例如,成功时发出静态字符串
            ``"工具 FooBar 执行成功。"``,失败时才显示完整负载(包括所有传递的参数等)。

            **限制**:该可调用对象*不可序列化*;通过 YAML/JSON 配置文件提供的值将被忽略。

    .. note::

        `tool_call_summary_formatter` 仅适用于代码内使用。目前无法通过配置文件保存或恢复。

        memory (Sequence[Memory] | None, optional): 代理使用的内存存储。默认为 `None`。
        metadata (Dict[str, str] | None, optional): 可选的跟踪元数据。

    Raises:
        ValueError: 工具名称不唯一时。
        ValueError: 移交名称不唯一时。
        ValueError: 移交名称与工具名称不唯一时。
        ValueError: 最大工具迭代次数小于 1 时。

    示例:

        **示例 1:基础代理**

        以下示例演示如何创建带有模型客户端的助手代理,并为简单任务生成响应。

        .. code-block:: python

            import asyncio
            from autogen_ext.models.openai import OpenAIChatCompletionClient
            from autogen_agentchat.agents import AssistantAgent


            async def main() -> None:
                model_client = OpenAIChatCompletionClient(
                    model="gpt-4o",
                    # api_key = "your_openai_api_key"
                )
                agent = AssistantAgent(name="assistant", model_client=model_client)

                result = await agent.run(task="说出北美两个城市的名字。")
                print(result)


            asyncio.run(main())

        **示例 2:模型客户端令牌流式传输**

        此示例演示如何创建带有模型客户端的助手代理,并通过设置 `model_client_stream=True` 生成令牌流。

        .. code-block:: python

            import asyncio
            from autogen_ext.models.openai import OpenAIChatCompletionClient
            from autogen_agentchat.agents import AssistantAgent


            async def main() -> None:
                model_client = OpenAIChatCompletionClient(
                    model="gpt-4o",
                    # api_key = "your_openai_api_key"
                )
                agent = AssistantAgent(
                    name="assistant",
                    model_client=model_client,
                    model_client_stream=True,
                )

                stream = agent.run_stream(task="说出北美两个城市的名字。")
                async for message in stream:
                    print(message)


            asyncio.run(main())

        .. code-block:: text

            source='user' models_usage=None metadata={} content='说出北美两个城市的名字。' type='TextMessage'
            source='assistant' models_usage=None metadata={} content='两个' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='城市' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='在' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='北美' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='是' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='纽约' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='市' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='和' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='多伦多' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='。' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=None metadata={} content='终止' type='ModelClientStreamingChunkEvent'
            source='assistant' models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0) metadata={} content='北美两个城市是纽约市和多伦多。终止' type='TextMessage'
            messages=[TextMessage(source='user', models_usage=None, metadata={}, content='说出北美两个城市的名字。', type='TextMessage'), TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0), metadata={}, content='北美两个城市是纽约市和多伦多。终止', type='TextMessage')] stop_reason=None

        **示例 3:带工具的代理**

        以下示例演示如何创建带有模型客户端和工具的助手代理,为任务生成消息流,
        并使用 :class:`~autogen_agentchat.ui.Console` 将消息打印到控制台。

        该工具是一个返回当前时间的简单函数。
        在底层,该函数被包装在 :class:`~autogen_core.tools.FunctionTool` 中,
        并与代理的模型客户端一起使用。函数的文档字符串用作工具描述,
        函数名用作工具名称,函数签名(包括类型提示)用作工具参数。

        .. code-block:: python

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


            async def get_current_time() -> str:
                return "当前时间是中午12点。"


            async def main() -> None:
                model_client = OpenAIChatCompletionClient(
                    model="gpt-4o",
                    # api_key = "your_openai_api_key"
                )
                agent = AssistantAgent(name="assistant", model_client=model_client, tools=[get_current_time])
                await Console(agent.run_stream(task="现在几点了?"))


            asyncio.run(main())

        **示例 4:带 max_tool_iterations 的代理**

        以下示例演示如何使用 `max_tool_iterations` 参数控制代理在单次运行中可执行工具调用的次数。
        当您希望代理执行多个连续的工具操作以达成目标时,这非常有用。

        .. code-block:: python

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


            # 全局计数器状态
            counter = 0


            def increment_counter() -> str:
                \"\"\"将计数器加1并返回当前值。\"\"\"
                global counter
                counter += 1
                return f"计数器已增加至:{counter}"


            def get_counter() -> str:
                \"\"\"获取当前计数器值。\"\"\"
                global counter
                return f"当前计数器值:{counter}"


            async def main() -> None:
                model_client = OpenAIChatCompletionClient(
                    model="gpt-4o",
                    # api_key = "your_openai_api_key"
                )

                # 创建允许最多5次工具调用迭代的代理
                agent = AssistantAgent(
                    name="assistant",
                    model_client=model_client,
                    tools=[increment_counter, get_counter],
                    max_tool_iterations=5,  # 允许最多5次工具调用迭代
                    reflect_on_tool_use=True,  # 在工具调用后获取最终摘要
                )

                await Console(agent.run_stream(task="将计数器增加3次,然后告诉我最终值。"))


            asyncio.run(main())

        **示例 5:带 Model-Context Protocol (MCP) 工坊的代理**

        以下示例演示如何创建带有模型客户端和 :class:`~autogen_ext.tools.mcp.McpWorkbench` 的助手代理,
        用于与 Model-Context Protocol (MCP) 服务器交互。

        .. code-block:: python

            import asyncio
            from autogen_agentchat.agents import AssistantAgent
            from autogen_agentchat.ui import Console
            from autogen_ext.models.openai import OpenAIChatCompletionClient
            from autogen_ext.tools.mcp import StdioServerParams, McpWorkbench


            async def main() -> None:
                params = StdioServerParams(
                    command="uvx",
                    args=["mcp-server-fetch"],
                    read_timeout_seconds=60,
                )

                # 您也可以使用 `start()` 和 `stop()` 管理会话。
                async with McpWorkbench(server_params=params) as workbench:
                    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")
                    assistant = AssistantAgent(
                        name="Assistant",
                        model_client=model_client,
                        workbench=workbench,
                        reflect_on_tool_use=True,
                    )
                    await Console(
                        assistant.run_stream(task="访问 https://github.com/microsoft/autogen 并告诉我你看到了什么。")
                    )


            asyncio.run(main())

        **示例 6:带结构化输出和工具的代理**

        以下示例演示如何创建配置为使用结构化输出和工具的助手代理。
        注意,您需要使用 :class:`~autogen_core.tools.FunctionTool` 创建工具,
        且结构化输出模式要求 `strict=True`。
        由于模型配置为使用结构化输出,输出反思响应将是 JSON 格式的字符串。

        .. code-block:: python

            import asyncio
            from typing import Literal

            from autogen_agentchat.agents import AssistantAgent
            from autogen_agentchat.ui import Console
            from autogen_core.tools import FunctionTool
            from autogen_ext.models.openai import OpenAIChatCompletionClient
            from pydantic import BaseModel


            # 定义结构化输出格式。
            class AgentResponse(BaseModel):
                thoughts: str
                response: Literal["happy", "sad", "neutral"]


            # 定义作为工具调用的函数。
            def sentiment_analysis(text: str) -> str:
                \"\"\"给定一段文本,返回其情感。\"\"\"
                return "happy" if "happy" in text else "sad" if "sad" in text else "neutral"


            # 创建一个 FunctionTool 实例,设置 `strict=True`,
            # 这是结构化输出模式所必需的。
            tool = FunctionTool(sentiment_analysis, description="情感分析", strict=True)

            # 创建支持结构化输出的 OpenAIChatCompletionClient 实例。
            model_client = OpenAIChatCompletionClient(
                model="gpt-4o-mini",
            )

            # 创建使用工具和模型客户端的 AssistantAgent 实例。
            agent = AssistantAgent(
                name="assistant",
                model_client=model_client,
                tools=[tool],
                system_message="使用工具进行情感分析。",
                output_content_type=AgentResponse,
            )


            async def main() -> None:
                stream = agent.run_stream(task="我今天很开心!")
                await Console(stream)


            asyncio.run(main())

        .. code-block:: text

            ---------- assistant ----------
            [FunctionCall(id='call_tIZjAVyKEDuijbBwLY6RHV2p', arguments='{"text":"我今天很开心!"}', name='sentiment_analysis')]
            ---------- assistant ----------
            [FunctionExecutionResult(content='happy', call_id='call_tIZjAVyKEDuijbBwLY6RHV2p', is_error=False)]
            ---------- assistant ----------
            {"thoughts":"用户通过表示今天很开心,表达了明确的积极情绪,表明心情愉快。","response":"happy"}

        **示例 7:带有限模型上下文的代理**

        以下示例展示了如何使用 :class:`~autogen_core.model_context.BufferedChatCompletionContext`,
        该上下文仅保留最后2条消息(1条用户消息 + 1条助手消息)。
        有限模型上下文在模型有令牌处理限制时非常有用。

        .. code-block:: python

            import asyncio

            from autogen_agentchat.agents import AssistantAgent
            from autogen_core.model_context import BufferedChatCompletionContext
            from autogen_ext.models.openai import OpenAIChatCompletionClient


            async def main() -> None:
                # 创建模型客户端。
                model_client = OpenAIChatCompletionClient(
                    model="gpt-4o-mini",
                    # api_key = "your_openai_api_key"
                )

                # 创建仅保留最后2条消息(1条用户消息 + 1条助手消息)的模型上下文。
                model_context = BufferedChatCompletionContext(buffer_size=2)

                # 创建带有模型客户端和上下文的 AssistantAgent 实例。
                agent = AssistantAgent(
                    name="assistant",
                    model_client=model_client,
                    model_context=model_context,
                    system_message="你是一个乐于助人的助手。",
                )

                result = await agent.run(task="说出北美两个城市的名字。")
                print(result.messages[-1].content)  # type: ignore

                result = await agent.run(task="我最喜欢的颜色是蓝色。")
                print(result.messages[-1].content)  # type: ignore

                result = await agent.run(task="我问过你任何问题吗?")
                print(result.messages[-1].content)  # type: ignore


            asyncio.run(main())

        .. code-block:: text

            北美两个城市是纽约市和多伦多。
            很棒!蓝色常与平静和宁静相关联。你有特别喜欢的蓝色色调吗?或者有什么特别的原因让它成为你的最爱?
            没有,你没有问过问题。对于任何误解我表示歉意。如果你有想讨论或询问的具体内容,请随时告诉我!

        **示例 8:带内存的代理**

        以下示例展示了如何在助手代理中使用基于列表的内存。
        内存预加载了一些初始内容。
        在底层,内存用于在推理前使用 :meth:`~autogen_core.memory.Memory.update_context` 方法更新模型上下文。

        .. code-block:: python

            import asyncio

            from autogen_agentchat.agents import AssistantAgent
            from autogen_core.memory import ListMemory, MemoryContent
            from autogen_ext.models.openai import OpenAIChatCompletionClient


            async def main() -> None:
                # 创建模型客户端。
                model_client = OpenAIChatCompletionClient(
                    model="gpt-4o-mini",
                    # api_key = "your_openai_api_key"
                )

                # 创建带有初始内容的基于列表的内存。
                memory = ListMemory()
                await memory.add(MemoryContent(content="用户喜欢披萨。", mime_type="text/plain"))
                await memory.add(MemoryContent(content="用户不喜欢奶酪。", mime_type="text/plain"))

                # 创建带有模型客户端和内存的 AssistantAgent 实例。
                agent = AssistantAgent(
                    name="assistant",
                    model_client=model_client,
                    memory=[memory],
                    system_message="你是一个乐于助人的助手。",
                )

                result = await agent.run(task="推荐一个好的晚餐主意?")
                print(result.messages[-1].content)  # type: ignore


            asyncio.run(main())

        .. code-block:: text

            如何制作一份不含奶酪的美味披萨?您可以制作一份带有多种配料的风味蔬菜披萨。这里有一个快速的想法:

            **蔬菜番茄酱披萨**
            - 从披萨饼皮开始(可以购买现成的或自制)。
            - 在饼皮上均匀涂抹一层番茄酱或意面酱。
            - 加上您喜欢的蔬菜,如甜椒、蘑菇、洋葱、橄榄和菠菜。
            - 如果喜欢,可以添加一些蛋白质,如烤鸡肉或意大利辣香肠(确保不含奶酪)。
            - 撒上牛至和罗勒等香草,再淋上一点橄榄油。
            - 根据饼皮说明烘烤,直到边缘呈金黄色且蔬菜熟透。

            搭配沙拉或蒜香面包,完成这顿美餐!祝您用餐愉快!

        **示例 9:使用 `o1-mini` 的代理**

        以下示例展示了如何在助手代理中使用 `o1-mini` 模型。

        .. code-block:: python

            import asyncio
            from autogen_ext.models.openai import OpenAIChatCompletionClient
            from autogen_agentchat.agents import AssistantAgent


            async def main() -> None:
                model_client = OpenAIChatCompletionClient(
                    model="o1-mini",
                    # api_key = "your_openai_api_key"
                )
                # o1 系列模型不支持系统消息。
                agent = AssistantAgent(name="assistant", model_client=model_client, system_message=None)

                result = await agent.run(task="法国的首都是哪里?")
                print(result.messages[-1].content)  # type: ignore


            asyncio.run(main())

        .. note::

            `o1-preview` 和 `o1-mini` 模型不支持系统消息和函数调用。
            因此应将 `system_message` 设置为 `None`,且不应设置 `tools` 和 `handoffs`。
            详情请参阅 `o1 beta 限制 <https://platform.openai.com/docs/guides/reasoning#beta-limitations>`_。

        **示例 10:使用自定义模型上下文的推理模型代理**

        以下示例展示了如何在助手代理中使用推理模型(DeepSeek R1)。
        模型上下文用于过滤掉助手消息中的思考字段。

        .. code-block:: python

            import asyncio
            from typing import List

            from autogen_agentchat.agents import AssistantAgent
            from autogen_core.model_context import UnboundedChatCompletionContext
            from autogen_core.models import AssistantMessage, LLMMessage, ModelFamily
            from autogen_ext.models.ollama import OllamaChatCompletionClient


            class ReasoningModelContext(UnboundedChatCompletionContext):
                \"\"\"用于推理模型的模型上下文。\"\"\"

                async def get_messages(self) -> List[LLMMessage]:
                    messages = await super().get_messages()
                    # 过滤掉 AssistantMessage 中的思考字段。
                    messages_out: List[LLMMessage] = []
                    for message in messages:
                        if isinstance(message, AssistantMessage):
                            message.thought = None
                        messages_out.append(message)
                    return messages_out


            # 为本地 Ollama 上托管的 DeepSeek R1 创建模型客户端实例。
            model_client = OllamaChatCompletionClient(
                model="deepseek-r1:8b",
                model_info={
                    "vision": False,
                    "function_calling": False,
                    "json_output": False,
                    "family": ModelFamily.R1,
                    "structured_output": True,
                },
            )

            agent = AssistantAgent(
                "reasoning_agent",
                model_client=model_client,
                model_context=ReasoningModelContext(),  # 使用自定义模型上下文。
            )


            async def run_reasoning_agent() -> None:
                result = await agent.run(task="法国的首都是哪里?")
                print(result)


            asyncio.run(run_reasoning_agent())

    有关详细示例和用法,请参阅下面的示例部分。
    
import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import MagenticOneGroupChat
from autogen_agentchat.ui import Console


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    assistant = AssistantAgent(
        "Assistant",
        model_client=model_client,
    )
    team = MagenticOneGroupChat([assistant], model_client=model_client)
    await Console(team.run_stream(task="Provide a different proof to Fermat last theorem"))


asyncio.run(main())

tools

import asyncio

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

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")
    writer = AssistantAgent(
        name="writer",
        description="A writer agent for generating text.",
        model_client=model_client,
        system_message="Write well.",
    )
    writer_tool = AgentTool(agent=writer)

    # Create model client with parallel tool calls disabled for the main agent
    main_model_client = OpenAIChatCompletionClient(model="gpt-4.1", parallel_tool_calls=False)
    assistant = AssistantAgent(
        name="assistant",
        model_client=main_model_client,
        tools=[writer_tool],
        system_message="You are a helpful assistant.",
    )
    await Console(assistant.run_stream(task="Write a poem about the sea."))

asyncio.run(main())

team.tools

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import SourceMatchTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.tools import TeamTool
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    # Disable parallel tool calls when using TeamTool
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    writer = AssistantAgent(name="writer", model_client=model_client, system_message="You are a helpful assistant.")
    reviewer = AssistantAgent(
        name="reviewer", model_client=model_client, system_message="You are a critical reviewer."
    )
    summarizer = AssistantAgent(
        name="summarizer",
        model_client=model_client,
        system_message="You combine the review and produce a revised response.",
    )
    team = RoundRobinGroupChat(
        [writer, reviewer, summarizer], termination_condition=SourceMatchTermination(sources=["summarizer"])
    )

    # Create a TeamTool that uses the team to run tasks, returning the last message as the result.
    tool = TeamTool(
        team=team,
        name="writing_team",
        description="A tool for writing tasks.",
        return_value_as_last_message=True,
    )

    # Create model client with parallel tool calls disabled for the main agent
    main_model_client = OpenAIChatCompletionClient(model="gpt-4.1", parallel_tool_calls=False)
    main_agent = AssistantAgent(
        name="main_agent",
        model_client=main_model_client,
        system_message="You are a helpful assistant that can use the writing tool.",
        tools=[tool],
    )
    # For handling each events manually.
    # async for message in main_agent.run_stream(
    #     task="Write a short story about a robot learning to love.",
    # ):
    #     print(message)
    # Use Console to display the messages in a more readable format.
    await Console(
        main_agent.run_stream(
            task="Write a short story about a robot learning to love.",
        )
    )


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

message:

from pydantic import BaseModel
from autogen_agentchat.messages import StructuredMessage


class MyMessageContent(BaseModel):
    text: str
    number: int


message = StructuredMessage[MyMessageContent](
    content=MyMessageContent(text="Hello", number=42),
    source="agent",
    format_string="Hello, {text} {number}!",
)

print(message.to_text())  # Hello, agent 42!

agents:

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


# Global counter state
counter = 0


def increment_counter() -> str:
    """Increment the counter by 1 and return the current value."""
    global counter
    counter += 1
    return f"Counter incremented to: {counter}"


def get_counter() -> str:
    """Get the current counter value."""
    global counter
    return f"Current counter value: {counter}"


async def main() -> None:
    model_client = OpenAIChatCompletionClient(
        model="gpt-4o",
        # api_key = "your_openai_api_key"
    )

    # Create agent with max_tool_iterations=5 to allow multiple tool calls
    agent = AssistantAgent(
        name="assistant",
        model_client=model_client,
        tools=[increment_counter, get_counter],
        max_tool_iterations=5,  # Allow up to 5 tool call iterations
        reflect_on_tool_use=True,  # Get a final summary after tool calls
    )

    await Console(agent.run_stream(task="Increment the counter 3 times and then tell me the final value."))


asyncio.run(main())

agent with memory:

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_core.memory import ListMemory, MemoryContent
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    # Create a model client.
    model_client = OpenAIChatCompletionClient(
        model="gpt-4o-mini",
        # api_key = "your_openai_api_key"
    )

    # Create a list-based memory with some initial content.
    memory = ListMemory()
    await memory.add(MemoryContent(content="User likes pizza.", mime_type="text/plain"))
    await memory.add(MemoryContent(content="User dislikes cheese.", mime_type="text/plain"))

    # Create an AssistantAgent instance with the model client and memory.
    agent = AssistantAgent(
        name="assistant",
        model_client=model_client,
        memory=[memory],
        system_message="You are a helpful assistant.",
    )

    result = await agent.run(task="What is a good dinner idea?")
    print(result.messages[-1].content)  # type: ignore


asyncio.run(main())

MagenticOneGroupChat:

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


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    assistant = AssistantAgent(
        "Assistant",
        model_client=model_client,
    )
    team = MagenticOneGroupChat([assistant], model_client=model_client)
    await Console(team.run_stream(task="Provide a different proof to Fermat last theorem"))


asyncio.run(main())

好的,遵照您的要求,我已经对文章进行了修改和润色。以下是优化后的版本,去除了引用标记和不必要的格式,语言风格更贴近中文技术博客,同时保持了原有的内容结构和长度。


AutoGen生态系统概览:核心组件与分层设计

AutoGen的生态系统是围绕其精巧的分层架构构建的,旨在为开发者提供从底层精细控制到高层快速抽象的灵活选择,让每个人都能找到最适合自己的开发方式。

  • 核心API (autogen_core): 作为整个架构的基石,Core API 负责处理智能体之间最基础的消息传递、事件驱动机制以及本地和分布式环境的运行。它甚至支持Python和.NET的跨语言操作,为构建大规模、高性能的多智能体系统提供了坚实的基础。如果你需要对智能体的行为和通信流进行最精细的控制,那么直接与这一层交互将是你的不二之选。
  • AgentChat API (autogen_agentchat): 这一层建立在Core API之上,提供了一套更高级、更易于上手的接口,是快速原型开发的利器。它封装了许多常见的对话模式,如双智能体对话和群组聊天,使得开发者可以用更少的代码实现复杂的智能体交互逻辑。对于希望快速构建聊天或协作型AI应用的开发者来说,AgentChat API无疑是首选。
  • 扩展API (autogen_ext): 这一层是AutoGen保持活力与先进性的关键,旨在持续扩展框架的功能。它允许轻松集成各种官方和第三方的组件,例如对接OpenAI、AzureOpenAI等不同的大型语言模型客户端,或是集成代码执行环境等外部工具。通过扩展API,AutoGen能够始终与最新的AI技术和外部服务保持同步,确保了框架的生命力与适应性。

除了上述的API层级,AutoGen生态系统还提供了两款极其实用的开发者工具:

  • AutoGen Studio: 这是一个充满未来感的低代码/无代码图形化界面(GUI),用户可以通过简单的拖放操作和直观的配置,就能快速构建、测试和运行多智能体工作流。AutoGen Studio极大地降低了多智能体开发的门槛,使得非编程背景的用户也能参与到AI智能体的设计与部署中,从而极大地加速了原型开发和AI应用的普及。
  • AutoGen Bench: 作为一个专业的基准测试套件,AutoGen Bench为开发者提供了衡量和优化其AI应用性能的科学工具,帮助你的智能体和多智能体系统表现得更加卓越。

这些核心组件和工具共同构成了一个全面且强大的生态系统,使得AutoGen能够从前沿的学术研究到稳健的企业级应用开发,提供全方位的支持。

环境搭建与准备

要开启AutoGen的探索之旅,首先需要配置一个干净、隔离的开发环境,这是项目顺利进行的基础。

Python环境要求与虚拟环境最佳实践

AutoGen框架要求Python版本为3.10或更高。为了避免不同项目之间的依赖库发生冲突,并确保环境的纯净与可复现性,我们强烈建议使用Python虚拟环境来管理AutoGen及其所有依赖项。

使用虚拟环境不仅是Python开发的通用最佳实践,对于AutoGen这样需要集成多种库(如LLM客户端、外部工具,甚至Docker)的复杂AI框架而言,更是至关重要。依赖冲突是一个非常常见且难以诊断的问题。通过在入门阶段就养成使用虚拟环境的习惯,AutoGen旨在帮助用户从一开始就避开“环境地狱”的困扰,从而显著降低上手难度。请记住,在构建和部署复杂AI应用时,优秀的环境管理是确保系统稳定性和可复现性的基石。

友情提示:社区中有一个名为 pyautogen 的库,它由原AutoGen的创建者维护,但与我们现在讨论的 autogen-agentchat 是不同的框架。安装时请务必区分清楚,以避免遇到 AttributeError 这类因包名不匹配而导致的问题。此外,如果你正在从AutoGen v0.2版本升级,请务必参考官方的迁移指南,以确保你的代码和配置能够平滑过渡。

配置API密钥:连接大型语言模型

为了让AutoGen智能体能够与大型语言模型(LLMs)进行交互,你需要配置API密钥。安全、便捷地管理这些密钥,是实际AI应用开发中不可忽视的重要环节。

  • 推荐方式:使用 .env 文件管理:这是最安全、最便捷的方式,可以有效避免将敏感信息(如API密钥)硬编码到代码中。

    1. 创建 .env 文件: 在你的项目根目录下,使用命令行创建一个名为 .env 的文件。
    2. 添加API密钥: 打开这个文件,并添加你的API密钥。例如,如果你使用OpenAI模型,格式如下:
      OPENAI_API_KEY=your-openai-api-key
      
      如果你使用其他第三方服务(如ElevenLabs或Stability AI),也应在此处添加相应的密钥。
    3. 在Python代码中加载: 为了让你的代码能读取到这些密钥,首先需要安装 python-dotenv 库 (pip install python-dotenv)。然后,在你的主代码文件中这样加载:
      import os
      from dotenv import load_dotenv
      
      # 这行代码会自动加载.env文件中的所有环境变量
      load_dotenv() 
      
      # 现在,你可以通过os.getenv()安全地访问API密钥了
      openai_api_key = os.getenv("OPENAI_API_KEY")
      # elevenlabs_api_key = os.getenv("ELEVENLABS_API_KEY")
      ```    这种做法将密钥与代码完全分离可以有效防止在代码共享或提交到版本控制系统如Git时发生密钥泄露同时这也让在不同环境开发测试生产之间切换配置变得异常轻松是构建健壮安全和可维护的生产级AI应用的基础
      
  • 替代方式:直接在代码中配置:你也可以在初始化模型客户端时直接传入API密钥,但这通常不推荐用于生产环境,因为它可能导致敏感信息泄露。 例如:model_client = OpenAIChatCompletionClient(model="gpt-4o", api_key="YOUR_API_KEY")

AutoGen Studio:可视化构建与快速原型

AutoGen Studio是一个专为快速原型设计和运行多智能体工作流而生的低代码/无代码图形界面(GUI)。它的出现,是AutoGen生态系统向更广泛用户群体开放、加速AI应用普及和创新的关键一步。

  • 安装与运行:
    • 安装:在激活的虚拟环境中运行 pip install -U autogenstudio
    • 运行:执行 autogenstudio ui --port 8080 --appdir ./my-app。此命令将在指定端口启动应用,并为你的智能体和工作流配置指定一个存储目录。
    • 启动后,在浏览器中访问 http://127.0.0.1:8080 即可开始使用AutoGen Studio的仪表盘。
  • 核心功能与优势:
    • 用户友好的界面: AutoGen Studio提供直观的拖放界面,让你能够轻松创建多智能体系统并进行实时测试,极大地降低了开发门槛。
    • 团队构建器: 你可以轻松定义和修改智能体工作流,包括创建团队、向团队中添加不同角色的智能体、为智能体配置模型和工具,以及定义团队的完成条件。
    • 组件画廊 (Gallery): 提供一个丰富的组件库,用户可以在这里分享和重用已经定义好的团队、智能体、模型和工具配置,极大地促进了社区内的协作与创新。
    • 实验场 (Playground): 提供一个交互式环境,你可以在这里测试智能体团队在特定任务上的表现,审查它们生成的产物(如图像、代码、文本),甚至监控团队的“内心独白”和执行过程,从而全方位地评估和优化你的AI应用。

AutoGen Studio不仅仅是Python库的一个附属品,它是一个独立且功能丰富的GUI工具。这一设计明确地表明,AutoGen致力于通过降低编程的复杂性来拥抱更广泛的用户群体。它提供了一个从开发、测试到分享的完整闭环体验,不仅加速了原型开发和概念验证,也为AI应用从实验室走向更广阔的业务场景铺平了道路。

核心概念与基本使用

理解AutoGen的核心概念是构建出色多智能体应用的关键。本节将深入探讨AutoGen中不同类型的智能体、它们如何通过对话协作,以及如何通过工具和代码执行来扩展它们的能力。

智能体类型:AutoGen的“角色”

AutoGen提供了一套丰富的预设智能体,每个智能体都像一个拥有特定技能和个性的“角色”。所有这些智能体都派生自一个共同的 ConversableAgent 基类,因此它们都拥有一些核心的属性和方法,如唯一的名称、角色描述,以及执行任务的 runrun_stream 方法。AutoGen智能体的多样性与高度可定制性,是其实现复杂任务分解和专业化协作的基础,完美体现了“分而治之”的AI设计理念。

AssistantAgent:通用AI助手

AssistantAgent 是AutoGen中一个内置的、功能强大的智能体,它能够使用语言模型进行思考,并具备调用外部工具来执行任务的能力。官方将其描述为“用于原型设计和教学的‘万能’智能体”,因其高度的通用性而广受欢迎。

AssistantAgent 提供了广泛的定制选项,以适应各种任务需求:

  • 工具集成: 它可以调用外部工具来执行特定操作,如从API获取数据或查询数据库。这些工具可以是简单的Python函数,也可以是一组共享状态的 Workbench
  • 并行工具调用: 如果底层的语言模型支持,AssistantAgent 默认会并行调用多个工具以提高效率。
  • 结构化输出: AssistantAgent 能够按照预定义的格式(如Pydantic模型)返回结构化的JSON文本,这使得它的响应可以直接被下游系统作为结构化对象使用。
  • 上下文管理: 你可以精细地控制发送给语言模型的对话历史长度,例如只发送最近的N条消息,或根据Token数量进行限制,以优化成本和性能。

UserProxyAgent:用户代理与代码执行

UserProxyAgent 是一个非常特殊的智能体,它扮演着人类用户的“代理人”角色,负责接收你的输入,并与其他智能体进行沟通。它在多智能体协作中是至关重要的“人机接口”。

UserProxyAgent 的核心功能包括:

  • 代码执行: UserProxyAgent 能够执行其他智能体(特别是 AssistantAgent)生成的代码。它可以被配置在本地环境或更安全的Docker容器中执行代码,这是AutoGen实现自动化编程和解决问题的关键一环。在许多应用场景中,它接收LLM生成的代码,执行后将结果(无论成功还是失败)反馈给生成代码的智能体,从而形成一个自动化的“生成-执行-调试”闭环。
  • 人类输入与干预: 你可以配置 UserProxyAgent 在特定条件下暂停对话,等待你的输入或批准。这使得人类能够无缝地参与到智能体的工作流中,提供关键的指导或进行必要的修正。

其他预设智能体简介

除了上述两种核心智能体,AutoGen还提供了许多其他预设的“专家”角色:

  • CodeExecutorAgent: 专门用于执行代码的智能体,更侧重于自动化执行。
  • SocietyOfMindAgent: 一个能够管理“心智社会”的智能体,用于处理更复杂的任务和组织结构。
  • OpenAIAssistantAgent: 一个由OpenAI Assistant API驱动的智能体,能够利用OpenAI平台提供的各种强大功能。
  • MultimodalWebSurfer: 一个多模态智能体,能够像人一样浏览网页、理解内容并提取信息。
  • FileSurferVideoSurfer: 分别用于在本地文件中搜索信息和“观看”视频以提取内容的智能体。
智能体类型描述主要用途核心能力示例应用场景
AssistantAgent通用AI助手,可调用LLM和工具。任务规划、代码生成、信息处理。LLM交互、工具调用、结构化输出、上下文管理。软件开发助手、数据分析师、内容创作者。
UserProxyAgent代理人类用户,接收输入,执行代码。人机交互、代码执行与调试、人类反馈集成。用户输入处理、代码解释器、终止条件管理。自动化编程、交互式数据探索、审批流程。
CodeExecutorAgent专门用于执行代码。安全、隔离地执行代码。代码执行(本地/Docker)、结果反馈。自动化脚本运行、测试环境。
MultimodalWebSurfer多模态智能体,可搜索和浏览网页。网页信息提取、在线研究。网页浏览、内容理解、信息检索。市场调研、新闻摘要、竞品分析。
OpenAIAssistantAgent基于OpenAI Assistant API的智能体。利用OpenAI平台高级功能。OpenAI API集成、自定义工具使用。复杂问答、知识库查询。
SocietyOfMindAgent管理“心智社会”的智能体。复杂任务的组织与协调。智能体团队管理、任务分解。复杂项目管理、多领域问题解决。

这种“乐高积木”式的智能体设计,使得AutoGen能够灵活应对从简单问答到复杂软件开发流程的各种场景。通过组合和定制不同功能的智能体,开发者可以构建出高度专业化、高效且适应性强的AI系统。

多智能体对话机制:智能体如何协作

AutoGen的核心魅力在于其强大的多智能体对话能力。智能体之间通过消息传递进行通信,像一个真正的团队一样协作解决复杂任务。这种对话机制是AutoGen实现自主问题解决和工作流自动化的基础。

一对一对话:基础交互

一对一对话是最简单、最基础的交互模式,通常涉及两个智能体之间的直接交流。例如,一个 AssistantAgent 可以与一个 UserProxyAgent 对话,以响应用户查询或执行特定任务。

示例代码结构:

import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def simple_chat_example():
    # 假设已配置好API密钥
    model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
    assistant = AssistantAgent("assistant", model_client=model_client)
    user_proxy = UserProxyAgent("user_proxy", human_input_mode="NEVER") # 示例中设置为无人干预

    # 启动对话
    response = await user_proxy.run(
        recipient=assistant, 
        task="请帮我写一首关于海洋的四行诗。"
    )
    print(response)
    
    await model_client.close()

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

此代码展示了 UserProxyAgent 如何向 AssistantAgent 发送一条消息,并启动一个简单的对话。

群组对话:复杂任务的协作模式

群组对话是AutoGen中一种更高级的协作模式,它允许多个智能体在一个共享的聊天室中进行交流,非常适合需要多角色协作和动态任务分解的复杂场景。

在群组对话中,每个智能体通常都专注于特定任务,就像一个人类团队:可能有负责撰写文本的“作家”,负责生成图像的“插画师”,以及负责审查和指导的“编辑”。人类用户也可以作为团队的一员参与对话,提供指导或在关键时刻进行干预。

群组对话的核心由一个特殊的“群聊管理器”(Group Chat Manager)智能体来协调。它负责决定在任何时刻应该由哪个智能体发言。这种选择机制可以是简单的轮流发言,也可以是利用LLM根据对话上下文动态地决定最佳发言者,使其更接近人类团队的自然工作方式。这种模式通过智能体之间的专业分工和协调,能够处理单一智能体无法完成的复杂任务,例如完整的软件开发流程或多媒体内容创作。

工具与函数调用:扩展智能体能力

大型语言模型虽然强大,但它们本身无法直接与外部世界互动。为了弥补这一局限,AutoGen的智能体可以利用外部工具来执行特定动作,例如查询最新的天气信息、操作数据库或调用任何第三方API。这种函数调用机制是弥补LLM内在局限性、实现与真实世界交互的关键桥梁。

现代LLMs能够理解你提供给它的工具列表(包括每个工具的功能描述和参数),并根据任务需求,智能地决定何时以及如何调用这些工具。

定义与注册自定义工具

在AutoGen中,创建一个自定义工具可以像定义一个普通的Python函数一样简单。AssistantAgent 会自动分析你的函数签名和文档字符串,并将其转换成LLM能够理解的格式。为函数的参数和返回值添加清晰的类型注解和描述,对于帮助LLM正确使用你的工具至关重要。

Workbench 与内置工具

Workbench 是一个工具的集合,它们可以共享状态和资源,非常适合管理一组相关的工具。此外,AutoGen还提供了多种内置工具,如 http(用于发送网络请求)、langchain(用于集成LangChain生态)等,进一步扩展了智能体的能力。

智能体作为工具

AutoGen的一项高级功能是,你可以将一个智能体本身封装成一个工具,供另一个智能体调用。这使得构建动态的、由模型驱动的多智能体工作流成为可能,其中一个“主管”智能体可以将子任务“委托”给其他“专家”智能体来完成。这极大地拓宽了AI应用的边界,使其能够真正融入到复杂的业务流程中。

代码生成、执行与调试:自动化编程

AutoGen的一项标志性能力是其智能体能够自动生成、执行和调试代码。这对于自动化软件工程、数据分析和科学计算等任务至关重要,能够显著减少人工干预并加速开发周期。

智能体生成、运行和修正代码的流程

在一个典型的工作流中,一个LLM驱动的智能体(如 AssistantAgent)会根据任务需求生成代码。然后,这些代码会被交给另一个智能体(通常是 UserProxyAgentCodeExecutorAgent)来执行。

执行结果,无论是成功输出还是错误信息,都会被反馈给生成代码的智能体。基于这些反馈,智能体能够识别问题、诊断错误,并尝试修正代码,然后再次执行。这个“生成-执行-反馈-调试”的自动化循环是AutoGen实现自主解决问题能力的核心。它使得智能体不仅能提出解决方案,还能主动验证方案的有效性,并在遇到问题时进行迭代优化,直至成功。

本地与Docker环境下的代码执行

为了安全起见,AutoGen默认在隔离的Docker容器中执行由LLM生成的代码,以避免对你的本地系统造成潜在风险。当然,你也可以根据需要将其配置为在本地执行,但这通常不推荐用于生产环境。你甚至可以完全禁用某个智能体的代码执行能力,以增强安全性。这种在实用性与安全性之间的权衡考量,为运行不确定的AI生成代码提供了必要的保护。

人类参与:Human-in-the-Loop (HIL)

AutoGen框架的一个显著特点是它能够无缝地支持人类参与,允许你在智能体工作流的任何阶段提供输入和反馈。这种“人类在环”(Human-in-the-Loop, HIL)机制确保了AI系统在处理复杂或关键任务时,能够得到人类的监督、指导和纠正,从而大大提高系统的可靠性和准确性。

运行时反馈:即时干预

这种方式允许你在智能体团队执行任务的过程中进行即时反馈。当团队需要人类决策时,UserProxyAgent 会暂停整个流程,等待你的输入。一旦你提供了反馈,团队就会根据你的新指令继续执行。

需要注意的是,这种方式是阻塞式的,意味着团队的执行会完全停止直到你回应。因此,它最适合需要即时决策的短交互,例如批准一个操作,或在紧急情况下需要你立即关注的场景。

跨运行反馈:异步协作

这种方式更像是一种异步的、回合制的协作。智能体团队完成一次运行并终止后,你可以从容地审查其工作,提供反馈,然后让团队带着新的反馈再次启动。这种模式特别适用于需要持久化会话和异步沟通的场景。

你可以通过设置最大对话轮次(max_turns)或定义特定的终止条件(如 HandoffTermination,即智能体主动请求人类接管)来实现这种模式。这使得智能体能够“记住”之前的上下文,即使在长时间中断后也能继续工作,非常适合构建需要持续用户参与的聊天机器人或复杂的项目管理助手。

模式实现方式特点适用场景优点缺点
运行时反馈UserProxyAgent 在任务执行期间被调用。阻塞式:团队暂停执行,等待人类输入。需要即时决策的短交互,如审批、紧急警报。确保人类即时干预和控制。阻塞团队执行,可能导致不稳定状态。
跨运行反馈使用 max_turns 或特定终止条件。非阻塞式:团队完成一次运行后终止,等待人类反馈后再次启动。需要持久化会话和异步通信的场景。允许异步协作,支持状态持久化,更灵活。需要额外的状态管理机制。

这种灵活的HIL机制是AutoGen在企业级应用中实现落地和广泛采纳的关键。它允许企业在AI自动化和人类监督之间找到最佳平衡点,让AI系统更好地融入现有业务流程,实现真正的智能自动化。

结论

AutoGen作为一个由微软及合作高校共同打造的前沿框架,在构建下一代大型语言模型应用方面展现出了卓越的能力和远见。通过其独特的多智能体对话机制,AutoGen将AI应用开发从传统的单点智能范式提升至协作智能的新高度,有效解决了单一LLM在处理复杂、多步骤任务时所面临的局限性。

该框架的分层与可扩展设计是其核心优势。从提供坚实基础的Core API,到加速原型开发的AgentChat API,再到确保与时俱进的Extensions API,这种模块化和开放性使得AutoGen能够灵活适应飞速发展的AI技术格局。与此同时,AutoGen Studio的引入更是极大地降低了多智能体开发的门槛,让更广泛的创新者能够参与到这场AI变革中。

AutoGen在工具集成和代码执行方面的能力尤为突出,它让智能体能够突破语言模型的限制,与真实世界交互,甚至形成自主“生成-执行-调试”的闭环,这在自动化软件开发、数据分析等领域具有革命性的意义。此外,其对“人类在环”的无缝支持,确保了AI系统在处理关键任务时,始终能够得到人类的监督和指导,兼顾了效率与可靠性。

综上所述,AutoGen不仅仅是一个LLM编排工具,它提供了一个全面的生态系统,赋能开发者构建能够自主协作、解决复杂问题并与人类无缝交互的智能系统。它为自动化数据分析、软件工程、智能客服等众多领域带来了巨大的潜力,并有望成为未来智能自动化和Agentic AI领域的基石框架。