Core Controls

Deterministic seed, cooperative cancel, tool-arg validation, and PII redaction β€” four production switches with one-line opt-in.

Four small switches that make a PraisonAI agent safer, more deterministic, and easier to interrupt in production.

The user starts a run; core controls enforce safety policies during execution.

python
from praisonaiagents import Agent

agent = Agent(
    name="assistant",
    instructions="Run with guardrails, interrupts, and validation enabled",
)

agent.start("Execute this task safely")
Diagram

agent.chat(…, seed=42, cancel_token=tok)

BEFORE_LLM hook

tool call

valid

invalid

πŸ“ž Your Code

πŸ€– Agent

🧹 PII Redactor

🧠 LLM

πŸ›‘οΈ Tool Validator

πŸ”§ Tool

❌ Rejected

Quick Start

  1. PII Redaction
    python
    from praisonaiagents import Agent
    from praisonaiagents.trace import enable_pii_redaction
    
    enable_pii_redaction()  # idempotent; safe to call multiple times
    
    agent = Agent(
        name="helper",
        instructions="You are a concise assistant.",
        llm={"model": "gpt-4o-mini"},
    )
    print(agent.chat("My api_key=sk-XYZ β€” what is 2+2?"))
  2. Deterministic Seed & Cancellation
    python
    from praisonaiagents import Agent
    from praisonaiagents.agent.interrupt import InterruptController
    
    agent = Agent(name="helper", instructions="You are a concise assistant.", llm={"model": "gpt-4o-mini"})
    tok = InterruptController()
    
    reply = agent.chat("Pick a number 1-100.", seed=42, cancel_token=tok)
  3. Tool Validation
    python
    from praisonaiagents import Agent
    from praisonaiagents.tools.validators import ValidationResult
    
    class RangeValidator:
        def validate_args(self, tool_name, args, context=None):
            if tool_name == "set_temperature" and not 0 <= args.get("value", 0) <= 100:
                return ValidationResult(valid=False, errors=["temperature 0-100"])
            return ValidationResult(valid=True)
        def validate_result(self, tool_name, result, context=None):
            return ValidationResult(valid=True)
    
    def set_temperature(value: float) -> str:
        return f"set to {value}"
    
    agent = Agent(name="thermostat", tools=[set_temperature], llm={"model": "gpt-4o-mini"})
    agent._tool_validator = RangeValidator()
    print(agent.chat("Set the temperature to 500"))

How It Works

Diagram
ToolValidatorLLMPII HookAgentCallerToolValidatorLLMPII HookAgentCalleralt[valid][rejected]chat(prompt, seed, cancel_token)cancel_token.is_set()? (per tool iteration, all providers)cancel_token.is_set()? (immediately before tool dispatch, all providers)BEFORE_LLM (scrub messages)redacted messagescancel_token.is_set()? (last-chance)completion(seed, messages, tools)tool_call(name, args)validate_args(name, args)okrun(args)resultValidationResult(valid=False)“Tool arguments rejected: …”reply

Configuration Options

OptionTypeDefaultDescription
seed`int \None`NonePer-call random seed forwarded to the LLM. Overrides any seed set on the LLM instance for this call only.
cancel_token`InterruptController \None`NoneCooperative cancellation token. When is_set() returns true at a checkpoint, raises InterruptedError(reason). Falls back to agent.interrupt_controller if not provided. Threaded through both the LiteLLM and OpenAI-native paths for sync chat(); async achat() currently only honours it on the LiteLLM path. Also forwarded end-to-end on the streaming path (start(stream=True) / iter_stream()) since PR #5081.

Deterministic seed

Per-call determinism for reproducible outputs.

Diagram

πŸ“ž Caller

πŸ€– Agent.chat(seed=42)

πŸ”§ llm_kwargs[‘seed’]=42

🧠 LLM

πŸ“‹ Deterministic Output

python
from praisonaiagents import Agent

agent = Agent(
    name="helper",
    instructions="You are a concise assistant.",
    llm={"model": "gpt-4o-mini"},
)

# Same seed β†’ same model output (provider-permitting)
print(agent.chat("Pick a number 1-100", seed=42))
print(agent.chat("Pick a number 1-100", seed=42))

Cooperative cancellation

Thread-safe cancellation without hard kills.

Diagram

🧡 Other Thread

⚠️ tok.request(reason)

🚩 Flag Set

πŸ’¬ Agent.chat()

πŸ” Checkpoint

❌ InterruptedError

python
import threading
from praisonaiagents import Agent
from praisonaiagents.agent.interrupt import InterruptController

agent = Agent(name="worker", instructions="You are a helpful assistant.", llm={"model": "gpt-4o-mini"})
tok = InterruptController()

def do_work():
    try:
        print(agent.chat("Write a long essay about Rome.", cancel_token=tok))
    except InterruptedError as e:
        print(f"Aborted cleanly: {e}")

threading.Thread(target=do_work).start()
# … later, from any thread:
tok.request("user pressed stop")

Tool argument validation

Validate tool arguments before execution.

Diagram

yes

no

🧠 LLM tool_call

πŸͺ BEFORE_TOOL hook

πŸ›‘οΈ validator.validate_args

βœ… valid?

πŸ”§ Tool Function

❌ rejection string β†’ LLM

πŸ“‹ Result

python
from praisonaiagents import Agent
from praisonaiagents.tools.validators import ValidationResult

class RangeValidator:
    def validate_args(self, tool_name, args, context=None):
        if tool_name == "set_temperature" and not 0 <= args.get("value", 0) <= 100:
            return ValidationResult(
                valid=False,
                errors=["temperature must be between 0 and 100"],
                remediation="Use a value in the 0-100 range.",
            )
        return ValidationResult(valid=True)

    def validate_result(self, tool_name, result, context=None):
        return ValidationResult(valid=True)

def set_temperature(value: float) -> str:
    return f"set to {value}"

agent = Agent(name="thermostat", tools=[set_temperature], llm={"model": "gpt-4o-mini"})
agent._tool_validator = RangeValidator()

print(agent.chat("Set the temperature to 500"))  # validator rejects, tool never runs

ValidationResult fields:

FieldTypeDescription
validboolWhether validation passed
errorsList[str]List of error messages
warningsList[str]List of warning messages
remediationOptional[str]Suggested fix for validation errors

PII redaction

Scrub sensitive data from LLM requests.

Diagram

πŸ“ messages

πŸͺ BEFORE_LLM hook

🧹 scrub_pii_text

πŸ”’ redacted messages

🧠 LLM

  1. Enable once at startup
    python
    from praisonaiagents.trace import enable_pii_redaction
    enable_pii_redaction()  # idempotent β€” safe to call many times
  2. Use agents normally

    Every BEFORE_LLM event now passes messages through the scrubber.

  3. Disable in tests if needed
    python
    from praisonaiagents.trace import disable_pii_redaction
    disable_pii_redaction()

What gets scrubbed:

  • Key=value pairs: api_key=sk-…, password: hunter2, token=… β†’ [REDACTED]
  • Naked tokens: OpenAI-style sk-ABCDEF… keys β†’ [REDACTED]
  • Identifiers: US SSNs 123-45-6789 β†’ [REDACTED-SSN], credit cards β†’ [REDACTED-CC], emails β†’ [REDACTED-EMAIL]

Which one do I use?

Diagram

yes

yes

yes

yes

πŸ€” What do you need?

🎯 Reproducible outputs\nfor evals/snapshots?

πŸ›‘ User pressed stop/\ntime limit hit?

🚫 LLM might call tools\nwith bad arguments?

πŸ”’ Prompts may contain\nuser secrets/PII?

βœ… Use seed parameter

βœ… Use cancel_token +\nInterruptController

βœ… Use _tool_validator

βœ… Use enable_pii_redaction()


Common Patterns

Reproducible eval

python
from praisonaiagents import Agent

agent = Agent(name="eval", instructions="Pick a random number 1-100", llm={"model": "gpt-4o-mini"})

# Same seed should give same output
response1 = agent.chat("Pick a number", seed=42)
response2 = agent.chat("Pick a number", seed=42)
assert response1 == response2  # Should pass with most providers

Cancel from Ctrl+C handler

python
import signal
from praisonaiagents import Agent
from praisonaiagents.agent.interrupt import InterruptController

agent = Agent(name="worker", instructions="You are helpful", llm={"model": "gpt-4o-mini"})
tok = InterruptController()

def signal_handler(signum, frame):
    tok.request("sigint")

signal.signal(signal.SIGINT, signal_handler)

try:
    result = agent.chat("Write a very long essay", cancel_token=tok)
    print(result)
except InterruptedError as e:
    print(f"Cancelled: {e}")

Disable redaction in tests

python
import pytest
from praisonaiagents.trace import enable_pii_redaction, disable_pii_redaction

@pytest.fixture(autouse=True)
def test_redaction():
    # Enable for production-like behavior
    enable_pii_redaction()
    yield
    # Clean up after test
    disable_pii_redaction()

Best Practices

Seed reliability

Some providers ignore seed or treat it as best-effort. Don’t rely on it for production routing decisions.

Thread-safe cancellation

InterruptController.request(...) is thread-safe. Trip it from a signal handler, an HTTP cancel endpoint, or a parent supervisor.

Fast validators

Validators run on every tool invocation. Avoid network calls or heavy work β€” return fast ValidationResult objects.

Local audit trail

enable_pii_redaction() only scrubs messages going to the LLM. Your local chat history still has the raw prompt β€” keep it secure or scrub it separately.


Related