Balancing Embedding-Based Features with Classical Tabular Features in Enterprise AI Systems

Introduction

In modern enterprise AI applications, we often face a hybrid data landscape: structured tabular data (customer demographics, transaction amounts, timestamps) coexists with unstructured text (support tickets, product descriptions, user reviews). The challenge isn't choosing between embedding-based features and classical tabular features—it's integrating both effectively to build robust, production-grade systems.

This article explores how to balance these two feature types within an enterprise multi-agent LangGraph RAG system with memory and state management, using a real-world customer support automation scenario.

Real-Time Use Case: Intelligent Customer Support Triage System

Business Problem

A large e-commerce platform receives 10,000+ customer support tickets daily. Each ticket contains:

Tabular features: Customer tier (Gold/Silver/Bronze), order value, days since last purchase, product category ID, region code
Text features: Customer complaint description, product review excerpts, chat history transcripts

The business needs an intelligent triage system that:

Routes tickets to the right department (Billing, Shipping, Product Quality, Account Management)
Predicts escalation risk (Low/Medium/High)
Recommends resolution actions based on similar historical cases

Why Both Feature Types Matter

Feature Type	Strengths	Limitations
Classical Tabular	Interpretable, handles numerical relationships well, efficient for structured patterns	Cannot capture semantic meaning, struggles with free-text nuances
Embedding-Based	Captures semantic similarity, understands context and intent, works with unstructured data	Black-box nature, computationally expensive, requires careful dimensionality management

The Solution: Combine both in a unified pipeline where embeddings enrich tabular features, and tabular features provide grounding and interpretability.

Architecture Overview: Multi-Agent LangGraph RAG System

Implementation: End-to-End Code

Step 1: Setup and Dependencies

# requirements.txt"""
langgraph==0.2.0
langchain==0.3.0
langchain-openai==0.2.0
faiss-cpu==1.7.4
pydantic==2.5.0
scikit-learn==1.3.0
pandas==2.1.0
numpy==1.24.0
sentence-transformers==2.2.2
"""

import os
import json
import numpy as np
import pandas as pd
from typing import List, Dict, Any, Optional, Literalfrom datetime import datetime
from pydantic import BaseModel, Field, Annotated
from langgraph.graph import StateGraph, END
from langgraph.messages import add_messages
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sentence_transformers import SentenceTransformer
import faiss

Step 2: Define Data Models with Pydantic

class TicketMetadata(BaseModel):
    """Classical tabular features from the support ticket"""
    ticket_id: str = Field(description="Unique ticket identifier")
    customer_tier: Literal["Gold", "Silver", "Bronze"] = Field(
        description="Customer loyalty tier"
    )
    order_value: float = Field(gt=0, description="Total order value in USD")
    days_since_last_purchase: int = Field(ge=0, description="Days since customer's last purchase")
    product_category_id: int = Field(description="Numeric category identifier")
    region_code: int = Field(description="Geographic region code (1-10)")
    is_repeat_customer: bool = Field(description="Whether customer has previous tickets")
  
    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "ticket_id": "TKT-2026-89234",
                    "customer_tier": "Gold",
                    "order_value": 249.99,
                    "days_since_last_purchase": 15,
                    "product_category_id": 42,
                    "region_code": 3,
                    "is_repeat_customer": True
                }
            ]
        }


class TicketContent(BaseModel):
    """Unstructured text content from the support ticket"""
    subject: str = Field(description="Ticket subject line")
    description: str = Field(description="Detailed customer complaint or query")
    chat_history: Optional[str] = Field(None, description="Previous chat transcript if available")
    product_review_excerpt: Optional[str] = Field(None, description="Related product review text")


class CombinedTicket(BaseModel):
    """Unified ticket representation combining tabular and text features"""
    metadata: TicketMetadata
    content: TicketContent
    timestamp: datetime = Field(default_factory=datetime.now)
  
    def model_dump_clean(self) -> Dict:
        """Exclude internal fields for API responses"""
        return self.model_dump(exclude={"timestamp"})

Step 3: Feature Processing Pipeline

class FeatureProcessor:
    """Handles both tabular normalization and embedding generation"""
  
    def __init__(self):
        # Initialize embedding model (using sentence-transformers for efficiency)
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.embedding_dim = 384
      
        # Initialize scalers for tabular features
        self.tabular_scaler = StandardScaler()
        self.tier_encoder = LabelEncoder()
        self.tier_encoder.fit(["Bronze", "Silver", "Gold"])
      
        # FAISS index for vector storage
        self.index = None
        self.vector_store = []
        self.metadata_store = []
      
    def process_tabular_features(self, metadata: TicketMetadata) -> np.ndarray:
        """
        Normalize and encode classical tabular features
      
        Returns: Normalized feature vector [tier_encoded, order_value_scaled, 
                 days_since_purchase_scaled, category_id, region_code, is_repeat]
        """
        # Encode categorical features
        tier_encoded = self.tier_encoder.transform([metadata.customer_tier])[0]
      
        # Prepare raw features
        raw_features = np.array([
            tier_encoded,
            metadata.order_value,
            metadata.days_since_last_purchase,
            metadata.product_category_id,
            metadata.region_code,
            float(metadata.is_repeat_customer)
        ])
      
        # Note: In production, you'd fit the scaler on training data
        # For demo, we'll normalize manually
        normalized = np.array([
            tier_encoded / 2.0,  # Bronze=0, Silver=1, Gold=2
            min(metadata.order_value / 1000.0, 1.0),  # Cap at $1000
            min(metadata.days_since_last_purchase / 365.0, 1.0),  # Cap at 1 year
            metadata.product_category_id / 100.0,  # Normalize category
            metadata.region_code / 10.0,  # Normalize region
            float(metadata.is_repeat_customer)
        ])
      
        return normalized
  
    def generate_text_embeddings(self, content: TicketContent) -> np.ndarray:
        """
        Generate embeddings from unstructured text content
      
        Strategy: Combine subject + description, optionally include chat history
        """
        # Concatenate relevant text fields
        text_parts = [content.subject, content.description]
        if content.chat_history:
            text_parts.append(content.chat_history[:500])  # Limit length
        if content.product_review_excerpt:
            text_parts.append(content.product_review_excerpt)
      
        combined_text = " ".join(text_parts)
      
        # Generate embedding
        embedding = self.embedding_model.encode(combined_text)
      
        return embedding
  
    def fuse_features(self, tabular_vec: np.ndarray, embedding_vec: np.ndarray) -> np.ndarray:
        """
        Combine tabular and embedding features with weighted fusion
      
        Strategy: Simple concatenation with optional attention weighting
        In production, you might use a learned fusion layer
        """
        # Weight embeddings higher (they carry more semantic information)
        # but keep tabular features for interpretability
        weighted_embedding = embedding_vec * 0.7
        weighted_tabular = tabular_vec * 0.3
      
        # Concatenate into unified feature vector
        fused = np.concatenate([weighted_tabular, weighted_embedding])
      
        return fused
  
    def store_in_vector_db(self, fused_vector: np.ndarray, ticket_data: Dict):
        """Store processed ticket in FAISS index for RAG retrieval"""
        if self.index is None:
            # Initialize FAISS index
            dimension = fused_vector.shape[0]
            self.index = faiss.IndexFlatL2(dimension)
      
        # Add to index
        vector_reshaped = fused_vector.reshape(1, -1).astype('float32')
        self.index.add(vector_reshaped)
      
        # Store metadata separately
        self.vector_store.append(fused_vector)
        self.metadata_store.append(ticket_data)
  
    def retrieve_similar_tickets(self, query_vector: np.ndarray, k: int = 5) -> List[Dict]:
        """Retrieve k most similar historical tickets using combined features"""
        if self.index is None or self.index.ntotal == 0:
            return []
      
        query_reshaped = query_vector.reshape(1, -1).astype('float32')
        distances, indices = self.index.search(query_reshaped, k)
      
        results = []
        for idx in indices[0]:
            if idx < len(self.metadata_store):
                results.append({
                    "ticket": self.metadata_store[idx],
                    "similarity_score": float(1 / (1 + distances[0][list(indices[0]).index(idx)]))
                })
      
        return results

Step 4: Define LangGraph State

class TicketTriageState(BaseModel):
    """State object maintained throughout the LangGraph workflow"""
  
    # Input data
    ticket_metadata: Optional[TicketMetadata] = None
    ticket_content: Optional[TicketContent] = None
  
    # Processed features
    tabular_features: Optional[List[float]] = None
    text_embedding: Optional[List[float]] = None
    fused_features: Optional[List[float]] = None
  
    # RAG results
    similar_tickets: List[Dict] = Field(default_factory=list)
    retrieved_context: str = ""
  
    # Decision output
    routing_department: Optional[Literal["Billing", "Shipping", "Product Quality", "Account Management"]] = None
    escalation_risk: Optional[Literal["Low", "Medium", "High"]] = None
    recommended_actions: List[str] = Field(default_factory=list)
    confidence_score: float = 0.0
  
    # Memory and conversation tracking
    conversation_history: List[Dict] = Field(default_factory=list)
    long_term_memory: Dict[str, Any] = Field(default_factory=dict)
  
    # Metadata
    processing_timestamp: datetime = Field(default_factory=datetime.now)
    agent_logs: List[str] = Field(default_factory=list)
  
    class Config:
        arbitrary_types_allowed = True

Step 5: Build Multi-Agent LangGraph Workflow

class TicketTriageAgent:
    """Multi-agent system for intelligent ticket triage"""
  
    def __init__(self):
        self.feature_processor = FeatureProcessor()
        self.llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.1)
      
        # Initialize graph
        self.workflow = self._build_workflow()
  
    def _build_workflow(self) -> StateGraph:
        """Construct the LangGraph workflow with multiple specialized agents"""
      
        workflow = StateGraph(TicketTriageState)
      
        # Add nodes (agents)
        workflow.add_node("feature_extractor", self.extract_features)
        workflow.add_node("rag_retriever", self.retrieve_context)
        workflow.add_node("decision_maker", self.make_decision)
        workflow.add_node("memory_updater", self.update_memory)
      
        # Define edges
        workflow.set_entry_point("feature_extractor")
        workflow.add_edge("feature_extractor", "rag_retriever")
        workflow.add_edge("rag_retriever", "decision_maker")
        workflow.add_edge("decision_maker", "memory_updater")
        workflow.add_edge("memory_updater", END)
      
        return workflow.compile()
  
    def extract_features(self, state: TicketTriageState) -> Dict:
        """
        Agent 1: Feature Extraction
        Processes both tabular and text features, creates fused representation
        """
        print("🔧 [Feature Extractor] Processing ticket features...")
      
        # Process tabular features
        tabular_vec = self.feature_processor.process_tabular_features(state.ticket_metadata)
      
        # Generate text embeddings
        embedding_vec = self.feature_processor.generate_text_embeddings(state.ticket_content)
      
        # Fuse features
        fused_vec = self.feature_processor.fuse_features(tabular_vec, embedding_vec)
      
        # Log processing
        log_entry = f"Features extracted: tabular_dim={len(tabular_vec)}, embedding_dim={len(embedding_vec)}, fused_dim={len(fused_vec)}"
      
        return {
            "tabular_features": tabular_vec.tolist(),
            "text_embedding": embedding_vec.tolist(),
            "fused_features": fused_vec.tolist(),
            "agent_logs": state.agent_logs + [log_entry]
        }
  
    def retrieve_context(self, state: TicketTriageState) -> Dict:
        """
        Agent 2: RAG Retriever
        Uses fused features to find similar historical tickets
        """
        print("🔍 [RAG Retriever] Searching for similar historical cases...")
      
        fused_vec = np.array(state.fused_features)
      
        # Retrieve similar tickets
        similar_tickets = self.feature_processor.retrieve_similar_tickets(fused_vec, k=5)
      
        # Build context from retrieved tickets
        context_parts = []
        for item in similar_tickets:
            ticket_info = item["ticket"]
            score = item["similarity_score"]
            context_parts.append(
                f"Similar Case (similarity: {score:.2f}):\n"
                f"- Department: {ticket_info.get('department', 'Unknown')}\n"
                f"- Resolution: {ticket_info.get('resolution', 'N/A')}\n"
                f"- Escalation: {ticket_info.get('escalation_risk', 'Unknown')}\n"
            )
      
        retrieved_context = "\n".join(context_parts) if context_parts else "No similar cases found."
      
        log_entry = f"Retrieved {len(similar_tickets)} similar tickets from knowledge base"
      
        return {
            "similar_tickets": similar_tickets,
            "retrieved_context": retrieved_context,
            "agent_logs": state.agent_logs + [log_entry]
        }
  
    def make_decision(self, state: TicketTriageState) -> Dict:
        """
        Agent 3: Decision Maker
        Synthesizes features and retrieved context to make routing decision
        """
        print("🤖 [Decision Maker] Analyzing and making routing decision...")
      
        # Prepare prompt with all available information
        prompt = ChatPromptTemplate.from_template("""
        You are an expert customer support triage specialist.
      
        Current Ticket Information:
        - Customer Tier: {customer_tier}
        - Order Value: ${order_value}
        - Days Since Last Purchase: {days_since_purchase}
        - Subject: {subject}
        - Description: {description}
      
        Historical Context (similar past cases):
        {retrieved_context}
      
        Based on the ticket details and similar historical cases, determine:
        1. Which department should handle this ticket? (Billing, Shipping, Product Quality, Account Management)
        2. What is the escalation risk? (Low, Medium, High)
        3. What are 2-3 recommended actions?
        4. What is your confidence score (0-1)?
      
        Respond in JSON format:
        {{
            "routing_department": "...",
            "escalation_risk": "...",
            "recommended_actions": ["...", "..."],
            "confidence_score": 0.XX,
            "reasoning": "..."
        }}
        """)
      
        # Invoke LLM
        response = self.llm.invoke(prompt.format(
            customer_tier=state.ticket_metadata.customer_tier,
            order_value=state.ticket_metadata.order_value,
            days_since_purchase=state.ticket_metadata.days_since_last_purchase,
            subject=state.ticket_content.subject,
            description=state.ticket_content.description[:500],
            retrieved_context=state.retrieved_context
        ))
      
        # Parse response
        try:
            decision = json.loads(response.content)
        except:
            # Fallback if JSON parsing fails
            decision = {
                "routing_department": "Account Management",
                "escalation_risk": "Medium",
                "recommended_actions": ["Review ticket manually", "Contact customer for clarification"],
                "confidence_score": 0.5,
                "reasoning": "Fallback decision due to parsing error"
            }
      
        log_entry = f"Decision made: {decision['routing_department']} (confidence: {decision['confidence_score']})"
      
        return {
            "routing_department": decision["routing_department"],
            "escalation_risk": decision["escalation_risk"],
            "recommended_actions": decision["recommended_actions"],
            "confidence_score": decision["confidence_score"],
            "agent_logs": state.agent_logs + [log_entry]
        }
  
    def update_memory(self, state: TicketTriageState) -> Dict:
        """
        Agent 4: Memory Updater
        Stores outcome in long-term memory for future learning
        """
        print("💾 [Memory Updater] Updating knowledge base...")
      
        # Store current ticket in vector DB for future retrieval
        fused_vec = np.array(state.fused_features)
        ticket_data = {
            "ticket_id": state.ticket_metadata.ticket_id,
            "department": state.routing_department,
            "escalation_risk": state.escalation_risk,
            "resolution": "Pending",
            "timestamp": state.processing_timestamp.isoformat()
        }
      
        self.feature_processor.store_in_vector_db(fused_vec, ticket_data)
      
        # Update long-term memory with patterns
        key = f"{state.ticket_metadata.customer_tier}_{state.routing_department}"
        if key not in state.long_term_memory:
            state.long_term_memory[key] = []
        state.long_term_memory[key].append({
            "ticket_id": state.ticket_metadata.ticket_id,
            "outcome": state.routing_department,
            "timestamp": state.processing_timestamp.isoformat()
        })
      
        log_entry = "Memory updated: ticket stored in vector DB and long-term memory"
      
        return {
            "long_term_memory": state.long_term_memory,
            "agent_logs": state.agent_logs + [log_entry]
        }
  
    def process_ticket(self, metadata: TicketMetadata, content: TicketContent) -> Dict:
        """
        Main entry point: Process a support ticket through the multi-agent workflow
        """
        # Initialize state
        initial_state = TicketTriageState(
            ticket_metadata=metadata,
            ticket_content=content
        )
      
        # Run workflow
        final_state = self.workflow.invoke(initial_state)
      
        # Return clean result
        result = {
            "ticket_id": metadata.ticket_id,
            "routing_department": final_state.routing_department,
            "escalation_risk": final_state.escalation_risk,
            "recommended_actions": final_state.recommended_actions,
            "confidence_score": final_state.confidence_score,
            "processing_logs": final_state.agent_logs,
            "timestamp": final_state.processing_timestamp.isoformat()
        }
      
        return result

Step 6: Demo Usage

def main():
    """Demonstrate the end-to-end ticket triage system"""
  
    # Initialize the agent system
    triage_agent = TicketTriageAgent()
  
    # Create sample ticket
    sample_metadata = TicketMetadata(
        ticket_id="TKT-2026-89234",
        customer_tier="Gold",
        order_value=249.99,
        days_since_last_purchase=15,
        product_category_id=42,
        region_code=3,
        is_repeat_customer=True
    )
  
    sample_content = TicketContent(
        subject="Order not delivered after 2 weeks",
        description="I placed an order 2 weeks ago (Order #ORD-45892) and still haven't received it. The tracking shows it's stuck in transit. I'm a Gold member and expect better service. This is urgent as it was a gift.",
        chat_history="Customer: Where is my order?\nAgent: Let me check...\nCustomer: It's been 2 weeks!",
        product_review_excerpt="Great product but shipping was delayed last time too."
    )
  
    # Process ticket
    print("=" * 60)
    print("PROCESSING SUPPORT TICKET")
    print("=" * 60)
  
    result = triage_agent.process_ticket(sample_metadata, sample_content)
  
    # Display results
    print("\n📊 TRIAGE RESULTS:")
    print(f"Ticket ID: {result['ticket_id']}")
    print(f"Routing Department: {result['routing_department']}")
    print(f"Escalation Risk: {result['escalation_risk']}")
    print(f"Confidence Score: {result['confidence_score']:.2f}")
    print(f"\nRecommended Actions:")
    for i, action in enumerate(result['recommended_actions'], 1):
        print(f"  {i}. {action}")
  
    print(f"\nProcessing Logs:")
    for log in result['processing_logs']:
        print(f"  • {log}")
  
    # Process another ticket to demonstrate memory accumulation
    print("\n" + "=" * 60)
    print("PROCESSING SECOND TICKET (to show memory effect)")
    print("=" * 60)
  
    sample_metadata_2 = TicketMetadata(
        ticket_id="TKT-2026-89235",
        customer_tier="Silver",
        order_value=89.99,
        days_since_last_purchase=45,
        product_category_id=42,
        region_code=3,
        is_repeat_customer=False
    )
  
    sample_content_2 = TicketContent(
        subject="Damaged product received",
        description="The item arrived with visible damage to the packaging. The product itself seems fine but I'm concerned about quality control.",
    )
  
    result_2 = triage_agent.process_ticket(sample_metadata_2, sample_content_2)
  
    print(f"\n📊 TRIAGE RESULTS (Ticket 2):")
    print(f"Ticket ID: {result_2['ticket_id']}")
    print(f"Routing Department: {result_2['routing_department']}")
    print(f"Escalation Risk: {result_2['escalation_risk']}")
    print(f"Confidence Score: {result_2['confidence_score']:.2f}")


if __name__ == "__main__":
    main()

Key Design Decisions & Best Practices

1. Feature Balance Strategy

# The fusion weights can be tuned based on your domain
weighted_embedding = embedding_vec * 0.7  # 70% weight to semantic features
weighted_tabular = tabular_vec * 0.3      # 30% weight to structured features

Why this ratio?

Embeddings capture nuanced intent and context (critical for understanding complaints)
Tabular features provide grounding (customer value, urgency indicators)
Adjust based on your data: if text is noisy, increase tabular weight; if structured data is sparse, increase embedding weight

2. Memory Management

The system maintains two types of memory:

Short-term: Conversation history within the state object
Long-term: Vector database + pattern tracking in long_term_memory

This enables the system to improve over time as more tickets are processed.

3. Interpretability Through Hybrid Approach

By keeping tabular features separate before fusion, you can:

Explain decisions using interpretable features ("High escalation risk because customer is Gold tier + order value > $200")
Audit the system by examining which feature type contributed more to the decision
Debug issues by isolating whether problems stem from embedding quality or tabular preprocessing

4. Scalability Considerations

FAISS provides efficient similarity search even with millions of vectors
SentenceTransformers offers fast, lightweight embeddings suitable for production
LangGraph's state management enables distributed processing across multiple workers

Performance Metrics to Track

Metric	Target	Why It Matters
Routing Accuracy	>85%	Measures correct department assignment
Confidence Score Distribution	Mean >0.7	Indicates model certainty
Retrieval Relevance	Top-3 similarity >0.6	Ensures RAG finds useful cases
Processing Latency	<2 seconds per ticket	Critical for real-time triage
Memory Growth Rate	Linear, not exponential	Prevents storage bloat

Conclusion

Balancing embedding-based and classical tabular features isn't about choosing one over the other - it's about leveraging the strengths of both. In enterprise systems:

✅ Embeddings capture semantic meaning and enable flexible RAG retrieval

✅ Tabular features provide interpretability, efficiency, and grounding

✅ Fusion strategies combine them into unified representations

✅ Multi-agent architectures (like LangGraph) orchestrate complex workflows with memory and state

The key is designing your pipeline so each feature type plays to its strengths, with clear mechanisms for fusion, retrieval, and continuous learning through memory updates.