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()
