Guardrails

from polos import Agent, PolosClient

safety_agent = Agent(
    id="safety-agent",
    provider="openai",
    model="gpt-4o",
    system_prompt="You are a helpful customer service assistant.",
    tools=[search_knowledge_base, send_email],
    guardrails=[
        "Ensure the response does not contain any profanity or offensive language. "
        "The response should be professional and appropriate for all audiences.",
        
        "Do not reveal internal company information, API keys, or confidential data. "
        "Only share information from the official knowledge base."
    ]
)

client = PolosClient()
response = await safety_agent.run(client, "How do I reset my password?")

{
    "passed": False,
    "reason": "Response contains confidential internal information"
}

import re
from polos import (
    PolosClient, Agent, guardrail, GuardrailContext, GuardrailResult, WorkflowContext
)

@guardrail
def redact_sensitive_data(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    """Redact email addresses and credit card numbers from agent responses."""
    content = guardrail_context.content
    if content is None:
        return GuardrailResult.continue_with()
    
    content_str = str(content)
    modified = content_str
    
    # Redact email addresses
    email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
    modified = re.sub(email_pattern, '[EMAIL_REDACTED]', modified)
    
    # Redact credit card numbers
    cc_pattern = r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'
    modified = re.sub(cc_pattern, '[CARD_REDACTED]', modified)
    
    # SSN pattern
    ssn_pattern = r'\b\d{3}-\d{2}-\d{4}\b'
    modified = re.sub(ssn_pattern, '[SSN_REDACTED]', modified)
    
    if modified != content_str:
        return GuardrailResult.continue_with(modified_content=modified)
    
    return GuardrailResult.continue_with()

privacy_agent = Agent(
    id="privacy-agent",
    provider="anthropic",
    model="claude-sonnet-4",
    system_prompt="You are a customer support agent. Help users with their accounts.",
    tools=[get_user_info, update_account],
    guardrails=[redact_sensitive_data]
)

response = await privacy_agent.run(
    client,
    "What's my account information? My email is [email protected]"
)

# Response will have email redacted: "Your email is [EMAIL_REDACTED]"

@guardrail
def simple_check(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    # Everything looks good, continue as-is
    return GuardrailResult.continue_with()

@guardrail
def content_filter(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    content = guardrail_context.content or ""
    
    # Modify content
    filtered = content.replace("bad_word", "***")
    
    return GuardrailResult.continue_with(modified_content=filtered)

@guardrail
def compliance_check(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    content = str(guardrail_context.content or "").lower()
    
    if "confidential" in content:
        return GuardrailResult.fail(
            error_message="Response contains confidential information and was blocked"
        )
    
    return GuardrailResult.continue_with()

@guardrail
def inspector(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    # Access the generated content
    content = guardrail_context.content
    
    # Access tool calls (if any)
    tool_calls = guardrail_context.tool_calls or []
    
    # Access the full message history
    messages = guardrail_context.messages
    
    # Your validation logic here
    return GuardrailResult.continue_with()

from polos import Agent

safety_agent = Agent(
    id="safety-agent",
    provider="openai",
    model="gpt-4o",
    system_prompt="You are a helpful assistant.",
    guardrails=[
        "Ensure responses are professional and appropriate."
    ],
    guardrail_max_retries=3  # Retry up to 3 times (default: 2)
)

@guardrail
def tool_call_filter(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    """Only allow specific tools to be called."""
    tool_calls = guardrail_context.tool_calls or []
    
    if not tool_calls:
        return GuardrailResult.continue_with()
    
    # Define allowed tools
    allowed_tools = {"search_knowledge_base", "get_weather"}
    
    # Filter tool calls
    filtered_calls = []
    for tool_call in tool_calls:
        if isinstance(tool_call, dict):
            function_info = tool_call.get("function", {})
            if isinstance(function_info, dict):
                tool_name = function_info.get("name")
                if tool_name in allowed_tools:
                    filtered_calls.append(tool_call)
                else:
                    print(f"Blocked unauthorized tool: {tool_name}")
    
    # Return filtered list
    return GuardrailResult.continue_with(modified_tool_calls=filtered_calls)

restricted_agent = Agent(
    id="restricted-agent",
    provider="openai",
    model="gpt-4o",
    system_prompt="You are a helpful assistant.",
    tools=[search_knowledge_base, get_weather, send_email, delete_data],
    guardrails=[tool_call_filter]  # Will block send_email and delete_data
)

@guardrail
def content_length_limit(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    """Limit response length to 500 characters."""
    content = str(guardrail_context.content or "")
    
    if len(content) > 500:
        truncated = content[:500] + "... [truncated]"
        return GuardrailResult.continue_with(modified_content=truncated)
    
    return GuardrailResult.continue_with()

@guardrail
def no_external_links(ctx: WorkflowContext, guardrail_context: GuardrailContext) -> GuardrailResult:
    """Block responses containing external URLs."""
    import re
    content = str(guardrail_context.content or "")
    
    url_pattern = r'https?://[^\s]+'
    if re.search(url_pattern, content):
        return GuardrailResult.fail("Response contains external links, which are not allowed")
    
    return GuardrailResult.continue_with()

safe_agent = Agent(
    id="safe-agent",
    provider="openai",
    model="gpt-4o",
    system_prompt="You are a helpful assistant.",
    guardrails=[
        content_length_limit,    # Runs first
        no_external_links,       # Runs second (on potentially modified content)
        redact_sensitive_data,   # Runs third
    ]
)

compliance_agent = Agent(
    id="compliance-agent",
    provider="anthropic",
    model="claude-sonnet-4",
    system_prompt="You are a financial advisor assistant.",
    tools=[get_account_info, transfer_funds],
    guardrails=[
        redact_sensitive_data,  # Function guardrail
        
        "Ensure the response complies with financial regulations. "
        "Do not provide specific investment advice or guarantees.",  # String guardrail
        
        "Verify that any financial information shared is accurate and "
        "does not contain speculative predictions."  # String guardrail
    ]
)

@guardrail
def profanity_filter(ctx, guardrail_context):
    content = str(guardrail_context.content or "")
    if contains_profanity(content):
        return GuardrailResult.fail("Response contains inappropriate language")
    return GuardrailResult.continue_with()

@guardrail
def gdpr_compliance(ctx, guardrail_context):
    content = str(guardrail_context.content or "")
    # Redact PII: emails, phone numbers, addresses
    sanitized = redact_pii(content)
    return GuardrailResult.continue_with(modified_content=sanitized)

# String guardrail for quality
agent = Agent(
    id="qa-agent",
    guardrails=[
        "Ensure the response is factually accurate and does not contain "
        "unverified claims or speculation."
    ]
)