This text demonstrates methods to construct an clever routing system powered by Anthropic’s Claude fashions. This technique improves response effectivity and high quality by mechanically classifying consumer requests and directing them to specialised handlers. The workflow analyses incoming queries, determines their intent, and routes them to applicable processing pipelines—whether or not for buyer help, technical help, or different domain-specific responses.
Step 1:Â Set up the required Python packages
!pip set up anthropic pandas scikit-learnStep 2:Â Import the mandatory libraries for the venture
import os
import json
import time
import pandas as pd
import numpy as np
from anthropic import Anthropic
from IPython.show import show, Markdown
from sklearn.metrics import classification_reportStep 3: Arrange the Anthropic API authentication by defining your API key and initialising the Anthropic shopper
ANTHROPIC_API_KEY = "{Your API KEY}"
shopper = Anthropic(api_key=ANTHROPIC_API_KEY)Step 4: Create a pattern dataset of buyer queries with related classes for coaching and testing the 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 can I find your privacy policy?", "category": "General Question"},
{"id": 5, "query": "The app keeps crashing when I try to upload photos.", "category": "Technical Support"},
{"id": 6, "query": "I ordered the wrong size, can I get my money back?", "category": "Refund Request"},
{"id": 7, "query": "Do you ship internationally?", "category": "General Question"},
{"id": 8, "query": "My account is showing incorrect information.", "category": "Technical Support"},
{"id": 9, "query": "I was charged twice for my order.", "category": "Refund Request"},
{"id": 10, "query": "What payment methods do you accept?", "category": "General Question"}
]
Step 5: Convert the shopper queries checklist right into a pandas DataFrame for simpler manipulation and evaluation. Then, show the DataFrame within the pocket book to visualise the coaching dataset construction.
df = pd.DataFrame(customer_queries)
show(df)
Step 6: Outline the core routing perform that makes use of Claude 3.7 Sonnet to categorise buyer queries into predefined classes.
def route_query(question, shopper):
"""
Route a buyer question to the suitable class utilizing Claude 3.5 Haiku.
Args:
question (str): The client question to categorise
shopper: Anthropic shopper
Returns:
str: The categorised class
"""
system_prompt = """
You're a question classifier for a customer support system.
Your job is to categorize buyer queries into precisely one in all these classes:
1. Common Query - Primary inquiries concerning the firm, merchandise, insurance policies, and so forth.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Assist - Questions on technical issues, bugs, or methods to use merchandise
Reply with ONLY the class title, nothing else.
"""
strive:
response = shopper.messages.create(
mannequin="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
class = response.content material[0].textual content.strip()
valid_categories = ["General Question", "Refund Request", "Technical Support"]
for valid_cat in valid_categories:
if valid_cat.decrease() in class.decrease():
return valid_cat
return "Common Query"
besides Exception as e:
print(f"Error in routing: {e}")
return "Common Query"Step 7: Outline three specialised handler features for every question class, every utilizing Claude 3.5 Sonnet with a category-specific system immediate.
def handle_general_question(question, shopper):
"""Deal with basic inquiries utilizing Claude 3.5 Haiku."""
system_prompt = """
You're a customer support consultant answering basic questions on our firm.
Be useful, concise, and pleasant. Present direct solutions to buyer queries.
"""
strive:
response = shopper.messages.create(
mannequin="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
return response.content material[0].textual content.strip()
besides Exception as e:
print(f"Error usually query handler: {e}")
return "I apologize, however I am having bother processing your request. Please strive once more later."def handle_refund_request(question, shopper):
"""Deal with refund requests utilizing Claude 3.5 Sonnet for extra nuanced responses."""
system_prompt = """
You're a customer support consultant specializing in refunds and billing points.
Reply to refund requests professionally and helpfully.
For any refund request, clarify the refund coverage clearly and supply subsequent steps.
Be empathetic however comply with firm coverage.
"""
strive:
response = shopper.messages.create(
mannequin="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
return response.content material[0].textual content.strip()
besides Exception as e:
print(f"Error in refund request handler: {e}")
return "I apologize, however I am having bother processing your refund request. Please contact our help workforce instantly."def handle_technical_support(question, shopper):
"""Deal with technical help queries utilizing Claude 3.5 Sonnet for extra detailed technical responses."""
system_prompt = """
You're a technical help specialist.
Present clear, step-by-step options to technical issues.
In case you want extra data to resolve a difficulty, specify what data you want.
Prioritize easy options first earlier than suggesting advanced troubleshooting.
"""
strive:
response = shopper.messages.create(
mannequin="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
return response.content material[0].textual content.strip()
besides Exception as e:
print(f"Error in technical help handler: {e}")
return "I apologize, however I am having bother processing your technical help request. Please strive our information base or contact our help workforce."Step 8: Create the primary workflow perform that orchestrates your entire routing course of. This perform first classifies a question, tracks timing metrics, directs it to the suitable specialised handler primarily based on class, and returns a complete outcomes dictionary with efficiency statistics.
def process_customer_query(question, shopper):
"""
Course of a buyer question by way of the whole routing workflow.
Args:
question (str): The client question
shopper: Anthropic shopper
Returns:
dict: Details about the question processing, together with class and response
"""
start_time = time.time()
class = route_query(question, shopper)
routing_time = time.time() - start_time
start_time = time.time()
if class == "Common Query":
response = handle_general_question(question, shopper)
model_used = "claude-3-5-haiku-20240307"
elif class == "Refund Request":
response = handle_refund_request(question, shopper)
model_used = "claude-3-5-sonnet-20240620"
elif class == "Technical Assist":
response = handle_technical_support(question, shopper)
model_used = "claude-3-5-sonnet-20240620"
else:
response = handle_general_question(question, shopper)
model_used = "claude-3-5-haiku-20240307"
handling_time = time.time() - start_time
total_time = routing_time + handling_time
return {
"question": question,
"routed_category": class,
"response": response,
"model_used": model_used,
"routing_time": routing_time,
"handling_time": handling_time,
"total_time": total_time
}Step 9: Course of every question within the pattern dataset by way of the routing workflow, accumulate the outcomes with precise vs. predicted classes, and consider the system’s efficiency.
outcomes = []
for _, row in df.iterrows():
question = row['query']
outcome = process_customer_query(question, shopper)
outcome["actual_category"] = row['category']
outcomes.append(outcome)
results_df = pd.DataFrame(outcomes)
show(results_df[["query", "actual_category", "routed_category", "model_used", "total_time"]])
accuracy = (results_df["actual_category"] == results_df["routed_category"]).imply()
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 outcomes.
simulated_results = []
for _, row in df.iterrows():
question = row['query']
actual_category = row['category']
if "hours" in question.decrease() or "coverage" in question.decrease() or "ship" in question.decrease() or "fee" in question.decrease():
routed_category = "Common Query"
model_used = "claude-3-5-haiku-20240307"
elif "refund" in question.decrease() or "a refund" in question.decrease() or "charged" in question.decrease():
routed_category = "Refund Request"
model_used = "claude-3-5-sonnet-20240620"
else:
routed_category = "Technical Assist"
model_used = "claude-3-5-sonnet-20240620"
simulated_results.append({
"question": question,
"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"]
show(simulated_df[["query", "actual_category", "routed_category", "model_used", "total_time"]])
Step 11: Calculate and show the accuracy of the simulated routing system by evaluating predicted classes with precise classes.
accuracy = (simulated_df["actual_category"] == simulated_df["routed_category"]).imply()
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.show import HTML, show, clear_output
from ipywidgets import widgets
def create_demo_interface():
query_input = widgets.Textarea(
worth="",
placeholder="Enter your customer support question right here...",
description='Question:',
disabled=False,
structure=widgets.Structure(width="80%", top="100px")
)
output = widgets.Output()
button = widgets.Button(
description='Course of Question',
disabled=False,
button_style="major",
tooltip='Click on to course of the question',
icon='examine'
)
def on_button_clicked(b):
with output:
clear_output()
question = query_input.worth
if not question.strip():
print("Please enter a question.")
return
if "hours" in question.decrease() or "coverage" in question.decrease() or "ship" in question.decrease() or "fee" in question.decrease():
class = "Common Query"
mannequin = "claude-3-5-haiku-20240307"
response = "Our commonplace enterprise hours are Monday by way of Friday, 9 AM to six PM Jap Time. Our customer support workforce is out there throughout these hours to help you."
elif "refund" in question.decrease() or "a refund" in question.decrease() or "charged" in question.decrease():
class = "Refund Request"
mannequin = "claude-3-5-sonnet-20240620"
response = "I perceive you are in search of a refund. Our refund coverage permits returns inside 30 days of buy with a legitimate receipt. To provoke your refund, please present your order quantity and the rationale for the return."
else:
class = "Technical Assist"
mannequin = "claude-3-5-sonnet-20240620"
response = "I am sorry to listen to you are experiencing technical points. Let's troubleshoot this step-by-step. First, strive restarting the applying. If that does not work, please examine if the app is up to date to the newest model."
print(f"Routed to: {class}")
print(f"Utilizing mannequin: {mannequin}")
print("nResponse:")
print(response)
button.on_click(on_button_clicked)
return widgets.VBox([query_input, button, output])
Step 13: Implement a complicated routing perform that not solely classifies queries but additionally supplies confidence scores and reasoning for every classification.
def advanced_route_query(question, shopper):
"""
A sophisticated routing perform that features confidence scores and fallback mechanisms.
Args:
question (str): The client question to categorise
shopper: Anthropic shopper
Returns:
dict: Classification outcome with class and confidence
"""
system_prompt = """
You're a question classifier for a customer support system.
Your job is to categorize buyer queries into precisely one in all these classes:
1. Common Query - Primary inquiries concerning the firm, merchandise, insurance policies, and so forth.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Assist - Questions on technical issues, bugs, or methods to use merchandise
Reply in JSON format with:
1. "class": The almost certainly class
2. "confidence": A confidence rating between 0 and 1
3. "reasoning": A short clarification of your classification
Instance response:
{
"class": "Common Query",
"confidence": 0.85,
"reasoning": "The question asks about enterprise hours, which is primary firm data."
}
"""
strive:
response = shopper.messages.create(
mannequin="claude-3-5-sonnet-20240620",
max_tokens=150,
system=system_prompt,
messages=[{"role": "user", "content": query}]
)
response_text = response.content material[0].textual content.strip()
strive:
outcome = json.masses(response_text)
if "class" not in outcome or "confidence" not in outcome:
elevate ValueError("Incomplete classification outcome")
return outcome
besides json.JSONDecodeError:
print("Didn't parse JSON response. Utilizing easy classification.")
if "basic" in response_text.decrease():
return {"class": "Common Query", "confidence": 0.6, "reasoning": "Fallback classification"}
elif "refund" in response_text.decrease():
return {"class": "Refund Request", "confidence": 0.6, "reasoning": "Fallback classification"}
else:
return {"class": "Technical Assist", "confidence": 0.6, "reasoning": "Fallback classification"}
besides Exception as e:
print(f"Error in superior routing: {e}")
return {"class": "Common Query", "confidence": 0.3, "reasoning": "Error fallback"}Step 14: Create an enhanced question processing workflow with confidence-based routing that escalates low-confidence queries to specialised dealing with, incorporating simulated classification for demonstration functions.
def advanced_process_customer_query(question, shopper, confidence_threshold=0.7):
"""
Course of a buyer question with confidence-based routing.
Args:
question (str): The client question
shopper: Anthropic shopper
confidence_threshold (float): Minimal confidence rating for automated routing
Returns:
dict: Details about the question processing
"""
start_time = time.time()
if "hours" in question.decrease() or "coverage" in question.decrease() or "ship" in question.decrease() or "fee" in question.decrease():
classification = {
"class": "Common Query",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Question associated to enterprise data"
}
elif "refund" in question.decrease() or "a refund" in question.decrease() or "charged" in question.decrease():
classification = {
"class": "Refund Request",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Question mentions refunds or billing points"
}
elif "password" in question.decrease() or "crash" in question.decrease() or "account" in question.decrease():
classification = {
"class": "Technical Assist",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Question mentions technical issues"
}
else:
classes = ["General Question", "Refund Request", "Technical Support"]
classification = {
"class": np.random.selection(classes),
"confidence": np.random.uniform(0.4, 0.65),
"reasoning": "Unsure classification"
}
routing_time = time.time() - start_time
start_time = time.time()
if classification["confidence"] >= confidence_threshold:
class = classification["category"]
if class == "Common Query":
response = "SIMULATED GENERAL QUESTION RESPONSE: I might be comfortable to assist along with your query about our enterprise."
model_used = "claude-3-5-haiku-20240307"
elif class == "Refund Request":
response = "SIMULATED REFUND REQUEST RESPONSE: I perceive you are in search of a refund. Let me provide help to with that course of."
model_used = "claude-3-5-sonnet-20240620"
elif class == "Technical Assist":
response = "SIMULATED TECHNICAL SUPPORT RESPONSE: I see you are having a technical difficulty. Let's troubleshoot this collectively."
model_used = "claude-3-5-sonnet-20240620"
else:
response = "I apologize, however I am unsure methods to categorize your request."
model_used = "claude-3-5-sonnet-20240620"
else:
response = "SIMULATED ESCALATION RESPONSE: Your question requires particular consideration. I will have our superior help system provide help to with this advanced 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 {
"question": question,
"routed_category": classification["category"],
"confidence": classification["confidence"],
"reasoning": classification["reasoning"],
"final_category": class,
"response": response,
"model_used": model_used,
"routing_time": routing_time,
"handling_time": handling_time,
"total_time": total_time
}Step 15: Take a look at the superior routing system with numerous pattern queries.
test_queries = [
"What are your business hours?",
"I need a refund for my order #12345",
"My app keeps crashing when I try to save photos",
"I received the wrong item in my shipment",
"How do I change my shipping address?",
"I'm not sure if my payment went through",
"The product description was misleading"
]
advanced_results = []
for question in test_queries:
outcome = advanced_process_customer_query(question, None, 0.7)
advanced_results.append(outcome)
advanced_df = pd.DataFrame(advanced_results)
show(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'].imply():.2f}")
escalated = (advanced_df["final_category"] == "Escalated (Low Confidence)").sum()
print(f"Escalated Queries: {escalated} ({escalated/len(advanced_df):.1%})")
Step 16: Outline a utility perform to calculate key efficiency metrics for the routing system, together with processing occasions, confidence ranges, escalation charges, and class distribution statistics.
def calculate_routing_metrics(results_df):
"""
Calculate key metrics for routing efficiency.
Args:
results_df (DataFrame): Outcomes of routing checks
Returns:
dict: Key efficiency metrics
"""
metrics = {
"total_queries": len(results_df),
"avg_routing_time": results_df["routing_time"].imply(),
"avg_handling_time": results_df["handling_time"].imply(),
"avg_total_time": results_df["total_time"].imply(),
"avg_confidence": results_df["confidence"].imply(),
"escalation_rate": (results_df["final_category"] == "Escalated (Low Confidence)").imply(),
}
category_distribution = results_df["routed_category"].value_counts(normalize=True).to_dict()
metrics["category_distribution"] = category_distribution
return metricsStep 17: Generate and show a complete efficiency report for the routing system.
metrics = calculate_routing_metrics(advanced_df)
print("Routing System Efficiency Metrics:")
print(f"Whole Queries: {metrics['total_queries']}")
print(f"Common Routing Time: {metrics['avg_routing_time']:.3f} seconds")
print(f"Common Dealing with Time: {metrics['avg_handling_time']:.3f} seconds")
print(f"Common Whole Time: {metrics['avg_total_time']:.3f} seconds")
print(f"Common Confidence: {metrics['avg_confidence']:.2f}")
print(f"Escalation Charge: {metrics['escalation_rate']:.1%}")
print("nCategory Distribution:")
for class, share in metrics["category_distribution"].gadgets():
print(f" {class}: {share:.1%}")
This clever request routing system demonstrates how Claude fashions can effectively classify and deal with numerous buyer queries. By implementing category-specific handlers with applicable mannequin choice, the system delivers tailor-made responses whereas sustaining excessive accuracy. The boldness-based routing with escalation paths ensures advanced queries obtain specialised consideration, creating a sturdy, scalable customer support answer.
Try the Colab Pocket book right here. Additionally, don’t overlook to comply with us on Twitter.
Right here’s a quick overview of what we’re constructing at Marktechpost:
Asjad is an intern guide at Marktechpost. He’s persuing B.Tech in mechanical engineering on the Indian Institute of Know-how, Kharagpur. Asjad is a Machine studying and deep studying fanatic who’s all the time researching the functions of machine studying in healthcare.
