autogen_agentchat.agents#
这个模块初始化了包提供的各种预定义代理。BaseChatAgent 是 AgentChat 中所有代理的基类。
- class AssistantAgent(name: str, model_client: ChatCompletionClient, *, tools: List[BaseTool[Any, Any] | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, workbench: Workbench | None = None, handoffs: List[Handoff | str] | None = None, model_context: ChatCompletionContext | None = None, description: str = 'An agent that provides assistance with ability to use tools.', system_message: str | None = 'You are a helpful AI assistant. Solve tasks using your tools. Reply with TERMINATE when the task has been completed.', model_client_stream: bool = False, reflect_on_tool_use: bool | None = None, tool_call_summary_format: str = '{result}', 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)[source]#
继承自：
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中返回。- 当模型返回工具调用时，它们将立即执行
当 reflect_on_tool_use 为 False 时，工具调用结果会作为
ToolCallSummaryMessage在chat_message中返回。 可以使用 tool_call_summary_format 自定义工具调用摘要。当 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。
提示
默认情况下，当进行工具调用时，工具调用结果作为响应返回。 因此，建议注意工具返回值格式，特别是当另一个代理以特定格式期望它们时。 如果需要，可以使用 tool_call_summary_format 自定义工具调用摘要。
切换行为
如果触发了切换，则会在
chat_message中返回HandoffMessage。如果存在工具调用，它们也将在返回切换之前立即执行。
工具调用和结果通过
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 | 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响应内容的格式字符串。tool_call_summary_format (str, optional) – 用于创建
ToolCallSummaryMessage响应内容的格式字符串。 该格式字符串用于格式化每个工具调用结果的工具调用摘要。 默认为“{result}”。当reflect_on_tool_use为False时，所有工具调用摘要的连接（以换行符“\n”分隔）将作为响应返回。 可用变量：{tool_name}, {arguments}, {result}。 例如，“{tool_name}: {result}”将创建一个类似“tool_name: result”的摘要。memory (Sequence[Memory] | None, optional) – 代理使用的内存存储。默认为None。
- Raises:
ValueError – 如果工具名称不唯一。
ValueError – 如果移交名称不唯一。
ValueError – 如果移交名称与工具名称不唯一。
ValueError – 如果工具迭代的最大次数小于 1。
Examples
Example 1: basic agent
以下示例演示了如何创建一个具有模型客户端的助理代理，并生成对简单任务的响应。
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：具有模型-上下文协议（MCP）工作台的代理
以下示例演示了如何创建一个具有模型客户端和
McpWorkbench的助理代理，用于与模型-上下文协议（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())
示例5：具有结构化输出和工具的代理
以下示例演示了如何创建一个具有模型客户端的助理代理，该模型客户端配置为使用结构化输出和工具。 请注意，您需要使用
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"}示例6：具有有界模型上下文的代理
以下示例显示了如何使用仅保留最后2条消息（1个用户+ 1个助手）的
BufferedChatCompletionContext。 当模型对其可以处理的令牌数量有限制时，有界模型上下文非常有用。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!
示例7：具有内存的代理
以下示例显示了如何将基于列表的内存与助理代理一起使用。 内存预先加载了一些初始内容。 在底层，内存用于在使用
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!
示例8：带有`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 limitations。
示例9：使用推理模型和自定义模型上下文的代理。
以下示例显示了如何将推理模型（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 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 on_reset(cancellation_token: CancellationToken) None[source]#
将助手代理重置为其初始化状态。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- class BaseChatAgent(name: str, description: str)[source]#
基类：
ChatAgent,ABC,ComponentBase[BaseModel]聊天代理的基类。
这个抽象类为
ChatAgent提供了一个基本实现。要创建一个新的聊天代理，请继承这个类并实现on_messages()、on_reset()和produced_message_types。如果需要流式传输，还应实现on_messages_stream()方法。代理被认为是状态ful的，并在调用
on_messages()或on_messages_stream()方法之间维护其状态。代理应该将其状态存储在代理实例中。代理还应该实现on_reset()方法，以便将代理重置为其初始化状态。注意
调用者应该只在每次调用
on_messages()或on_messages_stream()方法时，才将新消息传递给代理。不要在每次调用时将整个对话历史记录传递给代理。在创建新代理时，必须遵循此设计原则。- async close() None[source]#
释放代理持有的任何资源。这在
BaseChatAgent类中默认是一个空操作。子类可以重写此方法以实现自定义关闭行为。
- component_type: ClassVar[ComponentType] = 'agent'#
组件的逻辑类型。
- abstract 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_pause(cancellation_token: CancellationToken) None[source]#
当代理在其
on_messages()或on_messages_stream()方法中运行时暂停时调用。 默认情况下，在BaseChatAgent类中，这是一个空操作。 子类可以覆盖此方法以实现自定义的暂停行为。
- abstract async on_reset(cancellation_token: CancellationToken) None[source]#
将代理重置为其初始化状态。
- async on_resume(cancellation_token: CancellationToken) None[source]#
当代理在其
on_messages()或on_messages_stream()方法中运行时，从暂停状态恢复时调用。 默认情况下，在BaseChatAgent类中，这是一个空操作。 子类可以覆盖此方法以实现自定义的恢复行为。
- abstract property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- async run(*, task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None, cancellation_token: CancellationToken | None = None) TaskResult[source]#
使用给定的任务运行代理并返回结果。
- async run_stream(*, task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None, cancellation_token: CancellationToken | None = None) AsyncGenerator[BaseAgentEvent | BaseChatMessage | TaskResult, 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)[source]#
基类：
BaseChatAgent,Component[CodeExecutorAgentConfig]（实验性）一个根据用户指令生成和执行代码片段的代理。
注意
此代理是实验性的，将来版本可能会更改。
它通常在团队中与另一个生成要执行的代码片段的代理一起使用，或者在提供 model_client 的情况下单独使用，以便它可以根据用户查询生成代码，执行代码并反思代码结果。
当与 model_client 一起使用时，它将使用模型生成代码片段，并使用提供的 code_executor 执行它们。 该模型还将反思代码执行结果。 该代理将从模型中产生最终的思考结果作为最终响应。
当不使用 model_client 时，它将仅执行在
TextMessage消息中找到的代码块，并返回代码执行的输出。注意
使用带有
PythonCodeExecutionTool的AssistantAgent是此代理的替代方法。 但是，该代理的模型将必须生成正确转义的代码字符串作为该工具的参数。- 参数:
name (str) – 代理的名称。
code_executor (CodeExecutor) – 代码执行器负责执行在消息中收到的代码（推荐使用
DockerCommandLineCodeExecutor。 见下面的例子）model_client (ChatCompletionClient, 可选) – 用于推理和生成代码的模型客户端。如果未提供，代理将仅执行输入消息中找到的代码块。目前，模型必须支持结构化输出模式，这是自动重试机制工作所必需的。
model_client_stream (bool, 可选) – 如果为 True，模型客户端将以流模式使用。
on_messages_stream()和BaseChatAgent.run_stream()方法还将生成ModelClientStreamingChunkEvent消息，因为模型客户端会生成响应块。默认为 False。description (str, 可选) – 代理的描述。如果未提供，将使用
DEFAULT_AGENT_DESCRIPTION。system_message (str, 可选) – 模型的系统消息。如果提供，它将在进行推理时添加到模型上下文中的消息之前。设置为 None 以禁用。默认为
DEFAULT_SYSTEM_MESSAGE。这仅在提供 model_client 时使用。sources (Sequence[str], 可选) – 仅检查来自指定代理的消息以执行代码。当代理是群聊的一部分，并且您希望将代码执行限制为来自特定代理的消息时，这很有用。如果未提供，将检查所有消息中的代码块。这仅在未提供 model_client 时使用。
max_retries_on_error (int, 可选) – 出错时的最大重试次数。如果代码执行失败，代理将重试最多此次数。如果在此重试次数后代码执行仍然失败，代理将产生一个反射结果。
注意
建议 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 from autogen_agentchat.messages import TextMessage from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor from autogen_core import CancellationToken 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) # 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 ```bash 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 from autogen_agentchat.conditions import MaxMessageTermination from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.ui import Console termination_condition = MaxMessageTermination(3) 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) 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_agentchat.agents import CodeExecutorAgent 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") # 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 ) 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_AGENT_DESCRIPTION = '一个代码执行代理，它根据用户指令生成并执行 Python 和 shell 脚本。Python 代码应以 ```python 代码块形式提供，而 sh shell 脚本应以 ```sh 代码块形式提供以供执行。它确保正确性、效率和最小的错误，同时优雅地处理边缘情况。'#
- DEFAULT_SYSTEM_MESSAGE = '你是一个代码执行代理。你的角色是根据用户指令生成并执行 Python 代码，确保正确性、效率和最小的错误。优雅地处理边缘情况。'#
- DEFAULT_TERMINAL_DESCRIPTION = '一个计算机终端，除了运行 Python 脚本（以 ```python 代码块形式引用提供给它），或 sh shell 脚本（以 ```sh 代码块形式引用提供给它）外，不执行任何其他操作。'#
- NO_CODE_BLOCKS_FOUND_MESSAGE = '在线程中未找到代码块。请提供至少一个 Markdown 编码的代码块来执行（即，在 ```python 或 ```sh 代码块中引用代码）。'#
- classmethod _from_config(config: CodeExecutorAgentConfig) Self[source]#
从配置对象创建一个组件的新实例。
- 参数:
config (T) – 配置对象。
- 返回:
Self – 组件的新实例。
- component_config_schema#
CodeExecutorAgentConfig的别名
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.CodeExecutorAgent'#
覆盖组件的提供程序字符串。 这应用于防止内部模块名称成为模块名称的一部分。
- async execute_code_block(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) CodeResult[source]#
- async extract_code_blocks_from_messages(messages: Sequence[BaseChatMessage]) List[CodeBlock][source]#
- 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 on_reset(cancellation_token: CancellationToken) None[source]#
由于代码执行器代理没有可变状态，所以这是一个空操作。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代码执行器代理生成的message的类型。
- class MessageFilterAgent(name: str, wrapped_agent: BaseChatAgent, filter: MessageFilterConfig)[source]#
基类:
BaseChatAgent,Component[MessageFilterAgentConfig]一个包装代理，在将传入的消息传递给内部代理之前对其进行过滤。
警告
这是一个实验性功能，API 将在未来的版本中更改。
这在多代理工作流等场景中非常有用，在这种场景中，代理应该只处理完整消息历史的子集——例如，只有来自每个上游代理的最后一条消息，或者只有来自特定来源的第一条消息。
过滤使用
MessageFilterConfig进行配置，它支持： - 按消息来源进行过滤（例如，仅来自“user”或其他代理的消息） - 选择来自每个来源的前 N 条或后 N 条消息 - 如果 position 为 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", ... )
这将确保每个代理只看到其决策或行动逻辑所需的内容。
- classmethod _from_config(config: MessageFilterAgentConfig) MessageFilterAgent[source]#
从配置对象创建一个组件的新实例。
- 参数:
config (T) – 配置对象。
- 返回:
Self – 组件的新实例。
- component_config_schema#
别名为
MessageFilterAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.MessageFilterAgent'#
覆盖组件的提供程序字符串。 这应用于防止内部模块名称成为模块名称的一部分。
- 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]#
将代理重置为其初始化状态。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- 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] [必需]#
- 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)
- 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, 可选) – 使用内部团队的消息生成响应时要使用的指令。默认为
DEFAULT_INSTRUCTION。 它承担 'system' 的角色。response_prompt (str, 可选) – 使用内部团队的消息生成响应时要使用的响应提示。默认为
DEFAULT_RESPONSE_PROMPT。 它承担 'system' 的角色。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())
- DEFAULT_DESCRIPTION = 'An agent that uses an inner team of agents to generate responses.'#
SocietyOfMindAgent 的默认描述。
- 类型:
- DEFAULT_INSTRUCTION = 'Earlier you were asked to fulfill a request. You and your team worked diligently to address that request. Here is a transcript of that conversation:'#
使用内部团队的消息生成响应时要使用的默认指令。在使用模型生成响应时，该指令将添加到内部团队的消息之前。 它承担 'system' 的角色。
- 类型:
- DEFAULT_RESPONSE_PROMPT = 'Output a standalone response to the original request, without mentioning any of the intermediate discussion.'#
使用内部团队的消息生成响应时要使用的默认响应提示。 它承担 'system' 的角色。
- 类型:
- classmethod _from_config(config: SocietyOfMindAgentConfig) Self[source]#
从配置对象创建一个组件的新实例。
- 参数:
config (T) – 配置对象。
- 返回:
Self – 组件的新实例。
- component_config_schema#
别名
SocietyOfMindAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.SocietyOfMindAgent'#
覆盖组件的提供程序字符串。 这应用于防止内部模块名称成为模块名称的一部分。
- 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]#
将代理重置为其初始化状态。
- property produced_message_types: Sequence[type[BaseChatMessage]]#
代理在
Response.chat_message字段中生成的消息类型。它们必须是BaseChatMessage类型。
- class UserProxyAgent(name: str, *, description: str = 'A human user', 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}")
- classmethod _from_config(config: UserProxyAgentConfig) Self[source]#
从配置对象创建一个组件的新实例。
- 参数:
config (T) – 配置对象。
- 返回:
Self – 组件的新实例。
- component_config_schema#
alias of
UserProxyAgentConfig
- component_provider_override: ClassVar[str | None] = 'autogen_agentchat.agents.UserProxyAgent'#
覆盖组件的提供程序字符串。 这应用于防止内部模块名称成为模块名称的一部分。
- component_type: ClassVar[ComponentType] = 'agent'#
组件的逻辑类型。
- 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]#
Reset agent state.
- property produced_message_types: Sequence[type[BaseChatMessage]]#
Message types this agent can produce.