AutoGen-AgentChat-1-整体了解
import os
from dotenv import load_dotenv
load_dotenv()
# print(os.environ.get("OPENAI_API_KEY"))
1、整体示例
from autogen_agentchat.agents import AssistantAgent # 导入助手代理类
from autogen_agentchat.ui import Console # 导入用于控制台交互的 UI
from autogen_ext.models.openai import OpenAIChatCompletionClient # 导入 OpenAI 模型客户端# 定义一个模型客户端。你也可以使用其他实现了 `ChatCompletionClient` 接口的模型。
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini", # 使用 GPT-4o 模型# api_key="YOUR_API_KEY", # 如果需要,可以在此处填入你的 OpenAI API 密钥
)# 定义一个异步工具函数,该函数可以被智能体调用。
# 在这个示例中,我们创建一个模拟的天气查询工具。
async def get_weather(city: str) -> str:"""获取指定城市的天气信息。参数:city (str): 城市名称。返回:str: 天气描述(这里是伪造的示例数据)。"""return f"{city} 当前的天气是 73 华氏度,晴朗。"# 创建一个 AssistantAgent(助手代理),并为其配置模型客户端、工具函数、系统提示等参数。
agent = AssistantAgent(name="weather_agent", # 代理名称model_client=model_client, # 使用上面定义的模型客户端tools=[get_weather], # 代理可使用的工具函数列表system_message="你是一个乐于助人的助手。", # 系统提示,用自然语言定义代理角色reflect_on_tool_use=True, # 启用工具使用后的反思能力model_client_stream=True, # 启用模型输出的流式传输(逐词显示)
)# 定义主运行函数,用于执行任务并将结果流式输出到控制台。
async def main() -> None:# 使用 Console 界面运行任务并实时显示模型生成的内容。await Console(agent.run_stream(task="北京的天气如何?"))# 任务完成后关闭模型客户端的连接。await model_client.close()
# 如果你在脚本中运行这个函数,请使用 asyncio.run(main())。
# 在支持顶层 await 的环境(如 Jupyter Notebook)中,可以直接使用:
await main()
---------- TextMessage (user) ----------
北京的天气如何?
---------- ToolCallRequestEvent (weather_agent) ----------
[FunctionCall(id='call_w9SI4bFM3irKI5MnXT2btwiX', arguments='{"city":"北京"}', name='get_weather')]
---------- ToolCallExecutionEvent (weather_agent) ----------
[FunctionExecutionResult(content='北京 当前的天气是 73 华氏度,晴朗。', name='get_weather', call_id='call_w9SI4bFM3irKI5MnXT2btwiX', is_error=False)]
---------- ModelClientStreamingChunkEvent (weather_agent) ----------
北京当前的天气是73华氏度(约23摄氏度),
2、模型构建
from autogen_ext.models.openai import OpenAIChatCompletionClientopenai_model_client = OpenAIChatCompletionClient(model="gpt-4o-mini",# api_key="sk-...", # Optional if you have an OPENAI_API_KEY environment variable set.
)
# 要测试模型客户端,可以使用以下代码:from autogen_core.models import UserMessageresult = await openai_model_client.create([UserMessage(content="中国的首都是哪里?", source="user")])
print(result)
await openai_model_client.close()
finish_reason='stop' content='中国的首都是北京。' usage=RequestUsage(prompt_tokens=14, completion_tokens=6) cached=False logprobs=None thought=None
3、构建消息
from autogen_agentchat.messages import TextMessage# 创建一个文本消息
text_message = TextMessage(content="你好,世界!", source="User")
print(f"消息内容: {text_message.content}")
print(f"消息来源: {text_message.source}")
print(f"消息类型: {type(text_message)}")
消息内容: 你好,世界!
消息来源: User
消息类型: <class 'autogen_agentchat.messages.TextMessage'>
# 测试消息的其他属性
print(f"消息字典形式: {text_message.model_dump()}")
消息字典形式: {'source': 'User', 'models_usage': None, 'metadata': {}, 'created_at': datetime.datetime(2025, 6, 30, 8, 30, 17, 310021, tzinfo=datetime.timezone.utc), 'content': '你好,世界!', 'type': 'TextMessage'}
4、构建代理
# 定义一个用于在网络上搜索信息的工具函数
# 为了简单起见,这里我们使用一个模拟函数,返回静态字符串
async def web_search(query: str) -> str:"""在网络上查找信息"""return "AutoGen 是一个用于构建多智能体应用的编程框架。"# 创建一个使用 OpenAI GPT-4o 模型的代理客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini", # 指定所使用的模型为 gpt-4o-mini 版本# api_key="YOUR_API_KEY", # 如果使用实际 API,需要在此处填写 OpenAI API 密钥
)# 创建一个助手智能体(Agent),用于执行任务
agent = AssistantAgent(name="assistant", # 设置智能体的名称为“assistant”model_client=model_client, # 指定使用的模型客户端tools=[web_search], # 向智能体注册工具,这里是网络搜索函数 web_searchsystem_message="使用工具来完成任务。", # 设置系统提示语,引导智能体优先使用工具
)
# 查看效果
# 如果在脚本中运行,请使用 asyncio.run(agent.run(...)) 的方式调用
# 这里使用 await 是因为当前代码假设运行在异步环境(如 Jupyter Notebook 或异步函数中)# 调用智能体执行任务,这里任务是“查找关于 AutoGen 的信息”# 1、使用消息方式
result = await agent.run(task=text_message)# 2、直接提问
# result = await agent.run(task="查找关于 AutoGen 的信息")# 打印智能体在执行过程中记录的所有消息(包括请求、回复、工具调用等)
print(result.messages)
[TextMessage(source='User', models_usage=None, metadata={}, created_at=datetime.datetime(2025, 6, 30, 8, 30, 17, 310021, tzinfo=datetime.timezone.utc), content='你好,世界!', type='TextMessage'), TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=62, completion_tokens=11), metadata={}, created_at=datetime.datetime(2025, 6, 30, 8, 30, 28, 818922, tzinfo=datetime.timezone.utc), content='你好! 有什么我可以帮助你的吗?', type='TextMessage')]
流式输出
# 定义一个异步函数,用于以流式方式运行智能体任务并输出消息
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="查找关于 AutoGen 的信息"), # 以流式方式运行任务output_stats=True, # 启用统计信息输出,例如时间、消息数量等)
# 如果在脚本中运行此函数,请使用 asyncio.run(assistant_run_stream()) 的方式
# 这里假设处于异步上下文中,因此使用 await
await assistant_run_stream()
---------- TextMessage (user) ----------
查找关于 AutoGen 的信息
---------- ToolCallRequestEvent (assistant) ----------
[FunctionCall(id='call_UWFTuAuxKLRbbkjB1wWDZvZR', arguments='{"query":"AutoGen"}', name='web_search')]
[Prompt tokens: 88, Completion tokens: 15]
---------- ToolCallExecutionEvent (assistant) ----------
[FunctionExecutionResult(content='AutoGen 是一个用于构建多智能体应用的编程框架。', name='web_search', call_id='call_UWFTuAuxKLRbbkjB1wWDZvZR', is_error=False)]
---------- ToolCallSummaryMessage (assistant) ----------
AutoGen 是一个用于构建多智能体应用的编程框架。
---------- Summary ----------
Number of messages: 4
Finish reason: None
Total prompt tokens: 88
Total completion tokens: 15
Duration: 1.59 seconds
函数工具
该web_search_func工具是功能工具的一个示例。架构是自动生成的。
from autogen_core.tools import FunctionTool# 使用 Python 函数定义一个工具函数,用于模拟网络搜索功能
async def web_search_func(query: str) -> str:"""在网络上查找信息"""return "AutoGen 是一个用于构建多智能体应用的编程框架。"# 这一步在 AssistantAgent 中如果传入的是 Python 函数,会自动完成注册
# 此处我们手动将函数封装为 FunctionTool,便于更灵活使用
web_search_function_tool = FunctionTool(web_search_func,description="在网络上查找信息" # 向大模型描述此工具的用途,模型会据此判断是否调用该工具
)
# 工具的 schema(规范/结构)会在 AssistantAgent 调用 on_messages 时提供给模型,用于理解工具参数等信息
web_search_function_tool.schema # 查看生成的工具 schema(包含输入参数、描述、调用格式等信息)
{'name': 'web_search_func','description': '在网络上查找信息','parameters': {'type': 'object','properties': {'query': {'description': 'query','title': 'Query','type': 'string'}},'required': ['query'],'additionalProperties': False},'strict': False}
使用工具和工作台
大型语言模型 (LLM) 通常仅限于生成文本或代码响应。然而,许多复杂任务受益于使用执行特定操作的外部工具的能力,例如从 API 或数据库获取数据。
为了解决这一限制,现代 LLM 现在可以接受可用工具模式列表(工具及其参数的描述),并生成工具调用消息。此功能称为工具调用或函数调用,并且正在成为构建基于智能代理的应用程序的流行模式。有关 LLM 中工具调用的更多信息,请参阅OpenAI 和Anthropic的文档。
在 AgentChat 中,AssistantAgent可以使用工具执行特定操作。该web_search工具就是这样一种工具,它允许助理代理在网络上搜索信息。单个自定义工具可以是 Python 函数或 的子类BaseTool。
另一方面,aWorkbench是共享状态和资源的工具集合。
内置工具和工作台
AutoGen 扩展程序提供了一组可与助手一起使用的内置工具。请前往API 文档,查看命名空间下所有可用的工具autogen_ext.tools。例如,您可以找到以下工具:
- graphrag:使用 GraphRAG 索引的工具。
- http:用于发出 HTTP 请求的工具。
- langchain:使用 LangChain 工具的适配器。
- mcp:使用模型聊天协议 (MCP) 服务器的工具和工作台。
模型上下文协议 (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 的抓取工具参数(通过 uvx 启动服务)
fetch_mcp_server = StdioServerParams(command="uvx", args=["mcp-server-fetch"])# 创建一个 MCP 工作台,它提供与 MCP 服务交互的会话上下文
async with McpWorkbench(fetch_mcp_server) as workbench: # type: ignore# 创建一个可以使用抓取工具的智能体model_client = OpenAIChatCompletionClient(model="gpt-4o-mini") # 指定使用 GPT-4.1 nano 模型fetch_agent = AssistantAgent(name="fetcher", # 智能体名称model_client=model_client, # 使用的模型客户端workbench=workbench, # 指定 MCP 工作台,用于支持工具调用reflect_on_tool_use=True # 启用工具使用结果反思能力,可增强工具调用后的输出分析)# 让智能体抓取指定网页内容并进行总结result = await fetch_agent.run(task="总结 https://en.wikipedia.org/wiki/Seattle 的内容")# 确保返回的最后一条消息是文本类型(即总结结果)assert isinstance(result.messages[-1], TextMessage)# 打印智能体输出的最后一条消息内容(即总结文本)print(result.messages[-1].content)# 关闭与模型客户端的连接,释放资源await model_client.close()
我无法直接访问外部链接。您可以告诉我您想要了解的关于西雅图的具体信息,或者我可以提供一些西雅图的基本概况。西雅图是美国华盛顿州的一座城市,以其美丽的自然风光、咖啡文化和科技产业而闻名。西雅图是太平洋西北地区的重要城市,坐落在普吉特湾和喀尔文山之间。它也是许多知名公司的总部所在地,如微软和亚马逊。城市的文化丰富,拥有多样的音乐、艺术和美食场所,著名的景点包括太空针塔(Space Needle)、西雅图公共市场(Pike Place Market)等。西雅图的气候以湿润和温和著称,常年降雨较多。如果您需要更多的特定信息,请告诉我!
# asyncio.run(run())
并行工具调用
某些模型支持并行工具调用,这对于需要同时调用多个工具的任务非常有用。默认情况下,如果模型客户端生成多个工具调用,AssistantAgent 则会并行调用这些工具。
当工具具有可能相互干扰的副作用,或者代理行为需要在不同的模型之间保持一致时,您可能需要禁用并行工具调用。这应该在模型客户端级别完成。
对于OpenAIChatCompletionClient和AzureOpenAIChatCompletionClient,设置parallel_tool_calls=False为禁用并行工具调用。
# 创建一个 OpenAI 模型客户端,禁用工具并行调用功能
model_client_no_parallel_tool_call = OpenAIChatCompletionClient(model="gpt-4o-mini", # 使用 OpenAI 最新的 gpt-4o-mini 模型parallel_tool_calls=False, # 禁用并行工具调用(默认模型可能支持同时调用多个工具,此处强制关闭) # type: ignore 为类型检查器忽略此参数类型问题
)# 创建一个智能体,并绑定上面禁用并行工具调用的模型客户端
agent_no_parallel_tool_call = AssistantAgent(name="assistant", # 设置智能体名称为 assistantmodel_client=model_client_no_parallel_tool_call, # 使用上方定义的模型客户端tools=[web_search], # 为智能体注册工具列表,这里包含一个自定义的 web_search 工具system_message="使用工具来解决任务。", # 系统提示,告诉模型遇到任务时优先使用工具
)
结构化输出
结构化输出允许模型返回具有应用程序提供的预定义模式的结构化 JSON 文本。与 JSON 模式不同,该模式可以作为Pydantic BaseModel 类提供,该类也可用于验证输出。
output_content_type一旦您在构造函数的参数中指定了基础模型类AssistantAgent,代理就会响应一个StructuredMessage 其content类型为基础模型类类型的 。
这样,您可以将代理的响应直接集成到您的应用程序中,并将模型的输出用作结构化对象。
设置后output_content_type,默认情况下,代理会反映工具的使用情况,并根据工具调用结果返回结构化的输出消息。您可以通过reflect_on_tool_use=False明确设置来禁用此行为。
结构化输出也有助于在代理的响应中融入思维链推理。请参阅下方示例,了解如何将结构化输出与助理代理配合使用。
from typing import Literal
from pydantic import BaseModel
from autogen_agentchat.messages import StructuredMessage# 定义代理智能体输出结果的结构类型,继承自 Pydantic 的 BaseModel
# 其中包含两项字段:思考过程 thoughts(字符串)和分类结果 response(三选一)
class AgentResponse(BaseModel):thoughts: str # 智能体的思考过程response: Literal["快乐", "悲伤", "中性"] # 限定分类输出只能是 happy/sad/neutral(快乐/悲伤/中性)之一# 创建一个使用 OpenAI GPT-4o 模型的模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")# 创建智能体实例,指定输出类型和系统指令
agent = AssistantAgent("assistant", # 智能体名称model_client=model_client, # 指定所使用的语言模型system_message="将输入分类为 happy、sad 或 neutral,并以 JSON 格式返回。", # 提示语告诉模型必须输出结构化 JSONoutput_content_type=AgentResponse, # 指定输出内容的结构,模型必须严格返回 AgentResponse 格式
)# 使用 Console 流式运行任务,同时显示实时输出
# 任务内容是 “我很高兴.”
result = await Console(agent.run_stream(task="我很高兴."))# 验证返回结果的最后一条消息是结构化消息,并确保其内容符合 AgentResponse 类型
assert isinstance(result.messages[-1], StructuredMessage)
assert isinstance(result.messages[-1].content, AgentResponse)# 输出智能体的思考过程和最终分类结果
print("思考过程(thoughts): ", result.messages[-1].content.thoughts)
print("分类结果(response): ", result.messages[-1].content.response)# 关闭模型客户端,释放资源
await model_client.close()
---------- TextMessage (user) ----------
我很高兴.
---------- StructuredMessage[AgentResponse] (assistant) ----------
{"thoughts":"用户表达了他们的情绪,明确表示感到高兴。这是一个积极的表述,因此可以将其归类为'快乐'。","response":"快乐"}
思考过程(thoughts): 用户表达了他们的情绪,明确表示感到高兴。这是一个积极的表述,因此可以将其归类为'快乐'。
分类结果(response): 快乐
流式响应
您可以通过设置来流式传输模型客户端生成的令牌model_client_stream=True。这将导致代理ModelClientStreamingChunkEvent在中产生消息run_stream()。
底层模型 API 必须支持流式令牌才能正常工作。请咨询您的模型提供商,确认其是否支持此功能。
# 创建一个 OpenAI GPT-4o 的模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")# 创建一个支持流式输出的智能体(AssistantAgent)
streaming_assistant = AssistantAgent(name="assistant", # 智能体名称model_client=model_client, # 使用 GPT-4o 模型客户端system_message="你是一个乐于助人的助手。", # 系统提示语,用于设定智能体行为风格model_client_stream=True, # 启用模型的流式输出(逐词/逐句输出)
)# 在异步函数中运行此代码,如果是在脚本中请用 asyncio.run() 包裹执行
# 使用 run_stream 方法流式执行任务,任务内容为“列举两个亚洲的城市”
async for message in streaming_assistant.run_stream(task="列举两个亚洲的城市"): # type: ignoreprint(message) # 每接收到一个响应片段(如一段文字、工具调用等)就立即输出
source='user' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 2, 135844, tzinfo=datetime.timezone.utc) content='列举两个亚洲的城市' type='TextMessage'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 2, 989012, tzinfo=datetime.timezone.utc) content='两个' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 2, 990291, tzinfo=datetime.timezone.utc) content='亚洲' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 3248, tzinfo=datetime.timezone.utc) content='的' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 4104, tzinfo=datetime.timezone.utc) content='城市' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 37875, tzinfo=datetime.timezone.utc) content='是' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 38380, tzinfo=datetime.timezone.utc) content=':' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 60267, tzinfo=datetime.timezone.utc) content='东京' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 60864, tzinfo=datetime.timezone.utc) content='(' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 105395, tzinfo=datetime.timezone.utc) content='日本' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 105763, tzinfo=datetime.timezone.utc) content=')' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 129749, tzinfo=datetime.timezone.utc) content='和' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 130288, tzinfo=datetime.timezone.utc) content='北京' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 157990, tzinfo=datetime.timezone.utc) content='(' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 158512, tzinfo=datetime.timezone.utc) content='中国' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 171437, tzinfo=datetime.timezone.utc) content=')。' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0) metadata={} created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 173660, tzinfo=datetime.timezone.utc) content='两个亚洲的城市是:东京(日本)和北京(中国)。' type='TextMessage'
messages=[TextMessage(source='user', models_usage=None, metadata={}, created_at=datetime.datetime(2025, 6, 30, 9, 26, 2, 135844, tzinfo=datetime.timezone.utc), content='列举两个亚洲的城市', type='TextMessage'), TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0), metadata={}, created_at=datetime.datetime(2025, 6, 30, 9, 26, 3, 173660, tzinfo=datetime.timezone.utc), content='两个亚洲的城市是:东京(日本)和北京(中国)。', type='TextMessage')] stop_reason=None
您可以在上面的输出中看到流式数据块。这些数据块由模型客户端生成,并在代理收到后立即生成。最终响应(所有数据块的串联)在最后一个数据块之后立即生成。
使用模型 上下文
AssistantAgent有一个model_context 可用于传入ChatCompletionContext 对象的参数。这允许代理使用不同的模型上下文,例如 BufferedChatCompletionContext限制发送给模型的上下文。
默认情况下,AssistantAgent使用 ,UnboundedChatCompletionContext 将完整的对话历史记录发送到模型。要将上下文限制为最后一条n消息,可以使用BufferedChatCompletionContext。要按标记计数限制上下文,可以使用 TokenLimitedChatCompletionContext。
from autogen_core.model_context import BufferedChatCompletionContext# 创建一个智能体(AssistantAgent),上下文仅使用最近的 5 条消息进行生成
agent = AssistantAgent(name="assistant", # 智能体名称model_client=model_client, # 使用的语言模型客户端(如 GPT-4o)tools=[web_search], # 注册可用工具列表,此处包含自定义的 web_search 工具system_message="使用工具来解决任务。", # 提示语言模型优先考虑工具来完成任务model_context=BufferedChatCompletionContext(buffer_size=5), # 设置上下文为带缓冲区的模式,仅保留最近 5 条消息用于推理(可控制上下文长度,节省 token 消耗)
)
其他预设代理
有以下预设代理可用:
- UserProxyAgent:接受用户输入的代理将其作为响应返回。
- CodeExecutorAgent:可以执行代码的代理。
- OpenAIAssistantAgent:由 OpenAI Assistant 支持的代理,能够使用自定义工具。
- MultimodalWebSurfer:可以搜索网络并访问网页以获取信息的多模式代理。
- FileSurfer:可以搜索和浏览本地文件以获取信息的代理。
- VideoSurfer:可以观看视频来获取信息的代理。