added operative models

autogen/structure/models/field_model.py

@@ -0,0 +1,116 @@
+# copied from https://github.com/lion-agi/lion-os/blob/main/lion/core/models/field_model.py
+# copyright by HaiyangLi, APACHE LICENSE 2.0
+
+from typing import Any, Callable
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic.fields import FieldInfo
+from pydantic_core import PydanticUndefined
+
+
+class FieldModel(BaseModel):
+    """Model for defining and managing field definitions.
+
+    Provides a structured way to define fields with:
+    - Type annotations and validation
+    - Default values and factories
+    - Documentation and metadata
+    - Serialization options
+
+    Example:
+        ```python
+        field = FieldModel(
+            name="age",
+            annotation=int,
+            default=0,
+            description="User age in years",
+            validator=lambda v: v if v >= 0 else 0
+        )
+        ```
+
+    Attributes:
+        default: Default field value
+        default_factory: Function to generate default value
+        title: Field title for documentation
+        description: Field description
+        examples: Example values
+        validators: Validation functions
+        exclude: Exclude from serialization
+        deprecated: Mark as deprecated
+        frozen: Mark as immutable
+        alias: Alternative field name
+        alias_priority: Priority for alias resolution
+        name: Field name (required)
+        annotation: Type annotation
+        validator: Validation function
+        validator_kwargs: Validator parameters
+
+    Notes:
+        - All attributes except 'name' can be UNDEFINED
+        - validator_kwargs are passed to field_validator decorator
+        - Cannot have both default and default_factory
+    """
+
+    model_config = ConfigDict(
+        extra="allow",
+        validate_default=False,
+        arbitrary_types_allowed=True,
+        use_enum_values=True,
+    )
+
+    # Field configuration attributes
+    default: Any = PydanticUndefined  # Default value
+    default_factory: Callable = PydanticUndefined  # Factory function for default value
+    title: str = PydanticUndefined  # Field title
+    description: str = PydanticUndefined  # Field description
+    examples: list = PydanticUndefined  # Example values
+    validators: list = PydanticUndefined  # Validation functions
+    exclude: bool = PydanticUndefined  # Exclude from serialization
+    deprecated: bool = PydanticUndefined  # Mark as deprecated
+    frozen: bool = PydanticUndefined  # Mark as immutable
+    alias: str = PydanticUndefined  # Alternative field name
+    alias_priority: int = PydanticUndefined  # Priority for alias resolution
+
+    # Core field attributes
+    name: str = Field(..., exclude=True)  # Field name (required)
+    annotation: type | Any = Field(PydanticUndefined, exclude=True)  # Type annotation
+    validator: Callable | Any = Field(PydanticUndefined, exclude=True)  # Validation function
+    validator_kwargs: dict | Any = Field(default_factory=dict, exclude=True)  # Validator parameters
+
+    @property
+    def field_info(self) -> FieldInfo:
+        """Generate Pydantic FieldInfo object from field configuration.
+
+        Returns:
+            FieldInfo object with all configured attributes
+
+        Notes:
+            - Uses clean dict to exclude UNDEFINED values
+            - Sets annotation to Any if not specified
+            - Preserves all metadata in field_info
+        """
+        annotation = self.annotation if self.annotation is not PydanticUndefined else Any
+        field_obj: FieldInfo = Field(**self.to_dict(True))  # type: ignore
+        field_obj.annotation = annotation
+        return field_obj
+
+    @property
+    def field_validator(self) -> dict[str, Callable] | None:
+        """Generate field validator configuration.
+
+        Returns:
+            Dictionary mapping validator name to function,
+            or None if no validator defined
+
+        Notes:
+            - Validator name is f"{field_name}_validator"
+            - Uses validator_kwargs if provided
+            - Returns None if validator is UNDEFINED
+        """
+        if self.validator is PydanticUndefined:
+            return None
+        kwargs = self.validator_kwargs or {}
+        return {f"{self.name}_validator": field_validator(self.name, **kwargs)(self.validator)}
+
+
+__all__ = ["FieldModel"]

autogen/structure/models/instruct.py

@@ -0,0 +1,98 @@
+# copied from https://github.com/lion-agi/lion-os/blob/main/lion/protocols/operatives/instruct.py
+# copyright by HaiyangLi, APACHE LICENSE 2.0
+
+from typing import Any
+
+from pydantic import BaseModel, JsonValue, field_validator
+
+from .field_model import FieldModel
+from .prompts import (
+    context_examples,
+    context_field_description,
+    guidance_examples,
+    guidance_field_description,
+    instruction_examples,
+    instruction_field_description,
+)
+
+
+def validate_instruction(cls, value) -> JsonValue | None:
+    """Validates that the instruction is not empty or None and is in the correct format.
+
+    Args:
+        cls: The validator class method.
+        value (JsonValue | None): The instruction value to validate.
+
+    Returns:
+        JsonValue | None: The validated instruction or None if invalid.
+    """
+    if value is None or (isinstance(value, str) and not value.strip()):
+        return None
+    return value
+
+
+# Field Models
+INSTRUCTION_FIELD = FieldModel(
+    name="instruction",
+    annotation=JsonValue | None,
+    default=None,
+    title="Primary Instruction",
+    description=instruction_field_description,
+    examples=instruction_examples,
+    validator=validate_instruction,
+    validator_kwargs={"mode": "before"},
+)
+
+GUIDANCE_FIELD = FieldModel(
+    name="guidance",
+    annotation=JsonValue | None,
+    default=None,
+    title="Execution Guidance",
+    description=guidance_field_description,
+    examples=guidance_examples,
+)
+
+CONTEXT_FIELD = FieldModel(
+    name="context",
+    annotation=JsonValue | None,
+    default=None,
+    title="Task Context",
+    description=context_field_description,
+    examples=context_examples,
+)
+
+
+class Instruct(BaseModel):
+    """Model for defining instruction parameters and execution requirements.
+
+    Attributes:
+        instruction (JsonValue | None): The primary instruction.
+        guidance (JsonValue | None): Execution guidance.
+        context (JsonValue | None): Task context.
+        reason (bool): Whether to include reasoning.
+        actions (bool): Whether specific actions are required.
+    """
+
+    instruction: JsonValue | None = INSTRUCTION_FIELD.field_info
+    guidance: JsonValue | None = GUIDANCE_FIELD.field_info
+    context: JsonValue | None = CONTEXT_FIELD.field_info
+
+    @field_validator("instruction", **INSTRUCTION_FIELD.validator_kwargs)
+    def _validate_instruction(cls, v):
+        """Field validator for the 'instruction' field.
+
+        Args:
+            v: The value to validate.
+
+        Returns:
+            JsonValue | None: The validated instruction value.
+        """
+        return INSTRUCTION_FIELD.validator(cls, v)
+
+
+class InstructResponse(BaseModel):
+    instruct: Instruct
+    response: Any = None
+
+
+__all__ = ["Instruct", "InstructResponse"]

autogen/structure/models/new_model_params.py

@@ -0,0 +1,170 @@
+# copied from https://github.com/lion-agi/lion-os/blob/main/lion/core/models/new_model_params.py
+# copyright by HaiyangLi, APACHE LICENSE 2.0
+
+
+import inspect
+from typing import Callable
+
+from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, create_model, field_validator, model_validator
+from pydantic.fields import FieldInfo
+
+from .field_model import FieldModel
+
+
+class NewModelParams(BaseModel):
+    """Configuration class for dynamically creating new Pydantic models."""
+
+    model_config = ConfigDict(
+        extra="forbid",
+        arbitrary_types_allowed=True,
+        use_enum_values=True,
+    )
+
+    name: str | None = None
+    parameter_fields: dict[str, FieldInfo] = Field(default_factory=dict)
+    base_type: type[BaseModel] = Field(default=BaseModel)
+    field_models: list[FieldModel] = Field(default_factory=list)
+    exclude_fields: list = Field(default_factory=list)
+    field_descriptions: dict = Field(default_factory=dict)
+    inherit_base: bool = Field(default=True)
+    config_dict: dict | None = Field(default=None)
+    doc: str | None = Field(default=None)
+    frozen: bool = False
+    _validators: dict[str, Callable] | None = PrivateAttr(default=None)
+    _use_keys: set[str] = PrivateAttr(default_factory=set)
+
+    @property
+    def use_fields(self):
+        """Get field definitions to use in new model."""
+        params = {k: v for k, v in self.parameter_fields.items() if k in self._use_keys}
+        params.update({f.name: f.field_info for f in self.field_models if f.name in self._use_keys})
+        return {k: (v.annotation, v) for k, v in params.items()}
+
+    @field_validator("parameter_fields", mode="before")
+    def validate_parameters(cls, value):
+        """Validate parameter field definitions."""
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            raise ValueError("Fields must be a dictionary.")
+        for k, v in value.items():
+            if not isinstance(k, str):
+                raise ValueError("Field names must be strings.")
+            if not isinstance(v, FieldInfo):
+                raise ValueError("Field values must be FieldInfo objects.")
+        return value.copy()
+
+    @field_validator("base_type", mode="before")
+    def validate_base(cls, value) -> type[BaseModel]:
+        """Validate base model type."""
+        if value is None:
+            return BaseModel
+        if isinstance(value, type) and issubclass(value, BaseModel):
+            return value
+        if isinstance(value, BaseModel):
+            return value.__class__
+        raise ValueError("Base must be a BaseModel subclass or instance.")
+
+    @field_validator("exclude_fields", mode="before")
+    def validate_fields(cls, value) -> list[str]:
+        """Validate excluded fields list."""
+        if value is None:
+            return []
+        if isinstance(value, dict):
+            value = list(value.keys())
+        if isinstance(value, set | tuple):
+            value = list(value)
+        if isinstance(value, list):
+            if not all(isinstance(i, str) for i in value):
+                raise ValueError("Field names must be strings.")
+            return value.copy()
+        raise ValueError("Fields must be a list, set, or dictionary.")
+
+    @field_validator("field_descriptions", mode="before")
+    def validate_field_descriptions(cls, value) -> dict[str, str]:
+        """Validate field descriptions dictionary."""
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            raise ValueError("Field descriptions must be a dictionary.")
+        for k, v in value.items():
+            if not isinstance(k, str):
+                raise ValueError("Field names must be strings.")
+            if not isinstance(v, str):
+                raise ValueError("Field descriptions must be strings.")
+        return value
+
+    @field_validator("name", mode="before")
+    def validate_name(cls, value) -> str:
+        """Validate model name."""
+        if value is None:
+            return "StepModel"
+        if not isinstance(value, str):
+            raise ValueError("Name must be a string.")
+        return value
+
+    @field_validator("field_models", mode="before")
+    def _validate_field_models(cls, value):
+        """Validate field model definitions."""
+        if value is None:
+            return []
+        value = [value] if not isinstance(value, list) else value
+        if not all(isinstance(i, FieldModel) for i in value):
+            raise ValueError("Field models must be FieldModel objects.")
+        return value
+
+    @model_validator(mode="after")
+    def validate_param_model(self):
+        """Validate complete model configuration."""
+        if self.base_type is not None:
+            self.parameter_fields.update(self.base_type.model_fields)
+
+        self.parameter_fields.update({f.name: f.field_info for f in self.field_models})
+
+        use_keys = list(self.parameter_fields.keys())
+        use_keys.extend(list(self._use_keys))
+
+        if self.exclude_fields:
+            use_keys = [i for i in use_keys if i not in self.exclude_fields]
+
+        self._use_keys = set(use_keys)
+
+        validators = {}
+
+        for i in self.field_models:
+            if i.field_validator is not None:
+                validators.update(i.field_validator)
+        self._validators = validators
+
+        if self.field_descriptions:
+            for i in self.field_models:
+                if i.name in self.field_descriptions:
+                    i.description = self.field_descriptions[i.name]
+
+        if not isinstance(self.name, str):
+            if hasattr(self.base_type, "class_name"):
+                if callable(self.base_type.class_name):
+                    self.name = self.base_type.class_name()
+                else:
+                    self.name = self.base_type.class_name
+            elif inspect.isclass(self.base_type):
+                self.name = self.base_type.__name__
+
+        return self
+
+    def create_new_model(self) -> type[BaseModel]:
+        """Create new Pydantic model with specified configuration."""
+        a: type[BaseModel] = create_model(
+            self.name,
+            __config__=self.config_dict,
+            __doc__=self.doc,
+            __base__=self.base_type if self.inherit_base else None,
+            __validators__=self._validators,
+            **self.use_fields,
+        )
+        if self.frozen:
+            a.model_config["frozen"] = True
+        return a
+
+
+__all__ = ["NewModelParams"]