whylogs_container.whylabs.llm_validation.validator

Classes

AsyncValidationMessage(request, timestamp, ...)
QueueLock()
QueueLockEvents(work_event, enqueue_event)
ReturnType(result, score_result[, wf_metadata])
SynchronousValidator()	This class is a convenience wrapper that has all of the logic required for synchronously calling the ValidationActor.
ValidationActor(config_instance, ...)
ValidationMessage(request, timestamp, ...)

class whylogs_container.whylabs.llm_validation.validator.AsyncValidationMessage(request: bytes, timestamp: int, api_key: whylogs_container.whylabs.container.dependencies.api_key.WhyLabsApiKey, log: bool = True, trace: bool = True, id: str = <factory>, carrier: Optional[Dict[str, Any]] = None)

Bases: ValidationMessage

class whylogs_container.whylabs.llm_validation.validator.QueueLock

Bases: Thread

close() → None

request(work_event: Event) → Event

run() → None

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class whylogs_container.whylabs.llm_validation.validator.QueueLockEvents(work_event: threading.Event, enqueue_event: threading.Event)

Bases: object

enqueue_event: Event

work_event: Event

class whylogs_container.whylabs.llm_validation.validator.ReturnType(result: langkit.core.workflow.WorkflowResult, score_result: langkit.core.workflow.WorkflowResult | None, wf_metadata: Dict[str, Dict[str, Any]] | None = None)

Bases: object

result: WorkflowResult

score_result: WorkflowResult | None

wf_metadata: Dict[str, Dict[str, Any]] | None = None

class whylogs_container.whylabs.llm_validation.validator.SynchronousValidator

Bases: ABC

This class is a convenience wrapper that has all of the logic required for synchronously calling the ValidationActor. There shouldn’t be any real logic in here that does validation, just calls to actor.send and synchronous utilities.

abstract close_validator() → None

evaluate(request: LLMValidateRequest, active_predicate: Callable[[], bool] | Coroutine[None, None, bool], api_key: WhyLabsApiKey, log: bool = True, trace: bool = True, carrier: Dict[str, Any] | None = None) → ReturnType

Perform validation on the given LLM message and wait for the result.

abstract start_validator() → None

warm_caches() → None

class whylogs_container.whylabs.llm_validation.validator.ValidationActor(config_instance: ConfigInstance, rolling_logger: ContainerProcessRollingLogger | None, event_loop: AbstractEventLoop)

Bases: ProcessActor[ValidationMessage | AsyncValidationMessage, ReturnType | None], SynchronousValidator

close_validator() → None

process_async_validation_message(messages: List[AsyncValidationMessage]) → None

process_batch(batch, batch_type) → None

process_validation_message(messages: List[ValidationMessage]) → None

process_validation_messages(messages: List[AsyncValidationMessage] | List[ValidationMessage], return_results: bool) → None

run() → None

Method to be run in sub-process; can be overridden in sub-class

start() → None

The process version of the actor apparently has to be manually started after it’s created, unlike the thread version which can just be automatically started from within its init. There must be some post-init setup that needs to be done.

start_validator() → None

class whylogs_container.whylabs.llm_validation.validator.ValidationMessage(request: bytes, timestamp: int, api_key: whylogs_container.whylabs.container.dependencies.api_key.WhyLabsApiKey, log: bool = True, trace: bool = True, id: str = <factory>, carrier: Optional[Dict[str, Any]] = None)

Bases: object

api_key: WhyLabsApiKey

carrier: Dict[str, Any] | None = None

id: str

log: bool = True

request: bytes

timestamp: int

to_validation_request() → LLMValidateRequest

trace: bool = True