从v0.2迁移到v0.4的指南#

这是为使用v0.2.*版本autogen-agentchat的用户准备的迁移指南，我们将引导他们升级到v0.4版本，该版本引入了一系列新的API和功能。v0.4版本包含了一些重大变更，请仔细阅读本指南。我们仍在0.2分支中维护v0.2版本；然而，我们强烈建议您升级到v0.4版本。

注意

我们不再拥有对pyautogen PyPI包的管理员访问权限，并且自0.2.34版本以来，该包的发布不再来自微软。要继续使用AutoGen的v0.2版本，请使用autogen-agentchat~=0.2进行安装。请阅读我们关于分叉的澄清声明。

什么是 v0.4？#

自从2023年发布AutoGen以来，我们深入聆听了来自小型初创公司和大型企业的社区和用户的反馈，收集了大量意见。基于这些反馈，我们构建了AutoGen v0.4，这是一个从头开始的重写版本，采用了异步、事件驱动的架构，以解决可观察性、灵活性、交互控制和规模等问题。

v0.4 API是分层的： Core API是基础层，提供了一个可扩展的、事件驱动的参与者框架，用于创建代理工作流； AgentChat API建立在Core之上，提供了一个任务驱动的高级框架，用于构建交互式代理应用程序。它是AutoGen v0.2的替代品。

本指南的大部分内容都集中在v0.4的AgentChat API上；然而，你也可以仅使用Core API构建自己的高级框架。

初次使用AutoGen?#

直接跳转到AgentChat教程开始使用v0.4。

本指南包含哪些内容？#

我们提供了一个详细的指南，介绍如何将现有的代码库从v0.2迁移到v0.4。

查看以下每个功能的详细信息，了解如何迁移。

目前处于 v0.2 的以下功能将在未来的 v0.4.* 版本中提供：

  • 模型客户端成本 #4835

  • 可教学代理

  • RAG代理

我们将在缺失的功能可用时更新本指南。

模型客户端#

在 v0.2 中，您可以按如下方式配置模型客户端，并创建 OpenAIWrapper 对象。

from autogen.oai import OpenAIWrapper

config_list = [
    {"model": "gpt-4o", "api_key": "sk-xxx"},
    {"model": "gpt-4o-mini", "api_key": "sk-xxx"},
]

model_client = OpenAIWrapper(config_list=config_list)

注意: 在 AutoGen 0.2 中，OpenAI 客户端会尝试列表中的配置，直到找到一个可用的配置。而在 0.4 版本中，期望选择一个特定的模型配置。

在 v0.4 中，我们提供两种创建模型客户端的方式。

使用组件配置#

AutoGen 0.4 拥有一个 通用组件配置系统。模型客户端是一个很好的使用案例。请参见以下内容，了解如何创建一个 OpenAI 聊天完成客户端。


from autogen_core.models import ChatCompletionClient

config = {
    "provider": "OpenAIChatCompletionClient",
    "config": {
        "model": "gpt-4o",
        "api_key": "sk-xxx" # os.environ["...']
    }
}

model_client = ChatCompletionClient.load_component(config)

直接使用模型客户端类#

开放AI:

from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o", api_key="sk-xxx")

Azure OpenAI:

from autogen_ext.models.openai import AzureOpenAIChatCompletionClient

model_client = AzureOpenAIChatCompletionClient(
    azure_deployment="gpt-4o",
    azure_endpoint="https://<your-endpoint>.openai.azure.com/",
    model="gpt-4o",
    api_version="2024-09-01-preview",
    api_key="sk-xxx",
)

了解更多关于 OpenAIChatCompletionClient。

适用于OpenAI兼容API的模型客户端#

你可以使用 OpenAIChatCompletionClient 来连接到一个 OpenAI 兼容的 API，但你需要指定 base_url 和 model_info。

from autogen_ext.models.openai import OpenAIChatCompletionClient

custom_model_client = OpenAIChatCompletionClient(
    model="custom-model-name",
    base_url="https://custom-model.com/reset/of/the/path",
    api_key="placeholder",
    model_info={
        "vision": True,
        "function_calling": True,
        "json_output": True,
        "family": "unknown",
    },
)

注意: 我们不会测试所有与OpenAI兼容的API，尽管它们可能声称支持，但许多API的工作方式与OpenAI API不同。请在使用前进行测试。

阅读有关Model Clients 在AgentChat教程中的内容，以及更多详细信息请查看Core API Docs

未来将添加对其他托管模型的支持。

模型客户端缓存#

在v0.2中，您可以通过LLM配置中的cache_seed参数设置缓存种子。 默认情况下，缓存是启用的。

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
    "cache_seed": 42,
}

在v0.4中，默认情况下未启用缓存，要使用它，您需要在模型客户端周围使用ChatCompletionCache包装器。

你可以使用DiskCacheStore或RedisStore来存储缓存。

pip install -U "autogen-ext[openai, diskcache, redis]"

以下是使用 diskcache 进行本地缓存的示例：

import asyncio
import tempfile

from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.models.cache import ChatCompletionCache, CHAT_CACHE_VALUE_TYPE
from autogen_ext.cache_store.diskcache import DiskCacheStore
from diskcache import Cache


async def main():
    with tempfile.TemporaryDirectory() as tmpdirname:
        # Initialize the original client
        openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")

        # Then initialize the CacheStore, in this case with diskcache.Cache.
        # You can also use redis like:
        # from autogen_ext.cache_store.redis import RedisStore
        # import redis
        # redis_instance = redis.Redis()
        # cache_store = RedisCacheStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
        cache_store = DiskCacheStore[CHAT_CACHE_VALUE_TYPE](Cache(tmpdirname))
        cache_client = ChatCompletionCache(openai_model_client, cache_store)

        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # Should print response from OpenAI
        response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
        print(response)  # Should print cached response


asyncio.run(main())

助手代理#

在 v0.2 中，您可以按以下方式创建一个助理代理：

from autogen.agentchat import AssistantAgent

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant.",
    llm_config=llm_config,
)

在v0.4中，它类似，但你需要指定model_client而不是llm_config。

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

model_client = OpenAIChatCompletionClient(model="gpt-4o", api_key="sk-xxx", seed=42, temperature=0)

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant.",
    model_client=model_client,
)

然而，使用方式有些不同。在 v0.4 中，不是调用 assistant.send，而是调用 assistant.on_messages 或 assistant.on_messages_stream 来处理传入的消息。此外，on_messages 和 on_messages_stream 方法是异步的，后者返回一个异步生成器以流式传输代理的内部思考。

以下是如何直接从上面的示例中调用v0.4版本的助手代理：

import asyncio
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
    )

    cancellation_token = CancellationToken()
    response = await assistant.on_messages([TextMessage(content="Hello!", source="user")], cancellation_token)
    print(response)

asyncio.run(main())

CancellationToken 可用于在调用 cancellation_token.cancel() 时异步取消请求，这会导致 on_messages 调用中的 await 引发 CancelledError。

了解更多关于 代理教程 和 AssistantAgent的内容。

多模态代理#

v0.4 中的 AssistantAgent 如果模型客户端支持，它支持多模态输入。 模型客户端的 vision 功能用于确定代理是否支持多模态输入。

import asyncio
from pathlib import Path
from autogen_agentchat.messages import MultiModalMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken, Image
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
    )

    cancellation_token = CancellationToken()
    message = MultiModalMessage(
        content=["Here is an image:", Image.from_file(Path("test.png"))],
        source="user",
    )
    response = await assistant.on_messages([message], cancellation_token)
    print(response)

asyncio.run(main())

用户代理#

在 v0.2 中，你可以按如下方式创建一个用户代理：

from autogen.agentchat import UserProxyAgent

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config=False,
    llm_config=False,
)

这个用户代理将通过控制台接收用户的输入，并在传入的消息以“TERMINATE”结尾时终止。

在 v0.4 中，用户代理只是一个仅接受用户输入的代理，不需要其他特殊配置。您可以按如下方式创建用户代理：

from autogen_agentchat.agents import UserProxyAgent

user_proxy = UserProxyAgent("user_proxy")

参见UserProxyAgent了解更多详情以及如何使用超时功能自定义输入函数。

可对话代理和注册回复#

在 v0.2 中，您可以创建一个可对话的代理并注册一个回复函数，如下所示：

from typing import Any, Dict, List, Optional, Tuple, Union
from autogen.agentchat import ConversableAgent

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

conversable_agent = ConversableAgent(
    name="conversable_agent",
    system_message="You are a helpful assistant.",
    llm_config=llm_config,
    code_execution_config={"work_dir": "coding"},
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
)

def reply_func(
    recipient: ConversableAgent,
    messages: Optional[List[Dict]] = None,
    sender: Optional[Agent] = None,
    config: Optional[Any] = None,
) -> Tuple[bool, Union[str, Dict, None]]:
    # Custom reply logic here
    return True, "Custom reply"

# Register the reply function
conversable_agent.register_reply([ConversableAgent], reply_func, position=0)

# NOTE: An async reply function will only be invoked with async send.

与其猜测reply_func的作用、其所有参数，以及position在v0.4中应该是什么，我们可以简单地创建一个自定义代理并实现on_messages、on_reset和produced_message_types方法。

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

class CustomAgent(BaseChatAgent):
    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        return Response(chat_message=TextMessage(content="Custom reply", source=self.name))

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

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

然后你可以像使用AssistantAgent一样使用自定义代理。 更多详情请参见自定义代理教程。

保存和加载代理状态#

在 v0.2 版本中，没有内置的方法来保存和加载代理的状态：你需要通过导出 ConversableAgent 的 chat_messages 属性，并通过 chat_messages 参数将其导入来实现这个功能。

在v0.4版本中，您可以在代理上调用save_state和load_state方法来保存和加载它们的状态。

import asyncio
import json
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
    )

    cancellation_token = CancellationToken()
    response = await assistant.on_messages([TextMessage(content="Hello!", source="user")], cancellation_token)
    print(response)

    # Save the state.
    state = await assistant.save_state()

    # (Optional) Write state to disk.
    with open("assistant_state.json", "w") as f:
        json.dump(state, f)

    # (Optional) Load it back from disk.
    with open("assistant_state.json", "r") as f:
        state = json.load(f)
        print(state) # Inspect the state, which contains the chat history.

    # Carry on the chat.
    response = await assistant.on_messages([TextMessage(content="Tell me a joke.", source="user")], cancellation_token)
    print(response)

    # Load the state, resulting the agent to revert to the previous state before the last message.
    await assistant.load_state(state)

    # Carry on the same chat again.
    response = await assistant.on_messages([TextMessage(content="Tell me a joke.", source="user")], cancellation_token)

asyncio.run(main())

你也可以在任何团队上调用save_state和load_state，例如RoundRobinGroupChat 来保存和加载整个团队的状态。

双代理聊天#

在 v0.2 中，您可以按如下方式创建一个用于代码执行的双代理聊天：

from autogen.coding import LocalCommandLineCodeExecutor
from autogen.agentchat import AssistantAgent, UserProxyAgent

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

assistant = AssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant. Write all code in python. Reply only 'TERMINATE' if the task is done.",
    llm_config=llm_config,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"code_executor": LocalCommandLineCodeExecutor(work_dir="coding")},
    llm_config=False,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

chat_result = user_proxy.initiate_chat(assistant, message="Write a python script to print 'Hello, world!'")
# Intermediate messages are printed to the console directly.
print(chat_result)

要在v0.4中获得相同的行为，你可以在RoundRobinGroupChat中同时使用AssistantAgent和CodeExecutorAgent。

import asyncio
from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant. Write all code in python. Reply only 'TERMINATE' if the task is done.",
        model_client=model_client,
    )

    code_executor = CodeExecutorAgent(
        name="code_executor",
        code_executor=LocalCommandLineCodeExecutor(work_dir="coding"),
    )

    # The termination condition is a combination of text termination and max message termination, either of which will cause the chat to terminate.
    termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(10)

    # The group chat will alternate between the assistant and the code executor.
    group_chat = RoundRobinGroupChat([assistant, code_executor], termination_condition=termination)

    # `run_stream` returns an async generator to stream the intermediate messages.
    stream = group_chat.run_stream(task="Write a python script to print 'Hello, world!'")
    # `Console` is a simple UI to display the stream.
    await Console(stream)

asyncio.run(main())

工具使用#

在v0.2中，要创建一个使用工具聊天的机器人，你必须有两个代理，一个用于调用工具，另一个用于执行工具。 你需要为每个用户请求发起一个双代理聊天。

from autogen.agentchat import AssistantAgent, UserProxyAgent, register_function

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

tool_caller = AssistantAgent(
    name="tool_caller",
    system_message="You are a helpful assistant. You can call tools to help user.",
    llm_config=llm_config,
    max_consecutive_auto_reply=1, # Set to 1 so that we return to the application after each assistant reply as we are building a chatbot.
)

tool_executor = UserProxyAgent(
    name="tool_executor",
    human_input_mode="NEVER",
    code_execution_config=False,
    llm_config=False,
)

def get_weather(city: str) -> str:
    return f"The weather in {city} is 72 degree and sunny."

# Register the tool function to the tool caller and executor.
register_function(get_weather, caller=tool_caller, executor=tool_executor)

while True:
    user_input = input("User: ")
    if user_input == "exit":
        break
    chat_result = tool_executor.initiate_chat(
        tool_caller,
        message=user_input,
        summary_method="reflection_with_llm", # To let the model reflect on the tool use, set to "last_msg" to return the tool call result directly.
    )
    print("Assistant:", chat_result.summary)

在v0.4中，你实际上只需要一个代理——AssistantAgent来处理工具调用和工具执行。

import asyncio
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage

def get_weather(city: str) -> str: # Async tool is possible too.
    return f"The weather in {city} is 72 degree and sunny."

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)
    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant. You can call tools to help user.",
        model_client=model_client,
        tools=[get_weather],
        reflect_on_tool_use=True, # Set to True to have the model reflect on the tool use, set to False to return the tool call result directly.
    )
    while True:
        user_input = input("User: ")
        if user_input == "exit":
            break
        response = await assistant.on_messages([TextMessage(content=user_input, source="user")], CancellationToken())
        print("Assistant:", response.chat_message.content)

asyncio.run(main())

在群聊中使用配备了工具的代理，如 RoundRobinGroupChat， 您只需按照上述方法将工具添加到代理中，并与这些代理创建一个群聊。

聊天结果#

在 v0.2 中，你可以从 initiate_chat 方法获得一个 ChatResult 对象。例如：

chat_result = tool_executor.initiate_chat(
    tool_caller,
    message=user_input,
    summary_method="reflection_with_llm",
)
print(chat_result.summary) # Get LLM-reflected summary of the chat.
print(chat_result.chat_history) # Get the chat history.
print(chat_result.cost) # Get the cost of the chat.
print(chat_result.human_input) # Get the human input solicited by the chat.

查看 ChatResult 文档 以获取更多详情。

在v0.4中，你可以从run或run_stream方法中获取一个TaskResult对象。TaskResult对象包含messages，这是聊天消息的历史记录，包括两个代理的私人消息（如工具调用等）和公共消息。

在 TaskResult 和 ChatResult 之间存在一些显著的差异：

  • TaskResult中的messages列表使用的消息格式与ChatResult.chat_history列表不同。

  • 没有summary字段。应用程序需要根据messages列表自行决定如何总结聊天的内容。

  • human_input 未在 TaskResult 对象中提供，因为用户输入可以通过使用 source 字段过滤从 messages 列表中提取。

  • cost 未在 TaskResult 对象中提供，但您可以根据 token 使用情况来计算成本。添加成本计算将是一个很好的社区扩展。请参阅 社区扩展。

v0.2 和 v0.4 消息之间的转换#

你可以使用以下转换函数在autogen_agentchat.base.TaskResult.messages中的v0.4消息和ChatResult.chat_history中的v0.2消息之间进行转换。

from typing import Any, Dict, List, Literal

from autogen_agentchat.messages import (
    AgentEvent,
    ChatMessage,
    HandoffMessage,
    MultiModalMessage,
    StopMessage,
    TextMessage,
    ToolCallExecutionEvent,
    ToolCallRequestEvent,
    ToolCallSummaryMessage,
)
from autogen_core import FunctionCall, Image
from autogen_core.models import FunctionExecutionResult


def convert_to_v02_message(
    message: AgentEvent | ChatMessage,
    role: Literal["assistant", "user", "tool"],
    image_detail: Literal["auto", "high", "low"] = "auto",
) -> Dict[str, Any]:
    """Convert a v0.4 AgentChat message to a v0.2 message.

    Args:
        message (AgentEvent | ChatMessage): The message to convert.
        role (Literal["assistant", "user", "tool"]): The role of the message.
        image_detail (Literal["auto", "high", "low"], optional): The detail level of image content in multi-modal message. Defaults to "auto".

    Returns:
        Dict[str, Any]: The converted AutoGen v0.2 message.
    """
    v02_message: Dict[str, Any] = {}
    if isinstance(message, TextMessage | StopMessage | HandoffMessage | ToolCallSummaryMessage):
        v02_message = {"content": message.content, "role": role, "name": message.source}
    elif isinstance(message, MultiModalMessage):
        v02_message = {"content": [], "role": role, "name": message.source}
        for modal in message.content:
            if isinstance(modal, str):
                v02_message["content"].append({"type": "text", "text": modal})
            elif isinstance(modal, Image):
                v02_message["content"].append(modal.to_openai_format(detail=image_detail))
            else:
                raise ValueError(f"Invalid multimodal message content: {modal}")
    elif isinstance(message, ToolCallRequestEvent):
        v02_message = {"tool_calls": [], "role": "assistant", "content": None, "name": message.source}
        for tool_call in message.content:
            v02_message["tool_calls"].append(
                {
                    "id": tool_call.id,
                    "type": "function",
                    "function": {"name": tool_call.name, "args": tool_call.arguments},
                }
            )
    elif isinstance(message, ToolCallExecutionEvent):
        tool_responses: List[Dict[str, str]] = []
        for tool_result in message.content:
            tool_responses.append(
                {
                    "tool_call_id": tool_result.call_id,
                    "role": "tool",
                    "content": tool_result.content,
                }
            )
        content = "\n\n".join([response["content"] for response in tool_responses])
        v02_message = {"tool_responses": tool_responses, "role": "tool", "content": content}
    else:
        raise ValueError(f"Invalid message type: {type(message)}")
    return v02_message


def convert_to_v04_message(message: Dict[str, Any]) -> AgentEvent | ChatMessage:
    """Convert a v0.2 message to a v0.4 AgentChat message."""
    if "tool_calls" in message:
        tool_calls: List[FunctionCall] = []
        for tool_call in message["tool_calls"]:
            tool_calls.append(
                FunctionCall(
                    id=tool_call["id"],
                    name=tool_call["function"]["name"],
                    arguments=tool_call["function"]["args"],
                )
            )
        return ToolCallRequestEvent(source=message["name"], content=tool_calls)
    elif "tool_responses" in message:
        tool_results: List[FunctionExecutionResult] = []
        for tool_response in message["tool_responses"]:
            tool_results.append(
                FunctionExecutionResult(
                    call_id=tool_response["tool_call_id"],
                    content=tool_response["content"],
                    is_error=False,
                    name=tool_response["name"],
                )
            )
        return ToolCallExecutionEvent(source="tools", content=tool_results)
    elif isinstance(message["content"], list):
        content: List[str | Image] = []
        for modal in message["content"]:  # type: ignore
            if modal["type"] == "text":  # type: ignore
                content.append(modal["text"])  # type: ignore
            else:
                content.append(Image.from_uri(modal["image_url"]["url"]))  # type: ignore
        return MultiModalMessage(content=content, source=message["name"])
    elif isinstance(message["content"], str):
        return TextMessage(content=message["content"], source=message["name"])
    else:
        raise ValueError(f"Unable to convert message: {message}")

群聊#

在 v0.2 中，你需要创建一个 GroupChat 类并将其传递给 GroupChatManager，并需要一个作为用户代理的参与者来启动聊天。对于一个简单的作家和评论家的场景，你可以按照以下步骤操作：

from autogen.agentchat import AssistantAgent, GroupChat, GroupChatManager

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-xxx"}],
    "seed": 42,
    "temperature": 0,
}

writer = AssistantAgent(
    name="writer",
    description="A writer.",
    system_message="You are a writer.",
    llm_config=llm_config,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("APPROVE"),
)

critic = AssistantAgent(
    name="critic",
    description="A critic.",
    system_message="You are a critic, provide feedback on the writing. Reply only 'APPROVE' if the task is done.",
    llm_config=llm_config,
)

# Create a group chat with the writer and critic.
groupchat = GroupChat(agents=[writer, critic], messages=[], max_round=12)

# Create a group chat manager to manage the group chat, use round-robin selection method.
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config, speaker_selection_method="round_robin")

# Initiate the chat with the editor, intermediate messages are printed to the console directly.
result = editor.initiate_chat(
    manager,
    message="Write a short story about a robot that discovers it has feelings.",
)
print(result.summary)

在 v0.4 中，你可以使用 RoundRobinGroupChat 来实现相同的行为。

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

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    writer = AssistantAgent(
        name="writer",
        description="A writer.",
        system_message="You are a writer.",
        model_client=model_client,
    )

    critic = AssistantAgent(
        name="critic",
        description="A critic.",
        system_message="You are a critic, provide feedback on the writing. Reply only 'APPROVE' if the task is done.",
        model_client=model_client,
    )

    # The termination condition is a text termination, which will cause the chat to terminate when the text "APPROVE" is received.
    termination = TextMentionTermination("APPROVE")

    # The group chat will alternate between the writer and the critic.
    group_chat = RoundRobinGroupChat([writer, critic], termination_condition=termination, max_turns=12)

    # `run_stream` returns an async generator to stream the intermediate messages.
    stream = group_chat.run_stream(task="Write a short story about a robot that discovers it has feelings.")
    # `Console` is a simple UI to display the stream.
    await Console(stream)

asyncio.run(main())

对于基于LLM的说话者选择，你可以使用SelectorGroupChat代替。 详情请参阅选择器群聊教程 和SelectorGroupChat。

注意: 在 v0.4 中，您不需要在用户代理上注册函数来在群聊中使用工具。您只需将工具函数传递给 AssistantAgent，如 工具使用 部分所示。 代理将在需要时自动调用工具。 如果您的工具没有输出格式良好的响应，您可以使用 reflect_on_tool_use 参数让模型反思工具的使用。

带简历的群聊#

在v0.2中，恢复群聊有点复杂。你需要明确保存群聊消息，并在想要恢复聊天时重新加载它们。详情请参见Resuming Group Chat in v0.2。

在v0.4中，您可以简单地使用同一个群聊对象再次调用run或run_stream来恢复聊天。为了导出和加载状态，您可以使用save_state和load_state方法。

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

def create_team() -> RoundRobinGroupChat:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    writer = AssistantAgent(
        name="writer",
        description="A writer.",
        system_message="You are a writer.",
        model_client=model_client,
    )

    critic = AssistantAgent(
        name="critic",
        description="A critic.",
        system_message="You are a critic, provide feedback on the writing. Reply only 'APPROVE' if the task is done.",
        model_client=model_client,
    )

    # The termination condition is a text termination, which will cause the chat to terminate when the text "APPROVE" is received.
    termination = TextMentionTermination("APPROVE")

    # The group chat will alternate between the writer and the critic.
    group_chat = RoundRobinGroupChat([writer, critic], termination_condition=termination)

    return group_chat


async def main() -> None:
    # Create team.
    group_chat = create_team()

    # `run_stream` returns an async generator to stream the intermediate messages.
    stream = group_chat.run_stream(task="Write a short story about a robot that discovers it has feelings.")
    # `Console` is a simple UI to display the stream.
    await Console(stream)

    # Save the state of the group chat and all participants.
    state = await group_chat.save_state()
    with open("group_chat_state.json", "w") as f:
        json.dump(state, f)

    # Create a new team with the same participants configuration.
    group_chat = create_team()

    # Load the state of the group chat and all participants.
    with open("group_chat_state.json", "r") as f:
        state = json.load(f)
    await group_chat.load_state(state)

    # Resume the chat.
    stream = group_chat.run_stream(task="Translate the story into Chinese.")
    await Console(stream)

asyncio.run(main())

保存和加载群聊状态#

在 v0.2 中，您需要显式地保存群聊消息，并在想要恢复聊天时重新加载它们。

在 v0.4 中，您可以简单地调用群聊对象上的 save_state 和 load_state 方法。 请参阅 Group Chat with Resume 以获取示例。

使用工具进行群聊#

在v0.2版本的群聊中，当涉及到工具时，您需要在用户代理上注册工具函数，并将用户代理包含在群聊中。其他代理发起的工具调用将被路由到用户代理来执行。

我们观察到这种方法存在许多问题，例如工具调用路由未能按预期工作，且工具调用请求和结果无法被不支持函数调用的模型接受。

在v0.4版本中，无需在用户代理上注册工具函数，因为工具直接在AssistantAgent中执行，该代理将工具的响应发布到群聊中。因此，群聊管理器不需要参与工具调用的路由。

参见Selector Group Chat 教程，了解在群聊中使用工具的示例。

使用自定义选择器进行群聊（Stateflow）#

在v0.2群聊中，当speaker_selection_method设置为自定义函数时，它可以覆盖默认的选择方法。这对于实现基于状态的选择方法非常有用。更多详情，请参见v0.2中的自定义发言者选择。

在v0.4中，你可以使用SelectorGroupChat与selector_func来实现相同的行为。selector_func是一个函数，它接收群聊中的当前消息线程并返回下一个发言者的名字。如果返回None，则将使用基于LLM的选择方法。

以下是使用基于状态的选择方法实现网络搜索/分析场景的示例。

import asyncio
from typing import Sequence
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.messages import AgentEvent, ChatMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Note: This example uses mock tools instead of real APIs for demonstration purposes
def search_web_tool(query: str) -> str:
    if "2006-2007" in query:
        return """Here are the total points scored by Miami Heat players in the 2006-2007 season:
        Udonis Haslem: 844 points
        Dwayne Wade: 1397 points
        James Posey: 550 points
        ...
        """
    elif "2007-2008" in query:
        return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2007-2008 is 214."
    elif "2008-2009" in query:
        return "The number of total rebounds for Dwayne Wade in the Miami Heat season 2008-2009 is 398."
    return "No data found."


def percentage_change_tool(start: float, end: float) -> float:
    return ((end - start) / start) * 100

def create_team() -> SelectorGroupChat:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    planning_agent = AssistantAgent(
        "PlanningAgent",
        description="An agent for planning tasks, this agent should be the first to engage when given a new task.",
        model_client=model_client,
        system_message="""
        You are a planning agent.
        Your job is to break down complex tasks into smaller, manageable subtasks.
        Your team members are:
            Web search agent: Searches for information
            Data analyst: Performs calculations

        You only plan and delegate tasks - you do not execute them yourself.

        When assigning tasks, use this format:
        1. <agent> : <task>

        After all tasks are complete, summarize the findings and end with "TERMINATE".
        """,
    )

    web_search_agent = AssistantAgent(
        "WebSearchAgent",
        description="A web search agent.",
        tools=[search_web_tool],
        model_client=model_client,
        system_message="""
        You are a web search agent.
        Your only tool is search_tool - use it to find information.
        You make only one search call at a time.
        Once you have the results, you never do calculations based on them.
        """,
    )

    data_analyst_agent = AssistantAgent(
        "DataAnalystAgent",
        description="A data analyst agent. Useful for performing calculations.",
        model_client=model_client,
        tools=[percentage_change_tool],
        system_message="""
        You are a data analyst.
        Given the tasks you have been assigned, you should analyze the data and provide results using the tools provided.
        """,
    )

    # The termination condition is a combination of text mention termination and max message termination.
    text_mention_termination = TextMentionTermination("TERMINATE")
    max_messages_termination = MaxMessageTermination(max_messages=25)
    termination = text_mention_termination | max_messages_termination

    # The selector function is a function that takes the current message thread of the group chat
    # and returns the next speaker's name. If None is returned, the LLM-based selection method will be used.
    def selector_func(messages: Sequence[AgentEvent | ChatMessage]) -> str | None:
        if messages[-1].source != planning_agent.name:
            return planning_agent.name # Always return to the planning agent after the other agents have spoken.
        return None

    team = SelectorGroupChat(
        [planning_agent, web_search_agent, data_analyst_agent],
        model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"), # Use a smaller model for the selector.
        termination_condition=termination,
        selector_func=selector_func,
    )
    return team

async def main() -> None:
    team = create_team()
    task = "Who was the Miami Heat player with the highest points in the 2006-2007 season, and what was the percentage change in his total rebounds between the 2007-2008 and 2008-2009 seasons?"
    await Console(team.run_stream(task=task))

asyncio.run(main())

嵌套聊天#

嵌套聊天允许你将整个团队或另一个代理嵌套在一个代理中。这对于创建代理的层次结构或“信息孤岛”非常有用，因为嵌套的代理无法直接与同一组之外的其他代理通信。

在v0.2版本中，通过在ConversableAgent类上使用register_nested_chats方法来支持嵌套聊天。 你需要使用字典来指定代理的嵌套序列， 详情请参见v0.2中的嵌套聊天。

在 v0.4 中，嵌套聊天是自定义代理的一个实现细节。 你可以创建一个自定义代理，它接受一个团队或另一个代理作为参数， 并实现 on_messages 方法来触发嵌套团队或代理。 应用程序需要决定如何传递或转换来自和去向嵌套团队或代理的消息。

以下示例展示了一个简单的嵌套聊天，用于计数数字。

import asyncio
from typing import Sequence
from autogen_core import CancellationToken
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.messages import TextMessage, ChatMessage
from autogen_agentchat.base import Response

class CountingAgent(BaseChatAgent):
    """An agent that returns a new number by adding 1 to the last number in the input messages."""
    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        if len(messages) == 0:
            last_number = 0 # Start from 0 if no messages are given.
        else:
            assert isinstance(messages[-1], TextMessage)
            last_number = int(messages[-1].content) # Otherwise, start from the last number.
        return Response(chat_message=TextMessage(content=str(last_number + 1), source=self.name))

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

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

class NestedCountingAgent(BaseChatAgent):
    """An agent that increments the last number in the input messages
    multiple times using a nested counting team."""
    def __init__(self, name: str, counting_team: RoundRobinGroupChat) -> None:
        super().__init__(name, description="An agent that counts numbers.")
        self._counting_team = counting_team

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # Run the inner team with the given messages and returns the last message produced by the team.
        result = await self._counting_team.run(task=messages, cancellation_token=cancellation_token)
        # To stream the inner messages, implement `on_messages_stream` and use that to implement `on_messages`.
        assert isinstance(result.messages[-1], TextMessage)
        return Response(chat_message=result.messages[-1], inner_messages=result.messages[len(messages):-1])

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        # Reset the inner team.
        await self._counting_team.reset()

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

async def main() -> None:
    # Create a team of two counting agents as the inner team.
    counting_agent_1 = CountingAgent("counting_agent_1", description="An agent that counts numbers.")
    counting_agent_2 = CountingAgent("counting_agent_2", description="An agent that counts numbers.")
    counting_team = RoundRobinGroupChat([counting_agent_1, counting_agent_2], max_turns=5)
    # Create a nested counting agent that takes the inner team as a parameter.
    nested_counting_agent = NestedCountingAgent("nested_counting_agent", counting_team)
    # Run the nested counting agent with a message starting from 1.
    response = await nested_counting_agent.on_messages([TextMessage(content="1", source="user")], CancellationToken())
    assert response.inner_messages is not None
    for message in response.inner_messages:
        print(message)
    print(response.chat_message)

asyncio.run(main())

你应该会看到以下输出：

source='counting_agent_1' models_usage=None content='2' type='TextMessage'
source='counting_agent_2' models_usage=None content='3' type='TextMessage'
source='counting_agent_1' models_usage=None content='4' type='TextMessage'
source='counting_agent_2' models_usage=None content='5' type='TextMessage'
source='counting_agent_1' models_usage=None content='6' type='TextMessage'

你可以查看SocietyOfMindAgent了解更多复杂的实现。

顺序聊天#

在 v0.2 中，通过使用 initiate_chats 函数支持顺序聊天。 它接收一个字典配置列表作为序列的每一步输入。 更多详情请参阅 v0.2 中的顺序聊天。

根据社区的反馈，initiate_chats函数过于主观，不够灵活，无法支持用户想要实现的多样化场景。我们经常发现用户在使用initiate_chats函数时遇到困难，尽管他们可以使用基本的Python代码轻松地将步骤拼凑在一起。因此，在v0.4中，我们不再在AgentChat API中提供用于顺序聊天的内置函数。

相反，你可以使用Core API创建一个事件驱动的顺序工作流，并使用AgentChat API提供的其他组件来实现工作流的每一步。查看Core API教程中的顺序工作流示例。

我们认识到工作流的概念是许多应用的核心，未来我们将为工作流提供更多内置支持。

GPTAssistantAgent#

在v0.2中，GPTAssistantAgent是一个特殊的代理类，由OpenAI Assistant API支持。

在v0.4中，等效的是OpenAIAssistantAgent类。它支持与v0.2中的GPTAssistantAgent相同的功能，并且增加了更多功能，如可自定义的线程和文件上传。更多详情请参见OpenAIAssistantAgent。

长上下文处理#

在 v0.2 中，超出模型上下文窗口的长上下文可以通过使用在 ConversableAgent 构建后添加的 transforms 功能来处理。

来自我们社区的反馈使我们相信这个功能是必不可少的，并且应该是AssistantAgent的内置组件，并且可以用于每个自定义代理。

在 v0.4 中，我们引入了 ChatCompletionContext 基类，该类管理消息历史记录并提供历史的虚拟视图。应用程序可以使用内置实现，例如 BufferedChatCompletionContext 来限制发送到模型的消息历史记录，或者提供自己的实现以创建不同的虚拟视图。

在聊天机器人场景中使用 BufferedChatCompletionContext 于 AssistantAgent 中。

import asyncio
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_core.model_context import BufferedChatCompletionContext
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)

    assistant = AssistantAgent(
        name="assistant",
        system_message="You are a helpful assistant.",
        model_client=model_client,
        model_context=BufferedChatCompletionContext(buffer_size=10), # Model can only view the last 10 messages.
    )
    while True:
        user_input = input("User: ")
        if user_input == "exit":
            break
        response = await assistant.on_messages([TextMessage(content=user_input, source="user")], CancellationToken())
        print("Assistant:", response.chat_message.content)

asyncio.run(main())

在这个例子中，聊天机器人只能读取历史记录中的最后10条消息。

可观察性与控制#

在v0.4版本的AgentChat中，你可以使用on_messages_stream方法来观察代理，该方法返回一个异步生成器，用于流式传输代理的内部思考和操作。对于团队，你可以使用run_stream方法来流式传输团队中代理之间的内部对话。你的应用程序可以使用这些流来实时观察代理和团队。

无论是 on_messages_stream 还是 run_stream 方法，都接受一个 CancellationToken 作为参数， 该参数可用于异步取消输出流并停止代理或团队。 对于团队，您还可以使用终止条件在满足特定条件时停止团队。 有关更多详细信息，请参阅 Termination Condition Tutorial。

与带有特殊日志模块的v0.2不同，v0.4 API 仅使用 Python 的logging模块来记录诸如模型客户端调用之类的事件。有关更多详细信息，请参阅核心API文档中的日志记录。

代码执行器#

v0.2 和 v0.4 中的代码执行器几乎相同，除了 v0.4 的执行器支持异步 API。如果代码执行时间过长，您还可以使用 CancellationToken 来取消代码执行。请参阅 Core API 文档中的 命令行代码执行器教程。

我们还添加了AzureContainerCodeExecutor，它可以使用Azure Container Apps (ACA)的动态会话来执行代码。请参阅ACA Dynamic Sessions Code Executor Docs。