Typesafe

zenml.integrations.typesafe

Initialization of the TypeSafe AI integration.

The integration ships hooks that ask TypeSafe's Jev model cheap, probability-scored questions about steps and runs and store the answers as ZenML run metadata.

Attributes

TYPESAFE = 'typesafe' module-attribute

Classes

Integration

Base class for integration in ZenML.

Methods:

activate() -> None classmethod

Abstract method to activate the integration.

Source code in src/zenml/integrations/integration.py

@classmethod
def activate(cls) -> None:
    """Abstract method to activate the integration."""

check_installation() -> bool classmethod

Method to check whether the required packages are installed.

Returns:

Type	Description
bool	True if all required packages are installed, False otherwise.

Source code in src/zenml/integrations/integration.py

@classmethod
def check_installation(cls) -> bool:
    """Method to check whether the required packages are installed.

    Returns:
        True if all required packages are installed, False otherwise.
    """
    for requirement in cls.get_requirements():
        parsed_requirement = Requirement(requirement)

        if not requirement_installed(parsed_requirement):
            logger.debug(
                "Requirement '%s' for integration '%s' is not installed "
                "or installed with the wrong version.",
                requirement,
                cls.NAME,
            )
            return False

        dependencies = get_dependencies(parsed_requirement)

        for dependency in dependencies:
            if not requirement_installed(dependency):
                logger.debug(
                    "Requirement '%s' for integration '%s' is not "
                    "installed or installed with the wrong version.",
                    dependency,
                    cls.NAME,
                )
                return False

    logger.debug(
        f"Integration '{cls.NAME}' is installed correctly with "
        f"requirements {cls.get_requirements()}."
    )
    return True

flavors() -> List[Type[Flavor]] classmethod

Abstract method to declare new stack component flavors.

Returns:

Type	Description
List[Type[Flavor]]	A list of new stack component flavors.

Source code in src/zenml/integrations/integration.py

@classmethod
def flavors(cls) -> List[Type[Flavor]]:
    """Abstract method to declare new stack component flavors.

    Returns:
        A list of new stack component flavors.
    """
    return []

get_requirements(target_os: Optional[str] = None, python_version: Optional[str] = None) -> List[str] classmethod

Method to get the requirements for the integration.

Parameters:

Name	Type	Description	Default
target_os	Optional[str]	The target operating system to get the requirements for.	None
python_version	Optional[str]	The Python version to use for the requirements.	None

Returns:

Type	Description
List[str]	A list of requirements.

Source code in src/zenml/integrations/integration.py

@classmethod
def get_requirements(
    cls,
    target_os: Optional[str] = None,
    python_version: Optional[str] = None,
) -> List[str]:
    """Method to get the requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.
        python_version: The Python version to use for the requirements.

    Returns:
        A list of requirements.
    """
    return cls.REQUIREMENTS

get_uninstall_requirements(target_os: Optional[str] = None) -> List[str] classmethod

Method to get the uninstall requirements for the integration.

Parameters:

Name	Type	Description	Default
target_os	Optional[str]	The target operating system to get the requirements for.	None

Returns:

Type	Description
List[str]	A list of requirements.

Source code in src/zenml/integrations/integration.py

@classmethod
def get_uninstall_requirements(
    cls, target_os: Optional[str] = None
) -> List[str]:
    """Method to get the uninstall requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.

    Returns:
        A list of requirements.
    """
    ret = []
    for each in cls.get_requirements(target_os=target_os):
        is_ignored = False
        for ignored in cls.REQUIREMENTS_IGNORED_ON_UNINSTALL:
            if each.startswith(ignored):
                is_ignored = True
                break
        if not is_ignored:
            ret.append(each)
    return ret

TypeSafeIntegration

Bases: Integration

Definition of TypeSafe AI integration for ZenML.

Modules

hooks

Initialization of the TypeSafe AI hooks module.

Classes

JevConfig

Bases: BaseModel

Settings shared by the Jev hooks.

Lifecycle hooks take no arguments, so the built-in hooks call JevConfig.load(), which finds the API key in the environment or the secret store. Custom hooks pass overrides to load() to change the rest.

Methods:

load(**overrides: Any) -> JevConfig classmethod

Build a config with the API key found in the environment or secrets.

Parameters:

Name	Type	Description	Default
**overrides	Any	Field values that replace the defaults. An explicit api_key skips the lookup.	{}

Returns:

Type	Description
JevConfig	The config.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

@classmethod
def load(cls, **overrides: Any) -> "JevConfig":
    """Build a config with the API key found in the environment or secrets.

    Args:
        **overrides: Field values that replace the defaults. An explicit
            `api_key` skips the lookup.

    Returns:
        The config.
    """
    if "api_key" not in overrides:
        overrides["api_key"] = _find_api_key()
    return cls(**overrides)

Functions:

build_failure_state(exception: BaseException, config: Optional[JevConfig] = None) -> Dict[str, Any]

Build the state the failure triage hook sends to Jev.

Custom failure hooks can reuse it to ask extra questions about the same traceback.

Parameters:

Name	Type	Description	Default
exception	BaseException	The exception that caused the failure.	required
config	Optional[JevConfig]	Hook configuration, for the traceback length. Defaults to JevConfig().	None

Returns:

Type	Description
Dict[str, Any]	The tail of the traceback and, inside a step, the step name.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def build_failure_state(
    exception: BaseException, config: Optional[JevConfig] = None
) -> Dict[str, Any]:
    """Build the state the failure triage hook sends to Jev.

    Custom failure hooks can reuse it to ask extra questions about the same
    traceback.

    Args:
        exception: The exception that caused the failure.
        config: Hook configuration, for the traceback length. Defaults to
            `JevConfig()`.

    Returns:
        The tail of the traceback and, inside a step, the step name.
    """
    max_chars = (config or JevConfig()).max_traceback_chars
    # The traceback's last line already carries the exception type and
    # message, so they are not sent separately.
    formatted = "".join(traceback.format_exception(exception))
    state: Dict[str, Any] = {"traceback": formatted[-max_chars:]}
    if step_context := StepContext.get():
        state["step_name"] = step_context.step_run.name
    return state

build_run_summary_state(exception: Optional[BaseException] = None, config: Optional[JevConfig] = None) -> Optional[Dict[str, Any]]

Build the state the run summary hook sends to Jev.

Only works as a run-level hook of a dynamic pipeline. Static pipelines copy pipeline hooks onto each step, so there is no finished run to assess.

Parameters:

Name	Type	Description	Default
exception	Optional[BaseException]	The exception that ended the run, if it failed.	None
config	Optional[JevConfig]	Hook configuration, for the baseline. Defaults to JevConfig().	None

Returns:

Type	Description
Optional[Dict[str, Any]]	The run's status, duration and steps next to a baseline of earlier
Optional[Dict[str, Any]]	runs, or None outside a run-level hook of a dynamic pipeline.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def build_run_summary_state(
    exception: Optional[BaseException] = None,
    config: Optional[JevConfig] = None,
) -> Optional[Dict[str, Any]]:
    """Build the state the run summary hook sends to Jev.

    Only works as a run-level hook of a dynamic pipeline. Static pipelines
    copy pipeline hooks onto each step, so there is no finished run to assess.

    Args:
        exception: The exception that ended the run, if it failed.
        config: Hook configuration, for the baseline. Defaults to
            `JevConfig()`.

    Returns:
        The run's status, duration and steps next to a baseline of earlier
        runs, or `None` outside a run-level hook of a dynamic pipeline.
    """
    run_context = DynamicPipelineRunContext.get()
    if run_context is None or StepContext.is_active():
        logger.warning(
            "Jev run summaries only work as a run-level hook of a dynamic "
            "pipeline. Skipping."
        )
        return None

    config = config or JevConfig()
    # The context holds the run as it was when execution started, so fetch it
    # again for the final status and end time.
    run = Client().get_pipeline_run(run_context.run.id)
    pipeline = run.pipeline
    state: Dict[str, Any] = {
        "pipeline_name": pipeline.name if pipeline else None,
        "status": run.status.value,
        "duration_seconds": _seconds(_run_duration(run)),
        "steps": [
            {
                "name": name,
                "status": step.status.value,
                "duration_seconds": _seconds(step.duration),
            }
            for name, step in run.steps.items()
        ],
        "recent_runs": _summarize_recent_runs(pipeline, run, config)
        if pipeline
        else [],
    }
    if exception is not None:
        state["exception"] = f"{type(exception).__name__}: {exception}"
    return state

jev_classify_and_log(state: JSONContent, questions: Mapping[str, Question], config: Optional[JevConfig] = None) -> Optional[SystemOneResponse]

Ask Jev questions about state and log the answers as metadata.

This is best-effort by design: a missing API key or an API error logs a warning and returns None instead of failing the step or run.

Example, as a custom success hook:

def triage_hook() -> None:
    jev_classify_and_log(
        state=get_step_context().step_run.config.parameters,
        questions={"risky": Noul(instructions="Is this config risky?")},
    )

Parameters:

Name	Type	Description	Default
state	JSONContent	Text, a JSON object or a JSON array for Jev to evaluate.	required
questions	Mapping[str, Question]	Jev questions keyed by the name used in metadata keys.	required
config	Optional[JevConfig]	Hook configuration. Defaults to JevConfig.load().	None

Returns:

Type	Description
Optional[SystemOneResponse]	The Jev response, or None if nothing was classified.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def jev_classify_and_log(
    state: JSONContent,
    questions: Mapping[str, Question],
    config: Optional[JevConfig] = None,
) -> Optional[SystemOneResponse]:
    """Ask Jev questions about `state` and log the answers as metadata.

    This is best-effort by design: a missing API key or an API error logs a
    warning and returns `None` instead of failing the step or run.

    Example, as a custom success hook:

        def triage_hook() -> None:
            jev_classify_and_log(
                state=get_step_context().step_run.config.parameters,
                questions={"risky": Noul(instructions="Is this config risky?")},
            )

    Args:
        state: Text, a JSON object or a JSON array for Jev to evaluate.
        questions: Jev questions keyed by the name used in metadata keys.
        config: Hook configuration. Defaults to `JevConfig.load()`.

    Returns:
        The Jev response, or `None` if nothing was classified.
    """
    config = config or JevConfig.load()
    if config.api_key is None:
        _warn_missing_api_key()
        return None

    try:
        with TypeSafeClient(
            api_key=config.api_key.get_secret_value()
        ) as client:
            response = client.system_one(
                state=state, questions=questions, model=config.model
            )
    except TypeSafeError as e:
        logger.warning("Jev classification failed, skipping: %s", e)
        return None

    metadata = response_to_metadata(response, prefix=config.metadata_prefix)
    if not _log_to_current_step_or_run(metadata):
        logger.warning(
            "Jev answered but there is no active step or dynamic pipeline "
            "run to attach the metadata to: %s",
            metadata,
        )
    return response

jev_failure_triage_hook(exception: BaseException) -> None

Failure hook that asks Jev why a step or dynamic run failed.

Logs jev.failure_category (infrastructure, data, code, dependency or credentials) and jev.retry_likely_to_help (the probability that an unchanged rerun succeeds) on the failed step run, or on the pipeline run when used as a run-level hook of a dynamic pipeline.

Parameters:

Name	Type	Description	Default
exception	BaseException	The exception that caused the failure.	required

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def jev_failure_triage_hook(exception: BaseException) -> None:
    """Failure hook that asks Jev why a step or dynamic run failed.

    Logs `jev.failure_category` (infrastructure, data, code, dependency or
    credentials) and `jev.retry_likely_to_help` (the probability that an
    unchanged rerun succeeds) on the failed step run, or on the pipeline run
    when used as a run-level hook of a dynamic pipeline.

    Args:
        exception: The exception that caused the failure.
    """
    config = JevConfig.load()
    jev_classify_and_log(
        state=build_failure_state(exception, config),
        questions=FAILURE_TRIAGE_QUESTIONS,
        config=config,
    )

jev_run_summary_hook(exception: Optional[BaseException] = None) -> None

Run end hook for dynamic pipelines that asks Jev whether a run is odd.

Sends Jev the finished run's status, duration and per-step results next to a summary of the pipeline's runs that finished before this one started, then logs jev.needs_attention (probability that a human should look) and jev.anomaly (0 = normal, 2 = very unusual) on the pipeline run.

Use it as @pipeline(dynamic=True, on_end=jev_run_summary_hook). Static pipelines only copy pipeline hooks onto each step, so there is no finished run to assess and the hook skips with a warning.

Parameters:

Name	Type	Description	Default
exception	Optional[BaseException]	The exception that ended the run, if it failed.	None

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def jev_run_summary_hook(exception: Optional[BaseException] = None) -> None:
    """Run end hook for dynamic pipelines that asks Jev whether a run is odd.

    Sends Jev the finished run's status, duration and per-step results next
    to a summary of the pipeline's runs that finished before this one
    started, then logs `jev.needs_attention` (probability that a human
    should look) and `jev.anomaly` (0 = normal, 2 = very unusual) on the
    pipeline run.

    Use it as `@pipeline(dynamic=True, on_end=jev_run_summary_hook)`. Static
    pipelines only copy pipeline hooks onto each step, so there is no finished
    run to assess and the hook skips with a warning.

    Args:
        exception: The exception that ended the run, if it failed.
    """
    config = JevConfig.load()
    if config.api_key is None:
        # Checked before building the state, which costs server requests.
        _warn_missing_api_key()
        return
    state = build_run_summary_state(exception, config)
    if state is not None:
        jev_classify_and_log(
            state=state, questions=RUN_SUMMARY_QUESTIONS, config=config
        )

Modules

jev_hooks

Hooks that ask TypeSafe's Jev model questions and log the answers.

Jev answers typed questions (yes/no, pick-one-label, rubric score) about a piece of text or JSON with probabilities attached, and never generates text. Each answer is flattened into scalar metadata keys such as jev.failure_category so they can be filtered on later, e.g. with Client().list_run_steps(run_metadata="jev.failure_category:data") for step hooks or Client().list_pipeline_runs(run_metadata=...) for run hooks.

Classes

JevConfig

Bases: BaseModel

Settings shared by the Jev hooks.

Lifecycle hooks take no arguments, so the built-in hooks call JevConfig.load(), which finds the API key in the environment or the secret store. Custom hooks pass overrides to load() to change the rest.

Methods:

load(**overrides: Any) -> JevConfig classmethod

Build a config with the API key found in the environment or secrets.

Parameters:

Name	Type	Description	Default
**overrides	Any	Field values that replace the defaults. An explicit api_key skips the lookup.	{}

Returns:

Type	Description
JevConfig	The config.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

@classmethod
def load(cls, **overrides: Any) -> "JevConfig":
    """Build a config with the API key found in the environment or secrets.

    Args:
        **overrides: Field values that replace the defaults. An explicit
            `api_key` skips the lookup.

    Returns:
        The config.
    """
    if "api_key" not in overrides:
        overrides["api_key"] = _find_api_key()
    return cls(**overrides)

Functions:

build_failure_state(exception: BaseException, config: Optional[JevConfig] = None) -> Dict[str, Any]

Build the state the failure triage hook sends to Jev.

Custom failure hooks can reuse it to ask extra questions about the same traceback.

Parameters:

Name	Type	Description	Default
exception	BaseException	The exception that caused the failure.	required
config	Optional[JevConfig]	Hook configuration, for the traceback length. Defaults to JevConfig().	None

Returns:

Type	Description
Dict[str, Any]	The tail of the traceback and, inside a step, the step name.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def build_failure_state(
    exception: BaseException, config: Optional[JevConfig] = None
) -> Dict[str, Any]:
    """Build the state the failure triage hook sends to Jev.

    Custom failure hooks can reuse it to ask extra questions about the same
    traceback.

    Args:
        exception: The exception that caused the failure.
        config: Hook configuration, for the traceback length. Defaults to
            `JevConfig()`.

    Returns:
        The tail of the traceback and, inside a step, the step name.
    """
    max_chars = (config or JevConfig()).max_traceback_chars
    # The traceback's last line already carries the exception type and
    # message, so they are not sent separately.
    formatted = "".join(traceback.format_exception(exception))
    state: Dict[str, Any] = {"traceback": formatted[-max_chars:]}
    if step_context := StepContext.get():
        state["step_name"] = step_context.step_run.name
    return state

build_run_summary_state(exception: Optional[BaseException] = None, config: Optional[JevConfig] = None) -> Optional[Dict[str, Any]]

Build the state the run summary hook sends to Jev.

Only works as a run-level hook of a dynamic pipeline. Static pipelines copy pipeline hooks onto each step, so there is no finished run to assess.

Parameters:

Name	Type	Description	Default
exception	Optional[BaseException]	The exception that ended the run, if it failed.	None
config	Optional[JevConfig]	Hook configuration, for the baseline. Defaults to JevConfig().	None

Returns:

Type	Description
Optional[Dict[str, Any]]	The run's status, duration and steps next to a baseline of earlier
Optional[Dict[str, Any]]	runs, or None outside a run-level hook of a dynamic pipeline.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def build_run_summary_state(
    exception: Optional[BaseException] = None,
    config: Optional[JevConfig] = None,
) -> Optional[Dict[str, Any]]:
    """Build the state the run summary hook sends to Jev.

    Only works as a run-level hook of a dynamic pipeline. Static pipelines
    copy pipeline hooks onto each step, so there is no finished run to assess.

    Args:
        exception: The exception that ended the run, if it failed.
        config: Hook configuration, for the baseline. Defaults to
            `JevConfig()`.

    Returns:
        The run's status, duration and steps next to a baseline of earlier
        runs, or `None` outside a run-level hook of a dynamic pipeline.
    """
    run_context = DynamicPipelineRunContext.get()
    if run_context is None or StepContext.is_active():
        logger.warning(
            "Jev run summaries only work as a run-level hook of a dynamic "
            "pipeline. Skipping."
        )
        return None

    config = config or JevConfig()
    # The context holds the run as it was when execution started, so fetch it
    # again for the final status and end time.
    run = Client().get_pipeline_run(run_context.run.id)
    pipeline = run.pipeline
    state: Dict[str, Any] = {
        "pipeline_name": pipeline.name if pipeline else None,
        "status": run.status.value,
        "duration_seconds": _seconds(_run_duration(run)),
        "steps": [
            {
                "name": name,
                "status": step.status.value,
                "duration_seconds": _seconds(step.duration),
            }
            for name, step in run.steps.items()
        ],
        "recent_runs": _summarize_recent_runs(pipeline, run, config)
        if pipeline
        else [],
    }
    if exception is not None:
        state["exception"] = f"{type(exception).__name__}: {exception}"
    return state

jev_classify_and_log(state: JSONContent, questions: Mapping[str, Question], config: Optional[JevConfig] = None) -> Optional[SystemOneResponse]

Ask Jev questions about state and log the answers as metadata.

This is best-effort by design: a missing API key or an API error logs a warning and returns None instead of failing the step or run.

Example, as a custom success hook:

def triage_hook() -> None:
    jev_classify_and_log(
        state=get_step_context().step_run.config.parameters,
        questions={"risky": Noul(instructions="Is this config risky?")},
    )

Parameters:

Name	Type	Description	Default
state	JSONContent	Text, a JSON object or a JSON array for Jev to evaluate.	required
questions	Mapping[str, Question]	Jev questions keyed by the name used in metadata keys.	required
config	Optional[JevConfig]	Hook configuration. Defaults to JevConfig.load().	None

Returns:

Type	Description
Optional[SystemOneResponse]	The Jev response, or None if nothing was classified.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def jev_classify_and_log(
    state: JSONContent,
    questions: Mapping[str, Question],
    config: Optional[JevConfig] = None,
) -> Optional[SystemOneResponse]:
    """Ask Jev questions about `state` and log the answers as metadata.

    This is best-effort by design: a missing API key or an API error logs a
    warning and returns `None` instead of failing the step or run.

    Example, as a custom success hook:

        def triage_hook() -> None:
            jev_classify_and_log(
                state=get_step_context().step_run.config.parameters,
                questions={"risky": Noul(instructions="Is this config risky?")},
            )

    Args:
        state: Text, a JSON object or a JSON array for Jev to evaluate.
        questions: Jev questions keyed by the name used in metadata keys.
        config: Hook configuration. Defaults to `JevConfig.load()`.

    Returns:
        The Jev response, or `None` if nothing was classified.
    """
    config = config or JevConfig.load()
    if config.api_key is None:
        _warn_missing_api_key()
        return None

    try:
        with TypeSafeClient(
            api_key=config.api_key.get_secret_value()
        ) as client:
            response = client.system_one(
                state=state, questions=questions, model=config.model
            )
    except TypeSafeError as e:
        logger.warning("Jev classification failed, skipping: %s", e)
        return None

    metadata = response_to_metadata(response, prefix=config.metadata_prefix)
    if not _log_to_current_step_or_run(metadata):
        logger.warning(
            "Jev answered but there is no active step or dynamic pipeline "
            "run to attach the metadata to: %s",
            metadata,
        )
    return response

jev_failure_triage_hook(exception: BaseException) -> None

Failure hook that asks Jev why a step or dynamic run failed.

Logs jev.failure_category (infrastructure, data, code, dependency or credentials) and jev.retry_likely_to_help (the probability that an unchanged rerun succeeds) on the failed step run, or on the pipeline run when used as a run-level hook of a dynamic pipeline.

Parameters:

Name	Type	Description	Default
exception	BaseException	The exception that caused the failure.	required

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def jev_failure_triage_hook(exception: BaseException) -> None:
    """Failure hook that asks Jev why a step or dynamic run failed.

    Logs `jev.failure_category` (infrastructure, data, code, dependency or
    credentials) and `jev.retry_likely_to_help` (the probability that an
    unchanged rerun succeeds) on the failed step run, or on the pipeline run
    when used as a run-level hook of a dynamic pipeline.

    Args:
        exception: The exception that caused the failure.
    """
    config = JevConfig.load()
    jev_classify_and_log(
        state=build_failure_state(exception, config),
        questions=FAILURE_TRIAGE_QUESTIONS,
        config=config,
    )

jev_run_summary_hook(exception: Optional[BaseException] = None) -> None

Run end hook for dynamic pipelines that asks Jev whether a run is odd.

Sends Jev the finished run's status, duration and per-step results next to a summary of the pipeline's runs that finished before this one started, then logs jev.needs_attention (probability that a human should look) and jev.anomaly (0 = normal, 2 = very unusual) on the pipeline run.

Use it as @pipeline(dynamic=True, on_end=jev_run_summary_hook). Static pipelines only copy pipeline hooks onto each step, so there is no finished run to assess and the hook skips with a warning.

Parameters:

Name	Type	Description	Default
exception	Optional[BaseException]	The exception that ended the run, if it failed.	None

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def jev_run_summary_hook(exception: Optional[BaseException] = None) -> None:
    """Run end hook for dynamic pipelines that asks Jev whether a run is odd.

    Sends Jev the finished run's status, duration and per-step results next
    to a summary of the pipeline's runs that finished before this one
    started, then logs `jev.needs_attention` (probability that a human
    should look) and `jev.anomaly` (0 = normal, 2 = very unusual) on the
    pipeline run.

    Use it as `@pipeline(dynamic=True, on_end=jev_run_summary_hook)`. Static
    pipelines only copy pipeline hooks onto each step, so there is no finished
    run to assess and the hook skips with a warning.

    Args:
        exception: The exception that ended the run, if it failed.
    """
    config = JevConfig.load()
    if config.api_key is None:
        # Checked before building the state, which costs server requests.
        _warn_missing_api_key()
        return
    state = build_run_summary_state(exception, config)
    if state is not None:
        jev_classify_and_log(
            state=state, questions=RUN_SUMMARY_QUESTIONS, config=config
        )

response_to_metadata(response: SystemOneResponse, prefix: str = DEFAULT_METADATA_PREFIX) -> Dict[str, MetadataType]

Convert a Jev response into flat ZenML metadata.

Only yes/no, choice, and score answers are supported. Other answer types are skipped with a warning.

Parameters:

Name	Type	Description	Default
response	SystemOneResponse	The Jev response.	required
prefix	str	The prefix for all metadata keys.	DEFAULT_METADATA_PREFIX

Returns:

Type	Description
Dict[str, MetadataType]	The metadata entries for supported answers plus the model that answered.

Source code in src/zenml/integrations/typesafe/hooks/jev_hooks.py

def response_to_metadata(
    response: SystemOneResponse, prefix: str = DEFAULT_METADATA_PREFIX
) -> Dict[str, MetadataType]:
    """Convert a Jev response into flat ZenML metadata.

    Only yes/no, choice, and score answers are supported. Other answer types
    are skipped with a warning.

    Args:
        response: The Jev response.
        prefix: The prefix for all metadata keys.

    Returns:
        The metadata entries for supported answers plus the model that answered.
    """
    metadata: Dict[str, MetadataType] = {f"{prefix}.model": response.model}
    for name, answer in response.answers.items():
        metadata.update(_answer_to_metadata(f"{prefix}.{name}", answer))
    return metadata