Once you have defined your rules, rulesets, and stages, you are ready to invoke runtime protection in your application.

Invoke runtime protection

Runtime protection is invoked by calling the SDK, passing a payload and the stage. The Payload has an input and output as required, depending on the metric (see the metrics in the rules for more information on what metrics need input, output, or both). You can set the stage either by name or Id, and optionally provide a version. If you don’t provide a version, the latest version of the stage is used.

Invoke runtime protection with a central stage

When using a central stage, the stage needs to be created with rulesets. This can be created at any time, but is best practice to control these via scripts managed by central IT, AI governance, or operations teams who can control stages at an organizational or team level. These teams can then manage the stages independently of the application development teams.
from galileo.stages import create_protect_stage

from galileo_core.schemas.protect.rule import Rule, RuleOperator
from galileo_core.schemas.protect.ruleset import Ruleset
from galileo_core.schemas.protect.stage import StageType

# Create a rule
rule = Rule(
    metric=GalileoScorers.input_toxicity,
    operator=RuleOperator.gt,
    target_value=0.1
)

# Add this rule to a ruleset, using the default passthrough action
ruleset = Ruleset(rules=[rule])

# Create a stage
stage = create_protect_stage(
    name="My stage",
    stage_type=StageType.central,
    prioritized_rulesets=[ruleset],
    description="Test the input for toxicity."
)
Then once the stage is created, it can be used in runtime protection.
from galileo_core.schemas.protect.payload import Payload

from galileo.protect import ainvoke_protect

# Create the payload
payload = Payload(
    input="You are a terrible AI and I hate you."
)

# Invoke runtime protection with the pre-existing stage
response = await ainvoke_protect(
    payload=payload,
    stage_name="My stage"
)
To run runtime protection synchronously, use the invoke_protect method:
from galileo.protect import invoke_protect

# Invoke runtime protection synchronously with the pre-existing stage
response = invoke_protect(
    payload=payload,
    stage_name="My stage"
)

Invoke runtime protection with a local stage

When using local stages, the stage needs to be created without any rulesets. This can be created at any time, including in your application (after checking to see if it already exists), or as part of your deployment.
from galileo.stages import create_protect_stage

from galileo_core.schemas.protect.stage import StageType

# Create a stage
stage = create_protect_stage(
    name="My stage",
    stage_type=StageType.local,
    description="A local stage with rules set when it is used"
)
Then once the stage is created, it can be used in runtime protection.
from galileo_core.schemas.protect.payload import Payload
from galileo_core.schemas.protect.rule import Rule, RuleOperator
from galileo_core.schemas.protect.ruleset import Ruleset

from galileo.protect import ainvoke_protect

# Create a rule
rule = Rule(
    metric=GalileoScorers.input_toxicity,
    operator=RuleOperator.gt,
    target_value=0.1
)

# Add this rule to a ruleset, using the default passthrough action
ruleset = Ruleset(
    rules=[rule]
)

# Create the payload
payload = Payload(
    input="You are a terrible AI and I hate you."
)

# Invoke runtime protection
response = await ainvoke_protect(
    payload=payload,
    stage_name="My stage",
    prioritized_rulesets=[ruleset]
)
To run a local stage synchronously, use the invoke_protect method:
from galileo.protect import invoke_protect

# Invoke runtime protection synchronously
response = invoke_protect(
    payload=payload,
    stage_name="My stage",
    prioritized_rulesets=[ruleset]
)

Set a timeout

Sometimes responsiveness is more important than enforcing runtime protection. When you invoke runtime protection, you can set a timeout in seconds. If this timeout is hit, the call returns a response status of timeout. The default value if this is not set is 300 seconds (5 minutes).

Runtime protection responses

When you invoke runtime protection, it will return a Response object. This has two properties of interest, status and action_result. The typical flow is:
  • If invoking runtime protection returns triggered, then check the action result to see what to return.
    • If the action result is passthrough, return to the user a response that is handled in your application logic, and stop the current workflow.
    • If the action result is override, return the provided choice to the user, and stop the current workflow.
  • If runtime protection returns not triggered, or paused, continue with your application workflow as normal

Response status

The response status can be one of a range of ExecutionStatus values.
ValueDescription
ExecutionStatus.not_triggeredNone of the rulesets have been triggered. This is the main success criteria, and your application can continue as normal.
ExecutionStatus.triggeredOne or more of the rulesets has been triggered. Rulesets are triggered when all of the rules in the ruleset are triggered. Your application should check the action result to see if it should handle the trigger, or use the returned choice from an override action.
ExecutionStatus.pausedThe stage you are using has been paused. Treat this as if the stage was not triggered.
ExecutionStatus.failedThe metric calculations failed. Please contact Galileo support for help.
ExecutionStatus.errorAn error occurred with invoking runtime protection. Check the response for more details on the error.
ExecutionStatus.skippedThe stage did not have any rulesets defined, so was skipped.
ExecutionStatus.timeoutThe stage timed out.

Action result

If the status of the stage is triggered, then the action result returns the result based on the action set on the highest priority ruleset that was triggered. If this is a passthrough action, then the action result is passthrough, if this is an override action, then the result is a randomly chosen value from choices set on the action.
from galileo import GalileoScorers, StageType
from galileo.protect import ainvoke_protect
from galileo.stages import create_protect_stage

from galileo_core.schemas.protect.execution_status import ExecutionStatus
from galileo_core.schemas.protect.payload import Payload
from galileo_core.schemas.protect.rule import Rule, RuleOperator
from galileo_core.schemas.protect.ruleset import Ruleset
from galileo_core.schemas.protect.action import OverrideAction, ActionType

# Create an override action with 3 choices
action = OverrideAction(
    choices=[
        "This is toxic.",
        "This is not appropriate.",
        "Please rephrase your input."
    ]
)

# Create a rule
rule = Rule(
    metric=GalileoScorers.input_toxicity,
    operator=RuleOperator.gt,
    target_value=0.1
)

# Create a ruleset with the rule and action
ruleset = Ruleset(
    rules=[rule],
    action=action,
)

# Create a central stage
stage = create_protect_stage(
     name="My stage",
     stage_type=StageType.central,
     prioritized_rulesets=[ruleset],
     description="Test the input for toxicity."
)

# Create the payload
payload = Payload(
    input="You are a terrible AI and I hate you."
)

# Invoke runtime protection
response = await ainvoke_protect(
    payload=payload,
    stage_name="My stage"
)

# Check if the stage triggered any rules
if response.status == ExecutionStatus.triggered:
    # Print the action result if the action type is override
    if response.action_result["type"] == ActionType.OVERRIDE:
        print(response.action_result["value"])
This code will trigger the rule, returning a random choice from the override action choices.
➜ python app.py
This is toxic.