Introduction

In the rapidly evolving landscape of artificial intelligence, discriminative models stand as one of the most powerful and widely adopted approaches for solving real-world classification problems. Unlike generative models that learn to create new data samples, discriminative models focus on understanding the boundary between different classes by learning the conditional probability P(Y|X) – the probability of a label given the input features. This makes them exceptionally effective for tasks like spam detection, sentiment analysis, fraud identification, and medical diagnosis.

For enterprise applications, the challenge isn't just building a single model but orchestrating multiple specialized agents that can collaborate, maintain context through memory, and retrieve relevant knowledge using Graph RAG (Retrieval-Augmented Generation). Today, we'll build a complete proof-of-concept demonstrating how discriminative models integrate into an enterprise-grade multi-agent system using LangGraph, complete with state management, persistent memory, and intelligent retrieval capabilities.

Understanding Discriminative Models

Discriminative models directly model the decision boundary between classes. Popular examples include:

The key advantage? They require less data than generative models and excel when you need accurate predictions rather than data generation.

Real-Time Use Case: Enterprise Customer Support Ticket Classification

Imagine a large e-commerce platform receiving thousands of support tickets daily. Each ticket needs to be classified into categories like "Billing Issue," "Technical Problem," "Shipping Delay," or "Product Return." Misclassification leads to delayed responses and customer dissatisfaction. Our solution uses a multi-agent system where:

  1. Classifier Agent: Uses a discriminative model to categorize tickets

  2. Retriever Agent: Fetches similar historical cases using Graph RAG

  3. Resolver Agent: Suggests solutions based on classification and retrieved context

  4. Memory Manager: Maintains conversation state and learns from corrections

Step-by-Step Implementation

Step 1: Setting Up the Environment

# requirements.txt
langgraph==0.2.0
langchain==0.1.0
scikit-learn==1.3.0
fastapi==0.109.0
uvicorn==0.27.0
pydantic==2.5.0
chromadb==0.4.22
networkx==3.2.1

Step 2: Building the Discriminative Model

# models/classifier.py
import pickle
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
import numpy as np

class TicketClassifier:
    def __init__(self):
        self.pipeline = Pipeline([
            ('tfidf', TfidfVectorizer(max_features=5000, ngram_range=(1, 2))),
            ('classifier', RandomForestClassifier(n_estimators=100, random_state=42))
        ])
        self.categories = ["Billing", "Technical", "Shipping", "Returns"]
    
    def train(self, texts, labels):
        """Train the discriminative model"""
        self.pipeline.fit(texts, labels)
        self._save_model()
    
    def predict(self, text: str) -> dict:
        """Predict category with confidence scores"""
        prediction = self.pipeline.predict([text])[0]
        probabilities = self.pipeline.predict_proba([text])[0]
        
        return {
            "category": prediction,
            "confidence": float(max(probabilities)),
            "all_scores": {
                cat: float(score) 
                for cat, score in zip(self.categories, probabilities)
            }
        }
    
    def _save_model(self):
        with open('models/ticket_classifier.pkl', 'wb') as f:
            pickle.dump(self.pipeline, f)
    
    @staticmethod
    def load_model():
        with open('models/ticket_classifier.pkl', 'rb') as f:
            return pickle.load(f)

Step 3: Implementing Graph RAG with ChromaDB

# services/graph_rag.py
import chromadb
from chromadb.config import Settings
import networkx as nx
from typing import List, Dict

class GraphRAGService:
    def __init__(self):
        self.client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.client.get_or_create_collection("support_tickets")
        self.graph = nx.DiGraph()
    
    def add_document(self, doc_id: str, text: str, metadata: Dict):
        """Add document to vector store and graph"""
        self.collection.add(
            documents=[text],
            ids=[doc_id],
            metadatas=[metadata]
        )
        # Add to knowledge graph
        self.graph.add_node(doc_id, **metadata)
        if metadata.get('related_to'):
            self.graph.add_edge(doc_id, metadata['related_to'])
    
    def retrieve_similar(self, query: str, n_results: int = 3) -> List[Dict]:
        """Retrieve similar documents using vector similarity"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        return [{
            "id": results['ids'][0][i],
            "text": results['documents'][0][i],
            "metadata": results['metadatas'][0][i]
        } for i in range(len(results['ids'][0]))]
    
    def get_related_nodes(self, node_id: str) -> List[str]:
        """Get related nodes from knowledge graph"""
        if self.graph.has_node(node_id):
            return list(self.graph.neighbors(node_id))
        return []

Step 4: Creating the LangGraph Multi-Agent System

# agents/workflow.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional
from .classifier import TicketClassifier
from .graph_rag import GraphRAGService

class AgentState(TypedDict):
    ticket_text: str
    classification: Optional[dict]
    retrieved_context: List[dict]
    suggested_solution: Optional[str]
    conversation_history: List[dict]
    confidence_threshold: float

class SupportTicketAgent:
    def __init__(self):
        self.classifier = TicketClassifier.load_model()
        self.rag_service = GraphRAGService()
    
    def classify_ticket(self, state: AgentState) -> AgentState:
        """Classifier Agent Node"""
        result = self.classifier.predict(state['ticket_text'])
        state['classification'] = result
        
        # Route based on confidence
        if result['confidence'] < state.get('confidence_threshold', 0.7):
            state['needs_human_review'] = True
        
        return state
    
    def retrieve_context(self, state: AgentState) -> AgentState:
        """Retriever Agent Node"""
        similar_cases = self.rag_service.retrieve_similar(
            state['ticket_text'], 
            n_results=3
        )
        state['retrieved_context'] = similar_cases
        return state
    
    def generate_solution(self, state: AgentState) -> AgentState:
        """Resolver Agent Node"""
        category = state['classification']['category']
        context = state['retrieved_context']
        
        # Simple template-based solution generation
        solutions = {
            "Billing": "Our billing team will review your account within 24 hours.",
            "Technical": "Please try clearing your cache. If issue persists, our tech team will contact you.",
            "Shipping": "We'll track your shipment and provide an update within 2 hours.",
            "Returns": "You can initiate a return through your account dashboard."
        }
        
        state['suggested_solution'] = solutions.get(category, 
            "We'll escalate this to our specialist team.")
        
        # Update conversation history
        state['conversation_history'].append({
            "role": "assistant",
            "content": state['suggested_solution']
        })
        
        return state

def build_workflow():
    """Build the LangGraph workflow"""
    agent = SupportTicketAgent()
    
    workflow = StateGraph(AgentState)
    
    # Add nodes
    workflow.add_node("classify", agent.classify_ticket)
    workflow.add_node("retrieve", agent.retrieve_context)
    workflow.add_node("resolve", agent.generate_solution)
    
    # Define edges
    workflow.set_entry_point("classify")
    workflow.add_edge("classify", "retrieve")
    workflow.add_edge("retrieve", "resolve")
    workflow.add_edge("resolve", END)
    
    return workflow.compile()

Step 5: FastAPI Backend

# main.py
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
from agents.workflow import build_workflow

app = FastAPI(title="Enterprise Ticket Classifier")
workflow = build_workflow()

class TicketRequest(BaseModel):
    text: str
    conversation_history: List[dict] = []

class TicketResponse(BaseModel):
    category: str
    confidence: float
    solution: str
    similar_cases: List[dict]

@app.post("/classify", response_model=TicketResponse)
async def classify_ticket(request: TicketRequest):
    initial_state = {
        "ticket_text": request.text,
        "classification": None,
        "retrieved_context": [],
        "suggested_solution": None,
        "conversation_history": request.conversation_history,
        "confidence_threshold": 0.7
    }
    
    result = await workflow.ainvoke(initial_state)
    
    return TicketResponse(
        category=result['classification']['category'],
        confidence=result['classification']['confidence'],
        solution=result['suggested_solution'],
        similar_cases=result['retrieved_context']
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Step 6: Simple Frontend Interface

<!-- index.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Ticket Classifier</title>
    <style>
        body { font-family: Arial; max-width: 800px; margin: 50px auto; padding: 20px; }
        .result { background: #f0f0f0; padding: 15px; margin-top: 20px; border-radius: 5px; }
        button { background: #007bff; color: white; padding: 10px 20px; border: none; cursor: pointer; }
    </style>
</head>
<body>
    <h1>Enterprise Support Ticket Classifier</h1>
    <textarea id="ticketText" rows="5" cols="60" placeholder="Enter ticket description..."></textarea>
    <br><br>
    <button onclick="classifyTicket()">Classify Ticket</button>
    <div id="result" class="result" style="display:none;"></div>

    <script>
        async function classifyTicket() {
            const text = document.getElementById('ticketText').value;
            const response = await fetch('/classify', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({text: text})
            });
            const data = await response.json();
            
            document.getElementById('result').innerHTML = `
                <h3>Classification: ${data.category}</h3>
                <p>Confidence: ${(data.confidence * 100).toFixed(2)}%</p>
                <p>Solution: ${data.solution}</p>
                <h4>Similar Cases:</h4>
                <ul>${data.similar_cases.map(c => `<li>${c.text.substring(0, 100)}...</li>`).join('')}</ul>
            `;
            document.getElementById('result').style.display = 'block';
        }
    </script>
</body>
</html>

Conclusion

This proof-of-concept demonstrates how discriminative models can be effectively integrated into enterprise multi-agent systems using LangGraph. The combination of accurate classification, Graph RAG for contextual retrieval, and stateful memory management creates a robust solution for real-world problems like customer support automation.

Key takeaways:

This architecture is scalable, maintainable, and ready for production deployment. By combining traditional ML with modern AI orchestration tools, enterprises can build intelligent systems that truly understand and respond to complex business needs.