Adding sentiment detection to a support bot in LangChain

Adding sentiment detection to a LangChain support bot lets you catch frustrated users before they churn, route angry tickets to humans, and measure how your automated responses land. This tutorial walks through a production-ready pattern: a lightweight classifier that runs alongside your chain, a policy layer that decides when to escalate, and the observability hooks you need to iterate. The approach works whether you’re building on OpenAI, Anthropic, or a multi-provider gateway like n4n.ai that handles fallback and usage metering automatically.

Step 1: Choose a classifier that fits your latency budget

Sentiment detection runs on every user turn, so latency matters. You have three practical options:

Option A: Small fine-tuned transformer (recommended for most teams)
A DistilBERT or MiniLM model fine-tuned on customer support data runs in 15-30 ms on CPU. Hugging Face hosts several ready-to-use checkpoints like distilbert-base-uncased-finetuned-sst-2-english for general sentiment or cardiffnlp/twitter-roberta-base-sentiment-latest for social-style text. For support-specific language, fine-tune on your own labeled conversations.

Option B: LLM-as-judge (highest accuracy, highest latency)
Prompt a small model (gpt-4o-mini, claude-3-haiku) to classify sentiment. Expect 200-500 ms. Use this when you need nuanced categories (frustrated vs. confused vs. angry) and can tolerate the delay.

Option C: Rule-based / VADER (fastest, lowest accuracy)
NLTK’s VADER runs in <5 ms but struggles with sarcasm, negation scope, and domain-specific language. Acceptable only as a fallback or for very simple triage.

For this tutorial we’ll use Option A with a local ONNX model via optimum.onnxruntime — no GPU required, deterministic latency, and easy to swap later.

# sentiment_classifier.py
from optimum.onnxruntime import ORTModelForSequenceClassification
from transformers import AutoTokenizer
import numpy as np

class SentimentClassifier:
    def __init__(self, model_id: str = "distilbert-base-uncased-finetuned-sst-2-english"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_id)
        self.model = ORTModelForSequenceClassification.from_pretrained(model_id, export=True)
        self.labels = ["negative", "positive"]  # SST-2 is binary; adjust for your model

    def predict(self, text: str) -> dict:
        inputs = self.tokenizer(text, return_tensors="np", truncation=True, max_length=512)
        logits = self.model(**inputs).logits
        probs = np.exp(logits) / np.exp(logits).sum(axis=-1, keepdims=True)
        idx = int(probs.argmax())
        return {
            "label": self.labels[idx],
            "score": float(probs[0, idx]),
            "all_scores": {self.labels[i]: float(probs[0, i]) for i in range(len(self.labels))}
        }

# Usage
clf = SentimentClassifier()
print(clf.predict("I've been waiting three days for a refund and nobody replies."))
# {'label': 'negative', 'score': 0.987, 'all_scores': {'negative': 0.987, 'positive': 0.013}}

Verify: Run the classifier on 20 real support messages. You should see clear separation — angry/refund/cancel language scoring >0.9 negative, thank-you/confirmation language scoring >0.9 positive.

Step 2: Wrap the classifier as a LangChain runnable

LangChain’s Runnable interface lets you compose the classifier into chains with retries, fallbacks, and streaming. Keep it stateless and thread-safe.

# runnables/sentiment.py
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_core.outputs import Generation
from pydantic import BaseModel, Field
from typing import Any, Dict
from sentiment_classifier import SentimentClassifier

class SentimentOutput(BaseModel):
    label: str
    score: float
    all_scores: Dict[str, float]

class SentimentRunnable(Runnable[str, SentimentOutput]):
    def __init__(self, classifier: SentimentClassifier):
        super().__init__()
        self._classifier = classifier

    def invoke(self, input: str, config: RunnableConfig | None = None) -> SentimentOutput:
        result = self._classifier.predict(input)
        return SentimentOutput(**result)

    async def ainvoke(self, input: str, config: RunnableConfig | None = None) -> SentimentOutput:
        # CPU-bound; run in thread pool to avoid blocking event loop
        import asyncio
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self.invoke, input, config)

# Factory for dependency injection
def make_sentiment_runnable(model_id: str = "distilbert-base-uncased-finetuned-sst-2-english") -> SentimentRunnable:
    return SentimentRunnable(SentimentClassifier(model_id))

Verify: In a REPL, call runnable.invoke("This is great!") and runnable.invoke("This is terrible."). Confirm the output schema matches SentimentOutput and latency stays under 50 ms.

Step 3: Build a policy layer that maps sentiment to actions

Raw sentiment scores aren’t actionable. You need a policy that translates scores into routing decisions, response tone adjustments, and escalation triggers. Keep this separate from the classifier so you can tune thresholds without retraining.

# policy.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from runnables.sentiment import SentimentOutput

class Action(Enum):
    CONTINUE = "continue"           # Normal bot flow
    SOFTEN_TONE = "soften_tone"     # Add empathy preamble, avoid robotic phrasing
    OFFER_HUMAN = "offer_human"     # Proactively suggest handoff
    ESCALATE_NOW = "escalate_now"   # Immediate transfer, log urgency

@dataclass(frozen=True)
class PolicyDecision:
    action: Action
    reason: str
    metadata: dict

class SentimentPolicy:
    """
    Thresholds are starting points. Calibrate on your conversation logs.
    """
    def __init__(
        self,
        negative_threshold: float = 0.75,
        strong_negative_threshold: float = 0.90,
        consecutive_negative_turns: int = 2,
    ):
        self.negative_threshold = negative_threshold
        self.strong_negative_threshold = strong_negative_threshold
        self.consecutive_negative_turns = consecutive_negative_turns
        self._negative_streak: dict[str, int] = {}  # session_id -> count

    def decide(self, session_id: str, sentiment: SentimentOutput) -> PolicyDecision:
        is_negative = sentiment.label == "negative"
        score = sentiment.score

        # Track consecutive negative turns
        if is_negative and score >= self.negative_threshold:
            self._negative_streak[session_id] = self._negative_streak.get(session_id, 0) + 1
        else:
            self._negative_streak[session_id] = 0

        streak = self._negative_streak[session_id]

        # Strong negative on first turn -> immediate escalation offer
        if is_negative and score >= self.strong_negative_threshold and streak == 1:
            return PolicyDecision(
                action=Action.ESCALATE_NOW,
                reason=f"Strong negative sentiment (score={score:.2f}) on first turn",
                metadata={"sentiment_score": score, "streak": streak}
            )

        # Repeated negative -> escalate
        if streak >= self.consecutive_negative_turns:
            return PolicyDecision(
                action=Action.ESCALATE_NOW,
                reason=f"{streak} consecutive negative turns",
                metadata={"sentiment_score": score, "streak": streak}
            )

        # Single negative turn -> soften tone + offer human
        if is_negative and score >= self.negative_threshold:
            return PolicyDecision(
                action=Action.OFFER_HUMAN,
                reason=f"Negative sentiment detected (score={score:.2f})",
                metadata={"sentiment_score": score, "streak": streak}
            )

        return PolicyDecision(
            action=Action.CONTINUE,
            reason="Sentiment neutral or positive",
            metadata={"sentiment_score": score, "streak": streak}
        )

    def reset_session(self, session_id: str):
        self._negative_streak.pop(session_id, None)

Verify: Write unit tests covering each threshold boundary. Feed the policy a sequence of sentiments for a single session_id and assert the action escalates at the correct streak count.

Step 4: Compose the full support chain with sentiment gating

Now wire the classifier, policy, and your existing QA chain together. The pattern: run sentiment in parallel with retrieval, then let policy decide the next step before generating the final answer.

# chains/support_chain.py
from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from runnables.sentiment import make_sentiment_runnable
from policy import SentimentPolicy, Action
from typing import Dict, Any

# Your existing RAG chain (simplified)
def make_qa_chain():
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful support agent. Use the context to answer. Be concise."),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    # Assume you have a retriever bound elsewhere
    # return prompt | llm
    return prompt | llm  # replace with your actual chain

# Tone-adjusted prompt for negative sentiment
SOFTENED_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful support agent. The user is frustrated. "
               "Lead with empathy, acknowledge the problem, then solve. "
               "Keep responses under 3 sentences. Avoid corporate speak."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])

def make_support_chain(retriever, policy: SentimentPolicy, session_id: str):
    sentiment_runnable = make_sentiment_runnable()
    qa_chain = make_qa_chain()
    softened_chain = SOFTENED_PROMPT | ChatOpenAI(model="gpt-4o-mini", temperature=0)

    # Parallel: retrieve + classify sentiment
    parallel = RunnableParallel(
        context=retriever,
        sentiment=sentiment_runnable,
        question=RunnablePassthrough(),
    )

    def apply_policy(inputs: Dict[str, Any]) -> Dict[str, Any]:
        decision = policy.decide(session_id, inputs["sentiment"])
        inputs["policy_decision"] = decision
        return inputs

    def route_response(inputs: Dict[str, Any]) -> str:
        decision = inputs["policy_decision"]
        context = inputs["context"]
        question = inputs["question"]

        if decision.action == Action.ESCALATE_NOW:
            # Log and return handoff message; actual transfer handled upstream
            return ("I'm connecting you with a human agent who can help resolve this. "
                    "One moment please.")
        elif decision.action in (Action.OFFER_HUMAN, Action.SOFTEN_TONE):
            # Use softened prompt
            return softened_chain.invoke({"context": context, "question": question}).content
        else:
            return qa_chain.invoke({"context": context, "question": question}).content

    chain = parallel | RunnableLambda(apply_policy) | RunnableLambda(route_response)
    return chain

Verify: Run the chain end-to-end with three test inputs for the same session_id:

1. “How do I reset my password?” → neutral, normal response
2. “This is ridiculous, I’ve asked three times!” → negative, softened tone + human offer
3. “I want to cancel and get a refund right now!” → strong negative, immediate escalation message

Confirm the policy decision metadata is logged for each turn.

Step 5: Add observability — log every decision

You cannot tune thresholds without data. Log every sentiment score, policy action, and the final response template used. Structured JSON logs make this queryable in Datadog, Splunk, or a simple ELK stack.

# observability.py
import json
import logging
import time
from contextvars import ContextVar
from policy import PolicyDecision, Action

# Context variable carries session_id through async calls
session_var: ContextVar[str] = ContextVar("session_id", default="unknown")

logger = logging.getLogger("support_bot")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))  # raw JSON lines
logger.addHandler(handler)
logger.propagate = False

def log_sentiment_event(
    user_message: str,
    sentiment: dict,
    decision: PolicyDecision,
    response_template: str,
    latency_ms: float,
):
    session_id = session_var.get()
    event = {
        "timestamp": time.time(),
        "session_id": session_id,
        "user_message": user_message[:200],  # truncate PII risk
        "sentiment_label": sentiment["label"],
        "sentiment_score": sentiment["score"],
        "policy_action": decision.action.value,
        "policy_reason": decision.reason,
        "response_template": response_template,  # "standard" | "softened" | "escalation"
        "latency_ms": round(latency_ms, 1),
    }
    logger.info(json.dumps(event))

# Usage inside your chain (wrap route_response)
def make_observable_chain(chain, policy, session_id: str):
    def wrapped(input: str) -> str:
        token = session_var.set(session_id)
        start = time.perf_counter()
        try:
            # We need access to intermediate sentiment + decision.
            # Easiest: replicate the parallel step here or expose from chain.
            # For brevity, assume chain returns (response, sentiment, decision)
            response, sentiment, decision = chain.invoke(input)  # adjust your chain to return tuple
            latency = (time.perf_counter() - start) * 1000
            template = {
                Action.CONTINUE: "standard",
                Action.SOFTEN_TONE: "softened",
                Action.OFFER_HUMAN: "softened",
                Action.ESCALATE_NOW: "escalation",
            }[decision.action]
            log_sentiment_event(input, sentiment, decision, template, latency)
            return response
        finally:
            session_var.reset(token)
    return wrapped

Verify: Tail the log output during a test conversation. Each line should be valid JSON with all fields populated. Query for policy_action: "escalation" and confirm the associated sentiment_score exceeds your strong-negative threshold.

Step 6: Calibrate thresholds on real data

Default thresholds are guesses. Run a calibration script over your last 1,000 labeled conversations (or manually label a sample) to find the operating point that balances false escalations against missed frustration.

# calibrate.py
import json
from policy import SentimentPolicy, Action
from runnables.sentiment import make_sentiment_runnable

def load_labeled_conversations(path: str) -> list[dict]:
    # [{"messages": [...], "labels": [...]}]
    with open(path) as f:
        return json.load(f)

def evaluate(policy: SentimentPolicy, conversations: list[dict]) -> dict:
    sentiment_runnable = make_sentiment_runnable()
    stats = {"tp": 0, "fp": 0, "tn": 0, "fn": 0, "escalations": 0}
    for conv in conversations:
        policy.reset_session(conv["session_id"])
        for msg, true_label in zip(conv["messages"], conv["labels"]):
            sentiment = sentiment_runnable.invoke(msg)
            decision = policy.decide(conv["session_id"], sentiment)
            predicted_escalate = decision.action in (Action.ESCALATE_NOW, Action.OFFER_HUMAN)
            true_escalate = true_label in ("frustrated", "angry", "cancel_intent")
            if predicted_escalate and true_escalate:
                stats["tp"] += 1
            elif predicted_escalate and not true_escalate:
                stats["fp"] += 1
            elif not predicted_escalate and not true_escalate:
                stats["tn"] += 1
            else:
                stats["fn"] += 1
            if decision.action == Action.ESCALATE_NOW:
                stats["escalations"] += 1
    precision = stats["tp"] / (stats["tp"] + stats["fp"]) if (stats["tp"] + stats["fp"]) else 0
    recall = stats["tp"] / (stats["tp"] + stats["fn"]) if (stats["tp"] + stats["fn"]) else 0
    return {"precision": precision, "recall": recall, **stats}

# Grid search
for neg_thresh in [0.65, 0.70, 0.75, 0.80, 0.85]:
    for strong_thresh in [0.85, 0.90, 0.95]:
        for streak in [1, 2, 3]:
            policy = SentimentPolicy(neg_thresh, strong_thresh, streak)
            metrics = evaluate(policy, load_labeled_conversations("labeled_conversations.json"))
            print(f"neg={neg_thresh} strong={strong_thresh} streak={streak} "
                  f"P={metrics['precision']:.2f} R={metrics['recall']:.2f} esc={metrics['escalations']}")

Verify: Pick the threshold set that gives you ≥0.85 recall on frustration detection while keeping precision ≥0.70 (so agents aren’t drowning in false escalations). Record the chosen values in your config management.

Step 7: Handle edge cases in production

Multilingual users

The DistilBERT SST-2 model is English-only. For multilingual support, swap to cardiffnlp/twitter-xlm-roberta-base-sentiment (covers ~30 languages) or deploy a language detection step (fasttext or langdetect) that routes to per-language classifiers.

Short messages (“ok”, “thanks”, “???”)

Single-token inputs produce unreliable scores. Add a heuristic: if len(tokens) < 4, skip sentiment and default to CONTINUE.

def should_skip_sentiment(text: str, tokenizer) -> bool:
    return len(tokenizer.encode(text)) < 4

PII in logs

The observability snippet truncates user_message to 200 chars. For stricter compliance, hash the message (sha256(message)[:16]) and store the full text in an encrypted, access-controlled bucket with a separate retention policy.

Provider failures

If your LLM provider returns 429 or 5xx, the sentiment runnable (running locally) still works. Your chain should catch upstream failures and return a graceful degradation message — this is where a gateway with automatic fallback (like n4n.ai) saves you from writing custom retry logic per provider.

Step 8: Deploy and iterate

Package the classifier as a separate container (or Lambda layer) so you can update the model without rebuilding the entire bot. Version your policy thresholds in config, not code. Set up a weekly review: pull the last 7 days of sentiment logs, compute escalation rate, false positive rate, and average sentiment score per conversation. If escalation rate drifts >10% week-over-week, re-run calibration.

Final checklist before merging:

• Classifier latency p99 < 50 ms on target hardware
• Policy unit tests pass (threshold boundaries, streak reset)
• End-to-end integration test covers all four Action types
• Structured logs appear in your logging backend with correct schema
• Calibration script runs against labeled data and outputs chosen thresholds
• Rollback plan: feature flag to disable sentiment gating instantly

You now have a sentiment-aware support bot that escalates at the right moment, softens its tone when users are frustrated, and produces the data you need to keep improving both.