autogen_agentchat.agents#
此模块初始化包中提供的各种预定义代理。BaseChatAgent 是 AgentChat 中所有代理的基类。
- class BaseChatAgent(name: str, description: str)[来源]#
基类：
ChatAgent、ABC、ComponentBase[BaseModel]聊天代理的基类。
此抽象类为
ChatAgent提供了一个基本实现。要创建新的聊天代理，请继承此类并实现on_messages()、on_reset()和produced_message_types。如果需要流式传输，请同时实现on_messages_stream()方法。代理被认为是无状态的，并在调用
on_messages()或on_messages_stream()方法之间维护其状态。代理应将其状态存储在代理实例中。代理还应实现on_reset()方法以将代理重置为其初始化状态。注意
调用方在每次调用
on_messages()或on_messages_stream()方法时，只应将新消息传递给代理。不要在每次调用时将整个对话历史记录传递给代理。创建新代理时必须遵循此设计原则。- component_type: ClassVar[ComponentType] = 'agent'#
组件的逻辑类型。
- abstract property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- abstract async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) Response[来源]#
处理传入消息并返回响应。
注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][来源]#
处理传入消息并返回消息流，最后一个项目是响应。
BaseChatAgent中的基本实现只调用on_messages()并生成响应中的消息。注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async run(*, task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None, cancellation_token: CancellationToken | None = None, output_task_messages: bool = True) TaskResult[来源]#
运行带有给定任务的代理并返回结果。
- async run_stream(*, task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None, cancellation_token: CancellationToken | None = None, output_task_messages: bool = True) AsyncGenerator[BaseAgentEvent | BaseChatMessage | TaskResult, None]#
运行带有给定任务的代理，并返回消息流和作为流中最后一个项目的最终任务结果。
- 参数:
task – 要运行的任务。可以是字符串、单个消息或消息序列。
cancellation_token – 用于立即终止任务的取消令牌。
output_task_messages – 是否在输出流中包含任务消息。为了向后兼容，默认为 True。
- abstract async on_reset(cancellation_token: CancellationToken) None[来源]#
将代理重置为其初始化状态。
- async on_pause(cancellation_token: CancellationToken) None[来源]#
当代理在其
on_messages()或on_messages_stream()方法中运行时被暂停时调用。这在BaseChatAgent类中默认为空操作。子类可以覆盖此方法以实现自定义暂停行为。
- async on_resume(cancellation_token: CancellationToken) None[来源]#
当代理在其
on_messages()或on_messages_stream()方法中运行时，从暂停中恢复时调用。这在BaseChatAgent类中默认为空操作。子类可以覆盖此方法以实现自定义恢复行为。
- async close() None[来源]#
释放代理持有的所有资源。这在
BaseChatAgent类中默认为空操作。子类可以覆盖此方法以实现自定义关闭行为。
- class AssistantAgent(name: str, model_client: ChatCompletionClient, *, tools: List[BaseTool[Any, Any] | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, workbench: Workbench | Sequence[Workbench] | None = None, handoffs: List[Handoff | str] | None = None, model_context: ChatCompletionContext | None = None, description: str = '一个提供工具使用辅助的代理。', system_message: str | None = '你是一个乐于助人的人工智能助手。使用你的工具解决任务。当任务完成时，回复 TERMINATE。', model_client_stream: bool = False, reflect_on_tool_use: bool | None = None, max_tool_iterations: int = 1, tool_call_summary_format: str = '{result}', tool_call_summary_formatter: Callable[[FunctionCall, FunctionExecutionResult], str] | None = None, output_content_type: type[BaseModel] | None = None, output_content_type_format: str | None = None, memory: Sequence[Memory] | None = None, metadata: Dict[str, str] | None = None)[来源]#
基类：
BaseChatAgent、Component[AssistantAgentConfig]一个提供工具使用辅助的代理。
on_messages()返回一个Response，其中chat_message是最终的响应消息。on_messages_stream()创建一个异步生成器，该生成器在消息创建时生成内部消息，并在关闭生成器之前将Response对象作为最后一个项目。BaseChatAgent.run()方法返回一个TaskResult，其中包含代理生成的消息。在消息列表messages中，最后一条消息是最终的响应消息。BaseChatAgent.run_stream()方法创建一个异步生成器，该生成器在消息创建时生成内部消息，并在关闭生成器之前将TaskResult对象作为最后一个项目。注意
调用者在每次调用
on_messages()、on_messages_stream()、BaseChatAgent.run()或BaseChatAgent.run_stream()方法时，必须只将新消息传递给代理。代理在这些方法调用之间维护其状态。不要在每次调用时将整个对话历史记录传递给代理。警告
助手代理不是线程安全的，也不是协程安全的。它不应在多个任务或协程之间共享，也不应同时调用其方法。
下图显示了助手代理的工作原理
结构化输出
如果设置了 output_content_type，代理将默认在最终响应中返回
StructuredMessage，而不是TextMessage。注意
目前，设置 output_content_type 会阻止代理调用 load_component 和 dum_component 方法以进行可序列化配置。这将在未来尽快修复。
工具调用行为
如果模型没有返回工具调用，则响应将立即作为
TextMessage或StructuredMessage（使用结构化输出时）在chat_message中返回。这将结束工具调用迭代循环，无论 max_tool_iterations 设置如何。- 当模型返回工具调用时，它们将立即执行。
当 reflect_on_tool_use 为 False 时，工具调用结果作为
ToolCallSummaryMessage在chat_message中返回。您可以使用静态格式字符串 (tool_call_summary_format) 或 可调用对象 (tool_call_summary_formatter) 自定义摘要；可调用对象对每个工具调用评估一次。当 reflect_on_tool_use 为 True 时，将使用工具调用和结果进行另一次模型推断，最终响应将作为
TextMessage或StructuredMessage（使用结构化输出时）在chat_message中返回。当设置 output_content_type 时，reflect_on_tool_use 默认设置为 True。
当未设置 output_content_type 时，reflect_on_tool_use 默认设置为 False。
如果模型返回多个工具调用，它们将同时执行。要禁用并行工具调用，您需要配置模型客户端。例如，将
OpenAIChatCompletionClient和AzureOpenAIChatCompletionClient的 parallel_tool_calls 设置为 False。max_tool_iterations 参数控制代理在单次运行中可以执行多少次顺序工具调用迭代。当设置为 1（默认值）时，代理会执行工具调用一次并返回结果。当设置为更高的值时，如果模型继续请求，代理可以进行额外的模型调用以执行更多的工具调用，从而实现多步基于工具的工作流。当模型返回文本响应（而不是工具调用）或达到最大迭代次数时，代理停止。
提示
默认情况下，当进行工具调用时，工具调用结果将作为响应返回，因此请密切关注工具返回值是如何格式化的——特别是如果另一个代理期望特定的模式。
对于简单的静态模板，请使用 `tool_call_summary_format`。
对于完全程序化控制（例如，“隐藏大型成功负载，显示所有错误详细信息”），请使用 `tool_call_summary_formatter`。
注意：tool_call_summary_formatter 不可序列化，在从 YAML/JSON 配置文件加载或导出代理时将被忽略。
交接行为
如果触发了交接，
HandoffMessage将在chat_message中返回。如果存在工具调用，它们也将在返回交接之前立即执行。
工具调用和结果通过
context传递给目标代理。
注意
如果检测到多个交接，则只执行第一个交接。为避免这种情况，请在模型客户端配置中禁用并行工具调用。
限制发送到模型的上下文大小
您可以通过将 model_context 参数设置为
BufferedChatCompletionContext来限制发送到模型的消息数量。这将限制发送到模型的最新消息数量，当模型对可处理的令牌数量有限制时，这会很有用。另一种选择是使用TokenLimitedChatCompletionContext，它将限制发送到模型的令牌数量。您还可以通过继承ChatCompletionContext来创建自己的模型上下文。流模式
通过设置 model_client_stream=True，助手代理可以在流模式下使用。在此模式下，
on_messages_stream()和BaseChatAgent.run_stream()方法也会在模型客户端生成响应块时生成ModelClientStreamingChunkEvent消息。块消息将不包含在最终响应的内部消息中。- 参数:
name (str) – 代理的名称。
model_client (ChatCompletionClient) – 用于推理的模型客户端。
tools (List[BaseTool[Any, Any] | Callable[..., Any] | Callable[..., Awaitable[Any]]] | None, optional) – 要注册到代理的工具。
workbench (Workbench | Sequence[Workbench] | None, optional) – 用于代理的工作台或工作台列表。设置工作台时不能使用工具，反之亦然。
handoffs (List[HandoffBase | str] | None, optional) – 代理的交接配置，允许其通过响应
HandoffMessage转移给其他代理。仅当团队处于Swarm模式时才执行转移。如果交接是字符串，则应表示目标代理的名称。model_context (ChatCompletionContext | None, optional) – 用于存储和检索
LLMMessage的模型上下文。可以预加载初始消息。当代理重置时，初始消息将被清除。description (str, optional) – 代理的描述。
system_message (str, optional) – 模型的系统消息。如果提供，它将在进行推断时添加到模型上下文中的消息之前。设置为 None 以禁用。
model_client_stream (bool, optional) – 如果为 True，则模型客户端将以流模式使用。
on_messages_stream()和BaseChatAgent.run_stream()方法也会在模型客户端生成响应块时生成ModelClientStreamingChunkEvent消息。默认为 False。reflect_on_tool_use (bool, optional) – 如果为 True，代理将使用工具调用和结果进行另一次模型推断以生成响应。如果为 False，则工具调用结果将作为响应返回。默认情况下，如果设置了 output_content_type，则此值为 True；如果未设置 output_content_type，则此值为 False。
output_content_type (type[BaseModel] | None, optional) –
StructuredMessage响应的输出内容类型作为 Pydantic 模型。这将与模型客户端一起用于生成结构化输出。如果设置了此项，代理将在最终响应中返回StructuredMessage而不是TextMessage，除非 reflect_on_tool_use 为 False 且进行了工具调用。output_content_type_format (str | None, optional) – （实验性）用于
StructuredMessage响应内容的格式字符串。max_tool_iterations (int, optional) – 在模型停止进行工具调用之前执行的最大工具迭代次数。默认为 1，这意味着代理只会执行模型进行的一次工具调用，并将结果作为
ToolCallSummaryMessage，或作为最终响应在chat_message中以TextMessage或StructuredMessage（使用结构化输出时）返回。一旦模型停止进行工具调用，代理将停止执行工具调用并返回结果作为最终响应。值必须大于或等于 1。tool_call_summary_format (str, optional) – 应用于每个工具调用结果的静态格式字符串，用于构成
ToolCallSummaryMessage。默认为"{result}"。如果提供了 tool_call_summary_formatter，则忽略此项。当 reflect_on_tool_use 为False时，所有工具调用的摘要将用换行符（'n'）连接并作为响应返回。模板中可用的占位符有：{tool_name}、{arguments}、{result}、{is_error}。tool_call_summary_formatter (Callable[[FunctionCall, FunctionExecutionResult], str] | None, optional) –
一个可调用对象，接收
FunctionCall及其FunctionExecutionResult，并返回摘要字符串。如果提供，将覆盖 tool_call_summary_format，并允许条件逻辑——例如，在成功时发出静态字符串，如"Tool FooBar executed successfully."，并且只在失败时显示完整负载（包括所有传递的参数等）。限制：可调用对象 不可序列化；通过 YAML/JSON 配置提供的值将被忽略。
注意
tool_call_summary_formatter 仅用于代码内部使用。目前无法通过配置文件保存或恢复。
memory (Sequence[Memory] | None, optional): 用于代理的内存存储。默认为 None。metadata (Dict[str, str] | None, optional): 用于跟踪的可选元数据。
- 抛出:
ValueError – 如果工具名称不唯一。
ValueError – 如果交接名称不唯一。
ValueError – 如果交接名称与工具名称不唯一。
ValueError – 如果最大工具迭代次数小于 1。
示例
示例 1：基本代理
以下示例演示了如何创建一个带有模型客户端的助手代理，并生成对简单任务的响应。
import asyncio from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent async def main() -> None: model_client = OpenAIChatCompletionClient( model="gpt-4o", # api_key = "your_openai_api_key" ) agent = AssistantAgent(name="assistant", model_client=model_client) result = await agent.run(task="Name two cities in North America.") print(result) asyncio.run(main())
示例 2：模型客户端令牌流
此示例演示了如何使用模型客户端创建助手代理，并通过设置 model_client_stream=True 生成令牌流。
import asyncio from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent async def main() -> None: model_client = OpenAIChatCompletionClient( model="gpt-4o", # api_key = "your_openai_api_key" ) agent = AssistantAgent( name="assistant", model_client=model_client, model_client_stream=True, ) stream = agent.run_stream(task="Name two cities in North America.") async for message in stream: print(message) asyncio.run(main())
source='user' models_usage=None metadata={} content='Name two cities in North 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=' North' 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=' New' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content=' York' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content=' City' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content=' and' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content=' Toronto' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content='.' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content=' TERMIN' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=None metadata={} content='ATE' type='ModelClientStreamingChunkEvent' source='assistant' models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0) metadata={} content='Two cities in North America are New York City and Toronto. TERMINATE' type='TextMessage' messages=[TextMessage(source='user', models_usage=None, metadata={}, content='Name two cities in North America.', type='TextMessage'), TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0), metadata={}, content='Two cities in North America are New York City and Toronto. TERMINATE', type='TextMessage')] stop_reason=None示例 3：带工具的代理
以下示例演示了如何创建一个带有模型客户端和工具的助手代理，为任务生成消息流，并使用
Console将消息打印到控制台。该工具是一个返回当前时间的简单函数。在底层，该函数被封装在
FunctionTool中并与代理的模型客户端一起使用。函数的文档字符串用作工具描述，函数名称用作工具名称，包括类型提示的函数签名用作工具参数。import asyncio from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console async def get_current_time() -> str: return "The current time is 12:00 PM." async def main() -> None: model_client = OpenAIChatCompletionClient( model="gpt-4o", # api_key = "your_openai_api_key" ) agent = AssistantAgent(name="assistant", model_client=model_client, tools=[get_current_time]) await Console(agent.run_stream(task="What is the current time?")) asyncio.run(main())
示例 4：带有 max_tool_iterations 的代理
以下示例演示了如何使用 max_tool_iterations 参数来控制代理在单次运行中可以执行工具调用的次数。当您希望代理执行多个顺序工具操作以达到目标时，这会很有用。
import asyncio from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console # Global counter state counter = 0 def increment_counter() -> str: """Increment the counter by 1 and return the current value.""" global counter counter += 1 return f"Counter incremented to: {counter}" def get_counter() -> str: """Get the current counter value.""" global counter return f"Current counter value: {counter}" async def main() -> None: model_client = OpenAIChatCompletionClient( model="gpt-4o", # api_key = "your_openai_api_key" ) # Create agent with max_tool_iterations=5 to allow multiple tool calls agent = AssistantAgent( name="assistant", model_client=model_client, tools=[increment_counter, get_counter], max_tool_iterations=5, # Allow up to 5 tool call iterations reflect_on_tool_use=True, # Get a final summary after tool calls ) await Console(agent.run_stream(task="Increment the counter 3 times and then tell me the final value.")) asyncio.run(main())
示例 5：带有 Model-Context Protocol (MCP) 工作台的代理
以下示例演示了如何创建一个带有模型客户端和
McpWorkbench的助手代理，用于与 Model-Context Protocol (MCP) 服务器交互。import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_ext.tools.mcp import StdioServerParams, McpWorkbench async def main() -> None: params = StdioServerParams( command="uvx", args=["mcp-server-fetch"], read_timeout_seconds=60, ) # You can also use `start()` and `stop()` to manage the session. async with McpWorkbench(server_params=params) as workbench: model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano") assistant = AssistantAgent( name="Assistant", model_client=model_client, workbench=workbench, reflect_on_tool_use=True, ) await Console( assistant.run_stream(task="Go to https://github.com/microsoft/autogen and tell me what you see.") ) asyncio.run(main())
示例 6：带有结构化输出和工具的代理
以下示例演示了如何创建一个带有配置为使用结构化输出和工具的模型客户端的助手代理。请注意，您需要使用
FunctionTool来创建工具，并且结构化输出模式需要 strict=True。由于模型配置为使用结构化输出，因此输出反射响应将是一个 JSON 格式的字符串。import asyncio from typing import Literal from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console from autogen_core.tools import FunctionTool from autogen_ext.models.openai import OpenAIChatCompletionClient from pydantic import BaseModel # Define the structured output format. class AgentResponse(BaseModel): thoughts: str response: Literal["happy", "sad", "neutral"] # Define the function to be called as a tool. def sentiment_analysis(text: str) -> str: """Given a text, return the sentiment.""" return "happy" if "happy" in text else "sad" if "sad" in text else "neutral" # Create a FunctionTool instance with `strict=True`, # which is required for structured output mode. tool = FunctionTool(sentiment_analysis, description="Sentiment Analysis", strict=True) # Create an OpenAIChatCompletionClient instance that supports structured output. model_client = OpenAIChatCompletionClient( model="gpt-4o-mini", ) # Create an AssistantAgent instance that uses the tool and model client. agent = AssistantAgent( name="assistant", model_client=model_client, tools=[tool], system_message="Use the tool to analyze sentiment.", output_content_type=AgentResponse, ) async def main() -> None: stream = agent.run_stream(task="I am happy today!") await Console(stream) asyncio.run(main())
---------- assistant ---------- [FunctionCall(id='call_tIZjAVyKEDuijbBwLY6RHV2p', arguments='{"text":"I am happy today!"}', name='sentiment_analysis')] ---------- assistant ---------- [FunctionExecutionResult(content='happy', call_id='call_tIZjAVyKEDuijbBwLY6RHV2p', is_error=False)] ---------- assistant ---------- {"thoughts":"The user expresses a clear positive emotion by stating they are happy today, suggesting an upbeat mood.","response":"happy"}示例 7：带有受限模型上下文的代理
以下示例演示了如何使用
BufferedChatCompletionContext，它只保留最后 2 条消息（1 条用户消息 + 1 条助手消息）。当模型对可处理的令牌数量有限制时，受限模型上下文会很有用。import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_core.model_context import BufferedChatCompletionContext from autogen_ext.models.openai import OpenAIChatCompletionClient async def main() -> None: # Create a model client. model_client = OpenAIChatCompletionClient( model="gpt-4o-mini", # api_key = "your_openai_api_key" ) # Create a model context that only keeps the last 2 messages (1 user + 1 assistant). model_context = BufferedChatCompletionContext(buffer_size=2) # Create an AssistantAgent instance with the model client and context. agent = AssistantAgent( name="assistant", model_client=model_client, model_context=model_context, system_message="You are a helpful assistant.", ) result = await agent.run(task="Name two cities in North America.") print(result.messages[-1].content) # type: ignore result = await agent.run(task="My favorite color is blue.") print(result.messages[-1].content) # type: ignore result = await agent.run(task="Did I ask you any question?") print(result.messages[-1].content) # type: ignore asyncio.run(main())
Two cities in North America are New York City and Toronto. That's great! Blue is often associated with calmness and serenity. Do you have a specific shade of blue that you like, or any particular reason why it's your favorite? No, you didn't ask a question. I apologize for any misunderstanding. If you have something specific you'd like to discuss or ask, feel free to let me know!
示例 8：带有内存的代理
以下示例演示了如何将基于列表的内存与助手代理一起使用。内存预加载了一些初始内容。在底层，内存用于在进行推断之前更新模型上下文，使用
update_context()方法。import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_core.memory import ListMemory, MemoryContent from autogen_ext.models.openai import OpenAIChatCompletionClient async def main() -> None: # Create a model client. model_client = OpenAIChatCompletionClient( model="gpt-4o-mini", # api_key = "your_openai_api_key" ) # Create a list-based memory with some initial content. memory = ListMemory() await memory.add(MemoryContent(content="User likes pizza.", mime_type="text/plain")) await memory.add(MemoryContent(content="User dislikes cheese.", mime_type="text/plain")) # Create an AssistantAgent instance with the model client and memory. agent = AssistantAgent( name="assistant", model_client=model_client, memory=[memory], system_message="You are a helpful assistant.", ) result = await agent.run(task="What is a good dinner idea?") print(result.messages[-1].content) # type: ignore asyncio.run(main())
How about making a delicious pizza without cheese? You can create a flavorful veggie pizza with a variety of toppings. Here's a quick idea: **Veggie Tomato Sauce Pizza** - Start with a pizza crust (store-bought or homemade). - Spread a layer of marinara or tomato sauce evenly over the crust. - Top with your favorite vegetables like bell peppers, mushrooms, onions, olives, and spinach. - Add some protein if you'd like, such as grilled chicken or pepperoni (ensure it's cheese-free). - Sprinkle with herbs like oregano and basil, and maybe a drizzle of olive oil. - Bake according to the crust instructions until the edges are golden and the veggies are cooked. Serve it with a side salad or some garlic bread to complete the meal! Enjoy your dinner!
示例 9：带有 `o1-mini` 的代理
以下示例演示了如何将 o1-mini 模型与助手代理一起使用。
import asyncio from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent async def main() -> None: model_client = OpenAIChatCompletionClient( model="o1-mini", # api_key = "your_openai_api_key" ) # The system message is not supported by the o1 series model. agent = AssistantAgent(name="assistant", model_client=model_client, system_message=None) result = await agent.run(task="What is the capital of France?") print(result.messages[-1].content) # type: ignore asyncio.run(main())
注意
o1-preview 和 o1-mini 模型不支持系统消息和函数调用。因此，system_message 应设置为 None，并且不应设置 tools 和 handoffs。有关更多详细信息，请参阅 o1 beta 限制。
示例 10：使用推理模型和自定义模型上下文的代理。
以下示例演示了如何将推理模型 (DeepSeek R1) 与助手代理一起使用。模型上下文用于从助手消息中过滤掉思想字段。
import asyncio from typing import List from autogen_agentchat.agents import AssistantAgent from autogen_core.model_context import UnboundedChatCompletionContext from autogen_core.models import AssistantMessage, LLMMessage, ModelFamily from autogen_ext.models.ollama import OllamaChatCompletionClient class ReasoningModelContext(UnboundedChatCompletionContext): """A model context for reasoning models.""" async def get_messages(self) -> List[LLMMessage]: messages = await super().get_messages() # Filter out thought field from AssistantMessage. messages_out: List[LLMMessage] = [] for message in messages: if isinstance(message, AssistantMessage): message.thought = None messages_out.append(message) return messages_out # Create an instance of the model client for DeepSeek R1 hosted locally on Ollama. model_client = OllamaChatCompletionClient( model="deepseek-r1:8b", model_info={ "vision": False, "function_calling": False, "json_output": False, "family": ModelFamily.R1, "structured_output": True, }, ) agent = AssistantAgent( "reasoning_agent", model_client=model_client, model_context=ReasoningModelContext(), # Use the custom model context. ) async def run_reasoning_agent() -> None: result = await agent.run(task="What is the capital of France?") print(result) asyncio.run(run_reasoning_agent())
有关详细示例和用法，请参阅下面的“示例”部分。
- component_config_schema#
的别名
AssistantAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.AssistantAgent'#
覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
获取此代理可以生成的消息类型。
- 返回:
此代理可以生成的消息类型序列。
- property model_context: ChatCompletionContext#
获取此代理使用的模型上下文。
- 返回:
此代理的聊天完成上下文。
- async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) Response[source]#
处理传入消息并生成响应。
- 参数:
messages – 要处理的消息序列。
cancellation_token – 用于取消操作的令牌。
- 返回:
包含代理回复的响应。
- async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][source]#
处理消息并流式传输响应。
- 参数:
messages – 要处理的消息序列。
cancellation_token – 用于取消操作的令牌。
- 生成:
处理过程中的事件、消息和最终响应。
- async on_reset(cancellation_token: CancellationToken) None[source]#
将助手代理重置为其初始化状态。
- class CodeExecutorAgent(name: str, code_executor: CodeExecutor, *, model_client: ChatCompletionClient | None = None, model_context: ChatCompletionContext | None = None, model_client_stream: bool = False, max_retries_on_error: int = 0, description: str | None = None, system_message: str | None = DEFAULT_SYSTEM_MESSAGE, sources: Sequence[str] | None = None, supported_languages: List[str] | None = None, approval_func: Callable[[ApprovalRequest], ApprovalResponse] | Callable[[ApprovalRequest], Awaitable[ApprovalResponse]] | None = None)[source]#
基类：
BaseChatAgent,Component[CodeExecutorAgentConfig]（实验性）一个根据用户指令生成和执行代码片段的代理。
注意
此代理为实验性，未来版本可能会更改。
它通常与另一个生成要执行的代码片段的代理一起在团队中使用，或者单独提供 model_client，以便它可以根据用户查询生成代码、执行代码并反思代码结果。
当与 model_client 一起使用时，它将使用模型生成代码片段并使用提供的 code_executor 执行它们。模型还将反思代码执行结果。代理将把模型产生的最终反思结果作为最终响应。
当不使用 model_client 时，它只会执行
TextMessage消息中找到的代码块，并返回代码执行的输出。注意
使用
AssistantAgent和PythonCodeExecutionTool是此代理的替代方案。但是，该代理的模型必须生成正确转义的代码字符串作为工具的参数。- 参数:
name (str) – 代理的名称。
code_executor (CodeExecutor) – 负责执行消息中收到的代码的代码执行器（推荐使用
DockerCommandLineCodeExecutor。请参阅下面的示例）。model_client (ChatCompletionClient, optional) – 用于推理和生成代码的模型客户端。如果未提供，代理将只执行输入消息中找到的代码块。目前，模型必须支持结构化输出模式，这是自动重试机制工作所必需的。
model_client_stream (bool, optional) – 如果为 True，则模型客户端将以流模式使用。
on_messages_stream()和BaseChatAgent.run_stream()方法也将生成ModelClientStreamingChunkEvent消息，因为模型客户端会生成响应块。默认为 False。description (str, optional) – 代理的描述。如果未提供，将使用
DEFAULT_AGENT_DESCRIPTION。system_message (str, optional) – 模型的系统消息。如果提供，它将在进行推理时附加到模型上下文中的消息之前。设置为 None 以禁用。默认为
DEFAULT_SYSTEM_MESSAGE。仅在提供了 model_client 时使用。sources (Sequence[str], optional) – 仅检查来自指定代理的消息以执行代码。当代理是群聊的一部分并且您想将代码执行限制为来自特定代理的消息时，这很有用。如果未提供，所有消息都将检查代码块。仅在未提供 model_client 时使用。
max_retries_on_error (int, optional) – 出错时最大重试次数。如果代码执行失败，代理将重试此次数。如果代码执行在此重试次数后仍失败，代理将生成一个反思结果。
supported_languages (List[str], optional) – 将从代理响应中解析和执行的编程语言列表；其他语言将被忽略。默认为 DEFAULT_SUPPORTED_LANGUAGES。
approval_func (Optional[Union[Callable[[ApprovalRequest], ApprovalResponse], Callable[[ApprovalRequest], Awaitable[ApprovalResponse]]]], optional) – 在每次代码执行之前调用的函数，以获取批准。该函数接收一个包含要执行的代码和当前上下文的 ApprovalRequest，并返回一个 ApprovalResponse。该函数可以是同步或异步的。如果为 None（默认），所有代码执行将自动批准。如果设置，则无法使用
dump_component()序列化代理。
注意
建议 CodeExecutorAgent 代理使用 Docker 容器执行代码。这确保了模型生成的代码在隔离环境中执行。要使用 Docker，您的环境必须安装并运行 Docker。请遵循 Docker 的安装说明。
注意
代码执行器只处理使用三反引号在 Markdown 代码块中正确格式化的代码。例如：
```python print("Hello World") ``` # or ```sh echo "Hello World" ```在此示例中，我们展示了如何设置一个 CodeExecutorAgent 代理，该代理使用
DockerCommandLineCodeExecutor在 Docker 容器中执行代码片段。work_dir 参数指示所有执行文件在 Docker 容器中执行之前首次在本地保存的位置。import asyncio from autogen_agentchat.agents import CodeExecutorAgent, ApprovalRequest, ApprovalResponse from autogen_agentchat.messages import TextMessage from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor from autogen_core import CancellationToken def simple_approval_func(request: ApprovalRequest) -> ApprovalResponse: """Simple approval function that requests user input for code execution approval.""" print("Code execution approval requested:") print("=" * 50) print(request.code) print("=" * 50) while True: user_input = input("Do you want to execute this code? (y/n): ").strip().lower() if user_input in ['y', 'yes']: return ApprovalResponse(approved=True, reason='Approved by user') elif user_input in ['n', 'no']: return ApprovalResponse(approved=False, reason='Denied by user') else: print("Please enter 'y' for yes or 'n' for no.") async def run_code_executor_agent() -> None: # Create a code executor agent that uses a Docker container to execute code. code_executor = DockerCommandLineCodeExecutor(work_dir="coding") await code_executor.start() code_executor_agent = CodeExecutorAgent( "code_executor", code_executor=code_executor, approval_func=simple_approval_func ) # Run the agent with a given code snippet. task = TextMessage( content='''Here is some code ```python print('Hello world') ``` ''', source="user", ) response = await code_executor_agent.on_messages([task], CancellationToken()) print(response.chat_message) # Stop the code executor. await code_executor.stop() asyncio.run(run_code_executor_agent())
在此示例中，我们展示了如何设置一个 CodeExecutorAgent 代理，该代理使用
DeviceRequest将 GPU 暴露给容器以进行 CUDA 加速的代码执行。import asyncio from autogen_agentchat.agents import CodeExecutorAgent from autogen_agentchat.messages import TextMessage from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor from autogen_core import CancellationToken from docker.types import DeviceRequest async def run_code_executor_agent() -> None: # Create a code executor agent that uses a Docker container to execute code. code_executor = DockerCommandLineCodeExecutor( work_dir="coding", device_requests=[DeviceRequest(count=-1, capabilities=[["gpu"]])] ) await code_executor.start() code_executor_agent = CodeExecutorAgent("code_executor", code_executor=code_executor) # Display the GPU information task = TextMessage( content='''Here is some code ```sh nvidia-smi ``` ''', source="user", ) response = await code_executor_agent.on_messages([task], CancellationToken()) print(response.chat_message) # Stop the code executor. await code_executor.stop() asyncio.run(run_code_executor_agent())
在以下示例中，我们展示了如何设置不带 model_client 参数的 CodeExecutorAgent，用于在群聊中执行其他代理生成的代码块，使用
DockerCommandLineCodeExecutor。import asyncio from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent, ApprovalRequest, ApprovalResponse from autogen_agentchat.conditions import MaxMessageTermination from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.ui import Console termination_condition = MaxMessageTermination(3) def group_chat_approval_func(request: ApprovalRequest) -> ApprovalResponse: """Approval function for group chat that allows basic Python operations.""" # Allow common safe operations safe_operations = ["print(", "import ", "def ", "class ", "if ", "for ", "while "] if any(op in request.code for op in safe_operations): return ApprovalResponse(approved=True, reason='Safe Python operation') # Deny file system operations in group chat dangerous_operations = ["open(", "file(", "os.", "subprocess", "eval(", "exec("] if any(op in request.code for op in dangerous_operations): return ApprovalResponse(approved=False, reason='File system or dangerous operation not allowed') return ApprovalResponse(approved=True, reason='Operation approved') async def main() -> None: model_client = OpenAIChatCompletionClient(model="gpt-4o") # define the Docker CLI Code Executor code_executor = DockerCommandLineCodeExecutor(work_dir="coding") # start the execution container await code_executor.start() code_executor_agent = CodeExecutorAgent( "code_executor_agent", code_executor=code_executor, approval_func=group_chat_approval_func ) coder_agent = AssistantAgent("coder_agent", model_client=model_client) groupchat = RoundRobinGroupChat( participants=[coder_agent, code_executor_agent], termination_condition=termination_condition ) task = "Write python code to print Hello World!" await Console(groupchat.run_stream(task=task)) # stop the execution container await code_executor.stop() asyncio.run(main())
---------- user ---------- Write python code to print Hello World! ---------- coder_agent ---------- Certainly! Here's a simple Python code to print "Hello World!": ```python print("Hello World!") ``` You can run this code in any Python environment to display the message. ---------- code_executor_agent ---------- Hello World!在以下示例中，我们展示了如何设置带有 model_client 的 CodeExecutorAgent，它可以在无需其他代理帮助的情况下生成自己的代码，并在
DockerCommandLineCodeExecutor中执行。它还演示了使用基于模型的批准函数，在执行之前审查代码的安全性。import asyncio from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_core.models import SystemMessage, UserMessage from autogen_agentchat.agents import CodeExecutorAgent, ApprovalRequest, ApprovalResponse from autogen_agentchat.conditions import TextMessageTermination from autogen_agentchat.ui import Console termination_condition = TextMessageTermination("code_executor_agent") async def main() -> None: model_client = OpenAIChatCompletionClient(model="gpt-4o") async def model_client_approval_func(request: ApprovalRequest) -> ApprovalResponse: instruction = "Approve or reject the code in the last message based on whether it is dangerous or not. Use the following JSON format for your response: {approved: true/false, reason: 'your reason here'}" response = await model_client.create( messages=[SystemMessage(content=instruction)] + request.context + [UserMessage(content=request.code, source="user")], json_output=ApprovalResponse, ) assert isinstance(response.content, str) return ApprovalResponse.model_validate_json(response.content) # define the Docker CLI Code Executor code_executor = DockerCommandLineCodeExecutor(work_dir="coding") # start the execution container await code_executor.start() code_executor_agent = CodeExecutorAgent( "code_executor_agent", code_executor=code_executor, model_client=model_client, approval_func=model_client_approval_func, ) task = "Write python code to print Hello World!" await Console(code_executor_agent.run_stream(task=task)) # stop the execution container await code_executor.stop() asyncio.run(main())
---------- user ---------- Write python code to print Hello World! ---------- code_executor_agent ---------- Certainly! Here is a simple Python code to print "Hello World!" to the console: ```python print("Hello World!") ``` Let's execute it to confirm the output. ---------- code_executor_agent ---------- Hello World! ---------- code_executor_agent ---------- The code has been executed successfully, and it printed "Hello World!" as expected. If you have any more requests or questions, feel free to ask!- DEFAULT_TERMINAL_DESCRIPTION = '一个计算机终端，除了运行 Python 脚本（以 ```python 代码块引用提供）或 sh shell 脚本（以 ```sh 代码块引用提供）之外，不执行其他操作。'#
- DEFAULT_AGENT_DESCRIPTION = '一个代码执行代理，根据用户指令生成和执行 Python 和 shell 脚本。它确保正确性、效率和最小错误，同时优雅地处理边缘情况。'#
- DEFAULT_SYSTEM_MESSAGE = '你是一个代码执行代理。你的职责是根据用户指令生成和执行 Python 代码和 shell 脚本，确保正确性、效率和最小错误。优雅地处理边缘情况。Python 代码应以 ```python 代码块的形式提供，sh shell 脚本应以 ```sh 代码块的形式提供以供执行。'#
- NO_CODE_BLOCKS_FOUND_MESSAGE = '在线程中未找到代码块。请提供至少一个 Markdown 编码的代码块以执行（即，在 ```python 或 ```sh 代码块中引用代码）。'#
- DEFAULT_SUPPORTED_LANGUAGES = ['python', 'sh']#
- component_config_schema#
别名
CodeExecutorAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.CodeExecutorAgent'#
覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代码执行器代理生成的消息类型。
- property model_context: ChatCompletionContext#
代理正在使用的模型上下文。
- async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) Response[source]#
处理传入消息并返回响应。
注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][source]#
处理传入消息并流式传输事件/响应。
- async extract_code_blocks_from_messages(messages: Sequence[BaseChatMessage]) List[CodeBlock][source]#
- async execute_code_block(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) CodeResult[source]#
- async on_reset(cancellation_token: CancellationToken) None[source]#
它是一个无操作，因为代码执行器代理没有可变状态。
- class SocietyOfMindAgent(name: str, team: Team, model_client: ChatCompletionClient, *, description: str = DEFAULT_DESCRIPTION, instruction: str = DEFAULT_INSTRUCTION, response_prompt: str = DEFAULT_RESPONSE_PROMPT, model_context: ChatCompletionContext | None = None)[source]#
基类：
BaseChatAgent,Component[SocietyOfMindAgentConfig]一个使用内部代理团队生成响应的代理。
每次调用代理的
on_messages()或on_messages_stream()方法时，它都会运行内部代理团队，然后使用模型客户端根据内部团队的消息生成响应。生成响应后，代理通过调用Team.reset()重置内部团队。限制发送到模型的上下文大小
您可以通过将 model_context 参数设置为
BufferedChatCompletionContext来限制发送到模型的消息数量。这将限制发送到模型的最新消息数量，当模型对可处理的令牌数量有限制时，这很有用。您还可以通过子类化ChatCompletionContext来创建自己的模型上下文。- 参数:
name (str) – 代理的名称。
team (Team) – 要使用的代理团队。
model_client (ChatCompletionClient) – 用于准备响应的模型客户端。
description (str, optional) – 代理的描述。
instruction (str, optional) – 使用内部团队消息生成响应时使用的指令。默认为
DEFAULT_INSTRUCTION。它扮演“系统”的角色。response_prompt (str, optional) – 使用内部团队消息生成响应时使用的响应提示。默认为
DEFAULT_RESPONSE_PROMPT。它扮演“系统”的角色。model_context (ChatCompletionContext | None, optional) – 用于存储和检索
LLMMessage的模型上下文。可以预加载初始消息。当代理重置时，初始消息将被清除。
示例
import asyncio from autogen_agentchat.ui import Console from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.conditions import TextMentionTermination async def main() -> None: model_client = OpenAIChatCompletionClient(model="gpt-4o") agent1 = AssistantAgent("assistant1", model_client=model_client, system_message="You are a writer, write well.") agent2 = AssistantAgent( "assistant2", model_client=model_client, system_message="You are an editor, provide critical feedback. Respond with 'APPROVE' if the text addresses all feedbacks.", ) inner_termination = TextMentionTermination("APPROVE") inner_team = RoundRobinGroupChat([agent1, agent2], termination_condition=inner_termination) society_of_mind_agent = SocietyOfMindAgent("society_of_mind", team=inner_team, model_client=model_client) agent3 = AssistantAgent( "assistant3", model_client=model_client, system_message="Translate the text to Spanish." ) team = RoundRobinGroupChat([society_of_mind_agent, agent3], max_turns=2) stream = team.run_stream(task="Write a short story with a surprising ending.") await Console(stream) asyncio.run(main())
- component_config_schema#
别名
SocietyOfMindAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.SocietyOfMindAgent'#
覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。
- DEFAULT_INSTRUCTION = '之前你被要求完成一个请求。你和你的团队勤奋工作以解决该请求。以下是该对话的记录：'#
使用内部团队消息生成响应时使用的默认指令。在使用模型生成响应时，该指令将附加到内部团队的消息之前。它扮演“系统”的角色。
- 类型:
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- property model_context: ChatCompletionContext#
代理正在使用的模型上下文。
- async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) Response[source]#
处理传入消息并返回响应。
注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][source]#
处理传入消息并流式返回消息，最后一个项是响应。
BaseChatAgent中的基本实现只是调用on_messages()并生成响应中的消息。注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_reset(cancellation_token: CancellationToken) None[source]#
将代理重置为其初始化状态。
- class UserProxyAgent(name: str, *, description: str = '一个人类用户', input_func: Callable[[str], str] | Callable[[str, CancellationToken | None], Awaitable[str]] | None = None)[source]#
基类：
BaseChatAgent,Component[UserProxyAgentConfig]一个可以通过输入函数代表人类用户的代理。
此代理可以通过提供自定义输入函数来表示聊天系统中的人类用户。
注意
使用
UserProxyAgent会将正在运行的团队置于临时阻塞状态，直到用户响应。因此，如果用户不响应，请务必超时用户输入函数并使用CancellationToken取消。输入函数还应处理异常并在需要时返回默认响应。对于涉及缓慢人类响应的典型用例，建议使用终止条件，例如
HandoffTermination或SourceMatchTermination来停止正在运行的团队并将控制权返回给应用程序。您可以使用用户输入再次运行团队。这样，可以在用户响应时保存和恢复团队的状态。有关更多信息，请参阅 人类在环。
- 参数:
有关与 Web 和 UI 框架集成的示例，请参阅以下内容：
示例
简单使用案例
import asyncio from autogen_core import CancellationToken from autogen_agentchat.agents import UserProxyAgent from autogen_agentchat.messages import TextMessage async def simple_user_agent(): agent = UserProxyAgent("user_proxy") response = await asyncio.create_task( agent.on_messages( [TextMessage(content="What is your name? ", source="user")], cancellation_token=CancellationToken(), ) ) assert isinstance(response.chat_message, TextMessage) print(f"Your name is {response.chat_message.content}")
示例
可取消的使用案例
import asyncio from typing import Any from autogen_core import CancellationToken from autogen_agentchat.agents import UserProxyAgent from autogen_agentchat.messages import TextMessage token = CancellationToken() agent = UserProxyAgent("user_proxy") async def timeout(delay: float): await asyncio.sleep(delay) def cancellation_callback(task: asyncio.Task[Any]): token.cancel() async def cancellable_user_agent(): try: timeout_task = asyncio.create_task(timeout(3)) timeout_task.add_done_callback(cancellation_callback) agent_task = asyncio.create_task( agent.on_messages( [TextMessage(content="What is your name? ", source="user")], cancellation_token=token, ) ) response = await agent_task assert isinstance(response.chat_message, TextMessage) print(f"Your name is {response.chat_message.content}") except Exception as e: print(f"Exception: {e}") except BaseException as e: print(f"BaseException: {e}")
- component_type: ClassVar[ComponentType] = 'agent'#
组件的逻辑类型。
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.UserProxyAgent'#
覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。
- component_config_schema#
别名
UserProxyAgentConfig
- property produced_message_types: Sequence[type[BaseChatMessage]]#
此代理可以生成的消息类型。
- async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) Response[source]#
处理传入消息并返回响应。
注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][source]#
通过请求用户输入来处理传入消息。
- async on_reset(cancellation_token: CancellationToken | None = None) None[source]#
重置代理状态。
- class MessageFilterAgent(name: str, wrapped_agent: BaseChatAgent, filter: MessageFilterConfig)[source]#
基类：
BaseChatAgent,Component[MessageFilterAgentConfig]一个包装代理，在将传入消息传递给内部代理之前对其进行过滤。
警告
这是一个实验性功能，API 将在未来的版本中进行更改。
这在多代理工作流等场景中非常有用，例如，代理只应处理完整消息历史记录的子集——例如，仅处理来自每个上游代理的最后一条消息，或者仅处理来自特定源的第一条消息。
过滤使用
MessageFilterConfig进行配置，它支持：- 按消息源过滤（例如，仅来自“用户”或其他代理的消息）- 从每个源选择前 N 条或后 N 条消息- 如果位置是 None，则包含来自该源的所有消息此代理与直接消息传递和基于团队的执行（例如
GraphFlow）兼容。示例
>>> agent_a = MessageFilterAgent( ... name="A", ... wrapped_agent=some_other_agent, ... filter=MessageFilterConfig( ... per_source=[ ... PerSourceFilter(source="user", position="first", count=1), ... PerSourceFilter(source="B", position="last", count=2), ... ] ... ), ... )
- Graph 的示例用例
假设您有一个循环多代理图：A → B → A → B → C。
您希望：- A 只看到用户消息和来自 B 的最后一条消息- B 看到用户消息、来自 A 的最后一条消息以及它自己的先前响应（用于反思）- C 看到用户消息和来自 B 的最后一条消息
按如下方式包装代理
>>> agent_a = MessageFilterAgent( ... name="A", ... wrapped_agent=agent_a_inner, ... filter=MessageFilterConfig( ... per_source=[ ... PerSourceFilter(source="user", position="first", count=1), ... PerSourceFilter(source="B", position="last", count=1), ... ] ... ), ... )
>>> agent_b = MessageFilterAgent( ... name="B", ... wrapped_agent=agent_b_inner, ... filter=MessageFilterConfig( ... per_source=[ ... PerSourceFilter(source="user", position="first", count=1), ... PerSourceFilter(source="A", position="last", count=1), ... PerSourceFilter(source="B", position="last", count=10), ... ] ... ), ... )
>>> agent_c = MessageFilterAgent( ... name="C", ... wrapped_agent=agent_c_inner, ... filter=MessageFilterConfig( ... per_source=[ ... PerSourceFilter(source="user", position="first", count=1), ... PerSourceFilter(source="B", position="last", count=1), ... ] ... ), ... )
然后定义图
>>> graph = DiGraph( ... nodes={ ... "A": DiGraphNode(name="A", edges=[DiGraphEdge(target="B")]), ... "B": DiGraphNode( ... name="B", ... edges=[ ... DiGraphEdge(target="C", condition="exit"), ... DiGraphEdge(target="A", condition="loop"), ... ], ... ), ... "C": DiGraphNode(name="C", edges=[]), ... }, ... default_start_node="A", ... )
这将确保每个代理只看到其决策或行动逻辑所需的内容。
- component_config_schema#
别名
MessageFilterAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.MessageFilterAgent'#
覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) Response[source]#
处理传入消息并返回响应。
注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][source]#
处理传入消息并返回消息流，最后一个项目是响应。
BaseChatAgent中的基本实现只是调用on_messages()并生成响应中的消息。注意
代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。
- async on_reset(cancellation_token: CancellationToken) None[source]#
将代理重置为其初始化状态。
- classmethod _from_config(config: MessageFilterAgentConfig) MessageFilterAgent[source]#
从配置对象创建组件的新实例。
- 参数:
config (T) – 配置对象。
- 返回:
Self – 组件的新实例。
- pydantic model MessageFilterConfig[source]#
基类:
BaseModel显示 JSON 模式
{ "title": "MessageFilterConfig", "type": "object", "properties": { "per_source": { "items": { "$ref": "#/$defs/PerSourceFilter" }, "title": "Per Source", "type": "array" } }, "$defs": { "PerSourceFilter": { "properties": { "source": { "title": "Source", "type": "string" }, "position": { "anyOf": [ { "enum": [ "first", "last" ], "type": "string" }, { "type": "null" } ], "default": null, "title": "Position" }, "count": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": null, "title": "Count" } }, "required": [ "source" ], "title": "PerSourceFilter", "type": "object" } }, "required": [ "per_source" ] }
- 字段:
per_source (List[autogen_agentchat.agents._message_filter_agent.PerSourceFilter])
- field per_source: List[PerSourceFilter] [Required]#
- pydantic model PerSourceFilter[source]#
基类:
BaseModel显示 JSON 模式
{ "title": "PerSourceFilter", "type": "object", "properties": { "source": { "title": "Source", "type": "string" }, "position": { "anyOf": [ { "enum": [ "first", "last" ], "type": "string" }, { "type": "null" } ], "default": null, "title": "Position" }, "count": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "default": null, "title": "Count" } }, "required": [ "source" ] }
- 字段:
count (int | None)position (Literal['first', 'last'] | None)source (str)
- pydantic model ApprovalRequest[source]#
基类:
BaseModel代码执行的批准请求。
显示 JSON 模式
{ "title": "ApprovalRequest", "description": "Request for approval of code execution.", "type": "object", "properties": { "code": { "title": "Code", "type": "string" }, "context": { "items": { "discriminator": { "mapping": { "AssistantMessage": "#/$defs/AssistantMessage", "FunctionExecutionResultMessage": "#/$defs/FunctionExecutionResultMessage", "SystemMessage": "#/$defs/SystemMessage", "UserMessage": "#/$defs/UserMessage" }, "propertyName": "type" }, "oneOf": [ { "$ref": "#/$defs/SystemMessage" }, { "$ref": "#/$defs/UserMessage" }, { "$ref": "#/$defs/AssistantMessage" }, { "$ref": "#/$defs/FunctionExecutionResultMessage" } ] }, "title": "Context", "type": "array" } }, "$defs": { "AssistantMessage": { "description": "Assistant message are sampled from the language model.", "properties": { "content": { "anyOf": [ { "type": "string" }, { "items": { "$ref": "#/$defs/FunctionCall" }, "type": "array" } ], "title": "Content" }, "thought": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "default": null, "title": "Thought" }, "source": { "title": "Source", "type": "string" }, "type": { "const": "AssistantMessage", "default": "AssistantMessage", "title": "Type", "type": "string" } }, "required": [ "content", "source" ], "title": "AssistantMessage", "type": "object" }, "FunctionCall": { "properties": { "id": { "title": "Id", "type": "string" }, "arguments": { "title": "Arguments", "type": "string" }, "name": { "title": "Name", "type": "string" } }, "required": [ "id", "arguments", "name" ], "title": "FunctionCall", "type": "object" }, "FunctionExecutionResult": { "description": "Function execution result contains the output of a function call.", "properties": { "content": { "title": "Content", "type": "string" }, "name": { "title": "Name", "type": "string" }, "call_id": { "title": "Call Id", "type": "string" }, "is_error": { "anyOf": [ { "type": "boolean" }, { "type": "null" } ], "default": null, "title": "Is Error" } }, "required": [ "content", "name", "call_id" ], "title": "FunctionExecutionResult", "type": "object" }, "FunctionExecutionResultMessage": { "description": "Function execution result message contains the output of multiple function calls.", "properties": { "content": { "items": { "$ref": "#/$defs/FunctionExecutionResult" }, "title": "Content", "type": "array" }, "type": { "const": "FunctionExecutionResultMessage", "default": "FunctionExecutionResultMessage", "title": "Type", "type": "string" } }, "required": [ "content" ], "title": "FunctionExecutionResultMessage", "type": "object" }, "SystemMessage": { "description": "System message contains instructions for the model coming from the developer.\n\n.. note::\n\n Open AI is moving away from using 'system' role in favor of 'developer' role.\n See `Model Spec <https://cdn.openai.com/spec/model-spec-2024-05-08.html#definitions>`_ for more details.\n However, the 'system' role is still allowed in their API and will be automatically converted to 'developer' role\n on the server side.\n So, you can use `SystemMessage` for developer messages.", "properties": { "content": { "title": "Content", "type": "string" }, "type": { "const": "SystemMessage", "default": "SystemMessage", "title": "Type", "type": "string" } }, "required": [ "content" ], "title": "SystemMessage", "type": "object" }, "UserMessage": { "description": "User message contains input from end users, or a catch-all for data provided to the model.", "properties": { "content": { "anyOf": [ { "type": "string" }, { "items": { "anyOf": [ { "type": "string" }, {} ] }, "type": "array" } ], "title": "Content" }, "source": { "title": "Source", "type": "string" }, "type": { "const": "UserMessage", "default": "UserMessage", "title": "Type", "type": "string" } }, "required": [ "content", "source" ], "title": "UserMessage", "type": "object" } }, "required": [ "code", "context" ] }
- 字段:
code (str)context (List[autogen_core.models._types.SystemMessage | autogen_core.models._types.UserMessage | autogen_core.models._types.AssistantMessage | autogen_core.models._types.FunctionExecutionResultMessage])
- field context: List[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] [Required]#
- pydantic model ApprovalResponse[source]#
基类:
BaseModel对批准请求的响应。
显示 JSON 模式
{ "title": "ApprovalResponse", "description": "Response to approval request.", "type": "object", "properties": { "approved": { "title": "Approved", "type": "boolean" }, "reason": { "title": "Reason", "type": "string" } }, "required": [ "approved", "reason" ] }
- 字段:
approved (bool)reason (str)