AutoGen快速入门

引言

在大语言模型应用开发的浪潮中，已经诞生了多个流行的框架，它们分别解决了不同阶段的痛点：

LangChain
作为最早的 LLM 应用开发框架之一，LangChain 提供了丰富的组件和封装，帮助开发者快速构建对话、问答、检索增强（RAG）等应用。它的优势是生态广泛、组件齐全，适合快速原型开发。但随着项目规模增大，LangChain 的链式调用模型逐渐显得臃肿，不够灵活。

LangGraph
LangGraph 是在 LangChain 基础上发展起来的图结构工作流框架，它将复杂的应用逻辑抽象为“节点 + 边”的形式，更直观地表达多 Agent 协作、循环调用、条件分支等流程。它解决了 LangChain 在复杂逻辑上的局限性，但对初学者而言，学习曲线更陡峭，同时它本质上还是偏“编排层”。

AutoGen
与 LangChain 和 LangGraph 不同，AutoGen 的出发点是 “多智能体协作”。它把大模型视作一个个可交流的 Agent，这些 Agent 可以具备不同角色、目标和工具，并通过消息传递来协作完成任务。AutoGen 内置了 AgentChat 框架 和 扩展生态，让开发者能快速定义 Agent、绑定工具、调用不同模型，轻松实现 人机协作、多 Agent 协作、工具调用 等高级功能。

为什么要学习 AutoGen？

• 相比 LangChain：AutoGen 更加轻量和现代化，摆脱了冗余的链式调用，专注于 Agent 协作与实际应用。

• 相比 LangGraph：AutoGen 不仅能做流程编排，还提供了灵活的 AgentChat API，让你在几十行代码内就能跑起一个多 Agent 应用。

• AutoGen 的优势在于：

  • 原生支持 多 Agent 协作，无需复杂图结构即可实现灵活对话。
  • 生态丰富，内置对 OpenAI、Azure 及其他模型的支持，并能轻松扩展。
  • 更贴近真实应用场景，不仅能做实验，还能快速搭建生产级应用。

实际上，哪怕是 LangChain、LangGraph，甚至直接用 OpenAI 的官方 SDK，也都能实现 Agent 的构建。不同之处在于：

• 用 OpenAI SDK，开发者需要手写大部分逻辑，工作量最大，但灵活度最高；

• 用 LangChain / LangGraph，可以复用已有的链式调用或图结构编排，减少部分工作量；

• 而 AutoGen 提供了更进一步的封装，开发者甚至只需导入 autogen_chatagent，就能快速搭建起可用的多 Agent 系统。

如果说 LangChain 是“大模型应用开发的入门课”，LangGraph 是“复杂逻辑的强化班”，那么 AutoGen 就是“实战驱动的应用框架”，帮助开发者真正把 LLM 变成生产力。框架之间并没有绝对的好坏，更多是适用场景和开发体验的差别 —— 而 AutoGen 的高封装和即用性，正是它的最大魅力。

AutoGen 组件全景图

Studio （PyPi 包名：autogenstudio）
一个基于 Web 的 UI，用于无需写代码就能快速构建和原型化 Agent 应用，底层基于 AgentChat。

pip install -U autogenstudio
autogenstudio ui --port 8080 --appdir ./myapp

如果你刚接触 AutoGen，想要 零代码原型化 Agent 应用，从这里开始。

---

AgentChat （PyPi 包名：autogen-agentchat）
一个编程框架，用于构建 单 Agent 或多 Agent 的对话式应用。底层基于 Core，需要 Python 3.10+。

# 安装
pip install -U "autogen-agentchat" "autogen-ext[openai]"# 示例代码
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClientasync def main() -> None:agent = AssistantAgent("assistant", OpenAIChatCompletionClient(model="gpt-4o"))print(await agent.run(task="Say 'Hello World!'"))asyncio.run(main())

如果你想用 Python 来构建 Agent 原型，或者从 AutoGen 0.2 版本迁移，可以从这里开始。

---

Core （PyPi 包名：autogen-core）
一个事件驱动的编程框架，用于构建 可扩展的多 Agent AI 系统。适用场景包括：

• 面向业务流程的确定性和动态 Agent 工作流
• 多 Agent 协作研究
• 跨语言的分布式 Agent 应用

如果你准备 深入构建多 Agent 系统，从这里开始。

---

Extensions （PyPi 包名：autogen-ext）
实现了 Core 和 AgentChat 的扩展组件，用于与外部服务或其他库交互。你可以直接使用社区扩展，也可以自己编写。内置扩展示例：

• McpWorkbench：支持 Model-Context Protocol (MCP) 服务器
• OpenAIAssistantAgent：支持 OpenAI Assistant API
• DockerCommandLineCodeExecutor：在 Docker 容器中运行模型生成的代码
• GrpcWorkerAgentRuntime：实现分布式 Agent

总而言之：

组件	PyPI 包	定位	适用人群	如何开始
Studio	autogenstudio	基于 Web 的 UI，无需写代码即可快速原型化 Agent 应用，底层基于 AgentChat	零代码用户，想快速体验和原型化 Agent	bash<br>pip install -U autogenstudio<br>autogenstudio ui --port 8080 --appdir ./myapp<br>
AgentChat	autogen-agentchat	用 Python 编程构建 单/多 Agent 对话式应用，基于 Core	Python 开发者，想要构建对话型 Agent 原型	python<br>pip install -U "autogen-agentchat" "autogen-ext[openai]"<br><br>from autogen_agentchat.agents import AssistantAgent<br>from autogen_ext.models.openai import OpenAIChatCompletionClient<br>
Core	autogen-core	事件驱动框架，用于 大规模多 Agent 系统 的构建	研究人员 / 高级开发者，需要复杂工作流、多 Agent 协作或分布式系统	适用于业务流程自动化、多 Agent 协作研究、跨语言分布式系统
Extensions	autogen-ext	Core 和 AgentChat 的扩展库，用于对接外部服务	需要扩展功能的开发者，想连接更多服务/工具	内置扩展示例： • McpWorkbench（MCP 协议） • OpenAIAssistantAgent（Assistant API） • DockerCommandLineCodeExecutor（Docker 容器执行） • GrpcWorkerAgentRuntime（分布式 Agent）

在快速入门里面主要讲解 AgentChat，对于core而言，无论是langChain-core还是AutoGen-core其实内容都大差不差，封装Message，封装角色，封装存储…

agentchat

快速开始

通过 AgentChat，可以使用预设的 Agent 快速构建应用。我们先从创建一个可以使用工具的单个 Agent 开始。

首先，需要安装 AgentChat 和 Extension 包：

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

这个示例使用了 OpenAI 模型，但你也可以使用其他模型。只需将 model_client 替换为你想用的模型或模型客户端类即可。

如果想使用 Azure OpenAI 模型 并启用 AAD 认证，请按照官方说明操作；使用其他模型请参考 Models 文档。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient# 定义模型客户端。你也可以使用实现了 `ChatCompletionClient` 接口的其他客户端。
model_client = OpenAIChatCompletionClient(model="gpt-4o",# api_key="YOUR_API_KEY",
)# 定义一个简单的工具函数，供 Agent 使用。
# 本示例使用一个模拟天气工具用于演示。
async def get_weather(city: str) -> str:"""获取指定城市的天气。"""return f"{city} 的天气是 73 华氏度，晴朗。"# 定义一个 AssistantAgent，绑定模型、工具、系统消息，并启用反思功能。
# 系统消息用自然语言指导 Agent 行为。
agent = AssistantAgent(name="weather_agent",model_client=model_client,tools=[get_weather],system_message="你是一个乐于助人的助手。",reflect_on_tool_use=True,model_client_stream=True,  # 启用模型客户端的流式输出。
)# 运行 Agent，并将消息流输出到控制台。
async def main() -> None:await Console(agent.run_stream(task="纽约的天气如何？"))# 关闭模型客户端连接await model_client.close()# 注意：如果在 Python 脚本中运行，需要使用 asyncio.run(main())。
await main()

运行示例后，控制台输出示例：

---------- user ----------
纽约的天气如何？
---------- weather_agent ----------
[FunctionCall(id='call_bE5CYAwB7OlOdNAyPjwOkej1', arguments='{"city":"New York"}', name='get_weather')]
---------- weather_agent ----------
[FunctionExecutionResult(content='纽约的天气是 73 华氏度，晴朗。', call_id='call_bE5CYAwB7OlOdNAyPjwOkej1', is_error=False)]
---------- weather_agent ----------
当前纽约的天气是 73 华氏度，晴朗。

这里我们依然可以使用硅基流动的云模型：

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelInfomodel_info: ModelInfo = {"vision": False,                  # 该模型不支持图像输入"function_calling": True,         # 支持函数调用"json_output": True,              # 支持 JSON 输出"structured_output": True,        # 支持结构化输出"family": "qwen",                 # 模型属于 qwen 系列"multiple_system_messages": True, # 支持多系统消息
}# 定义模型客户端。你也可以使用实现了 `ChatCompletionClient` 接口的其他客户端。
model_client = OpenAIChatCompletionClient(model="Qwen/Qwen3-8B",api_key="your api key",base_url="https://api.siliconflow.cn",model_info=model_info
)# 定义一个简单的工具函数，供 Agent 使用。
# 本示例使用一个模拟天气工具用于演示。
async def get_weather(city: str) -> str:"""获取指定城市的天气。"""return f"{city} 的天气是 73 华氏度，晴朗。"# 定义一个 AssistantAgent，绑定模型、工具、系统消息，并启用反思功能。
# 系统消息用自然语言指导 Agent 行为。
agent = AssistantAgent(name="weather_agent",model_client=model_client,tools=[get_weather],system_message="你是一个乐于助人的助手。",reflect_on_tool_use=True,model_client_stream=True,  # 启用模型客户端的流式输出。
)# 运行 Agent，并将消息流输出到控制台。
async def main() -> None:await Console(agent.run_stream(task="纽约的天气如何？"))# 关闭模型客户端连接await model_client.close()# 注意：如果在 Python 脚本中运行，需要使用 asyncio.run(main())。
if __name__ == '__main__':asyncio.run(main())

输出如下：

---------- TextMessage (user) ----------
纽约的天气如何？
---------- ModelClientStreamingChunkEvent (weather_agent) ----------
<think>
好的，用户问的是纽约的天气如何。我需要调用get_weather这个函数来获取信息。首先，确认用户提供的城市名称是纽约，然后检查参数是否正确。函数需要的是一个city参数，类型是字符串，所以应该没问题。接下来，我需要生成一个工具调用的JSON对象，包含函数名和城市名称。确保没有额外的属性，并且格式正确。最后，把结果放在指定的XML标签里。这样用户就能得到他们需要的天气信息了。
</think>---------- ToolCallRequestEvent (weather_agent) ----------
[FunctionCall(id='0198d676f0e7da19f49edf5382f9acb1', arguments=' {"city": "纽约"}', name='get_weather')]
---------- ToolCallExecutionEvent (weather_agent) ----------
[FunctionExecutionResult(content='纽约 的天气是 73 华氏度，晴朗。', name='get_weather', call_id='0198d676f0e7da19f49edf5382f9acb1', is_error=False)]
---------- ModelClientStreamingChunkEvent (weather_agent) ----------
<think>
好的，用户之前询问了纽约的天气，我调用了get_weather工具，现在得到了响应：纽约的天气是73华氏度，晴朗。接下来我需要把这个信息用中文组织成自然的回复。首先，确认用户的需求。用户可能想知道当前的天气情况，或者计划出行，需要了解是否需要带伞或者穿什么衣服。由于回复是晴朗，可能用户关心的是是否适合户外活动。然后，检查回复内容是否完整。温度是73华氏度，转换成摄氏度大约是23度，可能用户更习惯使用摄氏度，但原数据是华氏度，所以保持原单位更准确。需要确认是否有必要转换，但根据工具返回的数据，保持华氏度即可。接下来，考虑是否需要补充其他信息，比如湿度、风速或降水概率，但工具只提供了温度和天气状况，所以只能基于这两点回答。可能用户没有提到其他需求，所以不需要额外猜测。然后，检查回复的语气是否友好。用户可能希望得到简洁明了的回答，同时带有建议。例如，晴朗的天气可以建议户外活动，但不需要太过冗长。保持自然，避免机械感。最后，确保回复格式正确，没有使用任何Markdown，只是纯文本。现在综合这些信息，给出一个合适的回答。
</think>纽约当前的天气是73华氏度，晴朗。这样的天气非常适合外出活动，建议可以享受户外时光，注意防晒哦！🌞
---------- TextMessage (weather_agent) ----------纽约当前的天气是73华氏度，晴朗。这样的天气非常适合外出活动，建议可以享受户外时光，注意防晒哦！🌞

基础功能

模型（Models）

在很多情况下，Agent 需要访问 LLM 模型服务，例如 OpenAI、Azure OpenAI 或本地模型。由于不同提供商的 API 各不相同，autogen-core 实现了一个用于模型客户端的协议，而 autogen-ext 则实现了一套针对常用模型服务的模型客户端。AgentChat 可以使用这些模型客户端与模型服务进行交互。

模型调用日志

AutoGen 使用标准 Python logging 模块来记录事件，例如模型调用和响应。

• 日志器名称：autogen_core.EVENT_LOGGER_NAME
• 事件类型：LLMCall

示例代码：

import logging
from autogen_core import EVENT_LOGGER_NAMElogging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(EVENT_LOGGER_NAME)
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.INFO)

OpenAI

要访问 OpenAI 模型，需要安装 openai 扩展，它允许你使用 OpenAIChatCompletionClient。

pip install "autogen-ext[openai]"

你还需要从 OpenAI 获取一个 API Key。

from autogen_ext.models.openai import OpenAIChatCompletionClientopenai_model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06",# api_key="sk-...", # 如果已设置 OPENAI_API_KEY 环境变量，可选
)

测试模型客户端的示例：

from autogen_core.models import UserMessageresult = await openai_model_client.create([UserMessage(content="法国的首都是哪里？", source="user")
])
print(result)
await openai_model_client.close()

输出示例：

CreateResult(finish_reason='stop', content='法国的首都是巴黎。', usage=RequestUsage(prompt_tokens=15, completion_tokens=7), cached=False, logprobs=None)

Azure OpenAI

同样地，需要安装 azure 和 openai 扩展 来使用 AzureOpenAIChatCompletionClient。

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

使用此客户端时，需要提供 deployment id、Azure Cognitive Services 端点、API 版本 和 模型能力。认证方式可使用 API Key 或 Azure AD (AAD) Token。

以下示例展示如何使用 AAD 认证（使用的身份需分配 Cognitive Services OpenAI User 角色）：

from autogen_core.models import UserMessage
from autogen_ext.auth.azure import AzureTokenProvider
from autogen_ext.models.openai import AzureOpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential# 创建 Token Provider
token_provider = AzureTokenProvider(DefaultAzureCredential(),"https://cognitiveservices.azure.com/.default",
)az_model_client = AzureOpenAIChatCompletionClient(azure_deployment="{your-azure-deployment}",model="{model-name, such as gpt-4o}",api_version="2024-06-01",azure_endpoint="https://{your-custom-endpoint}.openai.azure.com/",azure_ad_token_provider=token_provider,  # 如果使用 Key 认证，可选# api_key="sk-...", # Key 认证时使用
)result = await az_model_client.create([UserMessage(content="法国的首都是哪里？", source="user")
])
print(result)
await az_model_client.close()

Azure AI Foundry

Azure AI Foundry（之前称为 Azure AI Studio）提供托管在 Azure 上的模型。使用这些模型时，使用 AzureAIChatCompletionClient。

需要安装 azure 扩展：

pip install "autogen-ext[azure]"

下面是使用 GitHub Marketplace 上 Phi-4 模型 的示例：

import os
from autogen_core.models import UserMessage
from autogen_ext.models.azure import AzureAIChatCompletionClient
from azure.core.credentials import AzureKeyCredentialclient = AzureAIChatCompletionClient(model="Phi-4",endpoint="https://models.github.ai/inference",# 使用模型时需生成 GitHub Personal Access Token (PAT)credential=AzureKeyCredential(os.environ["GITHUB_TOKEN"]),model_info={"json_output": False,"function_calling": False,"vision": False,"family": "unknown","structured_output": False,},
)result = await client.create([UserMessage(content="法国的首都是哪里？", source="user")
])
print(result)
await client.close()

输出示例：

finish_reason='stop' content='法国的首都是巴黎。' usage=RequestUsage(prompt_tokens=14, completion_tokens=8) cached=False logprobs=None

更多客户端接入

注入接入ollama客户端等可以参考：https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/models.html#ollama-experimental

无非就是autogen-ext实现了autogen-core定义的接口，这里不再赘述。

消息（Messages）

在 AutoGen AgentChat 中，消息用于与其他 Agent、调度器以及应用程序进行通信和信息交换。AgentChat 支持多种消息类型，每种类型都有其特定用途。

从高层来看，AgentChat 中的消息可以分为两类：

1. Agent-与-Agent 消息（agent-agent messages）
2. Agent 的内部事件和消息（internal events and messages）

1. Agent-Agent 消息

AgentChat 支持多种 Agent 之间通信的消息类型，它们都继承自基类 BaseChatMessage。具体子类覆盖了基础文本和多模态通信，例如 TextMessage 和 MultiModalMessage。

• 创建文本消息示例：

from autogen_agentchat.messages import TextMessagetext_message = TextMessage(content="Hello, world!", source="User")

• 创建多模态消息示例（可包含文本和图像）：

from io import BytesIO
import requests
from autogen_agentchat.messages import MultiModalMessage
from autogen_core import Image as AGImage
from PIL import Imagepil_image = Image.open(BytesIO(requests.get("https://picsum.photos/300/200").content))
img = AGImage(pil_image)
multi_modal_message = MultiModalMessage(content=["Can you describe the content of this image?", img],source="User"
)

创建的 TextMessage 和 MultiModalMessage 可以通过 on_messages 方法直接传递给 Agent，或作为任务传递给team的 run() 方法。消息也会用于 Agent 的响应中。

2. 内部事件（Internal Events）

AgentChat 还支持 事件（events）——即属于 Agent 内部的消息。这些消息用于在 Agent 内部传递事件和操作信息，继承自基类 BaseAgentEvent。

• 示例：

  • ToolCallRequestEvent：表示发起了调用工具的请求
  • ToolCallExecutionEvent：包含工具调用的结果

通常，事件由 Agent 自身创建，并包含在 on_messages 返回的 Response 的 inner_messages 字段中。如果你构建自定义 Agent 并希望向其他实体（例如 UI）传递事件，也可以将这些事件包含在 Response.inner_messages 中。

自定义消息类型

你可以通过继承基类 BaseChatMessage 或 BaseAgentEvent 来创建 自定义消息类型。这允许你定义自己的消息格式和行为，以适应你的应用场景。自定义消息类型在编写自定义 Agent 时特别有用。

Agents（代理）

AutoGen AgentChat 提供了一组预设的代理（Agents），每个代理在响应消息的方式上有所不同。所有代理都共享以下属性和方法：

• name：代理的唯一名称。
• description：代理的文本描述。
• run：运行代理的方法，接收一个任务（可以是字符串或消息列表），返回一个 TaskResult。代理应当是有状态的，该方法期望接收新的消息，而不是完整的历史记录。
• run_stream：与 run() 类似，但返回一个消息迭代器。迭代器中的消息需要继承自 BaseAgentEvent 或 BaseChatMessage，最后返回一个 TaskResult。

Assistant Agent（助手代理）

AssistantAgent 是一个内置代理，它基于语言模型，并具备调用工具的能力。

⚠️ 注意
AssistantAgent 是一个“万用型”（kitchen sink）代理，主要用于原型设计和教学目的——它非常通用。请务必阅读相关文档和实现细节，以理解其设计取舍。在你完全理解设计之后，可以考虑实现你自己的代理。参见 Custom Agent（自定义代理）。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import StructuredMessage
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient# 定义一个搜索网络信息的工具。
# 为了简化，这里用一个返回静态字符串的模拟函数。
async def web_search(query: str) -> str:"""在网络上查找信息"""return "AutoGen 是一个用于构建多智能体应用的编程框架。"# 创建一个使用 OpenAI GPT-4o 模型的代理。
model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano",# api_key="YOUR_API_KEY",
)agent = AssistantAgent(name="assistant",model_client=model_client,tools=[web_search],system_message="使用工具解决任务。",
)

获取结果（Getting Result）

我们可以使用 run() 方法让代理在给定任务上运行。

# 在脚本中运行时，请使用 asyncio.run(agent.run(...))
result = await agent.run(task="Find information on AutoGen")
print(result.messages)[TextMessage(source='user',models_usage=None,metadata={},content='Find information on AutoGen',type='TextMessage'),ToolCallRequestEvent(source='assistant',models_usage=RequestUsage(prompt_tokens=61, completion_tokens=16),metadata={},content=[FunctionCall(id='call_703i17OLXfztkuioUbkESnea',arguments='{"query":"AutoGen"}',name='web_search')],type='ToolCallRequestEvent'),ToolCallExecutionEvent(source='assistant',models_usage=None,metadata={},content=[FunctionExecutionResult(content='AutoGen is a programming framework for building multi-agent applications.',name='web_search',call_id='call_703i17OLXfztkuioUbkESnea',is_error=False)],type='ToolCallExecutionEvent'),ToolCallSummaryMessage(source='assistant',models_usage=None,metadata={},content='AutoGen is a programming framework for building multi-agent applications.',type='ToolCallSummaryMessage')
]

调用 run() 方法会返回一个 TaskResult，其中包含 messages 属性。messages 保存了代理的“思考过程”，以及最终的响应。

💡 注意

• 状态更新：run() 会更新代理的内部状态 —— 它会把消息追加到代理的消息历史中。你也可以在 不传入任务 的情况下调用 run()，此时代理会基于当前状态生成响应。
• 与 v0.2 的区别：与 v0.2 的 AgentChat 不同，工具调用是在同一个代理、同一次 run() 调用中直接执行的。默认情况下，代理会将工具调用的结果作为最终响应返回。

多模态输入（Multi-Modal Input）

AssistantAgent 可以处理多模态输入，只需将输入封装为 MultiModalMessage。

from io import BytesIO
import PIL
import requests
from autogen_agentchat.messages import MultiModalMessage
from autogen_core import Image# 创建一个带随机图片和文本的多模态消息。
pil_image = PIL.Image.open(BytesIO(requests.get("https://picsum.photos/300/200").content))
img = Image(pil_image)
multi_modal_message = MultiModalMessage(content=["Can you describe the content of this image?", img],source="user"
)# 在脚本中运行时，请使用 asyncio.run(...)
result = await agent.run(task=multi_modal_message)
print(result.messages[-1].content)  # type: ignore

示例输出：

The image depicts a scenic mountain landscape under a clear blue sky. 
There are several rugged mountain peaks in the background, with some clouds scattered across the sky. 
In the valley below, there is a body of water, possibly a lake or river, surrounded by greenery. 
The overall scene conveys a sense of natural beauty and tranquility.

该图像展示了 蓝天下的山地风景：远处是嶙峋的山峰，天空中点缀着几朵云彩；山谷间有一片水域（可能是湖泊或河流），周围环绕着绿色植被。整体画面传递出一种自然之美与宁静感。

流式消息（Streaming Messages）

我们也可以使用 run_stream() 方法以流式的方式获取代理生成的每条消息，并通过 Console 将消息实时打印到控制台。

async def assistant_run_stream() -> None:# 方式一：逐条读取流中的消息（与前面示例类似）。# async for message in agent.run_stream(task="Find information on AutoGen"):#     print(message)# 方式二：使用 Console 实时打印所有消息。await Console(agent.run_stream(task="Find information on AutoGen"),output_stats=True,  # 启用统计信息输出。)# 在脚本中运行时，请使用 asyncio.run(assistant_run_stream())
await assistant_run_stream()

示例输出：

---------- TextMessage (user) ----------
Find information on AutoGen---------- ToolCallRequestEvent (assistant) ----------
[FunctionCall(id='call_HOTRhOzXCBm0zSqZCFbHD7YP', arguments='{"query":"AutoGen"}', name='web_search')]
[Prompt tokens: 61, Completion tokens: 16]---------- ToolCallExecutionEvent (assistant) ----------
[FunctionExecutionResult(content='AutoGen is a programming framework for building multi-agent applications.', name='web_search', call_id='call_HOTRhOzXCBm0zSqZCFbHD7YP', is_error=False)]---------- ToolCallSummaryMessage (assistant) ----------
AutoGen is a programming framework for building multi-agent applications.---------- Summary ----------
Number of messages: 4
Finish reason: None
Total prompt tokens: 61
Total completion tokens: 16
Duration: 0.52 seconds

调用 run_stream() 方法会返回一个 异步生成器（asynchronous generator）：

• 它会依次产出代理生成的每条消息。
• 最后一项是一个 TaskResult。

从输出的消息可以看到：

• 助手代理调用了 web_search 工具来获取信息；
• 随后基于搜索结果生成了最终回答。

使用工具与工作台（Using Tools and Workbench）

大型语言模型（LLMs）通常仅限于生成文本或代码响应。然而，许多复杂任务会从调用外部工具（如从 API 或数据库获取数据）中受益。

为了解决这一限制，现代 LLM 现在可以接受一组可用的工具模式（tool schema，工具及其参数的描述），并生成工具调用消息（Tool Call Message）。这一能力被称为 工具调用（Tool Calling） 或 函数调用（Function Calling），并逐渐成为构建智能体应用的重要模式。

在 AgentChat 中：

• AssistantAgent 可以使用工具来执行特定操作。

  • 例如，web_search 工具允许助手代理在网络上搜索信息。

• 一个自定义工具可以是：

  • 一个 Python 函数，或
  • 一个继承自 BaseTool 的子类。

另一方面，Workbench（工作台） 是一个共享状态和资源的工具集合。

💡 注意

• 关于如何将 模型客户端（model clients） 与 工具（tools） 和 工作台（workbench） 直接结合使用，请参考 核心用户指南（Core User Guide） 中的 Tools 与 Workbench 部分。
• 默认情况下，当 AssistantAgent 执行工具时，会在响应中将工具的输出以字符串形式返回，并封装在 ToolCallSummaryMessage 中。
• 如果工具没有返回自然语言格式的字符串，可以在代理构造函数中设置 reflect_on_tool_use=True，添加一个 反思步骤（reflection step），由模型对工具的输出进行总结。

内置工具与工作台（Built-in Tools and Workbench）

AutoGen Extension 提供了一组可与 AssistantAgent 搭配使用的内置工具。
完整的工具列表可在 autogen_ext.tools 命名空间下的 API 文档中找到。

例如，你可以使用以下工具：

• graphrag：用于操作 GraphRAG 索引 的工具。
• http：用于发送 HTTP 请求 的工具。
• langchain：用于接入 LangChain 工具 的适配器。
• mcp：用于操作 Model Chat Protocol (MCP) 服务器 的工具与工作台。

函数工具（Function Tool）

AssistantAgent 会自动将一个 Python 函数 转换为 FunctionTool，代理即可将其作为工具使用，并会根据函数的 签名（signature） 和 文档字符串（docstring） 自动生成工具模式（schema）。

例如，下面的 web_search_func 工具就是一个函数工具，其模式（schema）由系统自动生成：

from autogen_core.tools import FunctionTool# 使用 Python 函数定义一个工具。
async def web_search_func(query: str) -> str:"""在网络上查找信息"""return "AutoGen 是一个用于构建多智能体应用的编程框架。"# 如果工具是 Python 函数，这一步会在 AssistantAgent 内部自动完成。
web_search_function_tool = FunctionTool(web_search_func, description="在网络上查找信息"
)# 在 AssistantAgent 的 on_messages 调用中，schema 会被提供给模型。
web_search_function_tool.schema

示例输出：

{"name": "web_search_func","description": "Find information on the web","parameters": {"type": "object","properties": {"query": {"description": "query","title": "Query","type": "string"}},"required": ["query"],"additionalProperties": false},"strict": false
}

模型上下文协议（MCP）工作台（Model Context Protocol (MCP) Workbench）

AssistantAgent 还可以通过 McpWorkbench() 使用来自 Model Context Protocol (MCP) 服务器 提供的工具。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams# 从 mcp-server-fetch 获取 fetch 工具。
fetch_mcp_server = StdioServerParams(command="uvx", args=["mcp-server-fetch"])# 创建一个 MCP 工作台，提供与 MCP 服务器的会话。
async with McpWorkbench(fetch_mcp_server) as workbench:  # type: ignore# 创建一个可以使用 fetch 工具的代理。model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")fetch_agent = AssistantAgent(name="fetcher", model_client=model_client, workbench=workbench, reflect_on_tool_use=True)# 让代理获取指定 URL 的内容并进行总结。result = await fetch_agent.run(task="Summarize the content of https://en.wikipedia.org/wiki/Seattle")assert isinstance(result.messages[-1], TextMessage)print(result.messages[-1].content)# 关闭模型客户端的连接。await model_client.close()

示例输出：

Seattle is a major city located in the state of Washington, United States. 
It was founded on November 13, 1851, and incorporated as a town on January 14, 1865, 
and later as a city on December 2, 1869. The city is named after Chief Seattle. 
It covers an area of approximately 142 square miles, with a population of around 737,000 as of the 2020 Census, 
and an estimated 755,078 residents in 2023. 
Seattle is known by nicknames such as The Emerald City, Jet City, and Rain City, 
and has mottos including The City of Flowers and The City of Goodwill. 
The city operates under a mayor–council government system, with Bruce Harrell serving as mayor. 
Key landmarks include the Space Needle, Pike Place Market, Amazon Spheres, and the Seattle Great Wheel. 
It is situated on the U.S. West Coast, with a diverse urban and metropolitan area 
that extends to a population of over 4 million in the greater metropolitan region.

代理作为工具（Agent as a Tool）

任何 BaseChatAgent 都可以通过封装在 AgentTool 中被当作工具使用。
这使得代理能够调用其他代理作为工具，从而支持动态的、由模型驱动的多智能体工作流。

并行工具调用（Parallel Tool Calls）

部分模型支持并行工具调用，这对于需要同时调用多个工具的任务非常有用。
默认情况下，如果模型客户端产生多个工具调用，AssistantAgent 会并行执行它们。

⚠️ 需要注意：

• 如果工具具有可能相互干扰的副作用，或
• 代理行为需要在不同模型间保持一致性，

则应禁用并行工具调用。此操作应在 模型客户端级别 配置。

重要说明（Important）

• 当使用 AgentTool 或 TeamTool 时，必须禁用并行工具调用，以避免并发问题。

  • 因为代理和团队维护内部状态，若并行运行会产生冲突。

• 在 OpenAIChatCompletionClient 和 AzureOpenAIChatCompletionClient 中，可以通过设置 parallel_tool_calls=False 来关闭并行工具调用：

model_client_no_parallel_tool_call = OpenAIChatCompletionClient(model="gpt-4o",parallel_tool_calls=False,  # 禁用并行工具调用
)agent_no_parallel_tool_call = AssistantAgent(name="assistant",model_client=model_client_no_parallel_tool_call,tools=[web_search],system_message="Use tools to solve tasks.",
)

工具迭代（Tool Iterations）

一次模型调用，随后进行一次工具调用或并行工具调用，称为 单次工具迭代（single tool iteration）。
默认情况下，AssistantAgent 最多执行 一次迭代。

代理可以配置为执行 多次迭代，直到模型不再生成工具调用，或达到设置的最大迭代次数。
通过在 AssistantAgent 构造函数中设置 max_tool_iterations 参数，可以控制最大迭代次数。

agent_loop = AssistantAgent(name="assistant_loop",model_client=model_client_no_parallel_tool_call,tools=[web_search],system_message="Use tools to solve tasks.",max_tool_iterations=10,  # 在停止循环之前，最多执行 10 次工具迭代
)

结构化输出（Structured Output）

结构化输出允许模型根据应用提供的预定义 schema 返回 结构化 JSON 文本。
与 JSON 模式不同，这里的 schema 可以使用 Pydantic BaseModel 类提供，同时也可以用于验证输出。

一旦在 AssistantAgent 构造函数的 output_content_type 参数中指定了 BaseModel 类，代理将返回 StructuredMessage，其 content 的类型即为所指定的 BaseModel 类。

这样，你可以将代理的响应直接整合到应用中，并将模型输出作为结构化对象使用。

💡 注意

• 当设置了 output_content_type 时，默认会要求代理对工具使用进行反思（reflect on tool use），并根据工具调用结果返回结构化输出消息。
• 如果不希望启用此行为，可以显式设置 reflect_on_tool_use=False。
• 结构化输出还可以用于在代理响应中整合 Chain-of-Thought（推理链）。

示例代码：

from typing import Literal
from pydantic import BaseModel
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import StructuredMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.ui import Console# 定义代理的响应格式（Pydantic BaseModel）
class AgentResponse(BaseModel):thoughts: strresponse: Literal["happy", "sad", "neutral"]# 创建使用 OpenAI GPT-4o 模型的代理
model_client = OpenAIChatCompletionClient(model="gpt-4o")
agent = AssistantAgent("assistant",model_client=model_client,system_message="Categorize the input as happy, sad, or neutral following the JSON format.",output_content_type=AgentResponse,  # 定义代理的输出类型
)# 流式运行代理
result = await Console(agent.run_stream(task="I am happy."))# 检查最后一条消息类型，并打印 thoughts 和 response
assert isinstance(result.messages[-1], StructuredMessage)
assert isinstance(result.messages[-1].content, AgentResponse)
print("Thought: ", result.messages[-1].content.thoughts)
print("Response: ", result.messages[-1].content.response)
await model_client.close()

示例输出：

---------- user ----------
I am happy.---------- assistant ----------
{"thoughts": "The user explicitly states they are happy.","response": "happy"
}Thought:  The user explicitly states they are happy.
Response:  happy

反思（reflect）是指代理在生成结构化输出前，会理解、分析和总结工具调用的结果，然后把这些信息映射到结构化模型中，而不是简单地封装原始文本。

核心意思是：

1. 分析工具调用的结果

   • 当代理调用一个工具（比如 web_search 或 MCP 工具）后，它会得到工具的原始输出，这个输出可能是文本、JSON、网页内容或者其他格式。
   • 代理需要“理解”这个输出的意义，而不是直接把原始内容放入结构化字段。

2. 生成结构化输出

   • 在理解工具输出后，代理会根据你在 output_content_type 中定义的 BaseModel schema，将工具输出映射到相应的字段。
   • 例如，如果你定义了 thoughts 和 response 两个字段，代理会从工具结果中提取信息填充这些字段，而不仅仅是把原始文本放入某个字段。

3. 可能包含推理或总结

   • 如果工具输出复杂，代理可以做一些 Chain-of-Thought 风格的推理，总结、归纳或者分类工具结果，以便符合结构化输出要求。

流式令牌（Streaming Tokens）

你可以通过设置 model_client_stream=True 来流式获取模型客户端生成的 令牌（tokens）。
这样，在调用 run_stream() 时，代理会依次产生 ModelClientStreamingChunkEvent 消息。

注意：底层模型 API 必须支持流式输出令牌，才能启用此功能。请向你的模型提供商确认是否支持。

示例代码：

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
import asynciomodel_client = OpenAIChatCompletionClient(model="gpt-4o")streaming_assistant = AssistantAgent(name="assistant",model_client=model_client,system_message="You are a helpful assistant.",model_client_stream=True,  # 启用流式令牌
)# 使用异步函数，在脚本中使用 asyncio.run()
async for message in streaming_assistant.run_stream(task="Name two cities in South America"):  # type: ignoreprint(message)

示例输出（部分）：

source='user' models_usage=None metadata={} content='Name two cities in South America' type='TextMessage'
source='assistant' models_usage=None metadata={} content='Two' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' cities' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' in' type='ModelClientStreamingChunkEvent'
...
source='assistant' models_usage=None metadata={} content=' Brazil' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content='.' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0) metadata={} content='Two cities in South America are Buenos Aires in Argentina and São Paulo in Brazil.' type='TextMessage'

说明：

• 以上输出展示了 流式块（streaming chunks），它们是模型客户端生成并实时通过代理产出的。
• 最终的完整响应是将所有流式块 拼接后的内容，在最后一个块之后作为 TextMessage 返回。

使用模型上下文（Using Model Context）

AssistantAgent 提供了一个 model_context 参数，可以传入一个 ChatCompletionContext 对象。
这使得代理能够使用不同的模型上下文策略，例如：

• UnboundedChatCompletionContext（默认）：始终将完整的对话历史发送给模型。
• BufferedChatCompletionContext：仅保留最近的 n 条消息 作为上下文。
• TokenLimitedChatCompletionContext：通过 token 数量 限制上下文长度。

示例：仅使用最近 5 条消息作为上下文：

from autogen_core.model_context import BufferedChatCompletionContext# 创建一个仅使用最近 5 条消息作为上下文的代理
agent = AssistantAgent(name="assistant",model_client=model_client,tools=[web_search],system_message="Use tools to solve tasks.",model_context=BufferedChatCompletionContext(buffer_size=5),  # 仅使用最后 5 条消息
)

预置代理（Preset Agents）

除了 AssistantAgent，框架还提供了一些内置代理：

• UserProxyAgent：一个简单代理，直接返回用户输入作为响应。
• CodeExecutorAgent：可以执行代码的代理。
• OpenAIAssistantAgent：基于 OpenAI Assistant 的代理，支持使用自定义工具。
• MultimodalWebSurfer：一个多模态代理，可以搜索网页并访问网站获取信息。
• FileSurfer：可以搜索和浏览本地文件的代理。
• VideoSurfer：可以观看视频并提取信息的代理。

团队（Teams）

团队是由多个代理组成的群体，它们协作以实现共同的目标。

AgentChat 支持多种预置团队类型：

• RoundRobinGroupChat：团队以轮转方式进行群聊，每个参与者依次发言。
• SelectorGroupChat：团队在每条消息后，使用 ChatCompletion 模型选择下一位发言者。
• MagenticOneGroupChat：通用型多智能体系统，可在多个领域处理开放式网页和文件任务。
• Swarm：团队使用 HandoffMessage 信号在代理间进行切换。

💡 注意

何时使用团队？

• 团队适用于需要协作与多样化专业知识的复杂任务。
• 与单个代理相比，团队需要更多的框架和引导来控制行为。
• 虽然 AutoGen 简化了团队操作，但对于简单任务，建议先使用单个代理；当单个代理不足以解决任务时，再过渡到多智能体团队。
• 在过渡前，请确保单个代理已经通过合适的工具和指令进行了优化。

创建团队（Creating a Team）

RoundRobinGroupChat 是一种简单但高效的团队配置方式：

• 所有代理共享相同的上下文（context），
• 每个代理轮流发言（round-robin），
• 每位代理在轮到自己时，将响应广播给团队中的其他所有代理，确保团队保持一致的上下文。

下面示例中，我们创建了一个包含 两名 AssistantAgent 的团队，并设置 TextMentionTermination 条件，当代理响应中出现指定词汇时，团队任务终止。

• 该两代理团队实现了 反思模式（reflection pattern）：

  • 一名 primary agent 负责生成主要响应
  • 一名 critic agent 对 primary agent 的响应进行评价

• 反思模式可以参考 Core API 学习更多内容。

示例代码：

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient# 创建 OpenAI 模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06",# api_key="sk-...", # 如果已设置 OPENAI_API_KEY 可省略
)# 创建 primary agent
primary_agent = AssistantAgent("primary",model_client=model_client,system_message="You are a helpful AI assistant.",
)# 创建 critic agent
critic_agent = AssistantAgent("critic",model_client=model_client,system_message="Provide constructive feedback. Respond with 'APPROVE' when your feedback is addressed.",
)# 定义终止条件：当 critic 发送 “APPROVE” 时任务结束
text_termination = TextMentionTermination("APPROVE")# 创建包含 primary 和 critic 的团队
team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=text_termination)

运行团队（Running a Team）

调用 run() 方法即可启动团队执行任务：

# 异步运行
result = await team.run(task="Write a short poem about the fall season.")
print(result)

示例输出（简化说明）：

• 团队按轮转顺序运行代理，
• primary agent 生成诗歌内容，
• critic agent 提供反馈，
• primary agent 根据反馈修改诗歌，
• 当 critic agent 响应中出现 "APPROVE" 时，团队终止任务。

最终返回 TaskResult 对象，包含团队中所有代理生成的消息：

TaskResult(messages=[TextMessage(source='user', content='Write a short poem about the fall season.'),TextMessage(source='primary', content='Leaves of amber, gold, and rust...'),TextMessage(source='critic', content='Your poem beautifully captures the essence of fall...'),TextMessage(source='primary', content='Thank you for the feedback. Here's a revised version...'),TextMessage(source='critic', content='APPROVE')],stop_reason="Text 'APPROVE' mentioned"
)

• 团队会持续运行，直到满足 termination_condition。
• 在本例中，当代理响应中检测到 "APPROVE" 时任务结束。

观察团队（Observing a Team）

类似于代理的 on_messages_stream() 方法，你可以通过 run_stream() 在团队运行时实时获取团队生成的消息。

• 该方法返回一个 异步生成器（generator），
• 每条消息在生成时立即产出，
• 最后一项为 TaskResult 对象，包含团队执行任务的完整结果。

示例代码：

# 在脚本中使用异步主函数，并通过 asyncio.run() 调用
await team.reset()  # 重置团队以开始新任务async for message in team.run_stream(task="Write a short poem about the fall season."):  # type: ignoreif isinstance(message, TaskResult):print("Stop Reason:", message.stop_reason)else:print(message)

示例输出说明：

source='user' content='Write a short poem about the fall season.' type='TextMessage'
source='primary' content='Golden leaves in crisp air dance, ...' type='TextMessage'
source='critic' content='Your poem beautifully captures the essence of the fall season...' type='TextMessage'
source='primary' content='Thank you for your thoughtful feedback! Here is a revised version...' type='TextMessage'
source='critic' content='APPROVE' type='TextMessage'
Stop Reason: Text 'APPROVE' mentioned

• 你可以通过 stop_reason 属性查看团队停止的原因。
• Console() 方法提供了方便的方式，将团队消息以格式化的方式实时输出到控制台：

await team.reset()  # 重置团队
await Console(team.run_stream(task="Write a short poem about the fall season."))  # 流式输出消息到控制台

---------- user ----------
Write a short poem about the fall season.
---------- primary ----------
Golden leaves in crisp air dance,  
Whispering tales as they prance.  
Amber hues paint the ground,  
Nature's symphony all around.  Sweaters hug with tender grace,  
While pumpkins smile, a warm embrace.  
Chill winds hum through towering trees,  
A vibrant tapestry in the breeze.  Harvest moons in twilight glow,  
Casting magic on fields below.  
Fall's embrace, a gentle call,  
To savor beauty before snowfalls.  
[Prompt tokens: 28, Completion tokens: 99]
---------- critic ----------
Your poem beautifully captures the essence of the fall season, creating a vivid and cozy atmosphere. The imagery of golden leaves and amber hues paints a picturesque scene that many can easily relate to. I particularly appreciate the personification of pumpkins and the gentle embrace of sweaters, which adds warmth to your verses. To enhance the poem further, you might consider adding more sensory details to make the reader feel even more immersed in the experience. For example, including specific sounds, scents, or textures could deepen the connection to autumn's ambiance. Additionally, you could explore the emotional transitions as the season prepares for winter to provide a reflective element to the piece.Overall, it's a lovely and evocative depiction of fall, evoking feelings of comfort and appreciation for nature's changing beauty. Great work!
[Prompt tokens: 144, Completion tokens: 157]
---------- primary ----------
Thank you for your thoughtful feedback! I'm glad you enjoyed the imagery and warmth in the poem. To enhance the sensory experience and emotional depth, here's a revised version incorporating your suggestions:---Golden leaves in crisp air dance,  
Whispering tales as they prance.  
Amber hues paint the crunchy ground,  
Nature's symphony all around.  Sweaters hug with tender grace,  
While pumpkins grin, a warm embrace.  
Chill winds hum through towering trees,  
Crackling fires warm the breeze.  Apples in the orchard's glow,  
Sweet cider scents that overflow.  
Crunch of paths beneath our feet,  
Cinnamon spice and toasty heat.  Harvest moons in twilight's glow,  
Casting magic on fields below.  
Fall's embrace, a gentle call,  
Reflects on life's inevitable thaw.  --- I hope this version enhances the sensory and emotional elements of the season. Thank you again for your insights!
[Prompt tokens: 294, Completion tokens: 195]
---------- critic ----------
APPROVE
[Prompt tokens: 506, Completion tokens: 4]
---------- Summary ----------
Number of messages: 5
Finish reason: Text 'APPROVE' mentioned
Total prompt tokens: 972
Total completion tokens: 455
Duration: 11.78 seconds

TaskResult(messages=[TextMessage(source='user', models_usage=None, content='Write a short poem about the fall season.', type='TextMessage'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=28, completion_tokens=99), content="Golden leaves in crisp air dance,  \nWhispering tales as they prance.  \nAmber hues paint the ground,  \nNature's symphony all around.  \n\nSweaters hug with tender grace,  \nWhile pumpkins smile, a warm embrace.  \nChill winds hum through towering trees,  \nA vibrant tapestry in the breeze.  \n\nHarvest moons in twilight glow,  \nCasting magic on fields below.  \nFall's embrace, a gentle call,  \nTo savor beauty before snowfalls.  ", type='TextMessage'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=144, completion_tokens=157), content="Your poem beautifully captures the essence of the fall season, creating a vivid and cozy atmosphere. The imagery of golden leaves and amber hues paints a picturesque scene that many can easily relate to. I particularly appreciate the personification of pumpkins and the gentle embrace of sweaters, which adds warmth to your verses. \n\nTo enhance the poem further, you might consider adding more sensory details to make the reader feel even more immersed in the experience. For example, including specific sounds, scents, or textures could deepen the connection to autumn's ambiance. Additionally, you could explore the emotional transitions as the season prepares for winter to provide a reflective element to the piece.\n\nOverall, it's a lovely and evocative depiction of fall, evoking feelings of comfort and appreciation for nature's changing beauty. Great work!", type='TextMessage'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=294, completion_tokens=195), content="Thank you for your thoughtful feedback! I'm glad you enjoyed the imagery and warmth in the poem. To enhance the sensory experience and emotional depth, here's a revised version incorporating your suggestions:\n\n---\n\nGolden leaves in crisp air dance,  \nWhispering tales as they prance.  \nAmber hues paint the crunchy ground,  \nNature's symphony all around.  \n\nSweaters hug with tender grace,  \nWhile pumpkins grin, a warm embrace.  \nChill winds hum through towering trees,  \nCrackling fires warm the breeze.  \n\nApples in the orchard's glow,  \nSweet cider scents that overflow.  \nCrunch of paths beneath our feet,  \nCinnamon spice and toasty heat.  \n\nHarvest moons in twilight's glow,  \nCasting magic on fields below.  \nFall's embrace, a gentle call,  \nReflects on life's inevitable thaw.  \n\n--- \n\nI hope this version enhances the sensory and emotional elements of the season. Thank you again for your insights!", type='TextMessage'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=506, completion_tokens=4), content='APPROVE', type='TextMessage')], stop_reason="Text 'APPROVE' mentioned")

总结

• run_stream() 允许你实时观察团队内部的多智能体互动。
• 每条消息即时输出，有助于调试和理解团队行为。
• 团队停止时，会返回 TaskResult，包含所有生成的消息以及终止原因。

重置团队（Resetting a Team）

可以通过调用 reset() 方法重置团队。

• 此方法会清空团队状态，包括所有代理（agents）的状态。
• 会调用每个代理的 on_reset() 方法，以清除代理自身的状态。

示例代码：

await team.reset()  # 为下一次任务重置团队

使用建议：

• 如果下一次任务与前一次任务无关，建议先重置团队。
• 如果下一次任务与前一次任务相关，则无需重置，可以直接恢复团队状态继续执行。

停止团队（Stopping a Team）

除了基于内部状态的自动终止条件（如 TextMentionTermination）外，你还可以使用 ExternalTermination 从外部停止团队。

• 调用 set() 方法后，团队将在当前代理完成轮次后停止。
• 团队不会立即停止，而是让当前代理完成其回合并广播最终消息，保证团队状态一致。

示例代码：

from autogen_agentchat.conditions import ExternalTermination
from autogen_agentchat.teams import RoundRobinGroupChat
import asyncio# 创建外部终止条件
external_termination = ExternalTermination()# 创建团队，并组合终止条件
team = RoundRobinGroupChat([primary_agent, critic_agent],termination_condition=external_termination | text_termination,  # 使用按位或组合条件
)# 在后台任务中运行团队
run = asyncio.create_task(Console(team.run_stream(task="Write a short poem about the fall season.")))# 等待一段时间
await asyncio.sleep(0.1)# 外部请求停止团队
external_termination.set()# 等待团队完成当前回合
await run

示例输出说明：

• 团队停止的原因为 外部终止（External termination requested）。
• 发言中的代理能够完成当前轮次，保证团队状态一致。

---------- user ----------
Write a short poem about the fall season.
---------- primary ----------
Leaves of amber, gold, and red,  
Gently drifting from trees overhead.  
Whispers of wind through the crisp, cool air,  
Nature's canvas painted with care.  Harvest moons and evenings that chill,  
Fields of plenty on every hill.  
Sweaters wrapped tight as twilight nears,  
Fall's charming embrace, as warm as it appears.  Pumpkins aglow with autumn's light,  
Harvest feasts and stars so bright.  
In every leaf and breeze that calls,  
We find the magic of glorious fall.  
[Prompt tokens: 28, Completion tokens: 114]
---------- Summary ----------
Number of messages: 2
Finish reason: External termination requested
Total prompt tokens: 28
Total completion tokens: 114
Duration: 1.71 seconds

输出示例：

TaskResult(messages=[TextMessage(source='user', content='Write a short poem about the fall season.'),TextMessage(source='primary', content='Leaves of amber, gold, and red...'),],stop_reason='External termination requested'
)

可以看到，团队因外部终止条件触发而停止，但发言代理顺利完成了自己的回合。

这个机制可以确保：

1. 外部能够控制团队任务的终止，
2. 当前正在发言的代理有机会完成输出，
3. 团队状态保持一致，避免中途打断造成的数据不完整。

恢复团队（Resuming a Team）

团队（Teams）是有状态的，会在每次运行后保留对话历史和上下文，除非你重置了团队。

你可以通过再次调用 run() 或 run_stream() 方法来恢复团队，让它从上次停止的地方继续执行。

• 对于 RoundRobinGroupChat，团队会从上次发言的下一个代理开始轮转。

示例代码：

await Console(team.run_stream())  # 恢复团队，继续上次任务

---------- critic ----------
This poem beautifully captures the essence of the fall season with vivid imagery and a soothing rhythm. The descriptions of the changing leaves, cool air, and various autumn traditions make it easy for readers to envision and feel the charm of fall. Here are a few suggestions to enhance its impact:1. **Structure Variation**: Consider breaking some lines with a hyphen or ellipsis for dramatic effect or emphasis. For instance, “Sweaters wrapped tight as twilight nears— / Fall’s charming embrace, as warm as it appears."2. **Sensory Details**: While the poem already evokes visual and tactile senses, incorporating other senses such as sound or smell could deepen the immersion. For example, include the scent of wood smoke or the crunch of leaves underfoot.3. **Metaphorical Language**: Adding metaphors or similes can further enrich the imagery. For example, you might compare the leaves falling to a golden rain or the chill in the air to a gentle whisper.Overall, it’s a lovely depiction of fall. These suggestions are minor tweaks that might elevate the reader's experience even further. Nice work!Let me know if these feedbacks are addressed.
[Prompt tokens: 159, Completion tokens: 237]
---------- primary ----------
Thank you for the thoughtful feedback! Here’s a revised version, incorporating your suggestions:  Leaves of amber, gold—drifting like dreams,  
A golden rain from trees’ canopies.  
Whispers of wind—a gentle breath,  
Nature’s scented tapestry embracing earth.  Harvest moons rise as evenings chill,  
Fields of plenty paint every hill.  
Sweaters wrapped tight as twilight nears—  
Fall’s embrace, warm as whispered years.  Pumpkins aglow with autumn’s light,  
Crackling leaves underfoot in flight.  
In every leaf and breeze that calls,  
We find the magic of glorious fall.  I hope these changes enhance the imagery and sensory experience. Thank you again for your feedback!
[Prompt tokens: 389, Completion tokens: 150]
---------- critic ----------
Your revisions have made the poem even more evocative and immersive. The use of sensory details, such as "whispers of wind" and "crackling leaves," beautifully enriches the poem, engaging multiple senses. The metaphorical language, like "a golden rain from trees’ canopies" and "Fall’s embrace, warm as whispered years," adds depth and enhances the emotional warmth of the poem. The structural variation with the inclusion of dashes effectively adds emphasis and flow. Overall, these changes bring greater vibrancy and life to the poem, allowing readers to truly experience the wonders of fall. Excellent work on the revisions!APPROVE
[Prompt tokens: 556, Completion tokens: 132]
---------- Summary ----------
Number of messages: 3
Finish reason: Text 'APPROVE' mentioned
Total prompt tokens: 1104
Total completion tokens: 519
Duration: 9.79 seconds

示例说明：

• 在上面的输出中，你可以看到团队从上次停止的位置继续执行。
• 第一个消息来自上次停止后轮到的下一位代理。

你也可以在保持前一次任务上下文的情况下，给团队分配新的任务。

• 这样团队会在原有对话和状态的基础上执行新的任务。

示例代码：

# 新任务：将前一首诗改写成中文唐诗风格
await Console(team.run_stream(task="将这首诗用中文唐诗风格写一遍。"))

---------- user ----------
将这首诗用中文唐诗风格写一遍。
---------- primary ----------
朔风轻拂叶飘金，  
枝上斜阳染秋林。  
满山丰收人欢喜，  
月明归途衣渐紧。  南瓜影映灯火中，  
落叶沙沙伴归程。  
片片秋意随风起，  
秋韵悠悠心自明。  
[Prompt tokens: 700, Completion tokens: 77]
---------- critic ----------
这首改编的唐诗风格诗作成功地保留了原诗的意境与情感，体现出秋季特有的氛围和美感。通过“朔风轻拂叶飘金”、“枝上斜阳染秋林”等意象，生动地描绘出了秋天的景色，与唐诗中的自然意境相呼应。且“月明归途衣渐紧”、“落叶沙沙伴归程”让人感受到秋天的安宁与温暖。通过这些诗句，读者能够感受到秋天的惬意与宁静，勾起丰收与团圆的画面，是一次成功的翻译改编。APPROVE
[Prompt tokens: 794, Completion tokens: 161]
---------- Summary ----------
Number of messages: 3
Finish reason: Text 'APPROVE' mentioned
Total prompt tokens: 1494
Total completion tokens: 238
Duration: 3.89 seconds

TaskResult(messages=[TextMessage(source='user', models_usage=None, content='将这首诗用中文唐诗风格写一遍。', type='TextMessage'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=700, completion_tokens=77), content='朔风轻拂叶飘金，  \n枝上斜阳染秋林。  \n满山丰收人欢喜，  \n月明归途衣渐紧。  \n\n南瓜影映灯火中，  \n落叶沙沙伴归程。  \n片片秋意随风起，  \n秋韵悠悠心自明。  ', type='TextMessage'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=794, completion_tokens=161), content='这首改编的唐诗风格诗作成功地保留了原诗的意境与情感，体现出秋季特有的氛围和美感。通过“朔风轻拂叶飘金”、“枝上斜阳染秋林”等意象，生动地描绘出了秋天的景色，与唐诗中的自然意境相呼应。且“月明归途衣渐紧”、“落叶沙沙伴归程”让人感受到秋天的安宁与温暖。\n\n通过这些诗句，读者能够感受到秋天的惬意与宁静，勾起丰收与团圆的画面，是一次成功的翻译改编。\n\nAPPROVE', type='TextMessage')], stop_reason="Text 'APPROVE' mentioned")

说明：

• 新任务基于前一次任务的上下文进行，保留了团队的状态和信息。
• 批评者（critic）对改写结果进行了评价，并通过 APPROVE 确认任务完成。

这种机制使得团队能够在多轮任务中保持连续性，并在新的任务中充分利用之前积累的上下文和信息。

中止团队（Aborting a Team）

你可以在团队执行 run() 或 run_stream() 时，通过 cancellation_token 参数传入的 CancellationToken 来中止运行。

与停止团队（Stopping a Team）不同，中止团队会立即终止运行，并抛出 CancelledError 异常。

注意：

• 调用方会在团队被中止时收到 CancelledError 异常。

示例代码：

from autogen_core import CancellationToken
import asyncio# 创建一个取消令牌
cancellation_token = CancellationToken()# 使用协程运行团队
run = asyncio.create_task(team.run(task="Translate the poem to Spanish.",cancellation_token=cancellation_token,)
)# 取消运行
cancellation_token.cancel()try:result = await run  # 这将抛出 CancelledError
except asyncio.CancelledError:print("Task was cancelled.")

输出示例：

Task was cancelled.

说明：

• 使用 CancellationToken 可以在运行过程中立即中止团队任务，而不会等待当前代理完成发言。
• 适用于紧急停止或异常处理场景。

单代理团队（Single-Agent Team）

注意：

从 0.6.2 版本开始，你可以在 AssistantAgent 中使用 max_tool_iterations 来让代理多次调用工具循环执行。因此，如果只是想让代理在工具调用循环中运行，可能不需要单独使用单代理团队。

在某些情况下，你可能希望以团队配置运行单个代理。这在需要循环执行 AssistantAgent，直到满足某个终止条件时非常有用。

这与直接调用 AssistantAgent 的 run() 或 run_stream() 不同，后者只运行一次步骤并返回结果。

以下是一个示例：在 RoundRobinGroupChat 团队配置下运行单个代理，并使用 TextMessageTermination 条件。任务是使用工具将数字从 5 增加到 10。代理会不断调用工具，直到数字达到 10，然后返回最终 TextMessage，团队运行停止。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient# 创建模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o",parallel_tool_calls=False,  # 禁用并行工具调用
)# 创建一个递增数字的工具
def increment_number(number: int) -> int:"""数字加1"""return number + 1# 创建使用工具的代理
looped_assistant = AssistantAgent("looped_assistant",model_client=model_client,tools=[increment_number],system_message="你是一个有用的 AI 助手，使用工具将数字递增。",
)# 定义终止条件：当代理返回 TextMessage 时停止
termination_condition = TextMessageTermination("looped_assistant")# 创建单代理团队
team = RoundRobinGroupChat([looped_assistant],termination_condition=termination_condition,
)# 运行团队任务并输出消息
async for message in team.run_stream(task="Increment the number 5 to 10."):  # type: ignoreprint(type(message).__name__, message)await model_client.close()

输出示例显示了每一步工具调用和结果：

ToolCallRequestEvent -> ToolCallExecutionEvent -> ToolCallSummaryMessage
...
TextMessage: 'The number 5 incremented to 10 is 10.'

要点：

• 核心在于 终止条件。
• 示例中使用 TextMessageTermination 条件，当代理停止生成 ToolCallSummaryMessage 并返回最终 TextMessage 时，团队停止运行。
• 你也可以使用其他终止条件来控制代理运行。

人机协作（Human-in-the-Loop）

在前一节 Teams 中，我们已经了解了如何创建、观察和控制一个代理团队。本节将重点介绍如何从你的应用程序与团队交互，并向团队提供人工反馈。

从应用程序与团队交互主要有两种方式：

1. 在团队运行期间 —— 当 run() 或 run_stream() 正在执行时，通过 UserProxyAgent 提供反馈。
2. 运行结束后 —— 当一次运行结束后，通过下一次调用 run() 或 run_stream() 的输入提供反馈。

在运行期间提供反馈

UserProxyAgent 是一个特殊的内置代理，充当用户的代理，用于向团队提供反馈。

要使用 UserProxyAgent，可以创建它的实例，并在运行团队之前将其包含在团队中。团队会决定何时调用 UserProxyAgent 向用户请求反馈。

例如，在 RoundRobinGroupChat 团队中，UserProxyAgent 按传入团队的顺序被调用；而在 SelectorGroupChat 团队中，选择器提示或选择器函数会决定何时调用 UserProxyAgent。

下图展示了如何在团队运行期间使用 UserProxyAgent 从用户那里获取反馈：

粗体箭头表示团队运行期间的控制流：当团队调用 UserProxyAgent 时，控制权会转移到应用程序/用户，并等待用户反馈；一旦反馈提供完毕，控制权会转回给team，team继续执行。

注意

当 UserProxyAgent 在运行期间被调用时，它会阻塞团队的执行，直到用户提供反馈或发生错误。这会延缓团队的进度，并使团队处于无法保存或恢复的不稳定状态。

由于这种方法具有阻塞性，建议仅在需要用户即时反馈的短交互场景中使用，例如点击按钮以表示批准或拒绝，或弹出警报要求立即处理，否则任务将失败。

下面是一个在 RoundRobinGroupChat 中使用 UserProxyAgent 完成诗歌生成任务的示例：

from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient# 创建代理
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)
user_proxy = UserProxyAgent("user_proxy", input_func=input)  # 使用 input() 从控制台获取用户输入# 创建终止条件，当用户输入 "APPROVE" 时结束对话
termination = TextMentionTermination("APPROVE")# 创建团队
team = RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination)# 运行对话并输出到控制台
stream = team.run_stream(task="写一首关于海洋的四行诗。")
# 在脚本中运行时使用 asyncio.run(...)
await Console(stream)
await model_client.close()

运行结果示例：

---------- user ----------
Write a 4-line poem about the ocean.
---------- assistant ----------
In endless blue where whispers play,  
The ocean's waves dance night and day.  
A world of depths, both calm and wild,  
Nature's heart, forever beguiled.  
TERMINATE
---------- user_proxy ----------
APPROVE

TaskResult(messages=[TextMessage(source='user', models_usage=None, metadata={}, content='Write a 4-line poem about the ocean.', type='TextMessage'), TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=46, completion_tokens=43), metadata={}, content="In endless blue where whispers play,  \nThe ocean's waves dance night and day.  \nA world of depths, both calm and wild,  \nNature's heart, forever beguiled.  \nTERMINATE", type='TextMessage'), UserInputRequestedEvent(source='user_proxy', models_usage=None, metadata={}, request_id='2622a0aa-b776-4e54-9e8f-4ecbdf14b78d', content='', type='UserInputRequestedEvent'), TextMessage(source='user_proxy', models_usage=None, metadata={}, content='APPROVE', type='TextMessage')], stop_reason="Text 'APPROVE' mentioned")

从控制台输出可以看出，团队通过 user_proxy 向用户请求反馈，以批准生成的诗歌。

你可以为 UserProxyAgent 提供自定义的输入函数，以定制反馈流程。例如，当团队以 Web 服务形式运行时，你可以使用自定义输入函数来等待来自 WebSocket 连接的消息。下面的代码片段展示了在使用 FastAPI Web 框架时的自定义输入函数示例：

@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):await websocket.accept()async def _user_input(prompt: str, cancellation_token: CancellationToken | None) -> str:data = await websocket.receive_json()  # 等待来自 websocket 的用户消息message = TextMessage.model_validate(data)  # 假设用户消息是 TextMessage 类型return message.content# 使用自定义输入函数创建用户代理# 使用该用户代理运行团队# ...

向下一次运行提供反馈

在很多情况下，应用程序或用户会以交互循环的方式与代理团队进行交互：团队运行直到结束，应用程序或用户提供反馈，然后团队根据反馈再次运行。

这种方法在持久化会话中尤其有用，适用于团队与应用程序/用户之间的异步通信：当团队完成一次运行后，应用程序可以保存团队的状态，将其存储到持久化存储中，并在收到反馈时恢复团队运行。

下图展示了这种方法中的控制流程：

实现这种方法有两种方式：

1. 设置最大回合数：让团队在达到指定回合数后总是停止。

2. 使用终止条件：例如 TextMentionTermination 和 HandoffTermination，让团队根据自身的内部状态决定何时停止并将控制权交回。

你也可以将这两种方法结合使用，以实现期望的行为。

使用最大回合数

这种方法允许你通过设置最大回合数来暂停团队以等待用户输入。例如，你可以通过将 max_turns 设置为 1，让团队在第一个代理响应后就停止。这在需要持续用户参与的场景中尤其有用，比如聊天机器人。

要实现这一点，可以在 RoundRobinGroupChat() 构造函数中设置 max_turns 参数：

team = RoundRobinGroupChat([...], max_turns=1)

一旦团队停止，回合计数将被重置。当你恢复团队时，它将从 0 开始重新计数。然而，团队的内部状态会被保留，例如，RoundRobinGroupChat 会从列表中的下一个代理继续，且保持相同的对话历史。

注意

max_turn 是团队类特有的，目前仅被 RoundRobinGroupChat、SelectorGroupChat 和 Swarm 支持。当与终止条件一起使用时，只要满足任一条件，团队就会停止。

下面是一个在 RoundRobinGroupChat 中使用 max_turns 进行诗歌生成任务的示例，最大回合数为 1：

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient# 创建代理
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)# 创建团队，并设置最大回合数为 1
team = RoundRobinGroupChat([assistant], max_turns=1)task = "Write a 4-line poem about the ocean."
while True:# 运行对话并流式输出到控制台stream = team.run_stream(task=task)# 在脚本中运行时使用 asyncio.run(...)await Console(stream)# 获取用户反馈task = input("Enter your feedback (type 'exit' to leave): ")if task.lower().strip() == "exit":break
await model_client.close()

示例运行效果：

---------- user ----------
Write a 4-line poem about the ocean.
---------- assistant ----------
Endless waves in a dance with the shore,  
Whispers of secrets in tales from the roar,  
Beneath the vast sky, where horizons blend,  
The ocean’s embrace is a timeless friend.  
TERMINATE
[Prompt tokens: 46, Completion tokens: 48]
---------- Summary ----------
Number of messages: 2
Finish reason: Maximum number of turns 1 reached.
Total prompt tokens: 46
Total completion tokens: 48
Duration: 1.63 seconds
---------- user ----------
Can you make it about a person and its relationship with the ocean
---------- assistant ----------
She walks along the tide, where dreams intertwine,  
With every crashing wave, her heart feels aligned,  
In the ocean's embrace, her worries dissolve,  
A symphony of solace, where her spirit evolves.  
TERMINATE
[Prompt tokens: 117, Completion tokens: 49]
---------- Summary ----------
Number of messages: 2
Finish reason: Maximum number of turns 1 reached.
Total prompt tokens: 117
Total completion tokens: 49
Duration: 1.21 seconds

你可以看到，团队在一个代理响应后立即停止了。

使用终止条件

在前面的章节中，我们已经看到过几个终止条件的示例。在本节中，我们重点介绍 HandoffTermination，当代理发送 HandoffMessage 消息时，它会停止团队运行。

下面我们创建一个只有单个 AssistantAgent 的团队，并设置 handoff（移交）配置，然后运行一个任务，该任务需要用户提供额外输入，因为代理没有相关工具来继续处理任务。

注意

使用 AssistantAgent 的模型必须支持工具调用，才能使用 handoff 功能。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient# 创建 OpenAI 模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o",# api_key="sk-...", # 如果你已设置 OPENAI_API_KEY 环境变量，可选
)# 创建一个懒惰助手代理，总是将任务移交给用户
lazy_agent = AssistantAgent("lazy_assistant",model_client=model_client,handoffs=[Handoff(target="user", message="Transfer to user.")],system_message="If you cannot complete the task, transfer to user. Otherwise, when finished, respond with 'TERMINATE'.",
)# 定义一个终止条件，用于检查 handoff 消息
handoff_termination = HandoffTermination(target="user")
# 定义一个终止条件，用于检查特定文本提及
text_termination = TextMentionTermination("TERMINATE")# 创建一个单代理团队，包含懒惰助手和两个终止条件
lazy_agent_team = RoundRobinGroupChat([lazy_agent], termination_condition=handoff_termination | text_termination)# 运行团队并流式输出到控制台
task = "What is the weather in New York?"
await Console(lazy_agent_team.run_stream(task=task), output_stats=True)

示例运行效果：

---------- user ----------
What is the weather in New York?
---------- lazy_assistant ----------
[FunctionCall(id='call_EAcMgrLGHdLw0e7iJGoMgxuu', arguments='{}', name='transfer_to_user')]
[Prompt tokens: 69, Completion tokens: 12]
---------- lazy_assistant ----------
[FunctionExecutionResult(content='Transfer to user.', call_id='call_EAcMgrLGHdLw0e7iJGoMgxuu')]
---------- lazy_assistant ----------
Transfer to user.
---------- Summary ----------
Number of messages: 4
Finish reason: Handoff to user from lazy_assistant detected.
Total prompt tokens: 69
Total completion tokens: 12
Duration: 0.69 seconds
TaskResult(messages=[TextMessage(source='user', models_usage=None, content='What is the weather in New York?', type='TextMessage'), ToolCallRequestEvent(source='lazy_assistant', models_usage=RequestUsage(prompt_tokens=69, completion_tokens=12), content=[FunctionCall(id='call_EAcMgrLGHdLw0e7iJGoMgxuu', arguments='{}', name='transfer_to_user')], type='ToolCallRequestEvent'), ToolCallExecutionEvent(source='lazy_assistant', models_usage=None, content=[FunctionExecutionResult(content='Transfer to user.', call_id='call_EAcMgrLGHdLw0e7iJGoMgxuu')], type='ToolCallExecutionEvent'), HandoffMessage(source='lazy_assistant', models_usage=None, target='user', content='Transfer to user.', context=[], type='HandoffMessage')], stop_reason='Handoff to user from lazy_assistant detected.')

你可以看到，团队因为检测到 handoff 消息而停止。接下来，我们通过提供代理所需的信息来继续团队运行：

await Console(lazy_agent_team.run_stream(task="The weather in New York is sunny."))

示例运行效果：

---------- user ----------
The weather in New York is sunny.
---------- lazy_assistant ----------
Great! Enjoy the sunny weather in New York! Is there anything else you'd like to know?
---------- lazy_assistant ----------
TERMINATE
TaskResult(messages=[TextMessage(source='user', models_usage=None, content='The weather in New York is sunny.', type='TextMessage'), TextMessage(source='lazy_assistant', models_usage=RequestUsage(prompt_tokens=110, completion_tokens=21), content="Great! Enjoy the sunny weather in New York! Is there anything else you'd like to know?", type='TextMessage'), TextMessage(source='lazy_assistant', models_usage=RequestUsage(prompt_tokens=137, completion_tokens=5), content='TERMINATE', type='TextMessage')], stop_reason="Text 'TERMINATE' mentioned")

你可以看到，用户提供信息后，团队继续运行。

注意

如果你使用 Swarm 团队，并且 HandoffTermination 的目标是用户，要恢复团队运行，需要将任务设置为一个 HandoffMessage，目标设为你希望下一次运行的代理。更多详情请参见 Swarm。

怎么理解人机协作

UserProxyAgent` 并不一定非得对应“某个具体用户”，它本质上就是一个代理输入接口，只要你提供一个可调用的函数，这个函数最终返回一个字符串，它就能作为团队的“用户反馈”。

举几个例子：

1. 控制台输入：

   user_proxy = UserProxyAgent(input_func=input)
   

   用户在控制台输入内容，函数返回这个字符串。

2. Web 前端：
   你可以写一个 async 函数，通过 WebSocket 等待前端消息：

   async def web_input(prompt: str, cancellation_token=None):data = await websocket.receive_json()return data["message"]
   user_proxy = UserProxyAgent(input_func=web_input)
   

3. 短信/邮件/IM 通知专家：
   你完全可以实现一个函数，发送消息给专家，然后阻塞或轮询等待专家回复：

   def expert_sms_input(prompt: str):send_sms_to_expert(prompt)reply = wait_for_sms_reply()  # 轮询或阻塞直到收到专家短信return reply
   user_proxy = UserProxyAgent(input_func=expert_sms_input)
   

只要最终返回字符串，团队就能继续运行。

所以关键在于：UserProxy 只是一个回调函数接口，它不关心底层是 Web、短信还是控制台输入。

关于max_turns

team = RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination) 如果不指定退出条件，则默认的退出条件是什么？

在 RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination) 中，如果你没有显式指定 termination_condition，默认行为取决于团队类型。具体来说：

1. 默认退出条件

• RoundRobinGroupChat 的默认退出条件是 永不自动停止。

• 换句话说，如果你不提供 termination_condition，团队会一直轮询每个代理，直到：

  1. 达到 max_turns（如果你设置了 max_turns 参数），或者
  2. 某个代理在内部主动触发了终止（例如返回 "TERMINATE" 或发送 HandoffMessage 并触发 HandoffTermination）。

2. 总结默认逻辑

• 未设置 termination_condition → 不会主动停止（无限轮询）
• 设置 max_turns → 达到回合数后停止
• 设置终止条件 → 条件触发时停止（如 TextMentionTermination("TERMINATE") 或 HandoffTermination）

3. 建议

如果你的团队中有 UserProxyAgent，一般建议至少设置：

• TextMentionTermination("TERMINATE")，或者
• HandoffTermination，确保在任务完成或需要用户介入时能正常停止。

假设你有一个 RoundRobinGroupChat，里面有 3 个代理（AgentA、AgentB、AgentC），并且设置了 max_turns=3。

• max_turns 定义的是 整个团队的总轮次，不是每个代理单独的回合。
• 团队的运行逻辑是 Round Robin（轮流执行）：依次调用列表中的每个代理。

执行流程如下：

初始状态：

• 回合计数：0
• 当前执行代理：AgentA

第 1 个回合：

1. AgentA 执行任务 → 回合计数 +1 → 当前回合数 = 1
2. 团队检查回合数（1 < max_turns=3）→ 继续
3. 切换到下一个代理：AgentB

第 2 个回合：

1. AgentB 执行任务 → 回合计数 +1 → 当前回合数 = 2
2. 团队检查回合数（2 < max_turns=3）→ 继续
3. 切换到下一个代理：AgentC

第 3 个回合：

1. AgentC 执行任务 → 回合计数 +1 → 当前回合数 = 3
2. 团队检查回合数（3 = max_turns=3）→ 达到最大回合数，团队停止

小结

• 顺序执行：AgentA → AgentB → AgentC
• 回合计数：每轮调用一个代理算 1 回合
• 停止条件：当总回合数达到 max_turns 时停止
• 状态保存：团队内部状态会保留，下一次运行从下一个代理开始，且对话历史保持不变

终止条件（Termination）

一个运行可能会无限进行，在很多情况下，我们需要知道何时停止它们。这就是终止条件的作用。

AgentChat 通过提供一个基础类 TerminationCondition 及其多个子类，实现了多种终止条件。

• 终止条件是一个 可调用对象（callable），它接收自上一次调用以来的 BaseAgentEvent 或 BaseChatMessage 对象序列。
• 如果对话应该终止，则返回 StopMessage；否则返回 None。
• 一旦终止条件被触发，需要调用 reset() 才能再次使用。

重要说明

1. 终止条件是 有状态的，但在每次运行结束后（run() 或 run_stream()）会自动重置。
2. 可以使用 AND 和 OR 运算符组合多个终止条件。

注意

• 对于群聊团队（如 RoundRobinGroupChat、SelectorGroupChat 和 Swarm），终止条件会在每个代理响应后被调用。
• 虽然一次响应可能包含多个内部消息，但团队只会对单次响应的所有消息调用一次终止条件。
• 因此，终止条件接收的消息序列是自上一次调用以来的“增量序列（delta sequence）”。

内置终止条件（Built-In Termination Conditions）

1. MaxMessageTermination：当产生指定数量的消息（包括代理和任务消息）后停止。
2. TextMentionTermination：当消息中出现指定文本或字符串时停止（例如 "TERMINATE"）。
3. TokenUsageTermination：当使用的 prompt 或 completion token 达到一定数量时停止，需要代理在消息中报告 token 使用情况。
4. TimeoutTermination：指定秒数后停止。
5. HandoffTermination：当请求移交给特定目标时停止。Handoff 消息可用于构建如 Swarm 的模式，适用于希望在代理移交给应用或用户时暂停运行，以等待输入。
6. SourceMatchTermination：当特定代理响应后停止。
7. ExternalTermination：允许从外部程序控制终止，适合 UI 集成（例如聊天界面的“停止”按钮）。
8. StopMessageTermination：当代理生成 StopMessage 时停止。
9. TextMessageTermination：当代理生成 TextMessage 时停止。
10. FunctionCallTermination：当代理生成包含指定函数名的 ToolCallExecutionEvent 时停止。
11. FunctionalTermination：当对最新的增量消息序列评估的函数表达式返回 True 时停止，可快速创建不在内置列表中的自定义终止条件。

基本用法（Basic Usage）

为了演示终止条件的特点，我们创建一个由两个代理组成的团队：

• Primary Agent（主代理）负责文本生成
• Critic Agent（评论代理）负责审查生成的文本并提供反馈

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClientmodel_client = OpenAIChatCompletionClient(model="gpt-4o",temperature=1,# api_key="sk-...", # 如果已设置 OPENAI_API_KEY 环境变量，可选
)# 创建主代理
primary_agent = AssistantAgent("primary",model_client=model_client,system_message="You are a helpful AI assistant.",
)# 创建评论代理
critic_agent = AssistantAgent("critic",model_client=model_client,system_message="Provide constructive feedback for every message. Respond with 'APPROVE' to when your feedbacks are addressed.",
)

我们来看终止条件如何在每次 run 或 run_stream 调用后自动重置，从而允许团队从上次中断的位置继续对话。

max_msg_termination = MaxMessageTermination(max_messages=3)
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=max_msg_termination)# 如果作为独立脚本运行，使用 asyncio.run(...)
await Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))

---------- user ----------
Write a unique, Haiku about the weather in Paris
---------- primary ----------
Gentle rain whispers,  
Cobblestones glisten softly—  
Paris dreams in gray.
[Prompt tokens: 30, Completion tokens: 19]
---------- critic ----------
The Haiku captures the essence of a rainy day in Paris beautifully, and the imagery is vivid. However, it's important to ensure the use of the traditional 5-7-5 syllable structure for Haikus. Your current Haiku lines are composed of 4-7-5 syllables, which slightly deviates from the form. Consider revising the first line to fit the structure.For example:
Soft rain whispers down,  
Cobblestones glisten softly —  
Paris dreams in gray.This revision maintains the essence of your original lines while adhering to the traditional Haiku structure.
[Prompt tokens: 70, Completion tokens: 120]
---------- Summary ----------
Number of messages: 3
Finish reason: Maximum number of messages 3 reached, current message count: 3
Total prompt tokens: 100
Total completion tokens: 139
Duration: 3.34 seconds

TaskResult(messages=[TextMessage(source='user', models_usage=None, content='Write a unique, Haiku about the weather in Paris'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=30, completion_tokens=19), content='Gentle rain whispers,  \nCobblestones glisten softly—  \nParis dreams in gray.'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=70, completion_tokens=120), content="The Haiku captures the essence of a rainy day in Paris beautifully, and the imagery is vivid. However, it's important to ensure the use of the traditional 5-7-5 syllable structure for Haikus. Your current Haiku lines are composed of 4-7-5 syllables, which slightly deviates from the form. Consider revising the first line to fit the structure.\n\nFor example:\nSoft rain whispers down,  \nCobblestones glisten softly —  \nParis dreams in gray.\n\nThis revision maintains the essence of your original lines while adhering to the traditional Haiku structure.")], stop_reason='Maximum number of messages 3 reached, current message count: 3')

• 用户发的消息也算在消息总数内
• 代理的每次发言算一条消息
• 所以这一轮总共是 3 条消息，达到了 MaxMessageTermination(max_messages=3)，团队停止运行。

对话在达到最大消息数后停止。由于主代理尚未回应反馈，我们可以继续对话。

# 继续运行团队
await Console(round_robin_team.run_stream())

---------- primary ----------
Thank you for your feedback. Here is the revised Haiku:Soft rain whispers down,  
Cobblestones glisten softly —  
Paris dreams in gray.
[Prompt tokens: 181, Completion tokens: 32]
---------- critic ----------
The revised Haiku now follows the traditional 5-7-5 syllable pattern, and it still beautifully captures the atmospheric mood of Paris in the rain. The imagery and flow are both clear and evocative. Well done on making the adjustment! APPROVE
[Prompt tokens: 234, Completion tokens: 54]
---------- primary ----------
Thank you for your kind words and approval. I'm glad the revision meets your expectations and captures the essence of Paris. If you have any more requests or need further assistance, feel free to ask!
[Prompt tokens: 279, Completion tokens: 39]
---------- Summary ----------
Number of messages: 3
Finish reason: Maximum number of messages 3 reached, current message count: 3
Total prompt tokens: 694
Total completion tokens: 125
Duration: 6.43 seconds

TaskResult(messages=[TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=181, completion_tokens=32), content='Thank you for your feedback. Here is the revised Haiku:\n\nSoft rain whispers down,  \nCobblestones glisten softly —  \nParis dreams in gray.'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=234, completion_tokens=54), content='The revised Haiku now follows the traditional 5-7-5 syllable pattern, and it still beautifully captures the atmospheric mood of Paris in the rain. The imagery and flow are both clear and evocative. Well done on making the adjustment! \n\nAPPROVE'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=279, completion_tokens=39), content="Thank you for your kind words and approval. I'm glad the revision meets your expectations and captures the essence of Paris. If you have any more requests or need further assistance, feel free to ask!")], stop_reason='Maximum number of messages 3 reached, current message count: 3')

团队从上次中断的位置继续运行，主代理可以对反馈做出回应。

• 使用 MaxMessageTermination 可以在达到指定消息数后自动停止团队运行。
• 终止条件在每次 run 或 run_stream 结束后自动重置。
• 通过继续运行团队，可以让主代理接着回应评论代理的反馈，从而保持对话连续性。

组合终止条件（Combining Termination Conditions）

我们来看如何使用 AND（&） 和 OR（|） 运算符组合终止条件，以创建更复杂的停止逻辑。

例如，我们创建一个团队，当满足以下任意条件时就停止运行：

1. 生成了 10 条消息
2. 评论代理（Critic Agent）批准了一条消息

max_msg_termination = MaxMessageTermination(max_messages=10)
text_termination = TextMentionTermination("APPROVE")
combined_termination = max_msg_termination | text_terminationround_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=combined_termination)# 如果作为独立脚本运行，使用 asyncio.run(...)
await Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))

---------- user ----------
Write a unique, Haiku about the weather in Paris
---------- primary ----------
Spring breeze gently hums,  
Cherry blossoms in full bloom—  
Paris wakes to life.
[Prompt tokens: 467, Completion tokens: 19]
---------- critic ----------
The Haiku beautifully captures the awakening of Paris in the spring. The imagery of a gentle spring breeze and cherry blossoms in full bloom effectively conveys the rejuvenating feel of the season. The final line, "Paris wakes to life," encapsulates the renewed energy and vibrancy of the city. The Haiku adheres to the 5-7-5 syllable structure and portrays a vivid seasonal transformation in a concise and poetic manner. Excellent work!APPROVE
[Prompt tokens: 746, Completion tokens: 93]
---------- Summary ----------
Number of messages: 3
Finish reason: Text 'APPROVE' mentioned
Total prompt tokens: 1213
Total completion tokens: 112
Duration: 2.75 seconds

TaskResult(messages=[TextMessage(source='user', models_usage=None, content='Write a unique, Haiku about the weather in Paris'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=467, completion_tokens=19), content='Spring breeze gently hums,  \nCherry blossoms in full bloom—  \nParis wakes to life.'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=746, completion_tokens=93), content='The Haiku beautifully captures the awakening of Paris in the spring. The imagery of a gentle spring breeze and cherry blossoms in full bloom effectively conveys the rejuvenating feel of the season. The final line, "Paris wakes to life," encapsulates the renewed energy and vibrancy of the city. The Haiku adheres to the 5-7-5 syllable structure and portrays a vivid seasonal transformation in a concise and poetic manner. Excellent work!\n\nAPPROVE')], stop_reason="Text 'APPROVE' mentioned")

对话在评论代理批准消息后停止，尽管如果消息数达到 10 条，也会停止。

或者，如果我们希望 仅在两个条件都满足时 才停止运行，可以使用 AND（&） 运算符：

combined_termination = max_msg_termination & text_termination

自定义终止条件（Custom Termination Condition）

内置的终止条件已经能满足大多数使用场景。但有些情况下，你可能需要实现一个自定义的终止条件，这种条件不适用于现有的终止逻辑。你可以通过继承 TerminationCondition 类来实现。

在这个示例中，我们创建了一个自定义终止条件，当特定函数被调用时就停止对话。

from typing import Sequence
from autogen_agentchat.base import TerminatedException, TerminationCondition
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage, StopMessage, ToolCallExecutionEvent
from autogen_core import Component
from pydantic import BaseModel
from typing_extensions import Selfclass FunctionCallTerminationConfig(BaseModel):"""用于序列化和反序列化组件的终止条件配置。"""function_name: strclass FunctionCallTermination(TerminationCondition, Component[FunctionCallTerminationConfig]):"""当接收到特定名称的 FunctionExecutionResult 时，终止对话。"""component_config_schema = FunctionCallTerminationConfigcomponent_provider_override = "autogen_agentchat.conditions.FunctionCallTermination"def __init__(self, function_name: str) -> None:self._terminated = Falseself._function_name = function_name@propertydef terminated(self) -> bool:return self._terminatedasync def __call__(self, messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> StopMessage | None:if self._terminated:raise TerminatedException("Termination condition has already been reached")for message in messages:if isinstance(message, ToolCallExecutionEvent):for execution in message.content:if execution.name == self._function_name:self._terminated = Truereturn StopMessage(content=f"Function '{self._function_name}' was executed.",source="FunctionCallTermination",)return Noneasync def reset(self) -> None:self._terminated = Falsedef _to_config(self) -> FunctionCallTerminationConfig:return FunctionCallTerminationConfig(function_name=self._function_name)@classmethoddef _from_config(cls, config: FunctionCallTerminationConfig) -> Self:return cls(function_name=config.function_name)

接下来，我们使用这个新的终止条件，当批评代理（critic agent）通过 approve 函数批准消息时停止对话。

首先，定义一个简单的函数，当批评代理批准消息时调用：

def approve() -> None:"""当所有反馈都已处理时批准消息。"""pass

然后创建代理。批评代理配备了 approve 工具：

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClientmodel_client = OpenAIChatCompletionClient(model="gpt-4o",temperature=1,
)primary_agent = AssistantAgent("primary",model_client=model_client,system_message="You are a helpful AI assistant.",
)critic_agent = AssistantAgent("critic",model_client=model_client,tools=[approve],  # 注册 approve 函数为工具system_message="Provide constructive feedback. Use the approve tool to approve when all feedbacks are addressed.",
)

然后创建终止条件和团队，并执行写诗任务：

function_call_termination = FunctionCallTermination(function_name="approve")
round_robin_team = RoundRobinGroupChat([primary_agent, critic_agent], termination_condition=function_call_termination)await Console(round_robin_team.run_stream(task="Write a unique, Haiku about the weather in Paris"))
await model_client.close()

在实际运行中，你会看到，当批评代理使用 approve 函数批准消息时，对话就停止了。

例如输出：

---------- user ----------
Write a unique, Haiku about the weather in Paris
---------- primary ----------
Raindrops gently fall,  
Cobblestones shine in dim light—  
Paris dreams in grey.  
---------- critic ----------
This Haiku beautifully captures a melancholic yet romantic image of Paris in the rain. The use of sensory imagery like "Raindrops gently fall" and "Cobblestones shine" effectively paints a vivid picture. It could be interesting to experiment with more distinct seasonal elements of Paris, such as incorporating the Seine River or iconic landmarks in the context of the weather. Overall, it successfully conveys the atmosphere of Paris in subtle, poetic imagery.
---------- primary ----------
Thank you for your feedback! I’m glad you enjoyed the imagery. Here’s another Haiku that incorporates iconic Parisian elements:Eiffel stands in mist,  
Seine's ripple mirrors the sky—  
Spring whispers anew.  
---------- critic ----------
[FunctionCall(id='call_QEWJZ873EG4UIEpsQHi1HsAu', arguments='{}', name='approve')]
---------- critic ----------
[FunctionExecutionResult(content='None', name='approve', call_id='call_QEWJZ873EG4UIEpsQHi1HsAu', is_error=False)]
---------- critic ----------
None

TaskResult(messages=[TextMessage(source='user', models_usage=None, metadata={}, content='Write a unique, Haiku about the weather in Paris', type='TextMessage'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=30, completion_tokens=23), metadata={}, content='Raindrops gently fall,  \nCobblestones shine in dim light—  \nParis dreams in grey.  ', type='TextMessage'), TextMessage(source='critic', models_usage=RequestUsage(prompt_tokens=99, completion_tokens=90), metadata={}, content='This Haiku beautifully captures a melancholic yet romantic image of Paris in the rain. The use of sensory imagery like "Raindrops gently fall" and "Cobblestones shine" effectively paints a vivid picture. It could be interesting to experiment with more distinct seasonal elements of Paris, such as incorporating the Seine River or iconic landmarks in the context of the weather. Overall, it successfully conveys the atmosphere of Paris in subtle, poetic imagery.', type='TextMessage'), TextMessage(source='primary', models_usage=RequestUsage(prompt_tokens=152, completion_tokens=48), metadata={}, content="Thank you for your feedback! I’m glad you enjoyed the imagery. Here’s another Haiku that incorporates iconic Parisian elements:\n\nEiffel stands in mist,  \nSeine's ripple mirrors the sky—  \nSpring whispers anew.  ", type='TextMessage'), ToolCallRequestEvent(source='critic', models_usage=RequestUsage(prompt_tokens=246, completion_tokens=11), metadata={}, content=[FunctionCall(id='call_QEWJZ873EG4UIEpsQHi1HsAu', arguments='{}', name='approve')], type='ToolCallRequestEvent'), ToolCallExecutionEvent(source='critic', models_usage=None, metadata={}, content=[FunctionExecutionResult(content='None', name='approve', call_id='call_QEWJZ873EG4UIEpsQHi1HsAu', is_error=False)], type='ToolCallExecutionEvent'), ToolCallSummaryMessage(source='critic', models_usage=None, metadata={}, content='None', type='ToolCallSummaryMessage')], stop_reason="Function 'approve' was executed.")

可以看到，对话在批评代理调用 approve 函数后停止。

管理状态

到目前为止，我们已经讨论了如何在多代理应用中构建组件——包括代理、团队和终止条件。

在许多情况下，将这些组件的状态保存到磁盘并在之后加载回来是非常有用的。这在 Web 应用中尤其有用，因为无状态的端点在响应请求时，需要从持久化存储中加载应用的状态。

保存与加载代理

我们可以通过调用 AssistantAgent 的 save_state() 方法来获取代理的状态。

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClientmodel_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06")assistant_agent = AssistantAgent(name="assistant_agent",system_message="You are a helpful assistant",model_client=model_client,
)# 在脚本中运行时使用 asyncio.run(...)
response = await assistant_agent.on_messages([TextMessage(content="Write a 3 line poem on lake tangayika", source="user")],CancellationToken()
)
print(response.chat_message)

输出示例：

In Tanganyika's embrace so wide and deep,  
Ancient waters cradle secrets they keep,  
Echoes of time where horizons sleep.  

保存代理状态：

agent_state = await assistant_agent.save_state()
print(agent_state)

输出示例：

{'type': 'AssistantAgentState','version': '1.0.0','llm_messages': [{'content': 'Write a 3 line poem on lake tangayika', 'source': 'user', 'type': 'UserMessage'},{'content': "In Tanganyika's embrace so wide and deep,  \nAncient waters cradle secrets they keep,  \nEchoes of time where horizons sleep.", 'source': 'assistant_agent', 'type': 'AssistantMessage'}]
}

加载已保存的状态到新的代理实例：

model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06")new_assistant_agent = AssistantAgent(name="assistant_agent",system_message="You are a helpful assistant",model_client=model_client,
)
await new_assistant_agent.load_state(agent_state)response = await new_assistant_agent.on_messages([TextMessage(content="What was the last line of the previous poem you wrote", source="user")],CancellationToken()
)
print(response.chat_message)
await model_client.close()

输出示例：

The last line of the poem was: "Echoes of time where horizons sleep."

注意

对于 AssistantAgent 来说，其状态由 model_context 构成。如果你编写自己的自定义代理，建议重写 save_state() 和 load_state() 方法以自定义保存和加载行为。默认实现会保存和加载一个空状态。

保存与加载团队

我们可以通过调用团队的 save_state 方法来获取团队状态，并通过 load_state 方法将其加载回来。

当调用团队的 save_state 时，会保存团队中所有代理的状态。

下面我们创建一个简单的 RoundRobinGroupChat 团队，包含一个代理，并让它写一首三行诗：

model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06")# 定义团队
assistant_agent = AssistantAgent(name="assistant_agent",system_message="You are a helpful assistant",model_client=model_client,
)
agent_team = RoundRobinGroupChat([assistant_agent], termination_condition=MaxMessageTermination(max_messages=2))# 运行团队并将消息流输出到控制台
stream = agent_team.run_stream(task="Write a beautiful poem 3-line about lake tangayika")
await Console(stream)# 保存团队状态
team_state = await agent_team.save_state()

示例输出：

---------- user ----------
Write a beautiful poem 3-line about lake tangayika
---------- assistant_agent ----------
In Tanganyika's gleam, beneath the azure skies,  
Whispers of ancient waters, in tranquil guise,  
Nature's mirror, where dreams and serenity lie.
[Prompt tokens: 29, Completion tokens: 34]
---------- Summary ----------
Number of messages: 2
Finish reason: Maximum number of messages 2 reached, current message count: 2
Total prompt tokens: 29
Total completion tokens: 34
Duration: 0.71 seconds

如果我们重置团队（模拟重新实例化团队），再问 “What was the last line of the poem you wrote?”，团队将无法回答，因为没有前一次运行的上下文：

await agent_team.reset()
stream = agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)

输出示例：

---------- user ----------
What was the last line of the poem you wrote?
---------- assistant_agent ----------
I'm sorry, but I am unable to recall or access previous interactions, including any specific poem I may have composed in our past conversations. If you like, I can write a new poem for you.
[Prompt tokens: 28, Completion tokens: 40]
---------- Summary ----------
Number of messages: 2
Finish reason: Maximum number of messages 2 reached, current message count: 2
Total prompt tokens: 28
Total completion tokens: 40
Duration: 0.70 seconds

接下来，我们加载团队的状态并再次提问。此时团队能够准确返回上一次写的诗的最后一行：

print(team_state)# 加载团队状态
await agent_team.load_state(team_state)
stream = agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)

输出示例：

The last line of the poem I wrote is:  
"Nature's mirror, where dreams and serenity lie."

可以看到，通过保存和加载团队状态，团队能够记住之前的对话内容和生成的消息，从而保持上下文连续性。

持久化状态（文件或数据库）

在很多情况下，我们希望将团队的状态保存到磁盘（或数据库），并在之后加载回来。状态本质上是一个字典，可以序列化到文件或者写入数据库。

import json# 保存状态到磁盘
with open("coding/team_state.json", "w") as f:json.dump(team_state, f)# 从磁盘加载状态
with open("coding/team_state.json", "r") as f:team_state = json.load(f)# 创建新团队并加载状态
new_agent_team = RoundRobinGroupChat([assistant_agent], termination_condition=MaxMessageTermination(max_messages=2))
await new_agent_team.load_state(team_state)
stream = new_agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)
await model_client.close()

运行结果示例：

The last line of the poem I wrote is:  
"Nature's mirror, where dreams and serenity lie."

通过这种方式，即使重新启动应用程序或在不同实例中运行团队，也可以保持之前的上下文和状态，实现状态持久化。

结尾

综上来看，AgentChat 与 LangGraph 的 create_react_agent() 在定位上确实更接近：它们都提供了一个“快速创建可用 Agent”的能力，让开发者能以较低门槛启动实验和应用。区别在于，LangGraph 更强调“流程的可视化与编排”，而 AutoGen 的 AgentChat 则更关注“消息驱动的多 Agent 协作”。两者并不是谁替代谁，而是 两种并行的思路：

• 如果你想要的是“清晰的流程控制、可视化的逻辑编排”，LangGraph 更合适；
• 如果你更看重“快速上手、多角色对话、灵活的人机协作”，AutoGen AgentChat 会更自然。

最终，选择哪个框架，取决于你希望用 “流程” 来组织智能体，还是用 “对话” 来驱动智能体 —— 这正是 AutoGen 与 LangGraph 在同一层级上的不同切入点。