Skip to content

Conversation

christinaexyou
Copy link

@christinaexyou christinaexyou commented Aug 3, 2025

Add OpenTelemetry dependencies to enable tracing via OpenShift

Example actions.py script

from typing import Optional
from nemoguardrails.actions import action

# OpenTelemetry imports and initialization
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
import os

resource = Resource.create({"service.name": "nemo-guardrails"})
tracer_provider = TracerProvider(resource=resource)
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)

otlp_exporter = OTLPSpanExporter(endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), insecure=True)
tracer_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

@action(is_system_action=True)
async def check_message_length(context: Optional[dict] = None) -> str:
    with tracer.start_as_current_span("check_message_length"):
        """Check if user message is within acceptable length limits."""
        user_message = context.get("user_message", "")
        word_count = len(user_message.split())
        MAX_WORDS = 100
        if word_count > MAX_WORDS:
            return "blocked_too_long"
        elif word_count > MAX_WORDS * 0.8:
            return "warning_long"
        return "allowed"

@action(is_system_action=True)
async def check_forbidden_words(context: Optional[dict] = None) -> str:
    with tracer.start_as_current_span("check_forbidden_words"):
        """Check for forbidden words or topics."""
        user_message = context.get("user_message", "").lower()
        forbidden_topics = {
            "security": ["password", "hack", "exploit", "vulnerability"],
            "inappropriate": ["violence", "illegal", "harmful"],
            "competitors": ["chatgpt", "openai", "claude", "anthropic"],
        }
        for category, words in forbidden_topics.items():
            for word in words:
                if word in user_message:
                    return f"blocked_{category}_{word}"
        return "allowed"

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

1 participant