当前位置：首页 > news >正文

AutoGen Agent 使用指南

news 2025/7/31 16:46:30

Agent 核心属性和方法

所有 Agent 都具有以下属性和方法：

# 属性
name: str          # Agent 名称
description: str   # Agent 描述# 方法
async run(task, cancellation_token, output_task_messages) → TaskResult
async run_stream(task, cancellation_token, output_task_messages) → AsyncGenerator

run(): 返回 TaskResult 对象，包含 messages 和 stop_reason
run_stream(): 返回 AsyncGenerator，包含 BaseAgentEvent | BaseChatMessage | TaskResult

示例详解

1. 基础 Assistant Agent (`main`)

功能：创建一个具有工具调用能力的基础助手代理。

示例代码：

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Consoleasync def web_search(query: str) -> str:"""模拟一个网络搜索工具"""return "AutoGen is a programming framework for building agents..."async def main():assistant = AssistantAgent(name="Assistant",model_client=model_client,tools=[web_search],                    # 可用工具列表system_message="Use tools to solve tasks.",# reflect_on_tool_use=True,           # 是否反思工具使用结果)# 流式运行并在控制台显示result = await Console(assistant.run_stream(task="What is AutoGen?"))await model_client.close()

关键参数说明：

reflect_on_tool_use:
- False: 直接输出 ToolCallSummaryMessage
- True: 将工具结果重新发给模型处理，输出 TextMessage（需要两次 API 调用）

使用场景：

需要工具辅助的问答系统
智能助手应用
自动化任务处理

2. 多模态输入 (`multi_modal_message_example`)

功能：处理包含图片、文本等多种模态的输入。

示例代码：

from io import BytesIO
import PIL
from autogen_agentchat.messages import MultiModalMessage
from autogen_core import Image
import requestsasync def multi_modal_message_example():# 获取并处理图片pil_image = PIL.Image.open(BytesIO(requests.get("https://picsum.photos/300/200").content))img = Image(pil_image)# 创建多模态消息message = MultiModalMessage(content=["What is in this image?", img],  # 文本 + 图片source='user')assistant = AssistantAgent(name="Assistant",model_client=model_client,system_message="Answer the question based on the image.",)result = await assistant.run(task=message)print(result.messages[-1].content)

支持的模态类型：

文本 (Text)
图片 (Image)
音频 (Audio) - 需要模型支持
视频 (Video) - 需要模型支持

使用场景：

图像分析和描述
视觉问答系统
多媒体内容处理
OCR 和文档分析

3. 并行工具调用控制

功能：控制 Agent 是否并行调用多个工具。

示例代码：

禁用并行工具调用

async def disable_parallel_tool_calls():model_client_no_parallel = OpenAIChatCompletionClient(model="gpt-4o",parallel_tool_calls=False,  # 禁用并行调用)agent = AssistantAgent(name="assistant",model_client=model_client_no_parallel,tools=[web_search],system_message="Use tools to solve tasks.",)

限制工具调用次数

async def parallel_tool_calls():agent = AssistantAgent(name="assistant",model_client=model_client,tools=[web_search],system_message="Use tools to solve tasks.",max_tool_iterations=10,  # 限制最大工具调用次数)

配置选项：

parallel_tool_calls=False: 禁用并行工具调用，按顺序执行
max_tool_iterations: 限制工具调用的最大次数

使用场景：

需要顺序执行工具的任务
控制 API 调用成本
避免工具调用死循环

4. 结构化输出 (`main_structured_output`)

功能：让 Agent 返回预定义格式的结构化 JSON 响应。

示例代码：

from pydantic import BaseModel
from typing import Literal
from autogen_agentchat.messages import StructuredMessageclass AgentResponse(BaseModel):thoughts: strresponse: Literal["happy", "sad", "neutral"]async def main_structured_output():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()

重要说明：

设置 output_content_type 后，reflect_on_tool_use 默认为 True
可以显式设置 reflect_on_tool_use=False 来覆盖默认行为

5. 流式 Token 传输 (`main_stream`)

功能：实时流式接收模型生成的 token，提供更好的用户体验。

示例代码：

async def main_stream() -> None:streaming_assistant = AssistantAgent(name="assistant",model_client=model_client,system_message="You are a helpful assistant.",model_client_stream=True,  # 启用流式传输)async for message in streaming_assistant.run_stream(task="Name two cities in South America"):print(message)

输出示例：

source='user' content='Name two cities in South America' type='TextMessage'
source='assistant' content='Two' type='ModelClientStreamingChunkEvent'
source='assistant' content=' cities' type='ModelClientStreamingChunkEvent'
source='assistant' content=' in' type='ModelClientStreamingChunkEvent'
source='assistant' content=' South' type='ModelClientStreamingChunkEvent'
source='assistant' content=' America' type='ModelClientStreamingChunkEvent'

消息类型：

TextMessage: 完整的用户输入
ModelClientStreamingChunkEvent: 流式 token 片段

6. 模型上下文管理 (`model_context_sample`)

功能：控制 Agent 使用的对话历史范围，优化性能和成本。

示例代码：

from autogen_core.model_context import BufferedChatCompletionContextasync def model_context_sample() -> None:# 只使用最近 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),)