v0.2 至 v0.4 版本迁移指南#
本文档为使用 autogen-agentchat 的 v0.2.* 版本用户提供迁移至 v0.4 版本的指导。v0.4 版本引入了全新的 API 和功能特性，包含重大变更，请仔细阅读本指南。我们仍在 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 设计：
核心 API 是基础层，提供可扩展的事件驱动执行框架，用于创建智能工作流；
AgentChat API 构建在核心层之上，提供面向任务的高层框架，用于构建交互式智能应用。该 API 是 AutoGen
v0.2的替代方案。
本指南主要聚焦 v0.4 的 AgentChat API，但您也可以仅使用核心 API 来构建自定义的高层框架。
新用户入门？#
直接跳转至 AgentChat 教程 开始使用 v0.4 版本。
本指南内容#
我们提供了一份详细指南，介绍如何将现有代码库从 v0.2 迁移至 v0.4 版本。
请参阅以下各功能模块获取具体的迁移信息：
当前 v0.2 版本中的以下功能
将在未来的 v0.4.* 版本中提供：
模型客户端成本计算 #4835
可教学Agent
RAG Agent
当这些缺失功能可用时，我们将更新本指南。
模型客户端#
在 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)
直接使用模型客户端类#
OpenAI：
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",
"structured_output": True,
},
)
注意：我们并未测试所有兼容OpenAI的API，其中许多API的实际行为可能与OpenAI官方API存在差异，即使它们声称兼容。使用前请务必进行测试。
了解更多关于模型客户端的内容，可参阅AgentChat教程，或查看核心API文档获取更详细信息。
未来将增加对其他托管模型的支持。
模型客户端缓存#
在 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:
# 初始化原始客户端
openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")
# 然后初始化 CacheStore，本例使用 diskcache.Cache
# 您也可以使用 redis 例如：
# 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) # 应打印来自 OpenAI 的响应
response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
print(response) # 应打印缓存的响应
await openai_model_client.close()
asyncio.run(main())
Assistant Agent#
在 v0.2 版本中，你可以通过以下方式创建助手Agent：
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 方法是异步的，
后者会返回一个异步生成器来流式传输Agent的内部思考过程。
以下是 v0.4 中直接调用助手Agent的方式（接续上面的示例）：
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)
await model_client.close()
asyncio.run(main())
CancellationToken 可用于异步取消请求，
当你调用 cancellation_token.cancel() 时，
会导致 on_messages 调用的 await 抛出 CancelledError 异常。
了解更多请参阅 Agent 教程
和 AssistantAgent。
多模态Agent#
v0.4版本中的AssistantAgent支持多模态输入（前提是模型客户端支持此功能）。模型客户端的vision能力决定了Agent是否支持多模态输入。
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)
await model_client.close()
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版本中，用户代理简化为仅接收用户输入的Agent，无需其他特殊配置。创建方式如下：
from autogen_agentchat.agents import UserProxyAgent
user_proxy = UserProxyAgent("user_proxy")
更多详情及如何自定义带超时的输入函数，请参阅UserProxyAgent。
RAG Agent#
在 v0.2 版本中，存在可教学代理（teachable agents）的概念，以及可以接收数据库配置的 RAG agents。
teachable_agent = ConversableAgent(
name="teachable_agent",
llm_config=llm_config
)
# 实例化 Teachability 对象，所有参数均为可选
teachability = Teachability(
reset_db=False,
path_to_db_dir="./tmp/interactive/teachability_db"
)
teachability.add_to_agent(teachable_agent)
在 v0.4 版本中，您可以使用 Memory 类实现 RAG agent。具体而言，您可以定义一个内存存储类，并将其作为参数传递给助手代理。更多细节请参阅 Memory 教程。
这种关注点分离的设计允许您实现使用任意数据库或存储系统的内存存储（需继承自 Memory 类），并与助手代理配合使用。以下示例展示了如何将 ChromaDB 向量内存存储与助手代理结合使用。此外，您的应用逻辑应决定何时以及如何向内存存储添加内容。例如，您可以选择为助手代理的每个响应调用 memory.add，或使用单独的 LLM 调用来判断是否应将内容添加到内存存储。
# ...
# ChromaDBVectorMemory 类示例
chroma_user_memory = ChromaDBVectorMemory(
config=PersistentChromaDBVectorMemoryConfig(
collection_name="preferences",
persistence_path=os.path.join(str(Path.home()), ".chromadb_autogen"),
k=2, # 返回前 k 个结果
score_threshold=0.4, # 最小相似度分数
)
)
# 可添加逻辑，例如将内容添加到内存存储的文档索引器
assistant_agent = AssistantAgent(
name="assistant_agent",
model_client=OpenAIChatCompletionClient(
model="gpt-4o",
),
tools=[get_weather],
memory=[chroma_user_memory],
)
可对话Agent与注册回复函数#
在 v0.2 版本中，您可以按以下方式创建可对话agent并注册回复函数：
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="您是一个乐于助人的助手。",
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]]:
# 在此处实现自定义回复逻辑
return True, "自定义回复"
# 注册回复函数
conversable_agent.register_reply([ConversableAgent], reply_func, position=0)
注意：异步回复函数仅在异步发送时被调用#
在 v0.4 版本中，我们无需猜测 reply_func 的功能、参数含义以及 position 的用途，只需创建一个自定义 agent 并实现 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, BaseChatMessage
from autogen_agentchat.base import Response
class CustomAgent(BaseChatAgent):
async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
return Response(chat_message=TextMessage(content="自定义回复", source=self.name))
async def on_reset(self, cancellation_token: CancellationToken) -> None:
pass
@property
def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:
return (TextMessage,)
创建自定义 agent 后，您可以像使用 AssistantAgent 一样使用它。更多细节请参阅自定义 Agent 教程。
保存与加载Agent状态#
在v0.2版本中没有内置的方法来保存和加载agent的状态：你需要通过导出ConversableAgent的chat_messages属性，并通过chat_messages参数将其导入回来，自行实现这一功能。
在v0.4版本中，你可以调用agent的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)
# 保存状态
state = await assistant.save_state()
# （可选）将状态写入磁盘
with open("assistant_state.json", "w") as f:
json.dump(state, f)
# （可选）从磁盘加载回来
with open("assistant_state.json", "r") as f:
state = json.load(f)
print(state) # 检查状态，其中包含聊天历史记录
# 继续聊天
response = await assistant.on_messages([TextMessage(content="Tell me a joke.", source="user")], cancellation_token)
print(response)
# 加载状态，使agent恢复到上一条消息之前的状态
await assistant.load_state(state)
# 再次继续相同的聊天
response = await assistant.on_messages([TextMessage(content="Tell me a joke.", source="user")], cancellation_token)
# 关闭与模型客户端的连接
await model_client.close()
asyncio.run(main())
你也可以在任何团队上调用save_state和load_state，例如RoundRobinGroupChat，以保存和加载整个团队的状态。
双Agent对话#
在 v0.2 版本中，你可以通过以下方式创建用于代码执行的双Agent对话：
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="你是一个乐于助人的助手。请用python编写所有代码。当任务完成时仅回复'TERMINATE'。",
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="编写一个打印'Hello, world!'的python脚本")
中间消息直接打印到控制台#
print(chat_result)
要在 `v0.4` 版本中获得相同行为，可以结合使用 {py:class}`~autogen_agentchat.agents.AssistantAgent`
和 {py:class}`~autogen_agentchat.agents.CodeExecutorAgent`，通过 {py:class}`~autogen_agentchat.teams.RoundRobinGroupChat` 实现。
```python
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="你是一个乐于助人的助手。请用python编写所有代码。当任务完成时只需回复'TERMINATE'。",
model_client=model_client,
)
code_executor = CodeExecutorAgent(
name="code_executor",
code_executor=LocalCommandLineCodeExecutor(work_dir="coding"),
)
# 终止条件是文本终止和最大消息终止的组合，满足任一条件都会终止对话
termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(10)
# 群聊将在assistant和code_executor之间交替进行
group_chat = RoundRobinGroupChat([assistant, code_executor], termination_condition=termination)
# `run_stream`返回一个异步生成器用于流式传输中间消息
stream = group_chat.run_stream(task="编写一个打印'Hello, world!'的python脚本")
# `Console`是用于显示消息流的简单UI
await Console(stream)
# 关闭与模型客户端的连接
await model_client.close()
asyncio.run(main())
工具使用#
在 v0.2 版本中，要创建一个使用工具的聊天机器人，您需要配置两个 agents：一个负责调用工具，另一个负责执行工具。对于每个用户请求，都需要启动这两个 agents 之间的对话。
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="你是一个乐于助人的助手。你可以调用工具来帮助用户。",
llm_config=llm_config,
max_consecutive_auto_reply=1, # 设置为1，这样在构建聊天机器人时，每次助手回复后都会返回到应用程序。
)
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"{city}的天气是72华氏度，晴朗。"
将工具函数注册到工具调用者和执行器#
register_function(get_weather, caller=tool_caller, executor=tool_executor)
while True: user_input = input("用户输入: ") if user_input == "exit": break chat_result = tool_executor.initiate_chat( tool_caller, message=user_input, summary_method="reflection_with_llm", # 若要让模型对工具使用进行反思，设置为"reflection_with_llm"；若想直接返回工具调用结果，则设为"last_msg" ) print("助手回复:", chat_result.summary)
在 `v0.4` 版本中，你实际上只需要一个 agent —— {py:class}`~autogen_agentchat.agents.AssistantAgent` —— 就能同时处理工具调用和工具执行。
```python
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: # 异步工具同样适用
return f"{city}的天气是72华氏度且晴朗"
async def main() -> None:
model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)
assistant = AssistantAgent(
name="assistant",
system_message="你是一个乐于助人的助手。你可以调用工具来帮助用户。",
model_client=model_client,
tools=[get_weather],
reflect_on_tool_use=True, # 设为True让模型对工具使用进行反思，设为False则直接返回工具调用结果
)
while True:
user_input = input("用户输入: ")
if user_input == "exit":
break
response = await assistant.on_messages([TextMessage(content=user_input, source="user")], CancellationToken())
print("助手回复:", response.chat_message.to_text())
await model_client.close()
asyncio.run(main())
当在群聊（如 RoundRobinGroupChat）中使用配备工具的agents时，你只需按照上述相同方式为agents添加工具，然后用这些agents创建群聊即可。
聊天结果#
在 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) # 获取经过LLM反思的聊天摘要
print(chat_result.chat_history) # 获取聊天历史记录
print(chat_result.cost) # 获取聊天成本
print(chat_result.human_input) # 获取聊天过程中征询的人工输入
更多详情请参阅 ChatResult 文档。
在 v0.4 版本中，通过 run 或 run_stream 方法会返回一个 TaskResult 对象。该 TaskResult 对象包含 messages 字段，记录了聊天过程中的所有消息历史，包括Agent的私有消息（如工具调用等）和公开消息。
TaskResult 与 ChatResult 之间存在一些显著差异：
TaskResult中的messages列表使用不同于ChatResult.chat_history的消息格式不提供
summary字段。应用程序需要自行决定如何基于messages列表生成聊天摘要TaskResult对象不提供human_input字段，因为用户输入可以通过筛选messages列表中的source字段来提取TaskResult对象不提供cost字段，但您可以根据令牌使用量计算成本。添加成本计算功能将是一个很好的社区扩展，详见社区扩展
v0.2 与 v0.4 消息格式转换#
您可以使用以下转换函数在 v0.4 消息（位于 autogen_agentchat.base.TaskResult.messages）和 v0.2 消息（位于 ChatResult.chat_history）之间进行转换。
from typing import Any, Dict, List, Literal
from autogen_agentchat.messages import (
BaseAgentEvent,
BaseChatMessage,
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: BaseAgentEvent | BaseChatMessage,
role: Literal["assistant", "user", "tool"],
image_detail: Literal["auto", "high", "low"] = "auto",
) -> Dict[str, Any]:
"""将 v0.4 AgentChat 消息转换为 v0.2 消息格式。
Args:
message (BaseAgentEvent | BaseChatMessage): 待转换的消息
role (Literal["assistant", "user", "tool"]): 消息角色
image_detail (Literal["auto", "high", "low"], optional): 多模态消息中图像内容的细节级别。默认为 "auto"。
Returns:
Dict[str, Any]: 转换后的 AutoGen v0.2 格式消息
"""
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"无效的多模态消息内容: {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"无效的消息类型: {type(message)}")
return v02_message
def convert_to_v04_message(message: Dict[str, Any]) -> BaseAgentEvent | BaseChatMessage:
"""将 v0.2 消息转换为 v0.4 AgentChat 消息格式"""
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"无法转换的消息: {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,
)
# 创建包含作家和评论家的群组聊天
groupchat = GroupChat(agents=[writer, critic], messages=[], max_round=12)
# 创建群组聊天管理器来管理对话，使用轮询选择发言者的方式
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config, speaker_selection_method="round_robin")
通过编辑器发起对话，中间消息直接打印到控制台#
result = editor.initiate_chat( manager, message="写一个关于机器人发现自己有情感的短篇故事", ) print(result.summary)
在 `v0.4` 版本中，您可以使用 {py:class}`~autogen_agentchat.teams.RoundRobinGroupChat` 实现相同行为。
```python
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="作家",
system_message="你是一名作家",
model_client=model_client,
)
critic = AssistantAgent(
name="critic",
description="评论家",
system_message="你是一名评论家，对作品提供反馈。如果任务完成只需回复'APPROVE'",
model_client=model_client,
)
# 终止条件是文本终止，当收到"APPROVE"文本时聊天将终止
termination = TextMentionTermination("APPROVE")
# 群聊将在作家和评论家之间交替进行
group_chat = RoundRobinGroupChat([writer, critic], termination_condition=termination, max_turns=12)
# `run_stream`返回异步生成器以流式传输中间消息
stream = group_chat.run_stream(task="写一个关于机器人发现自己有情感的短篇故事")
# `Console`是用于显示流的简单UI
await Console(stream)
# 关闭与模型客户端的连接
await model_client.close()
asyncio.run(main())
对于基于LLM的发言者选择，您可以使用SelectorGroupChat替代。
详见选择器群聊教程
和SelectorGroupChat获取更多细节。
注意：在
v0.4版本中，您无需在用户代理上注册函数即可在群聊中使用工具。 如工具使用部分所示，您只需将工具函数传递给AssistantAgent。 代理将在需要时自动调用工具。 如果您的工具没有输出格式良好的响应，可以使用reflect_on_tool_use参数让模型对工具使用进行反思。
支持恢复的群组对话#
在 v0.2 版本中，恢复群组对话的操作较为复杂。您需要显式地保存群组对话消息，并在需要恢复对话时重新加载这些消息。详情请参阅 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(model_client : OpenAIChatCompletionClient) -> RoundRobinGroupChat:
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,
)
# 终止条件是一个文本终止条件，当收到"APPROVE"文本时将终止对话
termination = TextMentionTermination("APPROVE")
# 群组对话将在writer和critic之间交替进行
group_chat = RoundRobinGroupChat([writer, critic], termination_condition=termination)
return group_chat
async def main() -> None:
model_client = OpenAIChatCompletionClient(model="gpt-4o", seed=42, temperature=0)
# 创建团队
group_chat = create_team(model_client)
# `run_stream`返回一个异步生成器，用于流式传输中间消息
stream = group_chat.run_stream(task="Write a short story about a robot that discovers it has feelings.")
# `Console`是一个简单的UI，用于显示消息流
await Console(stream)
# 保存群组对话和所有参与者的状态
state = await group_chat.save_state()
with open("group_chat_state.json", "w") as f:
json.dump(state, f)
# 使用相同的参与者配置创建新团队
group_chat = create_team(model_client)
# 加载群组对话和所有参与者的状态
with open("group_chat_state.json", "r") as f:
state = json.load(f)
await group_chat.load_state(state)
# 恢复对话
stream = group_chat.run_stream(task="Translate the story into Chinese.")
await Console(stream)
# 关闭与模型客户端的连接
await model_client.close()
asyncio.run(main())
保存与加载群聊状态#
在 v0.2 版本中，您需要显式保存群聊消息，并在需要恢复聊天时重新加载它们。
在 v0.4 版本中，您可以直接调用群聊对象上的 save_state 和 load_state 方法。
具体示例请参阅可恢复的群聊。
使用工具的群聊#
在 v0.2 版本的群聊中，当涉及工具使用时，您需要在用户代理上注册工具函数，
并将该用户代理包含在群聊中。其他 agents 发起的工具调用
将被路由到用户代理执行。
我们已发现这种方法存在诸多问题，例如工具调用路由 未能按预期工作，以及不支持函数调用的模型无法接收 工具调用请求和结果。
在 v0.4 版本中，无需在用户代理上注册工具函数，
因为工具直接在 AssistantAgent 内部执行，
并将工具响应发布到群聊中。
因此群聊管理器无需参与工具调用路由。
关于在群聊中使用工具的示例， 请参阅选择器群聊教程。
带自定义选择器的群聊（状态流）#
在 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 BaseAgentEvent, BaseChatMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
注意：此示例使用模拟工具而非真实API进行演示#
def search_web_tool(query: str) -> str: if "2006-2007" in query: return """以下是迈阿密热火队球员在2006-2007赛季的总得分： 乌杜尼斯·哈斯勒姆：844分 德怀恩·韦德：1397分 詹姆斯·波西：550分 ... """ elif "2007-2008" in query: return "德怀恩·韦德在迈阿密热火队2007-2008赛季的总篮板数为214个。" elif "2008-2009" in query: return "德怀恩·韦德在迈阿密热火队2008-2009赛季的总篮板数为398个。" return "未找到数据。"
def percentage_change_tool(start: float, end: float) -> float: return ((end - start) / start) * 100
def create_team(model_client : OpenAIChatCompletionClient) -> SelectorGroupChat: planning_agent = AssistantAgent( "PlanningAgent", description="任务规划agent，当有新任务时应首先参与。", model_client=model_client, system_message=""" 你是一个规划agent。 你的工作是将复杂任务分解为更小、可管理的子任务。 你的团队成员包括： 网络搜索agent：负责信息检索 数据分析师：执行计算任务
你只负责规划和分配任务 - 不亲自执行任务。
分配任务时使用以下格式：
1. <agent> : <任务>
所有任务完成后，总结发现并以"TERMINATE"结束。
""",
)
web_search_agent = AssistantAgent(
"WebSearchAgent",
description="网络搜索agent。",
tools=[search_web_tool],
model_client=model_client,
system_message="""
你是一个网络搜索agent。
你唯一的工具是search_tool - 用它来查找信息。
你每次只进行一次搜索调用。
获得结果后，你从不基于结果进行计算。
""",
)
data_analyst_agent = AssistantAgent(
"DataAnalystAgent",
description="数据分析师agent，用于执行计算任务。",
model_client=model_client,
tools=[percentage_change_tool],
system_message="""
你是一名数据分析师。
根据分配给你的任务，你应该使用提供的工具分析数据并提供结果。
""",
)
# 终止条件是文本提及终止和最大消息数终止的组合
text_mention_termination = TextMentionTermination("TERMINATE")
max_messages_termination = MaxMessageTermination(max_messages=25)
termination = text_mention_termination | max_messages_termination
# 选择器函数接收当前群聊消息线程
# 并返回下一位发言者名称。如果返回None，将使用基于LLM的选择方法
def selector_func(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
if messages[-1].source != planning_agent.name:
return planning_agent.name # 其他agent发言后总是返回规划agent
return None
team = SelectorGroupChat(
[planning_agent, web_search_agent, data_analyst_agent],
model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"), # 为选择器使用较小模型
termination_condition=termination,
selector_func=selector_func,
)
return team
async def main() -> None: model_client = OpenAIChatCompletionClient(model="gpt-4o") team = create_team(model_client) task = "谁是迈阿密热火队在2006-2007赛季得分最高的球员？他在2007-2008赛季和2008-2009赛季之间的总篮板数百分比变化是多少？" await Console(team.run_stream(task=task))
asyncio.run(main())
## 嵌套对话
嵌套对话允许你将整个团队或另一个agent嵌套在一个agent内部。这种机制适用于创建agent的层级结构或"信息孤岛"，因为嵌套的agent无法直接与同组之外的其他agent通信。
在`v0.2`版本中，嵌套对话通过`ConversableAgent`类的`register_nested_chats`方法实现。需要使用字典指定嵌套的agent序列，详见[v0.2中的嵌套对话](https://microsoft.github.io/autogen/0.2/docs/tutorial/conversation-patterns#nested-chats)。
在`v0.4`版本中，嵌套对话是自定义agent的实现细节。你可以创建一个自定义agent，将团队或其他agent作为参数传入，并通过实现`on_messages`方法来触发嵌套团队或agent。应用程序需要自行决定如何传递或转换进出嵌套团队/agent的消息。
以下示例展示了一个简单的数字计数嵌套对话：
```python
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, BaseChatMessage
from autogen_agentchat.base import Response
class CountingAgent(BaseChatAgent):
"""一个通过将输入消息中的最后一个数字加1来返回新数字的agent"""
async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
if len(messages) == 0:
last_number = 0 # 如果没有消息则从0开始
else:
assert isinstance(messages[-1], TextMessage)
last_number = int(messages[-1].content) # 否则从最后一个数字开始
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[BaseChatMessage]]:
return (TextMessage,)
class NestedCountingAgent(BaseChatAgent):
"""使用嵌套计数团队多次递增输入消息中最后一个数字的agent"""
def __init__(self, name: str, counting_team: RoundRobinGroupChat) -> None:
super().__init__(name, description="一个计数数字的agent")
self._counting_team = counting_team
async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
# 使用给定消息运行内部团队，并返回团队生成的最后一条消息
result = await self._counting_team.run(task=messages, cancellation_token=cancellation_token)
# 要实现流式传输内部消息，请实现`on_messages_stream`并用其实现`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:
# 重置内部团队
await self._counting_team.reset()
@property
def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:
return (TextMessage,)
async def main() -> None:
# 创建一个包含两个计数agent的内部团队
counting_agent_1 = CountingAgent("counting_agent_1", description="一个计数数字的agent")
counting_agent_2 = CountingAgent("counting_agent_2", description="一个计数数字的agent")
counting_team = RoundRobinGroupChat([counting_agent_1, counting_agent_2], max_turns=5)
# 创建一个将内部团队作为参数的嵌套计数agent
nested_counting_agent = NestedCountingAgent("nested_counting_agent", counting_team)
# 使用从1开始的消息运行嵌套计数agent
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 函数设计过于主观，灵活性不足，难以支持用户想要实现的各种场景。我们经常发现用户在使用基础 Python 代码可以轻松串联步骤的情况下，却难以让 initiate_chats 函数正常工作。因此，在 v0.4 版本中，我们不再在 AgentChat API 中提供内置的顺序对话函数。
作为替代方案，您可以使用 Core API 创建事件驱动的顺序工作流，并利用 AgentChat API 提供的其他组件来实现工作流的每个步骤。顺序工作流的示例请参阅 Core API 教程。
我们认识到工作流概念是许多应用的核心，未来我们将为工作流提供更多内置支持。
GPTAssistantAgent#
在 v0.2 版本中，GPTAssistantAgent 是一个特殊的 agent 类，由 OpenAI Assistant API 提供支持。
在 v0.4 版本中，等效的类是 OpenAIAssistantAgent。它不仅支持 v0.2 中 GPTAssistantAgent 的所有功能，还新增了可自定义线程和文件上传等特性。更多详情请参阅 OpenAIAssistantAgent。
长上下文处理#
在 v0.2 版本中，对于超出模型上下文窗口的长上下文内容，可以通过使用 ConversableAgent 构造后添加的 transforms 功能来处理。
来自社区的反馈使我们确信这一功能至关重要，应作为 AssistantAgent 的内置组件，并可用于所有自定义 agent。
在 v0.4 版本中，我们引入了 ChatCompletionContext 基类来管理消息历史记录并提供历史的虚拟视图。应用程序可以使用内置实现（如 BufferedChatCompletionContext）来限制发送给模型的消息历史记录，或提供自己的实现以创建不同的虚拟视图。
以下是在聊天机器人场景中，在 AssistantAgent 中使用 BufferedChatCompletionContext 的示例：
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), # 模型只能查看最后10条消息
)
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.to_text())
await model_client.close()
asyncio.run(main())
在此示例中，聊天机器人只能读取历史记录中的最后10条消息。
可观测性与控制#
在 v0.4 版本的 AgentChat 中，您可以通过使用 on_messages_stream 方法来观察 agents 的行为，该方法会返回一个异步生成器，用于流式传输 agent 的内部思考和操作。对于团队场景，您可以使用 run_stream 方法来流式传输团队中 agents 之间的内部对话。您的应用程序可以利用这些流实时观察 agents 和团队的行为。
on_messages_stream 和 run_stream 方法都接受一个 CancellationToken 作为参数，该参数可用于异步取消输出流并停止 agent 或团队。对于团队场景，您还可以使用终止条件在满足特定条件时停止团队运行。更多详情请参阅终止条件教程。
与 v0.2 版本自带特殊日志模块不同，v0.4 API 直接使用 Python 的 logging 模块来记录诸如模型客户端调用等事件。更多详情请参阅核心 API 文档中的日志记录。
代码执行器#
v0.2 和 v0.4 版本的代码执行器几乎完全相同，只是 v0.4 的执行器支持异步 API。您也可以使用 CancellationToken 来取消耗时过长的代码执行。详情请参阅核心 API 文档中的命令行代码执行器教程。
我们还新增了 ACADynamicSessionsCodeExecutor，该执行器可以利用 Azure Container Apps (ACA) 动态会话进行代码执行。参见ACA 动态会话代码执行器文档。