Spaces:
Sleeping
Sleeping
| # function_orchestrator.py | |
| import anthropic | |
| import json | |
| import requests | |
| client = anthropic.Client() | |
| MODEL_NAME = "claude-3-sonnet-20240229" | |
| # Define the base URL for the FastAPI service | |
| BASE_URL = "https://dwb2023-blackbird-svc.hf.space" | |
| # 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"] | |
| } | |
| }, | |
| # other tools definitions... | |
| ] | |
| # 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 messages[-1].get("role") == "assistant": | |
| user_message = input("\nUser: ") | |
| messages.append({"role": "user", "content": user_message}) | |
| response = client.messages.create( | |
| model=MODEL_NAME, | |
| max_tokens=4096, | |
| tools=tools, | |
| messages=messages | |
| ) | |
| messages.append({"role": "assistant", "content": response.content}) | |
| if response.stop_reason == "tool_use": | |
| tool_use = response.content[-1] | |
| tool_name = tool_use.name | |
| tool_input = tool_use.input | |
| print(f"======Claude wants to use the {tool_name} tool======") | |
| tool_result = process_tool_call(tool_name, tool_input) | |
| messages.append({ | |
| "role": "user", | |
| "content": [{ | |
| "type": "tool_result", | |
| "tool_use_id": tool_use.id, | |
| "content": str(tool_result), | |
| }], | |
| }) | |
| else: | |
| print("\nTechNova Support: " + f"{response.content[0].text}") | |
| # Start the chat!! | |
| simple_chat() | |