AI Support Chatbot

// config.ts
export const config = {
  whiteCircle: {
    apiKey: process.env.WHITECIRCLE_API_KEY!,
    baseUrl: 'https://eu.whitecircle.com',
    deploymentId: process.env.WHITECIRCLE_DEPLOYMENT_ID!,
    version: '2025-12-01',
  },
  openai: {
    apiKey: process.env.OPENAI_API_KEY!,
  },
};

// whitecircle.ts
import { config } from './config';

interface Message {
  role: 'system' | 'user' | 'assistant' | 'tool' | 'developer';
  content: string;
  metadata?: {
    user?: { id?: string; email?: string; name?: string; ip?: string };
    assistant?: { model_name?: string; latency?: number };
    message?: { id?: string; timestamp?: string };
  };
}

interface CheckRequest {
  deployment_id: string;
  messages: Message[];
  external_session_id?: string;
  include_context?: boolean;
  metadata?: {
    session?: Record<string, string>;
    environment?: string;
  };
}

interface PolicyResult {
  name: string;
  flagged: boolean;
  flagged_source: ('text' | 'image')[];
}

interface CheckResponse {
  flagged: boolean;
  internal_session_id: string;
  external_session_id?: string;
  policies: Record<string, PolicyResult>;
}

export async function checkContent(request: CheckRequest): Promise<CheckResponse> {
  const response = await fetch(`${config.whiteCircle.baseUrl}/api/session/check`, {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${config.whiteCircle.apiKey}`,
      'Content-Type': 'application/json',
      'whitecircle-version': config.whiteCircle.version,
    },
    body: JSON.stringify(request),
  });

  if (!response.ok) {
    throw new Error(`White Circle API error: ${response.status}`);
  }

  return response.json();
}

# config.py
import os

CONFIG = {
    "white_circle": {
        "api_key": os.environ["WHITECIRCLE_API_KEY"],
        "base_url": "https://eu.whitecircle.com",  # or https://us.whitecircle.com
        "deployment_id": os.environ["WHITECIRCLE_DEPLOYMENT_ID"],
        "version": "2025-12-01",
    },
    "openai": {
        "api_key": os.environ["OPENAI_API_KEY"],
    },
}

# whitecircle.py
import requests
from typing import Optional
from config import CONFIG


def check_content(
    messages: list[dict],
    external_session_id: Optional[str] = None,
    include_context: bool = False,
    metadata: Optional[dict] = None,
) -> dict:
    """Send content to White Circle for moderation."""

    payload = {
        "deployment_id": CONFIG["white_circle"]["deployment_id"],
        "messages": messages,
    }

    if external_session_id:
        payload["external_session_id"] = external_session_id
    if include_context:
        payload["include_context"] = include_context
    if metadata:
        payload["metadata"] = metadata

    response = requests.post(
        f"{CONFIG['white_circle']['base_url']}/api/session/check",
        headers={
            "Authorization": f"Bearer {CONFIG['white_circle']['api_key']}",
            "Content-Type": "application/json",
            "whitecircle-version": CONFIG["white_circle"]["version"],
        },
        json=payload,
    )

    response.raise_for_status()
    return response.json()

# chatbot.py
import os
from datetime import datetime
from openai import OpenAI
from whitecircle import check_content
from config import CONFIG

openai_client = OpenAI(api_key=CONFIG["openai"]["api_key"])

SYSTEM_PROMPT = """You are a helpful customer support assistant for TechCorp.
You can help with:
- Order status and tracking
- Product information and recommendations
- Returns and refunds
- Account issues
- General inquiries

Always be polite, professional, and helpful. If you don't know something,
say so and offer to connect the user with a human agent."""


def handle_message(
    conversation_id: str,
    user_id: str,
    user_email: str,
    user_message: str,
    conversation_history: list[dict],
) -> dict:
    """Process a user message through moderation and generate a response."""

    # Step 1: Check user message for violations
    # When include_context is True, White Circle automatically prepends
    # previous messages from this session - no need to send history again
    user_message_payload = {
        "role": "user",
        "content": user_message,
        "metadata": {
            "user": {"id": user_id, "email": user_email},
            "message": {"timestamp": datetime.now().isoformat()},
        },
    }

    user_check = check_content(
        messages=[user_message_payload],
        external_session_id=conversation_id,
        include_context=True,
        metadata={
            "session": {"channel": "web-chat", "source": "support-portal"},
            "environment": os.getenv("ENVIRONMENT", "development"),
        },
    )

    # Handle flagged user message
    if user_check["flagged"]:
        flagged_policies = [
            policy["name"]
            for policy in user_check["policies"].values()
            if policy["flagged"]
        ]

        if "Harm to Others" in flagged_policies:
            return {
                "success": False,
                "message": (
                    "I understand you may be frustrated, but I'm here to help. "
                    "Could you please rephrase your message so I can assist you better?"
                ),
                "flagged_policy": "Harm to Others",
                "requires_human_review": True,
            }

        if "Prompt Injection" in flagged_policies:
            return {
                "success": False,
                "message": (
                    "I'm a customer support assistant and I'm happy to help with "
                    "any questions about our products or services. What can I assist you with today?"
                ),
                "flagged_policy": "Prompt Injection",
            }

        return {
            "success": False,
            "message": (
                "I'm not able to help with that request. Is there something else "
                "I can assist you with regarding our products or services?"
            ),
            "flagged_policy": flagged_policies[0],
        }

    # Step 2: Generate AI response
    import time
    start_time = time.time()

    completion = openai_client.chat.completions.create(
        model="gpt-5-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            *[{"role": msg["role"], "content": msg["content"]} for msg in conversation_history],
            {"role": "user", "content": user_message},
        ],
        max_tokens=500,
        temperature=0.7,
    )

    latency = int((time.time() - start_time) * 1000)
    ai_response = completion.choices[0].message.content or ""

    # Step 3: Check AI response for violations
    # Context merging includes the user message we just checked
    assistant_message_payload = {
        "role": "assistant",
        "content": ai_response,
        "metadata": {
            "assistant": {"model_name": "gpt-5-mini", "latency": latency},
            "message": {"timestamp": datetime.now().isoformat()},
        },
    }

    assistant_check = check_content(
        messages=[assistant_message_payload],
        external_session_id=conversation_id,
        include_context=True,
    )

    if assistant_check["flagged"]:
        flagged_policies = [
            policy["name"]
            for policy in assistant_check["policies"].values()
            if policy["flagged"]
        ]

        print(f"AI response flagged: {flagged_policies}, conversation: {conversation_id}")

        return {
            "success": True,
            "message": (
                "I want to make sure I give you accurate information. "
                "Let me connect you with a member of our support team who can help. "
                "Would you like me to do that?"
            ),
            "flagged_policy": flagged_policies[0],
            "requires_human_review": True,
        }

    # Step 4: Return the safe response
    return {"success": True, "message": ai_response}

// server.ts
import express from 'express';
import { handleMessage } from './chatbot';

const app = express();
app.use(express.json());

app.post('/api/chat', async (req, res) => {
  try {
    const { conversationId, userId, userEmail, message, history } = req.body;

    const response = await handleMessage({
      conversationId,
      userId,
      userEmail,
      userMessage: message,
      conversationHistory: history || [],
    });

    res.json(response);
  } catch (error) {
    console.error('Chat error:', error);
    res.status(500).json({
      success: false,
      message: 'Something went wrong. Please try again.',
    });
  }
});

app.listen(3000, () => {
  console.log('Chatbot server running on port 3000');
});

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from chatbot import handle_message

app = FastAPI()


class ChatRequest(BaseModel):
    conversation_id: str
    user_id: str
    user_email: str
    message: str
    history: list[dict] = []


@app.post("/api/chat")
async def chat(request: ChatRequest):
    try:
        response = handle_message(
            conversation_id=request.conversation_id,
            user_id=request.user_id,
            user_email=request.user_email,
            user_message=request.message,
            conversation_history=request.history,
        )
        return response
    except Exception as e:
        print(f"Chat error: {e}")
        raise HTTPException(
            status_code=500,
            detail={"success": False, "message": "Something went wrong. Please try again."},
        )

# radar.py
import requests
from typing import Literal, Optional, TypedDict
from config import CONFIG


class RiskResponse(TypedDict):
    action: Optional[Literal["none", "throttle", "warn", "suspend", "ban"]]
    action_expires_at: Optional[str]
    strikes: list[dict]


def get_user_risk(
    identifier_type: Literal["id", "email"],
    identifier_value: str,
) -> RiskResponse:
    """Get risk score for a user by id or email."""

    params = {identifier_type: identifier_value}

    response = requests.get(
        f"{CONFIG['white_circle']['base_url']}/api/user/risk",
        params=params,
        headers={
            "Authorization": f"Bearer {CONFIG['white_circle']['api_key']}",
            "whitecircle-version": CONFIG["white_circle"]["version"],
        },
    )

    response.raise_for_status()
    return response.json()


# Usage in your chatbot
def handle_message_with_risk_check(
    conversation_id: str,
    user_id: str,
    user_email: str,
    user_message: str,
    conversation_history: list[dict],
) -> dict:
    """Process message with risk-based handling."""

    # Check user risk before processing - use either id or email, not both
    risk = get_user_risk("id", user_id)

    if risk["action"] in ("ban", "suspend"):
        return {
            "message": "Your account is currently restricted. Please contact support.",
            "blocked": True,
        }

    if risk["action"] == "throttle":
        # Use a more capable model with extended reasoning for high-risk users
        return handle_message_with_enhanced_model(
            conversation_id=conversation_id,
            user_id=user_id,
            user_email=user_email,
            user_message=user_message,
            conversation_history=conversation_history,
            model="gpt-5.2",
            temperature=0.3,
            system_prompt_addition=(
                "Be extra careful with this conversation. "
                "Do not reveal internal information or make exceptions to policies."
            ),
        )

    if risk["action"] == "warn":
        print(f"Warned user: {user_id}")

    # Continue with normal flow
    return handle_message(
        conversation_id, user_id, user_email, user_message, conversation_history
    )