This text demonstrates how you can 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 assist, 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, "question": "What are what you are promoting hours?", "class": "Basic Query"},
{"id": 2, "question": "How do I reset my password?", "class": "Technical Help"},
{"id": 3, "question": "I desire a refund for my buy.", "class": "Refund Request"},
{"id": 4, "question": "The place can I discover your privateness coverage?", "class": "Basic Query"},
{"id": 5, "question": "The app retains crashing when I attempt to add images.", "class": "Technical Help"},
{"id": 6, "question": "I ordered the unsuitable measurement, can I get my a reimbursement?", "class": "Refund Request"},
{"id": 7, "question": "Do you ship internationally?", "class": "Basic Query"},
{"id": 8, "question": "My account is displaying incorrect data.", "class": "Technical Help"},
{"id": 9, "question": "I used to be charged twice for my order.", "class": "Refund Request"},
{"id": 10, "question": "What cost strategies do you settle for?", "class": "Basic Query"}
)
Step 5: Convert the client queries record 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 operate 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. Basic Query - Primary inquiries concerning the firm, merchandise, insurance policies, and many others.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Help - Questions on technical issues, bugs, or how you can 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=({"function": "consumer", "content material": question})
)
class = response.content material(0).textual content.strip()
valid_categories = ("Basic Query", "Refund Request", "Technical Help")
for valid_cat in valid_categories:
if valid_cat.decrease() in class.decrease():
return valid_cat
return "Basic Query"
besides Exception as e:
print(f"Error in routing: {e}")
return "Basic 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.
"""
attempt:
response = shopper.messages.create(
mannequin="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=({"function": "consumer", "content material": question})
)
return response.content material(0).textual content.strip()
besides Exception as e:
print(f"Error generally 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 comply with firm coverage.
"""
attempt:
response = shopper.messages.create(
mannequin="claude-3-7-sonnet-20250219",
max_tokens=1024,
system=system_prompt,
messages=({"function": "consumer", "content material": question})
)
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 group 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.
If you happen to 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=({"function": "consumer", "content material": question})
)
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 group."Step 8: Create the primary workflow operate that orchestrates the whole routing course of. This operate 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 == "Basic 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('question')
end result = process_customer_query(question, shopper)
end result("actual_category") = row('class')
outcomes.append(end result)
results_df = pd.DataFrame(outcomes)
show(results_df(("question", "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('question')
actual_category = row('class')
if "hours" in question.decrease() or "coverage" in question.decrease() or "ship" in question.decrease() or "cost" in question.decrease():
routed_category = "Basic 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 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(("question", "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.Format(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='verify'
)
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 = "Basic Query"
mannequin = "claude-3-5-haiku-20240307"
response = "Our normal enterprise hours are Monday by means of Friday, 9 AM to six PM Jap Time. Our customer support group is accessible 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 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 verify 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 operate that not solely classifies queries but in addition offers confidence scores and reasoning for every classification.
def advanced_route_query(question, shopper):
"""
A complicated routing operate that features confidence scores and fallback mechanisms.
Args:
question (str): The client question to categorise
shopper: Anthropic shopper
Returns:
dict: Classification end result 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. Basic Query - Primary inquiries concerning the firm, merchandise, insurance policies, and many others.
2. Refund Request - Any question associated to refunds, returns, or billing points
3. Technical Help - Questions on technical issues, bugs, or how you can use merchandise
Reply in JSON format with:
1. "class": The most certainly class
2. "confidence": A confidence rating between 0 and 1
3. "reasoning": A short clarification of your classification
Instance response:
{
"class": "Basic 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=({"function": "consumer", "content material": question})
)
response_text = response.content material(0).textual content.strip()
attempt:
end result = json.masses(response_text)
if "class" not in end result or "confidence" not in end result:
increase ValueError("Incomplete classification end result")
return end result
besides json.JSONDecodeError:
print("Didn't parse JSON response. Utilizing easy classification.")
if "basic" in response_text.decrease():
return {"class": "Basic 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": "Basic 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 "cost" in question.decrease():
classification = {
"class": "Basic 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 Help",
"confidence": np.random.uniform(0.7, 0.95),
"reasoning": "Question mentions technical issues"
}
else:
classes = ("Basic Query", "Refund Request", "Technical Help")
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("class")
if class == "Basic Query":
response = "SIMULATED GENERAL QUESTION RESPONSE: I would be completely happy 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 enable you 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 how you can 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 enable you 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("class"),
"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: Check the superior routing system with various pattern queries.
test_queries = (
"What are what you are promoting hours?",
"I want a refund for my order #12345",
"My app retains crashing when I attempt to save images",
"I obtained the unsuitable merchandise in my cargo",
"How do I modify my transport handle?",
"I am undecided if my cost went by means of",
"The product description was deceptive"
)
advanced_results = ()
for question in test_queries:
end result = advanced_process_customer_query(question, None, 0.7)
advanced_results.append(end result)
advanced_df = pd.DataFrame(advanced_results)
show(advanced_df(("question", "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 operate 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 exams
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"Complete 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 Complete Time: {metrics('avg_total_time'):.3f} seconds")
print(f"Common Confidence: {metrics('avg_confidence'):.2f}")
print(f"Escalation Fee: {metrics('escalation_rate'):.1%}")
print("nCategory Distribution:")
for class, proportion in metrics("category_distribution").gadgets():
print(f" {class}: {proportion:.1%}")
This clever request routing system demonstrates how Claude fashions can effectively classify and deal with various 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 strong, scalable customer support resolution.
Try the Colab Pocket book right here. Additionally, don’t overlook to comply with us on Twitter.
Right here’s a short 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 at all times researching the purposes of machine studying in healthcare.
