Grok LLM powers the intelligent task inference and context generation features using xAI’s official Python SDK.Model:grok-3-fast API Endpoint:https://api.x.ai/v1/chat Documentation:https://docs.x.ai/docs/guides/chat
Converts free-text user queries into structured service types.Function:infer_task(query: str) -> strExample:
task = await infer_task("my toilet is broken and leaking")# Returns: "plumber"
System Prompt:
You are a service task classifier. Given a user's request, identify the type of service professional needed.Respond with ONLY a single word or short phrase for the service type, such as:- plumber- electrician- house cleaner- painter- handyman- HVAC technician- locksmith
Converts first-person queries into second-person problem descriptions for AI agent.Function:format_problem_statement(original_query: str, task: str) -> strExample:
problem = await format_problem_statement("my lawn is too long", "landscaper")# Returns: "your lawn needs to be mowed"
Price Extraction
Analyzes call transcripts to extract negotiated prices.Function:extract_negotiated_price(transcript: List[Dict]) -> Optional[float]Example:
transcript = [ {"role": "assistant", "text": "How much would you charge?"}, {"role": "user", "text": "I can do it for $175"}, {"role": "assistant", "text": "That sounds good, let's go with $175"}]price = await extract_negotiated_price(transcript)# Returns: 175.0
Grok Realtime API provides low-latency voice conversations via WebSocket for the AI negotiation agent.Endpoint:wss://api.x.ai/v1/realtime Protocol: WebSocket Audio Format: PCM 24kHz (bidirectional)
def generate_prompt(provider: dict) -> str: return f"""YOU ARE a homeowner in zip code {provider['zip_code']} calling for service.{provider['problem']}.YOU ARE NOT an AI. You are a real person speaking to {provider['service_provider']}.Your tone should be casual, direct, and slightly cost-conscious.Refer to the following context if needed: {provider['context_answers']}1. You must begin with: "Hi, is this {provider['service_provider']}?"2. After confirming, ask for a price estimate.3. Negotiate to secure the lowest price, using **${provider['max_price']}** as target.4. Agreeing to a price up to ${provider['max_price']} is acceptable.End the call based on outcome: - No Agreement: "Thank you for the info. I need to think about it." - Price Agreed: "Thank you! I will reach out to you again shortly.""""
OpenAI’s Responses API with web search tool finds local service providers.Model:gpt-4o Tool:web_search_preview API Endpoint:https://api.openai.com/v1/responses
from openai import OpenAIclient = OpenAI( api_key=OPENAI_API_KEY, organization=OPENAI_ORG_API_KEY)search_prompt = f"""Find {task} services near zip code {zip_code}.Search the web for local {task}s and provide a list with:1. Business name2. Phone numberFormat each result as: NAME | PHONEFind up to 5 providers near {zip_code}."""response = client.responses.create( model="gpt-4o", tools=[{"type": "web_search_preview"}], input=search_prompt)full_response = response.output_text
Twilio provides phone call infrastructure for connecting to service providers.API Version: Twilio REST API Media Streams: WebSocket-based audio streaming
try: task = await infer_task(query)except Exception as e: print(f"Grok API exception: {e}") task = _fallback_infer_task(query) # Use rule-based fallback
OpenAI Search Errors
Common errors:
API key missing
Model not available
Search timeout
Handling:
if not OPENAI_API_KEY: print("⚠️ No OPENAI_API_KEY set - using fallback providers") return _fallback_providers(job)try: response = client.responses.create(...)except Exception as e: print(f"OpenAI Search API exception: {e}") return _fallback_providers(job)
Twilio Errors
Common errors:
Invalid phone number format
Call failed (busy, no answer, etc.)
WebSocket disconnection
Handling:
try: client.calls.create(to=phone, from_=FROM_NUMBER, url=twiml_url)except Exception as e: print(f"❌ Failed to dial {provider['service_provider']}: {e}") update_provider_call_status(provider_id, "failed")
Supabase Errors
Common errors:
Connection timeout
Invalid credentials
Row Level Security policy violation
Handling:
try: response = supabase.table("providers").insert(data).execute() if not response.data: raise Exception("Failed to create provider in Supabase")except Exception as e: print(f"Database error: {e}") # Log error, retry, or return error to user