autogen_agentchat.代理#

此模块初始化了由包提供的各种预定义代理。 BaseChatAgent是AgentChat中所有代理的基类。

class AssistantAgent(name: str, model_client: ChatCompletionClient, *, tools: 列表[BaseTool[任何, 任何] | Callable[[...], 任何] | Callable[[...], Awaitable[任何]]] | 无 = None, handoffs: 列表[切换 | str] | 无 = None, model_context: ChatCompletionContext | 无 = None, description: str = 'An agent that provides assistance with ability to use tools.', system_message: str | 无 = '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 = False, tool_call_summary_format: str = '{result}', memory: Sequence[内存] | 无 = None)[源代码]#

基础类: BaseChatAgent, Component[AssistantAgentConfig]

一个提供工具使用协助的代理。

on_messages() 返回一个 Response，其中 chat_message 是最终的响应消息。

on_messages_stream() 创建一个异步生成器，它在内部消息创建时生成这些消息，并在关闭生成器之前将 Response 对象作为最后一项生成。

注意

调用者必须在每次调用时只传递新消息给代理 方法 on_messages() 或 on_messages_stream()。 代理在这些方法调用之间保持其状态。 不要在每次调用时将整个对话历史传递给代理。

警告

助手代理不是线程安全或协程安全的。它不应在多个任务或协程之间共享，也不应同时调用其方法。

下图展示了助理代理的工作方式：

../../_images/assistant-agent.svg

工具调用行为：

  • 如果模型没有返回工具调用，那么响应将立即作为TextMessage在chat_message中返回。

  • When the model returns tool calls, they will be executed right away:

    • 当reflect_on_tool_use为False（默认值）时，工具调用结果将作为ToolCallSummaryMessage返回到chat_message中。tool_call_summary_format可用于自定义工具调用摘要。

    • 当reflect_on_tool_use为True时，使用工具调用和结果进行另一个模型推理，并将文本响应作为TextMessage返回在chat_message中。

  • 如果模型返回多个工具调用，它们将并发执行。要禁用并行工具调用，你需要配置模型客户端。例如，为 OpenAIChatCompletionClient 和 AzureOpenAIChatCompletionClient 设置 parallel_tool_calls=False。

提示

默认情况下，当进行工具调用时，工具调用的结果会作为响应返回。因此，建议注意工具返回值的格式，特别是如果另一个代理期望它们以特定格式返回。如果需要，使用tool_call_summary_format来自定义工具调用摘要。

移交行为：

  • 如果触发了交接，一个HandoffMessage将会在chat_message中返回。

  • 如果有工具调用，它们也会在返回交接之前立即执行。

  • 工具调用和结果通过context传递给目标代理。

注意

如果检测到多次切换，仅执行第一次切换。 为避免这种情况，请在模型客户端配置中禁用并行工具调用。

限制发送到模型的上下文大小：

您可以通过将model_context参数设置为BufferedChatCompletionContext来限制发送给模型的消息数量。这将限制发送给模型的最近消息数量，当模型处理令牌数量有限时，这可能非常有用。您还可以通过子类化ChatCompletionContext来创建自己的模型上下文。

流式模式：

助手代理可以通过设置model_client_stream=True以流模式使用。在此模式下，on_messages_stream()和BaseChatAgent.run_stream()方法也会在模型客户端生成响应块时产生ModelClientStreamingChunkEvent消息。这些块消息不会包含在最终响应的内部消息中。

Parameters:

  • name (str) – 代理的名称。

  • model_client (ChatCompletionClient) – 用于推理的模型客户端。

  • 工具 (列表[BaseTool[Any, Any] | Callable[..., Any] | Callable[..., Awaitable[Any]]] | 无, 可选) – 注册到代理的工具。

  • 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，工具调用结果将作为响应返回。默认为 False。

  • tool_call_summary_format (str, optional) – 用于为每个工具调用结果创建工具调用摘要的格式字符串。 默认为“{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。

示例

示例1：基础代理

以下示例展示了如何使用模型客户端创建一个助手代理，并生成对简单任务的响应。

import asyncio
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage


async def main() -> None:
    model_client = OpenAIChatCompletionClient(
        model="gpt-4o",
        # api_key = "your_openai_api_key"
    )
    agent = AssistantAgent(name="assistant", model_client=model_client)

    response = await agent.on_messages(
        [TextMessage(content="What is the capital of France?", source="user")], CancellationToken()
    )
    print(response)


asyncio.run(main())

示例2: 模型客户端令牌流

本示例展示了如何通过设置model_client_stream=True来创建一个带有模型客户端的助手代理，并生成一个令牌流。

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken


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.on_messages_stream(
        [TextMessage(content="Name two cities in North America.", source="user")], CancellationToken()
    )
    async for message in stream:
        print(message)


asyncio.run(main())

source='assistant' models_usage=None content='Two' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' cities' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' in' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' North' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' America' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' are' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' New' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' York' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' City' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' in' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' the' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' United' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' States' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' and' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' Toronto' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' in' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' Canada' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content='.' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content=' TERMIN' type='ModelClientStreamingChunkEvent'
source='assistant' models_usage=None content='ATE' type='ModelClientStreamingChunkEvent'
Response(chat_message=TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=0, completion_tokens=0), content='Two cities in North America are New York City in the United States and Toronto in Canada. TERMINATE', type='TextMessage'), inner_messages=[])

示例 3: 带有工具的代理

以下示例展示了如何创建一个带有模型客户端和工具的助手代理，为任务生成消息流，并使用 Console 将消息打印到控制台。

该工具是一个简单的函数，返回当前时间。 在内部，该函数被包装在FunctionTool 中，并与代理的模型客户端一起使用。函数的文档字符串被用作工具描述， 函数名称被用作工具名称， 包括类型提示的函数签名被用作工具参数。

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken


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.on_messages_stream(
            [TextMessage(content="What is the current time?", source="user")], CancellationToken()
        )
    )


asyncio.run(main())

示例4：具有结构化输出和工具的代理

以下示例演示了如何创建一个配置为使用结构化输出和工具的模型客户端的助手代理。 请注意，您需要使用FunctionTool来创建该工具， 并且在结构化输出模式下需要strict=True。 由于模型配置为使用结构化输出，反射响应将是一个JSON格式的字符串。

import asyncio
from typing import Literal

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
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 uses the structured output format.
model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    response_format=AgentResponse,  # type: ignore
)

# 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.",
    reflect_on_tool_use=True,  # Use reflection to have the agent generate a formatted response.
)


async def main() -> None:
    stream = agent.on_messages_stream([TextMessage(content="I am happy today!", source="user")], CancellationToken())
    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"}

示例 5：具有有限模型上下文的代理

以下示例展示了如何使用一个 BufferedChatCompletionContext ，该上下文仅保留最后两条消息（1条用户消息 + 1条助手消息）。 当模型对可以处理的令牌数量有限制时，有界的模型上下文非常有用。

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
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.",
    )

    response = await agent.on_messages(
        [TextMessage(content="Name two cities in North America.", source="user")], CancellationToken()
    )
    print(response.chat_message.content)  # type: ignore

    response = await agent.on_messages(
        [TextMessage(content="My favorite color is blue.", source="user")], CancellationToken()
    )
    print(response.chat_message.content)  # type: ignore

    response = await agent.on_messages(
        [TextMessage(content="Did I ask you any question?", source="user")], CancellationToken()
    )
    print(response.chat_message.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!

示例6：具有记忆的agent

以下示例展示了如何与助手代理一起使用基于列表的内存。 内存中预加载了一些初始内容。 在底层，内存用于在进行推理之前更新模型上下文，使用update_context()方法。

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
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.",
    )

    response = await agent.on_messages(
        [TextMessage(content="One idea for a dinner.", source="user")], CancellationToken()
    )
    print(response.chat_message.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!

示例 7: 使用 `o1-mini` 的代理

以下示例展示了如何使用o1-mini模型与助手代理。

import asyncio
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage


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)

    response = await agent.on_messages(
        [TextMessage(content="What is the capital of France?", source="user")], CancellationToken()
    )
    print(response)


asyncio.run(main())

注意

o1-preview 和 o1-mini 模型不支持系统消息和函数调用。 因此，system_message 应设置为 None，并且不应设置 tools 和 handoffs。 更多详情请参见 o1 beta limitations。

示例 8：使用自定义模型上下文的推理模型的代理。

下面的示例展示了如何在助手代理中使用推理模型（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,
    },
)

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 | 无] = 'autogen_agentchat.agents.AssistantAgent'#

覆盖组件的提供商字符串。这应用于防止内部模块名称成为模块名称的一部分。

async load_state(state: 映射[str, 任何]) → 无[源代码]#

加载助手代理的状态

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → 响应[源代码]#

处理传入的消息并返回响应。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | 响应, 无][源代码]#

使用助理代理处理传入的消息，并在事件/响应发生时生成它们。

async on_reset(cancellation_token: CancellationToken) → 无[源代码]#

将助手代理重置为其初始化状态。

property produced_message_types: Sequence[类型[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#

代理在Response.chat_message字段中生成的消息类型。它们必须是ChatMessage类型。

async save_state() → 映射[str, 任何][源代码]#

保存助手代理的当前状态。

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()方法时仅将新消息传递给代理。不要在每次调用时将整个对话历史记录传递给代理。这一设计原则在创建新代理时必须遵循。

async close() → 无[源代码]#

释放代理持有的所有资源。默认情况下，在BaseChatAgent类中这是一个无操作。子类可以重写此方法以实现自定义的关闭行为。

component_type: ClassVar[ComponentType] = 'agent'#

组件的逻辑类型。

property description: str#

代理的描述。团队使用此描述来决定使用哪些代理。描述应说明代理的能力以及如何与其互动。

async load_state(state: 映射[str, 任何]) → 无[源代码]#

从保存的状态恢复代理。无状态代理的默认实现。

property name: str#

代理的名称。团队使用此名称来唯一标识代理。它在团队内应该是唯一的。

abstract async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → 响应[源代码]#

处理传入的消息并返回响应。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | 响应, 无][源代码]#

处理传入的消息并返回消息流，最终项是响应。在 BaseChatAgent 中的基本实现只是调用 on_messages() 并生成响应中的消息。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_pause(cancellation_token: CancellationToken) → 无[源代码]#

当代理在运行其on_messages()或on_messages_stream()方法时暂停时调用。在BaseChatAgent类中，默认情况下这是一个空操作。子类可以重写此方法以实现自定义的暂停行为。

abstract async on_reset(cancellation_token: CancellationToken) → 无[源代码]#

将代理重置为其初始化状态。

async on_resume(cancellation_token: CancellationToken) → 无[源代码]#

当代理在其on_messages()或on_messages_stream()方法中从暂停状态恢复时调用。 在BaseChatAgent类中，默认情况下此方法不执行任何操作。 子类可以重写此方法以实现自定义的恢复行为。

abstract property produced_message_types: Sequence[类型[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#

代理在Response.chat_message字段中生成的消息类型。它们必须是ChatMessage类型。

async run(*, task: str | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | 无 = None, cancellation_token: CancellationToken | 无 = None) → 任务结果[源代码]#

运行具有给定任务的代理并返回结果。

async run_stream(*, task: str | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] | 无 = None, cancellation_token: CancellationToken | 无 = None) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | TaskResult, 无][源代码]#

使用给定的任务运行代理，并返回一系列消息，最后一项为最终任务结果。

async save_state() → 映射[str, 任何][源代码]#

导出状态。无状态代理的默认实现。

class CodeExecutorAgent(name: str, code_executor: CodeExecutor, *, description: str = 'A computer terminal that performs no other action than running Python scripts (provided to it quoted in ```python code blocks), or sh shell scripts (provided to it quoted in ```sh code blocks).', sources: Sequence[str] | 无 = None)[源代码]#

基础类: BaseChatAgent, Component[CodeExecutorAgentConfig]

一个提取并执行接收消息中找到的代码片段并返回输出的代理。

它通常在团队中与另一个生成要执行的代码片段的代理一起使用。

注意

考虑将PythonCodeExecutionTool作为此代理的替代方案。该工具允许在单个代理内执行Python代码，而不是将其发送到单独的代理执行。但是，代理的模型必须生成正确转义的代码字符串作为工具的参数。

Parameters:

  • name – 代理的名称。

  • code_executor – 负责执行消息中接收到的代码的CodeExecutor（推荐使用DockerCommandLineCodeExecutor。见下面的示例）

  • description (可选) – 代理的描述。

  • sources (可选的) – 仅检查来自指定代理的消息以执行代码。

注意

建议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())
classmethod _from_config(config: CodeExecutorAgentConfig) → 自我[源代码]#

从配置对象创建组件的新实例。

Parameters:

config (T) – 配置对象。

Returns:

Self – 组件的新实例。

_to_config() → CodeExecutorAgentConfig[源代码]#

导出配置，该配置将用于创建一个与此实例配置相匹配的组件新实例。

Returns:

T – 组件的配置。

component_config_schema#

CodeExecutorAgentConfig 的别名

component_provider_override: ClassVar[str | 无] = 'autogen_agentchat.agents.CodeExecutorAgent'#

覆盖组件的提供商字符串。这应用于防止内部模块名称成为模块名称的一部分。

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → 响应[源代码]#

处理传入的消息并返回响应。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_reset(cancellation_token: CancellationToken) → 无[源代码]#

这是一个无操作，因为代码执行代理没有可变状态。

property produced_message_types: Sequence[类型[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#

代码执行器代理生成的消息类型。

class SocietyOfMindAgent(name: str, team: 团队, model_client: ChatCompletionClient, *, description: str = DEFAULT_DESCRIPTION, instruction: str = DEFAULT_INSTRUCTION, response_prompt: str = DEFAULT_RESPONSE_PROMPT)[源代码]#

基础类：BaseChatAgent, Component[SocietyOfMindAgentConfig]

一个使用内部代理团队生成响应的代理。

每次调用代理的on_messages()或on_messages_stream()方法时，它会运行内部代理团队，然后使用模型客户端根据内部团队的消息生成响应。生成响应后，代理通过调用Team.reset()来重置内部团队。

Parameters:

  • name (str) – 代理的名称。

  • 团队 (Team) – 使用的代理团队。

  • model_client (ChatCompletionClient) – 用于准备响应的模型客户端。

  • description (str, optional) – 代理的描述。

  • instruction (str, optional) – 当使用内部团队的消息生成响应时使用的指令。默认为 DEFAULT_INSTRUCTION。它扮演‘系统’的角色。

  • response_prompt (str, 可选) – 使用内部团队的消息生成响应时使用的响应提示。默认为 DEFAULT_RESPONSE_PROMPT。它扮演‘system’的角色。

示例：

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 的默认描述。

Type:

str

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:'#

在使用内部团队的消息生成响应时使用的默认指令。该指令在使用模型生成响应时将被添加到内部团队的消息前。它承担了“系统”的角色。

Type:

str

DEFAULT_RESPONSE_PROMPT = 'Output a standalone response to the original request, without mentioning any of the intermediate discussion.'#

使用内部团队消息生成响应时使用的默认响应提示。它扮演‘system’角色。

Type:

str

classmethod _from_config(config: SocietyOfMindAgentConfig) → 自我[源代码]#

从配置对象创建组件的新实例。

Parameters:

config (T) – 配置对象。

Returns:

Self – 组件的新实例。

_to_config() → SocietyOfMindAgentConfig[源代码]#

导出配置，该配置将用于创建一个与此实例配置相匹配的组件新实例。

Returns:

T – 组件的配置。

component_config_schema#

SocietyOfMindAgentConfig 的别名

component_provider_override: ClassVar[str | 无] = 'autogen_agentchat.agents.SocietyOfMindAgent'#

覆盖组件的提供商字符串。这应用于防止内部模块名称成为模块名称的一部分。

async load_state(state: 映射[str, 任何]) → 无[源代码]#

从保存的状态恢复代理。无状态代理的默认实现。

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → 响应[源代码]#

处理传入的消息并返回响应。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | 响应, 无][源代码]#

处理传入的消息并返回消息流，最终项是响应。在 BaseChatAgent 中的基本实现只是调用 on_messages() 并生成响应中的消息。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_reset(cancellation_token: CancellationToken) → 无[源代码]#

将代理重置为其初始化状态。

property produced_message_types: Sequence[类型[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#

代理在Response.chat_message字段中生成的消息类型。它们必须是ChatMessage类型。

async save_state() → 映射[str, 任何][源代码]#

导出状态。无状态代理的默认实现。

class UserProxyAgent(name: str, *, description: str = 'A human user', input_func: Callable[[str], str] | Callable[[str, CancellationToken | 无], Awaitable[str]] | 无 = None)[源代码]#

基础类：BaseChatAgent, Component[UserProxyAgentConfig]

一个可以通过输入函数代表人类用户的代理。

该代理可用于通过提供自定义输入函数来代表聊天系统中的用户。

注意

使用UserProxyAgent会将正在运行的团队置于临时阻塞状态，直到用户响应。因此，重要的是对用户输入函数进行超时处理，并使用CancellationToken在用户不响应时取消操作。输入函数还应处理异常，并在需要时返回默认响应。

对于涉及较慢人工响应的典型用例，建议使用终止条件，例如HandoffTermination或SourceMatchTermination来停止运行团队并将控制权返回给应用程序。您可以在用户输入后再次运行团队。这样，团队的状态可以在用户响应时保存和恢复。

查看人在回路中获取更多信息。

Parameters:

  • name (str) – 代理的名称。

  • description (str, optional) – 对该代理的描述。

  • input_func (Optional[Callable[[str], str]], Callable[[str, Optional[CancellationToken]], Awaitable[str]]) – 一个函数，接收一个提示并返回用户输入的字符串。

有关与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(),
        )
    )
    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
        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}")
class InputRequestContext[源代码]#

基础: object

classmethod request_id() → str[源代码]#
classmethod _from_config(config: UserProxyAgentConfig) → 自我[源代码]#

从配置对象创建组件的新实例。

Parameters:

config (T) – 配置对象。

Returns:

Self – 组件的新实例。

_to_config() → UserProxyAgentConfig[源代码]#

导出配置，该配置将用于创建一个与此实例配置相匹配的组件新实例。

Returns:

T – 组件的配置。

component_config_schema#

alias of UserProxyAgentConfig

component_provider_override: ClassVar[str | 无] = 'autogen_agentchat.agents.UserProxyAgent'#

覆盖组件的提供商字符串。这应用于防止内部模块名称成为模块名称的一部分。

component_type: ClassVar[ComponentType] = 'agent'#

组件的逻辑类型。

async on_messages(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → 响应[源代码]#

处理传入的消息并返回响应。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应该在调用此方法之间保持其状态。例如，如果代理需要记住之前的消息以响应当前消息，它应该将之前的消息存储在代理状态中。

async on_messages_stream(messages: Sequence[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | 响应, 无][源代码]#

通过请求用户输入来处理传入的消息。

async on_reset(cancellation_token: CancellationToken | 无 = None) → 无[源代码]#

重置代理状态。

property produced_message_types: Sequence[类型[Annotated[TextMessage | MultiModalMessage | StopMessage | 工具调用摘要信息 | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#

此代理可以生成的消息类型。

On this page
在GitHub上编辑
显示源代码