Customer Service bot with tools

# Example conversation demonstrating order status inquiry:

User: Hello
Assistant: Hi there! How can I assist you today?

User: I placed an order the other day and I wonder when to expect it
Assistant: Could you please provide me with your order ID and the email address you used to place the order? This will help me check the status for you.

User: my order id is 123123
Assistant: Thank you for providing your order ID. Could you also share the email address you used to place the order so I can look up the details for you?

User: santaclause@clau.se
Assistant: I'm sorry, but I couldn't find an order with the ID 123123. Could you double-check the order ID and make sure it's correct? If you need further assistance, please let me know!

User: oh actually my email is santa@clau.se
Assistant: Thank you for the updated email address. I have found your order with ID 123123, and it is currently out for delivery to Snowy Mountain 123, 421 23, Greenland. If you need further assistance, please let me know!

User: What did I order? I forgot
Assistant: Currently, I don't have details on the contents of your order. For more information, please check your email or the confirmation message you received after placing the order. If you need further assistance, feel free to ask!

User: ok thanks! 
Assistant: You're welcome! If you have any more questions or need further assistance in the future, feel free to reach out. Have a great day!

from opperai import Opper
from pydantic import BaseModel, Field
from typing import Literal
import os

# Initialize Opper client
opper = Opper(http_bearer=os.getenv("OPPER_API_KEY"))

# A test "database"
orders = {
    123123: {
        "email": "santa@clau.se", 
        "status": "delivering",
        "created_date": "2024-11-03", 
        "updated_date": "2024-11-06",
        "adress": "Snowy Mountain 123, 421 23, Greenland",
        "purchase": "Large sled 1999 SEK"
        }
    }

# Input schema for intent classification
class ConversationMessages(BaseModel):
    messages: list[dict] = Field(description="List of conversation messages with role and content")

# Output schema for intent classification
class IntentClassification(BaseModel):
    thoughts: str = Field(description="The thoughts of the model while analyzing the intent")
    intent: Literal["get_order_status", "query_products", "unsupported"] = Field(description="The classified intent of the conversation")

# Function to determine intent of conversation
def determine_intent(messages, span=None):
    result = opper.call(
        name="determine_intent",
        instructions="Analyze the user message and determine their intent. Supported intents are get_order_status and query_products.",
        input_schema=ConversationMessages,
        output_schema=IntentClassification,
        input=ConversationMessages(messages=messages),
        parent_span_id=span.id if span else None
    )
    return result.json_payload
    
# Output schema for order extraction
class ParsedOrder(BaseModel):
    thoughts: str = Field(description="The thoughts of the model while extracting order information")
    order_id: int | None = Field(default=None, description="The order ID if found in the conversation")
    email: str | None = Field(default=None, description="The email address if found in the conversation")

# Function to extract order data from conversation
def extract_order_from_messages(messages, span=None):
    result = opper.call(
        name="extract_order_info",
        instructions="Extract order ID and email from the conversation if present",
        input_schema=ConversationMessages,
        output_schema=ParsedOrder,
        input=ConversationMessages(messages=messages),
        parent_span_id=span.id if span else None
    )
    return result.json_payload

# Function to get the requested order
def get_order(id, email):
    if id in orders and orders[id]["email"] == email:
        return orders[id]
    else:
        return None

# Function to process messages
def process_message(messages, span=None):

    # We first determine intent with the conversation
    intent = determine_intent(messages, span)

    # Process based on intent
    if intent["intent"] == "get_order_status":

        # Extract requested order information
        order_request = extract_order_from_messages(messages, span)

        # Verify we have all needed order info
        if not order_request["order_id"] or not order_request["email"]:
            return  f"Need {'order ID and email' if not order_request['order_id'] and not order_request['email'] else 'order ID' if not order_request['order_id'] else 'email'}"

        order = get_order(id=order_request["order_id"], email=order_request["email"])
        if order:
            return {
                    "order_id": order_request["order_id"],
                    "status": orders[order_request["order_id"]]["status"],
                    "email": orders[order_request["order_id"]]["email"],
                    "address": orders[order_request["order_id"]]["adress"]
                }
        else:
            return f"Could not find an order with id {order_request['order_id']}"

    # Here we could have different tools      
    #elif intent["intent"] == "query_products": 
    #    return None
    
    elif intent["intent"] == "unsupported": 
        return f"Request is currently not supported: {messages[-1]}"
    
    else:
        return None

# Input schema for response generation
class ResponseGenerationInput(BaseModel):
    messages: list[dict] = Field(description="List of conversation messages with role and content")

# Function to build a friendly AI response
def bake_response(messages, span=None):
    result = opper.call(
        name="generate_response",
        instructions="Generate a helpful, friendly but brief response to the user's message in the conversation.",
        input_schema=ResponseGenerationInput,
        input=ResponseGenerationInput(messages=messages),
        parent_span_id=span.id if span else None
    )
    return result.message

def run():
    # Here we have a conversatonal loop of user, function and assistant messages
    messages = []

    # Create a session span to track the entire conversation
    session_span = opper.spans.create(
        name="conversation_session"
    )

    while True:
        
        # Start a span for this message turn (child of session span)
        message_span = opper.spans.create(
            name="on_message",
            parent_id=session_span.id
        )

        # Get user input
        user_input = input("User: ")
        if user_input.lower() == "quit":
            break

        messages.append({
            "role": "user",
            "content": user_input
        })

        # Analyse the conversation and return an analysis
        analysis = process_message(messages, message_span)

        messages.append({
            "role": "function",
            "content": analysis
        })

        # Bake response to the user 
        response = bake_response(messages, message_span)

        print(f"Assistant: {response}")

        messages.append({
            "role": "assistant",
            "content": response
        })
        
        # Update the message span with input and output values
        opper.spans.update(
            span_id=message_span.id,
            input=user_input,
            output=response
        )
        
        # Here we could add thumbs up on this response 
        # opper.span_metrics.create_metric(
        #     span_id=message_span.id,
        #     dimension="thumbs_up", 
        #     value=1, 
        #     comment="User pressed thumbs up button"
        # )

    # Update the session span with final conversation summary
    opper.spans.update(
        span_id=session_span.id,
        input=f"Conversation started with {len(messages)} total messages",
        output=f"Conversation completed with {len(messages)} total messages",
        meta={"total_messages": len(messages)}
    )

if __name__ == "__main__":
    run()