File size: 8,228 Bytes
e3e7ab8 3dd44e9 e3e7ab8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 |
from smolagents import VisitWebpageTool, InferenceClientModel, CodeAgent, WebSearchTool, tool
from flask import Flask, request, jsonify
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.deep_crawling import BFSDeepCrawlStrategy
from crawl4ai.content_scraping_strategy import LXMLWebScrapingStrategy
from system_prompt import get_system_prompt
import xml.etree.ElementTree as ET
app = Flask(__name__)
model = InferenceClientModel(model_id="meta-llama/Llama-3.3-70B-Instruct", provider="together")
crawler_config = CrawlerRunConfig(
deep_crawl_strategy=BFSDeepCrawlStrategy(
max_depth=0,
include_external=False
),
scraping_strategy=LXMLWebScrapingStrategy(),
verbose=True
)
def extract_vehicle_info_as_string(adf_xml):
root = ET.fromstring(adf_xml)
# Find the vehicle element
vehicle = root.find('.//vehicle')
if vehicle is not None:
year = vehicle.find('year').text if vehicle.find('year') is not None else ""
make = vehicle.find('make').text if vehicle.find('make') is not None else ""
model = vehicle.find('model').text if vehicle.find('model') is not None else ""
vehicle_info = f"{year} {make} {model}".strip()
# Extract first name
first_name = ""
name_element = root.find('.//name[@part="first"]')
if name_element is not None:
first_name = name_element.text.strip() if name_element.text else ""
return first_name, vehicle_info
def safe_get_attr(obj, attr, default=None):
try:
return getattr(obj, attr, default)
except (AttributeError, TypeError):
return default
@tool
def custom_site_crawler(website_url: str) -> str:
"""
Crawl the car gurus site for the specific dealership and return the markdown content or crawl the dealership website and return the markdown content. Use this when the user asks about the car or dealership and cannot be answered by the ADF Lead.
Args:
website_url: The url of the website (car gurus or dealership website) to crawl.
Returns:
A string of markdown content of the website.
"""
import asyncio
async def _crawl_site(url):
async with AsyncWebCrawler() as crawler:
try:
result_container = await crawler.arun(url, config=crawler_config)
# Handle different result types from crawler
if hasattr(result_container, '__iter__') and not isinstance(result_container, str):
try:
# Try to get first result if it's a list-like container
result = next(iter(result_container))
except (StopIteration, TypeError):
result = result_container
else:
result = result_container
markdown_result = safe_get_attr(result, 'markdown', None)
if markdown_result:
raw_markdown = safe_get_attr(markdown_result, 'raw_markdown', '')
return raw_markdown if raw_markdown else ""
else:
return ""
except Exception as e:
print(f"Error crawling {url}: {e}")
return ""
return asyncio.run(_crawl_site(website_url))
agent = CodeAgent(tools=[VisitWebpageTool(), WebSearchTool(), custom_site_crawler], model=model, additional_authorized_imports=["xml.etree.ElementTree"])
dealership_phone = "(513) 800-0805"
car_gurus_site = "https://www.cargurus.com/Cars/m-Ohio-Cars-sp458596"
car_site = "https://www.Ohiocars.com"
adf_lead = "<?xml version=\"1.0\"?><?ADF version=\"1.0\"?><adf><prospect><requestdate>2025-05-12T13:59:30</requestdate><vehicle status=\"used\"><id source=\"CarsForSale.com\">16f3114e-825f-4eb0-8165-ce43fe5143b6</id><year>2016</year><make>Toyota</make><model>Corolla</model><vin>5YFBURHE4GP511115</vin><stock></stock><comments>DP</comments><colorcombination><exteriorcolor>Super White</exteriorcolor></colorcombination><miles>131024.0</miles><price type=\"asking\">9950</price></vehicle><customer><contact><name part=\"first\">Test</name><name part=\"last\">Lead</name><name part=\"full\">Test Lead</name><email>[email protected]</email><phone>2582584568</phone><address><city></city><state></state><postalcode></postalcode></address></contact><comments><![CDATA[I'm interested and want to know more about the 2016 Toyota Corolla S Plus you have listed for $9,950 on Cars For Sale.]]></comments><timeframe><description></description></timeframe></customer><provider><id>19971</id><name part=\"full\">Carsforsale.com</name><service>Carsforsale.com</service><phone>866-388-9778</phone></provider><vendor><id>114483</id><vendorname>Ohio Cars</vendorname></vendor></prospect></adf>"
first_name, vehicle_info = extract_vehicle_info_as_string(adf_lead)
agent.prompt_templates["system_prompt"] = agent.prompt_templates["system_prompt"] + get_system_prompt(car_gurus_site, car_site, adf_lead, dealership_phone)
@app.route('/chat', methods=['POST'])
def chat():
"""
Main chat endpoint to interact with the agent.
Expects JSON payload with 'message' field.
"""
try:
data = request.get_json()
if not data or 'message' not in data:
return jsonify({'error': 'Missing message field in request'}), 400
message = data['message']
response = agent.run(message)
return jsonify({
'response': response,
'status': 'success'
})
except Exception as e:
return jsonify({
'error': str(e),
'status': 'error'
}), 500
@app.route('/crawl', methods=['POST'])
def crawl_website():
"""
Endpoint to directly crawl a website using the custom site crawler.
Expects JSON payload with 'url' field.
"""
try:
data = request.get_json()
if not data or 'url' not in data:
return jsonify({'error': 'Missing url field in request'}), 400
url = data['url']
content = custom_site_crawler(url)
return jsonify({
'content': content,
'url': url,
'status': 'success'
})
except Exception as e:
return jsonify({
'error': str(e),
'status': 'error'
}), 500
@app.route('/health', methods=['GET'])
def health_check():
"""Health check endpoint."""
return jsonify({'status': 'healthy'})
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=4000)
# "(513) 800-0805"
# "https://www.cargurus.com/Cars/m-Ohio-Cars-sp458596"
# "https://www.Ohiocars.com"
# "<?xml version=\"1.0\"?><?ADF version=\"1.0\"?><adf><prospect><requestdate>2025-05-12T13:59:30</requestdate><vehicle status=\"used\"><id source=\"CarsForSale.com\">16f3114e-825f-4eb0-8165-ce43fe5143b6</id><year>2016</year><make>Toyota</make><model>Corolla</model><vin>5YFBURHE4GP511115</vin><stock></stock><comments>DP</comments><colorcombination><exteriorcolor>Super White</exteriorcolor></colorcombination><miles>131024.0</miles><price type=\"asking\">9950</price></vehicle><customer><contact><name part=\"first\">Test</name><name part=\"last\">Lead</name><name part=\"full\">Test Lead</name><email>[email protected]</email><phone>2582584568</phone><address><city></city><state></state><postalcode></postalcode></address></contact><comments><![CDATA[I'm interested and want to know more about the 2016 Toyota Corolla S Plus you have listed for $9,950 on Cars For Sale.]]></comments><timeframe><description></description></timeframe></customer><provider><id>19971</id><name part=\"full\">Carsforsale.com</name><service>Carsforsale.com</service><phone>866-388-9778</phone></provider><vendor><id>114483</id><vendorname>Ohio Cars</vendorname></vendor></prospect></adf>"
# "Hi {first_name}! The {vehicle_info} you're interested in is available at [OhioCars.com](https://www.ohiocars.com). Would you like to schedule a visit to check it out? We have appointment slots at 11 AM, 1 PM, or 3 PM. Which time works best for you?"
# first_name, vehicle_info = extract_vehicle_info_as_string(adf_lead.value)
|