This text demonstrates the way to construct an clever routing system powered by Anthropic’s Claude fashions. This technique improves response effectivity and high quality by mechanically classifying person requests and directing them to specialised handlers. The workflow analyses incoming queries, determines their intent, and routes them to acceptable processing pipelines—whether or not for buyer assist, technical help, or different domain-specific responses.
Step 1:Â Set up the required Python packages
!pip set up anthropic pandas scikit-learn
Step 2:Â Import the required libraries for the mission
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_report
Step 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 categorized class
"""
system_prompt = """
You're a question classifier for a customer support system.
Your job is to categorize buyer queries into precisely one among these classes:
1. Common Query - Primary inquiries concerning the firm, merchandise, insurance policies, and so on.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Assist - Questions on technical issues, bugs, or the way to use merchandise
Reply with ONLY the class identify, nothing else.
"""
attempt:
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 capabilities 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.
"""
attempt:
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 hassle processing your request. Please attempt 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 observe firm coverage.
"""
attempt:
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 hassle processing your refund request. Please contact our assist crew immediately."
def handle_technical_support(question, shopper):
"""Deal with technical assist queries utilizing Claude 3.5 Sonnet for extra detailed technical responses."""
system_prompt = """
You're a technical assist specialist.
Present clear, step-by-step options to technical issues.
When you want extra data to resolve a problem, specify what data you want.
Prioritize easy options first earlier than suggesting advanced troubleshooting.
"""
attempt:
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 assist handler: {e}")
return "I apologize, however I am having hassle processing your technical assist request. Please attempt our information base or contact our assist crew."
Step 8: Create the principle workflow perform that orchestrates the complete 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 means of the entire 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 means 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 reimbursement" 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,
format=widgets.Structure(width="80%", peak="100px")
)
output = widgets.Output()
button = widgets.Button(
description='Course of Question',
disabled=False,
button_style="main",
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 means of Friday, 9 AM to six PM Jap Time. Our customer support crew is on the market throughout these hours to help you."
elif "refund" in question.decrease() or "a reimbursement" 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 sound 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, attempt restarting the appliance. If that does not work, please examine if the app is up to date to the most recent 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 gives confidence scores and reasoning for every classification.
def advanced_route_query(question, shopper):
"""
A complicated 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 among these classes:
1. Common Query - Primary inquiries concerning the firm, merchandise, insurance policies, and so on.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Assist - Questions on technical issues, bugs, or the way to use merchandise
Reply in JSON format with:
1. "class": The more than likely class
2. "confidence": A confidence rating between 0 and 1
3. "reasoning": A quick clarification of your classification
Instance response:
{
"class": "Common Query",
"confidence": 0.85,
"reasoning": "The question asks about enterprise hours, which is primary firm data."
}
"""
attempt:
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()
attempt:
outcome = json.masses(response_text)
if "class" not in outcome or "confidence" not in outcome:
increase 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 reimbursement" 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.alternative(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 show you how 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 concern. Let's troubleshoot this collectively."
model_used = "claude-3-5-sonnet-20240620"
else:
response = "I apologize, however I am undecided the way to categorize your request."
model_used = "claude-3-5-sonnet-20240620"
else:
response = "SIMULATED ESCALATION RESPONSE: Your question requires particular consideration. I am going to have our superior assist system show you how 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 metrics
Step 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 Price: {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 acceptable 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.
Take a look at the Colab Pocket book right here. Additionally, don’t neglect to observe us on Twitter.
Right here’s a quick overview of what we’re constructing at Marktechpost: