intelligence_layer.core

Module contents

class intelligence_layer.core.TextChunk

Segment of a larger text.

This type infers that the string is smaller than the context size of the model where it is used.

LLMs can’t process documents larger than their context size. To handle this, documents have to be split up into smaller segments that fit within their context size. These smaller segments are referred to as chunks.

alias of str

class intelligence_layer.core.AlephAlphaChatModel(name: str, client: AlephAlphaClientProtocol | None = None)

Bases: ChatModel, ControlModel

Abstract base class for any model that supports chat and runs via the Aleph Alpha API.

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

echo_chat(messages: list[Message], response_prefix: str | None, expected_completion: str, tracer: Tracer) → Sequence[tuple[Any, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• messages – The messages to be used as prompt

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation.

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

generate_chat(messages: Sequence[Message], response_prefix: str | None, tracer: Tracer) → str

Generate a raw completion to messages for any AlephAlphaChatModel.

Parameters:

• messages – A number of messages to use as prompt for the model

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_chat_prompt(messages: Sequence[Message], response_prefix: str | None = None) → RichPrompt

Method to create a chat-RichPrompt object to use with any AlephAlphaModel.

Parameters:

• messages – A number of messages to use as prompt for the model

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation. Defaults to None.

Returns:

A RichPrompt object to be consumed by the Aleph Alpha client

abstract to_finetuning_sample(messages: Sequence[Message]) → Sequence[FinetuningMessage]

Abstract function allowing a user to define what the model’s finetuning samples should look like.

Parameters:

messages – The messages making up the finetuning sample

Returns:

A finetuning sample containing the input messages

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to use a chat model like an instruct model`.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarized

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – Instruction controls are not used but needed for the interface.

• input_controls – Input controls are not used but needed for the interface

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.AlephAlphaModel(name: str, client: AlephAlphaClientProtocol | None = None)

Bases: LanguageModel

Model-class for any model that uses the Aleph Alpha client.

Any class of Aleph Alpha model is implemented on top of this base class. Exposes methods that are available to all models, such as complete and tokenize. It is the central place for all things that are physically interconnected with a model, such as its tokenizer or prompt format used during training.

Parameters:

• name – The name of a valid model that can access an API using an implementation of the AlephAlphaClientProtocol.

• client – Aleph Alpha client instance for running model related API calls. Defaults to LimitedConcurrencyClient

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

class intelligence_layer.core.ChatModel(name: str)

Bases: LanguageModel

Abstract base class to implement any model that supports chat.

abstract echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Any, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

abstract echo_chat(messages: list[Message], response_prefix: str | None, expected_completion: str, tracer: Tracer) → Sequence[tuple[Any, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• messages – The messages to be used as prompt

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation.

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

abstract generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

abstract generate_chat(messages: list[Message], response_prefix: str | None, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• messages – The messages to be used as prompt

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation.

• tracer – Valid instance of a tracer

Returns:

An LLM completion

class intelligence_layer.core.Chunk(model: AlephAlphaModel, max_tokens_per_chunk: int = 512)

Bases: Task[ChunkInput, ChunkOutput]

Splits a longer text into smaller text chunks.

Provide a text of any length and chunk it into smaller pieces using a tokenizer that is available within the Aleph Alpha client.

Parameters:

• model – A valid Aleph Alpha model.

• max_tokens_per_chunk – The maximum number of tokens to fit into one chunk.

do_run(input: ChunkInput, task_span: TaskSpan) → ChunkOutput

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.ChunkInput(*, text: str)

Bases: BaseModel

The input for a Chunk-task.

text

A text of arbitrary length.

Type:

str

class intelligence_layer.core.ChunkOutput(*, chunks: Sequence[TextChunk])

Bases: BaseModel

The output of a ChunkTask.

chunks

A list of smaller sections of the input text.

Type:

collections.abc.Sequence[intelligence_layer.core.chunk.TextChunk]

class intelligence_layer.core.ChunkWithIndices(model: AlephAlphaModel, max_tokens_per_chunk: int = 512)

Bases: Task[ChunkInput, ChunkWithIndicesOutput]

Splits a longer text into smaller text chunks and returns the chunks’ start indices.

Provide a text of any length and chunk it into smaller pieces using a tokenizer that is available within the Aleph Alpha client. For each chunk, the respective start index relative to the document is also returned.

Parameters:

• model – A valid Aleph Alpha model.

• max_tokens_per_chunk – The maximum number of tokens to fit into one chunk.

do_run(input: ChunkInput, task_span: TaskSpan) → ChunkWithIndicesOutput

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.ChunkWithIndicesOutput(*, chunks_with_indices: Sequence[ChunkWithStartEndIndices])

Bases: BaseModel

The output of a ChunkWithIndices-task.

chunks_with_indices

A list of smaller sections of the input text with the respective start_index.

Type:

collections.abc.Sequence[intelligence_layer.core.chunk.ChunkWithStartEndIndices]

class intelligence_layer.core.ChunkWithStartEndIndices(*, chunk: TextChunk, start_index: int, end_index: int)

Bases: BaseModel

A TextChunk and its start_index and end_index within the given text.

chunk

The actual text.

Type:

intelligence_layer.core.chunk.TextChunk

start_index

The character start index of the chunk within the given text.

Type:

int

end_index

The character end index of the chunk within the given text.

Type:

int

class intelligence_layer.core.CompleteInput(*, prompt: Prompt, maximum_tokens: int | None = None, temperature: float = 0.0, top_k: int = 0, top_p: float = 0.0, presence_penalty: float = 0.0, frequency_penalty: float = 0.0, repetition_penalties_include_prompt: bool = False, use_multiplicative_presence_penalty: bool = False, penalty_bias: str | None = None, penalty_exceptions: List[str] | None = None, penalty_exceptions_include_stop_sequences: bool | None = None, best_of: int | None = None, n: int = 1, logit_bias: Dict[int, float] | None = None, log_probs: int | None = None, stop_sequences: List[str] | None = None, tokens: bool = False, disable_optimizations: bool = False, minimum_tokens: int = 0, echo: bool = False, use_multiplicative_frequency_penalty: bool = False, sequence_penalty: float = 0.0, sequence_penalty_min_length: int = 2, use_multiplicative_sequence_penalty: bool = False, completion_bias_inclusion: Sequence[str] | None = None, completion_bias_inclusion_first_token_only: bool = False, completion_bias_exclusion: Sequence[str] | None = None, completion_bias_exclusion_first_token_only: bool = False, contextual_control_threshold: float | None = None, control_log_additive: bool | None = True, repetition_penalties_include_completion: bool = True, raw_completion: bool = False)

Bases: BaseModel, CompletionRequest

The input for a Complete task.

class intelligence_layer.core.CompleteOutput(*, model_version: str, completions: Sequence[CompletionResult], num_tokens_prompt_total: int, num_tokens_generated: int, optimized_prompt: Prompt | None = None)

Bases: BaseModel, CompletionResponse

The output of a Complete task.

class intelligence_layer.core.CompositeTracer(tracers: Sequence[TracerVar])

Bases: Tracer, Generic[TracerVar]

A Tracer that allows for recording to multiple tracers simultaneously.

Each log-entry and span will be forwarded to all subtracers.

Parameters:

tracers – tracers that will be forwarded all subsequent log and span calls.

Example

>>> from intelligence_layer.core import InMemoryTracer, FileTracer, CompositeTracer, TextChunk
>>> from intelligence_layer.examples import PromptBasedClassify, ClassifyInput

>>> tracer_1 = InMemoryTracer()
>>> tracer_2 = InMemoryTracer()
>>> tracer = CompositeTracer([tracer_1, tracer_2])
>>> task = PromptBasedClassify()
>>> response = task.run(ClassifyInput(chunk=TextChunk("Cool"), labels=frozenset({"label", "other label"})), tracer)

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

span(name: str, timestamp: datetime | None = None) → CompositeSpan[Span]

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → CompositeTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.Context(*, trace_id: UUID, span_id: UUID)

Bases: BaseModel

class intelligence_layer.core.ControlModel(name: str, client: AlephAlphaClientProtocol | None = None)

Bases: AlephAlphaModel, ABC

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to create an instruct-RichPrompt object to use with any ControlModel.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarize

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – TextControls for the instruction part of the prompt. Only for text prompts.

• input_controls – TextControls for the input part of the prompt. Only for text prompts.

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.DetectLanguage(threshold: float = 0.5)

Bases: Task[DetectLanguageInput, DetectLanguageOutput]

Task that detects the language of a text.

Analyzes the likelihood that a given text is written in one of the possible_languages. Returns the best match or None.

Parameters:

threshold – Minimum probability value for a language to be considered the best_fit.

Example

>>> from intelligence_layer.core import (
...     DetectLanguage,
...     DetectLanguageInput,
...     InMemoryTracer,
...     Language,
... )

>>> task = DetectLanguage()
>>> input = DetectLanguageInput(
...     text="This is an English text.",
...     possible_languages=[Language(l) for l in ("en", "fr")],
... )
>>> output = task.run(input, InMemoryTracer())

do_run(input: DetectLanguageInput, task_span: TaskSpan) → DetectLanguageOutput

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.DetectLanguageInput(*, text: str, possible_languages: Sequence[Language])

Bases: BaseModel

The input for a DetectLanguage task.

text

The text to identify the language for.

Type:

str

possible_languages

All languages that should be considered during detection. Languages should be provided with their ISO 639-1 codes.

Type:

collections.abc.Sequence[intelligence_layer.core.detect_language.Language]

class intelligence_layer.core.DetectLanguageOutput(*, best_fit: Language | None)

Bases: BaseModel

The output of a DetectLanguage task.

best_fit

The prediction for the best matching language. Will be None if no language has a probability above the threshold.

Type:

intelligence_layer.core.detect_language.Language | None

class intelligence_layer.core.Echo(model: AlephAlphaModel)

Bases: Task[EchoInput, EchoOutput]

Task that returns probabilities of a completion given a prompt.

Analyzes the likelihood of generating tokens in the expected completion based on a given prompt and model. Does not generate any tokens.

Parameters:

model – A model to use in the task.

Example

>>> from aleph_alpha_client import Prompt
>>> from intelligence_layer.core import Echo, EchoInput, InMemoryTracer, LuminousControlModel

>>> model = LuminousControlModel(name="luminous-base-control")
>>> task = Echo(model)
>>> input = EchoInput(
...     prompt=Prompt.from_text("This is a "),
...     expected_completion="happy text",
... )
>>> tracer = InMemoryTracer()
>>> output = task.run(input, tracer)

do_run(input: EchoInput, task_span: TaskSpan) → EchoOutput

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.EchoInput(*, prompt: Prompt, expected_completion: str)

Bases: BaseModel

The input for an Echo task.

prompt

The input text that serves as the starting point for the LLM.

Type:

aleph_alpha_client.prompt.Prompt

expected_completion

The desired completion based on the prompt. The likelihood of the tokens in this will be examined.

Type:

str

class intelligence_layer.core.EchoOutput(*, tokens_with_log_probs: Sequence[TokenWithLogProb])

Bases: BaseModel

The output of an Echo task.

tokens_with_log_probs

Every token of the expected_completion of the EchoInput accompanied by its probability of having been generated in a completion scenario.

Type:

collections.abc.Sequence[intelligence_layer.core.echo.TokenWithLogProb]

class intelligence_layer.core.EndSpan(*, uuid: UUID, end: datetime, status_code: SpanStatus = SpanStatus.OK)

Bases: BaseModel

Represents the payload/entry of a log-line that indicates that a Span ended.

uuid

The uuid of the corresponding StartSpan.

Type:

uuid.UUID

end

the timestamp when this Span completed.

Type:

datetime.datetime

class intelligence_layer.core.EndTask(*, uuid: UUID, end: datetime, output: Annotated[PydanticSerializable, SerializeAsAny()], status_code: SpanStatus = SpanStatus.OK)

Bases: BaseModel

Represents the payload/entry of a log-line that indicates that a TaskSpan ended (i.e. the context-manager exited).

uuid

The uuid of the corresponding StartTask.

Type:

uuid.UUID

end

the timestamp when this Task completed (i.e. run returned).

Type:

datetime.datetime

output

the Output (i.e. return value of run) the Task returned.

Type:

PydanticSerializable

class intelligence_layer.core.ErrorValue(*, error_type: str, message: str, stack_trace: str)

Bases: BaseModel

class intelligence_layer.core.Event(*, name: str, message: str, body: Annotated[PydanticSerializable, SerializeAsAny()], timestamp: datetime = None)

Bases: BaseModel

class intelligence_layer.core.ExplainInput(*, prompt: Prompt, target: str, contextual_control_threshold: float | None = None, control_factor: float | None = None, control_token_overlap: ControlTokenOverlap | None = None, control_log_additive: bool | None = None, prompt_granularity: PromptGranularity | str | CustomGranularity | None = None, target_granularity: TargetGranularity | None = None, postprocessing: ExplanationPostprocessing | None = None, normalize: bool | None = None)

Bases: BaseModel, ExplanationRequest

The input for a Explain task.

class intelligence_layer.core.ExplainOutput(*, model_version: str, explanations: List[Explanation])

Bases: BaseModel, ExplanationResponse

The output of a Explain task.

class intelligence_layer.core.ExportedSpan(*, context: Context, name: str | None, parent_id: UUID | None, start_time: datetime, end_time: datetime, attributes: SpanAttributes | TaskSpanAttributes, events: Sequence[Event], status: SpanStatus)

Bases: BaseModel

intelligence_layer.core.ExportedSpanList

alias of RootModel[Sequence[ExportedSpan]]

class intelligence_layer.core.FileSpan(log_file_path: Path, context: Context | None = None)

Bases: PersistentSpan, FileTracer

A Span created by FileTracer.span.

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

span(name: str, timestamp: datetime | None = None) → FileSpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → FileTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

traces(trace_id: str | None = None) → InMemoryTracer

Returns all traces of the given tracer.

Returns:

An InMemoryTracer that contains all traces of the tracer.

class intelligence_layer.core.FileTaskSpan(log_file_path: Path, context: Context | None = None)

Bases: PersistentTaskSpan, FileSpan

A TaskSpan created by FileTracer.task_span.

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

record_output(output: PydanticSerializable) → None

Record Task output.

Since a Context Manager can’t capture output in the __exit__ method, output should be captured once it is generated.

Parameters:

output – The output of the task that is being logged.

span(name: str, timestamp: datetime | None = None) → FileSpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → FileTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

traces(trace_id: str | None = None) → InMemoryTracer

Returns all traces of the given tracer.

Returns:

An InMemoryTracer that contains all traces of the tracer.

class intelligence_layer.core.FileTracer(log_file_path: Path | str)

Bases: PersistentTracer

A Tracer that logs to a file.

Each log-entry is represented by a JSON object. The information logged allows to reconstruct the hierarchical nature of the logs, i.e. all entries have a _pointer_ to its parent element in form of a parent attribute containing the uuid of the parent.

Parameters:

log_file_path – Denotes the file to log to.

uuid

a uuid for the tracer. If multiple FileTracer instances log to the same file the child-elements for a tracer can be identified by referring to this id as parent.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

span(name: str, timestamp: datetime | None = None) → FileSpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → FileTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

traces(trace_id: str | None = None) → InMemoryTracer

Returns all traces of the given tracer.

Returns:

An InMemoryTracer that contains all traces of the tracer.

class intelligence_layer.core.FinetuningMessage(*, has_loss: bool, content: str, type: str = 'text')

Bases: BaseModel

Represent a prompt message in a finetuning sample as required to finetune an llm using [scaling](https://github.com/Aleph-Alpha/scaling).

Parameters:

• has_loss – Flag indicated whether loss should be applied to the message during training.

• content – The text in the message

• type – Should always be “text”

class intelligence_layer.core.InMemorySpan(name: str, context: Context | None = None, start_timestamp: datetime | None = None)

Bases: InMemoryTracer, Span

A span that keeps all important information in memory.

context

Ids that uniquely describe the span.

Type:

intelligence_layer.core.tracer.tracer.Context

parent_id

Id of the parent span. None if the span is a root span.

name

The name of the span.

start_timestamp

The start of the timestamp.

end_timestamp

The end of the timestamp. None until the span is closed.

status_code

The status of the context.

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

span(name: str, timestamp: datetime | None = None) → InMemorySpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → InMemoryTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.InMemoryTaskSpan(name: str, input: Annotated[PydanticSerializable, SerializeAsAny()], context: Context | None = None, start_timestamp: datetime | None = None)

Bases: InMemorySpan, TaskSpan

A span of a task that keeps all important information in memory.

context

Ids that uniquely describe the span.

Type:

intelligence_layer.core.tracer.tracer.Context

parent_id

Id of the parent span. None if the span is a root span.

name

The name of the span.

start_timestamp

The start of the timestamp.

end_timestamp

The end of the timestamp. None until the span is closed.

status_code

The status of the context.

input

The input of the task.

output

The output of the task.

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

record_output(output: PydanticSerializable) → None

Record Task output.

Since a Context Manager can’t capture output in the __exit__ method, output should be captured once it is generated.

Parameters:

output – The output of the task that is being logged.

span(name: str, timestamp: datetime | None = None) → InMemorySpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → InMemoryTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.InMemoryTracer

Bases: Tracer

Collects log entries in a nested structure, and keeps them in memory.

entries

A sequential list of log entries and/or nested InMemoryTracers with their own log entries.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

span(name: str, timestamp: datetime | None = None) → InMemorySpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → InMemoryTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.Instruct(model: ControlModel)

Bases: Task[InstructInput, CompleteOutput]

do_run(input: InstructInput, task_span: TaskSpan) → CompleteOutput

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.InstructInput(*, instruction: str, input: str | None = None, response_prefix: str | None = None, maximum_tokens: int = 128)

Bases: BaseModel

class intelligence_layer.core.JsonSerializer(root: RootModelRootType = PydanticUndefined)

Bases: RootModel[PydanticSerializable]

classmethod model_construct(root: RootModelRootType, _fields_set: set[str] | None = None) → Self

Create a new model using the provided root object and update fields set.

Parameters:

• root – The root object of the model.

• _fields_set – The set of fields to be updated.

Returns:

The new model.

Raises:

NotImplemented – If the model is not a subclass of RootModel.

class intelligence_layer.core.Language(iso_639_1: str)

Bases: object

A language identified by its ISO 639-1 code.

class intelligence_layer.core.LanguageModel(name: str)

Bases: ABC

Abstract base class to implement any LLM.

abstract echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Any, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

abstract generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

class intelligence_layer.core.Llama2InstructModel(name: str = 'llama-2-13b-chat', client: AlephAlphaClientProtocol | None = None)

Bases: ControlModel

A llama-2-*-chat model, prompt-optimized for single-turn instructions.

If possible, we recommend using Llama3InstructModel instead.

Parameters:

• name – The name of a valid llama-2 model. Defaults to llama-2-13b-chat

• client – Aleph Alpha client instance for running model related API calls. Defaults to LimitedConcurrencyClient

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to create an instruct-RichPrompt object to use with any ControlModel.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarize

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – TextControls for the instruction part of the prompt. Only for text prompts.

• input_controls – TextControls for the input part of the prompt. Only for text prompts.

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.Llama3ChatModel(name: str = 'llama-3.1-8b-instruct', client: AlephAlphaClientProtocol | None = None)

Bases: AlephAlphaChatModel

Chat model to be used for llama-3-* and llama-3.1-* models.

Parameters:

• name – The name of a valid llama-3 model. Defaults to llama-3-8b-instruct

• client – Aleph Alpha client instance for running model related API calls. Defaults to LimitedConcurrencyClient

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

echo_chat(messages: list[Message], response_prefix: str | None, expected_completion: str, tracer: Tracer) → Sequence[tuple[Any, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• messages – The messages to be used as prompt

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation.

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

generate_chat(messages: Sequence[Message], response_prefix: str | None, tracer: Tracer) → str

Generate a raw completion to messages for any AlephAlphaChatModel.

Parameters:

• messages – A number of messages to use as prompt for the model

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_chat_prompt(messages: Sequence[Message], response_prefix: str | None = None) → RichPrompt

Method to create a chat-RichPrompt object to use with any AlephAlphaModel.

Parameters:

• messages – A number of messages to use as prompt for the model

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation. Defaults to None.

Returns:

A RichPrompt object to be consumed by the Aleph Alpha client

to_finetuning_sample(messages: Sequence[Message]) → Sequence[FinetuningMessage]

Abstract function allowing a user to define what the model’s finetuning samples should look like.

Parameters:

messages – The messages making up the finetuning sample

Returns:

A finetuning sample containing the input messages

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to use a chat model like an instruct model`.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarized

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – Instruction controls are not used but needed for the interface.

• input_controls – Input controls are not used but needed for the interface

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.Llama3InstructModel(name: str = 'llama-3.1-8b-instruct', client: AlephAlphaClientProtocol | None = None)

Bases: ControlModel

A llama-3-*-instruct model.

Parameters:

• name – The name of a valid llama-3 model. Defaults to llama-3.1-8b-instruct

• client – Aleph Alpha client instance for running model related API calls. Defaults to LimitedConcurrencyClient

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to create an instruct-RichPrompt object to use with any ControlModel.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarize

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – TextControls for the instruction part of the prompt. Only for text prompts.

• input_controls – TextControls for the input part of the prompt. Only for text prompts.

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.LogEntry(*, message: str, value: Annotated[PydanticSerializable, SerializeAsAny()], timestamp: datetime = None, trace_id: UUID)

Bases: BaseModel

An individual log entry, currently used to represent individual logs by the InMemoryTracer.

message

A description of the value you are logging, such as the step in the task this is related to.

Type:

str

value

The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

Type:

PydanticSerializable

timestamp

The time that the log was emitted.

Type:

datetime.datetime

id

The ID of the trace to which this log entry belongs.

class intelligence_layer.core.LogLine(*, trace_id: UUID, entry_type: str, entry: Annotated[PydanticSerializable, SerializeAsAny()])

Bases: BaseModel

Represents a complete log-line.

entry_type

The type of the entry. This is the class-name of one of the classes representing a log-entry (e.g. “StartTask”).

Type:

str

entry

The actual entry.

Type:

PydanticSerializable

class intelligence_layer.core.LuminousControlModel(name: str = 'luminous-base-control', client: AlephAlphaClientProtocol | None = None)

Bases: ControlModel

An Aleph Alpha control model of the second generation.

Parameters:

• name – The name of a valid model second generation control model. Defaults to luminous-base-control

• client – Aleph Alpha client instance for running model related API calls. Defaults to LimitedConcurrencyClient

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to create an instruct-RichPrompt object to use with any ControlModel.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarize

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – TextControls for the instruction part of the prompt. Only for text prompts.

• input_controls – TextControls for the input part of the prompt. Only for text prompts.

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.Message(*, role: Literal['system', 'user', 'assistant'], content: str)

Bases: BaseModel

class intelligence_layer.core.NoOpTracer(context: Context | None = None)

Bases: TaskSpan

A no-op tracer.

Useful for cases, like testing, where a tracer is needed for a task, but you don’t have a need to collect or inspect the actual logs.

All calls to log won’t actually do anything.

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

record_output(output: PydanticSerializable) → None

Record Task output.

Since a Context Manager can’t capture output in the __exit__ method, output should be captured once it is generated.

Parameters:

output – The output of the task that is being logged.

span(name: str, timestamp: datetime | None = None) → NoOpTracer

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → NoOpTracer

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.OpenTelemetryTracer(tracer: Tracer)

Bases: Tracer

A Tracer that uses open telemetry.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

span(name: str, timestamp: datetime | None = None) → OpenTelemetrySpan

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → OpenTelemetryTaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.PersistentSpan(context: Context | None = None)

Bases: Span, PersistentTracer, ABC

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

abstract span(name: str, timestamp: datetime | None = None) → Span

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

abstract task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → TaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

abstract traces() → InMemoryTracer

Returns all traces of the given tracer.

Returns:

An InMemoryTracer that contains all traces of the tracer.

class intelligence_layer.core.PersistentTaskSpan(context: Context | None = None)

Bases: TaskSpan, PersistentSpan, ABC

end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

record_output(output: PydanticSerializable) → None

Record Task output.

Since a Context Manager can’t capture output in the __exit__ method, output should be captured once it is generated.

Parameters:

output – The output of the task that is being logged.

abstract span(name: str, timestamp: datetime | None = None) → Span

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

abstract task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → TaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

abstract traces() → InMemoryTracer

Returns all traces of the given tracer.

Returns:

An InMemoryTracer that contains all traces of the tracer.

class intelligence_layer.core.PersistentTracer

Bases: Tracer, ABC

export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

abstract span(name: str, timestamp: datetime | None = None) → Span

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

abstract task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → TaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

abstract traces() → InMemoryTracer

Returns all traces of the given tracer.

Returns:

An InMemoryTracer that contains all traces of the tracer.

class intelligence_layer.core.Pharia1ChatModel(name: str = 'pharia-1-llm-7b-control', client: AlephAlphaClientProtocol | None = None)

Bases: AlephAlphaChatModel

Chat model to be used for any “pharia-1-llm-* model.

Parameters:

• name – The name of a valid Pharia-1 model. Defaults to pharia-1-llm-7b-control

• client – Aleph Alpha client instance for running model related API calls. Defaults to LimitedConcurrencyClient

echo(prompt: str, expected_completion: str, tracer: Tracer) → Sequence[tuple[Token, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• prompt – The prompt to echo

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

echo_chat(messages: list[Message], response_prefix: str | None, expected_completion: str, tracer: Tracer) → Sequence[tuple[Any, float | None]]

Echos the log probs for each token of an expected completion given a prompt.

Parameters:

• messages – The messages to be used as prompt

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation.

• expected_completion – The expected completion to get log probs for

• tracer – Valid instance of a tracer

Returns:

A list of tuples with token identifier and log probability

generate(prompt: str, tracer: Tracer) → str

A completion function that takes a prompt and generates a completion.

Parameters:

• prompt – The prompt to generate a completion for

• tracer – Valid instance of a tracer

Returns:

An LLM completion

generate_chat(messages: Sequence[Message], response_prefix: str | None, tracer: Tracer) → str

Generate a raw completion to messages for any AlephAlphaChatModel.

Parameters:

• messages – A number of messages to use as prompt for the model

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• tracer – Valid instance of a tracer

Returns:

An LLM completion

to_chat_prompt(messages: Sequence[Message], response_prefix: str | None = None) → RichPrompt

Method to create a chat-RichPrompt object to use with any AlephAlphaModel.

Parameters:

• messages – A number of messages to use as prompt for the model

• response_prefix – Append the given string to the beginning of the final agent message to steer the generation. Defaults to None.

Returns:

A RichPrompt object to be consumed by the Aleph Alpha client

to_finetuning_sample(messages: Sequence[Message]) → Sequence[FinetuningMessage]

Abstract function allowing a user to define what the model’s finetuning samples should look like.

Parameters:

messages – The messages making up the finetuning sample

Returns:

A finetuning sample containing the input messages

to_instruct_prompt(instruction: str, input: str | None = None, response_prefix: str | None = None, instruction_controls: Sequence[TextControl] | None = None, input_controls: Sequence[TextControl] | None = None) → RichPrompt

Method to use a chat model like an instruct model`.

Parameters:

• instruction – The task the model should fulfill, for example summarization

• input – Any context necessary to solve the task, such as the text to be summarized

• response_prefix – Optional argument to append a string to the beginning of the final agent message to steer the generation

• instruction_controls – Instruction controls are not used but needed for the interface.

• input_controls – Input controls are not used but needed for the interface

Returns:

The rendered prompt with all variables filled in.

class intelligence_layer.core.PlainEntry(*, message: str, value: Annotated[PydanticSerializable, SerializeAsAny()], timestamp: datetime, parent: UUID, trace_id: UUID)

Bases: BaseModel

Represents a plain log-entry created through Tracer.log.

message

the message-parameter of Tracer.log

Type:

str

value

the value-parameter of Tracer.log

Type:

PydanticSerializable

timestamp

the timestamp when Tracer.log was called.

Type:

datetime.datetime

parent

The unique id of the parent element of the log. This could refer to either a surrounding TaskSpan, Span or the top-level Tracer.

Type:

uuid.UUID

trace_id

The ID of the trace this entry belongs to.

Type:

uuid.UUID

class intelligence_layer.core.PromptItemCursor(item: int)

Bases: object

Defines a position with a non-Text prompt item.

Parameters:

item – the index of the prompt item within the Prompt

class intelligence_layer.core.PromptRange(start: TextCursor | PromptItemCursor, end: TextCursor | PromptItemCursor)

Bases: object

Defines a range within a Prompt.

class intelligence_layer.core.PromptTemplate(template_str: str)

Bases: object

Allows to build a Prompt using the liquid template language.

To add non-text prompt items first you have to save it to the template with the template.placeholder() function. To embed the items in the template, pass the placeholder in the place(s) where you would like the items.

Example

>>> from aleph_alpha_client import CompletionRequest, Tokens

>>> from intelligence_layer.core import PromptTemplate

>>> tokens = Tokens.from_token_ids([1, 2, 3])
>>> template = PromptTemplate(
...     '''{%- for name in names -%}
...     Hello {{name}}!
...     {% endfor -%}
...     {{ image }}
...     ''')
>>> placeholder = template.placeholder(tokens)
>>> names = ["World", "Rutger"]
>>> prompt = template.to_rich_prompt(names=names, image=placeholder)
>>> request = CompletionRequest(prompt=prompt)

embed_prompt(prompt: Prompt) → str

Embeds a prompt in a prompt template.

Adds whitespace between text items if there is no whitespace between them. In case of non-text prompt items, this embeds them into the end result.

Parameters:

prompt – prompt to embed in the template

Example

>>> from aleph_alpha_client import Prompt, Text, Tokens

>>> from intelligence_layer.core import PromptTemplate

>>> user_prompt = Prompt([
... Tokens.from_token_ids([1, 2, 3]),
... Text.from_text("cool"),
... ])
>>> template = PromptTemplate("Question: {{user_prompt}}\n Answer: ")
>>> prompt = template.to_rich_prompt(user_prompt=template.embed_prompt(user_prompt))

Returns:

The prompt template with the embedded prompt.

placeholder(value: Image | Tokens) → Placeholder

Saves a non-text prompt item to the template and returns a placeholder.

The placeholder is used to embed the prompt item in the template

Parameters:

value – Tokens to store

Returns:

A placeholder for the given non-text item.

to_rich_prompt(**kwargs: Any) → RichPrompt

Creates a Prompt along with metadata from the template string and the given parameters.

Currently, the only metadata returned is information about ranges that are marked in the template. Provided parameters are passed to liquid.Template.render.

Parameters:

**kwargs – Parameters to enrich prompt with

Returns:

The rendered prompt as a RichPrompt

class intelligence_layer.core.RichPrompt(items: ~typing.Sequence[~aleph_alpha_client.prompt.Text | ~aleph_alpha_client.prompt.Tokens | ~aleph_alpha_client.prompt.Image], ranges: ~collections.abc.Mapping[str, ~collections.abc.Sequence[~intelligence_layer.core.prompt_template.PromptRange]] = <factory>)

Bases: Prompt

The Prompt along with some metadata generated when a PromptTemplate is turned into a Prompt.

Parameters:

ranges – A mapping of range name to a Sequence of corresponding PromptRange instances.

static from_tokens(tokens: Sequence[int], controls: Sequence[TokenControl] | None = None) → Prompt

Examples

>>> prompt = Prompt.from_tokens([1, 2, 3])

class intelligence_layer.core.ScoredTextHighlight(*, start: int, end: int, score: float)

Bases: BaseModel

A substring of the input prompt scored for relevance with regard to the output.

start

The start index of the highlight.

Type:

int

end

The end index of the highlight.

Type:

int

score

The score of the highlight. Normalized to be between zero and one, with higher being more important.

Type:

float

class intelligence_layer.core.Span(context: Context | None = None)

Bases: Tracer, AbstractContextManager[Span]

Captures a logical step within the overall workflow.

Logs and other spans can be nested underneath.

Can also be used as a context manager to easily capture the start and end time, and keep the span only in scope while it is open.

context

The context of the current span. If the span is a root span, the trace id will be equal to its span id.

Type:

intelligence_layer.core.tracer.tracer.Context

status_code

Status of the span. Will be “OK” unless the span was interrupted by an exception.

abstract end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

abstract export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

abstract log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

abstract span(name: str, timestamp: datetime | None = None) → Span

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

abstract task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → TaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.SpanAttributes(*, type: Literal[SpanType.SPAN] = SpanType.SPAN)

Bases: BaseModel

class intelligence_layer.core.SpanStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

class intelligence_layer.core.SpanType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

capitalize()

Return a capitalized version of the string.

More specifically, make the first character have upper case and the rest lower case.

casefold()

Return a version of the string suitable for caseless comparisons.

center(width, fillchar=' ', /)

Return a centered string of length width.

Padding is done using the specified fill character (default is a space).

count(sub[, start[, end]]) → int

Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.

encode(encoding='utf-8', errors='strict')

Encode the string using the codec registered for encoding.

encoding

The encoding in which to encode the string.

errors

The error handling scheme to use for encoding errors. The default is ‘strict’ meaning that encoding errors raise a UnicodeEncodeError. Other possible values are ‘ignore’, ‘replace’ and ‘xmlcharrefreplace’ as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.

endswith(suffix[, start[, end]]) → bool

Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.

expandtabs(tabsize=8)

Return a copy where all tab characters are expanded using spaces.

If tabsize is not given, a tab size of 8 characters is assumed.

find(sub[, start[, end]]) → int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

format(*args, **kwargs) → str

Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces (‘{’ and ‘}’).

format_map(mapping) → str

Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces (‘{’ and ‘}’).

index(sub[, start[, end]]) → int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

isalnum()

Return True if the string is an alpha-numeric string, False otherwise.

A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.

isalpha()

Return True if the string is an alphabetic string, False otherwise.

A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.

isascii()

Return True if all characters in the string are ASCII, False otherwise.

ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.

isdecimal()

Return True if the string is a decimal string, False otherwise.

A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.

isdigit()

Return True if the string is a digit string, False otherwise.

A string is a digit string if all characters in the string are digits and there is at least one character in the string.

isidentifier()

Return True if the string is a valid Python identifier, False otherwise.

Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as “def” or “class”.

islower()

Return True if the string is a lowercase string, False otherwise.

A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.

isnumeric()

Return True if the string is a numeric string, False otherwise.

A string is numeric if all characters in the string are numeric and there is at least one character in the string.

isprintable()

Return True if the string is printable, False otherwise.

A string is printable if all of its characters are considered printable in repr() or if it is empty.

isspace()

Return True if the string is a whitespace string, False otherwise.

A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.

istitle()

Return True if the string is a title-cased string, False otherwise.

In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.

isupper()

Return True if the string is an uppercase string, False otherwise.

A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.

join(iterable, /)

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: ‘.’.join([‘ab’, ‘pq’, ‘rs’]) -> ‘ab.pq.rs’

ljust(width, fillchar=' ', /)

Return a left-justified string of length width.

Padding is done using the specified fill character (default is a space).

lower()

Return a copy of the string converted to lowercase.

lstrip(chars=None, /)

Return a copy of the string with leading whitespace removed.

If chars is given and not None, remove characters in chars instead.

static maketrans()

Return a translation table usable for str.translate().

If there is only one argument, it must be a dictionary mapping Unicode ordinals (integers) or characters to Unicode ordinals, strings or None. Character keys will be then converted to ordinals. If there are two arguments, they must be strings of equal length, and in the resulting dictionary, each character in x will be mapped to the character at the same position in y. If there is a third argument, it must be a string, whose characters will be mapped to None in the result.

partition(sep, /)

Partition the string into three parts using the given separator.

This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing the original string and two empty strings.

removeprefix(prefix, /)

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

removesuffix(suffix, /)

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.

replace(old, new, count=-1, /)

Return a copy with all occurrences of substring old replaced by new.

count

Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.

If the optional argument count is given, only the first count occurrences are replaced.

rfind(sub[, start[, end]]) → int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

rindex(sub[, start[, end]]) → int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

rjust(width, fillchar=' ', /)

Return a right-justified string of length width.

Padding is done using the specified fill character (default is a space).

rpartition(sep, /)

Partition the string into three parts using the given separator.

This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing two empty strings and the original string.

rsplit(sep=None, maxsplit=-1)

Return a list of the substrings in the string, using sep as the separator string.

sep

The separator used to split the string.

When set to None (the default value), will split on any whitespace character (including n r t f and spaces) and will discard empty strings from the result.

maxsplit

Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the end of the string and works to the front.

rstrip(chars=None, /)

Return a copy of the string with trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

split(sep=None, maxsplit=-1)

Return a list of the substrings in the string, using sep as the separator string.

sep

The separator used to split the string.

When set to None (the default value), will split on any whitespace character (including n r t f and spaces) and will discard empty strings from the result.

maxsplit

Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the front of the string and works to the end.

Note, str.split() is mainly useful for data that has been intentionally delimited. With natural text that includes punctuation, consider using the regular expression module.

splitlines(keepends=False)

Return a list of the lines in the string, breaking at line boundaries.

Line breaks are not included in the resulting list unless keepends is given and true.

startswith(prefix[, start[, end]]) → bool

Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.

strip(chars=None, /)

Return a copy of the string with leading and trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

swapcase()

Convert uppercase characters to lowercase and lowercase characters to uppercase.

title()

Return a version of the string where each word is titlecased.

More specifically, words start with uppercased characters and all remaining cased characters have lower case.

translate(table, /)

Replace each character in the string using the given translation table.

table

Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.

The table must implement lookup/indexing via __getitem__, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.

upper()

Return a copy of the string converted to uppercase.

zfill(width, /)

Pad a numeric string with zeros on the left, to fill a field of the given width.

The string is never truncated.

class intelligence_layer.core.StartSpan(*, uuid: UUID, parent: UUID, name: str, start: datetime, trace_id: UUID)

Bases: BaseModel

Represents the payload/entry of a log-line indicating that a Span was opened through Tracer.span.

uuid

A unique id for the opened Span.

Type:

uuid.UUID

parent

The unique id of the parent element of opened TaskSpan. This could refer to either a surrounding TaskSpan, Span or the top-level Tracer.

Type:

uuid.UUID

name

The name of the task.

Type:

str

start

The timestamp when this Span was started.

Type:

datetime.datetime

trace_id

The ID of the trace this span belongs to.

Type:

uuid.UUID

class intelligence_layer.core.StartTask(*, uuid: UUID, parent: UUID, name: str, start: datetime, input: Annotated[PydanticSerializable, SerializeAsAny()], trace_id: UUID)

Bases: BaseModel

Represents the payload/entry of a log-line indicating that a TaskSpan was opened through Tracer.task_span.

uuid

A unique id for the opened TaskSpan.

Type:

uuid.UUID

parent

The unique id of the parent element of opened TaskSpan. This could refer to either a surrounding TaskSpan, Span or the top-level Tracer.

Type:

uuid.UUID

name

The name of the task.

Type:

str

start

The timestamp when this Task was started (i.e. run was called).

Type:

datetime.datetime

input

The Input (i.e. parameter for run) the Task was started with.

Type:

PydanticSerializable

trace_id

The trace id of the opened TaskSpan.

Type:

uuid.UUID

class intelligence_layer.core.Task

Bases: ABC, Generic[Input, Output]

Base task interface. This may consist of several sub-tasks to accomplish the given task.

Generics:

Input: Interface to be passed to the task with all data needed to run the process.

Ideally, these are specified in terms related to the use-case, rather than lower-level configuration options.

Output: Interface of the output returned by the task.

abstract do_run(input: Input, task_span: TaskSpan) → Output

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

final run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

final run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.TaskSpan(context: Context | None = None)

Bases: Span

Specialized span for instrumenting Task input, output, and nested spans and logs.

Generating this TaskSpan should capture the Task input, as well as the task name.

Can also be used as a Context Manager to easily capture the start and end time of the task, and keep the span only in scope while it is active

abstract end(timestamp: datetime | None = None) → None

Marks the Span as closed and sets the end time.

The Span should be regarded as complete, and no further logging should happen with it.

Ending a closed span in undefined behavior.

Parameters:

timestamp – Optional override of the timestamp. Defaults to call time.

abstract export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

abstract log(message: str, value: PydanticSerializable, timestamp: datetime | None = None) → None

Record a log of relevant information as part of a step within a task.

By default, the Input and Output of each Task are logged automatically, but you can log anything else that seems relevant to understanding the process of a given task.

Logging to closed spans is undefined behavior.

Parameters:

• message – A description of the value you are logging, such as the step in the task this is related to.

• value – The relevant data you want to log. Can be anything that is serializable by Pydantic, which gives the tracers flexibility in how they store and emit the logs.

• timestamp – optional override of the timestamp. Otherwise should default to now

abstract record_output(output: PydanticSerializable) → None

Record Task output.

Since a Context Manager can’t capture output in the __exit__ method, output should be captured once it is generated.

Parameters:

output – The output of the task that is being logged.

abstract span(name: str, timestamp: datetime | None = None) → Span

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

abstract task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → TaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

class intelligence_layer.core.TaskSpanAttributes(*, type: Literal[SpanType.TASK_SPAN] = SpanType.TASK_SPAN, input: Annotated[PydanticSerializable, SerializeAsAny()], output: Annotated[PydanticSerializable, SerializeAsAny()])

Bases: BaseModel

class intelligence_layer.core.TextCursor(item: int, position: int)

Bases: object

Defines a position with a Text prompt item.

Parameters:

• item – the index of the prompt item within the Prompt

• position – the character position in the text of the item.

Example: >>> from aleph_alpha_client import Prompt >>> from intelligence_layer.core import TextCursor >>> prompt = Prompt.from_text(“This is a text”) >>> # This denotes the “i” in “is” in the text-item of the Prompt above >>> cursor = TextCursor(item=0, position=5)

class intelligence_layer.core.TextHighlight(model: AlephAlphaModel, granularity: PromptGranularity | None = None, threshold: float = 0.1, clamp: bool = False)

Bases: Task[TextHighlightInput, TextHighlightOutput]

Generates text highlights given a prompt and completion.

For a given prompt and target (completion), extracts the parts of the prompt responsible for generation. The prompt can only contain text. A range can be provided via use of the liquid language (see the example). In this case, the highlights will only refer to text within this range.

Parameters:

• model – The model used throughout the task for model related API calls.

• granularity – At which granularity should the target be explained in terms of the prompt.

• threshold – After normalization, everything highlight below this value will be dropped.

• clamp – Control whether highlights should be clamped to a focus range if they intersect it.

Example

>>> import os

>>> from intelligence_layer.core import (
...     InMemoryTracer,
...     PromptTemplate,
...     TextHighlight,
...     TextHighlightInput,
...     AlephAlphaModel
... )

>>> model = AlephAlphaModel(name="luminous-base")
>>> text_highlight = TextHighlight(model=model)
>>> prompt_template_str = (
...             "{% promptrange r1 %}Question: What is 2 + 2?{% endpromptrange %}\nAnswer:"
...     )
>>> template = PromptTemplate(prompt_template_str)
>>> rich_prompt = template.to_rich_prompt()
>>> completion = " 4."
>>> model = "luminous-base"
>>> input = TextHighlightInput(
...         rich_prompt=rich_prompt, target=completion, focus_ranges=frozenset({"r1"})
... )
>>> output = text_highlight.run(input, InMemoryTracer())

do_run(input: TextHighlightInput, task_span: TaskSpan) → TextHighlightOutput

The implementation for this use case.

This takes an input and runs the implementation to generate an output. It takes a Span for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• task_span – The Span used for tracing.

Returns:

Generic output defined by the task implementation.

run(input: Input, tracer: Tracer) → Output

Executes the implementation of do_run for this use case.

This takes an input and runs the implementation to generate an output. It takes a Tracer for tracing of the process. The Input and Output are logged by default.

Parameters:

• input – Generic input defined by the task implementation

• tracer – The Tracer used for tracing.

Returns:

Generic output defined by the task implementation.

run_concurrently(inputs: Iterable[Input], tracer: Tracer, concurrency_limit: int = 20) → Sequence[Output]

Executes multiple processes of this task concurrently.

Each provided input is potentially executed concurrently to the others. There is a global limit on the number of concurrently executed tasks that is shared by all tasks of all types.

Parameters:

• inputs – The inputs that are potentially processed concurrently.

• tracer – The tracer passed on the run method when executing a task.

• concurrency_limit – An optional additional limit for the number of concurrently executed task for this method call. This can be used to prevent queue-full or similar error of downstream APIs when the global concurrency limit is too high for a certain task.

Returns:

The Outputs generated by calling run for each given Input. The order of Outputs corresponds to the order of the Inputs.

class intelligence_layer.core.TextHighlightInput(*, rich_prompt: RichPrompt, target: str, focus_ranges: frozenset[str] = frozenset({}))

Bases: BaseModel

The input for a text highlighting task.

rich_prompt

From client’s PromptTemplate. Includes both the actual ‘Prompt’ as well as text range information. Supports liquid-template-language-style {% promptrange range_name %}/{% endpromptrange %} for range.

Type:

intelligence_layer.core.prompt_template.RichPrompt

target

The target that should be explained. Expected to follow the prompt.

Type:

str

focus_ranges

The ranges contained in rich_prompt the returned highlights stem from. That means that each returned highlight overlaps with at least one character with one of the ranges listed here. If this set is empty highlights of the entire prompt are returned.

Type:

frozenset[str]

class intelligence_layer.core.TextHighlightOutput(*, highlights: Sequence[ScoredTextHighlight])

Bases: BaseModel

The output of a text highlighting task.

highlights

A sequence of ‘ScoredTextHighlight’s.

Type:

collections.abc.Sequence[intelligence_layer.core.text_highlight.ScoredTextHighlight]

class intelligence_layer.core.Token(*, token: str, token_id: int)

Bases: BaseModel

A token class containing it’s id and the raw token.

This is used instead of the Aleph Alpha client Token class since this one is serializable, while the one from the client is not.

class intelligence_layer.core.TokenWithLogProb(*, token: Token, prob: LogProb)

Bases: BaseModel

class intelligence_layer.core.Tracer

Bases: ABC

Provides a consistent way to instrument a Task with logging for each step of the workflow.

A tracer needs to provide a way to collect an individual log, which should be serializable, and a way to generate nested spans, so that sub-tasks can emit logs that are grouped together.

Implementations of how logs are collected and stored may differ. Refer to the individual documentation of each implementation to see how to use the resulting tracer.

abstract export_for_viewing() → Sequence[ExportedSpan]

Converts the trace to a format that can be read by the trace viewer.

The format is inspired by the OpenTelemetry Format, but does not abide by it. Specifically, it cuts away unused concepts, such as links.

Returns:

A list of spans which includes the current span and all its child spans.

abstract span(name: str, timestamp: datetime | None = None) → Span

Generate a span from the current span or logging instance.

Allows for grouping multiple logs and duration together as a single, logical step in the process.

Each tracer implementation can decide on how it wants to represent this, but they should all capture the hierarchical nature of nested spans, as well as the idea of the duration of the span.

Parameters:

• name – A descriptive name of what this span will contain logs about.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a Span.

abstract task_span(task_name: str, input: PydanticSerializable, timestamp: datetime | None = None) → TaskSpan

Generate a task-specific span from the current span or logging instance.

Allows for grouping multiple logs together, as well as the task’s specific input, output, and duration.

Each tracer implementation can decide on how it wants to represent this, but they should all allow for representing logs of a span within the context of a parent span.

Parameters:

• task_name – The name of the task that is being logged

• input – The input for the task that is being logged.

• timestamp – Override of the starting timestamp. Defaults to call time.

Returns:

An instance of a TaskSpan.

exception intelligence_layer.core.TracerLogEntryFailed(error_message: str, id: str)

Bases: Exception

add_note()

Exception.add_note(note) – add a note to the exception

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

intelligence_layer.core.utc_now() → datetime

Return datetime object with utc timezone.

datetime.utcnow() returns a datetime object without timezone, so this function is preferred.