Registering Custom Actions

1
# config/actions.py
2
from nemoguardrails.actions import action
3
4
@action()
5
async def my_action():
6
    return "result"
7
8
@action(name="custom_name")
9
async def another_action():
10
    return "another result"

1
# config/actions/__init__.py
2
from .validation import check_input, check_output
3
from .external import fetch_data, call_api

1
# config/actions/validation.py
2
from nemoguardrails.actions import action
3
4
@action()
5
async def check_input(text: str):
6
    return len(text) > 0
7
8
@action()
9
async def check_output(text: str):
10
    return "error" not in text.lower()

1
from nemoguardrails import LLMRails, RailsConfig
2
3
config = RailsConfig.from_path("config")
4
rails = LLMRails(config)
5
6
# Register a function as an action
7
async def my_dynamic_action(param: str):
8
    return f"Processed: {param}"
9
10
rails.register_action(my_dynamic_action, name="dynamic_action")

1
def setup_rails(environment: str):
2
    config = RailsConfig.from_path("config")
3
    rails = LLMRails(config)
4
5
    if environment == "production":
6
        rails.register_action(production_validator, "validate")
7
    else:
8
        rails.register_action(dev_validator, "validate")
9
10
    return rails

1
class DatabaseService:
2
    async def query(self, sql: str):
3
        # Database query logic
4
        pass
5
6
db = DatabaseService()
7
8
async def db_query_action(query: str):
9
    return await db.query(query)
10
11
rails.register_action(db_query_action, name="query_database")

1
from langchain_core.tools import tool
2
from nemoguardrails import LLMRails, RailsConfig
3
4
@tool
5
def get_weather(city: str) -> str:
6
    """Get weather for a city."""
7
    return f"Weather in {city}: Sunny, 72°F"
8
9
config = RailsConfig.from_path("config")
10
rails = LLMRails(config)
11
12
# Register the tool as an action
13
rails.register_action(get_weather, name="get_weather")

1
from langchain_core.tools import tool
2
3
@tool
4
def search_web(query: str) -> str:
5
    """Search the web."""
6
    return f"Results for: {query}"
7
8
@tool
9
def get_current_time(timezone: str) -> str:
10
    """Get the current time."""
11
    return f"Current time in {timezone}: 12:00 PM"
12
13
# Register multiple tools
14
tools = [search_web, get_current_time]
15
for t in tools:
16
    rails.register_action(t, name=t.name)

1
from langchain_core.runnables import RunnableLambda
2
from nemoguardrails import LLMRails, RailsConfig
3
4
# Create a runnable
5
process_text = RunnableLambda(lambda x: x.upper())
6
7
config = RailsConfig.from_path("config")
8
rails = LLMRails(config)
9
10
# Register the runnable
11
rails.register_action(process_text, name="process_text")

1
# config/config.py
2
from nemoguardrails import LLMRails
3
4
def init(app: LLMRails):
5
    """Custom initialization function."""
6
7
    # Register actions
8
    async def custom_action(param: str):
9
        return f"Custom: {param}"
10
11
    app.register_action(custom_action, name="custom_action")
12
13
    # Register action parameters
14
    db_connection = create_db_connection()
15
    app.register_action_param("db", db_connection)

1
# config/config.py
2
def init(app: LLMRails):
3
    # Create shared resources
4
    http_client = aiohttp.ClientSession()
5
    cache = RedisCache()
6
7
    # Register as action parameters
8
    app.register_action_param("http_client", http_client)
9
    app.register_action_param("cache", cache)

1
# config/actions.py
2
from nemoguardrails.actions import action
3
4
@action()
5
async def fetch_with_cache(
6
    url: str,
7
    http_client=None,  # Injected automatically
8
    cache=None         # Injected automatically
9
):
10
    # Check cache first
11
    cached = await cache.get(url)
12
    if cached:
13
        return cached
14
15
    # Fetch and cache
16
    response = await http_client.get(url)
17
    data = await response.json()
18
    await cache.set(url, data)
19
20
    return data

1
# Good
2
@action(name="validate_user_age")
3
async def validate_age(age: int):
4
    pass
5
6
# Avoid
7
@action(name="v_a")
8
async def validate_age(age: int):
9
    pass

1
@action()
2
async def search_knowledge_base(
3
    query: str,
4
    top_k: int = 5
5
) -> list:
6
    """
7
    Search the knowledge base for relevant documents.
8
9
    Args:
10
        query: The search query string
11
        top_k: Maximum number of results to return
12
13
    Returns:
14
        List of relevant document snippets
15
    """
16
    pass