core-crew/ecom.py

from crewai import Process, Agent, Task, Crew, tasks
from crewai_tools import BaseTool
import sqlite3
import os
import langchain_core

# Set environment variables for OpenAI
os.environ["OPENAI_API_KEY"] = "sk-kkk"
os.environ["OPENAI_MODEL_NAME"] = "anthropic.claude-3-sonnet-20240229-v1:0"
os.environ["OPENAI_API_BASE"] = "http://chat.the.mk:1337"

# Initialize LLM
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
    model="anthropic.claude-3-sonnet-20240229-v1:0",
    base_url="http://chat.the.mk:1337",
    temperature=0.1
)

# Create a new SQLite database file (if it doesn't exist)
conn = sqlite3.connect('your_database.db')
c = conn.cursor()

# Create tables for customers, orders, payments, inventory, and shipping
c.execute('''CREATE TABLE IF NOT EXISTS customers
             (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)''')
c.execute('''CREATE TABLE IF NOT EXISTS orders
             (id INTEGER PRIMARY KEY AUTOINCREMENT, customer_id INTEGER, items TEXT, total REAL)''')
c.execute('''CREATE TABLE IF NOT EXISTS payments
             (id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER, amount REAL, status TEXT)''')
c.execute('''CREATE TABLE IF NOT EXISTS inventory
             (id INTEGER PRIMARY KEY AUTOINCREMENT, product_id TEXT, quantity INTEGER)''')
c.execute('''CREATE TABLE IF NOT EXISTS shipping
             (id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER, shipping_address TEXT)''')

# Example usage
order_details = {
    'customer_id': 1234,
    'items': [
        {'product_id': 'SKU010', 'quantity': 2},
        {'product_id': 'SKU006', 'quantity': 1}
    ]
}

customer_inquiry = {
    'customer_id': 5678,
    'issue': 'Order status inquiry'
}

shipping_details = {
    'order_id': 9876,
    'shipping_address': '123 Main St, Anytown USA'
}

order = {
    'order_details': order_details,
    'customer_inquiry': customer_inquiry,
    'shipping_details': shipping_details
}

def create_customer(customer_data):
    name = customer_data.get("name")
    email = customer_data.get("email")
    c.execute("INSERT INTO customers (name, email) VALUES (?, ?)", (name, email))
    conn.commit()
    return c.lastrowid

def get_customer(customer_id):
    c.execute("SELECT * FROM customers WHERE id = ?", (customer_id,))
    customer = c.fetchone()
    if customer:
        return {"id": customer[0], "name": customer[1], "email": customer[2]}
    else:
        return None

def create_order(order_data):
    customer_id = order_data.get("customer_id")
    items = str(order_data.get("items"))
    total = order_data.get("total")
    c.execute("INSERT INTO orders (customer_id, items, total) VALUES (?, ?, ?)", (customer_id, items, total))
    conn.commit()
    return c.lastrowid

def get_order(order_id):
    c.execute("SELECT * FROM orders WHERE id = ?", (order_id,))
    order = c.fetchone()
    if order:
        return {"id": order[0], "customer_id": order[1], "items": eval(order[2]), "total": order[3]}
    else:
        return None

def process_payment(payment_data):
    order_id = payment_data.get("order_id")
    amount = payment_data.get("amount")
    # Implement payment processing logic using a payment gateway
    status = "paid"  # Replace with actual payment status
    c.execute("INSERT INTO payments (order_id, amount, status) VALUES (?, ?, ?)", (order_id, amount, status))
    conn.commit()
    return c.lastrowid

def get_payment(order_id):
    c.execute("SELECT * FROM payments WHERE order_id = ?", (order_id,))
    payment = c.fetchone()
    if payment:
        return {"id": payment[0], "order_id": payment[1], "amount": payment[2], "status": payment[3]}
    else:
        return None

def update_inventory(order_data):
    for item in order_data.get("items", []):
        product_id = item["product_id"]
        quantity = item["quantity"]
        c.execute("SELECT quantity FROM inventory WHERE product_id = ?", (product_id,))
        result = c.fetchone()
        if result:
            current_quantity = result[0]
            new_quantity = current_quantity - quantity
            c.execute("UPDATE inventory SET quantity = ? WHERE product_id = ?", (new_quantity, product_id))
        else:
            c.execute("INSERT INTO inventory (product_id, quantity) VALUES (?, ?)", (product_id, -quantity))
    conn.commit()

class CustomerManagementTool(BaseTool):
    name: str | None = "Customer Management"
    description: str | None = "A tool to manage customers."

    def _run(self, **kwargs):
        if "customer_data" in kwargs:
            customer_data = kwargs["customer_data"]
            customer_id = create_customer(customer_data)
            return f"Customer created successfully with ID {customer_id}."
        elif "customer_id" in kwargs:
            customer_id = kwargs["customer_id"]
            customer = get_customer(customer_id)
            if customer:
                return f"Customer details: {customer}"
            else:
                return f"Customer with ID {customer_id} not found."
        else:
            return "Invalid input. Please provide either 'customer_data' or 'customer_id'."

class OrderManagementTool(BaseTool):
    name: str | None = "Order Management"
    description: str | None = "A tool to manage orders."

    def _run(self, **kwargs):
        if "order_data" in kwargs:
            order_data = kwargs["order_data"]
            order_id = create_order(order_data)
            update_inventory(order_data)
            return f"Order created successfully with ID {order_id}."
        elif "order_id" in kwargs:
            order_id = kwargs["order_id"]
            order = get_order(order_id)
            if order:
                return f"Order details: {order}"
            else:
                return f"Order with ID {order_id} not found."
        else:
            return "Invalid input. Please provide either 'order_data' or 'order_id'."

class PaymentProcessingTool(BaseTool):
    name: str | None = "Payment Processing"
    description: str | None = "A tool to process payments."

    def _run(self, **kwargs):
        if "payment_data" in kwargs:
            payment_data = kwargs["payment_data"]
            payment_id = process_payment(payment_data)
            return f"Payment processed successfully with ID {payment_id}."
        elif "order_id" in kwargs:
            order_id = kwargs["order_id"]
            payment = get_payment(order_id)
            if payment:
                return f"Payment details: {payment}"
            else:
                return f"Payment not found for order with ID {order_id}."
        else:
            return "Invalid input. Please provide either 'payment_data' or 'order_id'."

class InventoryManagementTool(BaseTool):
    name: str | None = "Inventory Management"
    description: str | None = "A tool to manage inventory."

    def _run(self, **kwargs):
        c.execute("SELECT * FROM inventory")
        inventory_data = c.fetchall()
        inventory_report = "Current inventory:\n"
        for item in inventory_data:
            product_id, quantity = item
            inventory_report += f"{product_id}: {quantity}\n"
        return inventory_report

class ShippingTool(BaseTool):
    name: str | None = "Shipping"
    description: str | None = "A tool to handle shipping and delivery."

    def _run(self, **kwargs):
        if "order_id" in kwargs:
            order_id = kwargs["order_id"]
            shipping = get_shipping(order_id)
            if shipping:
                return f"Shipping details: {shipping}"
            else:
                return f"Shipping details not found for order with ID {order_id}."
        elif "shipping_data" in kwargs:
            shipping_data = kwargs["shipping_data"]
            order_id = shipping_data.get("order_id")
            recipient_address = shipping_data.get("recipient_address")
            package_details = shipping_data.get("package_details")

            if not order_id or not recipient_address or not package_details:
                return "Invalid shipping data. Please provide 'order_id', 'recipient_address', and 'package_details'."

            shipping_id = create_shipping(order_id, recipient_address, package_details)
            return f"Shipping details created with ID {shipping_id}."
        else:
            return "Invalid input. Please provide either 'order_id' or 'shipping_data'."

def create_shipping(order_id, recipient_address, package_details):
    name = recipient_address.get("name")
    street = recipient_address.get("street")
    city = recipient_address.get("city")
    state = recipient_address.get("state")
    zip_code = recipient_address.get("zip")
    weight = package_details.get("weight")
    length = package_details.get("dimensions", {}).get("length")
    width = package_details.get("dimensions", {}).get("width")
    height = package_details.get("dimensions", {}).get("height")

    shipping_address = f"{name}, {street}, {city}, {state} {zip_code}"

    c.execute("INSERT INTO shipping (order_id, shipping_address, weight, length, width, height) VALUES (?, ?, ?, ?, ?, ?)",
              (order_id, shipping_address, weight, length, width, height))
    conn.commit()
    return c.lastrowid

def get_shipping(order_id):
    c.execute("SELECT * FROM shipping WHERE order_id = ?", (order_id,))
    shipping = c.fetchone()
    if shipping:
        return {
            "id": shipping[0],
            "order_id": shipping[1],
            "shipping_address": shipping[2],
            "weight": shipping[3],
            "length": shipping[4],
            "width": shipping[5],
            "height": shipping[6]
        }
    else:
        return None
    
# Create instances of the custom tools
customer_management_tool = CustomerManagementTool()
order_management_tool = OrderManagementTool()
payment_processing_tool = PaymentProcessingTool()
inventory_management_tool = InventoryManagementTool()
shipping_tool = ShippingTool()

# ... (rest of the code remains the same)

# Callback functions
def order_callback(output):
    if isinstance(output, langchain_core.agents.AgentFinish):
        print("Order Agent finished")
    elif output and output[0]:
        print(output[0])
        if output[0][0]:
            print(output[0][0].log)

def customer_callback(output):
    if isinstance(output, langchain_core.agents.AgentFinish):
        print("Customer Agent finished")
    elif output and output[0]:
        print(output[0])
        if output[0][0]:
            print(output[0][0].log)

def inventory_callback(output):
    if isinstance(output, langchain_core.agents.AgentFinish):
        print("Inventory Agent finished")
    elif output and output[0]:
        print(output[0])
        if output[0][0]:
            print(output[0][0].log)

def logistics_callback(output):
    if isinstance(output, langchain_core.agents.AgentFinish):
        print("Logistics Agent finished")
    elif output and output[0]:
        print(output[0])
        if output[0][0]:
            print(output[0][0].log)

def payment_callback(output):
    if isinstance(output, langchain_core.agents.AgentFinish):
        print("Logistics Agent finished")
    elif output and output[0]:
        print(output[0])
        if output[0][0]:
            print(output[0][0].log)

# Define the Order Agent
order_agent = Agent(
    role='Order Manager',
    goal='Manage and process customer orders efficiently',
    backstory=(
        "You are the Order Manager at an e-commerce company."
        "Your responsibilities include receiving and processing customer orders,"
        "ensuring that orders are accurate and complete, and coordinating with other teams to fulfill orders."
        "You have access to the company's order management system."
    ),
    tools=[order_management_tool],
    max_rpm=100,
    step_callback=order_callback
)

order_task = Task(
    description=(
        "Process a new customer order."
        "Ensure the order details are accurate and complete."
        "Coordinate with other teams to fulfill the order."
    ),
    expected_output="Customer order processed and coordinated for fulfillment.",
    agent=order_agent
)

# Define the Customer Agent
customer_agent = Agent(
    role='Customer Support',
    goal='Provide excellent customer service and resolve inquiries quickly',
    backstory=(
        "You are a Customer Support Agent at an e-commerce company."
        "Your role is to assist customers with their inquiries, provide accurate information,"
        "and resolve any issues they may have."
        "You have access to the company's customer support system, knowledge base, order information, and issue tracking systems."
    ),
    tools=[customer_management_tool],
    max_rpm=100,
    step_callback=customer_callback
)

customer_task = Task(
    description=(
        "Handle the incoming customer inquiry."
        "Provide accurate information and resolve the issue."
        "Document the interaction in the customer support system."
    ),
    expected_output="Customer inquiry resolved, information provided, and interaction documented.",
    agent=customer_agent
)

# Define the Payment Agent
payment_agent = Agent(
    role='Payment Processor',
    goal='Process customer payments securely',
    backstory=(
        "You are the Payment Processor at an e-commerce company."
        "Your role is to process customer payments using a secure payment gateway."
        "You have access to the payment processing system."
    ),
    tools=[payment_processing_tool],
    max_rpm=100,
    step_callback=payment_callback
)

payment_task = Task(
    description=(
        "Process the payment for a customer order."
        "Ensure the payment is processed securely and accurately."
        "Update the order status and notify the customer."
    ),
    expected_output="Payment processed successfully, order status updated, and customer notified.",
    agent=payment_agent
)

# Define the Inventory Agent
inventory_agent = Agent(
    role='Inventory Manager',
    goal='Maintain accurate inventory levels and ensure product availability',
    backstory=(
        "You are the Inventory Manager at an e-commerce company."
        "Your role is to monitor and manage inventory levels, reorder products when needed,"
        "and ensure that the warehouse is well-stocked."
        "You have access to the company's inventory management system."
    ),
    tools=[inventory_management_tool],
    max_rpm=100,
    step_callback=inventory_callback
)

inventory_task = Task(
    description=(
        "Monitor inventory levels and reorder products as needed."
        "Coordinate with suppliers for timely restocking."
        "Optimize stock levels based on demand and sales data."
    ),
    expected_output="Inventory levels optimized, and restocking coordinated with suppliers.",
    agent=inventory_agent
)

# Define the Logistics Agent
logistics_agent = Agent(
    role='Logistics Coordinator',
    goal='Ensure timely and accurate shipping of customer orders',
    backstory=(
        "You are the Logistics Coordinator at an e-commerce company."
        "Your role is to manage the shipping process, coordinate with carriers,"
        "and track shipments to ensure timely delivery."
        "You have access to the company's order management system and carrier integrations."
    ),
    tools=[shipping_tool],
    max_rpm=100,
    step_callback=logistics_callback
)

logistics_task = Task(
    description=(
        "Coordinate the shipping of the processed order."
        "Select the appropriate carrier and shipping method."
        "Track the shipment and provide updates to the customer."
    ),
    expected_output="Order shipped successfully, and tracking information provided to the customer.",
    agent=logistics_agent
)

class ManagerTool(BaseTool):
    name: str | None = "Manager Tool"
    description: str | None = "A tool for the Manager Agent to coordinate other agents and handle business processes."

    # name = "Manager Tool"
    # description = "A tool for the Manager Agent to coordinate other agents and handle business processes."

    def _run(self, **kwargs) -> str:
        return f"I understand. I have done what is necessary."


manager_agent = Agent(
    role="Business Operations Manager",
    goal="Coordinate and manage business processes efficiently",
    backstory=(
        "You are the Business Operations Manager at an e-commerce company."
        "Your role is to oversee and coordinate various business processes,"
        "including order processing, customer support, inventory management, and logistics."
        "You have access to a tool that allows you to delegate tasks to specialized agents."
        "The agents you can delegate to are order_agent, customer_agent, inventory_agent, and logistics_agent"
    ),
    delegate=True,
    tools=[ManagerTool()],
    max_rpm=100,
)

manager_task = Task(
    description="Manage and coordinate the appropriate business processes based on the given input.",
    expected_output="Business processes executed successfully.",
    agent=manager_agent,
)

crew = Crew(
    name="eCommerce Team",
    agents=[
        manager_agent,
        order_agent,
        customer_agent,
        payment_agent,
        inventory_agent,
        logistics_agent
    ],
    tasks=[manager_task, order_task, customer_task, payment_task, inventory_task, logistics_task],
    manager_llm=ChatOpenAI(
        model="amazon-embeddings",
        base_url="http://chat.the.mk:1337",
        temperature=0.5
    ), # Mandatory for hierarchical process
    process=Process.hierarchical, # Specifies the hierarchical management approach
)

crew.kickoff()