Create function_orchestrator.py

function_orchestrator.py

@@ -0,0 +1,107 @@
+# 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()