verdict package

Submodules

verdict.config module

class verdict.config.Config(rate_limiter_disabled: bool = False)

Bases: object

rate_limiter_disabled: bool = False

verdict.dataset module

class verdict.dataset.DatasetWrapper(dataset: Dataset, input_fn: InputFn | None = None, columns: List[str] | None = None, max_samples: int | None = None)

Bases: Iterator[Tuple[Dict[str, Any], Schema]]

dataset: pd.DataFrame

static from_hf(dataset: Dict[str, 'Dataset'], input_fn: InputFn | None = None, columns: List[str] | None = None, max_samples: int | None = None, expand: bool = False) → Dict[str, 'DatasetWrapper']

static from_pandas(df: pd.DataFrame, input_fn: InputFn | None = None, columns: List[str] | None = None, split_column: str | None = None, max_samples: int | None = None)

input_fn: InputFn

static load(path: Path) → DatasetWrapper

max_samples: int | None

samples: pd.DataFrame

save(path: Path) → None

verdict.extractor module

class verdict.extractor.ArgmaxScoreExtractor

Bases: TokenProbabilityExtractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema, Usage]

class verdict.extractor.CustomExtractor

Bases: RawExtractor, ABC

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema, Usage]

inject(unit) → None

abstract post_extract(output: str, logger: Logger) → Dict[str, Any]

class verdict.extractor.Extractor

Bases: ABC

Represents a method of extracting a ResponseSchema from a provider call.

Some examples:

function-calling / structured output via instructor
obtaining probability using logprobs on some token support (eg, yes, no)
having a second LLM extract from a raw response string

abstract extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema | Iterator[Schema], Usage]

classmethod format() → str

inject(unit) → None

response_schema: Type[Schema]

streaming: bool = False

class verdict.extractor.PostHocExtractor(policy_or_name: str | Model | List[str | Model] | None = None, retries: int = 1, **inference_parameters)

Bases: StructuredOutputExtractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema | Iterator[Schema], Usage]

extract_client_wrappers: List[ClientWrapper] | None = None

format() → str

model_selection_policy: ModelSelectionPolicy | None = None

class verdict.extractor.RawExtractor

Bases: Extractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema | Iterator[Schema], Usage]

field_name: str

inject(unit) → None

class verdict.extractor.RegexExtractor(fields: Dict[str, str])

Bases: CustomExtractor

FIRST_FLOAT = '[+-]?\\d+(\\.\\d+)?'

FIRST_INT = '[+-]?\\d+'

fields: Dict[str, Pattern]

post_extract(output: str, logger: Logger) → Dict[str, Any]

class verdict.extractor.SampleScoreExtractor

Bases: TokenProbabilityExtractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema, Usage]

class verdict.extractor.StructuredOutputExtractor

Bases: Extractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema | Iterator[Schema], Usage]

class verdict.extractor.TokenProbabilityExtractor

Bases: Extractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema, Usage]

field_name: str

inject(unit) → None

scale: DiscreteScale

stream(stream: bool = False) → Self

class verdict.extractor.Usage(in_tokens: int, out_tokens: int)

Bases: object

in_tokens: int

is_unknown() → bool

out_tokens: int

static unknown() → Usage

class verdict.extractor.WeightedSummedScoreExtractor

Bases: TokenProbabilityExtractor

extract(client_wrapper: ClientWrapper, prompt_message: PromptMessage, logger: Logger) → Tuple[Schema, Usage]

inject(unit) → None

verdict.model module

class verdict.model.Client(complete: Callable, model: verdict.model.Model, inference_parameters: dict[str, Any])

Bases: object

complete: Callable

defaults(**inference_parameters_defaults) → ContextManager[None, bool | None]

inference_parameters: dict[str, Any]

model: Model

class verdict.model.ClientWrapper(model: Model, **inference_parameters)

Bases: object

encode(word: str) → List[int]

static from_model(model: Model, **inference_parameters) → ClientWrapper

inference_parameters: dict[str, Any]

model: Model

class verdict.model.Model

Bases: ABC

property char: str

property connection_parameters: dict[str, Any]

name: str

rate_limit: RateLimitPolicy = Field(name=None,type=None,default=<dataclasses._MISSING_TYPE object>,default_factory=<dataclasses._MISSING_TYPE object>,init=False,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=<dataclasses._MISSING_TYPE object>,_field_type=None)

rate_limiter: RateLimitPolicy | Dict[RateLimiter, str | RateLimiterMetric] | None = Field(name=None,type=None,default=None,default_factory=<dataclasses._MISSING_TYPE object>,init=True,repr=False,hash=None,compare=True,metadata=mappingproxy({}),kw_only=<dataclasses._MISSING_TYPE object>,_field_type=None)

use_nonce: bool = False

class verdict.model.ModelConfigurable

Bases: ABC

property model_selection_policy: ModelSelectionPolicy | None

abstract set(attr, value) → None

via(policy_or_name: ModelSelectionPolicy | str | Model | List[str | Model], retries: int = 1, **inference_parameters) → Self

class verdict.model.ModelSelectionPolicy

Bases: object

property char: str

static from_any(policy_or_name: ModelSelectionPolicy | str | Model | List[str | Model], retries: int = 1, **inference_parameters) → ModelSelectionPolicy

static from_name(model: str | Model, retries: int = 1, **inference_parameters) → ModelSelectionPolicy

static from_names(model_names: List[Tuple[str, int, dict]]) → ModelSelectionPolicy

get_clients() → Iterator[ClientWrapper]

class verdict.model.ProviderModel(name: str, use_nonce: bool = False, rate_limiter: verdict.util.ratelimit.RateLimitPolicy | Dict[verdict.util.ratelimit.RateLimiter, str | verdict.util.ratelimit.RateLimiterMetric] | NoneType = None)

Bases: Model

property connection_parameters: dict[str, Any]

name: str

rate_limiter: RateLimitPolicy | Dict[RateLimiter, str | RateLimiterMetric] | None = None

use_nonce: bool = False

class verdict.model.vLLMModel(name: str, api_base: str, api_key: str, rate_limiter: verdict.util.ratelimit.RateLimitPolicy | Dict[verdict.util.ratelimit.RateLimiter, str | verdict.util.ratelimit.RateLimiterMetric] | NoneType = None)

Bases: Model

api_base: str

api_key: str

property connection_parameters: dict[str, Any]

name: str

rate_limiter: RateLimitPolicy | Dict[RateLimiter, str | RateLimiterMetric] | None = None

verdict.prompt module

class verdict.prompt.PromptMessage(system: str | None, user: str, input_schema: verdict.schema.Schema | None = None)

Bases: object

input_schema: Schema | None = None

system: str | None

to_messages(add_nonce: bool = False) → List[Dict[str, str | List[Dict[str, str | Dict[str, str]]]]]

user: str

class verdict.prompt.PromptRegistry(name, bases, dct)

Bases: type

static extract_sections(template: str) → Tuple[str | None, str | None, bool]

static strip_prompt_template(prompt: str) → str

class verdict.prompt.Promptable

Bases: ABC

abstract populate_prompt_message(input: Schema, logger: Logger) → PromptMessage

verdict.scale module

class verdict.scale.BooleanScale(yes: List[str] = ['yes', 'Yes', 'YES'], no: List[str] = ['no', 'No', 'NO'])

Bases: DiscreteScale

pydantic_fields(key: str = 'output') → Dict[str, Tuple[Any, FieldInfo]]

token_support() → List[str]

value_mapping_fn(output: str) → bool

class verdict.scale.ContinuousScale(min_value: float, max_value: float, end_is_worst: bool = False)

Bases: Scale

prompt() → str

pydantic_fields(key: str = 'output') → Dict[str, Tuple[Any, FieldInfo]]

value_mapping_fn(output: float) → float

class verdict.scale.DiscreteScale(values: List[Any] | Tuple[Any, Any] | Tuple[Any, Any, int | None], end_is_worst: bool = False)

Bases: Scale

index(token: str) → int

prompt() → str

pydantic_fields(key: str = 'output') → Dict[str, Tuple[Any, FieldInfo]]

token_support() → List[str]

value_mapping_fn(output: str) → Any

values: List[Any]

verdict.scale.LikertScale(end_is_worst: bool = False) → Scale

class verdict.scale.Scale(scale_type: ScaleType, end_is_worst: bool = False)

Bases: ABC

T: Type

abstract prompt() → str

abstract pydantic_fields(key: str = 'output') → Dict[str, Tuple[Any, Any]]

class verdict.scale.ScaleType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

CONTINUOUS = 'continuous'

DISCRETE = 'discrete'

verdict.schema module

class verdict.schema.Schema

Bases: BaseModel, ABC

add(**kwargs) → Schema

classmethod append(**kwargs) → Type[Schema]

conform(expected: Type[Schema], logger: Logger | None = None) → Schema

Conform the current schema to the expected schema. For a missing field in this Schema (eg, in expected, but not in type(self)),

Check if there is a default factory for the field in expected. If so, use that.

Copy the first field in self that matches the type of the expected field.

classmethod empty() → Schema

escape() → str

static from_values(**kwargs) → Schema

static generate_key(field_info: FieldInfo) → str

classmethod get_scale(field_name: str) → Scale

static infer(**kwargs) → Type[Schema]

static infer_pydantic_annotation(obj: Any) → Any: Infer the appropriate Pydantic annotation for a given object.

static inline(**kwargs) → Type[Schema]

classmethod is_empty() → bool

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

static of(**kwargs) → Schema

classmethod prepend(**kwargs) → Type[Schema]

verdict.transform module

class verdict.transform.MapUnit(map_func: Callable[[Any | List[Any]], Any | List[Any]], **kwargs)

Bases: Unit

class InputSchema(*, values: Any | List[Any])

Bases: Schema

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None: We need to both initialize private attributes and call the user-defined model_post_init method.

values: Any | List[Any]

OutputSchema: alias of ResponseSchema

class ResponseSchema(*, values: Any | List[Any])

Bases: Schema

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None: We need to both initialize private attributes and call the user-defined model_post_init method.

values: Any | List[Any]

accumulate: bool = True

execute(input: InputSchema, execution_context: ExecutionContext | None = None) → ResponseSchema

Execute the unit with tracing.

Args:: input: The input schema. execution_context: The execution context for this execution (tracing, IDs, etc).
Returns:: The output schema.

lightweight: bool = True

map_func: Callable[[Any | List[Any]], Any | List[Any]]

class verdict.transform.MaxPoolUnit(fields: str | List[str] = [])

Bases: FieldMapUnit

fields: List[str]

class verdict.transform.MeanPoolUnit(fields: str | List[str] = [])

Bases: FieldMapUnit

fields: List[str]

class verdict.transform.MeanVariancePoolUnit(fields: str | List[str] = [])

Bases: FieldMapUnit

fields: List[str]

Module contents

class verdict.Block(name: str | None = None)

Bases: Graph[Unit | Layer], Node, ModelConfigurable

property char: str

leaf_view() → Block

link(other: Unit | Graph[Unit]) → None

materialize(_context: MaterializationContext | None = None) → Block

root_view() → Block

use_root: bool = True

verdict.Image(file: Path | str | None = None, mime: str | None = None, blob: str | None = None) → ImageModel

file: Path to the image file.

OR

mime: MIME type of the image. blob: base64-encoded image data.

class verdict.Layer(nodes: Node | List[Node], repeat: int = 1, inner: str | Inner = Inner.NONE, outer: str | Outer = Outer.DENSE)

Bases: Graph[Node], Node, ModelConfigurable

Ordered list of units.

property char: str

clone() → Self

Returns an associated deep copy of the node.

Used to create a new execution instance of a node.

copy() → Self

Returns a completely independent deep copy of the node.

Used in .from_sequence, etc.

how_inner: Inner

how_outer: Outer

leaf_idx: List[int]

property leaf_nodes: List[Node]

link(other: Unit | Layer | Block) → None

order: List[Node]

root_idx: List[int]

property root_nodes: List[Node]

sort() → List[Node]

with_leaf(idx: int | List[int]) → Layer

with_root(idx: int | List[int]) → Layer

class verdict.Pipeline(name: str = 'Pipeline', tracer: Tracer | List[Tracer] | None = None)

Bases: object

add_tracer(tracer: Tracer) → None

Add a tracer to the pipeline’s TracingManager.

Args:: tracer: The tracer to add.

block: Block

checkpoint(path: Path)

collect_outputs(executor: GraphExecutor, block_instance: Block) → Tuple[Dict[str, Schema], List[str]]

copy() → Pipeline

default_tracer: Tracer | List[Tracer] | None

executor: GraphExecutor

name: str

plot(display=False) → Image

restore(path: Path)

run(input_data: ~verdict.schema.Schema =, max_workers: int = 128, display: bool = False, graceful: bool = False, tracers: ~verdict.util.tracing.Tracer | ~typing.List[~verdict.util.tracing.Tracer] | None = None) → Tuple[Dict[str, Schema], List[str]]

Run the pipeline.

Args:: input_data: The input schema. max_workers: Number of workers. display: Whether to display live output. graceful: Whether to shut down gracefully. tracers: Optional tracer(s) to override the pipeline’s tracers for this run.
Returns:: Tuple of outputs and leaf node prefixes.

run_from_dataset(dataset: DatasetWrapper, max_workers: int = 128, experiment_config=None, display: bool = False, graceful: bool = False, tracers: Tracer | List[Tracer] | None = None) → Tuple['pd.DataFrame', List[str]]

Run the pipeline on a dataset.

Args:: dataset: The dataset wrapper. max_workers: Number of workers. experiment_config: Experiment config. display: Whether to display live output. graceful: Whether to shut down gracefully. tracers: Optional tracer(s) to override the pipeline’s tracers for this run.
Returns:: Tuple of DataFrame and leaf node prefixes.

run_from_list(dataset: List[Schema], max_workers: int = 128, experiment_config=None, display: bool = False, graceful: bool = False, tracers: Tracer | List[Tracer] | None = None) → Tuple[Dict[str, Schema], List[str]]

Run the pipeline on a list of Schemas.

Args:: dataset: List of input schemas. max_workers: Number of workers. experiment_config: Experiment config. display: Whether to display live output. graceful: Whether to shut down gracefully. tracers: Optional tracer(s) to override the pipeline’s tracers for this run.
Returns:: Tuple of outputs and leaf node prefixes.

via(policy_or_name: ModelSelectionPolicy | str, retries: int = 1, **inference_parameters) → Self

class verdict.Unit(**kwargs)

Bases: Node, Task, ModelConfigurable, Promptable, DataFlowSchema[InputSchemaT, ResponseSchemaT, OutputSchemaT]

property char: str

clone() → Self

Returns an associated deep copy of the node.

Used to create a new execution instance of a node.

copy() → Self

Returns a completely independent deep copy of the node.

Used in .from_sequence, etc.

data: UserState

property description: str | None

execute(input: InputSchemaT, execution_context: ExecutionContext | None = None) → OutputSchemaT

Execute the unit with tracing.

Args:: input: The input schema. execution_context: The execution context for this execution (tracing, IDs, etc).
Returns:: The output schema.

lightweight: bool = False

link(other: Unit | Graph[Unit]) → None

model_selection_policy: ModelSelectionPolicy | None = None

populate_prompt_message(input: Schema, logger: Logger) → PromptMessage

process(input: InputSchemaT, response: ResponseSchemaT) → OutputSchemaT | ResponseSchemaT

shared: SynchronizationState

validate(input: InputSchemaT, response: ResponseSchemaT) → None