Modules

Global

This package provides the core components required to define and configure cells within the system.

A cell is an autonomous agent characterized by its role, input and output formats, and the ability to execute specific tasks based on provided input data.

The package includes the following key components: - Cell: A concrete class representing a cell, which can be further subclassed for specific behaviors. - EnvVar: A model representing an environment variable as a key-value pair. - ConfigModel: A model for defining configuration parameters for a cell, including environment variables.

By creating a centralized package for these components, the system ensures consistency and ease of use when defining new types of cells and their configurations.

The __all__ list in this module specifies the public components that are available for import when the package is imported using the from package import * syntax.

Examples:

>>> from kinkernel import Cell
>>> from kinkernel import EnvVar, ConfigModel

# Define a custom cell by subclassing Cell # Define custom configuration models using ConfigModel and EnvVar

class kinkernel.Cell(*args, **kwargs)

Bases: BaseCell[InputModelT, OutputModelT]

A concrete implementation of the BaseCell class.

This class is meant to be used as a base for specific cell implementations that do not require additional attributes or methods beyond those provided by BaseCell. Subclasses of this class should provide specific role, description, input_format, output_format, and implement the execute method.

Example:

class MyCell(Cell[MyInputModel, MyOutputModel]):

role = ‘example_role’ description = ‘An example cell implementation.’ input_format = MyInputModel output_format = MyOutputModel

def execute(self, input_data: MyInputModel) -> MyOutputModel:

# Implementation of cell logic goes here. pass

Note that this class can be directly instantiated if no additional functionality is needed.

Usage:

# Assuming MyInputModel and MyOutputModel are defined and imported Pydantic models my_cell = Cell[MyInputModel, MyOutputModel]() result = my_cell.execute(my_input_data)

class kinkernel.ConfigModel(*, env_vars: List[EnvVar] | None = [])

Bases: BaseModel

Defines the configuration for a cell with an optional list of environment variables.

The model can be extended to include additional configuration parameters as needed.

Attributes:

env_vars (Optional[List[EnvVar]]): A list of EnvVar instances representing

environment variables. Defaults to an empty list if not provided.

env_vars: List[EnvVar] | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'env_vars': FieldInfo(annotation=Union[List[EnvVar], NoneType], required=False, default=[])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class kinkernel.EnvVar(*, key: str, value: str)

Bases: BaseModel

Represents an environment variable as a key-value pair.

Attributes:

key (str): The name of the environment variable. value (str): The value associated with the environment variable.

key: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'key': FieldInfo(annotation=str, required=True), 'value': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

value: str

Cells

This module initializes the cells folder, which includes the definitions of cell-related classes.

The cells folder provides the foundational abstract classes and concrete implementations required for creating and managing cells, which are autonomous agents within the system.

Classes:

BaseCell (class): Abstract base class for a Cell, defining the required interface and common behavior. Cell (class): A concrete implementation of BaseCell, intended to be subclassed for specific cell behaviors. ResponseModel (class): A Pydantic model representing the standard response format for a cell’s execution.

The __all__ list in this module specifies the public classes that are available for import when the package is imported using the from package import * syntax.

See Also:

kinkernel.cells.base kinkernel.cells.cell

class kinkernel.cells.BaseCell(*args, **kwargs)

Bases: Generic[InputModelT, OutputModelT], ABC

Abstract base class for a Cell, which is an autonomous agent in the system.

This class should be subclassed with specific implementations that define the role, description, input_format, output_format, and the execute method.

Parameters:

• role – The role of the cell within the system.

• description – A brief description of the cell’s purpose.

• input_format – A Pydantic model describing the expected input format.

• output_format – A Pydantic model describing the expected output format.

• config – An optional configuration model for the cell.

config: ConfigModel | None = None

description: str

classmethod get_description() → str

Get the description of the cell.

Returns:

The description of the cell.

Raises:

NotImplementedError – If the description is not defined.

classmethod get_input_format() → str

Get the JSON schema of the input format model.

Returns:

The JSON schema of the input format as a string.

Raises:

NotImplementedError – If the input_format is not defined.

classmethod get_input_schema() → dict

Retrieves the JSON schema for the input model of the cell.

The schema is cleaned to replace any $ref references with their actual definitions.

Returns:

The JSON schema of the input model.

Return type:

dict

Raises:

NotImplementedError – If the input_format is not defined in the subclass.

classmethod get_output_format() → str

Get the JSON schema of the output format model.

Returns:

The JSON schema of the output format as a string.

Raises:

NotImplementedError – If the output_format is not defined.

classmethod get_output_schema() → dict

Retrieves the JSON schema for the output model of the cell.

The schema is cleaned to replace any $ref references with their actual definitions.

Returns:

The JSON schema of the output model.

Return type:

dict

Raises:

NotImplementedError – If the output_format is not defined in the subclass.

classmethod get_role() → str

Get the role of the cell.

Returns:

The role of the cell.

Raises:

NotImplementedError – If the role is not defined.

input_format: Type[InputModelT]

output_format: Type[OutputModelT]

role: str

async run(input_json: str) → dict

Asynchronously process the input JSON string and return the result as a JSON.

This method acquires a mutex, validates the input JSON against the input_format Pydantic model, executes the cell’s logic by calling the _execute method, and then validates and serializes the output using the output_format Pydantic model.

Parameters:

input_json – A JSON string representing the input data to be processed.

Returns:

A JSON representing the response: ResponseModel, which could be the output data or an error message.

Raises:

• NotImplementedError – If the input_format or output_format is not defined.

• ValidationError – If the input data does not pass validation according to the input_format Pydantic model.

• Exception – If any other exception occurs during the processing of the input.

class kinkernel.cells.Cell(*args, **kwargs)

Bases: BaseCell[InputModelT, OutputModelT]

A concrete implementation of the BaseCell class.

This class is meant to be used as a base for specific cell implementations that do not require additional attributes or methods beyond those provided by BaseCell. Subclasses of this class should provide specific role, description, input_format, output_format, and implement the execute method.

Example:

class MyCell(Cell[MyInputModel, MyOutputModel]):

role = ‘example_role’ description = ‘An example cell implementation.’ input_format = MyInputModel output_format = MyOutputModel

def execute(self, input_data: MyInputModel) -> MyOutputModel:

# Implementation of cell logic goes here. pass

Note that this class can be directly instantiated if no additional functionality is needed.

Usage:

# Assuming MyInputModel and MyOutputModel are defined and imported Pydantic models my_cell = Cell[MyInputModel, MyOutputModel]() result = my_cell.execute(my_input_data)

class kinkernel.cells.ResponseModel(*, type: ResponseType, content: str)

Bases: BaseModel

A Pydantic model that represents a standard response from a cell.

Parameters:

• type – The type of response, indicating success or error.

• content – The content of the response, which may be a descriptive message or serialized data in JSON format.

content: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'content': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=ResponseType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: ResponseType

Config

This module serves as the entry point for the kinkernel.config package. It re-exports the EnvVar and ConfigModel classes from the kinkernel.config.base module.

The EnvVar class is used to represent environment variables as key-value pairs, while the ConfigModel class allows for defining and managing a set of configuration parameters for a cell, which can include environment variables and potentially other settings.

Example:

from kinkernel.config import EnvVar, ConfigModel

# Create an environment variable instance database_url = EnvVar(key=’DATABASE_URL’, value=’postgres://user:password@localhost/db’)

# Create a configuration model with the environment variable config = ConfigModel(env_vars=[database_url])

See Also:

kinkernel.config.base: The module where EnvVar and ConfigModel are defined.

Available Classes:

• EnvVar: Represents a single environment variable.

• ConfigModel: Holds the configuration for a cell, which may include multiple environment variables.

class kinkernel.config.ConfigModel(*, env_vars: List[EnvVar] | None = [])

Bases: BaseModel

Defines the configuration for a cell with an optional list of environment variables.

The model can be extended to include additional configuration parameters as needed.

Attributes:

env_vars (Optional[List[EnvVar]]): A list of EnvVar instances representing

environment variables. Defaults to an empty list if not provided.

env_vars: List[EnvVar] | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'env_vars': FieldInfo(annotation=Union[List[EnvVar], NoneType], required=False, default=[])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class kinkernel.config.EnvVar(*, key: str, value: str)

Bases: BaseModel

Represents an environment variable as a key-value pair.

Attributes:

key (str): The name of the environment variable. value (str): The value associated with the environment variable.

key: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'key': FieldInfo(annotation=str, required=True), 'value': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

value: str

Tools

This package provides tools for wrapping and interacting with OpenAI’s Functions.

The OpenAiFunctionCell class offers a high-level interface to execute BaseCell objects asynchronously and to access their properties and input arguments schema in order to use them with OpenAI functions.

The cell_to_openai_function function is a utility that converts a list of BaseCell objects into a list of OpenAiFunctionCell instances for easier manipulation.

Classes:

• OpenAiFunctionCell: Wraps a BaseCell to provide a convenient interface for execution and property access.

Functions:

• cell_to_openai_function(cells): Converts a list of BaseCell objects into a list of OpenAiFunctionCell instances.

class kinkernel.tools.OpenAiFunctionCell(cell: BaseCell)

Bases: object

A class representing an OpenAI function calls that wraps around a DK BaseCell object, providing a convenient interface to access its properties and execute it asynchronously.

Attributes:

cell (BaseCell): The underlying BaseCell instance. name (str): The role name of the cell. description (str): The description of the cell.

async ainvoke(input: str | Dict[str, Any]) → Any

Asynchronously invoke the cell with the given input.

Args:

input (Union[str, Dict[str, Any]]): The input data to pass to the cell. Can be a JSON string or a dictionary.

Returns:

Any: The output content from the cell after execution.

Raises:

TypeError: If the input cannot be serialized to JSON. KeyError: If the expected output key ‘content’ is missing from the cell’s response.

property args: dict

Retrieve the input arguments schema for the cell.

Returns:

Dict[str, Any]: A dictionary representing the input arguments schema.

Raises:

ValueError: If the input schema is not found or cannot be parsed.

kinkernel.tools.cell_to_openai_function(cells: List[BaseCell]) → List[OpenAiFunctionCell]

Convert a list of BaseCell instances to a list of OpenAiFunctionCell instances.

Args:

cells (List[BaseCell]): The list of BaseCell instances to convert.

Returns:

List[OpenAiFunctionCell]: A list of OpenAiFunctionCell instances.

Utils

This package provides utils functions.

The replace_refs_with_defs function cleans a JSON schema by replacing $ref references with their actual definitions.

Functions:

• replace_refs_with_defs(schema): Cleans a JSON schema by replacing $ref references with their actual definitions.

kinkernel.utils.replace_refs_with_defs(schema: dict) → dict

Replaces all $ref references in a JSON schema with their corresponding definitions from the $defs section.

Parameters:

schema (dict) – The JSON schema containing $ref references and a $defs section.

Returns:

The JSON schema with all $ref references replaced by their actual definitions.

Return type:

dict