autogen_core#

class Agent(*args, **kwargs)[源代码]#

基础类: Protocol

property metadata: AgentMetadata#

代理的元数据。

property id: AgentId#

代理的ID。

async on_message(message: 任何, ctx: MessageContext) → 任何[源代码]#

代理的消息处理器。此功能应仅由运行时调用，而不是由其他代理调用。

Parameters:

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

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

Returns:

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

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

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

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

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

Parameters:

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

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

当运行时关闭时调用

class AgentId(type: str | AgentType, key: str)[源代码]#

基础: object

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

更多信息请参见：代理身份和生命周期

classmethod from_str(agent_id: str) → 自我[源代码]#

将格式为 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: 任何, *, sender: AgentId, cancellation_token: CancellationToken | 无 = None, message_id: str | 无 = None) → 任何[源代码]#
async save_state() → 映射[str, 任何][源代码]#

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

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

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

Parameters:

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

class AgentMetadata[源代码]#

基础：TypedDict

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

基础类: Protocol

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

向代理发送消息并获取回复。

Parameters:

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

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

  • sender (AgentId | None, optional) – 发送消息的代理。如果此消息不是由任何代理发送的，例如直接从外部运行时发送，则只能为None。默认值为None。

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

Raises:
Returns:

任何 – 来自代理的响应。

async publish_message(message: 任何, topic_id: TopicId, *, sender: AgentId | 无 = None, cancellation_token: CancellationToken | 无 = None, message_id: str | 无 = None) → 无[源代码]#

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

发布时不期望有任何响应。

Parameters:

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

  • topic (TopicId) – 发布消息的主题。

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

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

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

Raises:

UndeliverableException – 如果消息无法被送达。

async register_factory(type: str | AgentType, agent_factory: Callable[[], T | Awaitable[T]], *, expected_class: 类型[T] | 无 = 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())
Parameters:

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

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

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

async try_get_underlying_agent_instance(id: AgentId, type: 类型[T] = Agent) → T[源代码]#

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

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

Parameters:

  • id (AgentId) – 代理ID。

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

Returns:

T – 具体的代理实例。

Raises:
async get(id: AgentId, /, *, lazy: bool = True) → AgentId[源代码]#
async get(type: AgentType | str, /, key: str = 'default', *, lazy: bool = True) → AgentId
async save_state() → 映射[str, 任何][源代码]#

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

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

Returns:

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

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

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

Parameters:

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

async agent_metadata(agent: AgentId) → AgentMetadata[源代码]#

获取代理的元数据。

Parameters:

代理 (AgentId) – 代理的 ID。

Returns:

AgentMetadata – 代理元数据。

async agent_save_state(agent: AgentId) → 映射[str, 任何][源代码]#

保存单个代理的状态。

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

Parameters:

代理 (AgentId) – 代理的 ID。

Returns:

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

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

加载单个代理的状态。

Parameters:

  • 代理 (AgentId) – 代理ID。

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

async add_subscription(subscription: 订阅) → 无[源代码]#

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

Parameters:

subscription (Subscription) – 添加的订阅

async remove_subscription(id: str) → 无[源代码]#

从运行时移除一个订阅

Parameters:

id (str) – 要移除的订阅的id

Raises:

LookupError – 如果订阅不存在

add_message_serializer(serializer: MessageSerializer[任何] | Sequence[MessageSerializer[任何]]) → 无[源代码]#

在运行时添加一个新的消息序列化序列化器

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

Parameters:

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

class BaseAgent(description: str)[源代码]#

基础：ABC, Agent

property metadata: AgentMetadata#

代理的元数据。

property type: str#
property id: AgentId#

代理的ID。

property runtime: AgentRuntime#
final async on_message(message: 任何, ctx: 消息上下文) → 任何[源代码]#

代理的消息处理器。此功能应仅由运行时调用，而不是由其他代理调用。

Parameters:

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

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

Returns:

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

Raises:
abstract async on_message_impl(message: 任何, ctx: MessageContext) → 任何[源代码]#
async send_message(message: 任何, recipient: AgentId, *, cancellation_token: CancellationToken | 无 = None, message_id: str | 无 = None) → 任何[源代码]#

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

async publish_message(message: 任何, topic_id: TopicId, *, cancellation_token: CancellationToken | 无 = None) → 无[源代码]#
async save_state() → 映射[str, 任何][源代码]#

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

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

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

Parameters:

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

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

当运行时关闭时调用

async classmethod register(runtime: AgentRuntime, type: str, factory: Callable[[], 自我 | Awaitable[自我]], *, skip_class_subscriptions: bool = False, skip_direct_message_subscription: bool = False) → AgentType[源代码]#

注册一个ABC的虚拟子类。

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

class CacheStore[源代码]#

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

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

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

component_type: ClassVar[ComponentType] = 'cache_store'#

组件的逻辑类型。

abstract get(key: str, default: T | 无 = None) → T | 无[源代码]#

从商店中检索一个项目。

Parameters:

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

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

Returns:

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

abstract set(key: str, value: T) → 无[源代码]#

在商店中设置一个项目。

Parameters:

  • key – 存储项的键。

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

class InMemoryStore[源代码]#

基础类: CacheStore[T], Component[InMemoryStoreConfig]

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

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

component_config_schema#

InMemoryStoreConfig 的别名

get(key: str, default: T | 无 = None) → T | 无[源代码]#

从商店中检索一个项目。

Parameters:

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

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

Returns:

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

set(key: str, value: T) → 无[源代码]#

在商店中设置一个项目。

Parameters:

  • key – 存储项的键。

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

_to_config() → InMemoryStoreConfig[源代码]#

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

Returns:

T – 组件的配置。

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

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

Parameters:

config (T) – 配置对象。

Returns:

Self – 组件的新实例。

class CancellationToken[源代码]#

基础: object

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

cancel() → 无[源代码]#

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

is_cancelled() → bool[源代码]#

检查CancellationToken是否已被使用

add_callback(callback: Callable[[], 无]) → 无[源代码]#

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

将一个待处理的异步调用链接到一个令牌以允许其取消

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)
        # 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定义了广播消息的范围。本质上，代理运行时通过其广播API实现了发布-订阅模型：在发布消息时，必须指定主题。

查看这里获取更多信息：主题

type: str#

此主题ID包含的事件类型。遵循云事件规范。

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

了解更多信息：cloudevents/spec

source: str#

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

了解更多请点击：cloudevents/spec

classmethod from_str(topic_id: str) → 自我[源代码]#

将格式为 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是否与订阅匹配。

Parameters:

topic_id (TopicId) – 要检查的TopicId。

Returns:

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

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

将topic_id映射到一个代理。仅当给定的topic_id返回is_match为True时才应调用。

Parameters:

topic_id (TopicId) – 要映射的TopicId。

Returns:

AgentId – 应处理 topic_id 的代理的 ID。

Raises:

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

class MessageContext(sender: AgentId | 无, topic_id: TopicId | 无, is_rpc: bool, cancellation_token: CancellationToken, message_id: str)[源代码]#

基础: object

sender: AgentId | 无#
topic_id: TopicId | 无#
is_rpc: bool#
cancellation_token: CancellationToken#
message_id: str#
class AgentType(type: str)[源代码]#

基础: object

type: str#

该代理类型的字符串表示。

class SubscriptionInstantiationContext[源代码]#

基础: object

classmethod agent_type() → AgentType[源代码]#
class MessageHandlerContext[源代码]#

基础: object

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) → 图片[源代码]#
classmethod from_uri(uri: str) → 图片[源代码]#
classmethod from_base64(base64_str: str) → 图片[源代码]#
to_base64() → str[源代码]#
classmethod from_file(file_path: 路径) → 图片[源代码]#
property data_uri: str#
to_openai_format(detail: 字面量['auto', 'low', 'high'] = 'auto') → 字典[str, 任何][源代码]#
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: 任何, ctx: MessageContext) → 任何 | 无[源代码]#

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

async on_unhandled_message(message: 任何, ctx: MessageContext) → 无[源代码]#

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

class ClosureAgent(description: str, closure: Callable[[ClosureContext, T, MessageContext], Awaitable[任何]], *, unknown_type_policy: 字面量['error', 'warn', 'ignore'] = 'warn')[源代码]#

基础类: BaseAgent, ClosureContext

property metadata: AgentMetadata#

代理的元数据。

property id: AgentId#

代理的ID。

property runtime: AgentRuntime#
async on_message_impl(message: 任何, ctx: MessageContext) → 任何[源代码]#
async save_state() → 映射[str, 任何][源代码]#

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

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

Closure agents 没有状态。所以这个方法什么也不做。

async classmethod register_closure(runtime: AgentRuntime, type: str, closure: Callable[[ClosureContext, T, 消息上下文], Awaitable[任何]], *, unknown_type_policy: 字面量['error', 'warn', 'ignore'] = 'warn', skip_direct_message_subscription: bool = False, description: str = '', subscriptions: Callable[[], list[订阅] | Awaitable[list[订阅]]] | 无 = 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())
Parameters:

  • 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, 可选) – 不为该代理添加直接消息订阅。默认为 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: 任何, recipient: AgentId, *, cancellation_token: CancellationToken | 无 = None, message_id: str | 无 = None) → 任何[源代码]#
async publish_message(message: 任何, topic_id: TopicId, *, cancellation_token: CancellationToken | 无 = None) → 无[源代码]#
message_handler(func: 无 | Callable[[AgentT, ReceivesT, 消息上下文], 协程[任何, 任何, ProducesT]] = None, *, strict: bool = True, match: 无 | Callable[[ReceivesT, 消息上下文], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, 消息上下文], 协程[任何, 任何, ProducesT]]], MessageHandler[AgentT, ReceivesT, ProducesT]] | MessageHandler[AgentT, ReceivesT, ProducesT][源代码]#

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

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

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

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

  • The method must have exactly 3 arguments:

    1. self

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

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

  • 该方法必须通过类型提示来指定它可以返回哪些消息类型作为响应，或者如果它不返回任何内容，则可以返回None。

Handlers 可以通过接受消息类型的联合来处理多种消息类型。它也可以通过返回消息类型的联合来返回多种消息类型。

Parameters:

  • func – 要装饰的函数。

  • strict – 如果为True，当消息类型或返回类型不在目标类型中时，处理器将引发异常。如果为False，它将记录一个警告。

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

event(func: 无 | Callable[[AgentT, ReceivesT, 消息上下文], 协程[任何, 任何, 无]] = None, *, strict: bool = True, match: 无 | Callable[[ReceivesT, 消息上下文], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, 消息上下文], 协程[任何, 任何, 无]]], MessageHandler[AgentT, ReceivesT, 无]] | MessageHandler[AgentT, ReceivesT, 无][源代码]#

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

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

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

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

  • The method must have exactly 3 arguments:

    1. self

    2. message: 要处理的事件消息，这必须与预期处理的消息类型进行类型提示。

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

  • 该方法必须返回 None。

处理程序可以通过接受消息类型的联合来处理多种消息类型。

Parameters:

  • func – 要装饰的函数。

  • strict – 如果为True，当消息类型不在目标类型中时，处理器将抛出异常。如果为False，它将记录一个警告。

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

rpc(func: 无 | Callable[[AgentT, ReceivesT, 消息上下文], 协程[任何, 任何, ProducesT]] = None, *, strict: bool = True, match: 无 | Callable[[ReceivesT, 消息上下文], bool] = None) → Callable[[Callable[[AgentT, ReceivesT, 消息上下文], 协程[任何, 任何, ProducesT]]], MessageHandler[AgentT, ReceivesT, ProducesT]] | MessageHandler[AgentT, ReceivesT, ProducesT][源代码]#

RPC消息处理器的装饰器。

把这个装饰器添加到RoutedAgent类中的方法上，这些方法旨在处理RPC消息。 这些方法必须遵循特定的签名，以确保其有效性：

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

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

  • The method must have exactly 3 arguments:

    1. self

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

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

  • 该方法必须通过类型提示来指定它可以返回哪些消息类型作为响应，或者如果它不返回任何内容，则可以返回None。

Handlers 可以通过接受消息类型的联合来处理多种消息类型。它也可以通过返回消息类型的联合来返回多种消息类型。

Parameters:

  • func – 要装饰的函数。

  • strict – 如果为True，当消息类型或返回类型不在目标类型中时，处理器将引发异常。如果为False，它将记录一个警告。

  • match – 一个以消息和上下文为参数并返回布尔值的函数。这用于消息类型之后的次级路由。对于处理相同消息类型的处理程序，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)[源代码]#

基础：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的代理处理。

Parameters:

  • 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是否与订阅匹配。

Parameters:

topic_id (TopicId) – 要检查的TopicId。

Returns:

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

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

将topic_id映射到一个代理。仅当给定的topic_id返回is_match为True时才应调用。

Parameters:

topic_id (TopicId) – 要映射的TopicId。

Returns:

AgentId – 应处理 topic_id 的代理的 ID。

Raises:

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

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

基础：TypeSubscription

默认订阅设计为仅需要全局范围的应用程序的合理默认设置。

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

Parameters:

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

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

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

基础：TopicId

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

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

Parameters:

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

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

default_subscription(cls: 类型[BaseAgentType] | 无 = None) → Callable[[类型[BaseAgentType]], 类型[BaseAgentType]] | 类型[BaseAgentType][源代码]#
type_subscription(topic_type: str) → Callable[[类型[BaseAgentType]], 类型[BaseAgentType]][源代码]#
class TypePrefixSubscription(topic_type_prefix: str, agent_type: str | AgentType, id: str | 无 = 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将由具有键s2的类型a1的代理处理。

Parameters:

  • 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是否与订阅匹配。

Parameters:

topic_id (TopicId) – 要检查的TopicId。

Returns:

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

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

将topic_id映射到一个代理。仅当给定的topic_id返回is_match为True时才应调用。

Parameters:

topic_id (TopicId) – 要映射的TopicId。

Returns:

AgentId – 应处理 topic_id 的代理的 ID。

Raises:

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

JSON_DATA_CONTENT_TYPE = 'application/json'#

JSON数据的内容类型。

PROTOBUF_DATA_CONTENT_TYPE = 'application/x-protobuf'#

Protobuf 数据类型的内容。

class SingleThreadedAgentRuntime(*, intervention_handlers: 列表[InterventionHandler] | 无 = None, tracer_provider: TracerProvider | 无 = None, ignore_unhandled_exceptions: bool = True)[源代码]#

基础类: AgentRuntime

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

注意

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

Parameters:

  • intervention_handlers (List[InterventionHandler], 可选) – 一组干预处理器，可以在消息发送或发布之前拦截它们。默认为 None。

  • tracer_provider (TracerProvider, optional) – 用于追踪的追踪提供者。默认为 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: 任何, recipient: AgentId, *, sender: AgentId | 无 = None, cancellation_token: CancellationToken | 无 = None, message_id: str | 无 = None) → 任何[源代码]#

向代理发送消息并获取回复。

Parameters:

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

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

  • sender (AgentId | None, optional) – 发送消息的代理。如果此消息不是由任何代理发送的，例如直接从外部运行时发送，则只能为None。默认值为None。

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

Raises:
Returns:

任何 – 来自代理的响应。

async publish_message(message: 任何, topic_id: TopicId, *, sender: AgentId | 无 = None, cancellation_token: CancellationToken | 无 = None, message_id: str | 无 = None) → 无[源代码]#

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

发布时不期望有任何响应。

Parameters:

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

  • topic (TopicId) – 发布消息的主题。

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

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

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

Raises:

UndeliverableException – 如果消息无法被送达。

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

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

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

Returns:

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

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

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

Parameters:

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

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

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

如果后台任务中出现未处理的异常，它将会在这里被抛出。在未处理异常被抛出后，不能再次调用process_next。

start() → 无[源代码]#

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

示例：

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() → 无[源代码]#

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

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

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

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

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

async stop_when(condition: Callable[[], bool]) → 无[源代码]#

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

注意

不推荐使用此方法，这里仅出于遗留原因保留。该方法将会生成一个繁忙循环来持续检查条件。调用stop_when_idle或stop会更为高效。如果需要基于条件停止运行时，建议使用后台任务和asyncio.Event来发出信号，当条件满足时，后台任务应调用stop。

async agent_metadata(agent: AgentId) → AgentMetadata[源代码]#

获取代理的元数据。

Parameters:

代理 (AgentId) – 代理的 ID。

Returns:

AgentMetadata – 代理元数据。

async agent_save_state(agent: AgentId) → 映射[str, 任何][源代码]#

保存单个代理的状态。

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

Parameters:

代理 (AgentId) – 代理的 ID。

Returns:

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

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

加载单个代理的状态。

Parameters:

  • 代理 (AgentId) – 代理ID。

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

async register_factory(type: str | AgentType, agent_factory: Callable[[], T | Awaitable[T]], *, expected_class: 类型[T] | 无 = 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())
Parameters:

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

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

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

async try_get_underlying_agent_instance(id: AgentId, type: 类型[T] = Agent) → T[源代码]#

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

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

Parameters:

  • id (AgentId) – 代理ID。

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

Returns:

T – 具体的代理实例。

Raises:
async add_subscription(subscription: 订阅) → 无[源代码]#

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

Parameters:

subscription (Subscription) – 添加的订阅

async remove_subscription(id: str) → 无[源代码]#

从运行时移除一个订阅

Parameters:

id (str) – 要移除的订阅的id

Raises:

LookupError – 如果订阅不存在

async get(id_or_type: AgentId | AgentType | str, /, key: str = 'default', *, lazy: bool = True) → AgentId[源代码]#
add_message_serializer(serializer: MessageSerializer[任何] | Sequence[MessageSerializer[任何]]) → 无[源代码]#

在运行时添加一个新的消息序列化序列化器

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

Parameters:

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) → 自我[源代码]#

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

Parameters:

config (T) – 配置对象。

Returns:

Self – 组件的新实例。

classmethod _from_config_past_version(config: 字典[str, 任何], version: int) → 自我[源代码]#

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

仅在配置对象的版本低于当前版本时调用，因为在这种情况下模式是未知的。

Parameters:

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

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

Returns:

Self – 组件的新实例。

class ComponentLoader[源代码]#

基础: object

classmethod load_component(model: ComponentModel | 字典[str, 任何], expected: 无 = None) → 自我[源代码]#
classmethod load_component(model: ComponentModel | 字典[str, 任何], expected: 类型[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)
Parameters:

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

  • model – _描述_

  • expected (Type[ExpectedType] | None, optional) – 仅在直接用于ComponentLoader时明确指定类型。默认为None。

Returns:

Self – 已加载的组件。

Raises:

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

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

Returns:

Self | ExpectedType – 加载的组件。

pydantic model ComponentModel[源代码]#

基础：BaseModel

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

Show 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"
               ],
               "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'] | str | None)

  • component_version (int | None)

  • config (dict[str, Any])

  • 描述 (str | None)

  • label (str | None)

  • provider (str)

  • version (int | None)

field provider: str [Required]#

描述组件如何被实例化。

field component_type: ComponentType | 无 = None#

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

field version: int | 无 = None#

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

field component_version: int | 无 = None#

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

field description: str | 无 = None#

组件的描述。

field label: str | 无 = None#

组件的可读标签。如果缺失，组件将假定提供者的类名。

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

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

class ComponentSchemaType[源代码]#

基础：Generic[ConfigT]

component_config_schema: 类型[ConfigT]#

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

required_class_vars = ['component_config_schema', 'component_type']#
class ComponentToConfig[源代码]#

基类：Generic[ToConfigT]

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

Parameters:

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

component_type: ClassVar[字面量['model', 'agent', 'tool', 'termination', 'token_provider'] | str]#

组件的逻辑类型。

component_version: ClassVar[int] = 1#

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

component_provider_override: ClassVar[str | 无] = None#

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

component_description: ClassVar[str | 无] = None#

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

component_label: ClassVar[str | 无] = None#

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

_to_config() → ToConfigT[源代码]#

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

Returns:

T – 组件的配置。

dump_component() → ComponentModel[源代码]#

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

Raises:

TypeError – 如果组件是一个本地类。

Returns:

ComponentModel – 表示组件的模型。

is_component_class(cls: 类型) → TypeGuard[类型[_ConcreteComponent[BaseModel]]][源代码]#
is_component_instance(cls: 任何) → 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: 任何, *, message_context: 消息上下文, recipient: AgentId) → 任何 | 类型[DropMessage][源代码]#

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

async on_publish(message: 任何, *, message_context: 消息上下文) → 任何 | 类型[DropMessage][源代码]#

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

async on_response(message: 任何, *, sender: AgentId, recipient: AgentId | 无) → 任何 | 类型[DropMessage][源代码]#

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

class DefaultInterventionHandler(*args, **kwargs)[源代码]#

基础：InterventionHandler

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

async on_send(message: 任何, *, message_context: 消息上下文, recipient: AgentId) → 任何 | 类型[DropMessage][源代码]#

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

async on_publish(message: 任何, *, message_context: 消息上下文) → 任何 | 类型[DropMessage][源代码]#

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

async on_response(message: 任何, *, sender: AgentId, recipient: AgentId | 无) → 任何 | 类型[DropMessage][源代码]#

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

ComponentType#

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

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