Sales Assistant
Build an intelligent sales assistant with BroxiAI for lead qualification, product recommendations, and automated sales processes
Learn how to create a comprehensive AI-powered sales assistant that can qualify leads, provide product recommendations, handle objections, and automate key parts of your sales process while maintaining human oversight.
What You'll Build
A sophisticated sales assistant that:
Qualifies leads through intelligent conversations
Provides personalized product recommendations
Handles common objections professionally
Schedules meetings and follow-ups
Integrates with CRM systems
Tracks sales metrics and performance
Escalates to human sales reps when needed
Prerequisites
BroxiAI account with API access
OpenAI API key or other LLM provider
CRM system access (Salesforce, HubSpot, etc.)
Customer database or lead sources
Calendar integration (optional)
Email automation platform (optional)
Architecture Overview
Step 1: Lead Qualification System
Lead Scoring Component
Lead Qualification Framework
# Lead qualification scoring
lead_qualification_agent = {
"component": "LeadQualificationAgent",
"config": {
"model": "gpt-4",
"temperature": 0.3,
"system_prompt": """
You are an expert sales qualification assistant. Your goal is to qualify leads using the BANT framework:
BUDGET: Do they have budget for our solution?
AUTHORITY: Are they a decision maker or influencer?
NEED: Do they have a clear need for our product?
TIMELINE: What's their purchasing timeline?
QUALIFICATION CRITERIA:
- High Quality Lead: 3-4 BANT criteria met
- Medium Quality Lead: 2 BANT criteria met
- Low Quality Lead: 0-1 BANT criteria met
CONVERSATION APPROACH:
1. Build rapport and understand their situation
2. Ask discovery questions naturally in conversation
3. Listen for buying signals and pain points
4. Score each BANT criteria (0-1) based on responses
5. Provide personalized value propositions
DISCOVERY QUESTIONS:
Budget: "What's your current investment in [relevant area]?"
Authority: "Who else would be involved in this decision?"
Need: "What challenges are you facing with [specific area]?"
Timeline: "When are you looking to implement a solution?"
Never be pushy or sales-y. Focus on understanding their needs first.
""",
"max_tokens": 1000,
"tools": [
"lead_scoring",
"crm_lookup",
"product_matcher"
]
}
}Lead Scoring Logic
class LeadScorer:
def __init__(self):
self.bant_criteria = {
"budget": 0,
"authority": 0,
"need": 0,
"timeline": 0
}
self.scoring_keywords = {
"budget": {
"positive": ["budget", "allocated", "approved", "investment", "spend"],
"negative": ["no budget", "tight budget", "free", "cheap"]
},
"authority": {
"positive": ["decision maker", "I decide", "my choice", "I approve"],
"negative": ["need approval", "not my decision", "my boss decides"]
},
"need": {
"positive": ["problem", "challenge", "need", "pain", "difficulty"],
"negative": ["working fine", "satisfied", "no issues"]
},
"timeline": {
"positive": ["soon", "quickly", "urgent", "this quarter", "ASAP"],
"negative": ["eventually", "someday", "no rush", "future"]
}
}
def score_conversation(self, conversation_text):
"""Score lead based on conversation content"""
scores = {}
for criterion, keywords in self.scoring_keywords.items():
positive_score = sum(1 for word in keywords["positive"]
if word.lower() in conversation_text.lower())
negative_score = sum(1 for word in keywords["negative"]
if word.lower() in conversation_text.lower())
# Calculate score (0-1 scale)
raw_score = max(0, positive_score - negative_score)
normalized_score = min(1.0, raw_score / 3) # Cap at 1.0
scores[criterion] = normalized_score
# Calculate overall score
overall_score = sum(scores.values()) / len(scores)
# Determine lead quality
if overall_score >= 0.75:
quality = "High"
elif overall_score >= 0.5:
quality = "Medium"
else:
quality = "Low"
return {
"overall_score": overall_score,
"quality": quality,
"bant_scores": scores,
"recommendations": self.get_recommendations(scores, quality)
}
def get_recommendations(self, scores, quality):
"""Get next steps recommendations"""
recommendations = []
if quality == "High":
recommendations.append("Schedule demo call immediately")
recommendations.append("Send pricing information")
recommendations.append("Involve sales manager")
elif quality == "Medium":
recommendations.extend([
"Gather more qualification information",
"Send case studies and ROI calculator",
"Schedule follow-up call"
])
else:
recommendations.extend([
"Add to nurture campaign",
"Send educational content",
"Re-qualify in 3 months"
])
# Specific recommendations based on weak areas
if scores["budget"] < 0.5:
recommendations.append("Explore budget and ROI justification")
if scores["authority"] < 0.5:
recommendations.append("Identify and engage decision makers")
if scores["need"] < 0.5:
recommendations.append("Develop pain points and business case")
if scores["timeline"] < 0.5:
recommendations.append("Create urgency and compelling events")
return recommendationsConversation Flow Management
Dynamic Conversation Handler
class SalesConversationManager:
def __init__(self):
self.conversation_stages = {
"greeting": self.handle_greeting,
"discovery": self.handle_discovery,
"qualification": self.handle_qualification,
"presentation": self.handle_presentation,
"objection": self.handle_objection,
"closing": self.handle_closing
}
self.current_stage = "greeting"
def handle_greeting(self, user_input, context):
"""Handle initial greeting and rapport building"""
greeting_prompt = f"""
Greet the prospect warmly and start building rapport.
Lead Information:
- Name: {context.get('name', 'there')}
- Company: {context.get('company', 'Unknown')}
- Source: {context.get('source', 'Website')}
Goals:
1. Make them feel welcome and valued
2. Confirm their interest/inquiry
3. Ask permission to ask a few questions
4. Transition naturally to discovery
Keep it conversational and professional.
"""
return self.generate_response(greeting_prompt, user_input)
def handle_discovery(self, user_input, context):
"""Handle discovery and needs assessment"""
discovery_prompt = f"""
You're in the discovery phase. Focus on understanding their:
BUSINESS SITUATION:
- Current challenges and pain points
- Existing solutions and what's not working
- Goals and desired outcomes
- Key metrics and success criteria
DISCOVERY QUESTIONS TO EXPLORE:
- "What's your biggest challenge with [relevant area]?"
- "How are you currently handling [specific process]?"
- "What would an ideal solution look like?"
- "What metrics matter most to your success?"
Listen actively and ask follow-up questions based on their responses.
Previous conversation context: {context.get('conversation_history', [])}
"""
return self.generate_response(discovery_prompt, user_input)
def handle_qualification(self, user_input, context):
"""Handle BANT qualification"""
qualification_prompt = f"""
Continue the discovery while gathering BANT qualification information:
BUDGET EXPLORATION:
- "What's your current investment in this area?"
- "Have you allocated budget for this type of solution?"
- "What's the cost of not solving this problem?"
AUTHORITY IDENTIFICATION:
- "Who else would be involved in evaluating a solution?"
- "What's your role in the decision-making process?"
- "Who would need to approve this investment?"
NEED VALIDATION:
- "How urgent is solving this problem?"
- "What happens if you don't address this soon?"
- "How much time/money is this costing you?"
TIMELINE ASSESSMENT:
- "When are you looking to implement a solution?"
- "What's driving the timing for this project?"
- "Are there any deadlines or events we should know about?"
Weave these questions naturally into the conversation.
"""
return self.generate_response(qualification_prompt, user_input)
def determine_next_stage(self, user_input, current_context):
"""Determine the next conversation stage"""
# Analyze conversation to determine stage transition
stage_analysis = self.analyze_conversation_stage(user_input, current_context)
if stage_analysis["ready_for_next_stage"]:
stage_order = ["greeting", "discovery", "qualification", "presentation", "closing"]
current_index = stage_order.index(self.current_stage)
if current_index < len(stage_order) - 1:
self.current_stage = stage_order[current_index + 1]
# Handle objections at any stage
if stage_analysis["objection_detected"]:
self.current_stage = "objection"
return self.current_stageStep 2: Product Recommendation Engine
Intelligent Product Matching
Product Recommendation System
class ProductRecommendationEngine:
def __init__(self):
self.product_catalog = self.load_product_catalog()
self.recommendation_rules = self.load_recommendation_rules()
def load_product_catalog(self):
"""Load product catalog with features and use cases"""
return {
"starter_plan": {
"name": "Starter Plan",
"price": "$99/month",
"features": [
"Up to 10 workflows",
"Basic integrations",
"Email support",
"Standard templates"
],
"ideal_for": [
"Small businesses",
"Startups",
"Individual professionals",
"Testing and evaluation"
],
"use_cases": [
"Simple chatbots",
"Basic automation",
"Content generation",
"Customer service"
],
"company_size": "1-10 employees",
"technical_level": "beginner"
},
"professional_plan": {
"name": "Professional Plan",
"price": "$299/month",
"features": [
"Unlimited workflows",
"Advanced integrations",
"Priority support",
"Custom templates",
"Analytics dashboard",
"Team collaboration"
],
"ideal_for": [
"Growing businesses",
"Marketing teams",
"Sales organizations",
"Customer service departments"
],
"use_cases": [
"Multi-agent systems",
"Complex workflows",
"CRM integration",
"Lead qualification",
"Content automation"
],
"company_size": "10-100 employees",
"technical_level": "intermediate"
},
"enterprise_plan": {
"name": "Enterprise Plan",
"price": "Custom pricing",
"features": [
"Unlimited everything",
"Custom integrations",
"Dedicated support",
"White-label options",
"Advanced security",
"Custom development",
"SLA guarantees"
],
"ideal_for": [
"Large enterprises",
"Complex organizations",
"High-volume users",
"Compliance-sensitive industries"
],
"use_cases": [
"Enterprise automation",
"Compliance workflows",
"Custom AI solutions",
"Large-scale deployments"
],
"company_size": "100+ employees",
"technical_level": "advanced"
}
}
def recommend_products(self, lead_profile, needs_assessment):
"""Generate personalized product recommendations"""
scores = {}
for product_id, product in self.product_catalog.items():
score = self.calculate_product_fit_score(
product, lead_profile, needs_assessment
)
scores[product_id] = score
# Sort by score and return top recommendations
sorted_products = sorted(scores.items(), key=lambda x: x[1], reverse=True)
recommendations = []
for product_id, score in sorted_products[:3]: # Top 3 recommendations
product = self.product_catalog[product_id]
recommendations.append({
"product": product,
"fit_score": score,
"reasoning": self.generate_recommendation_reasoning(
product, lead_profile, needs_assessment
)
})
return recommendations
def calculate_product_fit_score(self, product, lead_profile, needs_assessment):
"""Calculate how well a product fits the lead's needs"""
score = 0
# Company size fit
company_size = lead_profile.get("company_size", 0)
if product["company_size"] == "1-10 employees" and company_size <= 10:
score += 30
elif product["company_size"] == "10-100 employees" and 10 <= company_size <= 100:
score += 30
elif product["company_size"] == "100+ employees" and company_size > 100:
score += 30
# Technical level fit
tech_level = lead_profile.get("technical_level", "beginner")
if product["technical_level"] == tech_level:
score += 25
# Use case alignment
lead_use_cases = needs_assessment.get("use_cases", [])
matching_use_cases = set(lead_use_cases) & set(product["use_cases"])
score += len(matching_use_cases) * 10
# Industry fit
lead_industry = lead_profile.get("industry", "")
if lead_industry in product.get("target_industries", []):
score += 15
# Budget alignment
budget = lead_profile.get("budget", 0)
if product["price"] == "Custom pricing":
if budget > 1000: # Enterprise budget
score += 20
else:
product_price = int(product["price"].replace("$", "").replace("/month", ""))
if budget >= product_price:
score += 20
return min(score, 100) # Cap at 100
def generate_recommendation_reasoning(self, product, lead_profile, needs_assessment):
"""Generate explanation for why this product is recommended"""
reasons = []
# Add specific reasons based on fit factors
company_size = lead_profile.get("company_size", 0)
if product["company_size"] == "1-10 employees" and company_size <= 10:
reasons.append("Perfect fit for your team size")
use_cases = needs_assessment.get("use_cases", [])
matching_cases = set(use_cases) & set(product["use_cases"])
if matching_cases:
reasons.append(f"Supports your key use cases: {', '.join(matching_cases)}")
pain_points = needs_assessment.get("pain_points", [])
for pain_point in pain_points:
if pain_point.lower() in [feature.lower() for feature in product["features"]]:
reasons.append(f"Addresses your '{pain_point}' challenge")
return reasonsPersonalized Presentations
Dynamic Presentation Generator
class PresentationGenerator:
def __init__(self):
self.presentation_templates = self.load_templates()
def generate_personalized_presentation(self, lead_profile, recommended_products):
"""Generate a personalized presentation"""
presentation_prompt = f"""
Create a compelling, personalized product presentation for this prospect:
PROSPECT PROFILE:
- Company: {lead_profile.get('company', 'Unknown')}
- Industry: {lead_profile.get('industry', 'Unknown')}
- Size: {lead_profile.get('company_size', 'Unknown')} employees
- Role: {lead_profile.get('role', 'Unknown')}
- Main challenges: {lead_profile.get('pain_points', [])}
- Goals: {lead_profile.get('goals', [])}
RECOMMENDED SOLUTION:
{self.format_product_recommendation(recommended_products[0])}
PRESENTATION STRUCTURE:
1. Problem acknowledgment - Start by acknowledging their specific challenges
2. Solution overview - Present our solution in context of their needs
3. Key benefits - Focus on benefits that matter to them specifically
4. Success story - Share relevant case study from similar company
5. ROI projection - Quantify potential value for their situation
6. Implementation - Outline next steps and timeline
GUIDELINES:
- Make it conversational, not a formal pitch
- Use their company name and specific pain points
- Focus on outcomes, not just features
- Include specific metrics and ROI when possible
- End with a clear call to action
Keep each section concise but compelling.
"""
return self.generate_response(presentation_prompt)
def format_product_recommendation(self, recommendation):
"""Format product recommendation for presentation"""
product = recommendation["product"]
reasoning = recommendation["reasoning"]
return f"""
Product: {product["name"]} ({product["price"]})
Key Features:
{chr(10).join(f"• {feature}" for feature in product["features"])}
Why it's right for you:
{chr(10).join(f"• {reason}" for reason in reasoning)}
Fit Score: {recommendation["fit_score"]}/100
"""Step 3: Objection Handling System
Common Objections Database
Objection Recognition and Response
class ObjectionHandler:
def __init__(self):
self.objection_patterns = self.load_objection_patterns()
self.response_frameworks = self.load_response_frameworks()
def load_objection_patterns(self):
"""Load common sales objections and patterns"""
return {
"price": {
"patterns": [
"too expensive", "can't afford", "budget constraints",
"cheaper option", "price is high", "cost too much"
],
"category": "price",
"urgency": "high"
},
"timing": {
"patterns": [
"not ready", "bad timing", "maybe later", "in the future",
"next year", "too busy", "other priorities"
],
"category": "timing",
"urgency": "medium"
},
"authority": {
"patterns": [
"need approval", "talk to my boss", "committee decision",
"not my call", "someone else decides"
],
"category": "authority",
"urgency": "medium"
},
"need": {
"patterns": [
"don't need", "current solution works", "not a priority",
"happy with what we have", "no problems"
],
"category": "need",
"urgency": "high"
},
"trust": {
"patterns": [
"never heard of you", "too new", "not established",
"need references", "proof it works"
],
"category": "trust",
"urgency": "high"
},
"competition": {
"patterns": [
"looking at others", "comparing options", "competitor mentioned",
"already have quotes"
],
"category": "competition",
"urgency": "medium"
}
}
def detect_objection(self, user_input):
"""Detect objection type from user input"""
detected_objections = []
for objection_type, objection_data in self.objection_patterns.items():
for pattern in objection_data["patterns"]:
if pattern.lower() in user_input.lower():
detected_objections.append({
"type": objection_type,
"pattern": pattern,
"category": objection_data["category"],
"urgency": objection_data["urgency"]
})
break
return detected_objections
def handle_objection(self, objection_type, user_input, context):
"""Handle specific objection using appropriate framework"""
if objection_type == "price":
return self.handle_price_objection(user_input, context)
elif objection_type == "timing":
return self.handle_timing_objection(user_input, context)
elif objection_type == "authority":
return self.handle_authority_objection(user_input, context)
elif objection_type == "need":
return self.handle_need_objection(user_input, context)
elif objection_type == "trust":
return self.handle_trust_objection(user_input, context)
elif objection_type == "competition":
return self.handle_competition_objection(user_input, context)
else:
return self.handle_generic_objection(user_input, context)
def handle_price_objection(self, user_input, context):
"""Handle price-related objections"""
price_handling_prompt = f"""
The prospect has raised a price objection: "{user_input}"
Use this framework to respond:
1. ACKNOWLEDGE: "I understand price is an important consideration..."
2. ISOLATE: "If we could work out the investment, is there anything else that would prevent you from moving forward?"
3. VALUE: Reinforce the value and ROI:
- Cost of not solving their problem
- Quantified benefits and savings
- ROI timeframe
4. REFRAME:
- "Investment in solution vs. cost of problem"
- "Cost per employee/month breaks down to..."
- "Compared to cost of [their current approach]..."
5. OPTIONS: Offer alternatives:
- Different pricing tiers
- Phased implementation
- Payment terms
- Pilot program
Context: {context}
Be empathetic but confident in your value proposition.
"""
return self.generate_response(price_handling_prompt)
def handle_timing_objection(self, user_input, context):
"""Handle timing-related objections"""
timing_handling_prompt = f"""
The prospect has raised a timing objection: "{user_input}"
Response framework:
1. UNDERSTAND: "Help me understand what makes now not the right time?"
2. EXPLORE CONSEQUENCES:
- "What happens if you wait to address this?"
- "How much is this costing you each month you delay?"
- "What would need to change for timing to be right?"
3. CREATE URGENCY:
- Limited-time offers or bonuses
- Seasonal factors affecting their business
- Competitive advantages of acting now
- Implementation timeline benefits
4. ALTERNATIVE APPROACHES:
- Pilot program starting now
- Phased implementation
- Preparation work while waiting
Context: {context}
Focus on opportunity cost and competitive advantage.
"""
return self.generate_response(timing_handling_prompt)
def handle_authority_objection(self, user_input, context):
"""Handle authority/decision-making objections"""
authority_handling_prompt = f"""
The prospect has raised an authority objection: "{user_input}"
Response approach:
1. CLARIFY PROCESS: "Help me understand your decision-making process..."
2. IDENTIFY STAKEHOLDERS:
- "Who else would be involved in evaluating this?"
- "What criteria are most important to [decision maker]?"
- "What concerns might they have?"
3. BUILD COALITION:
- "How can I help you build a case for this?"
- "What information would be most valuable to share?"
- "Would it be helpful if I presented to the team?"
4. PROVIDE TOOLS:
- ROI calculator
- Executive summary
- Comparison charts
- Implementation timeline
Context: {context}
Position yourself as a helpful resource, not a pushy salesperson.
"""
return self.generate_response(authority_handling_prompt)Step 4: CRM Integration
Salesforce Integration
Lead Management and Tracking
from simple_salesforce import Salesforce
class SalesforceCRMIntegration:
def __init__(self, username, password, security_token):
self.sf = Salesforce(
username=username,
password=password,
security_token=security_token
)
def create_or_update_lead(self, lead_data, conversation_data):
"""Create or update lead in Salesforce"""
# Check if lead exists
existing_lead = self.find_existing_lead(lead_data["email"])
if existing_lead:
return self.update_lead(existing_lead["Id"], lead_data, conversation_data)
else:
return self.create_lead(lead_data, conversation_data)
def create_lead(self, lead_data, conversation_data):
"""Create new lead in Salesforce"""
lead_record = {
"FirstName": lead_data.get("first_name", ""),
"LastName": lead_data.get("last_name", "Unknown"),
"Email": lead_data["email"],
"Company": lead_data.get("company", "Unknown"),
"Title": lead_data.get("title", ""),
"Phone": lead_data.get("phone", ""),
"Website": lead_data.get("website", ""),
"Industry": lead_data.get("industry", ""),
"LeadSource": "AI Sales Assistant",
"Status": self.determine_lead_status(conversation_data),
"Rating": self.determine_lead_rating(conversation_data),
# Custom fields
"AI_Qualification_Score__c": conversation_data.get("qualification_score", 0),
"AI_Conversation_Summary__c": conversation_data.get("summary", ""),
"Recommended_Product__c": conversation_data.get("recommended_product", ""),
"Next_Action__c": conversation_data.get("next_action", ""),
"BANT_Budget__c": conversation_data.get("bant_scores", {}).get("budget", 0),
"BANT_Authority__c": conversation_data.get("bant_scores", {}).get("authority", 0),
"BANT_Need__c": conversation_data.get("bant_scores", {}).get("need", 0),
"BANT_Timeline__c": conversation_data.get("bant_scores", {}).get("timeline", 0)
}
result = self.sf.Lead.create(lead_record)
# Create activity record
self.create_activity_record(result["id"], conversation_data)
return result
def update_lead(self, lead_id, lead_data, conversation_data):
"""Update existing lead with new conversation data"""
update_data = {
"Status": self.determine_lead_status(conversation_data),
"Rating": self.determine_lead_rating(conversation_data),
"AI_Qualification_Score__c": conversation_data.get("qualification_score", 0),
"AI_Last_Interaction__c": datetime.utcnow().isoformat(),
"AI_Conversation_Count__c": conversation_data.get("interaction_count", 1),
"Recommended_Product__c": conversation_data.get("recommended_product", ""),
"Next_Action__c": conversation_data.get("next_action", "")
}
result = self.sf.Lead.update(lead_id, update_data)
# Create activity record
self.create_activity_record(lead_id, conversation_data)
return result
def create_activity_record(self, lead_id, conversation_data):
"""Create activity record for AI conversation"""
activity_record = {
"Subject": f"AI Sales Assistant Conversation - {conversation_data.get('stage', 'Discovery')}",
"Description": conversation_data.get("conversation_summary", ""),
"Status": "Completed",
"Type": "AI Conversation",
"WhoId": lead_id,
"ActivityDate": datetime.utcnow().date().isoformat(),
"Duration_in_Minutes__c": conversation_data.get("duration_minutes", 0),
"AI_Sentiment__c": conversation_data.get("sentiment", "Neutral"),
"AI_Intent__c": conversation_data.get("intent", "Information"),
"Objections_Raised__c": ", ".join(conversation_data.get("objections", [])),
"Follow_Up_Required__c": conversation_data.get("follow_up_required", False)
}
return self.sf.Task.create(activity_record)
def determine_lead_status(self, conversation_data):
"""Determine appropriate lead status based on conversation"""
quality = conversation_data.get("quality", "Low")
stage = conversation_data.get("stage", "discovery")
if quality == "High" and stage in ["presentation", "closing"]:
return "Working - Contacted"
elif quality == "Medium":
return "Open - Not Contacted"
else:
return "Unqualified"
def determine_lead_rating(self, conversation_data):
"""Determine lead rating based on qualification score"""
score = conversation_data.get("qualification_score", 0)
if score >= 0.8:
return "Hot"
elif score >= 0.6:
return "Warm"
else:
return "Cold"
def get_lead_history(self, email):
"""Get lead interaction history"""
# Query lead
lead_query = f"SELECT Id, Name, Email, Company, Status, Rating FROM Lead WHERE Email = '{email}'"
leads = self.sf.query(lead_query)
if not leads["records"]:
return None
lead = leads["records"][0]
# Get activities
activities_query = f"SELECT Id, Subject, Description, CreatedDate, Type FROM Task WHERE WhoId = '{lead['Id']}' ORDER BY CreatedDate DESC"
activities = self.sf.query(activities_query)
return {
"lead": lead,
"activities": activities["records"]
}HubSpot Integration
Alternative CRM Integration
import hubspot
from hubspot.crm.contacts import SimplePublicObjectInput
class HubSpotCRMIntegration:
def __init__(self, api_key):
self.client = hubspot.Client.create(api_token=api_key)
def create_or_update_contact(self, lead_data, conversation_data):
"""Create or update contact in HubSpot"""
# Check if contact exists
existing_contact = self.find_contact_by_email(lead_data["email"])
if existing_contact:
return self.update_contact(existing_contact.id, lead_data, conversation_data)
else:
return self.create_contact(lead_data, conversation_data)
def create_contact(self, lead_data, conversation_data):
"""Create new contact in HubSpot"""
properties = {
"email": lead_data["email"],
"firstname": lead_data.get("first_name", ""),
"lastname": lead_data.get("last_name", "Unknown"),
"company": lead_data.get("company", ""),
"jobtitle": lead_data.get("title", ""),
"phone": lead_data.get("phone", ""),
"website": lead_data.get("website", ""),
"industry": lead_data.get("industry", ""),
"lifecyclestage": self.determine_lifecycle_stage(conversation_data),
"lead_status": self.determine_lead_status(conversation_data),
# Custom properties
"ai_qualification_score": conversation_data.get("qualification_score", 0),
"ai_conversation_summary": conversation_data.get("summary", ""),
"recommended_product": conversation_data.get("recommended_product", ""),
"next_action": conversation_data.get("next_action", "")
}
simple_public_object_input = SimplePublicObjectInput(properties=properties)
result = self.client.crm.contacts.basic_api.create(simple_public_object_input)
# Create note
self.create_engagement_note(result.id, conversation_data)
return result
def create_engagement_note(self, contact_id, conversation_data):
"""Create engagement note for conversation"""
note_properties = {
"hs_note_body": f"""
AI Sales Assistant Conversation Summary
Stage: {conversation_data.get('stage', 'Discovery')}
Quality: {conversation_data.get('quality', 'Unknown')}
Qualification Score: {conversation_data.get('qualification_score', 0)}/1.0
Key Points:
{conversation_data.get('conversation_summary', 'No summary available')}
Recommended Product: {conversation_data.get('recommended_product', 'TBD')}
Next Action: {conversation_data.get('next_action', 'Follow up required')}
BANT Scores:
- Budget: {conversation_data.get('bant_scores', {}).get('budget', 0)}/1.0
- Authority: {conversation_data.get('bant_scores', {}).get('authority', 0)}/1.0
- Need: {conversation_data.get('bant_scores', {}).get('need', 0)}/1.0
- Timeline: {conversation_data.get('bant_scores', {}).get('timeline', 0)}/1.0
Objections Raised: {', '.join(conversation_data.get('objections', ['None']))}
""",
"hs_timestamp": int(time.time() * 1000)
}
note_input = SimplePublicObjectInput(properties=note_properties)
result = self.client.crm.objects.notes.basic_api.create(note_input)
# Associate note with contact
self.client.crm.objects.notes.associations_api.create(
object_id=result.id,
to_object_type="contacts",
to_object_id=contact_id,
association_type="note_to_contact"
)
return resultStep 5: Meeting Scheduling Integration
Calendar Integration
Automated Meeting Scheduling
import calendar_integration # Calendly, Google Calendar, etc.
class MeetingScheduler:
def __init__(self, calendar_provider="calendly"):
self.calendar_provider = calendar_provider
self.scheduler = self.init_scheduler()
def suggest_meeting(self, lead_data, conversation_context):
"""Suggest appropriate meeting based on lead qualification"""
quality = conversation_context.get("quality", "Low")
stage = conversation_context.get("stage", "discovery")
if quality == "High" and stage in ["presentation", "closing"]:
return self.suggest_demo_call(lead_data, conversation_context)
elif quality == "Medium":
return self.suggest_discovery_call(lead_data, conversation_context)
else:
return self.suggest_educational_session(lead_data, conversation_context)
def suggest_demo_call(self, lead_data, conversation_context):
"""Suggest demo call for qualified prospects"""
meeting_suggestion = {
"type": "product_demo",
"duration": 30,
"title": f"BroxiAI Demo for {lead_data.get('company', 'Your Team')}",
"description": f"""
Thank you for your interest in BroxiAI! Based on our conversation, I'd love to show you how our platform can specifically help with:
• {conversation_context.get('primary_pain_point', 'Your automation needs')}
• {conversation_context.get('recommended_product', 'Our recommended solution')}
This 30-minute demo will be customized to your specific use case and requirements.
What to expect:
• Live demonstration of BroxiAI features
• Personalized use case walkthrough
• Q&A session
• Next steps discussion
Please let me know what time works best for you!
""",
"meeting_url": self.generate_meeting_link("demo", lead_data),
"urgency": "high"
}
return meeting_suggestion
def suggest_discovery_call(self, lead_data, conversation_context):
"""Suggest discovery call for medium-qualified prospects"""
meeting_suggestion = {
"type": "discovery_call",
"duration": 20,
"title": f"Discovery Call - {lead_data.get('company', 'Your Business')} AI Needs",
"description": f"""
I'd love to learn more about your specific needs and see how BroxiAI might be able to help.
This brief 20-minute call will help us:
• Better understand your current challenges
• Explore potential AI automation opportunities
• Determine if BroxiAI is a good fit
• Answer any questions you have
No pressure - just a friendly conversation about your needs!
""",
"meeting_url": self.generate_meeting_link("discovery", lead_data),
"urgency": "medium"
}
return meeting_suggestion
def generate_meeting_link(self, meeting_type, lead_data):
"""Generate appropriate meeting link based on type"""
if self.calendar_provider == "calendly":
base_url = "https://calendly.com/broxi-sales"
if meeting_type == "demo":
return f"{base_url}/30-minute-demo"
elif meeting_type == "discovery":
return f"{base_url}/20-minute-discovery"
else:
return f"{base_url}/15-minute-chat"
# Add other calendar providers as needed
return "https://broxi.ai/schedule"
def schedule_follow_up(self, lead_data, conversation_context, follow_up_type="standard"):
"""Schedule automated follow-up based on conversation outcome"""
follow_up_sequences = {
"high_intent": {
"immediate": "Send demo link and meeting request",
"1_hour": "Follow up if no meeting scheduled",
"1_day": "Send case study and ROI calculator",
"3_days": "Personal video message from sales rep"
},
"medium_intent": {
"4_hours": "Send educational content",
"2_days": "Follow up with discovery call invitation",
"1_week": "Send industry-specific use cases",
"2_weeks": "Check in with new features/updates"
},
"low_intent": {
"1_day": "Add to nurture sequence",
"1_week": "Send weekly newsletter",
"1_month": "Re-qualification attempt",
"3_months": "Seasonal check-in"
}
}
intent_level = self.determine_intent_level(conversation_context)
sequence = follow_up_sequences.get(intent_level, follow_up_sequences["low_intent"])
return sequenceStep 6: Performance Analytics
Comprehensive Analytics Dashboard
Sales Performance Tracking
class SalesAnalytics:
def __init__(self):
self.metrics_collector = MetricsCollector()
self.reporting_engine = ReportingEngine()
def track_conversation_metrics(self, conversation_data):
"""Track key conversation metrics"""
metrics = {
"conversation_id": conversation_data["id"],
"timestamp": datetime.utcnow(),
"duration_minutes": conversation_data.get("duration", 0),
"message_count": conversation_data.get("message_count", 0),
"qualification_score": conversation_data.get("qualification_score", 0),
"lead_quality": conversation_data.get("quality", "Low"),
"stage_reached": conversation_data.get("final_stage", "greeting"),
"objections_count": len(conversation_data.get("objections", [])),
"meeting_scheduled": conversation_data.get("meeting_scheduled", False),
"product_recommended": conversation_data.get("recommended_product", ""),
"sentiment_score": conversation_data.get("sentiment", 0),
"conversion_outcome": conversation_data.get("outcome", "ongoing")
}
self.metrics_collector.record_metrics(metrics)
return metrics
def generate_daily_report(self, date=None):
"""Generate daily sales assistant performance report"""
if not date:
date = datetime.utcnow().date()
daily_metrics = self.metrics_collector.get_daily_metrics(date)
report = {
"date": date.isoformat(),
"summary": {
"total_conversations": daily_metrics["conversation_count"],
"high_quality_leads": daily_metrics["high_quality_count"],
"meetings_scheduled": daily_metrics["meetings_scheduled"],
"average_qualification_score": daily_metrics["avg_qualification_score"],
"conversion_rate": daily_metrics["conversion_rate"]
},
"conversation_stages": {
"greeting": daily_metrics["stage_breakdown"]["greeting"],
"discovery": daily_metrics["stage_breakdown"]["discovery"],
"qualification": daily_metrics["stage_breakdown"]["qualification"],
"presentation": daily_metrics["stage_breakdown"]["presentation"],
"closing": daily_metrics["stage_breakdown"]["closing"]
},
"lead_quality_distribution": {
"high": daily_metrics["quality_breakdown"]["high"],
"medium": daily_metrics["quality_breakdown"]["medium"],
"low": daily_metrics["quality_breakdown"]["low"]
},
"common_objections": daily_metrics["top_objections"],
"product_recommendations": daily_metrics["product_breakdown"],
"performance_trends": self.calculate_trends(date)
}
return report
def calculate_roi_metrics(self, period_days=30):
"""Calculate ROI metrics for sales assistant"""
metrics = self.metrics_collector.get_period_metrics(period_days)
# Calculate costs
ai_costs = metrics["total_ai_api_costs"]
platform_costs = metrics["platform_costs"]
total_costs = ai_costs + platform_costs
# Calculate revenue impact
meetings_scheduled = metrics["total_meetings_scheduled"]
estimated_close_rate = 0.25 # 25% close rate assumption
average_deal_size = metrics["average_deal_size"]
estimated_revenue = meetings_scheduled * estimated_close_rate * average_deal_size
# Calculate time savings
conversations_handled = metrics["total_conversations"]
avg_conversation_time = metrics["avg_conversation_duration"]
human_equivalent_time = conversations_handled * avg_conversation_time
hourly_rate = 50 # Sales rep hourly rate
time_savings_value = human_equivalent_time * hourly_rate / 60
roi_metrics = {
"period_days": period_days,
"total_costs": total_costs,
"estimated_revenue": estimated_revenue,
"time_savings_value": time_savings_value,
"total_value": estimated_revenue + time_savings_value,
"roi_percentage": ((estimated_revenue + time_savings_value - total_costs) / total_costs) * 100,
"conversations_handled": conversations_handled,
"meetings_generated": meetings_scheduled,
"cost_per_meeting": total_costs / meetings_scheduled if meetings_scheduled > 0 else 0,
"cost_per_conversation": total_costs / conversations_handled if conversations_handled > 0 else 0
}
return roi_metricsStep 7: Advanced Features
A/B Testing Framework
Conversation Optimization
class ConversationABTesting:
def __init__(self):
self.test_variants = {}
self.results_tracker = {}
def create_ab_test(self, test_name, variants):
"""Create A/B test for conversation approaches"""
self.test_variants[test_name] = {
"variants": variants,
"traffic_split": 1.0 / len(variants), # Equal split
"success_metric": "qualification_score",
"minimum_sample_size": 100,
"start_date": datetime.utcnow(),
"status": "active"
}
# Initialize results tracking
for variant in variants:
self.results_tracker[f"{test_name}_{variant['name']}"] = {
"conversations": 0,
"total_score": 0,
"meetings_scheduled": 0,
"objections_handled": 0
}
def get_variant_for_conversation(self, test_name, conversation_id):
"""Get A/B test variant for conversation"""
if test_name not in self.test_variants:
return None
test = self.test_variants[test_name]
if test["status"] != "active":
return None
# Use conversation ID hash for consistent assignment
import hashlib
hash_value = int(hashlib.md5(conversation_id.encode()).hexdigest(), 16)
variant_index = hash_value % len(test["variants"])
return test["variants"][variant_index]
def record_test_result(self, test_name, variant_name, conversation_result):
"""Record A/B test conversation result"""
key = f"{test_name}_{variant_name}"
if key not in self.results_tracker:
return
tracker = self.results_tracker[key]
tracker["conversations"] += 1
tracker["total_score"] += conversation_result.get("qualification_score", 0)
if conversation_result.get("meeting_scheduled", False):
tracker["meetings_scheduled"] += 1
tracker["objections_handled"] += len(conversation_result.get("objections", []))
def analyze_test_results(self, test_name):
"""Analyze A/B test results and determine winner"""
if test_name not in self.test_variants:
return None
test = self.test_variants[test_name]
results = {}
for variant in test["variants"]:
key = f"{test_name}_{variant['name']}"
tracker = self.results_tracker[key]
if tracker["conversations"] > 0:
results[variant["name"]] = {
"conversations": tracker["conversations"],
"avg_qualification_score": tracker["total_score"] / tracker["conversations"],
"meeting_rate": tracker["meetings_scheduled"] / tracker["conversations"],
"objection_handling_rate": tracker["objections_handled"] / tracker["conversations"]
}
# Determine statistical significance and winner
winner = self.determine_statistical_winner(results)
return {
"test_name": test_name,
"results": results,
"winner": winner,
"confidence_level": winner.get("confidence", 0) if winner else 0,
"recommendation": self.get_test_recommendation(results, winner)
}Sentiment Analysis Integration
Real-time Sentiment Monitoring
from textblob import TextBlob
import re
class SentimentAnalyzer:
def __init__(self):
self.sentiment_thresholds = {
"very_positive": 0.5,
"positive": 0.1,
"neutral": -0.1,
"negative": -0.5
}
def analyze_conversation_sentiment(self, conversation_history):
"""Analyze sentiment throughout conversation"""
sentiment_timeline = []
for message in conversation_history:
if message["role"] == "user":
sentiment = self.analyze_message_sentiment(message["content"])
sentiment_timeline.append({
"timestamp": message.get("timestamp", datetime.utcnow()),
"sentiment_score": sentiment["score"],
"sentiment_label": sentiment["label"],
"message": message["content"]
})
# Calculate overall sentiment trends
overall_sentiment = self.calculate_overall_sentiment(sentiment_timeline)
sentiment_trend = self.calculate_sentiment_trend(sentiment_timeline)
return {
"timeline": sentiment_timeline,
"overall_sentiment": overall_sentiment,
"sentiment_trend": sentiment_trend,
"alerts": self.generate_sentiment_alerts(sentiment_timeline)
}
def analyze_message_sentiment(self, message):
"""Analyze sentiment of individual message"""
# Clean message
cleaned_message = self.clean_message(message)
# Use TextBlob for basic sentiment analysis
blob = TextBlob(cleaned_message)
polarity = blob.sentiment.polarity # -1 to 1 scale
# Determine sentiment label
if polarity >= self.sentiment_thresholds["very_positive"]:
label = "very_positive"
elif polarity >= self.sentiment_thresholds["positive"]:
label = "positive"
elif polarity >= self.sentiment_thresholds["neutral"]:
label = "neutral"
elif polarity >= self.sentiment_thresholds["negative"]:
label = "negative"
else:
label = "very_negative"
return {
"score": polarity,
"label": label,
"confidence": abs(polarity)
}
def generate_sentiment_alerts(self, sentiment_timeline):
"""Generate alerts based on sentiment patterns"""
alerts = []
# Check for negative sentiment
recent_messages = sentiment_timeline[-3:] if len(sentiment_timeline) >= 3 else sentiment_timeline
recent_scores = [msg["sentiment_score"] for msg in recent_messages]
if len(recent_scores) >= 2:
avg_recent_sentiment = sum(recent_scores) / len(recent_scores)
if avg_recent_sentiment < -0.3:
alerts.append({
"type": "negative_sentiment",
"severity": "high",
"message": "Customer sentiment has turned negative",
"recommendation": "Consider escalating to human agent or addressing concerns"
})
elif avg_recent_sentiment < -0.1:
alerts.append({
"type": "declining_sentiment",
"severity": "medium",
"message": "Customer sentiment is declining",
"recommendation": "Acknowledge concerns and adjust approach"
})
# Check for sudden sentiment drops
if len(sentiment_timeline) >= 2:
current_sentiment = sentiment_timeline[-1]["sentiment_score"]
previous_sentiment = sentiment_timeline[-2]["sentiment_score"]
if current_sentiment - previous_sentiment < -0.4:
alerts.append({
"type": "sudden_sentiment_drop",
"severity": "high",
"message": "Sudden negative sentiment change detected",
"recommendation": "Immediate attention required - potential objection or concern"
})
return alertsBest Practices
Sales Assistant Optimization
Conversation Flow Optimization
Start with rapport building before qualification
Use natural, conversational language
Ask one question at a time
Listen actively and respond to specific points
Qualify naturally through conversation, not interrogation
Lead Scoring Accuracy
Continuously refine scoring algorithms based on actual outcomes
Include multiple data points beyond BANT
Consider industry-specific qualification criteria
Update scoring models based on closed-won/lost analysis
Balance automation with human judgment
Objection Handling Excellence
Maintain database of common objections and proven responses
Train on industry-specific objections
Use frameworks (Acknowledge, Isolate, Value, Close)
Escalate complex objections to human reps
Track objection resolution success rates
Integration Best Practices
CRM Data Quality
Ensure consistent data mapping between systems
Implement data validation and deduplication
Regular data cleanup and enrichment
Track data quality metrics
Maintain audit trails for compliance
Performance Monitoring
Set up real-time dashboards for key metrics
Monitor conversation quality scores
Track conversion rates by traffic source
Analyze drop-off points in conversations
Regular A/B testing of conversation approaches
Troubleshooting
Common Issues
Low Qualification Rates
Review and refine discovery questions
Improve rapport building techniques
Adjust qualification criteria for your market
Analyze lost opportunities for patterns
Train on industry-specific pain points
High Objection Rates
Review pricing presentation approach
Improve value proposition messaging
Address common objections proactively
Enhance trust-building content
Consider market positioning adjustments
Poor CRM Data Quality
Implement data validation rules
Regular data cleanup processes
Train team on data entry standards
Use data enrichment services
Set up automated data quality alerts
Next Steps
After implementing your sales assistant:
Performance Optimization: Continuously monitor and improve conversation quality
Training Enhancement: Regular updates to conversation scripts and objection handling
Integration Expansion: Connect additional tools and data sources
Team Training: Ensure smooth handoffs between AI and human sales reps
Scale Gradually: Expand to handle more complex sales scenarios
Related Examples
Customer Support: Cross-functional AI agent patterns
Document Q&A: Knowledge base integration
Basic Chatbot: Foundational conversation patterns
You've built a comprehensive AI-powered sales assistant! This system can handle lead qualification, product recommendations, and objection handling while seamlessly integrating with your existing sales infrastructure and processes.
Last updated