1 week ago
ARTICLE AD BOX
This article demonstrates really to build an intelligent routing strategy powered by Anthropic’s Claude models. This strategy improves consequence ratio and value by automatically classifying personification requests and directing them to specialised handlers. The workflow analyses incoming queries, determines their intent, and routes them to due processing pipelines—whether for customer support, method assistance, aliases different domain-specific responses.
Step 1: Install nan required Python packages
!pip instal anthropic pandas scikit-learn
Step 2: Import nan basal libraries for nan project
import os
import json
import time
import pandas arsenic pd
import numpy arsenic np
from anthropic import Anthropic
from IPython.display import display, Markdown
from sklearn.metrics import classification_report
Step 3: Set up nan Anthropic API authentication by defining your API cardinal and initialising nan Anthropic client
ANTHROPIC_API_KEY = "{Your API KEY}"
client = Anthropic(api_key=ANTHROPIC_API_KEY)
Step 4: Create a sample dataset of customer queries pinch associated categories for training and testing nan routing system.
customer_queries = [
{"id": 1, "query": "What are your business hours?", "category": "General Question"},
{"id": 2, "query": "How do I reset my password?", "category": "Technical Support"},
{"id": 3, "query": "I want a refund for my purchase.", "category": "Refund Request"},
{"id": 4, "query": "Where tin I find your privateness policy?", "category": "General Question"},
{"id": 5, "query": "The app keeps crashing erstwhile I effort to upload photos.", "category": "Technical Support"},
{"id": 6, "query": "I ordered nan incorrect size, tin I get my money back?", "category": "Refund Request"},
{"id": 7, "query": "Do you vessel internationally?", "category": "General Question"},
{"id": 8, "query": "My relationship is showing incorrect information.", "category": "Technical Support"},
{"id": 9, "query": "I was charged doubly for my order.", "category": "Refund Request"},
{"id": 10, "query": "What costs methods do you accept?", "category": "General Question"}
]
Step 5: Convert nan customer queries database into a pandas DataFrame for easier manipulation and analysis. Then, show nan DataFrame successful nan notebook to visualise nan training dataset structure.
df = pd.DataFrame(customer_queries)
display(df)
Step 6: Define nan halfway routing usability that uses Claude 3.7 Sonnet to categorize customer queries into predefined categories.
def route_query(query, client):
"""
Route a customer query to nan due class utilizing Claude 3.5 Haiku.
Args:
query (str): The customer query to classify
client: Anthropic client
Returns:
str: The classified category
"""
system_prompt = """
You are a query classifier for a customer work system.
Your occupation is to categorize customer queries into precisely 1 of these categories:
1. General Question - Basic inquiries astir nan company, products, policies, etc.
2. Refund Request - Any query related to refunds, returns, aliases billing issues
3. Technical Support - Questions astir method problems, bugs, aliases really to usage products
Respond pinch ONLY nan class name, thing else.
"""
try:
consequence = client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
class = response.content[0].text.strip()
valid_categories = ["General Question", "Refund Request", "Technical Support"]
for valid_cat successful valid_categories:
if valid_cat.lower() successful category.lower():
return valid_cat
return "General Question"
isolated from Exception arsenic e:
print(f"Error successful routing: {e}")
return "General Question"
Step 7: Define 3 specialised handler functions for each query category, each utilizing Claude 3.5 Sonnet pinch a category-specific strategy prompt.
def handle_general_question(query, client):
"""Handle wide inquiries utilizing Claude 3.5 Haiku."""
system_prompt = """
You are a customer work typical answering wide questions astir our company.
Be helpful, concise, and friendly. Provide nonstop answers to customer queries.
"""
try:
consequence = client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
return response.content[0].text.strip()
isolated from Exception arsenic e:
print(f"Error successful wide mobility handler: {e}")
return "I apologize, but I'm having problem processing your request. Please effort again later."
def handle_refund_request(query, client):
"""Handle refund requests utilizing Claude 3.5 Sonnet for much nuanced responses."""
system_prompt = """
You are a customer work typical specializing successful refunds and billing issues.
Respond to refund requests professionally and helpfully.
For immoderate refund request, explicate nan refund argumentation intelligibly and supply adjacent steps.
Be empathetic but travel institution policy.
"""
try:
consequence = client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
return response.content[0].text.strip()
isolated from Exception arsenic e:
print(f"Error successful refund petition handler: {e}")
return "I apologize, but I'm having problem processing your refund request. Please interaction our support squad directly."
def handle_technical_support(query, client):
"""Handle method support queries utilizing Claude 3.5 Sonnet for much elaborate method responses."""
system_prompt = """
You are a method support specialist.
Provide clear, step-by-step solutions to method problems.
If you request much accusation to resoluteness an issue, specify what accusation you need.
Prioritize elemental solutions first earlier suggesting analyzable troubleshooting.
"""
try:
consequence = client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
return response.content[0].text.strip()
isolated from Exception arsenic e:
print(f"Error successful method support handler: {e}")
return "I apologize, but I'm having problem processing your method support request. Please effort our knowledge guidelines aliases interaction our support team."
Step 8: Create nan main workflow usability that orchestrates nan full routing process. This usability first classifies a query, tracks timing metrics, directs it to nan due specialised handler based connected category, and returns a broad results dictionary pinch capacity statistics.
def process_customer_query(query, client):
"""
Process a customer query done nan complete routing workflow.
Args:
query (str): The customer query
client: Anthropic client
Returns:
dict: Information astir nan query processing, including class and response
"""
start_time = time.time()
class = route_query(query, client)
routing_time = time.time() - start_time
start_time = time.time()
if class == "General Question":
consequence = handle_general_question(query, client)
model_used = "claude-3-5-haiku-20240307"
elif class == "Refund Request":
consequence = handle_refund_request(query, client)
model_used = "claude-3-5-sonnet-20240620"
elif class == "Technical Support":
consequence = handle_technical_support(query, client)
model_used = "claude-3-5-sonnet-20240620"
else:
consequence = handle_general_question(query, client)
model_used = "claude-3-5-haiku-20240307"
handling_time = time.time() - start_time
total_time = routing_time + handling_time
return {
"query": query,
"routed_category": category,
"response": response,
"model_used": model_used,
"routing_time": routing_time,
"handling_time": handling_time,
"total_time": total_time
}
Step 9: Process each query successful nan sample dataset done nan routing workflow, cod nan results pinch existent vs. predicted categories, and measure nan system’s performance.
results = []
for _, statement successful df.iterrows():
query = row['query']
consequence = process_customer_query(query, client)
result["actual_category"] = row['category']
results.append(result)
results_df = pd.DataFrame(results)
display(results_df[["query", "actual_category", "routed_category", "model_used", "total_time"]])
accuracy = (results_df["actual_category"] == results_df["routed_category"]).mean()
print(f"Routing Accuracy: {accuracy:.2%}")
from sklearn.metrics import classification_report
print(classification_report(results_df["actual_category"], results_df["routed_category"]))
Step 10: Simulated results.
simulated_results = []
for _, statement successful df.iterrows():
query = row['query']
actual_category = row['category']
if "hours" successful query.lower() aliases "policy" successful query.lower() aliases "ship" successful query.lower() aliases "payment" successful query.lower():
routed_category = "General Question"
model_used = "claude-3-5-haiku-20240307"
elif "refund" successful query.lower() aliases "money back" successful query.lower() aliases "charged" successful query.lower():
routed_category = "Refund Request"
model_used = "claude-3-5-sonnet-20240620"
else:
routed_category = "Technical Support"
model_used = "claude-3-5-sonnet-20240620"
simulated_results.append({
"query": query,
"actual_category": actual_category,
"routed_category": routed_category,
"model_used": model_used,
"routing_time": np.random.uniform(0.2, 0.5),
"handling_time": np.random.uniform(0.5, 2.0)
})
simulated_df = pd.DataFrame(simulated_results)
simulated_df["total_time"] = simulated_df["routing_time"] + simulated_df["handling_time"]
display(simulated_df[["query", "actual_category", "routed_category", "model_used", "total_time"]])
Output
Step 11: Calculate and show nan accuracy of nan simulated routing strategy by comparing predicted categories pinch existent categories.
accuracy = (simulated_df["actual_category"] == simulated_df["routed_category"]).mean()
print(f"Simulated Routing Accuracy: {accuracy:.2%}")
print(classification_report(simulated_df["actual_category"], simulated_df["routed_category"]))
Step 12: Create an interactive demo interface utilizing IPython widgets.
from IPython.display import HTML, display, clear_output
from ipywidgets import widgets
def create_demo_interface():
query_input = widgets.Textarea(
value='',
placeholder='Enter your customer work query here...',
description='Query:',
disabled=False,
layout=widgets.Layout(width='80%', height='100px')
)
output = widgets.Output()
fastener = widgets.Button(
description='Process Query',
disabled=False,
button_style='primary',
tooltip='Click to process nan query',
icon='check'
)
def on_button_clicked(b):
pinch output:
clear_output()
query = query_input.value
if not query.strip():
print("Please participate a query.")
return
if "hours" successful query.lower() aliases "policy" successful query.lower() aliases "ship" successful query.lower() aliases "payment" successful query.lower():
class = "General Question"
exemplary = "claude-3-5-haiku-20240307"
consequence = "Our modular business hours are Monday done Friday, 9 AM to 6 PM Eastern Time. Our customer work squad is disposable during these hours to assistance you."
elif "refund" successful query.lower() aliases "money back" successful query.lower() aliases "charged" successful query.lower():
class = "Refund Request"
exemplary = "claude-3-5-sonnet-20240620"
consequence = "I understand you're looking for a refund. Our refund argumentation allows returns wrong 30 days of acquisition pinch a valid receipt. To initiate your refund, please supply your bid number and nan logic for nan return."
else:
class = "Technical Support"
exemplary = "claude-3-5-sonnet-20240620"
consequence = "I'm sorry to perceive you're experiencing method issues. Let's troubleshoot this measurement by step. First, effort restarting nan application. If that doesn't work, please cheque if nan app is updated to nan latest version."
print(f"Routed to: {category}")
print(f"Using model: {model}")
print("\nResponse:")
print(response)
button.on_click(on_button_clicked)
return widgets.VBox([query_input, button, output])
Output
Step 13: Implement an precocious routing usability that not only classifies queries but besides provides assurance scores and reasoning for each classification.
def advanced_route_query(query, client):
"""
An precocious routing usability that includes assurance scores and fallback mechanisms.
Args:
query (str): The customer query to classify
client: Anthropic client
Returns:
dict: Classification consequence pinch class and confidence
"""
system_prompt = """
You are a query classifier for a customer work system.
Your occupation is to categorize customer queries into precisely 1 of these categories:
1. General Question - Basic inquiries astir nan company, products, policies, etc.
2. Refund Request - Any query related to refunds, returns, aliases billing issues
3. Technical Support - Questions astir method problems, bugs, aliases really to usage products
Respond successful JSON format with:
1. "category": The astir apt category
2. "confidence": A assurance people betwixt 0 and 1
3. "reasoning": A little mentation of your classification
Example response:
{
"category": "General Question",
"confidence": 0.85,
"reasoning": "The query asks astir business hours, which is basal institution information."
}
"""
try:
consequence = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=150,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
response_text = response.content[0].text.strip()
try:
consequence = json.loads(response_text)
if "category" not successful consequence aliases "confidence" not successful result:
raise ValueError("Incomplete classification result")
return result
isolated from json.JSONDecodeError:
print("Failed to parse JSON response. Using elemental classification.")
if "general" successful response_text.lower():
return {"category": "General Question", "confidence": 0.6, "reasoning": "Fallback classification"}
elif "refund" successful response_text.lower():
return {"category": "Refund Request", "confidence": 0.6, "reasoning": "Fallback classification"}
else:
return {"category": "Technical Support", "confidence": 0.6, "reasoning": "Fallback classification"}
isolated from Exception arsenic e:
print(f"Error successful precocious routing: {e}")
return {"category": "General Question", "confidence": 0.3, "reasoning": "Error fallback"}
Step 14: Create an enhanced query processing workflow pinch confidence-based routing that escalates low-confidence queries to specialised handling, incorporating simulated classification for objection purposes.
def advanced_process_customer_query(query, client, confidence_threshold=0.7):
"""
Process a customer query pinch confidence-based routing.
Args:
query (str): The customer query
client: Anthropic client
confidence_threshold (float): Minimum assurance people for automated routing
Returns:
dict: Information astir nan query processing
"""
start_time = time.time()
if "hours" successful query.lower() aliases "policy" successful query.lower() aliases "ship" successful query.lower() aliases "payment" successful query.lower():
classification = {
"category": "General Question",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Query related to business information"
}
elif "refund" successful query.lower() aliases "money back" successful query.lower() aliases "charged" successful query.lower():
classification = {
"category": "Refund Request",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Query mentions refunds aliases billing issues"
}
elif "password" successful query.lower() aliases "crash" successful query.lower() aliases "account" successful query.lower():
classification = {
"category": "Technical Support",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Query mentions method problems"
}
else:
categories = ["General Question", "Refund Request", "Technical Support"]
classification = {
"category": np.random.choice(categories),
"confidence": np.random.uniform(0.4, 0.65),
"reasoning": "Uncertain classification"
}
routing_time = time.time() - start_time
start_time = time.time()
if classification["confidence"] >= confidence_threshold:
class = classification["category"]
if class == "General Question":
consequence = "SIMULATED GENERAL QUESTION RESPONSE: I'd beryllium happy to thief pinch your mobility astir our business."
model_used = "claude-3-5-haiku-20240307"
elif class == "Refund Request":
consequence = "SIMULATED REFUND REQUEST RESPONSE: I understand you're looking for a refund. Let maine thief you pinch that process."
model_used = "claude-3-5-sonnet-20240620"
elif class == "Technical Support":
consequence = "SIMULATED TECHNICAL SUPPORT RESPONSE: I spot you're having a method issue. Let's troubleshoot this together."
model_used = "claude-3-5-sonnet-20240620"
else:
consequence = "I apologize, but I'm not judge really to categorize your request."
model_used = "claude-3-5-sonnet-20240620"
else:
consequence = "SIMULATED ESCALATION RESPONSE: Your query requires typical attention. I'll person our precocious support strategy thief you pinch this analyzable request."
model_used = "claude-3-5-sonnet-20240620"
class = "Escalated (Low Confidence)"
handling_time = time.time() - start_time
total_time = routing_time + handling_time
return {
"query": query,
"routed_category": classification["category"],
"confidence": classification["confidence"],
"reasoning": classification["reasoning"],
"final_category": category,
"response": response,
"model_used": model_used,
"routing_time": routing_time,
"handling_time": handling_time,
"total_time": total_time
}
Step 15: Test nan precocious routing strategy pinch divers sample queries.
test_queries = [
"What are your business hours?",
"I request a refund for my bid #12345",
"My app keeps crashing erstwhile I effort to prevention photos",
"I received nan incorrect point successful my shipment",
"How do I alteration my shipping address?",
"I'm not judge if my costs went through",
"The merchandise explanation was misleading"
]
advanced_results = []
for query successful test_queries:
consequence = advanced_process_customer_query(query, None, 0.7)
advanced_results.append(result)
advanced_df = pd.DataFrame(advanced_results)
display(advanced_df[["query", "routed_category", "confidence", "final_category", "model_used"]])
print("\nRouting Distribution:")
print(advanced_df["final_category"].value_counts())
print(f"\nAverage Confidence: {advanced_df['confidence'].mean():.2f}")
escalated = (advanced_df["final_category"] == "Escalated (Low Confidence)").sum()
print(f"Escalated Queries: {escalated} ({escalated/len(advanced_df):.1%})")
Output
Step 16: Define a inferior usability to cipher cardinal capacity metrics for nan routing system, including processing times, assurance levels, escalation rates, and class distribution statistics.
def calculate_routing_metrics(results_df):
"""
Calculate cardinal metrics for routing performance.
Args:
results_df (DataFrame): Results of routing tests
Returns:
dict: Key capacity metrics
"""
metrics = {
"total_queries": len(results_df),
"avg_routing_time": results_df["routing_time"].mean(),
"avg_handling_time": results_df["handling_time"].mean(),
"avg_total_time": results_df["total_time"].mean(),
"avg_confidence": results_df["confidence"].mean(),
"escalation_rate": (results_df["final_category"] == "Escalated (Low Confidence)").mean(),
}
category_distribution = results_df["routed_category"].value_counts(normalize=True).to_dict()
metrics["category_distribution"] = category_distribution
return metrics
Step 17: Generate and show a broad capacity study for nan routing system.
metrics = calculate_routing_metrics(advanced_df)
print("Routing System Performance Metrics:")
print(f"Total Queries: {metrics['total_queries']}")
print(f"Average Routing Time: {metrics['avg_routing_time']:.3f} seconds")
print(f"Average Handling Time: {metrics['avg_handling_time']:.3f} seconds")
print(f"Average Total Time: {metrics['avg_total_time']:.3f} seconds")
print(f"Average Confidence: {metrics['avg_confidence']:.2f}")
print(f"Escalation Rate: {metrics['escalation_rate']:.1%}")
print("\nCategory Distribution:")
for category, percent successful metrics["category_distribution"].items():
print(f" {category}: {percentage:.1%}")
Output
This intelligent petition routing system demonstrates really Claude models tin efficiently categorize and grip divers customer queries. By implementing category-specific handlers pinch due exemplary selection, nan strategy delivers tailored responses while maintaining precocious accuracy. The confidence-based routing pinch escalation paths ensures analyzable queries person specialised attention, creating a robust, scalable customer work solution.
Check retired nan Colab Notebook here. Also, don’t hide to travel america on Twitter.
Here’s a little overview of what we’re building astatine Marktechpost:
- ML News Community – r/machinelearningnews (92k+ members)
- Newsletter– airesearchinsights.com/(30k+ subscribers)
- miniCON AI Events – minicon.marktechpost.com
- AI Reports & Magazines – magazine.marktechpost.com
- AI Dev & Research News – marktechpost.com (1M+ monthly readers)
Asjad is an intern advisor astatine Marktechpost. He is persuing B.Tech successful mechanical engineering astatine nan Indian Institute of Technology, Kharagpur. Asjad is simply a Machine learning and heavy learning enthusiast who is ever researching nan applications of instrumentality learning successful healthcare.
English (US) · 
Indonesian (ID) · 