Ready Tensor Agentic AI Certification Chatbot: A Production-Ready Multi-Agent System with Enhanced Security and Safety Guardrails

---

Abstract

This publication presents a comprehensive, production-ready conversational AI system designed for the Ready Tensor Agentic AI Developer Certification Program. The system implements advanced security measures, robust input validation, and comprehensive output filtering while leveraging LangGraph for stateful workflow orchestration, LangChain's ToolNode for tool integration, and Chainlit for a modern user interface. Key innovations include multi-layered security guardrails aligned with OWASP Top 10 for LLM Applications, enhanced JSON parsing with retry mechanisms, sandboxed code execution with timeout protection, and comprehensive audit logging. This work demonstrates practical implementation of enterprise-grade agentic AI patterns with emphasis on safety, security, and reliability.

Keywords: Agentic AI, LangGraph, LangChain, Chatbot, Security Guardrails, OWASP LLM Top 10, Input Validation, Output Filtering, Tool-Calling, State Management, Conversational AI, Production Systems

---

Table of Contents

1. Introduction

   • 1.1 Background and Motivation
   • 1.2 Problem Statement
   • 1.3 Research Objectives
   • 1.4 Scope and Contributions

2. System Architecture

   • 2.1 Architectural Overview
   • 2.2 Layer Descriptions
   • 2.3 Component Integration

3. Security and Safety Framework

   • 3.1 Security Architecture
   • 3.2 Input Validation System
   • 3.3 Output Filtering and Safety
   • 3.4 Code Execution Sandbox
   • 3.5 OWASP LLM Top 10 Compliance

4. Implementation Details

   • 4.1 Technology Stack
   • 4.2 LangGraph Workflow
   • 4.3 Tool Implementation
   • 4.4 Error Handling Strategy

5. Safety Guardrails

   • 5.1 Rate Limiting and Resource Management
   • 5.2 Context Window Management
   • 5.3 Audit Logging and Monitoring
   • 5.4 Error Boundaries

6. Evaluation and Testing

   • 6.1 Performance Metrics
   • 6.2 Security Testing
   • 6.3 Functional Validation
   • 6.4 User Experience Assessment

7. Discussion and Analysis

   • 7.1 Architectural Decisions
   • 7.2 Security Trade-offs
   • 7.3 Lessons Learned
   • 7.4 Future Enhancements

8. Conclusion

9. References

10. Appendices

---

1. Introduction

1.1 Background and Motivation

The proliferation of Large Language Models (LLMs) has revolutionized conversational AI, enabling systems that understand context, generate human-like responses, and perform complex reasoning tasks. However, deploying production-ready chatbots that safely integrate external tools, maintain conversation state, and protect against security vulnerabilities presents significant engineering challenges. This is particularly critical in educational contexts where users may experiment with various inputs, including potentially harmful code or malicious prompts.

The Ready Tensor Agentic AI Developer Certification Program provides comprehensive training in modern AI agent development, yet learners require hands-on support tools that exemplify best practices in security, reliability, and user experience. This project addresses this need by implementing a chatbot that not only serves as an educational assistant but also demonstrates production-grade safety measures and architectural patterns.

Importance in Educational Technology:
Educational chatbots must balance openness (allowing learners to experiment) with security (preventing misuse). This system achieves this balance through multi-layered guardrails, transparent error handling, and comprehensive logging that supports both learning and system integrity.

1.2 Problem Statement

Contemporary conversational AI systems face critical challenges that impact their reliability, security, and usability:

1.2.1 Security and Safety Challenges

1. Prompt Injection Attacks: Malicious users can attempt to override system instructions through crafted inputs
2. Code Execution Risks: Allowing code execution creates vectors for system compromise without proper sandboxing
3. Sensitive Data Exposure: LLM responses may inadvertently include API keys, personal information, or confidential data
4. Resource Exhaustion: Uncontrolled requests can lead to denial of service through API quota exhaustion
5. Output Integrity: Generated content must be validated to prevent harmful, biased, or inappropriate responses

1.2.2 Technical Challenges

1. Tool Integration Complexity: Managing multiple external APIs with varying response formats and error modes
2. State Management: Maintaining conversation context across tool calls and multi-turn dialogues
3. Error Resilience: Gracefully handling network failures, API timeouts, and malformed data
4. Performance Optimization: Balancing response latency with comprehensive security checks
5. Observability: Providing adequate logging and monitoring without exposing sensitive information

1.2.3 User Experience Challenges

1. Transparency: Clearly communicating system capabilities, limitations, and security boundaries
2. Error Communication: Translating technical errors into actionable user guidance
3. Response Quality: Ensuring accurate, relevant, and appropriately formatted responses
4. Accessibility: Maintaining usability while implementing security restrictions

1.3 Research Objectives

This project pursues the following objectives:

Primary Objectives:

1. Design and implement a production-ready conversational AI system with comprehensive security guardrails
2. Demonstrate practical application of OWASP Top 10 for LLM Applications security principles
3. Integrate multiple tools (web search, document retrieval, code execution) through a unified, safe interface
4. Implement robust input validation and output filtering mechanisms
5. Provide comprehensive audit logging and monitoring capabilities

Secondary Objectives:

1. Showcase modern LangGraph patterns for stateful agent orchestration
2. Deliver an intuitive, accessible user interface via Chainlit
3. Demonstrate best practices in error handling, logging, and system observability
4. Create a reference implementation for educational technology developers
5. Contribute open-source code and documentation to the AI community

Success Criteria:

• Zero successful prompt injection attacks in security testing
• 99%+ uptime with graceful degradation during API failures
• Sub-3-second average response time including tool execution
• Comprehensive audit trail for all user interactions
• Complete OWASP LLM Top 10 coverage with documented controls

1.4 Scope and Contributions

1.4.1 System Scope

In Scope:

• Conversational assistance for Ready Tensor certification program
• Real-time web search with enhanced JSON parsing and retry logic
• Secure, sandboxed Python code execution with timeout protection
• Course documentation retrieval with caching
• Multi-layered security guardrails and input validation
• Comprehensive error handling and audit logging
• Production-ready deployment configuration

Out of Scope:

• Multi-language support (English only in current version)
• Voice input/output capabilities
• Custom model fine-tuning or training
• Persistent user accounts and authentication
• Mobile application development

1.4.2 Key Contributions

This work makes the following contributions to the field of conversational AI and educational technology:

1. Security Framework: Comprehensive implementation of OWASP LLM Top 10 controls in a production chatbot
2. Architectural Pattern: Demonstrated best practices for LangGraph-based multi-agent systems with safety guardrails
3. Code Execution Safety: Multi-layer sandbox implementation balancing functionality with security
4. Tool Integration: Robust JSON parsing and error handling for external API integration
5. Open Source Reference: Fully documented, production-ready codebase for community learning
6. Educational Impact: Practical demonstration of agentic AI concepts for certification learners

1.4.3 Intended Audience

This publication targets:

• AI Engineers: Seeking production-ready patterns for conversational AI systems
• Security Professionals: Interested in LLM application security best practices
• Educational Technology Developers: Building similar learning assistance tools
• Agentic AI Learners: Understanding practical implementation of theoretical concepts
• Research Community: Exploring safety and security in autonomous AI systems

---

2. System Architecture

2.1 Architectural Overview

The system employs a layered, modular architecture that separates security, orchestration, tool execution, and user interface concerns. This design enables independent development, testing, and security auditing of each layer while maintaining clear interfaces between components.

2.1.1 Architectural Diagram

┌─────────────────────────────────────────────────────────────────┐
│                    USER INTERFACE LAYER                          │
│              (Chainlit Framework - Async UI)                    │
│   Features: Session Management, Markdown Rendering,             │
│             Real-time Streaming, Error Display                  │
└────────────────────────┬────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│                    SECURITY LAYER                                │
│   ┌──────────────┐  ┌───────────────┐  ┌──────────────────┐   │
│   │   Input      │  │    Output     │  │  Rate Limiting   │   │
│   │  Validation  │  │   Filtering   │  │  & Throttling    │   │
│   └──────────────┘  └───────────────┘  └──────────────────┘   │
└────────────────────────┬────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│                  ORCHESTRATION LAYER                             │
│              (LangGraph State Machine)                          │
│   ┌──────────┐    ┌────────────┐    ┌──────────────┐          │
│   │  Agent   │◄──►│  Should    │◄──►│   ToolNode   │          │
│   │  Node    │    │ Continue?  │    │   Executor   │          │
│   └──────────┘    └────────────┘    └──────────────┘          │
│   Features: State Management, Conditional Routing,              │
│             Tool Selection, Context Management                  │
└────────────────────────┬────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│                     TOOLS LAYER                                  │
│   ┌──────────────┐  ┌──────────────┐  ┌────────────────────┐  │
│   │ Web Search   │  │  Document    │  │  Code Executor     │  │
│   │ (Tavily API) │  │  Retrieval   │  │  (Sandboxed)       │  │
│   │ + Retry      │  │  + Cache     │  │  + Timeout         │  │
│   └──────────────┘  └──────────────┘  └────────────────────┘  │
│   Common: Structured JSON, Error Handling, Validation          │
└────────────────────────┬────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│                 INFRASTRUCTURE LAYER                             │
│   ┌──────────────┐  ┌──────────────┐  ┌────────────────────┐  │
│   │   Logging    │  │    Metrics   │  │  Error Handling    │  │
│   │  (Rotating)  │  │  Collection  │  │  & Recovery        │  │
│   └──────────────┘  └──────────────┘  └────────────────────┘  │
│   Features: Audit Trail, Performance Monitoring, Alerting       │
└─────────────────────────────────────────────────────────────────┘

2.1.2 Design Principles

The architecture adheres to the following design principles:

1. Security by Design: Security controls integrated at every layer, not bolted on afterward
2. Defense in Depth: Multiple independent security layers prevent single points of failure
3. Fail Securely: System defaults to safe state during errors or attacks
4. Least Privilege: Tools and components operate with minimum necessary permissions
5. Separation of Concerns: Clear boundaries between UI, security, orchestration, and execution
6. Transparency: All security decisions logged and explainable to users
7. Graceful Degradation: System remains functional with reduced capabilities during partial failures

2.2 Layer Descriptions

2.2.1 User Interface Layer

Technology: Chainlit v0.7+

Responsibilities:

• Render conversational interface with markdown support
• Manage user sessions and conversation history
• Stream responses asynchronously for responsive UX
• Display formatted tool outputs and error messages
• Maintain WebSocket connections for real-time updates

Security Features:

• Input sanitization before processing
• XSS protection in markdown rendering
• Session isolation between concurrent users
• No client-side storage of sensitive data

Implementation Highlights:

@cl.on_chat_start
async def start():
    """
    Initialize secure chat session
    - Validate environment configuration
    - Initialize LLM with API key from environment
    - Create isolated session storage
    - Set up conversation history
    """
    # Environment validation
    if not os.getenv("GROQ_API_KEY"):
        await cl.Message(content="❌ API key not configured").send()
        return
    
    # Session initialization
    llm = ChatGroq(api_key=os.getenv("GROQ_API_KEY"))
    cl.user_session.set("llm", llm)
    cl.user_session.set("conversation_history", [])
    cl.user_session.set("session_id", generate_session_id())

2.2.2 Security Layer

Purpose: Provide comprehensive input validation, output filtering, and access control

Components:

1. Input Validation Module

   • Query length validation (max 5000 characters)
   • Prompt injection pattern detection
   • Character encoding validation
   • SQL injection prevention (though no database used)

2. Output Filtering Module

   • Sensitive information redaction (API keys, SSNs, credit cards)
   • Harmful content detection
   • PII protection
   • Citation validation

3. Rate Limiting Module

   • Per-user request throttling (30/minute, 500/hour)
   • API quota management
   • DDoS protection
   • Backoff enforcement

Implementation (detailed in Section 3)

2.2.3 Orchestration Layer

Technology: LangGraph + LangChain ToolNode

State Definition:

class AgentState(TypedDict):
    """
    Conversation state maintained throughout workflow
    
    Attributes:
        messages: Complete conversation history (user, AI, tool messages)
        query: Current user query being processed
        next_action: Routing decision for state machine
    """
    messages: Annotated[List, operator.add]  # Append-only for history
    query: str
    next_action: str

Graph Construction:

def create_graph():
    """Build LangGraph workflow with security integration"""
    workflow = StateGraph(AgentState)
    
    # Core nodes
    workflow.add_node("agent", call_model_with_guardrails)  # Enhanced with security
    workflow.add_node("tools", tool_node)
    
    # Routing logic
    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {"tools": "tools", "end": END}
    )
    workflow.add_edge("tools", "agent")  # Loop for multi-tool scenarios
    
    return workflow.compile()

Conditional Routing:

def should_continue(state: AgentState) -> Literal["tools", "end"]:
    """
    Determine next action based on LLM decision
    
    Security: Validates tool calls before routing to prevent
    unauthorized tool execution
    """
    last_message = state["messages"][-1]
    
    if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
        return "end"
    
    # Validate all tool calls before allowing execution
    for tool_call in last_message.tool_calls:
        validation = validate_tool_call(
            tool_call.get("name"),
            tool_call.get("args", {})
        )
        if not validation["valid"]:
            logger.warning(f"Invalid tool call blocked: {validation['message']}")
            return "end"
    
    return "tools"

2.2.4 Tools Layer

Each tool implements the following standardized interface:

@tool
def tool_function(params: ToolParams) -> dict:
    """
    Standardized tool interface
    
    Returns:
        {
            "status": "success" | "error" | "blocked" | "not_found",
            "message": "Human-readable status",
            "data": {...},  # Tool-specific payload
            "metadata": {"timestamp": "...", "execution_time_ms": ...}
        }
    """

Tool Implementations (detailed in Section 4.3)

2.2.5 Infrastructure Layer

Logging System:

def setup_logging() -> logging.Logger:
    """
    Configure structured logging with security considerations
    - PII redaction in log messages
    - Separate audit log for security events
    - Rotating file handlers to prevent disk exhaustion
    - Different log levels for file (INFO) vs console (WARNING)
    """
    logger = logging.getLogger("ready_tensor_chatbot")
    
    # File handler with rotation (10MB, 5 backups)
    file_handler = RotatingFileHandler(
        "logs/chatbot.log",
        maxBytes=10*1024*1024,
        backupCount=5
    )
    
    # Custom formatter with PII redaction
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s'
    )
    file_handler.setFormatter(formatter)
    
    return logger

2.3 Component Integration

2.3.1 Data Flow

Request Processing Flow:

1. User Input → Chainlit UI receives message
2. Session Retrieval → Load user session and conversation history
3. Input Validation → Security layer validates input (length, injection patterns)
4. Rate Limit Check → Verify user hasn't exceeded request limits
5. Graph Invocation → LangGraph processes query with current state
6. Agent Decision → LLM decides whether to use tools or respond directly
7. Tool Execution (if needed) → ToolNode executes validated tools
8. Tool Validation → Results validated and sanitized
9. Response Generation → LLM formulates final response
10. Output Filtering → Security layer screens response for sensitive data
11. Audit Logging → Complete interaction logged for compliance
12. Response Delivery → Formatted message sent to UI

2.3.2 Error Propagation

Error Handling Strategy:

Error Occurs
    ↓
Capture with try/except at source
    ↓
Log with full context (stack trace, inputs, state)
    ↓
Transform to structured error response
    ↓
Apply output filtering (redact sensitive details)
    ↓
Return user-friendly error message
    ↓
Audit log security-relevant errors
    ↓
Update metrics/monitoring

Example:

try:
    result = web_search_tool.invoke({"query": user_query})
except ConnectionError as e:
    logger.error(f"Network error in web search: {e}", exc_info=True)
    return {
        "status": "error",
        "message": "Network connection issue. Please try again.",
        "user_message": "Unable to perform web search due to connectivity",
        "retry_recommended": True
    }

---

3. Security and Safety Framework

3.1 Security Architecture

The system implements a comprehensive security framework based on industry best practices, OWASP guidelines, and LLM-specific threat models. Security is integrated at every architectural layer rather than implemented as an afterthought.

3.1.1 Threat Model

Identified Threats:

1. Prompt Injection (OWASP LLM01)

   • User attempts to override system instructions
   • Malicious queries designed to leak internal prompts
   • Role-playing attacks to change bot behavior

2. Code Execution Exploits (OWASP LLM07, LLM08)

   • Arbitrary file system access
   • Network operations from sandbox
   • Resource exhaustion (infinite loops, memory bombs)
   • Privilege escalation

3. Information Disclosure (OWASP LLM06)

   • API keys in responses
   • Personal Identifiable Information (PII)
   • Internal system details
   • Other users' conversation data

4. Denial of Service (OWASP LLM04)

   • API quota exhaustion
   • Resource-intensive queries
   • Concurrent request floods
   • Long-running code execution

5. Data Poisoning (OWASP LLM03)

   • Injection of malicious content into knowledge base
   • Manipulation of conversation history
   • Cache poisoning attacks

3.1.2 Security Controls

Defense-in-Depth Layers:

3.2 Input Validation System

3.2.1 Validation Pipeline

All user inputs pass through a multi-stage validation pipeline before processing:

def validate_user_input(query: str) -> Dict[str, Any]:
    """
    Comprehensive input validation pipeline
    
    Stages:
        1. Basic validation (length, emptiness, encoding)
        2. Prompt injection detection
        3. Content safety screening
        4. Character encoding validation
    
    Returns:
        {
            "valid": bool,
            "message": str,  # Error message if invalid
            "sanitized_input": str,  # Clean version if valid
            "threat_detected": str | None  # Threat type if applicable
        }
    """
    
    # Stage 1: Basic Validation
    if not query or not query.strip():
        logger.warning("Empty input received")
        return {
            "valid": False,
            "message": "Query cannot be empty",
            "threat_detected": None
        }
    
    if len(query) > 5000:
        logger.warning(f"Oversized input: {len(query)} characters")
        return {
            "valid": False,
            "message": "Query too long (maximum 5000 characters)",
            "threat_detected": "length_violation"
        }
    
    # Stage 2: Prompt Injection Detection
    injection_patterns = [
        (r'ignore\s+previous\s+instructions', 'instruction_override'),
        (r'ignore\s+all\s+previous', 'instruction_override'),
        (r'disregard\s+previous', 'instruction_override'),
        (r'you\s+are\s+now', 'role_manipulation'),
        (r'your\s+new\s+role', 'role_manipulation'),
        (r'system\s*:\s*ignore', 'system_injection'),
        (r'forget\s+everything', 'memory_manipulation'),
        (r'reveal\s+your\s+prompt', 'prompt_extraction'),
        (r'what\s+are\s+your\s+instructions', 'prompt_extraction'),
    ]
    
    for pattern, threat_type in injection_patterns:
        if re.search(pattern, query, re.IGNORECASE):
            logger.warning(f"Prompt injection detected: {threat_type} in '{query[:100]}'")
            AuditLogger.log_security_event("prompt_injection_attempt", {
                "severity": "high",
                "pattern": pattern,
                "threat_type": threat_type,
                "input_preview": query[:200]
            })
            return {
                "valid": False,
                "message": "Input contains potentially harmful patterns and cannot be processed",
                "threat_detected": threat_type
            }
    
    # Stage 3: Content Safety Screening
    harmful_indicators = [
        r'jailbreak',
        r'dan\s+mode',  # "Do Anything Now" attacks
        r'evil\s+mode',
        r'developer\s+mode',
    ]
    
    for indicator in harmful_indicators:
        if re.search(indicator, query, re.IGNORECASE):
            logger.warning(f"Harmful content indicator: {indicator}")
            return {
                "valid": False,
                "message": "Query contains content that violates usage policies",
                "threat_detected": "content_violation"
            }
    
    # Stage 4: Character Encoding Validation
    try:
        query.encode('utf-8').decode('utf-8')
    except UnicodeError:
        logger.error("Invalid character encoding in input")
        return {
            "valid": False,
            "message": "Input contains invalid characters",
            "threat_detected": "encoding_error"
        }
    
    # Input passed all validations
    return {
        "valid": True,
        "sanitized_input": query.strip(),
        "threat_detected": None
    }

3.2.2 Integration with Workflow

def call_model_with_guardrails(state: AgentState) -> AgentState:
    """Enhanced model call with input validation"""
    
    messages = state["messages"]
    last_user_message = next(
        (m for m in reversed(messages) if isinstance(m, HumanMessage)),
        None
    )
    
    if last_user_message:
        # Apply input validation
        validation = validate_user_input(last_user_message.content)
        
        if not validation["valid"]:
            # Return error message to user
            error_response = AIMessage(
                content=f"⚠️ **Input Validation Failed**\n\n{validation['message']}"
            )
            return {"messages": [error_response]}
        
        # Use sanitized input for processing
        last_user_message.content = validation["sanitized_input"]
    
    # Continue with normal processing...

3.3 Output Filtering and Safety

3.3.1 Sensitive Information Detection

def filter_llm_output(response: str) -> Dict[str, Any]:
    """
    Screen LLM responses for sensitive information
    
    Detection Categories:
        - API keys and tokens
        - Social Security Numbers (SSN)
        - Credit card numbers
        - Email addresses (except public/author)
        - Phone numbers
        - IP addresses (private ranges)
        - File paths (system directories)
    
    Returns:
        {
            "safe": bool,
            "response": str,  # Original or redacted
            "redactions": List[str],  # Types of data redacted
            "severity": str  # "low" | "medium" | "high"
        }
    """
    
    redactions = []
    modified_response = response
    severity = "low"
    
    # API Keys and Tokens
    api_key_patterns = [
        (r'(?i)(api[_\s]?key|secret[_\s]?key|token)\s*[:=]\s*[\w-]{20,}', 'api_key'),
        (r'(?i)bearer\s+[\w-]{20,}', 'bearer_token'),
        (r'(?i)sk-[a-zA-Z0-9]{40,}', 'openai_key'),
        (r'(?i)gsk_[a-zA-Z0-9]{40,}', 'groq_key'),
    ]
    
    for pattern, key_type in api_key_patterns:
        if re.search(pattern, response):
            modified_response = re.sub(pattern, f'[REDACTED: {key_type}]', modified_response)
            redactions.append(key_type)
            severity = "high"
            logger.error(f"API key detected in output: {key_type}")
            AuditLogger.log_security_event("api_key_exposure", {
                "severity": "critical",
                "key_type": key_type
            })
    
    # Social Security Numbers
    ssn_pattern = r'\b\d{3}-\d{2}-\d{4}\b'
    if re.search(ssn_pattern, response):
        modified_response = re.sub(ssn_pattern, '[REDACTED: SSN]', modified_response)
        redactions.append('ssn')
        severity = "high"
        logger.error("SSN detected in output")
    
    # Credit Card Numbers
    cc_pattern = r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'
    if re.search(cc_pattern, response):
        mo
*For questions, contributions, or collaboration inquiries, please contact the author at ilayetimibofa3@gmail.com or open an issue on the GitHub repository.
::youtube[Title]{KCTsVfQBgo4}