autogen_core#

class Agent(*args, **kwargs)[source]#

基类: Protocol

property metadata: AgentMetadata#

代理的元数据。

property id: AgentId#

代理的 ID。

async on_message(message: Any, ctx: MessageContext) → Any[source]#

代理的消息处理程序。这应该只由运行时调用，而不是由其他代理调用。

参数:

  • message (Any) – 收到的消息。类型是 subscriptions 中的类型之一。

  • ctx (MessageContext) – 消息的上下文。

返回:

Any – 对消息的响应。可以为 None。

引发:
async save_state() → Mapping[str, Any][source]#

保存代理的状态。结果必须是 JSON 可序列化的。

async load_state(state: Mapping[str, Any]) → None[source]#

加载从 save_state 获取的代理的状态。

参数:

state (Mapping[str, Any]) – 代理的状态。必须是 JSON 可序列化的。

async close() → None[source]#

运行时关闭时调用

class AgentId(type: str | AgentType, key: str)[source]#

基类: object

Agent ID 唯一标识代理运行时中的代理实例 - 包括分布式运行时。它是用于接收消息的代理实例的“地址”。

有关更多信息，请参见此处： Agent Identity and Lifecycle

classmethod from_str(agent_id: str) → Self[source]#

将格式为 type/key 的字符串转换为 AgentId

property type: str#

一个将代理与特定工厂函数关联起来的标识符。

字符串只能由字母数字（a-z）和（0-9）或下划线（_）组成。

属性 key: str#

Agent 实例标识符。

字符串只能由字母数字（a-z）和（0-9）或下划线（_）组成。

类 AgentProxy(agent: AgentId, runtime: AgentRuntime)[源代码]#

基类: object

一个辅助类，允许您使用 AgentId 代替其关联的 Agent

属性 id: AgentId#

此代理的目标 Agent

属性 metadata: Awaitable[AgentMetadata]#

代理的元数据。

async send_message(message: Any, *, sender: AgentId, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any[源代码]#
async save_state() → Mapping[str, Any][源代码]#

保存代理的状态。结果必须是 JSON 可序列化的。

async load_state(state: Mapping[str, Any]) → None[源代码]#

加载从 save_state 获取的代理的状态。

参数:

state (Mapping[str, Any]) – 代理的状态。必须是 JSON 可序列化的。

类 AgentMetadata[源代码]#

基类: TypedDict

type: str#
key: str#
description: str#
类 AgentRuntime(*args, **kwargs)[源代码]#

基类: Protocol

async send_message(message: Any, recipient: AgentId, *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any[源代码]#

向 Agent 发送消息并获取响应。

参数:

  • message (Any) – 要发送的消息。

  • recipient (AgentId) – 要将消息发送到的 Agent。

  • sender (AgentId | None, 可选) – 发送消息的 Agent。 仅当从没有 Agent 发送（例如直接从外部发送到运行时）时才应为 None。 默认为 None。

  • cancellation_token (CancellationToken | None, 可选) – 用于取消正在进行的 。 默认为 None。

引发:
返回:

Any – 来自 Agent 的响应。

async publish_message(message: Any, topic_id: TopicId, *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → None[source]#

将消息发布到给定命名空间中的所有代理，如果未提供命名空间，则发布到发送者的命名空间。

发布操作不应期待任何响应。

参数:

  • message (Any) – 要发布的消息。

  • topic_id (TopicId) – 要将消息发布到的主题。

  • sender (AgentId | None, 可选) – 发送消息的代理。默认为 None。

  • cancellation_token (CancellationToken | None, 可选) – 用于取消正在进行中的操作的令牌。默认为 None。

  • message_id (str | None, 可选) – 消息 ID。 如果为 None，则将生成新的消息 ID。默认为 None。 此消息 ID 必须是唯一的，建议使用 UUID。

引发:

UndeliverableException – 如果消息无法传递。

async register_factory(type: str | AgentType, agent_factory: Callable[[], T | Awaitable[T]], *, expected_class: type[T] | None = None) → AgentType[source]#

向与特定类型关联的运行时注册一个代理工厂。类型必须是唯一的。 此 API 不添加任何订阅。

注意

这是一个低级 API，通常应该使用代理类的 register 方法，因为它也会自动处理订阅。

例子

from dataclasses import dataclass

from autogen_core import AgentRuntime, MessageContext, RoutedAgent, event
from autogen_core.models import UserMessage


@dataclass
class MyMessage:
    content: str


class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("My core agent")

    @event
    async def handler(self, message: UserMessage, context: MessageContext) -> None:
        print("Event received: ", message.content)


async def my_agent_factory():
    return MyAgent()


async def main() -> None:
    runtime: AgentRuntime = ...  # type: ignore
    await runtime.register_factory("my_agent", lambda: MyAgent())


import asyncio

asyncio.run(main())
参数:

  • type (str) – 此工厂创建的代理的类型。它与代理类名不同。 type 参数用于区分不同的工厂函数，而不是代理类。

  • agent_factory (Callable[[], T]) – 创建代理的工厂，其中 T 是一个具体的 Agent 类型。 在工厂内部，使用 autogen_core.AgentInstantiationContext 访问当前运行时和代理 ID 等变量。

  • expected_class (type[T] | None, 可选) – 代理的预期类，用于运行时的工厂验证。默认为 None。 如果为 None，则不执行任何验证。

async try_get_underlying_agent_instance(id: AgentId, type: Type[T] = Agent) → T[source]#

尝试按名称和命名空间获取底层代理实例。 通常不鼓励这样做（因此名称很长），但在某些情况下可能有用。

如果无法访问底层代理，这将引发异常。

参数:

  • id (AgentId) – 代理 ID。

  • type (Type[T], 可选) – 代理的预期类型。 默认为 Agent。

返回:

T – 具体代理实例。

引发:
async get(id: AgentId, /, *, lazy: bool = True) → AgentId[source]#
async get(type: AgentType | str, /, key: str = 'default', *, lazy: bool = True) → AgentId
async save_state() → Mapping[str, Any][source]#

保存整个运行时的状态，包括所有托管的代理。 恢复状态的唯一方法是将其传递给 load_state()。

状态的结构是实现定义的，可以是任何JSON可序列化的对象。

返回:

Mapping[str, Any] – 保存的状态。

async load_state(state: Mapping[str, Any]) → None[source]#

加载整个运行时的状态，包括所有托管的代理。 该状态应与 save_state() 返回的状态相同。

参数:

state (Mapping[str, Any]) – 保存的状态。

async agent_metadata(agent: AgentId) → AgentMetadata[source]#

获取代理的元数据。

参数:

agent (AgentId) – 代理ID。

返回:

AgentMetadata – 代理元数据。

async agent_save_state(agent: AgentId) → Mapping[str, Any][source]#

保存单个代理的状态。

状态的结构是实现定义的，可以是任何JSON可序列化的对象。

参数:

agent (AgentId) – 代理ID。

返回:

Mapping[str, Any] – 保存的状态。

async agent_load_state(agent: AgentId, state: Mapping[str, Any]) → None[source]#

加载单个代理的状态。

参数:

  • agent (AgentId) – 代理ID。

  • state (Mapping[str, Any]) – 保存的状态。

async add_subscription(subscription: Subscription) → None[source]#

添加一个新的订阅，运行时应在处理发布的消息时履行该订阅

参数:

subscription (Subscription) – 要添加的订阅

async remove_subscription(id: str) → None[source]#

从运行时删除订阅

参数:

id (str) – 要删除的订阅的ID

引发:

LookupError – 如果订阅不存在

add_message_serializer(serializer: MessageSerializer[Any] | Sequence[MessageSerializer[Any]]) → None[source]#

向运行时添加新的消息序列化器

注意：这将根据 type_name 和 data_content_type 属性删除重复的序列化器

参数:

serializer (MessageSerializer[Any] | Sequence[MessageSerializer[Any]]) – 要添加的序列化器

class BaseAgent(description: str)[source]#

基类： ABC, Agent

property metadata: AgentMetadata#

代理的元数据。

property type: str#
property id: AgentId#

代理的 ID。

property runtime: AgentRuntime#
final async on_message(message: Any, ctx: MessageContext) → Any[source]#

代理的消息处理程序。这应该只由运行时调用，而不是由其他代理调用。

参数:

  • message (Any) – 收到的消息。类型是 subscriptions 中的类型之一。

  • ctx (MessageContext) – 消息的上下文。

返回:

Any – 对消息的响应。可以为 None。

引发:
abstract async on_message_impl(message: Any, ctx: MessageContext) → Any[source]#
async send_message(message: Any, recipient: AgentId, *, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any[source]#

更多信息请参见 autogen_core.AgentRuntime.send_message()。

async publish_message(message: Any, topic_id: TopicId, *, cancellation_token: CancellationToken | None = None) → None[source]#
async save_state() → Mapping[str, Any][source]#

保存代理的状态。结果必须是 JSON 可序列化的。

async load_state(state: Mapping[str, Any]) → None[source]#

加载从 save_state 获取的代理的状态。

参数:

state (Mapping[str, Any]) – 代理的状态。必须是 JSON 可序列化的。

async close() → None[source]#

运行时关闭时调用

async classmethod register(runtime: AgentRuntime, type: str, factory: Callable[[], Self | Awaitable[Self]], *, skip_class_subscriptions: bool = False, skip_direct_message_subscription: bool = False) → AgentType[source]#

注册 ABC 的一个虚拟子类。

返回子类，以允许用作类装饰器。

class CacheStore[source]#

基类： ABC, Generic[T], ComponentBase[BaseModel]

该协议定义了存储/缓存操作的基本接口。

子类应处理底层存储的生命周期。

component_type: ClassVar[ComponentType] = 'cache_store'#

组件的逻辑类型。

abstract get(key: str, default: T | None = None) → T | None[source]#

从存储中检索一个项目。

参数:

  • key – 标识存储中项目的键。

  • default (可选) – 如果未找到键，则返回的默认值。默认为 None。

返回:

如果找到，则返回与键关联的值，否则返回默认值。

abstract set(key: str, value: T) → None[source]#

在存储中设置一个项目。

参数:

  • key – 将项目存储在其下的键。

  • value – 要存储在存储中的值。

class InMemoryStore[source]#

基类：CacheStore[T], Component[InMemoryStoreConfig]

component_provider_override: ClassVar[str | None] = 'autogen_core.InMemoryStore'#

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

component_config_schema#

InMemoryStoreConfig 的别名

get(key: str, default: T | None = None) → T | None[source]#

从存储中检索一个项目。

参数:

  • key – 标识存储中项目的键。

  • default (可选) – 如果未找到键，则返回的默认值。默认为 None。

返回:

如果找到，则返回与键关联的值，否则返回默认值。

set(key: str, value: T) → None[source]#

在存储中设置一个项目。

参数:

  • key – 将项目存储在其下的键。

  • value – 要存储在存储中的值。

_to_config() → InMemoryStoreConfig[source]#

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: InMemoryStoreConfig) → Self[source]#

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class CancellationToken[source]#

基类: object

用于取消挂起的异步调用的令牌

cancel() → None[source]#

取消链接到此取消令牌的挂起异步调用。

is_cancelled() → bool[source]#

检查是否已使用 CancellationToken

add_callback(callback: Callable[[], None]) → None[source]#

附加一个回调函数，该函数将在调用 cancel 时被调用

将挂起的异步调用链接到令牌以允许取消

class AgentInstantiationContext[source]#

基类: object

一个静态类，为代理实例化提供上下文。

这个静态类可用于在代理实例化期间（在工厂函数或代理的类构造函数中）访问当前运行时和代理 ID。

例子

在工厂函数和代理的构造函数中获取当前运行时和代理 ID

import asyncio
from dataclasses import dataclass

from autogen_core import (
    AgentId,
    AgentInstantiationContext,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    message_handler,
)


@dataclass
class TestMessage:
    content: str


class TestAgent(RoutedAgent):
    def __init__(self, description: str):
        super().__init__(description)
        # Get the current runtime -- we don't use it here, but it's available.
        _ = AgentInstantiationContext.current_runtime()
        # Get the current agent ID.
        agent_id = AgentInstantiationContext.current_agent_id()
        print(f"Current AgentID from constructor: {agent_id}")

    @message_handler
    async def handle_test_message(self, message: TestMessage, ctx: MessageContext) -> None:
        print(f"Received message: {message.content}")


def test_agent_factory() -> TestAgent:
    # Get the current runtime -- we don't use it here, but it's available.
    _ = AgentInstantiationContext.current_runtime()
    # Get the current agent ID.
    agent_id = AgentInstantiationContext.current_agent_id()
    print(f"Current AgentID from factory: {agent_id}")
    return TestAgent(description="Test agent")


async def main() -> None:
    # Create a SingleThreadedAgentRuntime instance.
    runtime = SingleThreadedAgentRuntime()

    # Start the runtime.
    runtime.start()

    # Register the agent type with a factory function.
    await runtime.register_factory("test_agent", test_agent_factory)

    # Send a message to the agent. The runtime will instantiate the agent and call the message handler.
    await runtime.send_message(TestMessage(content="Hello, world!"), AgentId("test_agent", "default"))

    # Stop the runtime.
    await runtime.stop()


asyncio.run(main())
classmethod current_runtime() → AgentRuntime
classmethod current_agent_id() → AgentId
class TopicId(type: str, source: str)

基类: object

TopicId 定义了广播消息的作用域。本质上，agent runtime 通过其广播 API 实现了一个发布-订阅模型：发布消息时，必须指定主题。

有关更多信息，请参见此处： Topic

type: str

此 topic_id 包含的事件类型。遵循云事件规范。

必须匹配模式：^[w-.:=]+Z

在此处了解更多信息： cloudevents/spec

source: str

标识事件发生的上下文。遵循云事件规范。

在此处了解更多信息： cloudevents/spec

classmethod from_str(topic_id: str) → Self

将 type/source 格式的字符串转换为 TopicId

class Subscription(*args, **kwargs)

基类: Protocol

订阅定义了 agent 感兴趣的主题。

property id: str

获取订阅的 ID。

实现应返回订阅的唯一 ID。通常这是一个 UUID。

返回:

str – 订阅的 ID。

is_match(topic_id: TopicId) → bool

检查给定的 topic_id 是否与订阅匹配。

参数:

topic_id (TopicId) – 要检查的 TopicId。

返回:

bool – 如果 topic_id 与订阅匹配，则为 True；否则为 False。

map_to_agent(topic_id: TopicId) → AgentId

将 topic_id 映射到 agent。 只有在给定的 topic_id 的 is_match 返回 True 时才应调用。

参数:

topic_id (TopicId) – 要映射的 TopicId。

返回:

AgentId – 应该处理 topic_id 的 agent 的 ID。

引发:

CantHandleException – 如果订阅无法处理 topic_id。

class MessageContext(sender: AgentId | None, topic_id: TopicId | None, is_rpc: bool, cancellation_token: CancellationToken, message_id: str)

基类: object

sender: AgentId | None
topic_id: TopicId | None
is_rpc: bool
cancellation_token: CancellationToken
message_id: str
class AgentType(type: str)

基类: object

type: str

此代理类型的字符串表示形式。

class SubscriptionInstantiationContext

基类: object

classmethod agent_type() → AgentType[source]#
class MessageHandlerContext[source]#

基类: object

classmethod agent_id() → AgentId[source]#
class MessageSerializer(*args, **kwargs)[source]#

基类: Protocol[T]

property data_content_type: str#
property type_name: str#
deserialize(payload: bytes) → T[source]#
serialize(message: T) → bytes[source]#
class UnknownPayload(type_name: str, data_content_type: str, payload: bytes)[source]#

基类: object

type_name: str#
data_content_type: str#
payload: bytes#
class Image(image: Image)[source]#

基类: object

表示一张图片。

例子

从 URL 加载图片

from autogen_core import Image
from PIL import Image as PILImage
import aiohttp
import asyncio


async def from_url(url: str) -> Image:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            content = await response.read()
            return Image.from_pil(PILImage.open(content))


image = asyncio.run(from_url("https://example.com/image"))
classmethod from_pil(pil_image: Image) → Image[source]#
classmethod from_uri(uri: str) → Image[source]#
classmethod from_base64(base64_str: str) → Image[source]#
to_base64() → str[source]#
classmethod from_file(file_path: Path) → Image[source]#
property data_uri: str#
to_openai_format(detail: Literal['auto', 'low', 'high'] = 'auto') → Dict[str, Any][source]#
class RoutedAgent(description: str)[source]#

基类： BaseAgent

代理的基础类，它根据消息类型和可选的匹配函数将消息路由到处理程序。

要创建路由代理，请继承此类并将消息处理程序添加为使用 event() 或 rpc() 装饰器装饰的方法。

例子

from dataclasses import dataclass
from autogen_core import MessageContext
from autogen_core import RoutedAgent, event, rpc


@dataclass
class Message:
    pass


@dataclass
class MessageWithContent:
    content: str


@dataclass
class Response:
    pass


class MyAgent(RoutedAgent):
    def __init__(self):
        super().__init__("MyAgent")

    @event
    async def handle_event_message(self, message: Message, ctx: MessageContext) -> None:
        assert ctx.topic_id is not None
        await self.publish_message(MessageWithContent("event handled"), ctx.topic_id)

    @rpc(match=lambda message, ctx: message.content == "special")  # type: ignore
    async def handle_special_rpc_message(self, message: MessageWithContent, ctx: MessageContext) -> Response:
        return Response()
async on_message_impl(message: Any, ctx: MessageContext) → Any | None[source]#

通过将消息路由到相应的消息处理程序来处理消息。不要在子类中覆盖此方法。而是将消息处理程序添加为使用 event() 或 rpc() 装饰器装饰的方法。

async on_unhandled_message(message: Any, ctx: MessageContext) → None[source]#

当收到没有匹配消息处理程序的消息时调用。默认实现记录一条信息消息。

class ClosureAgent(description: str, closure: Callable[[ClosureContext, T, MessageContext], Awaitable[Any]], *, unknown_type_policy: Literal['error', 'warn', 'ignore'] = 'warn')[source]#

基类： BaseAgent, ClosureContext

property metadata: AgentMetadata#

代理的元数据。

property id: AgentId#

代理的 ID。

property runtime: AgentRuntime#
async on_message_impl(message: Any, ctx: MessageContext) → Any[source]#
async save_state() → Mapping[str, Any][source]#

闭包代理没有状态。因此，此方法始终返回一个空字典。

async load_state(state: Mapping[str, Any]) → None[source]#

闭包代理没有状态。因此，此方法不执行任何操作。

async classmethod register_closure(runtime: AgentRuntime, type: str, closure: Callable[[ClosureContext, T, MessageContext], Awaitable[Any]], *, unknown_type_policy: Literal['error', 'warn', 'ignore'] = 'warn', skip_direct_message_subscription: bool = False, description: str = '', subscriptions: Callable[[], list[Subscription] | Awaitable[list[Subscription]]] | None = None) → AgentType[source]#

闭包代理允许您使用闭包（或函数）定义代理，而无需定义类。它允许从运行时提取值。

闭包可以定义期望的消息类型，或者可以使用 Any 接受任何类型的消息。

例子

import asyncio
from autogen_core import SingleThreadedAgentRuntime, MessageContext, ClosureAgent, ClosureContext
from dataclasses import dataclass

from autogen_core._default_subscription import DefaultSubscription
from autogen_core._default_topic import DefaultTopicId


@dataclass
class MyMessage:
    content: str


async def main():
    queue = asyncio.Queue[MyMessage]()

    async def output_result(_ctx: ClosureContext, message: MyMessage, ctx: MessageContext) -> None:
        await queue.put(message)

    runtime = SingleThreadedAgentRuntime()
    await ClosureAgent.register_closure(
        runtime, "output_result", output_result, subscriptions=lambda: [DefaultSubscription()]
    )

    runtime.start()
    await runtime.publish_message(MyMessage("Hello, world!"), DefaultTopicId())
    await runtime.stop_when_idle()

    result = await queue.get()
    print(result)


asyncio.run(main())
参数:

  • runtime (AgentRuntime) – 注册代理的运行时

  • type (str) – 注册代理的代理类型

  • closure (Callable[[ClosureContext, T, MessageContext], Awaitable[Any]]) – 处理消息的闭包

  • unknown_type_policy (Literal["error", "warn", "ignore"], optional) – 如果遇到与闭包类型不匹配的类型，该怎么做。默认为“warn”。

  • skip_direct_message_subscription (bool, optional) – 不为此代理添加直接消息订阅。默认为 False。

  • description (str, optional) – 代理的作用的描述。默认为“”。

  • subscriptions (Callable[[], list[Subscription] | Awaitable[list[Subscription]]] | None, optional) – 此闭包代理的订阅列表。默认为 None。

返回:

AgentType – 注册的代理的类型

class ClosureContext(*args, **kwargs)[source]#

基类: Protocol

property id: AgentId#
async send_message(message: Any, recipient: AgentId, *, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any[source]#
async publish_message(message: Any, topic_id: TopicId, *, cancellation_token: CancellationToken | None = None) → None[source]#
message_handler(func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]] = None, *, strict: bool = True, match: None | Callable[[ReceivesT, MessageContext], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]]], MessageHandler[AgentT, ReceivesT, ProducesT]] | MessageHandler[AgentT, ReceivesT, ProducesT][source]#

通用消息处理程序的装饰器。

将此装饰器添加到 RoutedAgent 类中旨在处理事件和 RPC 消息的方法。这些方法必须具有特定的签名，才能使其有效。

  • 该方法必须是一个 async 方法。

  • 该方法必须使用 @message_handler 装饰器进行装饰。

  • 该方法必须恰好有 3 个参数

    1. self

    2. message：要处理的消息，必须使用它要处理的消息类型进行类型提示。

    3. ctx：一个 autogen_core.MessageContext 对象。

  • 该方法必须进行类型提示，说明它可以作为响应返回哪些消息类型，如果它不返回任何内容，则可以返回 None。

处理程序可以通过接受消息类型的 Union 来处理多个消息类型。它还可以通过返回消息类型的 Union 来返回多个消息类型。

参数:

  • func – 要装饰的函数。

  • strict – 如果为 True，则如果消息类型或返回类型不在目标类型中，处理程序将引发异常。 如果为 False，它将改为记录警告。

  • match – 一个函数，它接受消息和上下文作为参数并返回一个布尔值。 这用于消息类型之后的辅助路由。 对于处理相同消息类型的处理程序，匹配函数按处理程序的字母顺序应用，并且将调用第一个匹配的处理程序，而其余的处理程序将被跳过。 如果为 None，则将调用按字母顺序排列的第一个匹配相同消息类型的处理程序。

event(func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, None]] = None, *, strict: bool = True, match: None | Callable[[ReceivesT, MessageContext], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, None]]], MessageHandler[AgentT, ReceivesT, None]] | MessageHandler[AgentT, ReceivesT, None][source]#

事件消息处理器的装饰器。

将此装饰器添加到 RoutedAgent 类中，用于处理事件消息的方法。 这些方法必须具有特定的签名才能有效。

  • 该方法必须是一个 async 方法。

  • 该方法必须使用 @message_handler 装饰器进行装饰。

  • 该方法必须恰好有 3 个参数

    1. self

    2. message: 要处理的事件消息，必须使用要处理的消息类型进行类型提示。

    3. ctx：一个 autogen_core.MessageContext 对象。

  • 该方法必须返回 None。

处理程序可以通过接受消息类型的 Union 来处理多个消息类型。

参数:

  • func – 要装饰的函数。

  • strict – 如果为 True，则如果消息类型不在目标类型中，处理程序将引发异常。 如果为 False，则会改为记录警告。

  • match – 一个函数，它接受消息和上下文作为参数并返回一个布尔值。 这用于消息类型之后的辅助路由。 对于处理相同消息类型的处理程序，匹配函数按处理程序的字母顺序应用，并且将调用第一个匹配的处理程序，而其余的处理程序将被跳过。 如果为 None，则将调用按字母顺序排列的第一个匹配相同消息类型的处理程序。

rpc(func: None | Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]] = None, *, strict: bool = True, match: None | Callable[[ReceivesT, MessageContext], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]]], MessageHandler[AgentT, ReceivesT, ProducesT]] | MessageHandler[AgentT, ReceivesT, ProducesT][source]#

RPC 消息处理程序的装饰器。

将此装饰器添加到 RoutedAgent 类中，用于处理 RPC 消息的方法。 这些方法必须具有特定的签名，以便有效。

  • 该方法必须是一个 async 方法。

  • 该方法必须使用 @message_handler 装饰器进行装饰。

  • 该方法必须恰好有 3 个参数

    1. self

    2. message：要处理的消息，必须使用它要处理的消息类型进行类型提示。

    3. ctx：一个 autogen_core.MessageContext 对象。

  • 该方法必须进行类型提示，说明它可以作为响应返回哪些消息类型，如果它不返回任何内容，则可以返回 None。

处理程序可以通过接受消息类型的 Union 来处理多个消息类型。它还可以通过返回消息类型的 Union 来返回多个消息类型。

参数:

  • func – 要装饰的函数。

  • strict – 如果为 True，则如果消息类型或返回类型不在目标类型中，处理程序将引发异常。 如果为 False，它将改为记录警告。

  • match – 一个函数，它接受消息和上下文作为参数并返回一个布尔值。 这用于消息类型之后的辅助路由。 对于处理相同消息类型的处理程序，匹配函数按处理程序的字母顺序应用，并且将调用第一个匹配的处理程序，而其余的处理程序将被跳过。 如果为 None，则将调用按字母顺序排列的第一个匹配相同消息类型的处理程序。

class FunctionCall(id: 'str', arguments: 'str', name: 'str')[source]#

基类: object

id: str#
arguments: str#
name: str#
class TypeSubscription(topic_type: str, agent_type: str | AgentType, id: str | None = None)[source]#

基类: Subscription

此订阅基于类型匹配主题，并使用主题的来源作为代理键映射到代理。

此订阅会导致每个源都有其自己的代理实例。

例子

from autogen_core import TypeSubscription

subscription = TypeSubscription(topic_type="t1", agent_type="a1")

在这种情况下

  • 类型为 t1 且源为 s1 的 topic_id 将由类型为 a1 且键为 s1 的代理处理

  • 类型为 t1 且源为 s2 的 topic_id 将由类型为 a1 且键为 s2 的代理处理。

参数:

  • topic_type (str) – 要匹配的主题类型

  • agent_type (str) – 处理此订阅的代理类型

property id: str#

获取订阅的 ID。

实现应返回订阅的唯一 ID。通常这是一个 UUID。

返回:

str – 订阅的 ID。

property topic_type: str#
property agent_type: str#
is_match(topic_id: TopicId) → bool[source]#

检查给定的 topic_id 是否与订阅匹配。

参数:

topic_id (TopicId) – 要检查的 TopicId。

返回:

bool – 如果 topic_id 与订阅匹配，则为 True；否则为 False。

map_to_agent(topic_id: TopicId) → AgentId[源代码]#

将 topic_id 映射到 agent。 只有在给定的 topic_id 的 is_match 返回 True 时才应调用。

参数:

topic_id (TopicId) – 要映射的 TopicId。

返回:

AgentId – 应该处理 topic_id 的 agent 的 ID。

引发:

CantHandleException – 如果订阅无法处理 topic_id。

class DefaultSubscription(topic_type: str = 'default', agent_type: str | AgentType | None = None)[源代码]#

基类： TypeSubscription

默认订阅旨在为只需要全局范围代理的应用程序提供合理的默认值。

默认情况下，此主题使用“default”主题类型，并尝试根据实例化上下文检测要使用的代理类型。

参数:

  • topic_type (str, optional) – 要订阅的主题类型。 默认为“default”。

  • agent_type (str, optional) – 用于订阅的代理类型。 默认为 None，在这种情况下，它将尝试根据实例化上下文检测代理类型。

class DefaultTopicId(type: str = 'default', source: str | None = None)[源代码]#

基类： TopicId

DefaultTopicId 为 TopicId 的 topic_id 和 source 字段提供了一个合理的默认值。

如果在消息处理程序的上下文中创建，则 source 将设置为消息处理程序的 agent_id，否则将设置为“default”。

参数:

  • type (str, optional) – 发布消息到的主题类型。 默认为“default”。

  • source (str | None, optional) – 发布消息到的主题来源。 如果为 None，则如果位于消息处理程序的上下文中，则 source 将设置为消息处理程序的 agent_id，否则将设置为“default”。 默认为 None。

default_subscription(cls: Type[BaseAgentType] | None = None) → Callable[[Type[BaseAgentType]], Type[BaseAgentType]] | Type[BaseAgentType][源代码]#
type_subscription(topic_type: str) → Callable[[Type[BaseAgentType]], Type[BaseAgentType]][源代码]#
class TypePrefixSubscription(topic_type_prefix: str, agent_type: str | AgentType, id: str | None = None)[源代码]#

基类: Subscription

此订阅基于类型的前缀匹配主题，并使用主题的来源作为代理键映射到代理。

此订阅会导致每个源都有其自己的代理实例。

例子

from autogen_core import TypePrefixSubscription

subscription = TypePrefixSubscription(topic_type_prefix="t1", agent_type="a1")

在这种情况下

  • 类型为 t1 且源为 s1 的 topic_id 将由类型为 a1 且键为 s1 的代理处理

  • 类型为 t1 且源为 s2 的 topic_id 将由类型为 a1 且键为 s2 的代理处理。

  • 类型为 t1SUFFIX 且来源为 s2 的 topic_id 将由类型为 a1 且键为 s2 的代理处理。

参数:

  • topic_type_prefix (str) – 要匹配的主题类型前缀

  • agent_type (str) – 处理此订阅的代理类型

property id: str#

获取订阅的 ID。

实现应返回订阅的唯一 ID。通常这是一个 UUID。

返回:

str – 订阅的 ID。

property topic_type_prefix: str#
property agent_type: str#
is_match(topic_id: TopicId) → bool[源代码]#

检查给定的 topic_id 是否与订阅匹配。

参数:

topic_id (TopicId) – 要检查的 TopicId。

返回:

bool – 如果 topic_id 与订阅匹配，则为 True；否则为 False。

map_to_agent(topic_id: TopicId) → AgentId[source]#

将 topic_id 映射到 agent。 只有在给定的 topic_id 的 is_match 返回 True 时才应调用。

参数:

topic_id (TopicId) – 要映射的 TopicId。

返回:

AgentId – 应该处理 topic_id 的 agent 的 ID。

引发:

CantHandleException – 如果订阅无法处理 topic_id。

JSON_DATA_CONTENT_TYPE = 'application/json'#

JSON数据的content type。

PROTOBUF_DATA_CONTENT_TYPE = 'application/x-protobuf'#

Protobuf数据的content type。

class SingleThreadedAgentRuntime(*, intervention_handlers: List[InterventionHandler] | None = None, tracer_provider: TracerProvider | None = None, ignore_unhandled_exceptions: bool = True)[source]#

基类: AgentRuntime

一个单线程的代理运行时，它使用单个 asyncio 队列处理所有消息。 消息按照接收顺序传递，运行时并发地在单独的 asyncio 任务中处理每个消息。

注意

此运行时适用于开发和独立应用程序。 它不适用于高吞吐量或高并发场景。

参数:

  • intervention_handlers (List[InterventionHandler], 可选) – 可以拦截消息在发送或发布之前的干预处理程序列表。 默认为 None。

  • tracer_provider (TracerProvider, 可选) – 用于跟踪的tracer provider。 默认为 None。

  • ignore_unhandled_exceptions (bool, 可选) – 是否忽略代理事件处理程序中发生的未处理异常。 任何后台异常将在下次调用 process_next 或从已等待的 stop、stop_when_idle 或 stop_when 中引发。 请注意，这不适用于 RPC 处理程序。 默认为 True。

示例

创建运行时、注册代理、发送消息并停止运行时的简单示例

import asyncio
from dataclasses import dataclass

from autogen_core import AgentId, MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler


@dataclass
class MyMessage:
    content: str


class MyAgent(RoutedAgent):
    @message_handler
    async def handle_my_message(self, message: MyMessage, ctx: MessageContext) -> None:
        print(f"Received message: {message.content}")


async def main() -> None:
    # Create a runtime and register the agent
    runtime = SingleThreadedAgentRuntime()
    await MyAgent.register(runtime, "my_agent", lambda: MyAgent("My agent"))

    # Start the runtime, send a message and stop the runtime
    runtime.start()
    await runtime.send_message(MyMessage("Hello, world!"), recipient=AgentId("my_agent", "default"))
    await runtime.stop()


asyncio.run(main())

创建运行时、注册代理、发布消息并停止运行时的示例

import asyncio
from dataclasses import dataclass

from autogen_core import (
    DefaultTopicId,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    default_subscription,
    message_handler,
)


@dataclass
class MyMessage:
    content: str


# The agent is subscribed to the default topic.
@default_subscription
class MyAgent(RoutedAgent):
    @message_handler
    async def handle_my_message(self, message: MyMessage, ctx: MessageContext) -> None:
        print(f"Received message: {message.content}")


async def main() -> None:
    # Create a runtime and register the agent
    runtime = SingleThreadedAgentRuntime()
    await MyAgent.register(runtime, "my_agent", lambda: MyAgent("My agent"))

    # Start the runtime.
    runtime.start()
    # Publish a message to the default topic that the agent is subscribed to.
    await runtime.publish_message(MyMessage("Hello, world!"), DefaultTopicId())
    # Wait for the message to be processed and then stop the runtime.
    await runtime.stop_when_idle()


asyncio.run(main())
property unprocessed_messages_count: int#
async send_message(message: Any, recipient: AgentId, *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any[source]#

向 Agent 发送消息并获取响应。

参数:

  • message (Any) – 要发送的消息。

  • recipient (AgentId) – 要将消息发送到的 Agent。

  • sender (AgentId | None, 可选) – 发送消息的 Agent。 仅当从没有 Agent 发送（例如直接从外部发送到运行时）时才应为 None。 默认为 None。

  • cancellation_token (CancellationToken | None, 可选) – 用于取消正在进行的 。 默认为 None。

引发:
返回:

Any – 来自 Agent 的响应。

async publish_message(message: Any, topic_id: TopicId, *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → None[source]#

将消息发布到给定命名空间中的所有代理，如果未提供命名空间，则发布到发送者的命名空间。

发布操作不应期待任何响应。

参数:

  • message (Any) – 要发布的消息。

  • topic_id (TopicId) – 要将消息发布到的主题。

  • sender (AgentId | None, 可选) – 发送消息的代理。默认为 None。

  • cancellation_token (CancellationToken | None, 可选) – 用于取消正在进行中的操作的令牌。默认为 None。

  • message_id (str | None, 可选) – 消息 ID。 如果为 None，则将生成新的消息 ID。默认为 None。 此消息 ID 必须是唯一的，建议使用 UUID。

引发:

UndeliverableException – 如果消息无法传递。

async save_state() → Mapping[str, Any][source]#

保存所有已实例化代理的状态。

此方法调用每个代理上的 save_state() 方法，并返回一个将代理 ID 映射到其状态的字典。

注意

此方法目前不保存订阅状态。 我们将在未来添加它。

返回:

一个将代理 ID 映射到其状态的字典。

async load_state(state: Mapping[str, Any]) → None[source]#

加载所有已实例化代理的状态。

此方法使用字典中提供的状态，对每个代理调用 load_state() 方法。字典的键是代理 ID，值是由 save_state() 方法返回的状态字典。

注意

此方法目前不加载订阅状态。我们将在未来添加此功能。

async process_next() → None[source]#

处理队列中的下一条消息。

如果后台任务中存在未处理的异常，将在此处引发。在引发未处理的异常后，不能再次调用 process_next 。

start() → None[source]#

启动运行时消息处理循环。 这在后台任务中运行。

例子

import asyncio
from autogen_core import SingleThreadedAgentRuntime


async def main() -> None:
    runtime = SingleThreadedAgentRuntime()
    runtime.start()

    # ... do other things ...

    await runtime.stop()


asyncio.run(main())
async close() → None[source]#

如果适用，调用 stop() 以及所有实例化的代理上的 Agent.close() 方法

async stop() → None[source]#

立即停止运行时消息处理循环。 当前正在处理的消息将完成，但所有后续消息将被丢弃。

async stop_when_idle() → None[source]#

当没有正在处理或排队的未完成消息时，停止运行时消息处理循环。 这是停止运行时的最常见方式。

async stop_when(condition: Callable[[], bool]) → None[source]#

当满足条件时，停止运行时消息处理循环。

注意

不建议使用此方法，此处仅出于历史原因。 它将产生一个繁忙的循环来不断检查条件。 调用 stop_when_idle 或 stop 效率更高。 如果您需要根据条件停止运行时，请考虑使用后台任务和 asyncio.Event 来发出条件满足的信号，并且后台任务应调用 stop。

async agent_metadata(agent: AgentId) → AgentMetadata[source]#

获取代理的元数据。

参数:

agent (AgentId) – 代理ID。

返回:

AgentMetadata – 代理元数据。

async agent_save_state(agent: AgentId) → Mapping[str, Any][source]#

保存单个代理的状态。

状态的结构是实现定义的，可以是任何JSON可序列化的对象。

参数:

agent (AgentId) – 代理ID。

返回:

Mapping[str, Any] – 保存的状态。

async agent_load_state(agent: AgentId, state: Mapping[str, Any]) → None[source]#

加载单个代理的状态。

参数:

  • agent (AgentId) – 代理ID。

  • state (Mapping[str, Any]) – 保存的状态。

async register_factory(type: str | AgentType, agent_factory: Callable[[], T | Awaitable[T]], *, expected_class: type[T] | None = None) → AgentType[source]#

向与特定类型关联的运行时注册一个代理工厂。类型必须是唯一的。 此 API 不添加任何订阅。

注意

这是一个低级 API，通常应该使用代理类的 register 方法，因为它也会自动处理订阅。

例子

from dataclasses import dataclass

from autogen_core import AgentRuntime, MessageContext, RoutedAgent, event
from autogen_core.models import UserMessage


@dataclass
class MyMessage:
    content: str


class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("My core agent")

    @event
    async def handler(self, message: UserMessage, context: MessageContext) -> None:
        print("Event received: ", message.content)


async def my_agent_factory():
    return MyAgent()


async def main() -> None:
    runtime: AgentRuntime = ...  # type: ignore
    await runtime.register_factory("my_agent", lambda: MyAgent())


import asyncio

asyncio.run(main())
参数:

  • type (str) – 此工厂创建的代理的类型。它与代理类名不同。 type 参数用于区分不同的工厂函数，而不是代理类。

  • agent_factory (Callable[[], T]) – 创建代理的工厂，其中 T 是一个具体的 Agent 类型。 在工厂内部，使用 autogen_core.AgentInstantiationContext 访问当前运行时和代理 ID 等变量。

  • expected_class (type[T] | None, 可选) – 代理的预期类，用于运行时的工厂验证。默认为 None。 如果为 None，则不执行任何验证。

async try_get_underlying_agent_instance(id: AgentId, type: Type[T] = Agent) → T[source]#

尝试按名称和命名空间获取底层代理实例。 通常不鼓励这样做（因此名称很长），但在某些情况下可能有用。

如果无法访问底层代理，这将引发异常。

参数:

  • id (AgentId) – 代理 ID。

  • type (Type[T], 可选) – 代理的预期类型。 默认为 Agent。

返回:

T – 具体代理实例。

引发:
async add_subscription(subscription: Subscription) → None[source]#

添加一个新的订阅，运行时应在处理发布的消息时履行该订阅

参数:

subscription (Subscription) – 要添加的订阅

async remove_subscription(id: str) → None[source]#

从运行时删除订阅

参数:

id (str) – 要删除的订阅的ID

引发:

LookupError – 如果订阅不存在

async get(id_or_type: AgentId | AgentType | str, /, key: str = 'default', *, lazy: bool = True) → AgentId[source]#
add_message_serializer(serializer: MessageSerializer[Any] | Sequence[MessageSerializer[Any]]) → None[source]#

向运行时添加新的消息序列化器

注意：这将根据 type_name 和 data_content_type 属性删除重复的序列化器

参数:

serializer (MessageSerializer[Any] | Sequence[MessageSerializer[Any]]) – 要添加的序列化器

ROOT_LOGGER_NAME = 'autogen_core'#

根 logger 的名称。

EVENT_LOGGER_NAME = 'autogen_core.events'#

用于结构化事件的 logger 的名称。

TRACE_LOGGER_NAME = 'autogen_core.trace'#

用于开发者预期的跟踪日志记录的 Logger 名称。 不应依赖此日志的内容和格式。

class Component[source]#

基类：ComponentFromConfig[ConfigT], ComponentSchemaType[ConfigT], Generic[ConfigT]

要创建组件类，请为具体类继承此类，为接口继承 ComponentBase。 然后实现两个类变量

  • component_config_schema - 一个 Pydantic 模型类，表示组件的配置。 这也是 Component 的类型参数。

  • component_type - 组件的逻辑类型是什么。

例子

from __future__ import annotations

from pydantic import BaseModel
from autogen_core import Component


class Config(BaseModel):
    value: str


class MyComponent(Component[Config]):
    component_type = "custom"
    component_config_schema = Config

    def __init__(self, value: str):
        self.value = value

    def _to_config(self) -> Config:
        return Config(value=self.value)

    @classmethod
    def _from_config(cls, config: Config) -> MyComponent:
        return cls(value=config.value)
class ComponentBase[source]#

基类：ComponentToConfig[ConfigT], ComponentLoader, Generic[ConfigT]

class ComponentFromConfig[source]#

基类：Generic[FromConfigT]

classmethod _from_config(config: FromConfigT) → Self[source]#

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

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

classmethod _from_config_past_version(config: Dict[str, Any], version: int) → Self[source]#

从配置对象的先前版本创建组件的新实例。

只有在配置对象的版本低于当前版本时才调用此方法，因为在这种情况下，架构未知。

参数:

  • config (Dict[str, Any]) – 配置对象。

  • version (int) – 配置对象的版本。

返回:

Self – 组件的新实例。

class ComponentLoader[source]#

基类: object

classmethod load_component(model: ComponentModel | Dict[str, Any], expected: None = None) → Self[source]#
classmethod load_component(model: ComponentModel | Dict[str, Any], expected: Type[ExpectedType]) → ExpectedType

从模型加载组件。旨在与 autogen_core.ComponentConfig.dump_component() 的返回类型一起使用。

例子

from autogen_core import ComponentModel
from autogen_core.models import ChatCompletionClient

component: ComponentModel = ...  # type: ignore

model_client = ChatCompletionClient.load_component(component)
参数:

  • model (ComponentModel) – 用于加载组件的模型。

  • model – _description_

  • expected (Type[ExpectedType] | None, optional) – 仅当直接在 ComponentLoader 上使用时才使用显式类型。默认为 None。

返回:

Self – 加载的组件。

引发:

  • ValueError – 如果 provider 字符串无效。

  • TypeError – Provider 不是 ComponentConfigImpl 的子类，或者预期类型不匹配。

返回:

Self | ExpectedType – 加载的组件。

pydantic model ComponentModel[source]#

基类： BaseModel

组件的模型类。包含实例化组件所需的所有信息。

显示 JSON schema
{
   "title": "ComponentModel",
   "description": "Model class for a component. Contains all information required to instantiate a component.",
   "type": "object",
   "properties": {
      "provider": {
         "title": "Provider",
         "type": "string"
      },
      "component_type": {
         "anyOf": [
            {
               "enum": [
                  "model",
                  "agent",
                  "tool",
                  "termination",
                  "token_provider",
                  "workbench"
               ],
               "type": "string"
            },
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Component Type"
      },
      "version": {
         "anyOf": [
            {
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Version"
      },
      "component_version": {
         "anyOf": [
            {
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Component Version"
      },
      "description": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Description"
      },
      "label": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Label"
      },
      "config": {
         "title": "Config",
         "type": "object"
      }
   },
   "required": [
      "provider",
      "config"
   ]
}

字段:

  • component_type (Literal['model', 'agent', 'tool', 'termination', 'token_provider', 'workbench'] | str | None)

  • component_version (int | None)

  • config (dict[str, Any])

  • description (str | None)

  • label (str | None)

  • provider (str)

  • version (int | None)

field provider: str [Required]#

描述如何实例化组件。

field component_type: ComponentType | None = None#

组件的逻辑类型。如果缺失，组件将假定 provider 的默认类型。

field version: int | None = None#

组件规范的版本。如果缺失，组件将假定用于加载它的库的当前版本。这显然是危险的，应该用于用户编写的临时配置。对于所有其他配置，都应指定版本。

field component_version: int | None = None#

组件的版本。如果缺失，组件将假定 provider 的默认版本。

field description: str | None = None#

组件的描述。

field label: str | None = None#

组件的人类可读标签。如果缺失，组件将假定 provider 的类名。

field config: dict[str, Any] [Required]#

schema 验证的 config 字段将传递给给定类的 autogen_core.ComponentConfigImpl._from_config() 实现，以创建组件类的新实例。

class ComponentSchemaType[source]#

基类： Generic[ConfigT]

component_config_schema: Type[ConfigT]#

Pydantic 模型类，表示组件的配置。

required_class_vars = ['component_config_schema', 'component_type']#
class ComponentToConfig[source]#

基类： Generic[ToConfigT]

一个类必须实现的两个方法才能成为组件。

参数:

Protocol (ConfigT) – 从 pydantic.BaseModel 派生的类型。

component_type: ClassVar[Literal['model', 'agent', 'tool', 'termination', 'token_provider', 'workbench'] | str]#

组件的逻辑类型。

component_version: ClassVar[int] = 1#

组件的版本。如果引入了模式不兼容性，则应更新此版本。

component_provider_override: ClassVar[str | None] = None#

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

component_description: ClassVar[str | None] = None#

组件的描述。 如果未提供，将使用该类的文档字符串。

component_label: ClassVar[str | None] = None#

组件的人工可读标签。 如果未提供，将使用组件类名。

_to_config() → ToConfigT[source]#

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

dump_component() → ComponentModel[source]#

将组件转储到可以重新加载的模型中。

引发:

TypeError – 如果组件是本地类。

返回:

ComponentModel – 表示组件的模型。

is_component_class(cls: type) → TypeGuard[Type[_ConcreteComponent[BaseModel]]][source]#
is_component_instance(cls: Any) → TypeGuard[_ConcreteComponent[BaseModel]][source]#
final class DropMessage[source]#

基类: object

标记类型，用于指示消息应被干预处理程序丢弃。 类型本身应从处理程序返回。

class InterventionHandler(*args, **kwargs)[source]#

基类: Protocol

干预处理程序是一个类，可用于修改，记录或删除由 autogen_core.base.AgentRuntime 处理的消息。

当消息提交到运行时时，将调用该处理程序。

目前，唯一支持此功能的运行时是 autogen_core.base.SingleThreadedAgentRuntime。

注意：从任何干预处理程序方法返回 None 将导致发出警告，并被视为“无更改”。 如果您打算删除消息，则应显式返回 DropMessage。

例子

from autogen_core import DefaultInterventionHandler, MessageContext, AgentId, SingleThreadedAgentRuntime
from dataclasses import dataclass
from typing import Any


@dataclass
class MyMessage:
    content: str


class MyInterventionHandler(DefaultInterventionHandler):
    async def on_send(self, message: Any, *, message_context: MessageContext, recipient: AgentId) -> MyMessage:
        if isinstance(message, MyMessage):
            message.content = message.content.upper()
        return message


runtime = SingleThreadedAgentRuntime(intervention_handlers=[MyInterventionHandler()])
async on_send(message: Any, *, message_context: MessageContext, recipient: AgentId) → Any | type[DropMessage][source]#

当使用 autogen_core.base.AgentRuntime.send_message() 将消息提交到 AgentRuntime 时调用。

async on_publish(message: Any, *, message_context: MessageContext) → Any | type[DropMessage][source]#

当使用 autogen_core.base.AgentRuntime.publish_message() 将消息发布到 AgentRuntime 时调用。

async on_response(message: Any, *, sender: AgentId, recipient: AgentId | None) → Any | type[DropMessage][source]#

当 AgentRuntime 从 Agent 的消息处理程序接收到一个返回值时调用。

class DefaultInterventionHandler(*args, **kwargs)[source]#

基类: InterventionHandler

一个简单的类，为所有干预处理程序方法提供默认实现，这些方法只是返回未更改的消息。 允许轻松进行子类化，以仅覆盖所需的方法。

async on_send(message: Any, *, message_context: MessageContext, recipient: AgentId) → Any | type[DropMessage][source]#

当使用 autogen_core.base.AgentRuntime.send_message() 将消息提交到 AgentRuntime 时调用。

async on_publish(message: Any, *, message_context: MessageContext) → Any | type[DropMessage][source]#

当使用 autogen_core.base.AgentRuntime.publish_message() 将消息发布到 AgentRuntime 时调用。

async on_response(message: Any, *, sender: AgentId, recipient: AgentId | None) → Any | type[DropMessage][source]#

当 AgentRuntime 从 Agent 的消息处理程序接收到一个返回值时调用。

ComponentType#

别名：Literal['model', 'agent', 'tool', 'termination', 'token_provider', 'workbench'] | str

页面内容
在 GitHub 上编辑
显示源代码