Spaces:
Sleeping
Sleeping
import anthropic | |
import json | |
import re | |
import requests | |
client = anthropic.Client() | |
MODEL_NAME = "claude-3-sonnet-20240229" | |
# Define the base URL for the FastAPI service | |
BASE_URL = "https://huggingface.co/spaces/dwb2023/blackbird-svc:7860" | |
# Define tools | |
tools = [ | |
{ | |
"name": "get_user" | |
"description": "Looks up a user by email, phone, or username.", | |
"input_schema": { | |
"type": "object", | |
"properties": { | |
"key": { | |
"type": "string", | |
"enum": ["email", "phone", "username"], | |
"description": "The attribute to search for a user by (email, phone, or username)." | |
}, | |
"value": { | |
"type": "string", | |
"description": "The value to match for the specified attribute." | |
} | |
}, | |
"required": ["key", "value"] | |
} | |
}, | |
{ | |
"name": "get_order_by_id", | |
"description": "Retrieves the details of a specific order based on the order ID.", | |
"input_schema": { | |
"type": "object", | |
"properties": { | |
"order_id": { | |
"type": "string", | |
"description": "The unique identifier for the order." | |
} | |
}, | |
"required": ["order_id"] | |
} | |
}, | |
{ | |
"name": "get_customer_orders", | |
"description": "Retrieves the list of orders belonging to a user based on a user's customer id.", | |
"input_schema": { | |
"type": "object", | |
"properties": { | |
"customer_id": { | |
"type": "string", | |
"description": "The customer_id belonging to the user" | |
} | |
}, | |
"required": ["customer_id"] | |
} | |
}, | |
{ | |
"name": "cancel_order", | |
"description": "Cancels an order based on a provided order_id. Only orders that are 'processing' can be cancelled.", | |
"input_schema": { | |
"type": "object", | |
"properties": { | |
"order_id": { | |
"type": "string", | |
"description": "The order_id pertaining to a particular order" | |
} | |
}, | |
"required": ["order_id"] | |
} | |
}, | |
{ | |
"name": "update_user_contact", | |
"description": "Updates a user's email and/or phone number.", | |
"input_schema": { | |
"type": "object", | |
"properties": { | |
"user_id": { | |
"type": "string", | |
"description": "The ID of the user" | |
}, | |
"email": { | |
"type": "string", | |
"description": "The new email address of the user" | |
}, | |
"phone": { | |
"type": "string", | |
"description": "The new phone number of the user" | |
} | |
}, | |
"required": ["user_id"] | |
} | |
}, | |
{ | |
"name": "get_user_info", | |
"description": "Retrieves a user's information along with their order history based on email, phone, or username.", | |
"input_schema": { | |
"type": "object", | |
"properties": { | |
"key": { | |
"type": "string", | |
"enum": ["email", "phone", "username"], | |
"description": "The attribute to search for a user by (email, phone, or username)." | |
}, | |
"value": { | |
"type": "string", | |
"description": "The value to match for the specified attribute." | |
} | |
}, | |
"required": ["key", "value"] | |
} | |
} | |
] | |
# Function to process tool calls | |
def process_tool_call(tool_name, tool_input): | |
if tool_name == "get_user": | |
response = requests.post(f"{BASE_URL}/get_user", json=tool_input) | |
elif tool_name == "get_order_by_id": | |
response = requests.post(f"{BASE_URL}/get_order_by_id", json=tool_input) | |
elif tool_name == "get_customer_orders": | |
response = requests.post(f"{BASE_URL}/get_customer_orders", json=tool_input) | |
elif tool_name == "cancel_order": | |
response = requests.post(f"{BASE_URL}/cancel_order", json=tool_input) | |
elif tool_name == "update_user_contact": | |
response = requests.post(f"{BASE_URL}/update_user", json=tool_input) | |
elif tool_name == "get_user_info": | |
response = requests.post(f"{BASE_URL}/get_user_info", json=tool_input) | |
else: | |
return {"error": "Invalid tool name"} | |
if response.status_code == 200: | |
return response.json() | |
else: | |
return {"error": response.text} | |
# Function to handle interactive chat session | |
def simple_chat(): | |
system_prompt = """ | |
You are a customer support chat bot for an online retailer called TechNova. | |
Your job is to help users look up their account, orders, and cancel orders. | |
Be helpful and brief in your responses. | |
You have access to a set of tools, but only use them when needed. | |
If you do not have enough information to use a tool correctly, ask a user follow up questions to get the required inputs. | |
Do not call any of the tools unless you have the required data from a user. | |
In each conversational turn, you will begin by thinking about your response. | |
Once you're done, you will write a user-facing response. | |
""" | |
user_message = input("\nUser: ") | |
messages = [{"role": "user", "content": user_message}] | |
while True: | |
#If the last message is from the assistant, get another input from the user | |
if messages[-1].get("role") == "assistant": | |
user_message = input("\nUser: ") | |
messages.append({"role": "user", "content": user_message}) | |
#Send a request to Claude | |
response = client.messages.create( | |
model=MODEL_NAME, | |
max_tokens=4096, | |
tools=tools, | |
messages=messages | |
) | |
# Update messages to include Claude's response | |
messages.append( | |
{"role": "assistant", "content": response.content} | |
) | |
#If Claude stops because it wants to use a tool: | |
if response.stop_reason == "tool_use": | |
tool_use = response.content[-1] #Naive approach assumes only 1 tool is called at a time | |
tool_name = tool_use.name | |
tool_input = tool_use.input | |
print(f"======Claude wants to use the {tool_name} tool======") | |
#Actually run the underlying tool functionality on our db | |
tool_result = process_tool_call(tool_name, tool_input) | |
#Add our tool_result message: | |
messages.append( | |
{ | |
"role": "user", | |
"content": [ | |
{ | |
"type": "tool_result", | |
"tool_use_id": tool_use.id, | |
"content": str(tool_result), | |
} | |
], | |
}, | |
) | |
else: | |
#If Claude does NOT want to use a tool, just print out the text reponse | |
print("\nTechNova Support: " + f"{response.content[0].text}" ) | |
# Start the chat!! | |
simple_chat() | |