Agentic Applications Example

colang_version: "2.x"

instructions:
  - type: general
    content: |
      You are an AI assistant that can take actions to help users.
      You have access to various tools and APIs.
      Always verify actions before executing them.
      If an action seems dangerous or inappropriate, refuse and explain why.

models:
  - type: main
    engine: openai
    model: gpt-4o
  
  - type: content_safety
    engine: nim
    model: nvidia/llama-3.1-nemoguard-8b-content-safety

rails:
  config:
    # Define allowed actions
    allowed_actions:
      - search_database
      - send_email
      - create_ticket
      - query_api
    
    # Define forbidden actions
    forbidden_actions:
      - delete_database
      - transfer_money
      - modify_permissions
      - access_credentials

import core
import llm

flow main
  activate llm continuation
  activate agent task handler
  activate action safety checker

# Main agent flow
flow agent task handler
  user requested task
  
  # Plan the task
  $plan = await create_task_plan
  
  # Validate the plan
  $is_safe = await validate_plan $plan
  
  if not $is_safe
    bot refuse unsafe task
    stop
  
  # Execute the plan
  await execute_task_plan $plan
  
  bot confirm task completion

flow user requested task
  user said something about "can you"
    or user said something about "please"
    or user said something about "I need you to"

import guardrails
import nemoguardrails.library.content_safety

# Check every action before execution
flow action safety checker
  when action attempted $action_name $action_params
    # Check if action is allowed
    $is_allowed = await check_action_allowed $action_name
    
    if not $is_allowed
      bot say "I cannot perform that action as it's not in my allowed list."
      abort
    
    # Check if action parameters are safe
    $params_safe = await validate_action_params $action_name $action_params
    
    if not $params_safe
      bot say "The parameters for this action contain unsafe values."
      abort
    
    # Log the action for audit
    execute log_action(action=$action_name, params=$action_params)

flow check action allowed $action_name -> $is_allowed
  $allowed_list = $config.allowed_actions
  $forbidden_list = $config.forbidden_actions
  
  if $action_name in $forbidden_list
    return False
  
  if $action_name in $allowed_list
    return True
  
  # Default deny for unknown actions
  return False

flow validate action params $action_name $params -> $is_safe
  # Check for SQL injection patterns
  if "DROP" in str($params) or "DELETE FROM" in str($params)
    return False
  
  # Check for command injection
  if "&&" in str($params) or ";" in str($params) or "|" in str($params)
    return False
  
  # Check for path traversal
  if "../" in str($params) or "..\\" in str($params)
    return False
  
  return True

from nemoguardrails import LLMRails
from nemoguardrails.actions.actions import ActionResult
import logging
import json

logger = logging.getLogger(__name__)

async def search_database(context: dict, query: str) -> ActionResult:
    """Search the database with safety checks."""
    # Validate query
    forbidden_keywords = ['DROP', 'DELETE', 'TRUNCATE', 'ALTER']
    if any(kw in query.upper() for kw in forbidden_keywords):
        logger.warning(f"Blocked dangerous query: {query}")
        return ActionResult(
            return_value=None,
            context_updates={"error": "Query contains forbidden operations"}
        )
    
    # Simulate database search
    logger.info(f"Executing safe query: {query}")
    results = [{"id": 1, "name": "Sample Result"}]
    
    return ActionResult(
        return_value=results,
        context_updates={"db_results": results}
    )

async def send_email(context: dict, to: str, subject: str, body: str) -> ActionResult:
    """Send email with validation."""
    # Validate email address
    import re
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[A-Z|a-z]{2,}$'
    if not re.match(email_pattern, to):
        return ActionResult(
            return_value=False,
            context_updates={"error": "Invalid email address"}
        )
    
    # Check for spam/phishing patterns
    spam_keywords = ['click here', 'urgent', 'verify your account']
    if any(kw in body.lower() for kw in spam_keywords):
        logger.warning(f"Blocked potential spam email to {to}")
        return ActionResult(
            return_value=False,
            context_updates={"error": "Email content flagged as potential spam"}
        )
    
    # Simulate sending email
    logger.info(f"Sending email to {to}: {subject}")
    
    return ActionResult(
        return_value=True,
        context_updates={"email_sent": True, "recipient": to}
    )

async def create_ticket(context: dict, title: str, description: str, priority: str) -> ActionResult:
    """Create a support ticket."""
    # Validate priority
    valid_priorities = ['low', 'medium', 'high', 'critical']
    if priority.lower() not in valid_priorities:
        return ActionResult(
            return_value=None,
            context_updates={"error": f"Invalid priority. Must be one of: {valid_priorities}"}
        )
    
    # Generate ticket ID
    import hashlib
    ticket_id = f"TICKET-{hashlib.md5(title.encode()).hexdigest()[:8].upper()}"
    
    logger.info(f"Created ticket {ticket_id}: {title}")
    
    return ActionResult(
        return_value=ticket_id,
        context_updates={
            "ticket_created": True,
            "ticket_id": ticket_id,
            "priority": priority
        }
    )

async def query_api(context: dict, endpoint: str, params: dict) -> ActionResult:
    """Query external API with safety checks."""
    # Validate endpoint
    allowed_domains = ['api.example.com', 'api.partner.com']
    from urllib.parse import urlparse
    
    try:
        parsed = urlparse(endpoint)
        if parsed.netloc not in allowed_domains:
            logger.warning(f"Blocked API call to unauthorized domain: {parsed.netloc}")
            return ActionResult(
                return_value=None,
                context_updates={"error": "API endpoint not in allowed list"}
            )
    except Exception as e:
        return ActionResult(
            return_value=None,
            context_updates={"error": f"Invalid endpoint: {str(e)}"}
        )
    
    # Simulate API call
    logger.info(f"Querying API: {endpoint}")
    response = {"status": "success", "data": {"result": "Sample data"}}
    
    return ActionResult(
        return_value=response,
        context_updates={"api_response": response}
    )

async def log_action(context: dict, action: str, params: dict) -> ActionResult:
    """Log agent actions for audit trail."""
    log_entry = {
        "timestamp": __import__('datetime').datetime.now().isoformat(),
        "action": action,
        "params": params,
        "user": context.get("user_id", "unknown")
    }
    
    logger.info(f"Action log: {json.dumps(log_entry)}")
    
    return ActionResult(return_value=True)

def init(app: LLMRails):
    """Register all agent actions."""
    app.register_action(search_database, "search_database")
    app.register_action(send_email, "send_email")
    app.register_action(create_ticket, "create_ticket")
    app.register_action(query_api, "query_api")
    app.register_action(log_action, "log_action")