camel.verifiers package

Submodules

camel.verifiers.base module

class camel.verifiers.base.BaseVerifier(extractor: BaseExtractor | None = None, max_parallel: int | None = None, timeout: float | None = None, max_retries: int = 3, retry_delay: float = 1.0, initial_batch_size: int | None = None, cpu_threshold: float = 80.0, memory_threshold: float = 85.0, **kwargs)

Bases: ABC

Base class for all verifiers.

Example

`python verifier = MyVerifier() await verifier.setup() result = await verifier.verify(response) await verifier.cleanup() `

Key Features: - Async verification with retry logic - Comprehensive error handling and logging - Configurable batch processing - Resource monitoring for adaptive scaling

async cleanup() → None

Clean up verifier resources.

Ensures: 1. Batch processor is reset 2. All internal states are cleared

Raises:

RuntimeError – If cleanup fails.

async setup(**kwargs) → None

Set up the verifier with necessary resources.

Initializes: 1. Batch processor with validated parameters 2. Any verifier-specific resources

Raises:

RuntimeError – If setup fails or resources cannot be initialized.

async verify(solution: str, ground_truth: str | None) → VerificationResult

Perform verification with full error handling.

This method verifies the correctness of a generated solution by comparing it against the provided ground truth. It handles execution errors, timeouts, and retry attempts to ensure robust validation.

Parameters:

• solution (str) – The generated response that needs verification.

• ground_truth (Optional[str]) – The expected correct answer to compare against.

Returns:

A structured object containing:

• status (SUCCESS/FAILURE/ERROR/TIMEOUT)

• result (str): The verification outcome or processed output.

• duration (float): Time taken for verification.

• metadata (dict): Additional details such as retry attempts.

• error_message (Optional[str]): Error description,

if applicable.

Return type:

VerificationResult

Raises:

• RuntimeError – If verification fails unexpectedly.

• asyncio.TimeoutError – If verification exceeds the time limit.

async verify_batch(solutions: List[str], ground_truths: List[str | None], raise_on_error: bool = False) → List[VerificationResult]

Verify multiple solutions in parallel with controlled concurrency.

This method verifies multiple generated solutions against their respective ground truths using parallel execution. It handles timeouts, execution errors, and batch processing optimizations.

Parameters:

• solutions (List[str]) – A list of generated solutions to be verified.

• ground_truths (List[Optional[str]]) – A list of expected outputs for comparison. Each element corresponds to a solution.

• raise_on_error (bool, optional) – If True, raises an exception if any verification fails. (default: False)

Returns:

A list of verification results, one per

input solution.

Return type:

List[VerificationResult]

Raises:

• RuntimeError – If any verification fails and raise_on_error is True.

• asyncio.TimeoutError – If verifications time out after maximum retries.

camel.verifiers.models module

class camel.verifiers.models.VerificationOutcome(value)

Bases: Enum

Enum representing the status of a verification.

ERROR = 'error'

FAILURE = 'failure'

SUCCESS = 'success'

TIMEOUT = 'timeout'

class camel.verifiers.models.VerificationResult(*, status: VerificationOutcome, result: str, duration: float = 0.0, timestamp: datetime = None, metadata: Dict[str, Any] = None, error_message: str | None = None)

Bases: BaseModel

Structured result from a verification.

duration: float

error_message: str | None

metadata: Dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'duration': FieldInfo(annotation=float, required=False, default=0.0, description='Duration of verification in seconds'), 'error_message': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Error message if verification failed'), 'metadata': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Additional metadata about the verification'), 'result': FieldInfo(annotation=str, required=True, description='Verification result'), 'status': FieldInfo(annotation=VerificationOutcome, required=True, description='Status of the verification'), 'timestamp': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method, description='When the verification was performed')}

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

This replaces Model.__fields__ from Pydantic V1.

result: str

status: VerificationOutcome

timestamp: datetime

class camel.verifiers.models.VerifierConfig(*, enabled: bool = True, strict_mode: bool = False, timeout: float | None = None, max_retries: int = 3, retry_delay: float = 1.0)

Bases: BaseModel

Configuration for verifier behavior.

enabled: bool

max_retries: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'enabled': FieldInfo(annotation=bool, required=False, default=True, description='Whether verification is enabled'), 'max_retries': FieldInfo(annotation=int, required=False, default=3, description='Maximum number of retry attempts'), 'retry_delay': FieldInfo(annotation=float, required=False, default=1.0, description='Delay between retries in seconds'), 'strict_mode': FieldInfo(annotation=bool, required=False, default=False, description='Whether to fail on any validation error'), 'timeout': FieldInfo(annotation=Union[float, NoneType], required=False, default=None, description='Verification timeout in seconds')}

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

This replaces Model.__fields__ from Pydantic V1.

retry_delay: float

strict_mode: bool

timeout: float | None

camel.verifiers.python_verifier module

class camel.verifiers.python_verifier.PythonVerifier(extractor: BaseExtractor | None = None, timeout: float | None = 30.0, required_packages: List[str] | None = None, **kwargs)

Bases: BaseVerifier

The PythonVerifier class verifies Python-based implementations by executing them in an isolated virtual environment.

Features: - Creates a virtual environment with a specified Python version. - Installs required packages before executing the provided script. - Executes the script and compares the output against a ground truth,

if supplied.

• Automatically cleans up the virtual environment after execution.

The verification process ensures that the code runs in a controlled environment, minimizing external dependencies and conflicts.

Module contents

class camel.verifiers.BaseVerifier(extractor: BaseExtractor | None = None, max_parallel: int | None = None, timeout: float | None = None, max_retries: int = 3, retry_delay: float = 1.0, initial_batch_size: int | None = None, cpu_threshold: float = 80.0, memory_threshold: float = 85.0, **kwargs)

Bases: ABC

Base class for all verifiers.

Example

`python verifier = MyVerifier() await verifier.setup() result = await verifier.verify(response) await verifier.cleanup() `

Key Features: - Async verification with retry logic - Comprehensive error handling and logging - Configurable batch processing - Resource monitoring for adaptive scaling

async cleanup() → None

Clean up verifier resources.

Ensures: 1. Batch processor is reset 2. All internal states are cleared

Raises:

RuntimeError – If cleanup fails.

async setup(**kwargs) → None

Set up the verifier with necessary resources.

Initializes: 1. Batch processor with validated parameters 2. Any verifier-specific resources

Raises:

RuntimeError – If setup fails or resources cannot be initialized.

async verify(solution: str, ground_truth: str | None) → VerificationResult

Perform verification with full error handling.

This method verifies the correctness of a generated solution by comparing it against the provided ground truth. It handles execution errors, timeouts, and retry attempts to ensure robust validation.

Parameters:

• solution (str) – The generated response that needs verification.

• ground_truth (Optional[str]) – The expected correct answer to compare against.

Returns:

A structured object containing:

• status (SUCCESS/FAILURE/ERROR/TIMEOUT)

• result (str): The verification outcome or processed output.

• duration (float): Time taken for verification.

• metadata (dict): Additional details such as retry attempts.

• error_message (Optional[str]): Error description,

if applicable.

Return type:

VerificationResult

Raises:

• RuntimeError – If verification fails unexpectedly.

• asyncio.TimeoutError – If verification exceeds the time limit.

async verify_batch(solutions: List[str], ground_truths: List[str | None], raise_on_error: bool = False) → List[VerificationResult]

Verify multiple solutions in parallel with controlled concurrency.

This method verifies multiple generated solutions against their respective ground truths using parallel execution. It handles timeouts, execution errors, and batch processing optimizations.

Parameters:

• solutions (List[str]) – A list of generated solutions to be verified.

• ground_truths (List[Optional[str]]) – A list of expected outputs for comparison. Each element corresponds to a solution.

• raise_on_error (bool, optional) – If True, raises an exception if any verification fails. (default: False)

Returns:

A list of verification results, one per

input solution.

Return type:

List[VerificationResult]

Raises:

• RuntimeError – If any verification fails and raise_on_error is True.

• asyncio.TimeoutError – If verifications time out after maximum retries.

class camel.verifiers.PythonVerifier(extractor: BaseExtractor | None = None, timeout: float | None = 30.0, required_packages: List[str] | None = None, **kwargs)

Bases: BaseVerifier

The PythonVerifier class verifies Python-based implementations by executing them in an isolated virtual environment.

Features: - Creates a virtual environment with a specified Python version. - Installs required packages before executing the provided script. - Executes the script and compares the output against a ground truth,

if supplied.

• Automatically cleans up the virtual environment after execution.

The verification process ensures that the code runs in a controlled environment, minimizing external dependencies and conflicts.

class camel.verifiers.VerificationOutcome(value)

Bases: Enum

Enum representing the status of a verification.

ERROR = 'error'

FAILURE = 'failure'

SUCCESS = 'success'

TIMEOUT = 'timeout'