autogen_core

class Agent(*args, **kwargs)

基类：Protocol

property metadata: AgentMetadata

代理的元数据。

property id: AgentId

代理的ID。

async bind_id_and_runtime(id: AgentId, runtime: AgentRuntime) → None

用于将Agent实例绑定到`AgentRuntime`的函数。

参数:

• agent_id (AgentId) -- 代理的ID。

• runtime (AgentRuntime) -- 要绑定代理的AgentRuntime实例。

async on_message(message: Any, ctx: MessageContext) → Any

代理的消息处理器。此方法应仅由运行时调用，而非其他代理。

参数:

• message (Any) -- 接收到的消息。类型为`subscriptions`中的一种。

• ctx (MessageContext) -- 消息上下文。

Returns:

Any -- 对消息的响应。可为None。

抛出:

• CancelledError -- 如果消息被取消。

• CantHandleException -- 如果代理无法处理该消息。

async save_state() → Mapping[str, Any]

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

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

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

参数:

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

async close() → None

当运行时关闭时调用

class AgentId(type: str | AgentType, key: str)

基类：object

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

更多信息请参见：Agent身份标识与生命周期

classmethod from_str(agent_id: str) → Self

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

property type: str

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

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

property key: str

代理实例标识符。

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

class AgentProxy(agent: AgentId, runtime: AgentRuntime)

基类：object

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

property id: AgentId

此代理的目标代理

property 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 序列化的。

class AgentMetadata

基类：TypedDict

type: str

key: str

description: str

class 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

向代理发送消息并获取响应。

参数:

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

• recipient (AgentId) -- 接收消息的代理。

• sender (AgentId | None, optional) -- 发送消息的代理。**仅**当消息不是由任何代理发送时（例如直接从外部发送到运行时）才应为None。默认为None。

• cancellation_token (CancellationToken | None, optional) -- 用于取消进行中的操作的令牌。默认为None。

抛出:

• CantHandleException -- 如果接收方无法处理该消息。

• UndeliverableException -- 如果消息无法送达。

• Other -- 接收方引发的任何其他异常。

Returns:

Any -- 来自代理的响应。

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

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

发布消息不期望获得响应。

参数:

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

• topic_id (TopicId) -- 发布消息的主题。

• sender (AgentId | None, optional) -- 发送消息的代理。默认为None。

• cancellation_token (CancellationToken | None, optional) -- 用于取消进行中的操作的令牌。默认为None。

• message_id (str | None, optional) -- 消息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

向运行时注册与特定类型关联的代理工厂。该类型必须唯一。此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是具体的代理类型。在工厂内部，使用`autogen_core.AgentInstantiationContext`访问当前运行时和代理ID等变量。

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

async register_agent_instance(agent_instance: Agent, agent_id: AgentId) → AgentId

向运行时注册一个代理实例。类型可以重复使用，但每个agent_id必须唯一。同一类型下的所有代理实例必须是相同的对象类型。此API不会添加任何订阅。

备注

这是一个底层API，通常应该使用代理类的`register_instance`方法，因为该方法还会自动处理订阅。

示例:

from dataclasses import dataclass

from autogen_core import AgentId, 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 main() -> None:
    runtime: AgentRuntime = ...  # type: ignore
    agent = MyAgent()
    await runtime.register_agent_instance(
        agent_instance=agent, agent_id=AgentId(type="my_agent", key="default")
    )


import asyncio

asyncio.run(main())

参数:

• agent_instance (Agent) -- 代理的具体实例。

• agent_id (AgentId) -- 代理的标识符。代理的类型是`agent_id.type`。

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

尝试通过名称和命名空间获取底层代理实例。通常不推荐这样做(因此名称较长)，但在某些情况下可能有用。

如果底层代理不可访问，将引发异常。

参数:

• id (AgentId) -- 代理ID。

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

Returns:

T -- 具体的代理实例。

抛出:

• LookupError -- 如果找不到代理。

• NotAccessibleError -- 如果代理不可访问，例如位于远程。

• TypeError -- 如果代理不是预期类型。

async get(id: AgentId, /, *, lazy: bool = True) → AgentId

async get(type: AgentType | str, /, key: str = 'default', *, lazy: bool = True) → AgentId

async save_state() → Mapping[str, Any]

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

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

Returns:

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

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

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

参数:

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

async agent_metadata(agent: AgentId) → AgentMetadata

获取代理的元数据。

参数:

agent (AgentId) -- 代理ID。

Returns:

AgentMetadata -- 代理元数据。

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

保存单个代理的状态。

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

参数:

agent (AgentId) -- 代理ID。

Returns:

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

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

加载单个代理的状态。

参数:

• agent (AgentId) -- 代理ID。

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

async add_subscription(subscription: Subscription) → None

添加运行时在处理发布消息时应满足的新订阅

参数:

subscription (Subscription) -- 要添加的订阅

async remove_subscription(id: str) → None

从运行时移除订阅

参数:

id (str) -- 要移除的订阅ID

抛出:

LookupError -- 如果订阅不存在

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

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

注意：这将根据 type_name 和 data_content_type 属性对序列化器进行去重

参数:

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

class BaseAgent(description: str)

基类：ABC, Agent

property metadata: AgentMetadata

代理的元数据。

async bind_id_and_runtime(id: AgentId, runtime: AgentRuntime) → None

用于将Agent实例绑定到`AgentRuntime`的函数。

参数:

• agent_id (AgentId) -- 代理的ID。

• runtime (AgentRuntime) -- 要绑定代理的AgentRuntime实例。

property type: str

property id: AgentId

代理的ID。

property runtime: AgentRuntime

final async on_message(message: Any, ctx: MessageContext) → Any

代理的消息处理器。此方法应仅由运行时调用，而非其他代理。

参数:

• message (Any) -- 接收到的消息。类型为`subscriptions`中的一种。

• ctx (MessageContext) -- 消息上下文。

Returns:

Any -- 对消息的响应。可为None。

抛出:

• CancelledError -- 如果消息被取消。

• CantHandleException -- 如果代理无法处理该消息。

abstract async on_message_impl(message: Any, ctx: MessageContext) → Any

async send_message(message: Any, recipient: AgentId, *, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any

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

async publish_message(message: Any, topic_id: TopicId, *, cancellation_token: CancellationToken | None = None) → None

async save_state() → Mapping[str, Any]

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

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

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

参数:

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

async close() → None

当运行时关闭时调用

async register_instance(runtime: AgentRuntime, agent_id: AgentId, *, skip_class_subscriptions: bool = True, skip_direct_message_subscription: bool = False) → AgentId

此函数与 register 类似，但用于注册代理实例。会创建一个基于代理 ID 的订阅并将其添加到运行时中。

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

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

class CacheStore

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

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

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

component_type: ClassVar[ComponentType] = 'cache_store'

组件的逻辑类型。

abstract get(key: str, default: T | None = None) → T | None

从存储中检索一个项目。

参数:

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

• default (optional) -- 如果未找到键时返回的默认值。 默认为 None。

Returns:

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

abstract set(key: str, value: T) → None

在存储中设置一个项目。

参数:

• key -- 用于存储项目的键。

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

class InMemoryStore

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

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

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

component_config_schema

InMemoryStoreConfig 的别名

get(key: str, default: T | None = None) → T | None

从存储中检索一个项目。

参数:

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

• default (optional) -- 如果未找到键时返回的默认值。 默认为 None。

Returns:

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

set(key: str, value: T) → None

在存储中设置一个项目。

参数:

• key -- 用于存储项目的键。

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

_to_config() → InMemoryStoreConfig

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:

T -- 组件的配置。

classmethod _from_config(config: InMemoryStoreConfig) → Self

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

参数:

config (T) -- 配置对象。

Returns:

Self -- 组件的新实例。

class CancellationToken

基类：object

用于取消待处理异步调用的令牌

cancel() → None

取消与此取消令牌关联的待处理异步调用

is_cancelled() → bool

检查 CancellationToken 是否已被使用

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

附加一个回调函数，当调用取消操作时将被触发

link_future(future: Future[Any]) → Future[Any]

将待处理的异步调用与令牌关联，以允许其被取消

class AgentInstantiationContext

基类：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)
        # 获取当前运行时——这里不使用但它是可用的
        _ = AgentInstantiationContext.current_runtime()
        # 获取当前代理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:
    # 获取当前运行时——这里不使用但它是可用的
    _ = AgentInstantiationContext.current_runtime()
    # 获取当前代理ID
    agent_id = AgentInstantiationContext.current_agent_id()
    print(f"Current AgentID from factory: {agent_id}")
    return TestAgent(description="Test agent")


async def main() -> None:
    # 创建SingleThreadedAgentRuntime实例
    runtime = SingleThreadedAgentRuntime()

    # 启动运行时
    runtime.start()

    # 用工厂函数注册代理类型
    await runtime.register_factory("test_agent", test_agent_factory)

    # 向代理发送消息。运行时将实例化代理并调用消息处理器
    await runtime.send_message(TestMessage(content="Hello, world!"), AgentId("test_agent", "default"))

    # 停止运行时
    await runtime.stop()


asyncio.run(main())

classmethod current_runtime() → AgentRuntime

classmethod current_agent_id() → AgentId

classmethod is_in_factory_call() → bool

class TopicId(type: str, source: str)

基类：object

TopicId 定义了广播消息的作用范围。本质上，代理运行时通过其广播 API 实现了发布-订阅模型：发布消息时必须指定主题。

更多信息请参阅：主题

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

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

property id: str

获取订阅的ID。

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

Returns:

str -- 订阅的ID。

is_match(topic_id: TopicId) → bool

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

参数:

topic_id (TopicId) -- 要检查的 TopicId。

Returns:

bool -- 如果 topic_id 匹配订阅则返回 True，否则返回 False。

map_to_agent(topic_id: TopicId) → AgentId

将 topic_id 映射到一个代理。仅当 is_match 对给定 topic_id 返回 True 时才应调用此方法。

参数:

topic_id (TopicId) -- 要映射的 TopicId。

Returns:

AgentId -- 应处理该 topic_id 的代理 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 populate_context(ctx: AgentType) → Generator[None, Any, None]

元数据 私有:

classmethod agent_type() → AgentType

class MessageHandlerContext

基类：object

classmethod populate_context(ctx: AgentId) → Generator[None, Any, None]

：元数据私有：

classmethod agent_id() → AgentId

class MessageSerializer(*args, **kwargs)

基类：Protocol[T]

property data_content_type: str

property type_name: str

deserialize(payload: bytes) → T

serialize(message: T) → bytes

class UnknownPayload(type_name: str, data_content_type: str, payload: bytes)

基类：object

type_name: str

data_content_type: str

payload: bytes

class Image(image: Image)

基类：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

classmethod from_uri(uri: str) → Image

classmethod from_base64(base64_str: str) → Image

to_base64() → str

classmethod from_file(file_path: Path) → Image

property data_uri: str

to_openai_format(detail: Literal['auto', 'low', 'high'] = 'auto') → Dict[str, Any]

class RoutedAgent(description: str)

基类：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

通过将消息路由到适当的消息处理程序来处理消息。 不要在子类中重写此方法。相反，应添加使用 event() 或 rpc() 装饰器装饰的消息处理方法。

async on_unhandled_message(message: Any, ctx: MessageContext) → None

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

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

基类：BaseAgent, ClosureContext

property metadata: AgentMetadata

代理的元数据。

property id: AgentId

代理的ID。

property runtime: AgentRuntime

async on_message_impl(message: Any, ctx: MessageContext) → Any

async save_state() → Mapping[str, Any]

闭包代理没有状态。因此该方法总是返回一个空字典。

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

闭包代理没有状态。因此该方法不做任何操作。

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

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

闭包可以定义预期的消息类型，或使用`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。

Returns:

AgentType -- 已注册代理的类型

class ClosureContext(*args, **kwargs)

基类：Protocol

property id: AgentId

async send_message(message: Any, recipient: AgentId, *, cancellation_token: CancellationToken | None = None, message_id: str | None = None) → Any

async publish_message(message: Any, topic_id: TopicId, *, cancellation_token: CancellationToken | None = None) → None

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]

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

将此装饰器添加到:class:`RoutedAgent`类中用于处理事件和RPC消息的方法上。 这些方法必须遵循特定的签名规范才能生效:

• 方法必须是`async`异步方法

• 方法必须使用`@message_handler`装饰器装饰

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

  1. self

  2. message: 要处理的消息，必须使用其目标处理的消息类型进行类型提示

  3. ctx: 一个:class:`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]

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

将此装饰器添加到:class:`RoutedAgent`类中用于处理事件消息的方法上。 这些方法必须遵循特定的签名规范才能生效:

• 方法必须是`async`异步方法

• 方法必须使用`@message_handler`装饰器装饰

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

  1. self

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

  3. ctx: 一个:class:`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]

RPC消息处理器的装饰器。

将此装饰器添加到:class:`RoutedAgent`类中用于处理RPC消息的方法上。 这些方法必须遵循特定的签名规范才能生效:

• 方法必须是`async`异步方法

• 方法必须使用`@message_handler`装饰器装饰

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

  1. self

  2. message: 要处理的消息，必须使用其目标处理的消息类型进行类型提示

  3. ctx: 一个:class:`autogen_core.MessageContext`对象

• 方法必须类型提示其可能返回的响应消息类型，如果不返回任何内容则可以返回`None`

处理器可以通过接受消息类型的Union来处理多种消息类型。也可以通过返回消息类型的Union来返回多种消息类型。

参数:

• func -- 要被装饰的函数

• strict -- 如果为`True`，当消息类型或返回类型不在目标类型中时会抛出异常。如果为`False`，则只记录警告

• match -- 一个接收消息和上下文作为参数并返回布尔值的函数。用于消息类型之后的二级路由。对于处理相同消息类型的处理器，匹配函数会按照处理器名称的字母顺序应用，第一个匹配的处理器会被调用，其余跳过。如果为`None`，则按字母顺序调用第一个匹配该消息类型的处理器。

class FunctionCall(id: 'str', arguments: 'str', name: 'str')

基类：object

id: str

arguments: str

name: str

class TypeSubscription(topic_type: str, agent_type: str | AgentType, id: str | None = None)

基类：Subscription

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

此订阅使每个来源拥有自己的代理实例。

示例：

from autogen_core import TypeSubscription

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

在此情况下：

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

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

参数:

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

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

property id: str

获取订阅的ID。

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

Returns:

str -- 订阅的ID。

property topic_type: str

property agent_type: str

is_match(topic_id: TopicId) → bool

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

参数:

topic_id (TopicId) -- 要检查的 TopicId。

Returns:

bool -- 如果 topic_id 匹配订阅则返回 True，否则返回 False。

map_to_agent(topic_id: TopicId) → AgentId

将 topic_id 映射到一个代理。仅当 is_match 对给定 topic_id 返回 True 时才应调用此方法。

参数:

topic_id (TopicId) -- 要映射的 TopicId。

Returns:

AgentId -- 应处理该 topic_id 的代理 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将由键为`s1`的`a1`类型代理处理

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

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

参数:

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

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

property id: str

获取订阅的ID。

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

Returns:

str -- 订阅的ID。

property topic_type_prefix: str

property agent_type: str

is_match(topic_id: TopicId) → bool

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

参数:

topic_id (TopicId) -- 要检查的 TopicId。

Returns:

bool -- 如果 topic_id 匹配订阅则返回 True，否则返回 False。

map_to_agent(topic_id: TopicId) → AgentId

将 topic_id 映射到一个代理。仅当 is_match 对给定 topic_id 返回 True 时才应调用此方法。

参数:

topic_id (TopicId) -- 要映射的 TopicId。

Returns:

AgentId -- 应处理该 topic_id 的代理 ID。

抛出:

CantHandleException -- 如果订阅无法处理该 topic_id。

JSON_DATA_CONTENT_TYPE = 'application/json'

JSON 数据的内容类型。

PROTOBUF_DATA_CONTENT_TYPE = 'application/x-protobuf'

Protobuf 数据的内容类型。

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

基类：AgentRuntime

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

备注

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

参数:

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

• tracer_provider (TracerProvider, optional) -- 用于追踪的追踪器提供者。默认为 None。

• ignore_unhandled_exceptions (bool, optional) -- 是否忽略代理事件处理程序中发生的未处理异常。任何后台异常将在下次调用 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

向代理发送消息并获取响应。

参数:

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

• recipient (AgentId) -- 接收消息的代理。

• sender (AgentId | None, optional) -- 发送消息的代理。**仅**当消息不是由任何代理发送时（例如直接从外部发送到运行时）才应为None。默认为None。

• cancellation_token (CancellationToken | None, optional) -- 用于取消进行中的操作的令牌。默认为None。

抛出:

• CantHandleException -- 如果接收方无法处理该消息。

• UndeliverableException -- 如果消息无法送达。

• Other -- 接收方引发的任何其他异常。

Returns:

Any -- 来自代理的响应。

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

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

发布消息不期望获得响应。

参数:

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

• topic_id (TopicId) -- 发布消息的主题。

• sender (AgentId | None, optional) -- 发送消息的代理。默认为None。

• cancellation_token (CancellationToken | None, optional) -- 用于取消进行中的操作的令牌。默认为None。

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

抛出:

UndeliverableException -- 如果消息无法送达。

async save_state() → Mapping[str, Any]

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

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

备注

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

Returns:

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

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

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

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

备注

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

async process_next() → None

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

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

start() → None

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

示例：

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

如果适用则调用 stop() 方法，并对所有已实例化的代理调用 Agent.close() 方法

async stop() → None

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

async stop_when_idle() → None

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

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

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

警告

不建议使用此方法，此处仅为向后兼容保留。 该方法会生成一个繁忙循环来持续检查条件。 调用`stop_when_idle`或`stop`会更加高效。 如果需要基于条件停止运行时，建议使用后台任务和asyncio.Event来 在条件满足时发出信号，然后由后台任务调用stop方法。

async agent_metadata(agent: AgentId) → AgentMetadata

获取代理的元数据。

参数:

agent (AgentId) -- 代理ID。

Returns:

AgentMetadata -- 代理元数据。

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

保存单个代理的状态。

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

参数:

agent (AgentId) -- 代理ID。

Returns:

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

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

加载单个代理的状态。

参数:

• 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

向运行时注册与特定类型关联的代理工厂。该类型必须唯一。此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是具体的代理类型。在工厂内部，使用`autogen_core.AgentInstantiationContext`访问当前运行时和代理ID等变量。

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

async register_agent_instance(agent_instance: Agent, agent_id: AgentId) → AgentId

向运行时注册一个代理实例。类型可以重复使用，但每个agent_id必须唯一。同一类型下的所有代理实例必须是相同的对象类型。此API不会添加任何订阅。

备注

这是一个底层API，通常应该使用代理类的`register_instance`方法，因为该方法还会自动处理订阅。

示例:

from dataclasses import dataclass

from autogen_core import AgentId, 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 main() -> None:
    runtime: AgentRuntime = ...  # type: ignore
    agent = MyAgent()
    await runtime.register_agent_instance(
        agent_instance=agent, agent_id=AgentId(type="my_agent", key="default")
    )


import asyncio

asyncio.run(main())

参数:

• agent_instance (Agent) -- 代理的具体实例。

• agent_id (AgentId) -- 代理的标识符。代理的类型是`agent_id.type`。

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

尝试通过名称和命名空间获取底层代理实例。通常不推荐这样做(因此名称较长)，但在某些情况下可能有用。

如果底层代理不可访问，将引发异常。

参数:

• id (AgentId) -- 代理ID。

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

Returns:

T -- 具体的代理实例。

抛出:

• LookupError -- 如果找不到代理。

• NotAccessibleError -- 如果代理不可访问，例如位于远程。

• TypeError -- 如果代理不是预期类型。

async add_subscription(subscription: Subscription) → None

添加运行时在处理发布消息时应满足的新订阅

参数:

subscription (Subscription) -- 要添加的订阅

async remove_subscription(id: str) → None

从运行时移除订阅

参数:

id (str) -- 要移除的订阅ID

抛出:

LookupError -- 如果订阅不存在

async get(id_or_type: AgentId | AgentType | str, /, key: str = 'default', *, lazy: bool = True) → AgentId

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

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

注意：这将根据 type_name 和 data_content_type 属性对序列化器进行去重

参数:

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

ROOT_LOGGER_NAME = 'autogen_core'

根日志记录器名称。

EVENT_LOGGER_NAME = 'autogen_core.events'

用于结构化事件的日志记录器名称。

TRACE_LOGGER_NAME = 'autogen_core.trace'

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

class Component

基类：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

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

class ComponentFromConfig

基类：Generic[FromConfigT]

classmethod _from_config(config: FromConfigT) → Self

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

参数:

config (T) -- 配置对象。

Returns:

Self -- 组件的新实例。

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

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

仅当配置对象版本低于当前版本时调用此方法，因为此时模式未知。

参数:

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

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

Returns:

Self -- 组件的新实例。

class ComponentLoader

基类：object

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

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。

Returns:

Self -- 加载后的组件。

抛出:

• ValueError -- 如果提供者字符串无效。

• TypeError -- 提供者不是 ComponentConfigImpl 的子类，或期望类型不匹配。

Returns:

Self | ExpectedType -- 加载后的组件。

pydantic model ComponentModel

基类：BaseModel

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

Show JSON schema

{
   "title": "ComponentModel",
   "description": "\u7ec4\u4ef6\u7684\u6a21\u578b\u7c7b\u3002\u5305\u542b\u5b9e\u4f8b\u5316\u7ec4\u4ef6\u6240\u9700\u7684\u5168\u90e8\u4fe1\u606f\u3002",
   "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"
   ]
}

Fields:

• 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

组件的逻辑类型。如果缺失，组件将采用提供者的默认类型。

field version: int | None = None

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

field component_version: int | None = None

组件的版本号。如果缺失，组件将假定使用提供者的默认版本。

field description: str | None = None

组件的描述信息。

field label: str | None = None

组件的可读标签。如果缺失，组件将使用提供者的类名作为默认值。

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

经过模式验证的配置字段会被传递给给定类的 autogen_core.ComponentConfigImpl._from_config() 方法实现，用于创建组件类的新实例。

class ComponentSchemaType

基类：Generic[ConfigT]

component_config_schema: Type[ConfigT]

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

required_class_vars = ['component_config_schema', 'component_type']

class ComponentToConfig

基类：Generic[ToConfigT]

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

参数:

Protocol (ConfigT) -- 继承自 pydantic.BaseModel 的类型。

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

组件的逻辑类型。

component_version: ClassVar[int] = 1

组件的版本号，如果引入了不兼容的schema变更则应更新此版本号。

component_provider_override: ClassVar[str | None] = None

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

component_description: ClassVar[str | None] = None

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

component_label: ClassVar[str | None] = None

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

_to_config() → ToConfigT

导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。

Returns:

T -- 组件的配置。

dump_component() → ComponentModel

将组件转储为可重新加载的模型。

抛出:

TypeError -- 如果组件是本地类。

Returns:

ComponentModel -- 表示组件的模型。

is_component_class(cls: type) → TypeGuard[Type[_ConcreteComponent[BaseModel]]]

is_component_instance(cls: Any) → TypeGuard[_ConcreteComponent[BaseModel]]

final class DropMessage

基类：object

用于标记消息应被干预处理器丢弃的标记类型。该类型本身应从处理器返回。

class InterventionHandler(*args, **kwargs)

基类：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]

当通过 autogen_core.base.AgentRuntime.send_message() 方法向 AgentRuntime 提交消息时调用。

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

当通过 autogen_core.base.AgentRuntime.publish_message() 方法向 AgentRuntime 发布消息时调用。

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

当 AgentRuntime 从 Agent 的消息处理器接收到返回值响应时调用。

class DefaultInterventionHandler(*args, **kwargs)

基类：InterventionHandler

提供所有干预处理器方法默认实现的简单类，这些方法会原样返回消息。便于通过子类化仅覆盖所需方法。

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

当通过 autogen_core.base.AgentRuntime.send_message() 方法向 AgentRuntime 提交消息时调用。

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

当通过 autogen_core.base.AgentRuntime.publish_message() 方法向 AgentRuntime 发布消息时调用。

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

当 AgentRuntime 从 Agent 的消息处理器接收到返回值响应时调用。

ComponentType

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