Implementing Custom Article Ranking for NICE CXone Agent Assist with Python

What You Will Build

• A Python service that intercepts knowledge base retrieval, applies a learned-to-rank scoring algorithm, routes sessions through an A/B testing framework, and pushes the reordered results to the NICE CXone Agent Assist API.
• This implementation uses the NICE CXone Agent Assist REST API endpoint /api/v2/agentassist/sessions/{sessionId}/suggestions and standard HTTP libraries.
• The tutorial covers Python 3.9+ with requests, numpy, and typing.

Prerequisites

• OAuth 2.0 Client Credentials grant with the agentassist:write scope assigned to the API key.
• NICE CXone API version 2 (/api/v2/).
• Python 3.9 or higher.
• External dependencies: requests>=2.28.0, numpy>=1.24.0, urllib3>=1.26.0. Install via pip install requests numpy urllib3.

Authentication Setup

NICE CXone uses standard OAuth 2.0 Client Credentials flow. The authentication client must cache tokens and refresh before expiration to avoid unnecessary network calls. The agentassist:write scope is mandatory for submitting suggestions.

import requests
import time
from typing import Optional

class CXoneAuth:
    """Handles OAuth 2.0 token retrieval and caching for NICE CXone."""
    def __init__(self, client_id: str, client_secret: str, region: str = "us-2"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://api.{region}.convoacloud.com/oauth/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0
        self._session = requests.Session()

    def get_token(self) -> str:
        """Returns a valid access token. Refreshes automatically if expired."""
        if self._token and time.time() < self._expires_at:
            return self._token
        
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        
        response = self._session.post(self.token_url, data=payload)
        response.raise_for_status()
        
        data = response.json()
        self._token = data["access_token"]
        # Subtract 60 seconds for safe refresh window
        self._expires_at = time.time() + data["expires_in"] - 60
        return self._token

The token cache uses a 60-second safety buffer before expiration. This prevents race conditions where multiple concurrent requests attempt simultaneous token refreshes.

Implementation

Step 1: Configure A/B Testing Router and Feedback Collection

A/B testing requires deterministic bucket assignment without external state. Hashing the session ID guarantees consistent routing across service restarts. The feedback collector tracks clicks and dismissals exclusively for treatment groups to prevent data contamination.

import hashlib
from collections import defaultdict
from typing import Dict, List

class ABTestRouter:
    """Deterministic session bucketing and feedback tracking."""
    def __init__(self):
        # Structure: {session_id: {"clicks": [article_ids], "dismissals": [article_ids], "is_treatment": bool}}
        self.feedback_store: Dict[str, Dict] = defaultdict(
            lambda: {"clicks": [], "dismissals": [], "is_treatment": False}
        )

    def assign_bucket(self, session_id: str) -> bool:
        """Returns True if session belongs to the treatment group (custom ranking)."""
        hash_digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest()
        hash_value = int(hash_digest, 16)
        # 50% treatment, 50% control
        return hash_value % 100 >= 50

    def record_feedback(self, session_id: str, article_id: str, action: str) -> None:
        """Records agent interaction only for treatment buckets."""
        if self.assign_bucket(session_id):
            self.feedback_store[session_id]["is_treatment"] = True
            if action == "click":
                self.feedback_store[session_id]["clicks"].append(article_id)
            elif action == "dismiss":
                self.feedback_store[session_id]["dismissals"].append(article_id)

The hash modulo operation distributes traffic evenly. Control groups continue to use native CXone relevance scoring, while treatment groups receive the learned-to-rank output.

Step 2: Feature Extraction and Learned-to-Rank Scoring

Listwise ranking outperforms pointwise scoring because it optimizes the entire suggestion list against a global objective function. The feature vector below represents a simplified production model. You will replace the linear weight multiplication with a lightgbm or xgboost model trained on historical click_through_rate and resolution_time labels.

import numpy as np
import time
from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class KnowledgeArticle:
    id: str
    title: str
    category: str
    last_updated: float
    historical_ctr: float
    url: str

class LearntToRankScorer:
    """Applies feature weighting to reorder knowledge base articles."""
    def __init__(self):
        # Production systems load these weights from a trained model file
        # Feature order: title_match, category_match, recency_score, historical_ctr
        self.feature_weights = np.array([0.35, 0.25, 0.20, 0.20])

    def extract_features(self, article: KnowledgeArticle, query: str) -> List[float]:
        """Converts article metadata and query into a numerical feature vector."""
        title_match = 1.0 if query.lower() in article.title.lower() else 0.0
        category_match = 1.0 if article.category.lower() in query.lower() else 0.0
        
        # Recency decays exponentially: newer articles score higher
        days_old = max(0.0, (time.time() - article.last_updated) / 86400.0)
        recency_score = np.exp(-0.1 * days_old)
        
        return [title_match, category_match, recency_score, article.historical_ctr]

    def score_and_rank(self, articles: List[KnowledgeArticle], query: str) -> List[Tuple[KnowledgeArticle, float]]:
        """Returns articles sorted by descending relevance score."""
        scored_articles: List[Tuple[KnowledgeArticle, float]] = []
        
        for article in articles:
            features = self.extract_features(article, query)
            # Linear combination mimics a trained LTR model output
            relevance_score = float(np.dot(features, self.feature_weights))
            scored_articles.append((article, relevance_score))
            
        # Sort descending by score
        scored_articles.sort(key=lambda x: x[1], reverse=True)
        return scored_articles

The extract_features method normalizes inputs to a 0-1 range where applicable. The exponential decay function ensures recently updated articles receive priority without completely discarding evergreen content.

Step 3: Paginated Knowledge Base Retrieval and Ranking

NICE CXone Knowledge API returns results in paginated blocks. The ranking service must fetch all relevant pages, flatten the results, and apply the scoring function before submission. This step demonstrates proper pagination handling and memory-efficient processing.

from typing import List

class KnowledgeRetriever:
    """Simulates paginated CXone Knowledge API retrieval."""
    def __init__(self, base_url: str, auth: CXoneAuth):
        self.base_url = base_url
        self.auth = auth
        self.session = requests.Session()

    def fetch_all_articles(self, query: str, limit: int = 20) -> List[KnowledgeArticle]:
        """Paginates through CXone Knowledge API and returns flattened results."""
        all_articles: List[KnowledgeArticle] = []
        offset = 0
        headers = {"Authorization": f"Bearer {self.auth.get_token()}"}
        
        while True:
            url = f"{self.base_url}/api/v2/knowledge/articles"
            params = {"query": query, "limit": limit, "offset": offset}
            
            response = self.session.get(url, params=params, headers=headers)
            response.raise_for_status()
            
            data = response.json()
            articles_page = data.get("articles", [])
            if not articles_page:
                break
                
            for art in articles_page:
                all_articles.append(KnowledgeArticle(
                    id=art["id"],
                    title=art["title"],
                    category=art.get("category", "general"),
                    last_updated=art.get("lastUpdatedDate", time.time()),
                    historical_ctr=art.get("historicalCtR", 0.0),
                    url=art.get("url", "")
                ))
                
            # Pagination check
            if len(articles_page) < limit:
                break
            offset += limit
            
        return all_articles

The pagination loop terminates when the returned page contains fewer items than the requested limit. This matches the standard CXone API pagination contract.

Step 4: Submit Ranked Suggestions with Retry Logic

The Agent Assist API enforces strict rate limits. The submission client uses urllib3 retry logic with exponential backoff for 429 and 5xx responses. The payload structure must match the CXone suggestion schema exactly.

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from typing import List, Tuple, Dict

class CXoneAgentAssistClient:
    """Handles suggestion submission with production-grade retry logic."""
    def __init__(self, auth: CXoneAuth, region: str = "us-2"):
        self.auth = auth
        self.base_url = f"https://api.{region}.convoacloud.com"
        self.session = requests.Session()
        
        # Configure retry strategy for transient failures and rate limits
        retry_strategy = Retry(
            total=3,
            backoff_factor=1.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)

    def submit_ranked_suggestions(self, session_id: str, ranked_articles: List[Tuple[KnowledgeArticle, float]], max_suggestions: int = 5) -> Dict:
        """Pushes the top N ranked articles to the Agent Assist session."""
        url = f"{self.base_url}/api/v2/agentassist/sessions/{session_id}/suggestions"
        
        headers = {
            "Authorization": f"Bearer {self.auth.get_token()}",
            "Content-Type": "application/json"
        }
        
        # Format payload according to CXone Agent Assist schema
        suggestions_payload = []
        for article, score in ranked_articles[:max_suggestions]:
            suggestions_payload.append({
                "type": "article",
                "content": {
                    "id": article.id,
                    "title": article.title,
                    "url": article.url,
                    "preview": article.title[:100] + "..." if len(article.title) > 100 else article.title
                },
                "score": round(score, 4)
            })
            
        payload = {"suggestions": suggestions_payload}
        
        response = self.session.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()

The retry strategy applies exponential backoff (1.5s, 2.25s, 3.375s) for 429 responses. The allowed_methods parameter restricts retries to POST operations. The payload truncates preview text to prevent JSON payload size violations.

Complete Working Example

The following script combines all components into a runnable service. Replace the credential placeholders before execution.

import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import numpy as np
import hashlib
from collections import defaultdict
from dataclasses import dataclass
from typing import List, Tuple, Dict, Optional

@dataclass
class KnowledgeArticle:
    id: str
    title: str
    category: str
    last_updated: float
    historical_ctr: float
    url: str

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, region: str = "us-2"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://api.{region}.convoacloud.com/oauth/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0
        self._session = requests.Session()

    def get_token(self) -> str:
        if self._token and time.time() < self._expires_at:
            return self._token
        payload = {"grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret}
        response = self._session.post(self.token_url, data=payload)
        response.raise_for_status()
        data = response.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"] - 60
        return self._token

class ABTestRouter:
    def __init__(self):
        self.feedback_store: Dict[str, Dict] = defaultdict(
            lambda: {"clicks": [], "dismissals": [], "is_treatment": False}
        )

    def assign_bucket(self, session_id: str) -> bool:
        hash_digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest()
        return int(hash_digest, 16) % 100 >= 50

    def record_feedback(self, session_id: str, article_id: str, action: str) -> None:
        if self.assign_bucket(session_id):
            self.feedback_store[session_id]["is_treatment"] = True
            if action == "click":
                self.feedback_store[session_id]["clicks"].append(article_id)
            elif action == "dismiss":
                self.feedback_store[session_id]["dismissals"].append(article_id)

class LearntToRankScorer:
    def __init__(self):
        self.feature_weights = np.array([0.35, 0.25, 0.20, 0.20])

    def extract_features(self, article: KnowledgeArticle, query: str) -> List[float]:
        title_match = 1.0 if query.lower() in article.title.lower() else 0.0
        category_match = 1.0 if article.category.lower() in query.lower() else 0.0
        days_old = max(0.0, (time.time() - article.last_updated) / 86400.0)
        recency_score = np.exp(-0.1 * days_old)
        return [title_match, category_match, recency_score, article.historical_ctr]

    def score_and_rank(self, articles: List[KnowledgeArticle], query: str) -> List[Tuple[KnowledgeArticle, float]]:
        scored_articles: List[Tuple[KnowledgeArticle, float]] = []
        for article in articles:
            features = self.extract_features(article, query)
            relevance_score = float(np.dot(features, self.feature_weights))
            scored_articles.append((article, relevance_score))
        scored_articles.sort(key=lambda x: x[1], reverse=True)
        return scored_articles

class CXoneAgentAssistClient:
    def __init__(self, auth: CXoneAuth, region: str = "us-2"):
        self.auth = auth
        self.base_url = f"https://api.{region}.convoacloud.com"
        self.session = requests.Session()
        retry_strategy = Retry(total=3, backoff_factor=1.5, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["POST"])
        self.session.mount("https://", HTTPAdapter(max_retries=retry_strategy))

    def submit_ranked_suggestions(self, session_id: str, ranked_articles: List[Tuple[KnowledgeArticle, float]], max_suggestions: int = 5) -> Dict:
        url = f"{self.base_url}/api/v2/agentassist/sessions/{session_id}/suggestions"
        headers = {"Authorization": f"Bearer {self.auth.get_token()}", "Content-Type": "application/json"}
        suggestions_payload = []
        for article, score in ranked_articles[:max_suggestions]:
            suggestions_payload.append({
                "type": "article",
                "content": {"id": article.id, "title": article.title, "url": article.url, "preview": article.title},
                "score": round(score, 4)
            })
        payload = {"suggestions": suggestions_payload}
        response = self.session.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()

def main():
    # Configuration
    CLIENT_ID = "YOUR_CLIENT_ID"
    CLIENT_SECRET = "YOUR_CLIENT_SECRET"
    REGION = "us-2"
    SESSION_ID = "agent-session-8821"
    QUERY = "billing payment failed"
    
    auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, REGION)
    router = ABTestRouter()
    scorer = LearntToRankScorer()
    client = CXoneAgentAssistClient(auth, REGION)
    
    # Simulated retrieved articles (replace with actual Knowledge API pagination call)
    retrieved_articles = [
        KnowledgeArticle("art-001", "Payment Failed Troubleshooting", "billing", time.time() - 86400, 0.75, "https://kb.example.com/art-001"),
        KnowledgeArticle("art-002", "General Account Settings", "account", time.time() - 172800, 0.30, "https://kb.example.com/art-002"),
        KnowledgeArticle("art-003", "Billing Cycle Updates", "billing", time.time() - 3600, 0.82, "https://kb.example.com/art-003")
    ]
    
    is_treatment = router.assign_bucket(SESSION_ID)
    print(f"Session {SESSION_ID} assigned to {'Treatment' if is_treatment else 'Control'} group.")
    
    if is_treatment:
        ranked = scorer.score_and_rank(retrieved_articles, QUERY)
        print("Applying learned-to-rank scoring...")
        for article, score in ranked:
            print(f"  {article.title}: {score:.4f}")
            
        try:
            result = client.submit_ranked_suggestions(SESSION_ID, ranked)
            print(f"Suggestions submitted successfully. Response: {result}")
        except requests.exceptions.HTTPError as e:
            print(f"API submission failed: {e.response.status_code} - {e.response.text}")
    else:
        print("Control group: Skipping custom ranking. Using native CXone relevance.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token has expired, the client credentials are incorrect, or the API key lacks the agentassist:write scope.
• Fix: Verify the client ID and secret match the credentials registered in the CXone Admin Portal. Confirm the API key has the agentassist:write scope assigned. The CXoneAuth class automatically refreshes tokens before expiration, but network timeouts during refresh will trigger this error.
• Code showing the fix: The get_token method already implements TTL-based caching. If you encounter persistent 401 errors, add explicit scope validation during token retrieval by checking the response.json() payload for scope grants, or regenerate the API key in the CXone console.

Error: 429 Too Many Requests

• Cause: The service exceeds CXone rate limits for the Agent Assist endpoint. CXone enforces limits per tenant and per API key.
• Fix: The HTTPAdapter with Retry handles automatic backoff. If retries exhaust, implement a token bucket or leaky bucket rate limiter at the application layer. Batch suggestion submissions to reduce call frequency.
• Code showing the fix: The retry strategy in CXoneAgentAssistClient already covers this. For high-volume deployments, add a pre-flight check:

import time
class RateLimiter:
    def __init__(self, max_calls: int = 10, period: int = 60):
        self.max_calls = max_calls
        self.period = period
        self.calls = []
    def acquire(self):
        now = time.time()
        self.calls = [t for t in self.calls if now - t < self.period]
        if len(self.calls) >= self.max_calls:
            wait_time = self.period - (now - self.calls[0])
            time.sleep(wait_time)
        self.calls.append(now)

Error: 400 Bad Request (Invalid Suggestion Payload)

• Cause: The JSON payload structure does not match the CXone Agent Assist schema. Common mistakes include missing type fields, incorrect content nesting, or exceeding the maximum suggestion count per request.
• Fix: Validate the payload against the official schema before submission. Ensure type is set to article and content contains id, title, and url. Limit submissions to 5-10 suggestions per call to match UI rendering constraints.
• Code showing the fix: The submit_ranked_suggestions method enforces max_suggestions=5 and structures the payload exactly as required. Add a validation step before the POST call:

if not payload["suggestions"]:
    raise ValueError("Cannot submit empty suggestion list.")
for s in payload["suggestions"]:
    if "type" not in s or "content" not in s:
        raise ValueError("Invalid suggestion schema.")

Official References

• NICE CXone Agent Assist API Documentation
• NICE CXone OAuth 2.0 Authentication Guide