This text demonstrates learn how to construct an clever routing system powered by Anthropic’s Claude fashions. This method 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 acceptable 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-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 listing 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 shopper 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. Normal Query - Fundamental inquiries in regards to the firm, merchandise, insurance policies, and so forth.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Help - Questions on technical issues, bugs, or learn how to use merchandise
Reply with ONLY the class title, 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 "Normal Query"
besides Exception as e:
print(f"Error in routing: {e}")
return "Normal 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 normal inquiries utilizing Claude 3.5 Haiku."""
system_prompt = """
You're a customer support consultant answering normal 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 basically 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 help workforce straight."
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.
When you want extra data to resolve a problem, specify what data you want.
Prioritize easy options first earlier than suggesting complicated 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 help handler: {e}")
return "I apologize, however I am having hassle processing your technical help request. Please attempt our data base or contact our help workforce."
Step 8: Create the principle workflow perform that orchestrates all the 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 whole routing workflow.
Args:
question (str): The shopper 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 == "Normal 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 Help":
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, gather the outcomes with precise vs. predicted classes, and consider the system’s efficiency.
outcomes = []
for _, row in df.iterrows():
question = row['query']
consequence = process_customer_query(question, shopper)
consequence["actual_category"] = row['category']
outcomes.append(consequence)
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 "cost" in question.decrease():
routed_category = "Normal 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 Help"
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%", 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 "cost" in question.decrease():
class = "Normal 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 workforce is accessible 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 explanation for the return."
else:
class = "Technical Help"
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 sophisticated 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 shopper question to categorise
shopper: Anthropic shopper
Returns:
dict: Classification consequence 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. Normal Query - Fundamental inquiries in regards to the firm, merchandise, insurance policies, and so forth.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Help - Questions on technical issues, bugs, or learn how to use merchandise
Reply in JSON format with:
1. "class": The most probably class
2. "confidence": A confidence rating between 0 and 1
3. "reasoning": A short rationalization of your classification
Instance response:
{
"class": "Normal 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:
consequence = json.masses(response_text)
if "class" not in consequence or "confidence" not in consequence:
increase ValueError("Incomplete classification consequence")
return consequence
besides json.JSONDecodeError:
print("Did not parse JSON response. Utilizing easy classification.")
if "normal" in response_text.decrease():
return {"class": "Normal 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 Help", "confidence": 0.6, "reasoning": "Fallback classification"}
besides Exception as e:
print(f"Error in superior routing: {e}")
return {"class": "Normal 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 shopper 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 "cost" in question.decrease():
classification = {
"class": "Normal 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 Help",
"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 == "Normal Query":
response = "SIMULATED GENERAL QUESTION RESPONSE: I would be comfortable to assist together 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 Help":
response = "SIMULATED TECHNICAL SUPPORT RESPONSE: I see you are having a technical subject. Let's troubleshoot this collectively."
model_used = "claude-3-5-sonnet-20240620"
else:
response = "I apologize, however I am undecided learn how 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 help system provide help to with this complicated 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:
consequence = advanced_process_customer_query(question, None, 0.7)
advanced_results.append(consequence)
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 instances, 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 assessments
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 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 acceptable mannequin choice, the system delivers tailor-made responses whereas sustaining excessive accuracy. The boldness-based routing with escalation paths ensures complicated queries obtain specialised consideration, creating a strong, scalable customer support resolution.
Take a look at the Colab Pocket book right here. Additionally, don’t overlook to observe us on Twitter.
Right here’s a quick overview of what we’re constructing at Marktechpost: