What is a Customer Support Bot with LangChain?
A customer support bot built with LangChain is an intelligent conversational agent that leverages large language models (LLMs) to understand, process, and resolve customer inquiries. LangChain provides the orchestration layer that connects the LLM to your company's knowledge base, APIs, databases, and business logic, creating a support system that goes far beyond simple FAQ matching.
Unlike traditional rule-based chatbots that follow rigid decision trees, a LangChain-powered bot uses chain-of-thought reasoning, semantic search over documents, and tool-calling capabilities to handle complex, multi-step customer issues. It can look up order statuses, initiate returns, check inventory, and escalate to human agents—all while maintaining context across the conversation.
Core Architecture Overview
At its heart, a LangChain customer support bot consists of several interconnected components:
- LLM Provider: OpenAI (GPT-4o, GPT-4), Anthropic (Claude), or open-source models via Ollama/HuggingFace
- Retrieval System: Vector store with embeddings for semantic search over knowledge base articles, product documentation, and past tickets
- Tool Framework: Functions the bot can call—order lookup, refund processing, inventory checks, CRM integration
- Memory Layer: Conversation history and session state management
- Routing & Guardrails: Intent classification, content moderation, and escalation logic
Why Building a LangChain Support Bot Matters
Customer support is undergoing a fundamental transformation. Organizations that deploy intelligent support bots see measurable impact across multiple dimensions:
Business Impact
- 24/7 Availability: Resolve customer issues at any hour without staffing overnight shifts
- Reduced Cost Per Ticket: Automate Tier-1 and Tier-2 support, saving an average of $5–$12 per resolved conversation
- Faster Resolution Times: Eliminate queue wait times—bots respond in seconds
- Scalability: Handle thousands of simultaneous conversations without degradation
Technical Advantages of LangChain Specifically
- Model Agnosticism: Swap between OpenAI, Anthropic, Cohere, or open-source models with minimal code changes
- Composable Chains: Build modular workflows—classification → retrieval → generation → tool execution
- Built-in Observability: LangSmith integration provides tracing, evaluation, and debugging
- Ecosystem Integrations: Pre-built connectors for Zendesk, Salesforce, Stripe, Shopify, Slack, and more
Project Setup and Dependencies
Let's start by setting up a complete project environment. Create a new directory and initialize your Python project:
# Create project directory
mkdir customer-support-bot
cd customer-support-bot
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install core dependencies
pip install langchain langchain-openai langchain-community
pip install chromadb tiktoken sentence-transformers
pip install python-dotenv fastapi uvicorn pydantic
pip install langsmith # For observability and tracingCreate a .env file for your API keys and configuration:
# .env file
OPENAI_API_KEY=your-openai-api-key-here
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your-langsmith-key-here
LANGCHAIN_PROJECT=customer-support-bot
# Optional: for production, add other provider keys
ANTHROPIC_API_KEY=your-anthropic-key
COHERE_API_KEY=your-cohere-keyBuilding the Knowledge Base with Document Loading and Chunking
The foundation of any effective support bot is a rich knowledge base. You'll load your support documentation, product manuals, FAQ pages, and historical ticket resolutions, then index them for semantic search.
Loading Documents from Multiple Sources
from langchain_community.document_loaders import (
TextLoader,
CSVLoader,
PyPDFLoader,
WebBaseLoader,
UnstructuredMarkdownLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pathlib import Path
# Load documents from various sources
documents = []
# 1. Load FAQ markdown files
faq_directory = Path("./knowledge_base/faqs")
for md_file in faq_directory.glob("*.md"):
loader = UnstructuredMarkdownLoader(str(md_file))
documents.extend(loader.load())
# 2. Load product documentation PDFs
pdf_directory = Path("./knowledge_base/product_docs")
for pdf_file in pdf_directory.glob("*.pdf"):
loader = PyPDFLoader(str(pdf_file))
documents.extend(loader.load())
# 3. Load support ticket history CSV
ticket_loader = CSVLoader(
file_path="./knowledge_base/resolved_tickets.csv",
source_column="ticket_description",
)
documents.extend(ticket_loader.load())
# 4. Load policy pages from company website
web_loader = WebBaseLoader([
"https://docs.yourcompany.com/returns",
"https://docs.yourcompany.com/shipping",
"https://docs.yourcompany.com/warranty",
])
documents.extend(web_loader.load())
print(f"Loaded {len(documents)} documents")Strategic Chunking for Support Content
Chunking strategy dramatically impacts retrieval quality. For support content, you need chunks that preserve complete problem-solution pairs:
from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
MarkdownHeaderTextSplitter,
)
# Primary splitter: respect markdown structure
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "h1_title"),
("##", "h2_section"),
("###", "h3_topic"),
],
strip_headers=False,
)
# Secondary splitter for oversized sections
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=800, # Larger chunks to preserve context
chunk_overlap=150, # Overlap prevents mid-concept splits
separators=["\n\n", "\n", ". ", "? ", "! ", " ", ""],
length_function=len,
add_start_index=True,
)
# Process documents
split_docs = []
for doc in documents:
# First, split by markdown headers
md_splits = markdown_splitter.split_text(doc.page_content)
for split in md_splits:
# Then split oversized sections
if len(split.page_content) > 1000:
sub_splits = text_splitter.split_documents([split])
split_docs.extend(sub_splits)
else:
split_docs.append(split)
print(f"Created {len(split_docs)} chunks after splitting")Creating the Vector Store with Embeddings
Now embed your chunks into a vector database for fast semantic retrieval:
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
import chromadb
# Initialize embeddings model
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small", # Cost-effective, high quality
dimensions=1536,
)
# Set up persistent Chroma client
chroma_client = chromadb.PersistentClient(
path="./vector_store/customer_support_db"
)
# Create collection with metadata filtering capability
collection = chroma_client.get_or_create_collection(
name="support_knowledge",
metadata={"hnsw:space": "cosine"},
)
# Create vector store
vector_store = Chroma(
client=chroma_client,
collection_name="support_knowledge",
embedding_function=embeddings,
)
# Batch insert with metadata for filtering
batch_size = 100
for i in range(0, len(split_docs), batch_size):
batch = split_docs[i:i + batch_size]
vector_store.add_documents(
documents=batch,
ids=[f"doc_{j}" for j in range(i, i + len(batch))],
)
print(f"Inserted batch {i//batch_size + 1}, total: {i+len(batch)}")
print(f"Vector store ready with {vector_store._collection.count()} embeddings")Building the Core Retrieval Chain
With the vector store populated, build a retrieval-augmented generation (RAG) chain that forms the backbone of your support bot:
from langchain_openai import ChatOpenAI
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainFilter
# Initialize the LLM
llm = ChatOpenAI(
model="gpt-4o",
temperature=0.3, # Lower temperature for factual consistency
max_tokens=1000,
)
# Create a sophisticated retriever with relevance filtering
base_retriever = vector_store.as_retriever(
search_type="similarity",
search_kwargs={
"k": 8, # Retrieve 8 candidates
"filter": None, # Add metadata filters if needed
},
)
# Add contextual compression to filter irrelevant chunks
compressor = LLMChainFilter.from_llm(llm)
compressed_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever,
)
# Define the support-specific prompt template
support_system_prompt = """You are a helpful customer support agent for OurCompany.
Use the following retrieved context to answer the customer's question.
Follow these rules strictly:
1. If the context contains the answer, provide it clearly and concisely.
2. If the context is partial, acknowledge gaps and offer to escalate.
3. If the context is irrelevant, say: "I don't have that information in my knowledge base.
Let me connect you with a human agent."
4. Always maintain a friendly, empathetic tone.
5. Never fabricate policies, prices, or procedures.
6. When quoting policies, cite the source document section.
Retrieved context:
{context}
Customer query: {input}
"""
rag_prompt = ChatPromptTemplate.from_messages([
("system", support_system_prompt),
("human", "{input}"),
])
# Build the document combination chain
combine_docs_chain = create_stuff_documents_chain(
llm=llm,
prompt=rag_prompt,
)
# Assemble the full retrieval chain
retrieval_chain = create_retrieval_chain(
retriever=compressed_retriever,
combine_docs_chain=combine_docs_chain,
)
# Test the chain
response = retrieval_chain.invoke({
"input": "What is your return policy for electronics purchased online?"
})
print(response["answer"])Adding Conversation Memory and Session Management
Customer support conversations often span multiple exchanges. You need persistent memory that tracks the full dialogue history and any actions taken:
from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.messages import HumanMessage, AIMessage
from datetime import datetime, timedelta
import json
class CustomerSessionManager:
"""Manages per-customer conversation sessions with TTL-based expiry."""
def __init__(self, llm, vector_store, ttl_minutes=30):
self.llm = llm
self.vector_store = vector_store
self.ttl_minutes = ttl_minutes
self.sessions = {} # In production, use Redis or a database
def get_or_create_session(self, session_id: str, customer_info: dict = None):
"""Retrieve existing session or create a new one."""
now = datetime.now()
if session_id in self.sessions:
session = self.sessions[session_id]
# Check if session has expired
if (now - session["last_accessed"]) < timedelta(minutes=self.ttl_minutes):
session["last_accessed"] = now
return session
else:
# Session expired, create new one
del self.sessions[session_id]
# Create fresh session
memory = ConversationSummaryBufferMemory(
llm=self.llm,
max_token_limit=2000, # Summarize when history exceeds this
return_messages=True,
memory_key="chat_history",
)
session = {
"session_id": session_id,
"memory": memory,
"created_at": now,
"last_accessed": now,
"customer_info": customer_info or {},
"escalation_count": 0,
"resolved_issues": [],
}
self.sessions[session_id] = session
return session
def add_escalation(self, session_id: str):
"""Track escalations to detect problematic sessions."""
if session_id in self.sessions:
self.sessions[session_id]["escalation_count"] += 1
def mark_resolved(self, session_id: str, issue: str):
"""Record resolved issues for analytics."""
if session_id in self.sessions:
self.sessions[session_id]["resolved_issues"].append({
"issue": issue,
"resolved_at": datetime.now().isoformat(),
})
# Initialize session manager
session_manager = CustomerSessionManager(llm, vector_store)Memory-Integrated Chat Chain
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
def build_memory_aware_chain(retrieval_chain, session_manager):
"""Wrap the retrieval chain with memory injection."""
def inject_memory(inputs: dict):
session_id = inputs.get("session_id", "default")
session = session_manager.get_or_create_session(session_id)
# Load conversation history
memory = session["memory"]
history = memory.load_memory_variables({})
chat_history = history.get("chat_history", [])
# Format history for the prompt
history_text = "\n".join([
f"{'Customer' if isinstance(m, HumanMessage) else 'Agent'}: {m.content}"
for m in chat_history[-10:] # Last 10 exchanges
])
# Augment the input with history context
augmented_input = f"""Previous conversation:
{history_text}
Current customer message: {inputs['input']}"""
return {"input": augmented_input, "session_id": session_id}
def store_interaction(outputs: dict):
session_id = outputs.get("session_id", "default")
session = session_manager.get_or_create_session(session_id)
memory = session["memory"]
# Store the exchange in memory
memory.save_context(
{"input": outputs.get("original_input", "")},
{"output": outputs.get("answer", "")}
)
return outputs
# Build the full chain
full_chain = (
RunnableLambda(inject_memory)
| retrieval_chain
| RunnableLambda(lambda x: {
**x,
"original_input": x.get("input", ""),
"session_id": x.get("session_id", "default"),
})
| RunnableLambda(store_interaction)
)
return full_chain
# Create the memory-aware chain
memory_chain = build_memory_aware_chain(retrieval_chain, session_manager)
# Test conversation continuity
response1 = memory_chain.invoke({
"input": "I need to return a laptop I bought last week.",
"session_id": "customer_42",
})
print("Agent:", response1["answer"])
response2 = memory_chain.invoke({
"input": "Yes, I have the order number. It's #ORD-98765.",
"session_id": "customer_42",
})
print("Agent:", response2["answer"])Building the Tool Framework for Action Execution
The true power of a LangChain support bot comes from its ability to take action—looking up orders, processing returns, checking inventory. Here's how to build a comprehensive tool suite:
from langchain.tools import Tool, tool, StructuredTool
from langchain_core.tools import ToolException
from pydantic import BaseModel, Field
from typing import Optional, List
import json
import httpx
# ---- Tool Input Schemas ----
class OrderLookupInput(BaseModel):
order_id: str = Field(description="The order ID (e.g., ORD-12345)")
customer_email: Optional[str] = Field(
default=None,
description="Customer email for verification"
)
class ReturnInitiateInput(BaseModel):
order_id: str = Field(description="Order ID for the return")
item_sku: str = Field(description="SKU of the item to return")
reason: str = Field(description="Return reason code: DEFECTIVE, WRONG_ITEM, CHANGED_MIND, DAMAGED")
customer_notes: Optional[str] = Field(default="", description="Additional customer notes")
class InventoryCheckInput(BaseModel):
sku: str = Field(description="Product SKU to check")
warehouse_id: Optional[str] = Field(default="MAIN", description="Warehouse identifier")
# ---- Tool Implementations ----
@tool("order_lookup", args_schema=OrderLookupInput)
def lookup_order(order_id: str, customer_email: str = None) -> str:
"""
Look up order details including status, items, and shipping information.
Use this when a customer asks about their order status or details.
"""
# In production, this calls your order management system API
# Simulated response for demonstration
orders_db = {
"ORD-98765": {
"status": "DELIVERED",
"order_date": "2024-12-15",
"delivery_date": "2024-12-22",
"items": [
{"sku": "LAPTOP-XPS-15", "name": "XPS 15 Laptop", "price": 1299.99},
{"sku": "CASE-LEATHER", "name": "Leather Laptop Case", "price": 49.99},
],
"total": 1349.98,
"eligible_return": True,
"return_deadline": "2025-01-22",
},
"ORD-12345": {
"status": "SHIPPED",
"order_date": "2024-12-18",
"estimated_delivery": "2024-12-26",
"items": [
{"sku": "HEADPHONES-PRO", "name": "Pro Headphones", "price": 249.99},
],
"total": 249.99,
"eligible_return": False,
},
}
order = orders_db.get(order_id)
if not order:
return f"Order {order_id} not found. Please verify the order ID and try again."
return json.dumps(order, indent=2)
@tool("initiate_return", args_schema=ReturnInitiateInput)
def initiate_return(order_id: str, item_sku: str, reason: str, customer_notes: str = "") -> str:
"""
Initiate a return for an item in an order.
Use this when a customer wants to return a product.
Requires the order ID, item SKU, and a valid reason code.
"""
valid_reasons = ["DEFECTIVE", "WRONG_ITEM", "CHANGED_MIND", "DAMAGED"]
if reason.upper() not in valid_reasons:
return f"Invalid reason code. Must be one of: {', '.join(valid_reasons)}"
# In production, this calls your returns management API
# Generate return authorization
import random
rma_number = f"RMA-{random.randint(100000, 999999)}"
return json.dumps({
"rma_number": rma_number,
"status": "INITIATED",
"order_id": order_id,
"item_sku": item_sku,
"reason": reason,
"next_steps": [
f"Print return label for RMA {rma_number}",
"Package item securely with original accessories",
"Drop off at nearest carrier location within 14 days",
],
"refund_timeline": "Refund processed within 5-7 business days after carrier scan",
}, indent=2)
@tool("check_inventory", args_schema=InventoryCheckInput)
def check_inventory(sku: str, warehouse_id: str = "MAIN") -> str:
"""
Check current inventory levels for a product SKU.
Use this when customers ask about product availability or stock status.
"""
# Simulated inventory database
inventory_db = {
"LAPTOP-XPS-15": {
"warehouses": {
"MAIN": 42,
"EAST": 15,
"WEST": 8,
},
"status": "IN_STOCK",
"next_restock": None,
},
"HEADPHONES-PRO": {
"warehouses": {
"MAIN": 0,
"EAST": 3,
"WEST": 0,
},
"status": "LOW_STOCK",
"next_restock": "2025-01-10",
},
}
item = inventory_db.get(sku)
if not item:
return f"SKU {sku} not found in inventory system."
wh_qty = item["warehouses"].get(warehouse_id, 0)
return json.dumps({
"sku": sku,
"status": item["status"],
"warehouse_quantity": wh_qty,
"total_available": sum(item["warehouses"].values()),
"next_restock": item.get("next_restock"),
}, indent=2)
@tool("escalate_to_human", return_direct=True)
def escalate_to_human(summary: str) -> str:
"""
Escalate the conversation to a human agent.
Use this when the bot cannot resolve the issue, the customer explicitly
requests a human, or the issue requires manual intervention.
Provide a concise summary of the conversation.
"""
return json.dumps({
"action": "ESCALATE",
"status": "TRANSFERRING",
"summary": summary,
"estimated_wait": "2-5 minutes",
"ticket_id": f"TKT-{hash(summary) % 1000000:06d}",
}, indent=2)
# Assemble the tool collection
support_tools = [
lookup_order,
initiate_return,
check_inventory,
escalate_to_human,
]
print(f"Registered {len(support_tools)} support tools")Building the Agent with Tool-Calling and Routing
Now assemble a full agent that can reason about when to use tools, retrieve knowledge, or escalate:
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.callbacks import StdOutCallbackHandler
# Define the agent's system prompt with routing logic
agent_system_prompt = """You are Aurora, a senior customer support agent for OurCompany.
You have access to knowledge base articles and several tools.
Your workflow:
1. Greet the customer warmly and understand their issue.
2. For order-related questions, use the order_lookup tool.
3. For return requests, first verify the order with order_lookup, then use initiate_return.
4. For product availability, use check_inventory.
5. For general policy questions, rely on your knowledge base context.
6. If you cannot resolve the issue after 2 attempts, or if the customer asks for a human,
use escalate_to_human with a detailed summary.
IMPORTANT RULES:
- Always verify order eligibility before initiating returns.
- Never promise refund amounts—defer to the returns system.
- If a tool fails, try an alternative approach once before escalating.
- Be empathetic: acknowledge frustration, apologize for issues.
- Protect customer data: never expose raw system responses with PII.
"""
# Create the agent prompt
agent_prompt = ChatPromptTemplate.from_messages([
("system", agent_system_prompt),
("system", "You have access to the following context from the knowledge base:\n{context}"),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# Initialize the LLM with tool-calling capabilities
agent_llm = ChatOpenAI(
model="gpt-4o",
temperature=0.2,
max_tokens=2000,
)
# Create the OpenAI tools agent
agent = create_openai_tools_agent(
llm=agent_llm,
tools=support_tools,
prompt=agent_prompt,
)
# Build the agent executor
agent_executor = AgentExecutor(
agent=agent,
tools=support_tools,
verbose=True,
max_iterations=6, # Prevent infinite loops
max_execution_time=30, # 30-second timeout
handle_parsing_errors=True,
return_intermediate_steps=False,
)
print("Agent executor ready for operation")Integrating Retrieval with the Agent
def build_full_support_agent(agent_executor, retrieval_chain, session_manager):
"""
Combine the agent with retrieval and memory for a complete support system.
"""
def retrieve_and_agent(inputs: dict):
session_id = inputs.get("session_id", "default")
customer_query = inputs["input"]
# First, retrieve relevant knowledge
retrieval_result = retrieval_chain.invoke({"input": customer_query})
context = retrieval_result.get("answer", "")
# Get session memory
session = session_manager.get_or_create_session(session_id)
memory = session["memory"]
history = memory.load_memory_variables({})
chat_history = history.get("chat_history", [])
# Run the agent with context and history
result = agent_executor.invoke({
"input": customer_query,
"context": context,
"chat_history": chat_history,
})
# Store the interaction
memory.save_context(
{"input": customer_query},
{"output": result.get("output", "")}
)
# Check for escalation and update session
if "ESCALATE" in result.get("output", ""):
session_manager.add_escalation(session_id)
return result
return retrieve_and_agent
# Build the complete agent
full_support_agent = build_full_support_agent(
agent_executor,
retrieval_chain,
session_manager,
)
# Test with a complex scenario
test_result = full_support_agent({
"input": "Hi, I received my order ORD-98765 yesterday but the laptop has a dead pixel cluster. "
"I want to return just the laptop and get a replacement. What are my options?",
"session_id": "customer_42",
})
print("\nFinal Response:", test_result.get("output", "No output"))Implementing Intent Classification and Intelligent Routing
Before the agent engages, classify the customer's intent to route them efficiently and apply the right business logic:
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional
class IntentCategory(str, Enum):
ORDER_STATUS = "order_status"
RETURN_REQUEST = "return_request"
PRODUCT_INQUIRY = "product_inquiry"
TECHNICAL_SUPPORT = "technical_support"
BILLING_ISSUE = "billing_issue"
POLICY_QUESTION = "policy_question"
COMPLAINT = "complaint"
GENERAL = "general"
class IntentClassification(BaseModel):
primary_intent: IntentCategory = Field(description="Primary customer intent")
confidence: float = Field(description="Confidence score 0-1", ge=0, le=1)
urgency: str = Field(description="LOW, MEDIUM, or HIGH")
requires_escalation: bool = Field(description="True if this likely needs a human")
summary: str = Field(description="One-line summary of the customer's issue")
# Create the intent classifier
intent_parser = PydanticOutputParser(pydantic_object=IntentClassification)
intent_prompt = PromptTemplate(
template="""Analyze this customer message and classify the intent.
Customer message: {message}
{format_instructions}
Classify carefully. If the customer is angry or mentions legal action, mark as HIGH urgency.""",
input_variables=["message"],
partial_variables={
"format_instructions": intent_parser.get_format_instructions(),
},
)
intent_chain = intent_prompt | llm | intent_parser
# Test classification
test_messages = [
"Where is my order? I placed it a week ago and haven't heard anything.",
"The laptop you sold me is defective. The screen flickers constantly. I want my money back NOW.",
"What's the battery life of the XPS 15 model?",
]
for msg in test_messages:
result = intent_chain.invoke({"message": msg})
print(f"Message: '{msg[:60]}...'")
print(f" Intent: {result.primary_intent}, Urgency: {result.urgency}")
print(f" Escalate: {result.requires_escalation}\n")Router Implementation
from typing import Dict, Callable
class SupportRouter:
"""Routes customer queries based on intent classification."""
def __init__(self, agent_executor, retrieval_chain, session_manager):
self.agent_executor = agent_executor
self.retrieval_chain = retrieval_chain
self.session_manager = session_manager
# Define routing rules
self.routing_map: Dict[IntentCategory, Callable] = {
IntentCategory.ORDER_STATUS: self.handle_order_status,
IntentCategory.RETURN_REQUEST: self.handle_return_request,
IntentCategory.PRODUCT_INQUIRY: self.handle_product_inquiry,
IntentCategory.TECHNICAL_SUPPORT: self.handle_technical_support,
IntentCategory.BILLING_ISSUE: self.handle_billing_issue,
IntentCategory.POLICY_QUESTION: self.handle_policy_question,
IntentCategory.COMPLAINT: self.handle_complaint,
IntentCategory.GENERAL: self.handle_general,
}
def route(self, message: str, session_id: str, customer_info: dict = None):
"""Classify intent and route to appropriate handler."""
# Classify the intent
classification = intent_chain.invoke({"message": message})
# Immediate escalation for high-urgency complaints
if classification.urgency == "HIGH" and classification.primary_intent == IntentCategory.COMPLAINT:
return self.immediate_escalation(message, classification, session_id)
# Route to handler
handler = self.routing_map.get(
classification.primary_intent,
self.handle_general,
)
return handler(message, session_id, classification, customer_info)
def handle_order_status(self, message, session_id, classification, customer_info):
"""Order status queries get priority tool access."""
return self.agent_executor.invoke({
"input": message,
"context": f"Customer wants order status information.",
"chat_history": self._get_history(session_id),
})
def handle_return_request(self, message, session_id, classification, customer_info):
"""Return requests require verification steps."""
session = self.session_manager.get_or_create_session(session_id)
context = self.retrieval_chain.invoke({"input": "return policy requirements"})
return self.agent_executor.invoke({
"input": f"{message}\n\nIMPORTANT: Verify order eligibility before processing return. "
f"Follow return policy from context.",
"context": context.get("answer", ""),
"chat_history": self._get_history(session_id),
})
def handle_complaint(self, message, session_id, classification, customer_info):
"""Complaints get empathetic handling with escalation readiness."""
return self.agent_executor.invoke({
"input": f"{message}\n\nThis customer may be frustrated. Be extra empathetic. "
f"If resolution isn't immediate, escalate.",
"context": "Handle complaint with priority.",
"chat_history": self._get_history(session_id),
})
def handle_policy_question(self, message, session_id, classification, customer_info):
"""Policy questions rely heavily on the knowledge base."""
context = self.retrieval_chain.invoke({"input": message})
return self.agent_executor.invoke({
"input": message,
"context": context.get("answer", ""),
"chat_history": self._get_history(session_id),
})
def handle_product_inquiry(self, message, session_id, classification, customer_info):
return self.agent_executor.invoke({
"input": message,
"context": "",
"chat_history": self._get_history(session_id),
})
def handle_technical_support(self, message, session_id, classification, customer_info):
context = self.retrieval_chain.invoke({"input": message})
return self.agent_executor.invoke({
"input": message,
"context": context.get("answer", ""),
"chat_history": self._get_history(session_id),
})
def handle_billing_issue(self, message, session_id, classification, customer_info):
"""Billing issues often need human intervention."""
return self.agent_executor.invoke({
"input": f"{message}\n\nIf billing details are involved, recommend escalation "
f"for security verification.",
"context": "",
"chat_history": self._get_history(session_id),
})
def handle_general(self, message, session_id, classification, customer_info):
return self.agent_executor.invoke({
"input": message,
"context": "",
"chat_history": self._get_history(session_id),
})
def immediate_escalation(self, message, classification, session_id):
"""Bypass agent for urgent complaints."""
return {
"output