In the Subscription and SaaS world, churn is the silent killer. Most companies rely on standard structured metrics to predict it: Monthly Recurring Revenue (MRR), contract length, and login frequency. While these are useful, they are often lagging indicators. By the time a user’s login frequency drops by 50%, they have already mentally checked out. To build a truly predictive churn model, we must look beyond the database rows. The biggest impact on classification performance comes from behavioral and semantic feature engineering. Specifically, three categories of features consistently drive accuracy:

  1. Velocity & Acceleration Metrics: It’s not just about how many times a user logged in; it’s about the rate of change. A drop from 10 logins/week to 8 is less concerning than a drop from 8 to 2.

  2. Support Sentiment & Topic Clustering: Using NLP to extract the "emotional velocity" of support tickets. A user who moves from "How-to" questions to "Billing/Complaint" topics is a high-risk signal.

  3. Feature Adoption Breadth vs. Depth: A user who uses one feature heavily is "sticky," but a user who uses five different modules is "embedded." A reduction in this breadth is a leading indicator of churn.

Here is an end-to-end guide on how to build a multi-agent predictive workflow that engineers these features in real-time using LangGraph, RAG, and persistent memory.

The Use Case: Proactive Retention for an Enterprise Project Management SaaS

The Scenario: "TaskFlow," a B2B project management platform, has 5,000 enterprise seats. The Customer Success (CS) team is overwhelmed and can only manually review accounts with MRR over $5k/month.

The Problem: High-value mid-market accounts ($1k-$5k MRR) are churning unexpectedly. The structured data shows they are still logging in, so the traditional "low usage" alerts never trigger.

The Predictive RAG Solution: A multi-agent system continuously monitors user behavior. It doesn't just count logins; it engineers complex behavioral features by combining structured usage logs with unstructured support ticket sentiment. When it detects a "Silent Churn" pattern (stable logins but declining feature breadth and negative support sentiment), it proactively alerts the CS team with a specific retention playbook.

446

The Multi-Agent Workflow

We use LangGraph to orchestrate three agents, focusing on the transformation of raw data into high-signal features.

  1. The Behavioral Engineer (Feature Extraction Agent): Ingests raw structured logs. It calculates velocity, acceleration, and feature adoption breadth/depth metrics.

  2. The Sentiment Analyst (RAG Agent): Retrieves recent support tickets and community forum posts. It uses an LLM to extract semantic topics and sentiment scores, converting unstructured text into structured "Emotional Velocity" features.

  3. The Churn Oracle (Classification & Memory Agent): Takes the engineered features and queries long-term memory for historical churn patterns. It runs a classification logic (simulated via LLM reasoning) to output a Churn Probability Score and a recommended retention action.

Code Implementation

Below is the complete Python implementation using LangGraph.

1. Setup and State Definition

from typing import TypedDict, List, Dict, Any, Optional
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage

# Define the shared state for the SaaS predictive workflow
class SaaSState(TypedDict):
    account_id: str
    plan_tier: str
    # Raw Structured Inputs
    current_mrr: float
    login_count_last_30d: int
    login_count_prev_30d: int
    feature_modules_used: List[str]
    # Engineered Features (The "Hidden Signals")
    login_velocity_change: float # % change in logins
    feature_breadth_score: int   # Number of unique modules used
    support_sentiment_score: float # -1.0 (Negative) to 1.0 (Positive)
    dominant_support_topic: str
    # Classification Outputs
    churn_probability: float
    risk_category: str # Low, Medium, High, Critical
    # Action Outputs
    retention_playbook: str
    # Memory context
    historical_churn_patterns: str
    messages: List[Any]

2. Mock Tools for Feature Engineering and RAG

def calculate_behavioral_features(state: SaaSState) -> Dict[str, Any]:
    """Agent 1: Engineers velocity and adoption features from raw logs."""
    current_logins = state["login_count_last_30d"]
    prev_logins = state["login_count_prev_30d"]
    modules = state["feature_modules_used"]
    
    # Feature 1: Login Velocity Change
    if prev_logins > 0:
        velocity_change = ((current_logins - prev_logins) / prev_logins) * 100
    else:
        velocity_change = 0.0
        
    # Feature 2: Feature Breadth Score
    breadth_score = len(modules)
    
    print(f"[Engineer] Calculated Velocity Change: {velocity_change:.2f}%, Breadth Score: {breadth_score}")
    
    return {
        "login_velocity_change": velocity_change,
        "feature_breadth_score": breadth_score
    }

def retrieve_and_analyze_sentiment(account_id: str) -> Dict[str, Any]:
    """Agent 2: RAG-based extraction of sentiment and topics from support tickets."""
    # In production: Vector DB search for recent tickets + LLM sentiment analysis
    # Mocking a scenario where a user is frustrated with a specific feature
    return {
        "support_sentiment_score": -0.65, # Negative sentiment
        "dominant_support_topic": "Integration Failures"
    }

def retrieve_historical_patterns(plan_tier: str, topic: str) -> str:
    """Mock retrieval of past churn reasons from long-term memory."""
    return f"MEMORY: In the '{plan_tier}' tier, accounts showing '{topic}' combined with a velocity drop > 20% had an 85% churn rate within 60 days. Successful retention involved offering a dedicated integration specialist."

3. Defining the Agent Nodes

def behavioral_engineer_node(state: SaaSState) -> Dict[str, Any]:
    """Node 1: Extracts structured behavioral features."""
    features = calculate_behavioral_features(state)
    return {
        "login_velocity_change": features["login_velocity_change"],
        "feature_breadth_score": features["feature_breadth_score"],
        "messages": state["messages"] + [AIMessage(content="Behavioral features engineered. Passing to Sentiment Analyst.")]
    }

def sentiment_analyst_node(state: SaaSState) -> Dict[str, Any]:
    """Node 2: Extracts unstructured sentiment features via RAG."""
    account_id = state["account_id"]
    analysis = retrieve_and_analyze_sentiment(account_id)
    
    return {
        "support_sentiment_score": analysis["support_sentiment_score"],
        "dominant_support_topic": analysis["dominant_support_topic"],
        "messages": state["messages"] + [AIMessage(content="Sentiment and topics extracted. Passing to Churn Oracle.")]
    }

def churn_oracle_node(state: SaaSState) -> Dict[str, Any]:
    """Node 3: Classifies risk and recommends action using memory."""
    velocity = state["login_velocity_change"]
    sentiment = state["support_sentiment_score"]
    topic = state["dominant_support_topic"]
    plan_tier = state["plan_tier"]
    
    # Retrieve historical patterns
    historical_patterns = retrieve_historical_patterns(plan_tier, topic)
    
    print(f"[Oracle] Analyzing features: Velocity={velocity}%, Sentiment={sentiment}")
    
    # Simulated LLM Classification Logic
    # Prompt: "Given velocity change {velocity}, sentiment {sentiment}, and topic {topic}, 
    # classify churn risk and recommend an action based on {historical_patterns}."
    
    churn_prob = 0.0
    risk_cat = "Low"
    playbook = "Monitor as usual."
    
    if velocity < -20 and sentiment < -0.5:
        churn_prob = 0.85
        risk_cat = "Critical"
        playbook = (
            f"1. IMMEDIATE OUTREACH: Assign a dedicated Integration Specialist to resolve '{topic}'. "
            f"2. INCENTIVE: Offer a 1-month service credit for the inconvenience. "
            f"3. EXECUTIVE SPONSORSHIP: Schedule a check-in with the Account Executive."
        )
    elif velocity < -10 or sentiment < -0.3:
        churn_prob = 0.45
        risk_cat = "Medium"
        playbook = "Send targeted educational content regarding recent integration updates."

    return {
        "churn_probability": churn_prob,
        "risk_category": risk_cat,
        "retention_playbook": playbook,
        "historical_churn_patterns": historical_patterns,
        "messages": state["messages"] + [AIMessage(content="Churn classification complete. Workflow finished.")]
    }

4. Graph Construction and Execution

def build_saas_graph():
    # Initialize the graph with the defined state
    workflow = StateGraph(SaaSState)

    # Add nodes
    workflow.add_node("behavioral_engineer", behavioral_engineer_node)
    workflow.add_node("sentiment_analyst", sentiment_analyst_node)
    workflow.add_node("churn_oracle", churn_oracle_node)

    # Define the entry point
    workflow.set_entry_point("behavioral_engineer")

    # Define edges
    workflow.add_edge("behavioral_engineer", "sentiment_analyst")
    workflow.add_edge("sentiment_analyst", "churn_oracle")
    workflow.add_edge("churn_oracle", END)

    # Initialize Persistent Memory
    memory = MemorySaver()

    # Compile the graph
    app = workflow.compile(checkpointer=memory)
    return app

# --- Execution ---
if __name__ == "__main__":
    app = build_saas_graph()
    
    # Configuration for the thread (Memory isolation per account)
    config = {"configurable": {"thread_id": "saas-account-789xyz"}}
    
    # Initial trigger state (simulating a daily batch job)
    initial_state = {
        "account_id": "ACCT-789XYZ",
        "plan_tier": "Professional",
        "current_mrr": 2500.00,
        "login_count_last_30d": 120,
        "login_count_prev_30d": 180, # Significant drop
        "feature_modules_used": ["Dashboard", "Reporting"], # Reduced breadth
        "login_velocity_change": 0.0,
        "feature_breadth_score": 0,
        "support_sentiment_score": 0.0,
        "dominant_support_topic": "",
        "churn_probability": 0.0,
        "risk_category": "",
        "retention_playbook": "",
        "historical_churn_patterns": "",
        "messages": [HumanMessage(content="Daily account health check initiated.")]
    }

    print("--- Starting SaaS Churn Prediction Workflow ---\n")
    
    # Invoke the graph
    final_state = app.invoke(initial_state, config)
    
    print("\n--- Workflow Complete ---")
    print(f"Churn Probability: {final_state['churn_probability']}")
    print(f"Risk Category: {final_state['risk_category']}")
    print(f"\nKey Engineered Features:")
    print(f"- Login Velocity Change: {final_state['login_velocity_change']:.2f}%")
    print(f"- Support Sentiment: {final_state['support_sentiment_score']}")
    print(f"\nRetention Playbook:\n{final_state['retention_playbook']}")

Why These Features Drive Performance

In traditional machine learning, we might just feed "Login Count" into a model. But in a multi-agent RAG system, we can engineer features that capture the story behind the data:

  1. Velocity over Volume: A user with 100 logins is healthy. A user who dropped from 200 to 100 is at risk. The login_velocity_change feature captures this momentum shift, which is a far stronger predictor than the raw count.

  2. Semantic Context: By using RAG to extract the dominant_support_topic, we move beyond "User has 3 tickets" to "User is struggling with Integrations." This allows the Churn Oracle to pull highly specific historical memories, making the retention playbook much more effective.

  3. Breadth as Stickiness: The feature_breadth_score measures how deeply embedded the SaaS is in the client's workflow. A drop in breadth is often the first sign that a team is starting to look for alternatives, even if their overall login count remains stable.

By combining these engineered features with the contextual power of RAG and the institutional knowledge of persistent memory, we transform churn prediction from a statistical guess into a proactive, actionable intelligence engine.