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
![457]()
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:
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.