Skip to content

Remote model classes

OpenAIModel #

OpenAIModel(
    name,
    *,
    genconf=None,
    schemaconf=None,
    ctx_len=None,
    max_tokens_limit=None,
    tokenizer=None,
    api_key=None,
    base_url=None,
    overhead_per_msg=None,
    token_estimation_factor=None,
    create_tokenizer=False,
    other_init_kwargs={}
)

Access an OpenAI model.

Supports constrained JSON output, via the OpenAI API tools mechanism.

Ref

https://platform.openai.com/docs/api-reference/chat/create

Create an OpenAI remote model.

Parameters:

Name Type Description Default
name str

Model name to resolve into an existing model.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None
schemaconf Optional[JSchemaConf]

Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.

None
ctx_len Optional[int]

Maximum context length to be used (shared for input and output). None for model's default.

None
max_tokens_limit Optional[int]

Maximum output tokens limit. None for model's default.

None
tokenizer Optional[Tokenizer]

An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.

None
api_key Optional[str]

OpenAI API key. Defaults to None, which will use env variable OPENAI_API_KEY.

None
base_url Optional[str]

Base location for API access. Defaults to None, which will use env variable OPENAI_BASE_URL or a default.

None
overhead_per_msg Optional[int]

Overhead tokens to account for when calculating token length. None for model's default.

None
token_estimation_factor Optional[float]

Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.

None
create_tokenizer bool

When no tokenizer is passed, should try to create one?

False
other_init_kwargs dict

Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

{}

Raises:

Type Description
ImportError

If OpenAI API is not installed.

NameError

If model name was not found or there's an API or authentication problem.

Source code in sibila/openai.py
def __init__(self,
             name: str,
             *,

             # common base model args
             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None,
             ctx_len: Optional[int] = None,
             max_tokens_limit: Optional[int] = None,
             tokenizer: Optional[Tokenizer] = None,

             # most important OpenAI-specific args
             api_key: Optional[str] = None,
             base_url: Optional[str] = None,
             overhead_per_msg: Optional[int] = None,
             token_estimation_factor: Optional[float] = None,
             create_tokenizer: bool = False,

             # other OpenAI-specific args
             other_init_kwargs: dict = {},
             ):
    """Create an OpenAI remote model.

    Args:
        name: Model name to resolve into an existing model.
        genconf: Model generation configuration. Defaults to None.
        schemaconf: Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.
        ctx_len: Maximum context length to be used (shared for input and output). None for model's default.
        max_tokens_limit: Maximum output tokens limit. None for model's default.
        tokenizer: An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.
        api_key: OpenAI API key. Defaults to None, which will use env variable OPENAI_API_KEY.
        base_url: Base location for API access. Defaults to None, which will use env variable OPENAI_BASE_URL or a default.
        overhead_per_msg: Overhead tokens to account for when calculating token length. None for model's default.
        token_estimation_factor: Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.
        create_tokenizer: When no tokenizer is passed, should try to create one?
        other_init_kwargs: Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

    Raises:
        ImportError: If OpenAI API is not installed.
        NameError: If model name was not found or there's an API or authentication problem.
    """


    if not has_openai:
        raise ImportError("Please install openai by running: pip install openai")

    self._client = self._client_async = None


    # also accept "provider:name" for ease of use
    provider_name = self.PROVIDER_NAME + ":"
    if name.startswith(provider_name):
        name = name[len(provider_name):]

    super().__init__(False,
                     genconf,
                     schemaconf,
                     tokenizer
                     )

    if (ctx_len is not None and
        max_tokens_limit is not None and
        overhead_per_msg is not None and
        token_estimation_factor is not None): # all elements given: probably created via Models.create()

        self._model_name = name
        default_ctx_len = ctx_len
        default_max_tokens_limit = max_tokens_limit
        default_overhead_per_msg = overhead_per_msg
        default_token_estimation_factor = token_estimation_factor

    else: # need to resolve
        settings = self.resolve_settings(self.PROVIDER_NAME,
                                         name,
                                         ["name", 
                                          "ctx_len", 
                                          "max_tokens_limit", 
                                          "overhead_per_msg",
                                          "token_estimation_factor"])
        self._model_name = settings.get("name") or name
        default_ctx_len = settings.get("ctx_len") # type: ignore[assignment]
        default_max_tokens_limit = settings.get("max_tokens_limit") # type: ignore[assignment]
        default_overhead_per_msg = settings.get("overhead_per_msg") # type: ignore[assignment]
        default_token_estimation_factor = settings.get("token_estimation_factor") # type: ignore[assignment]

        # all defaults are conservative values
        if ctx_len is None and default_ctx_len is None:
            default_ctx_len = 4096
            logger.warning(f"Model '{self._model_name}': unknown ctx_len, assuming {default_ctx_len}")

        if max_tokens_limit is None and default_max_tokens_limit is None:
            default_max_tokens_limit = ctx_len or default_ctx_len                
            # don't warn: assume equal to ctx_len: logger.warning(f"Model '{self._model_name}': unknown max_tokens_limit, assuming {default_max_tokens_limit}")

        if overhead_per_msg is None and default_overhead_per_msg is None:
            default_overhead_per_msg = 3
            # don't warn for this setting due to derived model classes (none uses it)

        if token_estimation_factor is None and default_token_estimation_factor is None:
            default_token_estimation_factor = self.DEFAULT_TOKEN_ESTIMATION_FACTOR
            logger.warning(f"Model '{self._model_name}': unknown token_estimation_factor, assuming {default_token_estimation_factor}")


    self.ctx_len = ctx_len or default_ctx_len

    self.max_tokens_limit = max_tokens_limit or default_max_tokens_limit
    self.max_tokens_limit = min(self.max_tokens_limit, self.ctx_len)

    self._overhead_per_msg = overhead_per_msg or default_overhead_per_msg

    self._token_estimation_factor = token_estimation_factor or default_token_estimation_factor

    self.maybe_image_input = True # True means maybe - always check model specs

    # only check for "json" text presence as json schema (including field descriptions) is requested with the tools facility.
    self.json_format_instructors["json_schema"] = self.json_format_instructors["json"]


    if self.tokenizer is None and create_tokenizer:
        try:
            self.tokenizer = OpenAITokenizer(self._model_name)
        except Exception as e:
            logger.warning(f"Could not create a local tokenizer for model '{self._model_name}' - "
                           "token length calculation will be disabled and assume defaults. "
                           "To support recent OpenAI models, install the latest tiktoken version with 'pip install -U tiktoken'. "
                           f"Internal error: {e}")


    self._client_init_kwargs = other_init_kwargs
    if api_key is not None:
        self._client_init_kwargs["api_key"] = api_key
    if base_url is not None:
        self._client_init_kwargs["base_url"] = base_url

extract #

extract(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
def extract(self,
            target: Any,

            query: Union[Thread,Msg,tuple,str],
            *,
            inst: Optional[str] = None,

            genconf: Optional[GenConf] = None,
            schemaconf: Optional[JSchemaConf] = None
            ) -> Any:        
    """Type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_extract(target,
                           thread,
                           genconf,
                           schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify #

classify(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
def classify(self,
             labels: Any,

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return self.extract(labels,
                        query,
                        inst=inst,
                        genconf=genconf,
                        schemaconf=schemaconf)

json #

json(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
def json(self,
         query: Union[Thread,Msg,tuple,str],
         *,
         json_schema: Union[dict,str,None] = None,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         massage_schema: bool = True,
         schemaconf: Optional[JSchemaConf] = None,
         ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = self.gen_json(thread,
                        json_schema,                            
                        genconf,
                        massage_schema,
                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass #

dataclass(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
def dataclass(self, # noqa: F811
              cls: Any, # a dataclass definition

              query: Union[Thread,Msg,tuple,str],
              *,
              inst: Optional[str] = None,

              genconf: Optional[GenConf] = None,
              schemaconf: Optional[JSchemaConf] = None
              ) -> Any: # a dataclass object
    """Constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_dataclass(cls,
                             thread,
                             genconf,
                             schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic #

pydantic(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
def pydantic(self,
             cls: Any, # a Pydantic BaseModel class

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any: # a Pydantic BaseModel object
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_pydantic(cls,
                            thread,
                            genconf,
                            schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call #

call(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def call(self,             
         query: Union[Thread,Msg,tuple,str],
         *,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         ok_length_is_error: bool = False
         ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen(thread=thread, 
                   genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

__call__ #

__call__(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def __call__(self,             
             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             ok_length_is_error: bool = False
             ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    return self.call(query,
                     inst=inst,
                     genconf=genconf,
                     ok_length_is_error=ok_length_is_error)

extract_async async #

extract_async(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Async type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
async def extract_async(self,
                        target: Any,

                        query: Union[Thread,Msg,tuple,str],
                        *,
                        inst: Optional[str] = None,

                        genconf: Optional[GenConf] = None,
                        schemaconf: Optional[JSchemaConf] = None
                        ) -> Any:        
    """Async type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_extract_async(target,
                                       thread,
                                       genconf,
                                       schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify_async async #

classify_async(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
async def classify_async(self,
                         labels: Any,

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return await self.extract_async(labels,
                                    query,
                                    inst=inst,
                                    genconf=genconf,
                                    schemaconf=schemaconf)

json_async async #

json_async(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
async def json_async(self,             
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     json_schema: Union[dict,str,None] = None,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     massage_schema: bool = True,
                     schemaconf: Optional[JSchemaConf] = None,
                     ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = await self.gen_json_async(thread,
                                    json_schema,
                                    genconf,
                                    massage_schema,
                                    schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass_async async #

dataclass_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
async def dataclass_async(self, # noqa: E811
                          cls: Any, # a dataclass definition

                          query: Union[Thread,Msg,tuple,str],
                          *,
                          inst: Optional[str] = None,

                          genconf: Optional[GenConf] = None,
                          schemaconf: Optional[JSchemaConf] = None
                          ) -> Any: # a dataclass object
    """Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_dataclass_async(cls,
                                         thread,
                                         genconf,
                                         schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic_async async #

pydantic_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
async def pydantic_async(self,
                         cls: Any, # a Pydantic BaseModel class

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any: # a Pydantic BaseModel object
    """Async constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_pydantic_async(cls,
                                        thread,
                                        genconf,
                                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call_async async #

call_async(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
async def call_async(self,
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     ok_length_is_error: bool = False
                     ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_async(thread=thread, 
                               genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

gen #

gen(thread, genconf=None)

Text generation from a Thread, used by the other model generation methods. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc.

GenOut

The output text is in GenOut.text.

Source code in sibila/openai.py
def gen(self, 
        thread: Thread,
        genconf: Optional[GenConf] = None,
        ) -> GenOut:
    """Text generation from a Thread, used by the other model generation methods.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc.
        The output text is in GenOut.text.
    """

    genconf2: GenConf
    kwargs, genconf2 = self._gen_pre(thread, genconf)

    self._ensure_client(False)

    try:
        # https://platform.openai.com/docs/api-reference/chat/create
        response = self._client.chat.completions.create(**kwargs) # type: ignore[attr-defined]

    except Exception as e:
        raise RuntimeError(f"Cannot generate. Internal error: {e}")


    return self._gen_post(response,
                          kwargs,
                          genconf2)

gen_json #

gen_json(
    thread,
    json_schema,
    genconf=None,
    massage_schema=True,
    schemaconf=None,
)

JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.

Source code in sibila/model.py
def gen_json(self,
             thread: Thread,
             json_schema: Union[dict,str,None],
             genconf: Optional[GenConf] = None,

             massage_schema: bool = True,
             schemaconf: Optional[JSchemaConf] = None,
             ) -> GenOut:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.
    """

    args = self._gen_json_pre(thread,
                              json_schema,
                              genconf,
                              massage_schema,
                              schemaconf)
    return self.gen(*args)

gen_dataclass #

gen_dataclass(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a dataclass definition. An initialized dataclass object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
thread Thread

The Thread object to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.

Source code in sibila/model.py
def gen_dataclass(self,
                  cls: Any, # a dataclass
                  thread: Thread,
                  genconf: Optional[GenConf] = None,
                  schemaconf: Optional[JSchemaConf] = None
                  ) -> GenOut:
    """Constrained generation after a dataclass definition.
    An initialized dataclass object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A dataclass definition.
        thread: The Thread object to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.
    """

    schema = self._gen_dataclass_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_dataclass_post(out,
                                    cls,
                                    schemaconf)

gen_pydantic #

gen_pydantic(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a Pydantic BaseModel-derived class definition. An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

TypeError

When cls is not a Pydantic BaseClass.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.

Source code in sibila/model.py
def gen_pydantic(self,
                 cls: Any, # a Pydantic BaseModel class
                 thread: Thread,
                 genconf: Optional[GenConf] = None,
                 schemaconf: Optional[JSchemaConf] = None
                 ) -> GenOut:
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.
        TypeError: When cls is not a Pydantic BaseClass.

    Returns:
        A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.
    """

    schema = self._gen_pydantic_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_pydantic_post(out,
                                   cls,
                                   schemaconf)

token_len #

token_len(thread_or_text, genconf=None)

Calculate or estimate the token length for a Thread or a plain text string. In some cases where it's not possible to calculate the exact token count, this function should give a conservative (upper bound) estimate. It's up to the implementation whether to account for side information like JSON Schema, but it must reflect the model's context token accounting. Thread or text must be the final text which will passed to model.

If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

More info on calculating function_call (and tools?) tokens:

https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

Parameters:

Name Type Description Default
thread_or_text Union[Thread, str]

For token length calculation.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Returns:

Type Description
int

Estimated number of tokens used.

Source code in sibila/openai.py
def token_len(self,
              thread_or_text: Union[Thread,str],
              genconf: Optional[GenConf] = None) -> int:
    """Calculate or estimate the token length for a Thread or a plain text string.
    In some cases where it's not possible to calculate the exact token count, 
    this function should give a conservative (upper bound) estimate.
    It's up to the implementation whether to account for side information like JSON Schema,
    but it must reflect the model's context token accounting.
    Thread or text must be the final text which will passed to model.

    If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

    From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

    More info on calculating function_call (and tools?) tokens:

    https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

    https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

    Args:
        thread_or_text: For token length calculation.
        genconf: Model generation configuration. Defaults to None.

    Returns:
        Estimated number of tokens used.
    """

    if isinstance(thread_or_text, Thread):
        thread = thread_or_text            
    else:
        thread = Thread.make_IN(thread_or_text)

    num_tokens = 0

    if self.tokenizer is None: # no tokenizer was found, so we'll have to do a conservative estimate

        OVERHEAD_PER_MSG = 3
        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            msg_tokens = len(str(message["content"])) * self._token_estimation_factor + OVERHEAD_PER_MSG
            # str(message["content"]): hacky way to deal with dict "content" key
            num_tokens += int(msg_tokens)

        if genconf is not None and genconf.json_schema is not None:
            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = len(js_str) * self._token_estimation_factor
            num_tokens += int(tools_num_tokens)
            # print("tools_num_tokens", tools_num_tokens)

    else: # do an "informed" token estimation from what is known of the OpenAI model's tokenization

        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            # print(message)
            num_tokens += self._overhead_per_msg
            for key, value in message.items():
                num_tokens += len(self.tokenizer.encode(str(value))) # str(value): hacky way to deal with dict "content" key

        # add extras + every reply is primed with <|start|>assistant<|message|>
        num_tokens += 32

        # print("text token_len", num_tokens)

        if genconf is not None and genconf.json_schema is not None:
            TOOLS_TOKEN_LEN_FACTOR = 1.2

            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = self.tokenizer.token_len(js_str)

            # this is an upper bound, as empirically tested with the api.
            tools_num_tokens = int(tools_num_tokens * TOOLS_TOKEN_LEN_FACTOR)
            # print("tools token_len", tools_num_tokens)

            num_tokens += tools_num_tokens


    return num_tokens

tokenizer instance-attribute #

tokenizer = OpenAITokenizer(_model_name)

ctx_len instance-attribute #

ctx_len = ctx_len or default_ctx_len

maybe_image_input instance-attribute #

maybe_image_input = True

known_models classmethod #

known_models(api_key=None)

List of model names that can be used. Some of the models are not chat models and cannot be used, for example embedding models.

Parameters:

Name Type Description Default
api_key Optional[str]

Requires OpenAI API key, passed as this arg or set in env variable OPENAI_API_KEY.

None

Returns:

Type Description
Union[list[str], None]

Returns a list of known models.

Source code in sibila/openai.py
@classmethod
def known_models(cls,
                 api_key: Optional[str] = None) -> Union[list[str], None]:
    """List of model names that can be used. Some of the models are not chat models and cannot be used,
    for example embedding models.

    Args:
        api_key: Requires OpenAI API key, passed as this arg or set in env variable OPENAI_API_KEY.

    Returns:
        Returns a list of known models.
    """

    client = openai.OpenAI(api_key=api_key)
    model_list = client.models.list()

    out = []
    for model in model_list.data:
        out.append(model.id)
    return sorted(out)

desc #

desc()

Model description.

Source code in sibila/openai.py
def desc(self) -> str:
    """Model description."""
    return f"{type(self).__name__}: '{self._model_name}'"

AnthropicModel #

AnthropicModel(
    name,
    *,
    genconf=None,
    schemaconf=None,
    ctx_len=None,
    max_tokens_limit=None,
    api_key=None,
    token_estimation_factor=None,
    anthropic_init_kwargs={}
)

Access an Anthropic model. Supports constrained JSON output, via the Anthropic API function calling mechanism.

Ref

https://docs.anthropic.com/claude/docs/intro-to-claude

Create an Anthropic remote model.

Parameters:

Name Type Description Default
name str

Model name to resolve into an existing model.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None
schemaconf Optional[JSchemaConf]

Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.

None
ctx_len Optional[int]

Maximum context length to be used (shared for input and output). None for model's default.

None
max_tokens_limit Optional[int]

Maximum output tokens limit. None for model's default.

None
api_key Optional[str]

Anthropic API key. Defaults to None, which will use env variable ANTHROPIC_API_KEY.

None
token_estimation_factor Optional[float]

Multiplication factor to estimate token usage: multiplies total text length to obtain token length.

None
anthropic_init_kwargs dict

Extra args for Anthropic() initialization. Defaults to {}.

{}

Raises:

Type Description
ImportError

If Anthropic API is not installed.

NameError

If model name was not found or there's an API or authentication problem.

Source code in sibila/anthropic.py
def __init__(self,
             name: str,
             *,

             # common base model args
             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None,
             ctx_len: Optional[int] = None,
             max_tokens_limit: Optional[int] = None,

             # most important Anthropic-specific args
             api_key: Optional[str] = None,
             token_estimation_factor: Optional[float] = None,

             # other Anthropic-specific args
             anthropic_init_kwargs: dict = {},
             ):
    """Create an Anthropic remote model.

    Args:
        name: Model name to resolve into an existing model.
        genconf: Model generation configuration. Defaults to None.
        schemaconf: Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.
        ctx_len: Maximum context length to be used (shared for input and output). None for model's default.
        max_tokens_limit: Maximum output tokens limit. None for model's default.
        api_key: Anthropic API key. Defaults to None, which will use env variable ANTHROPIC_API_KEY.
        token_estimation_factor: Multiplication factor to estimate token usage: multiplies total text length to obtain token length.
        anthropic_init_kwargs: Extra args for Anthropic() initialization. Defaults to {}.

    Raises:
        ImportError: If Anthropic API is not installed.
        NameError: If model name was not found or there's an API or authentication problem.
    """


    if not has_anthropic:
        raise ImportError("Please install anthropic API by running: pip install anthropic")

    self._client = self._client_async = None


    # also accept "provider:name" for ease of use
    provider_name = self.PROVIDER_NAME + ":"
    if name.startswith(provider_name):
        name = name[len(provider_name):]

    super().__init__(False,
                     genconf,
                     schemaconf,
                     None
                     )

    if (ctx_len is not None and
        max_tokens_limit is not None and
        token_estimation_factor is not None): # all elements given: probably created via Models.create()

        self._model_name = name
        default_ctx_len = ctx_len
        default_max_tokens_limit = max_tokens_limit
        default_token_estimation_factor = token_estimation_factor

    else: # need to resolve
        settings = self.resolve_settings(self.PROVIDER_NAME,
                                         name,
                                         ["name", 
                                          "ctx_len", 
                                          "max_tokens_limit", 
                                          "token_estimation_factor"])
        self._model_name = settings.get("name") or name
        default_ctx_len = settings.get("ctx_len") # type: ignore[assignment]
        default_max_tokens_limit = settings.get("max_tokens_limit") or default_ctx_len
        default_token_estimation_factor = settings.get("token_estimation_factor") # type: ignore[assignment]

        # all defaults are conservative values
        if default_ctx_len is None:
            default_ctx_len = 200000
            logger.warning(f"Model '{self._model_name}': unknown ctx_len, assuming {default_ctx_len}")
        if default_max_tokens_limit is None:
            default_max_tokens_limit = default_ctx_len
            logger.warning(f"Model '{self._model_name}': unknown max_tokens_limit, assuming {default_max_tokens_limit}")
        if default_token_estimation_factor is None:
            default_token_estimation_factor = self.DEFAULT_TOKEN_ESTIMATION_FACTOR
            logger.warning(f"Model '{self._model_name}': unknown token_estimation_factor, assuming {default_token_estimation_factor}")


    self.ctx_len = ctx_len or default_ctx_len

    self.max_tokens_limit = max_tokens_limit or default_max_tokens_limit

    self.max_tokens_limit = min(self.max_tokens_limit, self.ctx_len)

    self._token_estimation_factor = token_estimation_factor or default_token_estimation_factor

    self.maybe_image_input = True # currently all Anthropic models support image input - always check model specs

    # only check for "json" text presence as json schema (including field descriptions) is requested with the tools facility.
    self.json_format_instructors["json_schema"] = self.json_format_instructors["json"]

    self._client_init_kwargs = anthropic_init_kwargs

    if api_key is not None:
        self._client_init_kwargs["api_key"] = api_key    

extract #

extract(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
def extract(self,
            target: Any,

            query: Union[Thread,Msg,tuple,str],
            *,
            inst: Optional[str] = None,

            genconf: Optional[GenConf] = None,
            schemaconf: Optional[JSchemaConf] = None
            ) -> Any:        
    """Type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_extract(target,
                           thread,
                           genconf,
                           schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify #

classify(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
def classify(self,
             labels: Any,

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return self.extract(labels,
                        query,
                        inst=inst,
                        genconf=genconf,
                        schemaconf=schemaconf)

json #

json(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
def json(self,
         query: Union[Thread,Msg,tuple,str],
         *,
         json_schema: Union[dict,str,None] = None,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         massage_schema: bool = True,
         schemaconf: Optional[JSchemaConf] = None,
         ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = self.gen_json(thread,
                        json_schema,                            
                        genconf,
                        massage_schema,
                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass #

dataclass(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
def dataclass(self, # noqa: F811
              cls: Any, # a dataclass definition

              query: Union[Thread,Msg,tuple,str],
              *,
              inst: Optional[str] = None,

              genconf: Optional[GenConf] = None,
              schemaconf: Optional[JSchemaConf] = None
              ) -> Any: # a dataclass object
    """Constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_dataclass(cls,
                             thread,
                             genconf,
                             schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic #

pydantic(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
def pydantic(self,
             cls: Any, # a Pydantic BaseModel class

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any: # a Pydantic BaseModel object
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_pydantic(cls,
                            thread,
                            genconf,
                            schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call #

call(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def call(self,             
         query: Union[Thread,Msg,tuple,str],
         *,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         ok_length_is_error: bool = False
         ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen(thread=thread, 
                   genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

__call__ #

__call__(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def __call__(self,             
             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             ok_length_is_error: bool = False
             ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    return self.call(query,
                     inst=inst,
                     genconf=genconf,
                     ok_length_is_error=ok_length_is_error)

extract_async async #

extract_async(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Async type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
async def extract_async(self,
                        target: Any,

                        query: Union[Thread,Msg,tuple,str],
                        *,
                        inst: Optional[str] = None,

                        genconf: Optional[GenConf] = None,
                        schemaconf: Optional[JSchemaConf] = None
                        ) -> Any:        
    """Async type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_extract_async(target,
                                       thread,
                                       genconf,
                                       schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify_async async #

classify_async(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
async def classify_async(self,
                         labels: Any,

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return await self.extract_async(labels,
                                    query,
                                    inst=inst,
                                    genconf=genconf,
                                    schemaconf=schemaconf)

json_async async #

json_async(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
async def json_async(self,             
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     json_schema: Union[dict,str,None] = None,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     massage_schema: bool = True,
                     schemaconf: Optional[JSchemaConf] = None,
                     ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = await self.gen_json_async(thread,
                                    json_schema,
                                    genconf,
                                    massage_schema,
                                    schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass_async async #

dataclass_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
async def dataclass_async(self, # noqa: E811
                          cls: Any, # a dataclass definition

                          query: Union[Thread,Msg,tuple,str],
                          *,
                          inst: Optional[str] = None,

                          genconf: Optional[GenConf] = None,
                          schemaconf: Optional[JSchemaConf] = None
                          ) -> Any: # a dataclass object
    """Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_dataclass_async(cls,
                                         thread,
                                         genconf,
                                         schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic_async async #

pydantic_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
async def pydantic_async(self,
                         cls: Any, # a Pydantic BaseModel class

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any: # a Pydantic BaseModel object
    """Async constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_pydantic_async(cls,
                                        thread,
                                        genconf,
                                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call_async async #

call_async(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
async def call_async(self,
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     ok_length_is_error: bool = False
                     ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_async(thread=thread, 
                               genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

gen #

gen(thread, genconf=None)

Text generation from a Thread, used by the other model generation methods. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc.

GenOut

The output text is in GenOut.text.

Source code in sibila/anthropic.py
def gen(self, 
        thread: Thread,
        genconf: Optional[GenConf] = None,
        ) -> GenOut:
    """Text generation from a Thread, used by the other model generation methods.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc.
        The output text is in GenOut.text.
    """


    genconf2: GenConf
    kwargs, genconf2 = self._gen_pre(thread, genconf)

    self._ensure_client(False)

    try:
        if "tools" in kwargs:
            response = self._client.beta.tools.messages.create(**kwargs) # type: ignore[attr-defined]
        else:
            response = self._client.messages.create(**kwargs) # type: ignore[attr-defined]

    except Exception as e:
        raise RuntimeError(f"Cannot generate. Internal error: {e}")


    return self._gen_post(response,
                          kwargs,
                          genconf2)

gen_json #

gen_json(
    thread,
    json_schema,
    genconf=None,
    massage_schema=True,
    schemaconf=None,
)

JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.

Source code in sibila/model.py
def gen_json(self,
             thread: Thread,
             json_schema: Union[dict,str,None],
             genconf: Optional[GenConf] = None,

             massage_schema: bool = True,
             schemaconf: Optional[JSchemaConf] = None,
             ) -> GenOut:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.
    """

    args = self._gen_json_pre(thread,
                              json_schema,
                              genconf,
                              massage_schema,
                              schemaconf)
    return self.gen(*args)

gen_dataclass #

gen_dataclass(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a dataclass definition. An initialized dataclass object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
thread Thread

The Thread object to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.

Source code in sibila/model.py
def gen_dataclass(self,
                  cls: Any, # a dataclass
                  thread: Thread,
                  genconf: Optional[GenConf] = None,
                  schemaconf: Optional[JSchemaConf] = None
                  ) -> GenOut:
    """Constrained generation after a dataclass definition.
    An initialized dataclass object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A dataclass definition.
        thread: The Thread object to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.
    """

    schema = self._gen_dataclass_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_dataclass_post(out,
                                    cls,
                                    schemaconf)

gen_pydantic #

gen_pydantic(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a Pydantic BaseModel-derived class definition. An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

TypeError

When cls is not a Pydantic BaseClass.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.

Source code in sibila/model.py
def gen_pydantic(self,
                 cls: Any, # a Pydantic BaseModel class
                 thread: Thread,
                 genconf: Optional[GenConf] = None,
                 schemaconf: Optional[JSchemaConf] = None
                 ) -> GenOut:
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.
        TypeError: When cls is not a Pydantic BaseClass.

    Returns:
        A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.
    """

    schema = self._gen_pydantic_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_pydantic_post(out,
                                   cls,
                                   schemaconf)

token_len #

token_len(thread_or_text, genconf=None)

Calculate or estimate the token length for a Thread or a plain text string. In some cases where it's not possible to calculate the exact token count, this function should give a conservative (upper bound) estimate. It's up to the implementation whether to account for side information like JSON Schema, but it must reflect the model's context token accounting. Thread or text must be the final text which will passed to model.

Parameters:

Name Type Description Default
thread_or_text Union[Thread, str]

For token length calculation.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Returns:

Type Description
int

Estimated number of tokens occupied.

Source code in sibila/anthropic.py
def token_len(self,
              thread_or_text: Union[Thread,str],
              genconf: Optional[GenConf] = None) -> int:
    """Calculate or estimate the token length for a Thread or a plain text string.
    In some cases where it's not possible to calculate the exact token count, 
    this function should give a conservative (upper bound) estimate.
    It's up to the implementation whether to account for side information like JSON Schema,
    but it must reflect the model's context token accounting.
    Thread or text must be the final text which will passed to model.

    Args:
        thread_or_text: For token length calculation.
        genconf: Model generation configuration. Defaults to None.

    Returns:
        Estimated number of tokens occupied.
    """

    if isinstance(thread_or_text, Thread):
        thread = thread_or_text            
    else:
        thread = Thread.make_IN(thread_or_text)

    OVERHEAD_PER_MSG = 3
    num_tokens = 0
    for msg in thread.get_iter(True): # True for system message
        message = msg.as_chatml()
        msg_tokens = len(str(message["content"])) * self._token_estimation_factor + OVERHEAD_PER_MSG
        # str(message["content"]): hacky way to deal with dict "content" key
        num_tokens += int(msg_tokens)

    if genconf is not None and genconf.json_schema is not None:
        if isinstance(genconf.json_schema, str):
            js_str = genconf.json_schema
        else:
            js_str = json.dumps(genconf.json_schema)

        tools_num_tokens = len(js_str) * self._token_estimation_factor
        num_tokens += int(tools_num_tokens)
        # print("tools_num_tokens", tools_num_tokens)

    # print(num_tokens)
    return num_tokens

tokenizer instance-attribute #

tokenizer = tokenizer

Tokenizer used to encode text. Some remote models don't have tokenizer and token length is estimated

ctx_len instance-attribute #

ctx_len = ctx_len or default_ctx_len

maybe_image_input instance-attribute #

maybe_image_input = True

known_models classmethod #

known_models(api_key=None)

If the model can only use a fixed set of models, return their names. Otherwise, return None.

Parameters:

Name Type Description Default
api_key Optional[str]

If the model provider requires an API key, pass it here or set it in the respective env variable.

None

Returns:

Type Description
Union[list[str], None]

Returns a list of known models or None if unable to fetch it.

Source code in sibila/model.py
@classmethod
def known_models(cls,
                 api_key: Optional[str] = None) -> Union[list[str], None]:
    """If the model can only use a fixed set of models, return their names. Otherwise, return None.

    Args:
        api_key: If the model provider requires an API key, pass it here or set it in the respective env variable.

    Returns:
        Returns a list of known models or None if unable to fetch it.
    """
    return None

desc #

desc()

Model description.

Source code in sibila/anthropic.py
def desc(self) -> str:
    """Model description."""
    return f"AnthropicModel: {self._model_name}"

FireworksModel #

FireworksModel(
    name,
    *,
    genconf=None,
    schemaconf=None,
    ctx_len=None,
    max_tokens_limit=None,
    tokenizer=None,
    api_key=None,
    base_url=None,
    token_estimation_factor=None,
    other_init_kwargs={}
)

Access a Fireworks AI model with the OpenAI API. Supports constrained JSON output, via the response_format JSON Schema mechanism.

Ref

https://readme.fireworks.ai/docs/structured-response-formatting

https://readme.fireworks.ai/reference/createchatcompletion

Create a Fireworks AI remote model.

Parameters:

Name Type Description Default
name str

Model name to resolve into an existing model.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None
schemaconf Optional[JSchemaConf]

Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.

None
ctx_len Optional[int]

Maximum context length to be used (shared for input and output). None for model's default.

None
max_tokens_limit Optional[int]

Maximum output tokens limit. None for model's default.

None
tokenizer Optional[Tokenizer]

An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.

None
api_key Optional[str]

API key. Defaults to None, which will use env variable FIREWORKS_API_KEY.

None
base_url Optional[str]

Base location for API access. Defaults to None, which will use env variable FIREWORKS_BASE_URL or a default.

None
token_estimation_factor Optional[float]

Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.

None
other_init_kwargs dict

Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

{}

Raises:

Type Description
ImportError

If OpenAI API is not installed.

NameError

If model name was not found or there's an API or authentication problem.

Source code in sibila/schema_format_openai.py
def __init__(self,
             name: str,
             *,

             # common base model args
             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None,
             ctx_len: Optional[int] = None,
             max_tokens_limit: Optional[int] = None,
             tokenizer: Optional[Tokenizer] = None,

             # most important OpenAI API specific args
             api_key: Optional[str] = None,
             base_url: Optional[str] = None,
             token_estimation_factor: Optional[float] = None,

             # other OpenAI API specific args
             other_init_kwargs: dict = {},
             ):
    """Create a Fireworks AI remote model.

    Args:
        name: Model name to resolve into an existing model.
        genconf: Model generation configuration. Defaults to None.
        schemaconf: Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.
        ctx_len: Maximum context length to be used (shared for input and output). None for model's default.
        max_tokens_limit: Maximum output tokens limit. None for model's default.
        tokenizer: An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.
        api_key: API key. Defaults to None, which will use env variable FIREWORKS_API_KEY.
        base_url: Base location for API access. Defaults to None, which will use env variable FIREWORKS_BASE_URL or a default.
        token_estimation_factor: Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.
        other_init_kwargs: Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

    Raises:
        ImportError: If OpenAI API is not installed.
        NameError: If model name was not found or there's an API or authentication problem.
    """

    if api_key is None:
        api_key = os.environ.get("FIREWORKS_API_KEY")
    if base_url is None:
        base_url = os.environ.get("FIREWORKS_BASE_URL", self.DEFAULT_BASE_URL)

    super().__init__(name,
                     # common base model args
                     genconf=genconf,
                     schemaconf=schemaconf,
                     ctx_len=ctx_len,
                     max_tokens_limit=max_tokens_limit,
                     tokenizer=tokenizer,

                     # most important OpenAI API specific args
                     api_key=api_key,
                     base_url=base_url,
                     token_estimation_factor=token_estimation_factor,

                     # other OpenAI API specific args
                     other_init_kwargs=other_init_kwargs)

    self.maybe_image_input = False # no Fireworks models currently support image input - always check model specs

extract #

extract(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
def extract(self,
            target: Any,

            query: Union[Thread,Msg,tuple,str],
            *,
            inst: Optional[str] = None,

            genconf: Optional[GenConf] = None,
            schemaconf: Optional[JSchemaConf] = None
            ) -> Any:        
    """Type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_extract(target,
                           thread,
                           genconf,
                           schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify #

classify(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
def classify(self,
             labels: Any,

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return self.extract(labels,
                        query,
                        inst=inst,
                        genconf=genconf,
                        schemaconf=schemaconf)

json #

json(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
def json(self,
         query: Union[Thread,Msg,tuple,str],
         *,
         json_schema: Union[dict,str,None] = None,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         massage_schema: bool = True,
         schemaconf: Optional[JSchemaConf] = None,
         ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = self.gen_json(thread,
                        json_schema,                            
                        genconf,
                        massage_schema,
                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass #

dataclass(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
def dataclass(self, # noqa: F811
              cls: Any, # a dataclass definition

              query: Union[Thread,Msg,tuple,str],
              *,
              inst: Optional[str] = None,

              genconf: Optional[GenConf] = None,
              schemaconf: Optional[JSchemaConf] = None
              ) -> Any: # a dataclass object
    """Constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_dataclass(cls,
                             thread,
                             genconf,
                             schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic #

pydantic(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
def pydantic(self,
             cls: Any, # a Pydantic BaseModel class

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any: # a Pydantic BaseModel object
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_pydantic(cls,
                            thread,
                            genconf,
                            schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call #

call(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def call(self,             
         query: Union[Thread,Msg,tuple,str],
         *,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         ok_length_is_error: bool = False
         ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen(thread=thread, 
                   genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

__call__ #

__call__(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def __call__(self,             
             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             ok_length_is_error: bool = False
             ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    return self.call(query,
                     inst=inst,
                     genconf=genconf,
                     ok_length_is_error=ok_length_is_error)

extract_async async #

extract_async(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Async type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
async def extract_async(self,
                        target: Any,

                        query: Union[Thread,Msg,tuple,str],
                        *,
                        inst: Optional[str] = None,

                        genconf: Optional[GenConf] = None,
                        schemaconf: Optional[JSchemaConf] = None
                        ) -> Any:        
    """Async type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_extract_async(target,
                                       thread,
                                       genconf,
                                       schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify_async async #

classify_async(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
async def classify_async(self,
                         labels: Any,

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return await self.extract_async(labels,
                                    query,
                                    inst=inst,
                                    genconf=genconf,
                                    schemaconf=schemaconf)

json_async async #

json_async(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
async def json_async(self,             
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     json_schema: Union[dict,str,None] = None,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     massage_schema: bool = True,
                     schemaconf: Optional[JSchemaConf] = None,
                     ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = await self.gen_json_async(thread,
                                    json_schema,
                                    genconf,
                                    massage_schema,
                                    schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass_async async #

dataclass_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
async def dataclass_async(self, # noqa: E811
                          cls: Any, # a dataclass definition

                          query: Union[Thread,Msg,tuple,str],
                          *,
                          inst: Optional[str] = None,

                          genconf: Optional[GenConf] = None,
                          schemaconf: Optional[JSchemaConf] = None
                          ) -> Any: # a dataclass object
    """Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_dataclass_async(cls,
                                         thread,
                                         genconf,
                                         schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic_async async #

pydantic_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
async def pydantic_async(self,
                         cls: Any, # a Pydantic BaseModel class

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any: # a Pydantic BaseModel object
    """Async constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_pydantic_async(cls,
                                        thread,
                                        genconf,
                                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call_async async #

call_async(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
async def call_async(self,
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     ok_length_is_error: bool = False
                     ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_async(thread=thread, 
                               genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

gen #

gen(thread, genconf=None)

Text generation from a Thread, used by the other model generation methods. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc.

GenOut

The output text is in GenOut.text.

Source code in sibila/openai.py
def gen(self, 
        thread: Thread,
        genconf: Optional[GenConf] = None,
        ) -> GenOut:
    """Text generation from a Thread, used by the other model generation methods.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc.
        The output text is in GenOut.text.
    """

    genconf2: GenConf
    kwargs, genconf2 = self._gen_pre(thread, genconf)

    self._ensure_client(False)

    try:
        # https://platform.openai.com/docs/api-reference/chat/create
        response = self._client.chat.completions.create(**kwargs) # type: ignore[attr-defined]

    except Exception as e:
        raise RuntimeError(f"Cannot generate. Internal error: {e}")


    return self._gen_post(response,
                          kwargs,
                          genconf2)

gen_json #

gen_json(
    thread,
    json_schema,
    genconf=None,
    massage_schema=True,
    schemaconf=None,
)

JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.

Source code in sibila/model.py
def gen_json(self,
             thread: Thread,
             json_schema: Union[dict,str,None],
             genconf: Optional[GenConf] = None,

             massage_schema: bool = True,
             schemaconf: Optional[JSchemaConf] = None,
             ) -> GenOut:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.
    """

    args = self._gen_json_pre(thread,
                              json_schema,
                              genconf,
                              massage_schema,
                              schemaconf)
    return self.gen(*args)

gen_dataclass #

gen_dataclass(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a dataclass definition. An initialized dataclass object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
thread Thread

The Thread object to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.

Source code in sibila/model.py
def gen_dataclass(self,
                  cls: Any, # a dataclass
                  thread: Thread,
                  genconf: Optional[GenConf] = None,
                  schemaconf: Optional[JSchemaConf] = None
                  ) -> GenOut:
    """Constrained generation after a dataclass definition.
    An initialized dataclass object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A dataclass definition.
        thread: The Thread object to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.
    """

    schema = self._gen_dataclass_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_dataclass_post(out,
                                    cls,
                                    schemaconf)

gen_pydantic #

gen_pydantic(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a Pydantic BaseModel-derived class definition. An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

TypeError

When cls is not a Pydantic BaseClass.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.

Source code in sibila/model.py
def gen_pydantic(self,
                 cls: Any, # a Pydantic BaseModel class
                 thread: Thread,
                 genconf: Optional[GenConf] = None,
                 schemaconf: Optional[JSchemaConf] = None
                 ) -> GenOut:
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.
        TypeError: When cls is not a Pydantic BaseClass.

    Returns:
        A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.
    """

    schema = self._gen_pydantic_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_pydantic_post(out,
                                   cls,
                                   schemaconf)

token_len #

token_len(thread_or_text, genconf=None)

Calculate or estimate the token length for a Thread or a plain text string. In some cases where it's not possible to calculate the exact token count, this function should give a conservative (upper bound) estimate. It's up to the implementation whether to account for side information like JSON Schema, but it must reflect the model's context token accounting. Thread or text must be the final text which will passed to model.

If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

More info on calculating function_call (and tools?) tokens:

https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

Parameters:

Name Type Description Default
thread_or_text Union[Thread, str]

For token length calculation.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Returns:

Type Description
int

Estimated number of tokens used.

Source code in sibila/openai.py
def token_len(self,
              thread_or_text: Union[Thread,str],
              genconf: Optional[GenConf] = None) -> int:
    """Calculate or estimate the token length for a Thread or a plain text string.
    In some cases where it's not possible to calculate the exact token count, 
    this function should give a conservative (upper bound) estimate.
    It's up to the implementation whether to account for side information like JSON Schema,
    but it must reflect the model's context token accounting.
    Thread or text must be the final text which will passed to model.

    If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

    From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

    More info on calculating function_call (and tools?) tokens:

    https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

    https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

    Args:
        thread_or_text: For token length calculation.
        genconf: Model generation configuration. Defaults to None.

    Returns:
        Estimated number of tokens used.
    """

    if isinstance(thread_or_text, Thread):
        thread = thread_or_text            
    else:
        thread = Thread.make_IN(thread_or_text)

    num_tokens = 0

    if self.tokenizer is None: # no tokenizer was found, so we'll have to do a conservative estimate

        OVERHEAD_PER_MSG = 3
        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            msg_tokens = len(str(message["content"])) * self._token_estimation_factor + OVERHEAD_PER_MSG
            # str(message["content"]): hacky way to deal with dict "content" key
            num_tokens += int(msg_tokens)

        if genconf is not None and genconf.json_schema is not None:
            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = len(js_str) * self._token_estimation_factor
            num_tokens += int(tools_num_tokens)
            # print("tools_num_tokens", tools_num_tokens)

    else: # do an "informed" token estimation from what is known of the OpenAI model's tokenization

        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            # print(message)
            num_tokens += self._overhead_per_msg
            for key, value in message.items():
                num_tokens += len(self.tokenizer.encode(str(value))) # str(value): hacky way to deal with dict "content" key

        # add extras + every reply is primed with <|start|>assistant<|message|>
        num_tokens += 32

        # print("text token_len", num_tokens)

        if genconf is not None and genconf.json_schema is not None:
            TOOLS_TOKEN_LEN_FACTOR = 1.2

            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = self.tokenizer.token_len(js_str)

            # this is an upper bound, as empirically tested with the api.
            tools_num_tokens = int(tools_num_tokens * TOOLS_TOKEN_LEN_FACTOR)
            # print("tools token_len", tools_num_tokens)

            num_tokens += tools_num_tokens


    return num_tokens

tokenizer instance-attribute #

tokenizer = OpenAITokenizer(_model_name)

ctx_len instance-attribute #

ctx_len = ctx_len or default_ctx_len

maybe_image_input instance-attribute #

maybe_image_input = False

known_models classmethod #

known_models(api_key=None)

List of model names that can be used. Some of the models are not chat models and cannot be used, for example embedding models.

Parameters:

Name Type Description Default
api_key Optional[str]

If the model provider requires an API key, pass it here or set it in the respective env variable.

None

Returns:

Type Description
Union[list[str], None]

Returns a list of known models or None if unable to fetch it.

Source code in sibila/schema_format_openai.py
@classmethod
def known_models(cls,
                 api_key: Optional[str] = None) -> Union[list[str], None]:
    """List of model names that can be used. Some of the models are not chat models and cannot be used,
    for example embedding models.

    Args:
        api_key: If the model provider requires an API key, pass it here or set it in the respective env variable.

    Returns:
        Returns a list of known models or None if unable to fetch it.
    """
    return None

desc #

desc()

Model description.

Source code in sibila/openai.py
def desc(self) -> str:
    """Model description."""
    return f"{type(self).__name__}: '{self._model_name}'"

GroqModel #

GroqModel(
    name,
    *,
    genconf=None,
    schemaconf=None,
    ctx_len=None,
    max_tokens_limit=None,
    tokenizer=None,
    api_key=None,
    base_url=None,
    token_estimation_factor=None,
    other_init_kwargs={}
)

Access a Groq model with the OpenAI API. Supports constrained JSON output, via the response_format JSON Schema mechanism.

Ref

https://console.groq.com/docs/tool-use

https://github.com/groq/groq-api-cookbook/blob/main/parallel-tool-use/parallel-tool-use.ipynb

Create a Groq remote model.

Parameters:

Name Type Description Default
name str

Model name to resolve into an existing model.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None
schemaconf Optional[JSchemaConf]

Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.

None
ctx_len Optional[int]

Maximum context length to be used (shared for input and output). None for model's default.

None
max_tokens_limit Optional[int]

Maximum output tokens limit. None for model's default.

None
tokenizer Optional[Tokenizer]

An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.

None
api_key Optional[str]

API key. Defaults to None, which will use env variable GROQ_API_KEY.

None
base_url Optional[str]

Base location for API access. Defaults to None, which will use env variable GROQ_BASE_URL or a default.

None
token_estimation_factor Optional[float]

Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.

None
other_init_kwargs dict

Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

{}

Raises:

Type Description
ImportError

If OpenAI API is not installed.

NameError

If model name was not found or there's an API or authentication problem.

Source code in sibila/schema_format_openai.py
def __init__(self,
             name: str,
             *,

             # common base model args
             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None,
             ctx_len: Optional[int] = None,
             max_tokens_limit: Optional[int] = None,
             tokenizer: Optional[Tokenizer] = None,

             # most important OpenAI API specific args
             api_key: Optional[str] = None,
             base_url: Optional[str] = None,
             token_estimation_factor: Optional[float] = None,

             # other OpenAI API specific args
             other_init_kwargs: dict = {},
             ):
    """Create a Groq remote model.

    Args:
        name: Model name to resolve into an existing model.
        genconf: Model generation configuration. Defaults to None.
        schemaconf: Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.
        ctx_len: Maximum context length to be used (shared for input and output). None for model's default.
        max_tokens_limit: Maximum output tokens limit. None for model's default.
        tokenizer: An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.
        api_key: API key. Defaults to None, which will use env variable GROQ_API_KEY.
        base_url: Base location for API access. Defaults to None, which will use env variable GROQ_BASE_URL or a default.
        token_estimation_factor: Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.
        other_init_kwargs: Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

    Raises:
        ImportError: If OpenAI API is not installed.
        NameError: If model name was not found or there's an API or authentication problem.
    """

    if api_key is None:
        api_key = os.environ.get("GROQ_API_KEY")
    if base_url is None:
        base_url = os.environ.get("GROQ_BASE_URL", self.DEFAULT_BASE_URL)

    super().__init__(name,
                     # common base model args
                     genconf=genconf,
                     schemaconf=schemaconf,
                     ctx_len=ctx_len,
                     max_tokens_limit=max_tokens_limit,
                     tokenizer=tokenizer,

                     # most important OpenAI API specific args
                     api_key=api_key,
                     base_url=base_url,
                     token_estimation_factor=token_estimation_factor,

                     # other OpenAI API specific args
                     other_init_kwargs=other_init_kwargs)

    self.maybe_image_input = False # no Groq models currently support image input - always check model specs

extract #

extract(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
def extract(self,
            target: Any,

            query: Union[Thread,Msg,tuple,str],
            *,
            inst: Optional[str] = None,

            genconf: Optional[GenConf] = None,
            schemaconf: Optional[JSchemaConf] = None
            ) -> Any:        
    """Type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_extract(target,
                           thread,
                           genconf,
                           schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify #

classify(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
def classify(self,
             labels: Any,

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return self.extract(labels,
                        query,
                        inst=inst,
                        genconf=genconf,
                        schemaconf=schemaconf)

json #

json(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
def json(self,
         query: Union[Thread,Msg,tuple,str],
         *,
         json_schema: Union[dict,str,None] = None,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         massage_schema: bool = True,
         schemaconf: Optional[JSchemaConf] = None,
         ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = self.gen_json(thread,
                        json_schema,                            
                        genconf,
                        massage_schema,
                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass #

dataclass(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
def dataclass(self, # noqa: F811
              cls: Any, # a dataclass definition

              query: Union[Thread,Msg,tuple,str],
              *,
              inst: Optional[str] = None,

              genconf: Optional[GenConf] = None,
              schemaconf: Optional[JSchemaConf] = None
              ) -> Any: # a dataclass object
    """Constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_dataclass(cls,
                             thread,
                             genconf,
                             schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic #

pydantic(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
def pydantic(self,
             cls: Any, # a Pydantic BaseModel class

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any: # a Pydantic BaseModel object
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_pydantic(cls,
                            thread,
                            genconf,
                            schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call #

call(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def call(self,             
         query: Union[Thread,Msg,tuple,str],
         *,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         ok_length_is_error: bool = False
         ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen(thread=thread, 
                   genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

__call__ #

__call__(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def __call__(self,             
             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             ok_length_is_error: bool = False
             ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    return self.call(query,
                     inst=inst,
                     genconf=genconf,
                     ok_length_is_error=ok_length_is_error)

extract_async async #

extract_async(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Async type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
async def extract_async(self,
                        target: Any,

                        query: Union[Thread,Msg,tuple,str],
                        *,
                        inst: Optional[str] = None,

                        genconf: Optional[GenConf] = None,
                        schemaconf: Optional[JSchemaConf] = None
                        ) -> Any:        
    """Async type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_extract_async(target,
                                       thread,
                                       genconf,
                                       schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify_async async #

classify_async(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
async def classify_async(self,
                         labels: Any,

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return await self.extract_async(labels,
                                    query,
                                    inst=inst,
                                    genconf=genconf,
                                    schemaconf=schemaconf)

json_async async #

json_async(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
async def json_async(self,             
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     json_schema: Union[dict,str,None] = None,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     massage_schema: bool = True,
                     schemaconf: Optional[JSchemaConf] = None,
                     ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = await self.gen_json_async(thread,
                                    json_schema,
                                    genconf,
                                    massage_schema,
                                    schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass_async async #

dataclass_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
async def dataclass_async(self, # noqa: E811
                          cls: Any, # a dataclass definition

                          query: Union[Thread,Msg,tuple,str],
                          *,
                          inst: Optional[str] = None,

                          genconf: Optional[GenConf] = None,
                          schemaconf: Optional[JSchemaConf] = None
                          ) -> Any: # a dataclass object
    """Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_dataclass_async(cls,
                                         thread,
                                         genconf,
                                         schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic_async async #

pydantic_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
async def pydantic_async(self,
                         cls: Any, # a Pydantic BaseModel class

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any: # a Pydantic BaseModel object
    """Async constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_pydantic_async(cls,
                                        thread,
                                        genconf,
                                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call_async async #

call_async(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
async def call_async(self,
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     ok_length_is_error: bool = False
                     ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_async(thread=thread, 
                               genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

gen #

gen(thread, genconf=None)

Text generation from a Thread, used by the other model generation methods. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc.

GenOut

The output text is in GenOut.text.

Source code in sibila/openai.py
def gen(self, 
        thread: Thread,
        genconf: Optional[GenConf] = None,
        ) -> GenOut:
    """Text generation from a Thread, used by the other model generation methods.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc.
        The output text is in GenOut.text.
    """

    genconf2: GenConf
    kwargs, genconf2 = self._gen_pre(thread, genconf)

    self._ensure_client(False)

    try:
        # https://platform.openai.com/docs/api-reference/chat/create
        response = self._client.chat.completions.create(**kwargs) # type: ignore[attr-defined]

    except Exception as e:
        raise RuntimeError(f"Cannot generate. Internal error: {e}")


    return self._gen_post(response,
                          kwargs,
                          genconf2)

gen_json #

gen_json(
    thread,
    json_schema,
    genconf=None,
    massage_schema=True,
    schemaconf=None,
)

JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.

Source code in sibila/model.py
def gen_json(self,
             thread: Thread,
             json_schema: Union[dict,str,None],
             genconf: Optional[GenConf] = None,

             massage_schema: bool = True,
             schemaconf: Optional[JSchemaConf] = None,
             ) -> GenOut:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.
    """

    args = self._gen_json_pre(thread,
                              json_schema,
                              genconf,
                              massage_schema,
                              schemaconf)
    return self.gen(*args)

gen_dataclass #

gen_dataclass(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a dataclass definition. An initialized dataclass object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
thread Thread

The Thread object to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.

Source code in sibila/model.py
def gen_dataclass(self,
                  cls: Any, # a dataclass
                  thread: Thread,
                  genconf: Optional[GenConf] = None,
                  schemaconf: Optional[JSchemaConf] = None
                  ) -> GenOut:
    """Constrained generation after a dataclass definition.
    An initialized dataclass object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A dataclass definition.
        thread: The Thread object to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.
    """

    schema = self._gen_dataclass_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_dataclass_post(out,
                                    cls,
                                    schemaconf)

gen_pydantic #

gen_pydantic(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a Pydantic BaseModel-derived class definition. An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

TypeError

When cls is not a Pydantic BaseClass.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.

Source code in sibila/model.py
def gen_pydantic(self,
                 cls: Any, # a Pydantic BaseModel class
                 thread: Thread,
                 genconf: Optional[GenConf] = None,
                 schemaconf: Optional[JSchemaConf] = None
                 ) -> GenOut:
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.
        TypeError: When cls is not a Pydantic BaseClass.

    Returns:
        A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.
    """

    schema = self._gen_pydantic_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_pydantic_post(out,
                                   cls,
                                   schemaconf)

token_len #

token_len(thread_or_text, genconf=None)

Calculate or estimate the token length for a Thread or a plain text string. In some cases where it's not possible to calculate the exact token count, this function should give a conservative (upper bound) estimate. It's up to the implementation whether to account for side information like JSON Schema, but it must reflect the model's context token accounting. Thread or text must be the final text which will passed to model.

If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

More info on calculating function_call (and tools?) tokens:

https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

Parameters:

Name Type Description Default
thread_or_text Union[Thread, str]

For token length calculation.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Returns:

Type Description
int

Estimated number of tokens used.

Source code in sibila/openai.py
def token_len(self,
              thread_or_text: Union[Thread,str],
              genconf: Optional[GenConf] = None) -> int:
    """Calculate or estimate the token length for a Thread or a plain text string.
    In some cases where it's not possible to calculate the exact token count, 
    this function should give a conservative (upper bound) estimate.
    It's up to the implementation whether to account for side information like JSON Schema,
    but it must reflect the model's context token accounting.
    Thread or text must be the final text which will passed to model.

    If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

    From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

    More info on calculating function_call (and tools?) tokens:

    https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

    https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

    Args:
        thread_or_text: For token length calculation.
        genconf: Model generation configuration. Defaults to None.

    Returns:
        Estimated number of tokens used.
    """

    if isinstance(thread_or_text, Thread):
        thread = thread_or_text            
    else:
        thread = Thread.make_IN(thread_or_text)

    num_tokens = 0

    if self.tokenizer is None: # no tokenizer was found, so we'll have to do a conservative estimate

        OVERHEAD_PER_MSG = 3
        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            msg_tokens = len(str(message["content"])) * self._token_estimation_factor + OVERHEAD_PER_MSG
            # str(message["content"]): hacky way to deal with dict "content" key
            num_tokens += int(msg_tokens)

        if genconf is not None and genconf.json_schema is not None:
            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = len(js_str) * self._token_estimation_factor
            num_tokens += int(tools_num_tokens)
            # print("tools_num_tokens", tools_num_tokens)

    else: # do an "informed" token estimation from what is known of the OpenAI model's tokenization

        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            # print(message)
            num_tokens += self._overhead_per_msg
            for key, value in message.items():
                num_tokens += len(self.tokenizer.encode(str(value))) # str(value): hacky way to deal with dict "content" key

        # add extras + every reply is primed with <|start|>assistant<|message|>
        num_tokens += 32

        # print("text token_len", num_tokens)

        if genconf is not None and genconf.json_schema is not None:
            TOOLS_TOKEN_LEN_FACTOR = 1.2

            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = self.tokenizer.token_len(js_str)

            # this is an upper bound, as empirically tested with the api.
            tools_num_tokens = int(tools_num_tokens * TOOLS_TOKEN_LEN_FACTOR)
            # print("tools token_len", tools_num_tokens)

            num_tokens += tools_num_tokens


    return num_tokens

tokenizer instance-attribute #

tokenizer = OpenAITokenizer(_model_name)

ctx_len instance-attribute #

ctx_len = ctx_len or default_ctx_len

maybe_image_input instance-attribute #

maybe_image_input = False

known_models classmethod #

known_models(api_key=None)

List of model names that can be used. Some of the models are not chat models and cannot be used, for example embedding models.

Parameters:

Name Type Description Default
api_key Optional[str]

If the model provider requires an API key, pass it here or set it in the respective env variable.

None

Returns:

Type Description
Union[list[str], None]

Returns a list of known models or None if unable to fetch it.

Source code in sibila/schema_format_openai.py
@classmethod
def known_models(cls,
                 api_key: Optional[str] = None) -> Union[list[str], None]:
    """List of model names that can be used. Some of the models are not chat models and cannot be used,
    for example embedding models.

    Args:
        api_key: If the model provider requires an API key, pass it here or set it in the respective env variable.

    Returns:
        Returns a list of known models or None if unable to fetch it.
    """
    return None

desc #

desc()

Model description.

Source code in sibila/openai.py
def desc(self) -> str:
    """Model description."""
    return f"{type(self).__name__}: '{self._model_name}'"

MistralModel #

MistralModel(
    name,
    *,
    genconf=None,
    schemaconf=None,
    ctx_len=None,
    max_tokens_limit=None,
    api_key=None,
    token_estimation_factor=None,
    mistral_init_kwargs={}
)

Access a Mistral AI model. Supports constrained JSON output, via the Mistral API function calling mechanism.

Ref

https://docs.mistral.ai/guides/function-calling/

Create a Mistral AI remote model.

Parameters:

Name Type Description Default
name str

Model name to resolve into an existing model.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None
schemaconf Optional[JSchemaConf]

Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.

None
ctx_len Optional[int]

Maximum context length to be used (shared for input and output). None for model's default.

None
max_tokens_limit Optional[int]

Maximum output tokens limit. None for model's default.

None
api_key Optional[str]

Mistral API key. Defaults to None, which will use env variable MISTRAL_API_KEY.

None
token_estimation_factor Optional[float]

Multiplication factor to estimate token usage: multiplies total text length to obtain token length.

None
mistral_init_kwargs dict

Extra args for mistral.MistralClient() initialization. Defaults to {}.

{}

Raises:

Type Description
ImportError

If Mistral API is not installed.

NameError

If model name was not found or there's an API or authentication problem.

Source code in sibila/mistral.py
def __init__(self,
             name: str,
             *,

             # common base model args
             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None,
             ctx_len: Optional[int] = None,
             max_tokens_limit: Optional[int] = None,

             # most important Mistral-specific args
             api_key: Optional[str] = None,
             token_estimation_factor: Optional[float] = None,

             # other Mistral-specific args
             mistral_init_kwargs: dict = {},
             ):
    """Create a Mistral AI remote model.

    Args:
        name: Model name to resolve into an existing model.
        genconf: Model generation configuration. Defaults to None.
        schemaconf: Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.
        ctx_len: Maximum context length to be used (shared for input and output). None for model's default.
        max_tokens_limit: Maximum output tokens limit. None for model's default.
        api_key: Mistral API key. Defaults to None, which will use env variable MISTRAL_API_KEY.
        token_estimation_factor: Multiplication factor to estimate token usage: multiplies total text length to obtain token length.
        mistral_init_kwargs: Extra args for mistral.MistralClient() initialization. Defaults to {}.

    Raises:
        ImportError: If Mistral API is not installed.
        NameError: If model name was not found or there's an API or authentication problem.
    """


    if not has_mistral:
        raise ImportError("Please install mistral by running: pip install mistralai")

    self._client = self._client_async = None


    # also accept "provider:name" for ease of use
    provider_name = self.PROVIDER_NAME + ":"
    if name.startswith(provider_name):
        name = name[len(provider_name):]

    super().__init__(False,
                     genconf,
                     schemaconf,
                     None
                     )

    if (ctx_len is not None and
        max_tokens_limit is not None and
        token_estimation_factor is not None): # all elements given: probably created via Models.create()

        self._model_name = name
        default_ctx_len = ctx_len
        default_max_tokens_limit = max_tokens_limit
        default_token_estimation_factor = token_estimation_factor

    else: # need to resolve
        settings = self.resolve_settings(self.PROVIDER_NAME,
                                         name,
                                         ["name", 
                                          "ctx_len", 
                                          "max_tokens_limit", 
                                          "token_estimation_factor"])
        self._model_name = settings.get("name") or name
        default_ctx_len = settings.get("ctx_len") # type: ignore[assignment]
        default_max_tokens_limit = settings.get("max_tokens_limit") or default_ctx_len
        default_token_estimation_factor = settings.get("token_estimation_factor") # type: ignore[assignment]

        # all defaults are conservative values
        if default_ctx_len is None:
            default_ctx_len = 32768
            logger.warning(f"Model '{self._model_name}': unknown ctx_len, assuming {default_ctx_len}")
        if default_max_tokens_limit is None:
            default_max_tokens_limit = default_ctx_len
            logger.warning(f"Model '{self._model_name}': unknown max_tokens_limit, assuming {default_max_tokens_limit}")
        if default_token_estimation_factor is None:
            default_token_estimation_factor = self.DEFAULT_TOKEN_ESTIMATION_FACTOR
            logger.warning(f"Model '{self._model_name}': unknown token_estimation_factor, assuming {default_token_estimation_factor}")


    self.ctx_len = ctx_len or default_ctx_len

    self.max_tokens_limit = max_tokens_limit or default_max_tokens_limit

    self.max_tokens_limit = min(self.max_tokens_limit, self.ctx_len)

    self._token_estimation_factor = token_estimation_factor or default_token_estimation_factor

    self.maybe_image_input = False # no Mistral models currently support image input - always check model specs

    # only check for "json" text presence as json schema (including field descriptions) is requested with the tools facility.
    self.json_format_instructors["json_schema"] = self.json_format_instructors["json"]

    self._client_init_kwargs = mistral_init_kwargs

    if api_key is not None:
        self._client_init_kwargs["api_key"] = api_key    
    elif "api_key" not in self._client_init_kwargs and "MISTRAL_API_KEY" in os.environ:
        # "MISTRAL_API_KEY" env key is ignored in pytest?
        self._client_init_kwargs["api_key"] = os.environ["MISTRAL_API_KEY"]

extract #

extract(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
def extract(self,
            target: Any,

            query: Union[Thread,Msg,tuple,str],
            *,
            inst: Optional[str] = None,

            genconf: Optional[GenConf] = None,
            schemaconf: Optional[JSchemaConf] = None
            ) -> Any:        
    """Type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_extract(target,
                           thread,
                           genconf,
                           schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify #

classify(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
def classify(self,
             labels: Any,

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return self.extract(labels,
                        query,
                        inst=inst,
                        genconf=genconf,
                        schemaconf=schemaconf)

json #

json(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
def json(self,
         query: Union[Thread,Msg,tuple,str],
         *,
         json_schema: Union[dict,str,None] = None,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         massage_schema: bool = True,
         schemaconf: Optional[JSchemaConf] = None,
         ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = self.gen_json(thread,
                        json_schema,                            
                        genconf,
                        massage_schema,
                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass #

dataclass(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
def dataclass(self, # noqa: F811
              cls: Any, # a dataclass definition

              query: Union[Thread,Msg,tuple,str],
              *,
              inst: Optional[str] = None,

              genconf: Optional[GenConf] = None,
              schemaconf: Optional[JSchemaConf] = None
              ) -> Any: # a dataclass object
    """Constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_dataclass(cls,
                             thread,
                             genconf,
                             schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic #

pydantic(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
def pydantic(self,
             cls: Any, # a Pydantic BaseModel class

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any: # a Pydantic BaseModel object
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_pydantic(cls,
                            thread,
                            genconf,
                            schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call #

call(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def call(self,             
         query: Union[Thread,Msg,tuple,str],
         *,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         ok_length_is_error: bool = False
         ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen(thread=thread, 
                   genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

__call__ #

__call__(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def __call__(self,             
             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             ok_length_is_error: bool = False
             ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    return self.call(query,
                     inst=inst,
                     genconf=genconf,
                     ok_length_is_error=ok_length_is_error)

extract_async async #

extract_async(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Async type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
async def extract_async(self,
                        target: Any,

                        query: Union[Thread,Msg,tuple,str],
                        *,
                        inst: Optional[str] = None,

                        genconf: Optional[GenConf] = None,
                        schemaconf: Optional[JSchemaConf] = None
                        ) -> Any:        
    """Async type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_extract_async(target,
                                       thread,
                                       genconf,
                                       schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify_async async #

classify_async(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
async def classify_async(self,
                         labels: Any,

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return await self.extract_async(labels,
                                    query,
                                    inst=inst,
                                    genconf=genconf,
                                    schemaconf=schemaconf)

json_async async #

json_async(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
async def json_async(self,             
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     json_schema: Union[dict,str,None] = None,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     massage_schema: bool = True,
                     schemaconf: Optional[JSchemaConf] = None,
                     ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = await self.gen_json_async(thread,
                                    json_schema,
                                    genconf,
                                    massage_schema,
                                    schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass_async async #

dataclass_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
async def dataclass_async(self, # noqa: E811
                          cls: Any, # a dataclass definition

                          query: Union[Thread,Msg,tuple,str],
                          *,
                          inst: Optional[str] = None,

                          genconf: Optional[GenConf] = None,
                          schemaconf: Optional[JSchemaConf] = None
                          ) -> Any: # a dataclass object
    """Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_dataclass_async(cls,
                                         thread,
                                         genconf,
                                         schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic_async async #

pydantic_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
async def pydantic_async(self,
                         cls: Any, # a Pydantic BaseModel class

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any: # a Pydantic BaseModel object
    """Async constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_pydantic_async(cls,
                                        thread,
                                        genconf,
                                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call_async async #

call_async(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
async def call_async(self,
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     ok_length_is_error: bool = False
                     ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_async(thread=thread, 
                               genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

gen #

gen(thread, genconf=None)

Text generation from a Thread, used by the other model generation methods. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc.

GenOut

The output text is in GenOut.text.

Source code in sibila/mistral.py
def gen(self, 
        thread: Thread,
        genconf: Optional[GenConf] = None,
        ) -> GenOut:
    """Text generation from a Thread, used by the other model generation methods.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc.
        The output text is in GenOut.text.
    """


    genconf2: GenConf
    kwargs, genconf2 = self._gen_pre(thread, genconf)

    self._ensure_client(False)

    try:
        response = self._client.chat(**kwargs) # type: ignore[attr-defined]

    except Exception as e:
        raise RuntimeError(f"Cannot generate. Internal error: {e}")


    return self._gen_post(response,
                          kwargs,
                          genconf2)

gen_json #

gen_json(
    thread,
    json_schema,
    genconf=None,
    massage_schema=True,
    schemaconf=None,
)

JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.

Source code in sibila/model.py
def gen_json(self,
             thread: Thread,
             json_schema: Union[dict,str,None],
             genconf: Optional[GenConf] = None,

             massage_schema: bool = True,
             schemaconf: Optional[JSchemaConf] = None,
             ) -> GenOut:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.
    """

    args = self._gen_json_pre(thread,
                              json_schema,
                              genconf,
                              massage_schema,
                              schemaconf)
    return self.gen(*args)

gen_dataclass #

gen_dataclass(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a dataclass definition. An initialized dataclass object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
thread Thread

The Thread object to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.

Source code in sibila/model.py
def gen_dataclass(self,
                  cls: Any, # a dataclass
                  thread: Thread,
                  genconf: Optional[GenConf] = None,
                  schemaconf: Optional[JSchemaConf] = None
                  ) -> GenOut:
    """Constrained generation after a dataclass definition.
    An initialized dataclass object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A dataclass definition.
        thread: The Thread object to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.
    """

    schema = self._gen_dataclass_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_dataclass_post(out,
                                    cls,
                                    schemaconf)

gen_pydantic #

gen_pydantic(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a Pydantic BaseModel-derived class definition. An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

TypeError

When cls is not a Pydantic BaseClass.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.

Source code in sibila/model.py
def gen_pydantic(self,
                 cls: Any, # a Pydantic BaseModel class
                 thread: Thread,
                 genconf: Optional[GenConf] = None,
                 schemaconf: Optional[JSchemaConf] = None
                 ) -> GenOut:
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.
        TypeError: When cls is not a Pydantic BaseClass.

    Returns:
        A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.
    """

    schema = self._gen_pydantic_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_pydantic_post(out,
                                   cls,
                                   schemaconf)

token_len #

token_len(thread_or_text, genconf=None)

Calculate or estimate the token length for a Thread or a plain text string. In some cases where it's not possible to calculate the exact token count, this function should give a conservative (upper bound) estimate. It's up to the implementation whether to account for side information like JSON Schema, but it must reflect the model's context token accounting. Thread or text must be the final text which will passed to model.

Parameters:

Name Type Description Default
thread_or_text Union[Thread, str]

For token length calculation.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Returns:

Type Description
int

Estimated number of tokens occupied.

Source code in sibila/mistral.py
def token_len(self,
              thread_or_text: Union[Thread,str],
              genconf: Optional[GenConf] = None) -> int:
    """Calculate or estimate the token length for a Thread or a plain text string.
    In some cases where it's not possible to calculate the exact token count, 
    this function should give a conservative (upper bound) estimate.
    It's up to the implementation whether to account for side information like JSON Schema,
    but it must reflect the model's context token accounting.
    Thread or text must be the final text which will passed to model.

    Args:
        thread_or_text: For token length calculation.
        genconf: Model generation configuration. Defaults to None.

    Returns:
        Estimated number of tokens occupied.
    """

    if isinstance(thread_or_text, Thread):
        thread = thread_or_text            
    else:
        thread = Thread.make_IN(thread_or_text)

    OVERHEAD_PER_MSG = 3
    num_tokens = 0
    for msg in thread.get_iter(True): # True for system message
        message = msg.as_chatml()
        msg_tokens = len(str(message["content"])) * self._token_estimation_factor + OVERHEAD_PER_MSG
        # str(message["content"]): hacky way to deal with dict "content" key
        num_tokens += int(msg_tokens)

    if genconf is not None and genconf.json_schema is not None:
        if isinstance(genconf.json_schema, str):
            js_str = genconf.json_schema
        else:
            js_str = json.dumps(genconf.json_schema)

        tools_num_tokens = len(js_str) * self._token_estimation_factor
        num_tokens += int(tools_num_tokens)
        # print("tools_num_tokens", tools_num_tokens)

    # print(num_tokens)
    return num_tokens

tokenizer instance-attribute #

tokenizer = tokenizer

Tokenizer used to encode text. Some remote models don't have tokenizer and token length is estimated

ctx_len instance-attribute #

ctx_len = ctx_len or default_ctx_len

maybe_image_input instance-attribute #

maybe_image_input = False

known_models classmethod #

known_models(api_key=None)

If the model can only use a fixed set of models, return their names. Otherwise, return None.

Parameters:

Name Type Description Default
api_key Optional[str]

If the model provider requires an API key, pass it here or set it in the respective env variable.

None

Returns:

Type Description
Union[list[str], None]

Returns a list of known models or None if unable to fetch it.

Source code in sibila/mistral.py
@classmethod
def known_models(cls,
                 api_key: Optional[str] = None) -> Union[list[str], None]:
    """If the model can only use a fixed set of models, return their names. Otherwise, return None.

    Args:
        api_key: If the model provider requires an API key, pass it here or set it in the respective env variable.

    Returns:
        Returns a list of known models or None if unable to fetch it.
    """

    args = {}
    if api_key is not None:
        args["api_key"] = api_key
    model = MistralClient(**args) # type: ignore[arg-type]

    model_list = model.list_models()
    del model

    out = []
    for mod in model_list.data:
        out.append(mod.id)

    return sorted(out)

desc #

desc()

Model description.

Source code in sibila/mistral.py
def desc(self) -> str:
    """Model description."""
    return f"MistralModel: {self._model_name}"

TogetherModel #

TogetherModel(
    name,
    *,
    genconf=None,
    schemaconf=None,
    ctx_len=None,
    max_tokens_limit=None,
    tokenizer=None,
    api_key=None,
    base_url=None,
    token_estimation_factor=None,
    other_init_kwargs={}
)

Access a together.ai model with the OpenAI API. Supports constrained JSON output, via the response_format JSON Schema mechanism.

Ref

https://docs.together.ai/docs/json-mode

https://docs.together.ai/reference/chat-completions

Create a together.ai remote model.

Parameters:

Name Type Description Default
name str

Model name to resolve into an existing model.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None
schemaconf Optional[JSchemaConf]

Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.

None
ctx_len Optional[int]

Maximum context length to be used (shared for input and output). None for model's default.

None
max_tokens_limit Optional[int]

Maximum output tokens limit. None for model's default.

None
tokenizer Optional[Tokenizer]

An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.

None
api_key Optional[str]

API key. Defaults to None, which will use env variable TOGETHER_API_KEY.

None
base_url Optional[str]

Base location for API access. Defaults to None, which will use env variable TOGETHER_BASE_URL or a default.

None
token_estimation_factor Optional[float]

Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.

None
other_init_kwargs dict

Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

{}

Raises:

Type Description
ImportError

If OpenAI API is not installed.

NameError

If model name was not found or there's an API or authentication problem.

Source code in sibila/schema_format_openai.py
def __init__(self,
             name: str,
             *,

             # common base model args
             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None,
             ctx_len: Optional[int] = None,
             max_tokens_limit: Optional[int] = None,
             tokenizer: Optional[Tokenizer] = None,

             # most important OpenAI API specific args
             api_key: Optional[str] = None,
             base_url: Optional[str] = None,
             token_estimation_factor: Optional[float] = None,

             # other OpenAI API specific args
             other_init_kwargs: dict = {},
             ):
    """Create a together.ai remote model.

    Args:
        name: Model name to resolve into an existing model.
        genconf: Model generation configuration. Defaults to None.
        schemaconf: Default configuration for JSON schema validation, used if generation call doesn't supply one. Defaults to None.
        ctx_len: Maximum context length to be used (shared for input and output). None for model's default.
        max_tokens_limit: Maximum output tokens limit. None for model's default.
        tokenizer: An external initialized tokenizer to use instead of the created from the GGUF file. Defaults to None.
        api_key: API key. Defaults to None, which will use env variable TOGETHER_API_KEY.
        base_url: Base location for API access. Defaults to None, which will use env variable TOGETHER_BASE_URL or a default.
        token_estimation_factor: Used when no tokenizer is available. Multiplication factor to estimate token usage: multiplies total text length to obtain token length.
        other_init_kwargs: Extra args for OpenAI.OpenAI() initialization. Defaults to {}.

    Raises:
        ImportError: If OpenAI API is not installed.
        NameError: If model name was not found or there's an API or authentication problem.
    """

    if api_key is None:
        api_key = os.environ.get("TOGETHER_API_KEY")
    if base_url is None:
        base_url = os.environ.get("TOGETHER_BASE_URL", self.DEFAULT_BASE_URL)

    super().__init__(name,
                     # common base model args
                     genconf=genconf,
                     schemaconf=schemaconf,
                     ctx_len=ctx_len,
                     max_tokens_limit=max_tokens_limit,
                     tokenizer=tokenizer,

                     # most important OpenAI API specific args
                     api_key=api_key,
                     base_url=base_url,
                     token_estimation_factor=token_estimation_factor,

                     # other OpenAI API specific args
                     other_init_kwargs=other_init_kwargs)

    self.maybe_image_input = False # no together.ai models currently support image input - always check model specs

extract #

extract(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
def extract(self,
            target: Any,

            query: Union[Thread,Msg,tuple,str],
            *,
            inst: Optional[str] = None,

            genconf: Optional[GenConf] = None,
            schemaconf: Optional[JSchemaConf] = None
            ) -> Any:        
    """Type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_extract(target,
                           thread,
                           genconf,
                           schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify #

classify(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
def classify(self,
             labels: Any,

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return self.extract(labels,
                        query,
                        inst=inst,
                        genconf=genconf,
                        schemaconf=schemaconf)

json #

json(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
def json(self,
         query: Union[Thread,Msg,tuple,str],
         *,
         json_schema: Union[dict,str,None] = None,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         massage_schema: bool = True,
         schemaconf: Optional[JSchemaConf] = None,
         ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = self.gen_json(thread,
                        json_schema,                            
                        genconf,
                        massage_schema,
                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass #

dataclass(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
def dataclass(self, # noqa: F811
              cls: Any, # a dataclass definition

              query: Union[Thread,Msg,tuple,str],
              *,
              inst: Optional[str] = None,

              genconf: Optional[GenConf] = None,
              schemaconf: Optional[JSchemaConf] = None
              ) -> Any: # a dataclass object
    """Constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_dataclass(cls,
                             thread,
                             genconf,
                             schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic #

pydantic(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
def pydantic(self,
             cls: Any, # a Pydantic BaseModel class

             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             schemaconf: Optional[JSchemaConf] = None
             ) -> Any: # a Pydantic BaseModel object
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen_pydantic(cls,
                            thread,
                            genconf,
                            schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call #

call(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def call(self,             
         query: Union[Thread,Msg,tuple,str],
         *,
         inst: Optional[str] = None,

         genconf: Optional[GenConf] = None,
         ok_length_is_error: bool = False
         ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = self.gen(thread=thread, 
                   genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

__call__ #

__call__(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
def __call__(self,             
             query: Union[Thread,Msg,tuple,str],
             *,
             inst: Optional[str] = None,

             genconf: Optional[GenConf] = None,
             ok_length_is_error: bool = False
             ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods. Same as call().

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    return self.call(query,
                     inst=inst,
                     genconf=genconf,
                     ok_length_is_error=ok_length_is_error)

extract_async async #

extract_async(
    target,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Async type-constrained generation: an instance of the given type will be initialized with the model's output. The following target types are accepted:

  • prim_type:

    • bool
    • int
    • float
    • str
  • enums:

    • [1, 2, 3] or ["a","b"] - all items of the same prim_type
    • Literal['year', 'name'] - all items of the same prim_type
    • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type
  • datetime/date/time

  • a list in the form:

    • list[type]

    For example list[int]. The list can be annotated: Annotated[list[T], "List desc"] And/or the list item type can be annotated: list[Annotated[T, "Item desc"]]

  • dataclass with fields of the above supported types (or dataclass).

  • Pydantic BaseModel

All types can be Annotated[T, "Desc"], for example: count: int Can be annotated as: count: Annotated[int, "How many units?"]

Parameters:

Name Type Description Default
target Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A value of target arg type instantiated with the model's output.

Source code in sibila/model.py
async def extract_async(self,
                        target: Any,

                        query: Union[Thread,Msg,tuple,str],
                        *,
                        inst: Optional[str] = None,

                        genconf: Optional[GenConf] = None,
                        schemaconf: Optional[JSchemaConf] = None
                        ) -> Any:        
    """Async type-constrained generation: an instance of the given type will be initialized with the model's output.
    The following target types are accepted:

    - prim_type:

        - bool
        - int
        - float
        - str

    - enums:

        - [1, 2, 3] or ["a","b"] - all items of the same prim_type
        - Literal['year', 'name'] - all items of the same prim_type
        - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    - datetime/date/time

    - a list in the form:
        - list[type]

        For example list[int]. The list can be annotated:
            Annotated[list[T], "List desc"]
        And/or the list item type can be annotated:
            list[Annotated[T, "Item desc"]]

    - dataclass with fields of the above supported types (or dataclass).

    - Pydantic BaseModel

    All types can be Annotated[T, "Desc"], for example: 
        count: int
    Can be annotated as:
        count: Annotated[int, "How many units?"]

    Args:
        target: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A value of target arg type instantiated with the model's output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_extract_async(target,
                                       thread,
                                       genconf,
                                       schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

classify_async async #

classify_async(
    labels,
    query,
    *,
    inst=None,
    genconf=None,
    schemaconf=None
)

Returns a classification from one of the given enumeration values The following ways to specify the valid labels are accepted:

  • [1, 2, 3] or ["a","b"] - all items of the same prim_type
  • Literal['year', 'name'] - all items of the same prim_type
  • Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

Parameters:

Name Type Description Default
labels Any

One of the above types.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

One of the given labels, as classified by the model.

Source code in sibila/model.py
async def classify_async(self,
                         labels: Any,

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any:
    """Returns a classification from one of the given enumeration values
    The following ways to specify the valid labels are accepted:

    - [1, 2, 3] or ["a","b"] - all items of the same prim_type
    - Literal['year', 'name'] - all items of the same prim_type
    - Enum, EnumInt, EnumStr, (Enum, int),... - all items of the same prim_type

    Args:
        labels: One of the above types.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        One of the given labels, as classified by the model.
    """

    # verify it's a valid enum "type"
    type_,_ = get_enum_type(labels)
    if type_ is None:
        raise TypeError("Arg labels must be one of Literal, Enum class or a list of str, float or int items")

    return await self.extract_async(labels,
                                    query,
                                    inst=inst,
                                    genconf=genconf,
                                    schemaconf=schemaconf)

json_async async #

json_async(
    query,
    *,
    json_schema=None,
    inst=None,
    genconf=None,
    massage_schema=True,
    schemaconf=None
)

JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema. Raises GenError if unable to get a valid/schema-validated JSON.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

None
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid JSON schema output error. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
dict

A dict from model's JSON response, following genconf.jsonschema, if provided.

Source code in sibila/model.py
async def json_async(self,             
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     json_schema: Union[dict,str,None] = None,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     massage_schema: bool = True,
                     schemaconf: Optional[JSchemaConf] = None,
                     ) -> dict:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, constrained or not by a JSON schema.
    Raises GenError if unable to get a valid/schema-validated JSON.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid JSON schema output error. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A dict from model's JSON response, following genconf.jsonschema, if provided.
    """        

    thread = Thread.ensure(query, inst)

    out = await self.gen_json_async(thread,
                                    json_schema,
                                    genconf,
                                    massage_schema,
                                    schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.dic # type: ignore[return-value]

dataclass_async async #

dataclass_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response. Raises GenError if unable to get a valid response that follows the dataclass definition.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example invalid object initialization. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

An object of class cls (derived from dataclass) initialized from the constrained JSON output.

Source code in sibila/model.py
async def dataclass_async(self, # noqa: E811
                          cls: Any, # a dataclass definition

                          query: Union[Thread,Msg,tuple,str],
                          *,
                          inst: Optional[str] = None,

                          genconf: Optional[GenConf] = None,
                          schemaconf: Optional[JSchemaConf] = None
                          ) -> Any: # a dataclass object
    """Async constrained generation after a dataclass definition, resulting in an object initialized with the model's response.
    Raises GenError if unable to get a valid response that follows the dataclass definition.

    Args:
        cls: A dataclass definition.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example invalid object initialization. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        An object of class cls (derived from dataclass) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_dataclass_async(cls,
                                         thread,
                                         genconf,
                                         schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

pydantic_async async #

pydantic_async(
    cls, query, *, inst=None, genconf=None, schemaconf=None
)

Async constrained generation after a Pydantic BaseModel-derived class definition. Results in an object initialized with the model response. Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
GenError

If an error occurred, for example an invalid BaseModel object. See GenError.

RuntimeError

If unable to generate.

Returns:

Type Description
Any

A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.

Source code in sibila/model.py
async def pydantic_async(self,
                         cls: Any, # a Pydantic BaseModel class

                         query: Union[Thread,Msg,tuple,str],
                         *,
                         inst: Optional[str] = None,

                         genconf: Optional[GenConf] = None,
                         schemaconf: Optional[JSchemaConf] = None
                         ) -> Any: # a Pydantic BaseModel object
    """Async constrained generation after a Pydantic BaseModel-derived class definition.
    Results in an object initialized with the model response.
    Raises GenError if unable to get a valid dict that follows the BaseModel class definition.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        GenError: If an error occurred, for example an invalid BaseModel object. See GenError.
        RuntimeError: If unable to generate.

    Returns:
        A Pydantic object of class cls (derived from BaseModel) initialized from the constrained JSON output.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_pydantic_async(cls,
                                        thread,
                                        genconf,
                                        schemaconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=False) # as valid JSON can still be produced

    return out.value

call_async async #

call_async(
    query,
    *,
    inst=None,
    genconf=None,
    ok_length_is_error=False
)

Text generation from a Thread or plain text, used by the other model generation methods.

Parameters:

Name Type Description Default
query Union[Thread, Msg, tuple, str]

A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.

required
inst Optional[str]

Instruction message for model. Will override Thread's inst, if set. Defaults to None.

None
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
ok_length_is_error bool

Should a result of GenRes.OK_LENGTH be considered an error and raise?

False

Raises:

Type Description
GenError

If an error occurred. This can be a model error, or an invalid JSON output error.

RuntimeError

If unable to generate.

Returns:

Type Description
str

Text generated by model.

Source code in sibila/model.py
async def call_async(self,
                     query: Union[Thread,Msg,tuple,str],
                     *,
                     inst: Optional[str] = None,

                     genconf: Optional[GenConf] = None,
                     ok_length_is_error: bool = False
                     ) -> str:
    """Text generation from a Thread or plain text, used by the other model generation methods.

    Args:
        query: A Thread or a single IN message given as Msg, list, tuple or str. List and tuple should contain the same args as for creating Msg.
        inst: Instruction message for model. Will override Thread's inst, if set. Defaults to None.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        ok_length_is_error: Should a result of GenRes.OK_LENGTH be considered an error and raise?

    Raises:
        GenError: If an error occurred. This can be a model error, or an invalid JSON output error.
        RuntimeError: If unable to generate.

    Returns:
        Text generated by model.
    """

    thread = Thread.ensure(query, inst)

    out = await self.gen_async(thread=thread, 
                               genconf=genconf)

    GenError.raise_if_error(out,
                            ok_length_is_error=ok_length_is_error)

    return out.text

gen #

gen(thread, genconf=None)

Text generation from a Thread, used by the other model generation methods. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc.

GenOut

The output text is in GenOut.text.

Source code in sibila/openai.py
def gen(self, 
        thread: Thread,
        genconf: Optional[GenConf] = None,
        ) -> GenOut:
    """Text generation from a Thread, used by the other model generation methods.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc.
        The output text is in GenOut.text.
    """

    genconf2: GenConf
    kwargs, genconf2 = self._gen_pre(thread, genconf)

    self._ensure_client(False)

    try:
        # https://platform.openai.com/docs/api-reference/chat/create
        response = self._client.chat.completions.create(**kwargs) # type: ignore[attr-defined]

    except Exception as e:
        raise RuntimeError(f"Cannot generate. Internal error: {e}")


    return self._gen_post(response,
                          kwargs,
                          genconf2)

gen_json #

gen_json(
    thread,
    json_schema,
    genconf=None,
    massage_schema=True,
    schemaconf=None,
)

JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema. Doesn't raise an exception if an error occurs, always returns GenOut.

Parameters:

Name Type Description Default
thread Thread

The Thread to use as model input.

required
json_schema Union[dict, str, None]

A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
massage_schema bool

Simplify schema. Defaults to True.

True
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.

Source code in sibila/model.py
def gen_json(self,
             thread: Thread,
             json_schema: Union[dict,str,None],
             genconf: Optional[GenConf] = None,

             massage_schema: bool = True,
             schemaconf: Optional[JSchemaConf] = None,
             ) -> GenOut:
    """JSON/JSON-schema constrained generation, returning a Python dict of values, conditioned or not by a JSON schema.
    Doesn't raise an exception if an error occurs, always returns GenOut.

    Args:
        thread: The Thread to use as model input.
        json_schema: A JSON schema describing the dict fields that will be output. None means no schema (free JSON output).
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        massage_schema: Simplify schema. Defaults to True.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The output dict is in GenOut.dic.
    """

    args = self._gen_json_pre(thread,
                              json_schema,
                              genconf,
                              massage_schema,
                              schemaconf)
    return self.gen(*args)

gen_dataclass #

gen_dataclass(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a dataclass definition. An initialized dataclass object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A dataclass definition.

required
thread Thread

The Thread object to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.

Source code in sibila/model.py
def gen_dataclass(self,
                  cls: Any, # a dataclass
                  thread: Thread,
                  genconf: Optional[GenConf] = None,
                  schemaconf: Optional[JSchemaConf] = None
                  ) -> GenOut:
    """Constrained generation after a dataclass definition.
    An initialized dataclass object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A dataclass definition.
        thread: The Thread object to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.

    Returns:
        A GenOut object with result, generated text, etc. The initialized dataclass object is in GenOut.value.
    """

    schema = self._gen_dataclass_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_dataclass_post(out,
                                    cls,
                                    schemaconf)

gen_pydantic #

gen_pydantic(cls, thread, genconf=None, schemaconf=None)

Constrained generation after a Pydantic BaseModel-derived class definition. An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict. Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

Parameters:

Name Type Description Default
cls Any

A class derived from a Pydantic BaseModel class.

required
thread Thread

The Thread to use as model input.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None, which uses model's default.

None
schemaconf Optional[JSchemaConf]

JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

None

Raises:

Type Description
RuntimeError

If unable to generate.

TypeError

When cls is not a Pydantic BaseClass.

Returns:

Type Description
GenOut

A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.

Source code in sibila/model.py
def gen_pydantic(self,
                 cls: Any, # a Pydantic BaseModel class
                 thread: Thread,
                 genconf: Optional[GenConf] = None,
                 schemaconf: Optional[JSchemaConf] = None
                 ) -> GenOut:
    """Constrained generation after a Pydantic BaseModel-derived class definition.
    An initialized Pydantic BaseModel object is returned in the "value" field of the returned dict.
    Doesn't raise an exception if an error occurs, always returns GenOut containing the created object.

    Args:
        cls: A class derived from a Pydantic BaseModel class.
        thread: The Thread to use as model input.
        genconf: Model generation configuration. Defaults to None, which uses model's default.
        schemaconf: JSchemaConf object that controls schema simplification. Defaults to None, which uses model's default.

    Raises:
        RuntimeError: If unable to generate.
        TypeError: When cls is not a Pydantic BaseClass.

    Returns:
        A GenOut object with result, generated text, etc. The initialized Pydantic BaseModel-derived object is in GenOut.value.
    """

    schema = self._gen_pydantic_pre(cls)

    out = self.gen_json(thread,
                        schema,
                        genconf,
                        massage_schema=True,
                        schemaconf=schemaconf)

    return self._gen_pydantic_post(out,
                                   cls,
                                   schemaconf)

token_len #

token_len(thread_or_text, genconf=None)

Calculate or estimate the token length for a Thread or a plain text string. In some cases where it's not possible to calculate the exact token count, this function should give a conservative (upper bound) estimate. It's up to the implementation whether to account for side information like JSON Schema, but it must reflect the model's context token accounting. Thread or text must be the final text which will passed to model.

If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

More info on calculating function_call (and tools?) tokens:

https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

Parameters:

Name Type Description Default
thread_or_text Union[Thread, str]

For token length calculation.

required
genconf Optional[GenConf]

Model generation configuration. Defaults to None.

None

Returns:

Type Description
int

Estimated number of tokens used.

Source code in sibila/openai.py
def token_len(self,
              thread_or_text: Union[Thread,str],
              genconf: Optional[GenConf] = None) -> int:
    """Calculate or estimate the token length for a Thread or a plain text string.
    In some cases where it's not possible to calculate the exact token count, 
    this function should give a conservative (upper bound) estimate.
    It's up to the implementation whether to account for side information like JSON Schema,
    but it must reflect the model's context token accounting.
    Thread or text must be the final text which will passed to model.

    If a json_schema is provided in genconf, we use its string's token_len as upper bound for the extra prompt tokens.

    From https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb

    More info on calculating function_call (and tools?) tokens:

    https://community.openai.com/t/how-to-calculate-the-tokens-when-using-function-call/266573/24

    https://gist.github.com/CGamesPlay/dd4f108f27e2eec145eedf5c717318f5

    Args:
        thread_or_text: For token length calculation.
        genconf: Model generation configuration. Defaults to None.

    Returns:
        Estimated number of tokens used.
    """

    if isinstance(thread_or_text, Thread):
        thread = thread_or_text            
    else:
        thread = Thread.make_IN(thread_or_text)

    num_tokens = 0

    if self.tokenizer is None: # no tokenizer was found, so we'll have to do a conservative estimate

        OVERHEAD_PER_MSG = 3
        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            msg_tokens = len(str(message["content"])) * self._token_estimation_factor + OVERHEAD_PER_MSG
            # str(message["content"]): hacky way to deal with dict "content" key
            num_tokens += int(msg_tokens)

        if genconf is not None and genconf.json_schema is not None:
            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = len(js_str) * self._token_estimation_factor
            num_tokens += int(tools_num_tokens)
            # print("tools_num_tokens", tools_num_tokens)

    else: # do an "informed" token estimation from what is known of the OpenAI model's tokenization

        for msg in thread.get_iter(True): # True for system message
            message = msg.as_chatml()
            # print(message)
            num_tokens += self._overhead_per_msg
            for key, value in message.items():
                num_tokens += len(self.tokenizer.encode(str(value))) # str(value): hacky way to deal with dict "content" key

        # add extras + every reply is primed with <|start|>assistant<|message|>
        num_tokens += 32

        # print("text token_len", num_tokens)

        if genconf is not None and genconf.json_schema is not None:
            TOOLS_TOKEN_LEN_FACTOR = 1.2

            if isinstance(genconf.json_schema, str):
                js_str = genconf.json_schema
            else:
                js_str = json.dumps(genconf.json_schema)

            tools_num_tokens = self.tokenizer.token_len(js_str)

            # this is an upper bound, as empirically tested with the api.
            tools_num_tokens = int(tools_num_tokens * TOOLS_TOKEN_LEN_FACTOR)
            # print("tools token_len", tools_num_tokens)

            num_tokens += tools_num_tokens


    return num_tokens

tokenizer instance-attribute #

tokenizer = OpenAITokenizer(_model_name)

ctx_len instance-attribute #

ctx_len = ctx_len or default_ctx_len

maybe_image_input instance-attribute #

maybe_image_input = False

known_models classmethod #

known_models(api_key=None)

List of model names that can be used. Some of the models are not chat models and cannot be used, for example embedding models.

Parameters:

Name Type Description Default
api_key Optional[str]

If the model provider requires an API key, pass it here or set it in the respective env variable.

None

Returns:

Type Description
Union[list[str], None]

Returns a list of known models or None if unable to fetch it.

Source code in sibila/schema_format_openai.py
@classmethod
def known_models(cls,
                 api_key: Optional[str] = None) -> Union[list[str], None]:
    """List of model names that can be used. Some of the models are not chat models and cannot be used,
    for example embedding models.

    Args:
        api_key: If the model provider requires an API key, pass it here or set it in the respective env variable.

    Returns:
        Returns a list of known models or None if unable to fetch it.
    """
    return None

desc #

desc()

Model description.

Source code in sibila/openai.py
def desc(self) -> str:
    """Model description."""
    return f"{type(self).__name__}: '{self._model_name}'"