Creating Custom Actions

1
from nemoguardrails.actions import action
2
3
@action()
4
async def my_custom_action():
5
    """A simple custom action."""
6
    return "result"

1
@action(name="validate_user_input")
2
async def check_input(text: str):
3
    """Validates user input."""
4
    return len(text) > 0

1
@action(is_system_action=True)
2
async def check_policy_compliance(context: Optional[dict] = None):
3
    """Check if message complies with policy."""
4
    message = context.get("last_user_message", "")
5
    # Validation logic
6
    return True

1
@action(execute_async=True)
2
async def call_external_api(endpoint: str):
3
    """Call an external API without blocking event processing."""
4
    response = await http_client.get(endpoint)
5
    return response.json()

1
@action(output_mapping=lambda value: value)
2
async def check_hallucination(context: Optional[dict] = None):
3
    """Return True if hallucination detected (blocked), False if safe."""
4
    return detect_hallucination(context.get("bot_message", ""))

1
@action(is_system_action=True, output_mapping=lambda value: not value)
2
async def check_output_safety(context: Optional[dict] = None):
3
    """Return True if safe (allowed), mapped to not-blocked."""
4
    return is_safe(context.get("bot_message", ""))

1
def my_custom_mapping(result):
2
    if isinstance(result, dict):
3
        return result.get("score", 1.0) < 0.7
4
    return False
5
6
@action(output_mapping=my_custom_mapping)
7
async def score_safety(context: Optional[dict] = None):
8
    """Return a dict with a safety score."""
9
    return {"score": compute_score(context.get("bot_message", ""))}

1
@action()
2
async def greet_user(name: str, formal: bool = False):
3
    """Generate a greeting."""
4
    if formal:
5
        return f"Good day, {name}."
6
    return f"Hello, {name}!"

1
@action()
2
async def search_documents(
3
    query: str,
4
    max_results: int = 10,
5
    include_metadata: bool = False
6
):
7
    """Search documents with optional parameters."""
8
    results = perform_search(query, limit=max_results)
9
    if include_metadata:
10
        return {"results": results, "count": len(results)}
11
    return results

1
@action()
2
async def get_status():
3
    return "active"

1
@action()
2
async def get_user_info(user_id: str):
3
    return {
4
        "id": user_id,
5
        "name": "John Doe",
6
        "role": "admin"
7
    }

1
@action(is_system_action=True)
2
async def is_safe_content(context: Optional[dict] = None):
3
    content = context.get("bot_message", "")
4
    # Returns True if safe, False if blocked
5
    return not contains_harmful_content(content)

1
@action()
2
async def fetch_data(url: str):
3
    """Fetch data with error handling."""
4
    try:
5
        response = await http_client.get(url)
6
        response.raise_for_status()
7
        return response.json()
8
    except Exception as e:
9
        # Log the error
10
        print(f"Error fetching data: {e}")
11
        # Return a safe default or raise
12
        return None

1
from typing import Optional
2
from nemoguardrails.actions import action
3
4
@action(is_system_action=True)
5
async def check_input_length(context: Optional[dict] = None):
6
    """Ensure user input is not too long."""
7
    user_message = context.get("last_user_message", "")
8
    max_length = 1000
9
10
    if len(user_message) > max_length:
11
        return False  # Block the input
12
13
    return True  # Allow the input

1
@action(is_system_action=True)
2
async def filter_sensitive_data(context: Optional[dict] = None):
3
    """Check for sensitive data in bot response."""
4
    bot_response = context.get("bot_message", "")
5
6
    sensitive_patterns = [
7
        r"\b\d{3}-\d{2}-\d{4}\b",  # SSN pattern
8
        r"\b\d{16}\b",              # Credit card pattern
9
    ]
10
11
    import re
12
    for pattern in sensitive_patterns:
13
        if re.search(pattern, bot_response):
14
            return True  # Contains sensitive data
15
16
    return False  # No sensitive data found

1
import aiohttp
2
3
@action(execute_async=True)
4
async def query_knowledge_base(query: str, top_k: int = 5):
5
    """Query an external knowledge base API."""
6
    async with aiohttp.ClientSession() as session:
7
        async with session.post(
8
            "https://api.example.com/search",
9
            json={"query": query, "limit": top_k}
10
        ) as response:
11
            data = await response.json()
12
            return data.get("results", [])