common_types

Class info

Classes

Name	Children	Inherits
BaseCode llmling_agent.common_types Base class for syntax-validated code.	ConfigCode	BaseModel
ConfigCode llmling_agent.common_types Base class for configuration code that validates against a specific type.		BaseCode Generic
SupportsRunStream llmling_agent.common_types Protocol for nodes that support streaming via run_stream().		Protocol
SupportsStructuredOutput llmling_agent.common_types Protocol for nodes that support structured output via run().		Protocol

🛈 DocStrings

Type definitions.

QueueStrategy module-attribute

QueueStrategy = Literal['concat', 'latest', 'buffer']

The strategy for handling multiple tool calls when a final result is found.

• 'early': Stop processing other tool calls once a final result is found
• 'exhaustive': Process all tool calls even after finding a final result

BaseCode

Bases: BaseModel

Base class for syntax-validated code.

Source code in src/llmling_agent/common_types.py

class BaseCode(BaseModel):
    """Base class for syntax-validated code."""

    code: str
    """The source code."""

    @field_validator("code")
    @classmethod
    def validate_syntax(cls, code: str) -> str:
        """Override in subclasses."""
        return code

    model_config = ConfigDict(use_attribute_docstrings=True)

code instance-attribute

code: str

The source code.

validate_syntax classmethod

validate_syntax(code: str) -> str

Override in subclasses.

Source code in src/llmling_agent/common_types.py

@field_validator("code")
@classmethod
def validate_syntax(cls, code: str) -> str:
    """Override in subclasses."""
    return code

ConfigCode

Bases: BaseCode

Base class for configuration code that validates against a specific type.

Generic type T specifies the type to validate against.

Source code in src/llmling_agent/common_types.py

class ConfigCode[T](BaseCode):
    """Base class for configuration code that validates against a specific type.

    Generic type T specifies the type to validate against.
    """

    validator_type: ClassVar[type]

    @field_validator("code")
    @classmethod
    def validate_syntax(cls, code: str) -> str:
        """Validate both YAML syntax and type constraints."""
        import yamling

        try:
            # First validate YAML syntax
            data = yamling.load(code, mode="yaml")

            # Then validate against target type
            match cls.validator_type:
                case type() as model_cls if issubclass(model_cls, BaseModel):
                    model_cls.model_validate(data)
                case _ if origin := get_origin(cls.validator_type):
                    # Handle generics like dict[str, int]
                    if not isinstance(data, origin):
                        msg = f"Expected {origin.__name__}, got {type(data).__name__}"
                        raise ValueError(msg)  # noqa: TRY004, TRY301
                    # Validate type arguments if present
                    if args := get_args(cls.validator_type):
                        _validate_type_args(data, args)
                case _:
                    msg = f"Unsupported validation type: {cls.validator_type}"
                    raise TypeError(msg)  # noqa: TRY301

        except Exception as e:
            msg = f"Invalid YAML for {cls.validator_type.__name__}: {e}"
            raise ValueError(msg) from e

        return code

    @classmethod
    def for_config[TConfig](
        cls,
        base_type: type[TConfig],
        *,
        name: str | None = None,
        error_msg: str | None = None,
    ) -> type[ConfigCode[TConfig]]:
        """Create a new ConfigCode class for a specific type.

        Args:
            base_type: The type to validate against
            name: Optional name for the new class
            error_msg: Optional custom error message

        Returns:
            New ConfigCode subclass with type-specific validation
        """

        class TypedConfigCode(ConfigCode[TConfig]):
            validator_type = base_type

            @field_validator("code")
            @classmethod
            def validate_syntax(cls, code: str) -> str:
                try:
                    return super().validate_syntax(code)
                except ValueError as e:
                    msg = error_msg or str(e)
                    raise ValueError(msg) from e

        if name:
            TypedConfigCode.__name__ = name

        return TypedConfigCode

for_config classmethod

for_config(
    base_type: type[TConfig], *, name: str | None = None, error_msg: str | None = None
) -> type[ConfigCode[TConfig]]

Create a new ConfigCode class for a specific type.

Parameters:

Name	Type	Description	Default
base_type	type[TConfig]	The type to validate against	required
name	str | None	Optional name for the new class	None
error_msg	str | None	Optional custom error message	None

Returns:

Type	Description
type[ConfigCode[TConfig]]	New ConfigCode subclass with type-specific validation

Source code in src/llmling_agent/common_types.py

@classmethod
def for_config[TConfig](
    cls,
    base_type: type[TConfig],
    *,
    name: str | None = None,
    error_msg: str | None = None,
) -> type[ConfigCode[TConfig]]:
    """Create a new ConfigCode class for a specific type.

    Args:
        base_type: The type to validate against
        name: Optional name for the new class
        error_msg: Optional custom error message

    Returns:
        New ConfigCode subclass with type-specific validation
    """

    class TypedConfigCode(ConfigCode[TConfig]):
        validator_type = base_type

        @field_validator("code")
        @classmethod
        def validate_syntax(cls, code: str) -> str:
            try:
                return super().validate_syntax(code)
            except ValueError as e:
                msg = error_msg or str(e)
                raise ValueError(msg) from e

    if name:
        TypedConfigCode.__name__ = name

    return TypedConfigCode

validate_syntax classmethod

validate_syntax(code: str) -> str

Validate both YAML syntax and type constraints.

Source code in src/llmling_agent/common_types.py

@field_validator("code")
@classmethod
def validate_syntax(cls, code: str) -> str:
    """Validate both YAML syntax and type constraints."""
    import yamling

    try:
        # First validate YAML syntax
        data = yamling.load(code, mode="yaml")

        # Then validate against target type
        match cls.validator_type:
            case type() as model_cls if issubclass(model_cls, BaseModel):
                model_cls.model_validate(data)
            case _ if origin := get_origin(cls.validator_type):
                # Handle generics like dict[str, int]
                if not isinstance(data, origin):
                    msg = f"Expected {origin.__name__}, got {type(data).__name__}"
                    raise ValueError(msg)  # noqa: TRY004, TRY301
                # Validate type arguments if present
                if args := get_args(cls.validator_type):
                    _validate_type_args(data, args)
            case _:
                msg = f"Unsupported validation type: {cls.validator_type}"
                raise TypeError(msg)  # noqa: TRY301

    except Exception as e:
        msg = f"Invalid YAML for {cls.validator_type.__name__}: {e}"
        raise ValueError(msg) from e

    return code

SupportsRunStream

Bases: Protocol

Protocol for nodes that support streaming via run_stream().

Used by Team and TeamRun to check if a node can be streamed.

Return type is a union because: - Agent returns AsyncIterator[RichAgentStreamEvent[TResult]] - Team/TeamRun return AsyncIterator[tuple[MessageNode, RichAgentStreamEvent]]

Source code in src/llmling_agent/common_types.py

@runtime_checkable
class SupportsRunStream[TResult](Protocol):
    """Protocol for nodes that support streaming via run_stream().

    Used by Team and TeamRun to check if a node can be streamed.

    Return type is a union because:
    - Agent returns `AsyncIterator[RichAgentStreamEvent[TResult]]`
    - Team/TeamRun return `AsyncIterator[tuple[MessageNode, RichAgentStreamEvent]]`
    """

    def run_stream(
        self, *prompts: Any, **kwargs: Any
    ) -> AsyncIterator[RichAgentStreamEvent[TResult] | TeamStreamEvent]: ...

SupportsStructuredOutput

Bases: Protocol

Protocol for nodes that support structured output via run().

This protocol is used for components that need to call run() with an output_type parameter (e.g., picker agents, Interactions).

Source code in src/llmling_agent/common_types.py

class SupportsStructuredOutput(Protocol):
    """Protocol for nodes that support structured output via run().

    This protocol is used for components that need to call run() with
    an output_type parameter (e.g., picker agents, Interactions).
    """

    async def run(self, *prompts: Any, output_type: Any = ...) -> ChatMessage[Any]: ...