智能体#
AutoGen AgentChat 提供了一组预设智能体，每个智能体在响应消息时可能有不同的行为方式。 所有智能体共享以下属性和方法：
name: 智能体的唯一名称description: 智能体的文本描述run: 给定字符串任务或消息列表运行智能体的方法，返回一个TaskResult。智能体应保持状态，此方法预期使用新消息调用，而非完整历史记录run_stream: 与run()相同，但返回一个迭代器，生成继承自BaseAgentEvent或BaseChatMessage的消息，最后一项是TaskResult
有关 AgentChat 消息类型的更多信息，请参阅 autogen_agentchat.messages
助手智能体#
AssistantAgent 是一个内置智能体，
它使用语言模型并具备使用工具的能力。
警告
AssistantAgent 是一个"大杂烩"智能体，
主要用于原型设计和教育目的——它非常通用。
请确保阅读文档和实现代码以理解设计选择。
一旦完全理解设计后，您可能需要实现自己的智能体。
参见自定义智能体。
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:
"""Find information on the web"""
return "AutoGen is a programming framework for building multi-agent applications."
# 创建一个使用 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="Use tools to solve tasks.",
)
获取结果#
我们可以使用 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 属性
存储了代理的"思考过程"以及最终响应。
备注
与 v0.2 AgentChat 不同，工具是由同一个代理在
调用 run() 时直接执行的。
默认情况下，代理会返回工具调用的结果作为最终响应。
多模态输入#
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")
img
# 在脚本中使用 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.
流式消息#
我们还可以通过使用 run_stream() 方法
来流式传输代理生成的每条消息，
并使用 Console 将消息
实时打印到控制台。
async def assistant_run_stream() -> None:
# 选项1：从流中读取每条消息（如前一示例所示）。 async for
# message in agent.run_stream(task="查找关于AutoGen的信息"):
# print(message)
# 选项2：使用 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() 方法
返回一个异步生成器，该生成器会依次产生代理生成的每条消息，
最后一项是 TaskResult。
从这些消息中，你可以观察到助手代理使用了 web_search 工具来
收集信息，并根据搜索结果做出响应。
使用工具与工作台#
大型语言模型(LLMs)通常仅限于生成文本或代码响应。 然而，许多复杂任务需要能够使用执行特定操作的外部工具， 例如从API或数据库获取数据。
为了解决这一限制，现代LLMs现在可以接受可用工具模式列表 (工具及其参数的描述)并生成工具调用消息。 这种能力被称为工具调用或函数调用， 正成为构建基于智能代理的应用程序的流行模式。 有关LLM中工具调用的更多信息，请参考OpenAI 和Anthropic的文档。
在AgentChat中，AssistantAgent可以使用工具执行特定操作。
web_search工具就是其中之一，它允许助手代理在网络上搜索信息。
单个自定义工具可以是Python函数或BaseTool的子类。
另一方面，Workbench是共享状态和资源的工具集合。
默认情况下，当AssistantAgent执行工具时，
它会在响应中以ToolCallSummaryMessage形式返回工具输出的字符串。
如果你的工具没有返回自然语言格式良好的字符串，
你可以通过设置AssistantAgent构造函数中的reflect_on_tool_use=True参数，
添加一个反思步骤让模型总结工具的输出。
内置工具与工作台#
AutoGen扩展提供了一组可以与助手代理一起使用的内置工具。
前往API文档查看autogen_ext.tools命名空间下
所有可用工具。例如，你可以找到以下工具：
graphrag: 用于GraphRAG索引的工具http: 用于发起HTTP请求的工具langchain: 用于LangChain工具的适配器mcp: 用于模型聊天协议(MCP)服务器的工具和工作台
函数工具#
AssistantAgent会自动
将Python函数转换为FunctionTool，
该工具可以被代理使用，并自动从函数签名和文档字符串生成工具模式。
web_search_func工具就是一个函数工具的例子。
其模式是自动生成的。
from autogen_core.tools import FunctionTool
# 定义一个使用Python函数的工具
async def web_search_func(query: str) -> str:
"""Find information on the web"""
return "AutoGen is a programming framework for building multi-agent applications."
# 如果工具是Python函数，这一步会在AssistantAgent内部自动执行
web_search_function_tool = FunctionTool(web_search_func, description="Find information on the web")
# 在AssistantAgent的on_messages调用期间，该模式会被提供给模型
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)工作台#
AssistantAgent也可以使用来自
模型上下文协议(MCP)服务器的工具，通过
McpWorkbench()实现
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.
并行工具调用#
部分模型支持并行工具调用，这对于需要同时调用多个工具的任务非常有用。
默认情况下，如果模型客户端生成多个工具调用，AssistantAgent
会并行执行这些工具调用。
当工具之间存在可能相互干扰的副作用时，或者需要确保不同模型间代理行为一致时， 您可能需要禁用并行工具调用。 这应该在模型客户端层面进行设置。
对于OpenAIChatCompletionClient和AzureOpenAIChatCompletionClient，
设置parallel_tool_calls=False即可禁用并行工具调用。
model_client_no_parallel_tool_call = OpenAIChatCompletionClient(
model="gpt-4o",
parallel_tool_calls=False, # type: ignore
)
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.",
)
在循环中运行代理#
AssistantAgent采用分步执行机制：
每次执行一个模型调用，接着执行一个工具调用（或并行工具调用），
然后进行可选的反思步骤。
如需在循环中运行代理（例如持续运行直到不再产生工具调用）， 请参考单代理团队章节。
结构化输出#
结构化输出允许模型根据应用程序预定义的schema返回结构化JSON文本。 与JSON模式不同，该schema可以定义为Pydantic BaseModel 类，同时也可用于输出验证。
当您在AssistantAgent构造函数的output_content_type参数中
指定基模型类后，代理将返回StructuredMessage类型响应，
其content字段的类型即为基模型类的类型。
通过这种方式，您可以直接将代理响应集成到应用程序中， 并将模型输出作为结构化对象使用。
备注
当设置output_content_type时，默认要求代理对工具使用进行反思，
并根据工具调用结果返回结构化输出消息。
您可以通过显式设置reflect_on_tool_use=False来禁用此行为。
结构化输出对于在代理响应中实现思维链(Chain-of-Thought) 推理也非常有用。 下方示例展示了如何在助理代理中使用结构化输出功能。
from typing import Literal
from pydantic import BaseModel
# 代理响应格式的Pydantic基模型定义。
class AgentResponse(BaseModel):
thoughts: str
response: 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."))
# 检查结果中的最后一条消息，验证其类型，并打印思考过程和响应。
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
流式令牌#
通过设置 model_client_stream=True 可以流式传输模型客户端生成的令牌。
这将使智能体在 run_stream() 方法中产生 ModelClientStreamingChunkEvent 消息。
底层模型 API 必须支持流式令牌才能实现此功能。 请咨询您的模型提供商以确认是否支持此功能。
model_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: ignore
print(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=' South' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' America' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' are' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' Buenos' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' Aires' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' in' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' Argentina' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' and' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' São' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} content=' Paulo' 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'
messages=[TextMessage(source='user', models_usage=None, metadata={}, content='Name two cities in South America', type='TextMessage'), TextMessage(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')] stop_reason=None
你可以在上面的输出中看到流式传输的数据块。 这些数据块由模型客户端生成，并在接收时由代理逐个输出。 最终响应（即所有数据块的拼接结果）会在最后一个数据块之后立即输出。
使用模型上下文#
AssistantAgent具有一个model_context
参数，可用于传入ChatCompletionContext
对象。这使得代理能够使用不同的模型上下文，例如
BufferedChatCompletionContext来
限制发送给模型的上下文内容。
默认情况下，AssistantAgent使用
UnboundedChatCompletionContext
它会将完整的对话历史发送给模型。若要限制上下文为最后n条消息，
可以使用BufferedChatCompletionContext。
若要按token数量限制上下文，可以使用
TokenLimitedChatCompletionContext。
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条消息。
)
其他预设代理#
以下是可用的预设代理：
UserProxyAgent：接收用户输入并将其作为响应返回的代理。CodeExecutorAgent：可以执行代码的代理。OpenAIAssistantAgent：由OpenAI Assistant支持的代理，能够使用自定义工具。MultimodalWebSurfer：能够搜索网页并访问网页获取信息的多模态代理。FileSurfer：能够搜索和浏览本地文件获取信息的代理。VideoSurfer：能够观看视频获取信息的代理。
下一步#
在探索了AssistantAgent的用法后，我们现在可以继续下一节学习AgentChat中的团队功能。