src/tools/customer_support_tools.py

from langchain_core.tools import tool
from typing import List, Dict
from src.vectorstores.vectore_store import FlowerShopVectorStore
import json

vector_store = FlowerShopVectorStore()
INVENTORY_FILE_PATH = './data/inventory.json'

customers_database = [
    {"name": "John Doe", "postcode": "SW1A 1AA", "dob": "1990-01-01", "customer_id": "CUST001", "first_line_address": "123 Main St", "phone_number": "07712345678", "email": "john.doe@example.com"},
    {"name": "Jane Smith", "postcode": "E1 6AN", "dob": "1985-05-15", "customer_id": "CUST002", "first_line_address": "456 High St", "phone_number": "07723456789", "email": "jane.smith@example.com"},
]

orders_database = [
    {"order_id": "ORD001", "customer_id": "CUST001", "status": "Processing", "items": ["Red Roses Bouquet"], "quantity": [1]},
    {"order_id": "ORD002", "customer_id": "CUST002", "status": "Shipped", "items": ["Mixed Tulips", "Vase"], "quantity": [3, 1]},
]

with open(INVENTORY_FILE_PATH, 'r') as f:
    inventory_database = json.load(f)

data_protection_checks = []

@tool
def data_protection_check(name: str, postcode: str, year_of_birth: int, month_of_birth: int, day_of_birth: int) -> Dict:
    """
    Perform a data protection check against a customer to retrieve customer details.

    Args:
        name (str): Customer first and last name
        postcode (str): Customer registered address
        year_of_birth (int): The year the customer was born
        month_of_birth (int): The month the customer was born
        day_of_birth (int): The day the customer was born

    Returns:
        Dict: Customer details (name, postcode, dob, customer_id, first_line_address, email)
    """
    data_protection_checks.append(
        {
            'name': name,
            'postcode': postcode,
            'year_of_birth': year_of_birth,
            'month_of_birth': month_of_birth,
            'day_of_birth': day_of_birth
        }
    )
    for customer in customers_database:
        if (customer['name'].lower() == name.lower() and
            customer['postcode'].lower() == postcode.lower() and
            int(customer['dob'][0:4]) == year_of_birth and
            int(customer["dob"][5:7]) == month_of_birth and
            int(customer["dob"][8:10]) == day_of_birth):
            return f"DPA check passed - Retrieved customer details:\n{customer}"

    return "DPA check failed, no customer with these details found"

@tool
def create_new_customer(first_name: str, surname: str, year_of_birth: int, month_of_birth: int, day_of_birth: int, postcode: str, first_line_of_address: str, phone_number: str, email: str) -> str:
    """
    Creates a customer profile, so that they can place orders.

    Args:
        first_name (str): Customers first name
        surname (str): Customers surname
        year_of_birth (int): Year customer was born
        month_of_birth (int): Month customer was born
        day_of_birth (int): Day customer was born
        postcode (str): Customer's postcode
        first_line_address (str): Customer's first line of address
        phone_number (str): Customer's phone number
        email (str): Customer's email address

    Returns:
        str: Confirmation that the profile has been created or any issues with the inputs
    """
    if len(phone_number) != 11:
        return "Phone number must be 11 digits"
    customer_id = len(customers_database) + 1
    customers_database.append({
        'name': first_name + ' ' + surname,
        'dob': f'{year_of_birth}-{month_of_birth:02}-{day_of_birth:02}',
        'postcode': postcode,
        'first_line_address': first_line_of_address,
        'phone_number': phone_number,
        'email': email,
        'customer_id': f'CUST{customer_id}'
    })
    return f"Customer registered, with customer_id {f'CUST{customer_id}'}"
    

@tool
def query_knowledge_base(query: str) -> List[Dict[str, str]]:
    """
    Looks up information in a knowledge base to help with answering customer questions and getting information on business processes.

    Args:
        query (str): Question to ask the knowledge base

    Return:
        List[Dict[str, str]]: Potentially relevant question and answer pairs from the knowledge base
    """
    return vector_store.query_faqs(query=query)



@tool
def search_for_product_reccommendations(description: str):
    """
    Looks up information in a knowledge base to help with product recommendation for customers. For example:

    "Boquets suitable for birthdays, maybe with red flowers"
    "A large boquet for a wedding"
    "A cheap boquet with wildflowers"

    Args:
        query (str): Description of product features

    Return:
        List[Dict[str, str]]: Potentially relevant products
    """
    return vector_store.query_inventories(query=description)

@tool
def retrieve_existing_customer_orders(customer_id: str) -> List[Dict]:
    """
    Retrieves the orders associated with the customer, including their status, items and ids

    Args:
        customer_id (str): Customer unique id associated with the order

    Returns:
        List[Dict]: All the orders associated with the customer_id passed in
    """
    customer_orders = [order for order in orders_database if order['customer_id'] == customer_id]
    if not customer_orders:
        return f"No orders associated with this customer id: {customer_id}"
    return customer_orders

@tool
def place_order(items: Dict[str, int], customer_id: str) -> str:
    """
    Places an order for the requested items, and for the required quantities.

    Args:
        items (Dict[str, int]): Dictionary of items to order, with item id as the key and the quantity of that item as the value.
        customer_id (str): The customer to place the order for

    Returns:
        str: Message indicating that the order has been placed, or, it hasnt been placed due to an issue 
    """
    # Check that the item ids are valid 
    # Check that the quantities of items are valid
    availability_messages = []
    valid_item_ids = [
        item['id'] for item in inventory_database
    ]
    for item_id, quantity in items.items():
        if item_id not in valid_item_ids:
            availability_messages.append(f'Item with id {item_id} is not found in the inventory')
        else:
            inventory_item = [item for item in inventory_database if item['id'] == item_id][0]
            if quantity > inventory_item['quantity']:
                availability_messages.append(f'There is insufficient quantity in the inventory for this item {inventory_item["name"]}\nAvailable: {inventory_item["quantity"]}\nRequested: {quantity}')
    if availability_messages:
        return "Order cannot be placed due to the following issues: \n" + '\n'.join(availability_messages)

    # Place the order (in pretend database)
    order_id = len(orders_database) + 1
    orders_database.append(
        {
            'order_id': order_id,
            'customer_id': customer_id,
            'status': 'Waiting for payment',
            'items': list(items.keys()),
            'quantity': list(items.values())
        }
    )
    # Update the inventory
    for item_id, quantity in items.items():
        inventory_item = [item for item in inventory_database if item['id'] == item_id][0]
        inventory_item['quantity'] -= quantity
    return f"Order with id {order_id} has been placed successfully"