from __future__ import annotations
import importlib
import warnings
from typing import Any, ClassVar, Dict, Generic, Literal, Type, TypeGuard, cast, overload
from pydantic import BaseModel
from typing_extensions import Self, TypeVar
ComponentType = Literal["model", "agent", "tool", "termination", "token_provider", "workbench"] | str
ConfigT = TypeVar("ConfigT", bound=BaseModel)
FromConfigT = TypeVar("FromConfigT", bound=BaseModel, contravariant=True)
ToConfigT = TypeVar("ToConfigT", bound=BaseModel, covariant=True)
T = TypeVar("T", bound=BaseModel, covariant=True)
[文档]
class ComponentModel(BaseModel):
"""组件的模型类。包含实例化组件所需的全部信息。"""
provider: str
"""描述组件如何被实例化。"""
component_type: ComponentType | None = None
"""组件的逻辑类型。如果缺失，组件将采用提供者的默认类型。"""
version: int | None = None
"""组件规范的版本号。如果缺失，组件将假定使用加载它的库的当前版本。这显然存在风险，应仅用于用户编写的临时配置。所有其他配置都应指定版本号。"""
component_version: int | None = None
"""组件的版本号。如果缺失，组件将假定使用提供者的默认版本。"""
description: str | None = None
"""组件的描述信息。"""
label: str | None = None
"""组件的可读标签。如果缺失，组件将使用提供者的类名作为默认值。"""
config: dict[str, Any]
"""经过模式验证的配置字段会被传递给给定类的 :py:meth:`autogen_core.ComponentConfigImpl._from_config` 方法实现，用于创建组件类的新实例。"""
def _type_to_provider_str(t: type) -> str:
return f"{t.__module__}.{t.__qualname__}"
WELL_KNOWN_PROVIDERS = {
"azure_openai_chat_completion_client": "autogen_ext.models.openai.AzureOpenAIChatCompletionClient",
"AzureOpenAIChatCompletionClient": "autogen_ext.models.openai.AzureOpenAIChatCompletionClient",
"openai_chat_completion_client": "autogen_ext.models.openai.OpenAIChatCompletionClient",
"OpenAIChatCompletionClient": "autogen_ext.models.openai.OpenAIChatCompletionClient",
}
[文档]
class ComponentFromConfig(Generic[FromConfigT]):
[文档]
@classmethod
def _from_config(cls, config: FromConfigT) -> Self:
"""从配置对象创建组件的新实例。
Args:
config (T): 配置对象。
Returns:
Self: 组件的新实例。
:meta public:
"""
raise NotImplementedError("This component does not support dumping to config")
[文档]
@classmethod
def _from_config_past_version(cls, config: Dict[str, Any], version: int) -> Self:
"""从旧版配置对象创建组件的新实例。
仅当配置对象版本低于当前版本时调用此方法，因为此时模式未知。
Args:
config (Dict[str, Any]): 配置对象。
version (int): 配置对象的版本。
Returns:
Self: 组件的新实例。
:meta public:
"""
raise NotImplementedError("This component does not support loading from past versions")
[文档]
class ComponentToConfig(Generic[ToConfigT]):
"""类必须实现的两种方法才能成为组件。
Args:
Protocol (ConfigT): 继承自 :py:class:`pydantic.BaseModel` 的类型。
"""
component_type: ClassVar[ComponentType]
"""组件的逻辑类型。"""
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
"""组件的人类可读标签。如果未提供，将使用组件类名。"""
[文档]
def _to_config(self) -> ToConfigT:
"""导出当前组件实例的配置，该配置可用于创建具有相同配置的新组件实例。
Returns:
T: 组件的配置。
:meta public:
"""
raise NotImplementedError("This component does not support dumping to config")
[文档]
def dump_component(self) -> ComponentModel:
"""将组件转储为可重新加载的模型。
Raises:
TypeError: 如果组件是本地类。
Returns:
ComponentModel: 表示组件的模型。
"""
if self.component_provider_override is not None:
provider = self.component_provider_override
else:
provider = _type_to_provider_str(self.__class__)
# Warn if internal module name is used,
if "._" in provider:
warnings.warn(
"Internal module name used in provider string. This is not recommended and may cause issues in the future. Silence this warning by setting component_provider_override to this value.",
stacklevel=2,
)
if "<locals>" in provider:
raise TypeError("Cannot dump component with local class")
if not hasattr(self, "component_type"):
raise AttributeError("component_type not defined")
description = self.component_description
if description is None and self.__class__.__doc__:
# use docstring as description
docstring = self.__class__.__doc__.strip()
for marker in ["\n\nArgs:", "\n\nParameters:", "\n\nAttributes:", "\n\n"]:
docstring = docstring.split(marker)[0]
description = docstring.strip()
obj_config = self._to_config().model_dump(exclude_none=True)
model = ComponentModel(
provider=provider,
component_type=self.component_type,
version=self.component_version,
component_version=self.component_version,
description=description,
label=self.component_label or self.__class__.__name__,
config=obj_config,
)
return model
ExpectedType = TypeVar("ExpectedType")
[文档]
class ComponentLoader:
@overload
@classmethod
def load_component(cls, model: ComponentModel | Dict[str, Any], expected: None = None) -> Self: ...
@overload
@classmethod
def load_component(cls, model: ComponentModel | Dict[str, Any], expected: Type[ExpectedType]) -> ExpectedType: ...
[文档]
@classmethod
def load_component(
cls, model: ComponentModel | Dict[str, Any], expected: Type[ExpectedType] | None = None
) -> Self | ExpectedType:
"""从模型加载组件。该方法设计用于与 :py:meth:`autogen_core.ComponentConfig.dump_component` 的返回类型配合使用。
Example:
.. code-block:: python
from autogen_core import ComponentModel
from autogen_core.models import ChatCompletionClient
component: ComponentModel = ... # type: ignore
model_client = ChatCompletionClient.load_component(component)
Args:
model (ComponentModel): 用于加载组件的模型。
Returns:
Self: 加载后的组件。
Args:
model (ComponentModel): _description_
expected (Type[ExpectedType] | None, optional): 仅在直接用于 ComponentLoader 时需要显式类型。默认为 None。
Raises:
ValueError: 如果提供者字符串无效。
TypeError: 提供者不是 ComponentConfigImpl 的子类，或期望类型不匹配。
Returns:
Self | ExpectedType: 加载后的组件。
"""
# Use global and add further type checks
if isinstance(model, dict):
loaded_model = ComponentModel(**model)
else:
loaded_model = model
# First, do a look up in well known providers
if loaded_model.provider in WELL_KNOWN_PROVIDERS:
loaded_model.provider = WELL_KNOWN_PROVIDERS[loaded_model.provider]
output = loaded_model.provider.rsplit(".", maxsplit=1)
if len(output) != 2:
raise ValueError("Invalid")
module_path, class_name = output
module = importlib.import_module(module_path)
component_class = module.__getattribute__(class_name)
if not is_component_class(component_class):
raise TypeError("Invalid component class")
# We need to check the schema is valid
if not hasattr(component_class, "component_config_schema"):
raise AttributeError("component_config_schema not defined")
if not hasattr(component_class, "component_type"):
raise AttributeError("component_type not defined")
loaded_config_version = loaded_model.component_version or component_class.component_version
if loaded_config_version < component_class.component_version:
try:
instance = component_class._from_config_past_version(loaded_model.config, loaded_config_version) # type: ignore
except NotImplementedError as e:
raise NotImplementedError(
f"Tried to load component {component_class} which is on version {component_class.component_version} with a config on version {loaded_config_version} but _from_config_past_version is not implemented"
) from e
else:
schema = component_class.component_config_schema # type: ignore
validated_config = schema.model_validate(loaded_model.config)
# We're allowed to use the private method here
instance = component_class._from_config(validated_config) # type: ignore
if expected is None and not isinstance(instance, cls):
raise TypeError("Expected type does not match")
elif expected is None:
return cast(Self, instance)
elif not isinstance(instance, expected):
raise TypeError("Expected type does not match")
else:
return cast(ExpectedType, instance)
[文档]
class ComponentSchemaType(Generic[ConfigT]):
# Ideally would be ClassVar[Type[ConfigT]], but this is disallowed https://github.com/python/typing/discussions/1424 (despite being valid in this context)
component_config_schema: Type[ConfigT]
"""表示组件配置的Pydantic模型类。"""
required_class_vars = ["component_config_schema", "component_type"]
def __init_subclass__(cls, **kwargs: Any):
super().__init_subclass__(**kwargs)
if cls.__name__ != "Component" and not cls.__name__ == "_ConcreteComponent":
# TODO: validate provider is loadable
for var in cls.required_class_vars:
if not hasattr(cls, var):
warnings.warn(
f"Class variable '{var}' must be defined in {cls.__name__} to be a valid component",
stacklevel=2,
)
[文档]
class ComponentBase(ComponentToConfig[ConfigT], ComponentLoader, Generic[ConfigT]): ...
[文档]
class Component(
ComponentFromConfig[ConfigT],
ComponentSchemaType[ConfigT],
Generic[ConfigT],
):
"""要创建组件类，具体类需继承自本类，接口需继承自ComponentBase。然后实现两个类变量：
- :py:attr:`component_config_schema` - 表示组件配置的Pydantic模型类。这也是Component的类型参数。
- :py:attr:`component_type` - 组件的逻辑类型。
示例：
.. code-block:: python
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)
"""
def __init_subclass__(cls, **kwargs: Any):
super().__init_subclass__(**kwargs)
if not is_component_class(cls):
warnings.warn(
f"Component class '{cls.__name__}' must subclass the following: ComponentFromConfig, ComponentToConfig, ComponentSchemaType, ComponentLoader, individually or with ComponentBase and Component. Look at the component config documentation or how OpenAIChatCompletionClient does it.",
stacklevel=2,
)
# Should never be used directly, only for type checking
class _ConcreteComponent(
ComponentFromConfig[ConfigT],
ComponentSchemaType[ConfigT],
ComponentToConfig[ConfigT],
ComponentLoader,
Generic[ConfigT],
): ...
[文档]
def is_component_instance(cls: Any) -> TypeGuard[_ConcreteComponent[BaseModel]]:
return (
isinstance(cls, ComponentFromConfig)
and isinstance(cls, ComponentToConfig)
and isinstance(cls, ComponentSchemaType)
and isinstance(cls, ComponentLoader)
)
[文档]
def is_component_class(cls: type) -> TypeGuard[Type[_ConcreteComponent[BaseModel]]]:
return (
issubclass(cls, ComponentFromConfig)
and issubclass(cls, ComponentToConfig)
and issubclass(cls, ComponentSchemaType)
and issubclass(cls, ComponentLoader)
)
