Spaces:
Sleeping
Sleeping
File size: 7,339 Bytes
15bcec8 48e7fd1 15bcec8 3775a55 15bcec8 3775a55 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 48e7fd1 15bcec8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 |
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()
|