test / chatbot_langgraph.py

add

e983a89 verified 12 months ago

21.7 kB

"""
LangGraph-based Car Finder Chatbot

This implementation uses:
- Template-based SQL queries (no SQL generation)
- LangGraph for agent orchestration
- Tool-based architecture for security
- State management for conversation flow
"""

from typing import TypedDict, Annotated, Optional, Literal
from operator import add
import sqlite3
import os
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel, Field

# Load environment variables
load_dotenv()

# Validate API key
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
    raise ValueError("OPENAI_API_KEY environment variable is not set")

# Initialize LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7, api_key=api_key)

# Load database schema with error handling
try:
    with open('database_schema.txt', 'r') as f:
        SCHEMA_DESCRIPTION = f.read()
except FileNotFoundError:
    raise FileNotFoundError("database_schema.txt not found. Please ensure it exists in the current directory.")

# Constants
MIN_RESULTS = 1
MAX_RESULTS = 20
DB_PATH = 'cars.db'


# ============================================================================
# PYDANTIC MODELS FOR STRUCTURED TOOL INPUTS
# ============================================================================

class SearchParameters(BaseModel):
    """Parameters for searching cars using template-based SQL"""
    min_price: Optional[int] = Field(None, description="Minimum price in USD", ge=0, le=100000)
    max_price: Optional[int] = Field(None, description="Maximum price in USD", ge=0, le=100000)
    fuel_type: Optional[Literal["Gasoline", "Diesel", "Electric", "Hybrid", "Plug-in Hybrid"]] = Field(None, description="Type of fuel")
    is_suv: Optional[bool] = Field(None, description="True for SUVs, False for sedans/coupes")
    min_seating: Optional[int] = Field(None, description="Minimum seating capacity", ge=4, le=8)
    max_seating: Optional[int] = Field(None, description="Maximum seating capacity", ge=4, le=8)
    drivetrain: Optional[Literal["FWD", "RWD", "AWD", "4WD"]] = Field(None, description="Drive system")
    min_fuel_efficiency_city: Optional[float] = Field(None, description="Minimum city MPG", ge=0)
    min_cargo_space: Optional[int] = Field(None, description="Minimum cargo space in cubic feet", ge=0)
    has_sunroof: Optional[bool] = Field(None, description="Must have sunroof")
    has_leather_seats: Optional[bool] = Field(None, description="Must have leather seats")
    has_navigation: Optional[bool] = Field(None, description="Must have navigation system")
    has_backup_camera: Optional[bool] = Field(None, description="Must have backup camera")
    min_safety_rating: Optional[float] = Field(None, description="Minimum safety rating", ge=0, le=5)


# ============================================================================
# SQL TEMPLATE-BASED TOOLS (Secure, No SQL Generation)
# ============================================================================

@tool
def search_cars(
    min_price: Optional[int] = None,
    max_price: Optional[int] = None,
    fuel_type: Optional[str] = None,
    is_suv: Optional[bool] = None,
    min_seating: Optional[int] = None,
    max_seating: Optional[int] = None,
    drivetrain: Optional[str] = None,
    min_fuel_efficiency_city: Optional[float] = None,
    min_cargo_space: Optional[int] = None,
    has_sunroof: Optional[bool] = None,
    has_leather_seats: Optional[bool] = None,
    has_navigation: Optional[bool] = None,
    has_backup_camera: Optional[bool] = None,
    min_safety_rating: Optional[float] = None,
) -> dict:
    """
    Search for cars using a secure template-based SQL query.

    Returns a dictionary with:
    - count: number of cars found
    - cars: list of matching cars (limited to first 20)
    - status: 'too_few', 'good', or 'too_many'
    """
    # Build WHERE clause from parameters using template
    conditions = []
    params = []

    if min_price is not None:
        conditions.append("price >= ?")
        params.append(min_price)

    if max_price is not None:
        conditions.append("price <= ?")
        params.append(max_price)

    if fuel_type is not None:
        conditions.append("fuel_type = ?")
        params.append(fuel_type)

    if is_suv is not None:
        conditions.append("is_suv = ?")
        params.append(1 if is_suv else 0)

    if min_seating is not None:
        conditions.append("seating_capacity >= ?")
        params.append(min_seating)

    if max_seating is not None:
        conditions.append("seating_capacity <= ?")
        params.append(max_seating)

    if drivetrain is not None:
        conditions.append("drivetrain = ?")
        params.append(drivetrain)

    if min_fuel_efficiency_city is not None:
        conditions.append("fuel_efficiency_city >= ?")
        params.append(min_fuel_efficiency_city)

    if min_cargo_space is not None:
        conditions.append("cargo_space >= ?")
        params.append(min_cargo_space)

    if has_sunroof is not None:
        conditions.append("has_sunroof = ?")
        params.append(1 if has_sunroof else 0)

    if has_leather_seats is not None:
        conditions.append("has_leather_seats = ?")
        params.append(1 if has_leather_seats else 0)

    if has_navigation is not None:
        conditions.append("has_navigation = ?")
        params.append(1 if has_navigation else 0)

    if has_backup_camera is not None:
        conditions.append("has_backup_camera = ?")
        params.append(1 if has_backup_camera else 0)

    if min_safety_rating is not None:
        conditions.append("safety_rating >= ?")
        params.append(min_safety_rating)

    # Build SQL query using template
    where_clause = " AND ".join(conditions) if conditions else "1=1"
    query = f"SELECT * FROM cars WHERE {where_clause} ORDER BY price LIMIT 21"  # Get 21 to check if > 20

    try:
        # Use context manager to ensure connection is closed
        with sqlite3.connect(DB_PATH) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute(query, params)
            results = cursor.fetchall()

            # Convert to list of dicts
            cars = [dict(row) for row in results]
            count = len(cars)

            # Determine status
            if count < MIN_RESULTS:
                status = "too_few"
            elif count > MAX_RESULTS:
                status = "too_many"
                cars = cars[:MAX_RESULTS]  # Limit results
            else:
                status = "good"

            return {
                "count": count,
                "cars": cars,
                "status": status,
                "params_used": {k: v for k, v in [
                    ("min_price", min_price),
                    ("max_price", max_price),
                    ("fuel_type", fuel_type),
                    ("is_suv", is_suv),
                    ("min_seating", min_seating),
                    ("max_seating", max_seating),
                    ("drivetrain", drivetrain),
                    ("min_fuel_efficiency_city", min_fuel_efficiency_city),
                    ("min_cargo_space", min_cargo_space),
                    ("has_sunroof", has_sunroof),
                    ("has_leather_seats", has_leather_seats),
                    ("has_navigation", has_navigation),
                    ("has_backup_camera", has_backup_camera),
                    ("min_safety_rating", min_safety_rating),
                ] if v is not None}
            }

    except sqlite3.Error as e:
        return {
            "count": 0,
            "cars": [],
            "status": "error",
            "error": str(e)
        }


@tool
def count_cars_only(
    min_price: Optional[int] = None,
    max_price: Optional[int] = None,
    fuel_type: Optional[str] = None,
    is_suv: Optional[bool] = None,
    min_seating: Optional[int] = None,
    max_seating: Optional[int] = None,
    drivetrain: Optional[str] = None,
    min_fuel_efficiency_city: Optional[float] = None,
    min_cargo_space: Optional[int] = None,
    has_sunroof: Optional[bool] = None,
    has_leather_seats: Optional[bool] = None,
    has_navigation: Optional[bool] = None,
    has_backup_camera: Optional[bool] = None,
    min_safety_rating: Optional[float] = None,
) -> dict:
    """
    Count how many cars match the criteria without returning full results.
    Useful for checking if we need to refine search parameters.
    """
    # Build WHERE clause from parameters
    conditions = []
    params = []

    if min_price is not None:
        conditions.append("price >= ?")
        params.append(min_price)

    if max_price is not None:
        conditions.append("price <= ?")
        params.append(max_price)

    if fuel_type is not None:
        conditions.append("fuel_type = ?")
        params.append(fuel_type)

    if is_suv is not None:
        conditions.append("is_suv = ?")
        params.append(1 if is_suv else 0)

    if min_seating is not None:
        conditions.append("seating_capacity >= ?")
        params.append(min_seating)

    if max_seating is not None:
        conditions.append("seating_capacity <= ?")
        params.append(max_seating)

    if drivetrain is not None:
        conditions.append("drivetrain = ?")
        params.append(drivetrain)

    if min_fuel_efficiency_city is not None:
        conditions.append("fuel_efficiency_city >= ?")
        params.append(min_fuel_efficiency_city)

    if min_cargo_space is not None:
        conditions.append("cargo_space >= ?")
        params.append(min_cargo_space)

    if has_sunroof is not None:
        conditions.append("has_sunroof = ?")
        params.append(1 if has_sunroof else 0)

    if has_leather_seats is not None:
        conditions.append("has_leather_seats = ?")
        params.append(1 if has_leather_seats else 0)

    if has_navigation is not None:
        conditions.append("has_navigation = ?")
        params.append(1 if has_navigation else 0)

    if has_backup_camera is not None:
        conditions.append("has_backup_camera = ?")
        params.append(1 if has_backup_camera else 0)

    if min_safety_rating is not None:
        conditions.append("safety_rating >= ?")
        params.append(min_safety_rating)

    where_clause = " AND ".join(conditions) if conditions else "1=1"
    query = f"SELECT COUNT(*) as count FROM cars WHERE {where_clause}"

    try:
        with sqlite3.connect(DB_PATH) as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            count = cursor.fetchone()[0]

            return {
                "count": count,
                "status": "too_few" if count < MIN_RESULTS else "too_many" if count > MAX_RESULTS else "good"
            }

    except sqlite3.Error as e:
        return {
            "count": 0,
            "status": "error",
            "error": str(e)
        }


# ============================================================================
# LANGGRAPH STATE DEFINITION
# ============================================================================

class ConversationState(TypedDict):
    """State for the conversation graph"""
    messages: Annotated[list, add]  # Conversation history
    search_params: Optional[dict]  # Current search parameters
    search_results: Optional[dict]  # Results from last search
    iteration_count: int  # Number of search iterations
    user_satisfied: bool  # Whether user is satisfied
    requires_search: bool  # Whether we need to search


# ============================================================================
# LANGGRAPH NODES
# ============================================================================

def gather_requirements(state: ConversationState) -> ConversationState:
    """
    Node: Gather requirements from user and determine search parameters.
    This node uses LLM to extract search criteria from conversation.
    """
    messages = state["messages"]

    system_prompt = f"""You are a friendly car shopping advisor. Analyze the conversation and extract search parameters.

{SCHEMA_DESCRIPTION}

Based on the user's requirements, determine:
1. What search parameters should be used (price range, fuel type, SUV/sedan, features, etc.)
2. Whether we have enough information to search (set requires_search=true)
3. Whether the user is satisfied with current results (set user_satisfied=true)

If the user just greeted you or asked a general question, be friendly and ask what they're looking for.
If previous search returned too few/many results, suggest adjustments.

You have access to these tools:
- search_cars: Search for cars with specific parameters
- count_cars_only: Check count before full search

IMPORTANT: Only use tools when you have concrete search criteria.
For greetings or clarifying questions, just respond conversationally without calling tools."""

    # Get LLM response with tools
    llm_with_tools = llm.bind_tools([search_cars, count_cars_only])
    response = llm_with_tools.invoke([SystemMessage(content=system_prompt)] + messages)

    # Update state with response
    new_state = {
        "messages": [response],
        "requires_search": bool(response.tool_calls),
        "iteration_count": state.get("iteration_count", 0)
    }

    # Check if user expressed satisfaction
    if "satisfied" in response.content.lower() or "perfect" in response.content.lower():
        new_state["user_satisfied"] = True

    return new_state


def execute_search(state: ConversationState) -> ConversationState:
    """
    Node: Execute the search using tool calls from the LLM.
    """
    messages = state["messages"]
    last_message = messages[-1]

    # Execute tool calls
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        tool_node = ToolNode([search_cars, count_cars_only])
        result = tool_node.invoke(state)

        # Extract search results
        tool_messages = result["messages"]
        if tool_messages:
            # Parse the tool response
            tool_response = tool_messages[-1]
            if hasattr(tool_response, 'content'):
                import json
                try:
                    search_results = json.loads(tool_response.content)
                except:
                    search_results = {"error": "Failed to parse tool response"}
            else:
                search_results = {}
        else:
            search_results = {}

        return {
            "messages": tool_messages,
            "search_results": search_results,
            "iteration_count": state.get("iteration_count", 0) + 1
        }

    return {"iteration_count": state.get("iteration_count", 0)}


def present_results(state: ConversationState) -> ConversationState:
    """
    Node: Present search results to user and provide guidance.
    """
    search_results = state.get("search_results", {})
    messages = state["messages"]

    system_prompt = """You are presenting car search results to the user.

Based on the search results:
- If status is 'good' (1-20 cars): Present the results enthusiastically and ask if they want details
- If status is 'too_few' (<1 cars): Suggest broadening criteria (increase price range, consider more fuel types, etc.)
- If status is 'too_many' (>20 cars): Suggest narrowing criteria (set budget, choose specific features, etc.)
- If status is 'error': Apologize and ask them to rephrase

Be conversational, helpful, and guide the user toward finding their perfect car."""

    # Create context message with results
    context = f"\nSearch Results: {search_results}"

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        *messages,
        HumanMessage(content=context)
    ])

    return {"messages": [response]}


def should_continue(state: ConversationState) -> Literal["execute_search", "present_results", "end"]:
    """
    Conditional edge: Determine next step in the graph.
    """
    # Check if user is satisfied
    if state.get("user_satisfied", False):
        return "end"

    # Check if we need to execute search
    if state.get("requires_search", False) and not state.get("search_results"):
        return "execute_search"

    # If we have search results, present them
    if state.get("search_results"):
        return "present_results"

    # Otherwise, end (probably a greeting or farewell)
    return "end"


# ============================================================================
# BUILD LANGGRAPH
# ============================================================================

def build_graph() -> StateGraph:
    """Build the LangGraph workflow"""

    workflow = StateGraph(ConversationState)

    # Add nodes
    workflow.add_node("gather_requirements", gather_requirements)
    workflow.add_node("execute_search", execute_search)
    workflow.add_node("present_results", present_results)

    # Set entry point
    workflow.set_entry_point("gather_requirements")

    # Add conditional edges
    workflow.add_conditional_edges(
        "gather_requirements",
        should_continue,
        {
            "execute_search": "execute_search",
            "present_results": "present_results",
            "end": END
        }
    )

    # After executing search, present results
    workflow.add_edge("execute_search", "present_results")

    # After presenting results, go back to gather requirements for next turn
    workflow.add_edge("present_results", END)

    return workflow.compile()


# ============================================================================
# HELPER FUNCTIONS
# ============================================================================

def format_car_display(car: dict) -> str:
    """Format a single car for display"""
    return f"""
{car['brand']} {car['model_name']} ({car['year']})
  Price: ${car['price']:,}
  Type: {'SUV' if car['is_suv'] else 'Sedan/Coupe'}
  Fuel: {car['fuel_type']}
  Seats: {car['seating_capacity']} | Cargo: {car['cargo_space']} cu ft
  Drivetrain: {car['drivetrain']} | Transmission: {car['transmission']}
  Features: {'Sunroof, ' if car['has_sunroof'] else ''}{'Leather, ' if car['has_leather_seats'] else ''}{'Navigation, ' if car['has_navigation'] else ''}{'Backup Camera' if car['has_backup_camera'] else ''}
"""


# ============================================================================
# MAIN CHATBOT LOOP
# ============================================================================

def main():
    """Main chatbot loop using LangGraph"""
    print("=" * 60)
    print("Welcome to Car Finder Chatbot (LangGraph Edition)")
    print("=" * 60)
    print("Tell me what kind of car you're looking for!")
    print("Type 'quit' to exit.\n")

    # Build the graph
    app = build_graph()

    # Initialize conversation state
    conversation_state = {
        "messages": [],
        "search_params": None,
        "search_results": None,
        "iteration_count": 0,
        "user_satisfied": False,
        "requires_search": False
    }

    while True:
        # Get user input
        user_input = input("You: ").strip()

        if user_input.lower() in ['quit', 'exit', 'bye']:
            print("\nThanks for using Car Finder! Goodbye! 👋")
            break

        if not user_input:
            continue

        # Add user message to state
        conversation_state["messages"].append(HumanMessage(content=user_input))

        # Run the graph
        try:
            result = app.invoke(conversation_state)

            # Update conversation state
            conversation_state = result

            # Display assistant's response
            if result["messages"]:
                last_message = result["messages"][-1]
                if hasattr(last_message, 'content'):
                    print(f"\nAssistant: {last_message.content}\n")

                    # If we have good search results, display them
                    if result.get("search_results", {}).get("status") == "good":
                        cars = result["search_results"].get("cars", [])
                        if cars:
                            print("=" * 60)
                            print("✅ MATCHING CARS:")
                            print("=" * 60)
                            for car in cars:
                                print(format_car_display(car))
                            print("=" * 60)
                            print()

            # Check if conversation should end
            if result.get("user_satisfied", False):
                print("\nThank you for using Car Finder! Have a great day! 👋")
                break

        except Exception as e:
            print(f"\n❌ Error: {str(e)}")
            print("Let's try again. Please rephrase your request.\n")
            # Remove the failed message
            conversation_state["messages"] = conversation_state["messages"][:-1]


if __name__ == "__main__":
    main()

	"""
	LangGraph-based Car Finder Chatbot

	This implementation uses:
	- Template-based SQL queries (no SQL generation)
	- LangGraph for agent orchestration
	- Tool-based architecture for security
	- State management for conversation flow
	"""

	from typing import TypedDict, Annotated, Optional, Literal
	from operator import add
	import sqlite3
	import os
	from dotenv import load_dotenv

	from langchain_openai import ChatOpenAI
	from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
	from langchain_core.tools import tool
	from langgraph.graph import StateGraph, END
	from langgraph.prebuilt import ToolNode
	from pydantic import BaseModel, Field

	# Load environment variables
	load_dotenv()

	# Validate API key
	api_key = os.environ.get("OPENAI_API_KEY")
	if not api_key:
	raise ValueError("OPENAI_API_KEY environment variable is not set")

	# Initialize LLM
	llm = ChatOpenAI(model="gpt-4o", temperature=0.7, api_key=api_key)

	# Load database schema with error handling
	try:
	with open('database_schema.txt', 'r') as f:
	SCHEMA_DESCRIPTION = f.read()
	except FileNotFoundError:
	raise FileNotFoundError("database_schema.txt not found. Please ensure it exists in the current directory.")

	# Constants
	MIN_RESULTS = 1
	MAX_RESULTS = 20
	DB_PATH = 'cars.db'


	# ============================================================================
	# PYDANTIC MODELS FOR STRUCTURED TOOL INPUTS
	# ============================================================================

	class SearchParameters(BaseModel):
	"""Parameters for searching cars using template-based SQL"""
	min_price: Optional[int] = Field(None, description="Minimum price in USD", ge=0, le=100000)
	max_price: Optional[int] = Field(None, description="Maximum price in USD", ge=0, le=100000)
	fuel_type: Optional[Literal["Gasoline", "Diesel", "Electric", "Hybrid", "Plug-in Hybrid"]] = Field(None, description="Type of fuel")
	is_suv: Optional[bool] = Field(None, description="True for SUVs, False for sedans/coupes")
	min_seating: Optional[int] = Field(None, description="Minimum seating capacity", ge=4, le=8)
	max_seating: Optional[int] = Field(None, description="Maximum seating capacity", ge=4, le=8)
	drivetrain: Optional[Literal["FWD", "RWD", "AWD", "4WD"]] = Field(None, description="Drive system")
	min_fuel_efficiency_city: Optional[float] = Field(None, description="Minimum city MPG", ge=0)
	min_cargo_space: Optional[int] = Field(None, description="Minimum cargo space in cubic feet", ge=0)
	has_sunroof: Optional[bool] = Field(None, description="Must have sunroof")
	has_leather_seats: Optional[bool] = Field(None, description="Must have leather seats")
	has_navigation: Optional[bool] = Field(None, description="Must have navigation system")
	has_backup_camera: Optional[bool] = Field(None, description="Must have backup camera")
	min_safety_rating: Optional[float] = Field(None, description="Minimum safety rating", ge=0, le=5)


	# ============================================================================
	# SQL TEMPLATE-BASED TOOLS (Secure, No SQL Generation)
	# ============================================================================

	@tool
	def search_cars(
	min_price: Optional[int] = None,
	max_price: Optional[int] = None,
	fuel_type: Optional[str] = None,
	is_suv: Optional[bool] = None,
	min_seating: Optional[int] = None,
	max_seating: Optional[int] = None,
	drivetrain: Optional[str] = None,
	min_fuel_efficiency_city: Optional[float] = None,
	min_cargo_space: Optional[int] = None,
	has_sunroof: Optional[bool] = None,
	has_leather_seats: Optional[bool] = None,
	has_navigation: Optional[bool] = None,
	has_backup_camera: Optional[bool] = None,
	min_safety_rating: Optional[float] = None,
	) -> dict:
	"""
	Search for cars using a secure template-based SQL query.

	Returns a dictionary with:
	- count: number of cars found
	- cars: list of matching cars (limited to first 20)
	- status: 'too_few', 'good', or 'too_many'
	"""
	# Build WHERE clause from parameters using template
	conditions = []
	params = []

	if min_price is not None:
	conditions.append("price >= ?")
	params.append(min_price)

	if max_price is not None:
	conditions.append("price <= ?")
	params.append(max_price)

	if fuel_type is not None:
	conditions.append("fuel_type = ?")
	params.append(fuel_type)

	if is_suv is not None:
	conditions.append("is_suv = ?")
	params.append(1 if is_suv else 0)

	if min_seating is not None:
	conditions.append("seating_capacity >= ?")
	params.append(min_seating)

	if max_seating is not None:
	conditions.append("seating_capacity <= ?")
	params.append(max_seating)

	if drivetrain is not None:
	conditions.append("drivetrain = ?")
	params.append(drivetrain)

	if min_fuel_efficiency_city is not None:
	conditions.append("fuel_efficiency_city >= ?")
	params.append(min_fuel_efficiency_city)

	if min_cargo_space is not None:
	conditions.append("cargo_space >= ?")
	params.append(min_cargo_space)

	if has_sunroof is not None:
	conditions.append("has_sunroof = ?")
	params.append(1 if has_sunroof else 0)

	if has_leather_seats is not None:
	conditions.append("has_leather_seats = ?")
	params.append(1 if has_leather_seats else 0)

	if has_navigation is not None:
	conditions.append("has_navigation = ?")
	params.append(1 if has_navigation else 0)

	if has_backup_camera is not None:
	conditions.append("has_backup_camera = ?")
	params.append(1 if has_backup_camera else 0)

	if min_safety_rating is not None:
	conditions.append("safety_rating >= ?")
	params.append(min_safety_rating)

	# Build SQL query using template
	where_clause = " AND ".join(conditions) if conditions else "1=1"
	query = f"SELECT * FROM cars WHERE {where_clause} ORDER BY price LIMIT 21" # Get 21 to check if > 20

	try:
	# Use context manager to ensure connection is closed
	with sqlite3.connect(DB_PATH) as conn:
	conn.row_factory = sqlite3.Row
	cursor = conn.cursor()
	cursor.execute(query, params)
	results = cursor.fetchall()

	# Convert to list of dicts
	cars = [dict(row) for row in results]
	count = len(cars)

	# Determine status
	if count < MIN_RESULTS:
	status = "too_few"
	elif count > MAX_RESULTS:
	status = "too_many"
	cars = cars[:MAX_RESULTS] # Limit results
	else:
	status = "good"

	return {
	"count": count,
	"cars": cars,
	"status": status,
	"params_used": {k: v for k, v in [
	("min_price", min_price),
	("max_price", max_price),
	("fuel_type", fuel_type),
	("is_suv", is_suv),
	("min_seating", min_seating),
	("max_seating", max_seating),
	("drivetrain", drivetrain),
	("min_fuel_efficiency_city", min_fuel_efficiency_city),
	("min_cargo_space", min_cargo_space),
	("has_sunroof", has_sunroof),
	("has_leather_seats", has_leather_seats),
	("has_navigation", has_navigation),
	("has_backup_camera", has_backup_camera),
	("min_safety_rating", min_safety_rating),
	] if v is not None}
	}

	except sqlite3.Error as e:
	return {
	"count": 0,
	"cars": [],
	"status": "error",
	"error": str(e)
	}


	@tool
	def count_cars_only(
	min_price: Optional[int] = None,
	max_price: Optional[int] = None,
	fuel_type: Optional[str] = None,
	is_suv: Optional[bool] = None,
	min_seating: Optional[int] = None,
	max_seating: Optional[int] = None,
	drivetrain: Optional[str] = None,
	min_fuel_efficiency_city: Optional[float] = None,
	min_cargo_space: Optional[int] = None,
	has_sunroof: Optional[bool] = None,
	has_leather_seats: Optional[bool] = None,
	has_navigation: Optional[bool] = None,
	has_backup_camera: Optional[bool] = None,
	min_safety_rating: Optional[float] = None,
	) -> dict:
	"""
	Count how many cars match the criteria without returning full results.
	Useful for checking if we need to refine search parameters.
	"""
	# Build WHERE clause from parameters
	conditions = []
	params = []

	if min_price is not None:
	conditions.append("price >= ?")
	params.append(min_price)

	if max_price is not None:
	conditions.append("price <= ?")
	params.append(max_price)

	if fuel_type is not None:
	conditions.append("fuel_type = ?")
	params.append(fuel_type)

	if is_suv is not None:
	conditions.append("is_suv = ?")
	params.append(1 if is_suv else 0)

	if min_seating is not None:
	conditions.append("seating_capacity >= ?")
	params.append(min_seating)

	if max_seating is not None:
	conditions.append("seating_capacity <= ?")
	params.append(max_seating)

	if drivetrain is not None:
	conditions.append("drivetrain = ?")
	params.append(drivetrain)

	if min_fuel_efficiency_city is not None:
	conditions.append("fuel_efficiency_city >= ?")
	params.append(min_fuel_efficiency_city)

	if min_cargo_space is not None:
	conditions.append("cargo_space >= ?")
	params.append(min_cargo_space)

	if has_sunroof is not None:
	conditions.append("has_sunroof = ?")
	params.append(1 if has_sunroof else 0)

	if has_leather_seats is not None:
	conditions.append("has_leather_seats = ?")
	params.append(1 if has_leather_seats else 0)

	if has_navigation is not None:
	conditions.append("has_navigation = ?")
	params.append(1 if has_navigation else 0)

	if has_backup_camera is not None:
	conditions.append("has_backup_camera = ?")
	params.append(1 if has_backup_camera else 0)

	if min_safety_rating is not None:
	conditions.append("safety_rating >= ?")
	params.append(min_safety_rating)

	where_clause = " AND ".join(conditions) if conditions else "1=1"
	query = f"SELECT COUNT(*) as count FROM cars WHERE {where_clause}"

	try:
	with sqlite3.connect(DB_PATH) as conn:
	cursor = conn.cursor()
	cursor.execute(query, params)
	count = cursor.fetchone()[0]

	return {
	"count": count,
	"status": "too_few" if count < MIN_RESULTS else "too_many" if count > MAX_RESULTS else "good"
	}

	except sqlite3.Error as e:
	return {
	"count": 0,
	"status": "error",
	"error": str(e)
	}


	# ============================================================================
	# LANGGRAPH STATE DEFINITION
	# ============================================================================

	class ConversationState(TypedDict):
	"""State for the conversation graph"""
	messages: Annotated[list, add] # Conversation history
	search_params: Optional[dict] # Current search parameters
	search_results: Optional[dict] # Results from last search
	iteration_count: int # Number of search iterations
	user_satisfied: bool # Whether user is satisfied
	requires_search: bool # Whether we need to search


	# ============================================================================
	# LANGGRAPH NODES
	# ============================================================================

	def gather_requirements(state: ConversationState) -> ConversationState:
	"""
	Node: Gather requirements from user and determine search parameters.
	This node uses LLM to extract search criteria from conversation.
	"""
	messages = state["messages"]

	system_prompt = f"""You are a friendly car shopping advisor. Analyze the conversation and extract search parameters.

	{SCHEMA_DESCRIPTION}

	Based on the user's requirements, determine:
	1. What search parameters should be used (price range, fuel type, SUV/sedan, features, etc.)
	2. Whether we have enough information to search (set requires_search=true)
	3. Whether the user is satisfied with current results (set user_satisfied=true)

	If the user just greeted you or asked a general question, be friendly and ask what they're looking for.
	If previous search returned too few/many results, suggest adjustments.

	You have access to these tools:
	- search_cars: Search for cars with specific parameters
	- count_cars_only: Check count before full search

	IMPORTANT: Only use tools when you have concrete search criteria.
	For greetings or clarifying questions, just respond conversationally without calling tools."""

	# Get LLM response with tools
	llm_with_tools = llm.bind_tools([search_cars, count_cars_only])
	response = llm_with_tools.invoke([SystemMessage(content=system_prompt)] + messages)

	# Update state with response
	new_state = {
	"messages": [response],
	"requires_search": bool(response.tool_calls),
	"iteration_count": state.get("iteration_count", 0)
	}

	# Check if user expressed satisfaction
	if "satisfied" in response.content.lower() or "perfect" in response.content.lower():
	new_state["user_satisfied"] = True

	return new_state


	def execute_search(state: ConversationState) -> ConversationState:
	"""
	Node: Execute the search using tool calls from the LLM.
	"""
	messages = state["messages"]
	last_message = messages[-1]

	# Execute tool calls
	if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
	tool_node = ToolNode([search_cars, count_cars_only])
	result = tool_node.invoke(state)

	# Extract search results
	tool_messages = result["messages"]
	if tool_messages:
	# Parse the tool response
	tool_response = tool_messages[-1]
	if hasattr(tool_response, 'content'):
	import json
	try:
	search_results = json.loads(tool_response.content)
	except:
	search_results = {"error": "Failed to parse tool response"}
	else:
	search_results = {}
	else:
	search_results = {}

	return {
	"messages": tool_messages,
	"search_results": search_results,
	"iteration_count": state.get("iteration_count", 0) + 1
	}

	return {"iteration_count": state.get("iteration_count", 0)}


	def present_results(state: ConversationState) -> ConversationState:
	"""
	Node: Present search results to user and provide guidance.
	"""
	search_results = state.get("search_results", {})
	messages = state["messages"]

	system_prompt = """You are presenting car search results to the user.

	Based on the search results:
	- If status is 'good' (1-20 cars): Present the results enthusiastically and ask if they want details
	- If status is 'too_few' (<1 cars): Suggest broadening criteria (increase price range, consider more fuel types, etc.)
	- If status is 'too_many' (>20 cars): Suggest narrowing criteria (set budget, choose specific features, etc.)
	- If status is 'error': Apologize and ask them to rephrase

	Be conversational, helpful, and guide the user toward finding their perfect car."""

	# Create context message with results
	context = f"\nSearch Results: {search_results}"

	response = llm.invoke([
	SystemMessage(content=system_prompt),
	*messages,
	HumanMessage(content=context)
	])

	return {"messages": [response]}


	def should_continue(state: ConversationState) -> Literal["execute_search", "present_results", "end"]:
	"""
	Conditional edge: Determine next step in the graph.
	"""
	# Check if user is satisfied
	if state.get("user_satisfied", False):
	return "end"

	# Check if we need to execute search
	if state.get("requires_search", False) and not state.get("search_results"):
	return "execute_search"

	# If we have search results, present them
	if state.get("search_results"):
	return "present_results"

	# Otherwise, end (probably a greeting or farewell)
	return "end"


	# ============================================================================
	# BUILD LANGGRAPH
	# ============================================================================

	def build_graph() -> StateGraph:
	"""Build the LangGraph workflow"""

	workflow = StateGraph(ConversationState)

	# Add nodes
	workflow.add_node("gather_requirements", gather_requirements)
	workflow.add_node("execute_search", execute_search)
	workflow.add_node("present_results", present_results)

	# Set entry point
	workflow.set_entry_point("gather_requirements")

	# Add conditional edges
	workflow.add_conditional_edges(
	"gather_requirements",
	should_continue,
	{
	"execute_search": "execute_search",
	"present_results": "present_results",
	"end": END
	}
	)

	# After executing search, present results
	workflow.add_edge("execute_search", "present_results")

	# After presenting results, go back to gather requirements for next turn
	workflow.add_edge("present_results", END)

	return workflow.compile()


	# ============================================================================
	# HELPER FUNCTIONS
	# ============================================================================

	def format_car_display(car: dict) -> str:
	"""Format a single car for display"""
	return f"""
	{car['brand']} {car['model_name']} ({car['year']})
	Price: ${car['price']:,}
	Type: {'SUV' if car['is_suv'] else 'Sedan/Coupe'}
	Fuel: {car['fuel_type']}
	Seats: {car['seating_capacity']} \| Cargo: {car['cargo_space']} cu ft
	Drivetrain: {car['drivetrain']} \| Transmission: {car['transmission']}
	Features: {'Sunroof, ' if car['has_sunroof'] else ''}{'Leather, ' if car['has_leather_seats'] else ''}{'Navigation, ' if car['has_navigation'] else ''}{'Backup Camera' if car['has_backup_camera'] else ''}
	"""


	# ============================================================================
	# MAIN CHATBOT LOOP
	# ============================================================================

	def main():
	"""Main chatbot loop using LangGraph"""
	print("=" * 60)
	print("Welcome to Car Finder Chatbot (LangGraph Edition)")
	print("=" * 60)
	print("Tell me what kind of car you're looking for!")
	print("Type 'quit' to exit.\n")

	# Build the graph
	app = build_graph()

	# Initialize conversation state
	conversation_state = {
	"messages": [],
	"search_params": None,
	"search_results": None,
	"iteration_count": 0,
	"user_satisfied": False,
	"requires_search": False
	}

	while True:
	# Get user input
	user_input = input("You: ").strip()

	if user_input.lower() in ['quit', 'exit', 'bye']:
	print("\nThanks for using Car Finder! Goodbye! 👋")
	break

	if not user_input:
	continue

	# Add user message to state
	conversation_state["messages"].append(HumanMessage(content=user_input))

	# Run the graph
	try:
	result = app.invoke(conversation_state)

	# Update conversation state
	conversation_state = result

	# Display assistant's response
	if result["messages"]:
	last_message = result["messages"][-1]
	if hasattr(last_message, 'content'):
	print(f"\nAssistant: {last_message.content}\n")

	# If we have good search results, display them
	if result.get("search_results", {}).get("status") == "good":
	cars = result["search_results"].get("cars", [])
	if cars:
	print("=" * 60)
	print("✅ MATCHING CARS:")
	print("=" * 60)
	for car in cars:
	print(format_car_display(car))
	print("=" * 60)
	print()

	# Check if conversation should end
	if result.get("user_satisfied", False):
	print("\nThank you for using Car Finder! Have a great day! 👋")
	break

	except Exception as e:
	print(f"\n❌ Error: {str(e)}")
	print("Let's try again. Please rephrase your request.\n")
	# Remove the failed message
	conversation_state["messages"] = conversation_state["messages"][:-1]


	if __name__ == "__main__":
	main()