Create app.py

app.py

@@ -0,0 +1,516 @@
+# ==============================================================================
+#  Tool World: Advanced Prototype (Hugging Face Space Version)
+# ==============================================================================
+#
+#  This script has been updated to run as a Hugging Face Space.
+#
+#  Key Upgrades from the original script:
+#  1.  **Hugging Face Model Integration**: Uses the 'google/gemma-3n-E4B' model
+#      from the Hugging Face Hub for argument extraction.
+#  2.  **Environment Variable Management**: Securely accesses the
+#      HUGGING_FACE_HUB_TOKEN using os.environ.get(), which is the standard
+#      for Hugging Face Spaces.
+#  3.  **Standard Dependencies**: All dependencies are managed via a
+#      `requirements.txt` file.
+#
+# ==============================================================================
+
+# ------------------------------
+#  1. INSTALL & IMPORT PACKAGES
+# ------------------------------
+import numpy as np
+import umap
+import gradio as gr
+from sentence_transformers import SentenceTransformer, util
+import matplotlib.pyplot as plt
+import json
+import os
+from datetime import datetime, timedelta
+import torch
+from transformers import AutoTokenizer, AutoModelForCausalLM
+
+# ------------------------------
+#  2. CONFIGURE & LOAD MODELS
+# ------------------------------
+
+print("⚙️ Loading embedding model...")
+# Using a powerful model for better semantic understanding
+embedder = SentenceTransformer('all-mpnet-base-v2')
+print("✅ Embedding model loaded.")
+
+# --- Configuration for Hugging Face Model-based Argument Extraction ---
+try:
+    HF_TOKEN = os.environ.get('HUGGING_FACE_HUB_TOKEN')
+    if HF_TOKEN is None:
+        raise ValueError("HUGGING_FACE_HUB_TOKEN secret not found.")
+
+    print("⚙️ Loading Hugging Face model for argument extraction...")
+    # Using the user-specified Gemma model
+    model_id = "google/gemma-3n-E4B"
+
+    hf_tokenizer = AutoTokenizer.from_pretrained(model_id, token=HF_TOKEN)
+
+    # --------------------------------------------------------------------------
+    #  ✅ FIX: Manually set the chat template for the Gemma model.
+    #  This is required because the specified model does not have a default
+    #  template set in its tokenizer config on the Hugging Face Hub.
+    # --------------------------------------------------------------------------
+    gemma_template = (
+        "{% for message in messages %}"
+        "{{'<start_of_turn>' + message['role'] + '\n' + message['content'] + '<end_of_turn>\n'}}"
+        "{% endfor %}"
+        "{% if add_generation_prompt %}"
+        "{{ '<start_of_turn>model\n' }}"
+        "{% endif %}"
+    )
+    hf_tokenizer.chat_template = gemma_template
+    # --------------------------------------------------------------------------
+
+    hf_model = AutoModelForCausalLM.from_pretrained(
+        model_id,
+        token=HF_TOKEN,
+        torch_dtype=torch.bfloat16, # Use bfloat16 for efficiency
+        device_map="auto" # Automatically use GPU if available
+    )
+    USE_HF_LLM = True
+    print(f"✅ Successfully loaded '{model_id}' model and set chat template.")
+
+except Exception as e:
+    USE_HF_LLM = False
+    print(f"⚠️ WARNING: Could not load the Hugging Face model. Reason: {e}")
+    print("   Argument extraction will be disabled.")
+
+
+# ------------------------------
+#  3. ADVANCED TOOL DEFINITION
+# ------------------------------
+
+class Tool:
+    """
+    Represents a tool with structured arguments and rich descriptive data
+    for high-quality embedding.
+    """
+    def __init__(self, name, description, args_schema, function, examples=None):
+        self.name = name
+        self.description = description
+        self.args_schema = args_schema
+        self.function = function
+        self.examples = examples or []
+        self.embedding = self._create_embedding()
+
+    def _create_embedding(self):
+        """
+        Creates a rich embedding by combining the tool's name, description,
+        argument structure, and examples.
+        """
+        schema_str = json.dumps(self.args_schema, indent=2)
+        examples_str = "\n".join([f" - Example: {ex['prompt']} -> Args: {json.dumps(ex['args'])}" for ex in self.examples])
+
+        embedding_text = (
+            f"Tool Name: {self.name}\n"
+            f"Description: {self.description}\n"
+            f"Argument Schema: {schema_str}\n"
+            f"Usage Examples:\n{examples_str}"
+        )
+        return embedder.encode(embedding_text, convert_to_tensor=True)
+
+    def __repr__(self):
+        return f"<Tool: {self.name}>"
+
+# ------------------------------
+#  4. TOOL IMPLEMENTATIONS
+# ------------------------------
+
+def get_weather_forecast(location: str, days: int = 1):
+    """Simulates fetching a weather forecast."""
+    if not isinstance(location, str) or not isinstance(days, int):
+        return {"error": "Invalid argument types. 'location' must be a string and 'days' an integer."}
+
+    weather_conditions = ["Sunny", "Cloudy", "Rainy", "Windy", "Snowy"]
+    response = {"location": location, "forecast": []}
+
+    for i in range(days):
+        date = (datetime.now() + timedelta(days=i)).strftime('%Y-%m-%d')
+        condition = np.random.choice(weather_conditions)
+        temp = np.random.randint(5, 25)
+        response["forecast"].append({
+            "date": date,
+            "condition": condition,
+            "temperature_celsius": temp
+        })
+    return response
+
+def create_calendar_event(title: str, date: str, duration_minutes: int = 60, participants: list = None):
+    """Simulates creating a calendar event."""
+    try:
+        # Check for relative terms like "tomorrow"
+        if 'tomorrow' in date.lower():
+            event_base_date = datetime.now() + timedelta(days=1)
+            # Try to extract time, default to 9am if not specified
+            try:
+                time_part = datetime.strptime(date, '%I:%M %p').time()
+            except ValueError:
+                try:
+                    time_part = datetime.strptime(date, '%H:%M').time()
+                except ValueError:
+                    time_part = datetime.strptime('09:00', '%H:%M').time()
+            event_time = event_base_date.replace(hour=time_part.hour, minute=time_part.minute, second=0, microsecond=0)
+        else:
+            event_time = datetime.strptime(date, '%Y-%m-%d %H:%M')
+
+        return {
+            "status": "success",
+            "event_created": {
+                "title": title,
+                "start_time": event_time.isoformat(),
+                "end_time": (event_time + timedelta(minutes=duration_minutes)).isoformat(),
+                "participants": participants or ["organizer"]
+            }
+        }
+    except ValueError:
+        return {"error": "Invalid date format. Please use 'YYYY-MM-DD HH:MM' or a relative term like 'tomorrow at 10:00'."}
+
+def summarize_text(text: str, compression_level: str = 'medium'):
+    """Summarizes a given text based on a compression level."""
+    word_count = len(text.split())
+    ratios = {'high': 0.2, 'medium': 0.4, 'low': 0.7}
+    ratio = ratios.get(compression_level, 0.4)
+    summary_length = int(word_count * ratio)
+    summary = " ".join(text.split()[:summary_length])
+    return {"summary": summary + "...", "original_word_count": word_count, "summary_word_count": summary_length}
+
+def search_web(query: str, domain: str = None):
+    """Simulates a web search, with an optional domain filter."""
+    results = [
+        f"Simulated result 1 for '{query}'",
+        f"Simulated result 2 for '{query}'",
+        f"Simulated result 3 for '{query}'"
+    ]
+    if domain:
+        return {"status": f"Searching for '{query}' within '{domain}'...", "results": results}
+    return {"status": f"Searching for '{query}'...", "results": results}
+
+
+# ------------------------------
+#  5. DEFINE THE TOOLSET
+# ------------------------------
+
+tools = [
+    Tool(
+        name="weather_reporter",
+        description="Provides the weather forecast for a specific location for a given number of days.",
+        args_schema={
+            "type": "object",
+            "properties": {
+                "location": {"type": "string", "description": "The city and state, e.g., 'San Francisco, CA'"},
+                "days": {"type": "integer", "description": "The number of days to forecast", "default": 1}
+            },
+            "required": ["location"]
+        },
+        function=get_weather_forecast,
+        examples=[
+            {"prompt": "what's the weather like in London for the next 3 days", "args": {"location": "London", "days": 3}},
+            {"prompt": "forecast for New York tomorrow", "args": {"location": "New York", "days": 1}}
+        ]
+    ),
+    Tool(
+        name="calendar_creator",
+        description="Creates a new event in the user's calendar.",
+        args_schema={
+            "type": "object",
+            "properties": {
+                "title": {"type": "string", "description": "The title of the calendar event"},
+                "date": {"type": "string", "description": "The start date and time in 'YYYY-MM-DD HH:MM' format. Handles relative terms like 'tomorrow at 10:30 am'."},
+                "duration_minutes": {"type": "integer", "description": "The duration of the event in minutes", "default": 60},
+                "participants": {"type": "array", "items": {"type": "string"}, "description": "List of email addresses of participants"}
+            },
+            "required": ["title", "date"]
+        },
+        function=create_calendar_event,
+        examples=[
+            {"prompt": "Schedule a 'Project Sync' for tomorrow at 3pm with [email protected]", "args": {"title": "Project Sync", "date": (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d 15:00'), "participants": ["[email protected]"]}},
+            {"prompt": "new event: Dentist appointment on 2025-12-20 at 10:00 for 45 mins", "args": {"title": "Dentist appointment", "date": "2025-12-20 10:00", "duration_minutes": 45}}
+        ]
+    ),
+    Tool(
+        name="text_summarizer",
+        description="Summarizes a long piece of text. Can be set to high, medium, or low compression.",
+        args_schema={
+            "type": "object",
+            "properties": {
+                "text": {"type": "string", "description": "The text to be summarized."},
+                "compression_level": {"type": "string", "enum": ["high", "medium", "low"], "description": "The level of summarization.", "default": "medium"}
+            },
+            "required": ["text"]
+        },
+        function=summarize_text,
+        examples=[
+            {"prompt": "summarize this article for me, make it very short: [long text...]", "args": {"text": "[long text...]", "compression_level": "high"}}
+        ]
+    ),
+    Tool(
+        name="web_search",
+        description="Performs a web search to find information on a topic.",
+        args_schema={
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "The search query."},
+                "domain": {"type": "string", "description": "Optional: a specific website domain to search within (e.g., 'wikipedia.org')."}
+            },
+            "required": ["query"]
+        },
+        function=search_web,
+        examples=[
+            {"prompt": "who invented the light bulb", "args": {"query": "who invented the light bulb"}},
+            {"prompt": "search for 'transformer models' on arxiv.org", "args": {"query": "transformer models", "domain": "arxiv.org"}}
+        ]
+    )
+]
+
+print(f"✅ {len(tools)} tools defined and embedded.")
+
+# ------------------------------
+#  6. CORE LOGIC: TOOL SELECTION & ARGUMENT EXTRACTION
+# ------------------------------
+
+def find_best_tool(user_intent: str):
+    """Finds the most semantically similar tool for a user's intent."""
+    intent_embedding = embedder.encode(user_intent, convert_to_tensor=True)
+    # Move tool embeddings to the same device as the intent embedding
+    tool_embeddings = [tool.embedding.to(intent_embedding.device) for tool in tools]
+    similarities = [util.pytorch_cos_sim(intent_embedding, tool_emb).item() for tool_emb in tool_embeddings]
+    best_index = int(np.argmax(similarities))
+    best_tool = tools[best_index]
+    best_score = similarities[best_index]
+    return best_tool, best_score, similarities
+
+def extract_arguments_hf(user_prompt: str, tool: Tool):
+    """
+    Uses a local Hugging Face model to extract structured arguments.
+    """
+    system_prompt = f"""
+    You are an expert at extracting structured data from natural language.
+    Your task is to analyze the user's prompt and extract the arguments required to call the tool: '{tool.name}'.
+
+    You must adhere to the following JSON schema for the arguments:
+    {json.dumps(tool.args_schema, indent=2)}
+
+    - If a value is not present in the prompt for a non-required field, omit it from the JSON.
+    - If a required value is missing, return a JSON object with an "error" key explaining what is missing.
+    - Today's date is {datetime.now().strftime('%Y-%m-%d')}. If the user says "tomorrow", use {(datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')}.
+    - Respond ONLY with a valid JSON object. Do not include any other text, explanation, or markdown code blocks.
+    """
+
+    # Gemma instruction-following format
+    chat = [
+        # Gemma does not use a 'system' role. Instructions are part of the first user message.
+        {"role": "user", "content": f"{system_prompt}\n\nUser Prompt: \"{user_prompt}\""},
+    ]
+
+    prompt = hf_tokenizer.apply_chat_template(chat, tokenize=False, add_generation_prompt=True)
+
+    try:
+        inputs = hf_tokenizer.encode(prompt, add_special_tokens=False, return_tensors="pt").to(hf_model.device)
+
+        # Generate with the model
+        outputs = hf_model.generate(input_ids=inputs, max_new_tokens=256, do_sample=False)
+        decoded_output = hf_tokenizer.decode(outputs[0][len(inputs[0]):], skip_special_tokens=True)
+
+        # Clean the response to find the JSON object
+        json_str = decoded_output.strip()
+
+        # Find the first '{' and the last '}' to get the JSON part
+        json_start = json_str.find('{')
+        json_end = json_str.rfind('}')
+
+        if json_start != -1 and json_end != -1:
+            json_str = json_str[json_start : json_end + 1]
+            return json.loads(json_str)
+        else:
+            raise json.JSONDecodeError("No JSON object found in the model output.", json_str, 0)
+
+    except Exception as e:
+        print(f"Error during HF model inference or JSON parsing: {e}")
+        return {"error": f"Failed to extract arguments with the local LLM. Details: {str(e)}"}
+
+def execute_tool(user_prompt: str):
+    """The main pipeline: Find tool, extract args, execute."""
+    selected_tool, score, _ = find_best_tool(user_prompt)
+
+    if USE_HF_LLM:
+        print(f"⚙️ Selected Tool: {selected_tool.name}. Extracting arguments with Gemma...")
+        extracted_args = extract_arguments_hf(user_prompt, selected_tool)
+    else:
+        # Fallback if the model failed to load
+        extracted_args = {"error": "Argument extraction is disabled because the Hugging Face model could not be loaded."}
+
+    if 'error' in extracted_args:
+        print(f"❌ Argument extraction failed: {extracted_args['error']}")
+        # Ensure the final output string is valid JSON
+        final_output_str = json.dumps({
+            "error": "Execution failed during argument extraction.",
+            "details": extracted_args.get('error', 'Unknown extraction error')
+        })
+        return (
+            user_prompt,
+            selected_tool.name,
+            f"{score:.3f}",
+            json.dumps(extracted_args, indent=2),
+            final_output_str
+        )
+
+    print(f"✅ Arguments extracted: {json.dumps(extracted_args, indent=2)}")
+
+    try:
+        print(f"🚀 Executing tool function: {selected_tool.name}...")
+        output = selected_tool.function(**extracted_args)
+        print(f"✅ Execution complete.")
+        output_str = json.dumps(output, indent=2)
+    except Exception as e:
+        print(f"❌ Tool execution failed: {e}")
+        output_str = f'{{"error": "Tool execution failed", "details": "{str(e)}"}}'
+
+    return (
+        user_prompt,
+        selected_tool.name,
+        f"{score:.3f}",
+        json.dumps(extracted_args, indent=2),
+        output_str
+    )
+
+
+# ------------------------------
+#  7. VISUALIZATION
+# ------------------------------
+
+def plot_tool_world(user_intent=None):
+    """Generates a 2D UMAP plot of the tool latent space."""
+    tool_vectors = [tool.embedding.cpu().numpy() for tool in tools]
+    labels = [tool.name for tool in tools]
+    all_vectors = tool_vectors
+
+    if user_intent and user_intent.strip():
+        intent_vector = embedder.encode(user_intent, convert_to_tensor=True).cpu().numpy()
+        all_vectors.append(intent_vector)
+        labels.append("Your Intent")
+
+    # UMAP requires at least 2 neighbors
+    n_neighbors = min(len(all_vectors) - 1, 5)
+    if n_neighbors < 1:
+        n_neighbors = 1
+
+    reducer = umap.UMAP(n_neighbors=n_neighbors, min_dist=0.3, metric='cosine', random_state=42)
+
+    # UMAP fit_transform requires at least 2 samples
+    if len(all_vectors) < 2:
+         # Create a dummy plot if there's not enough data
+        fig, ax = plt.subplots(figsize=(10, 7))
+        ax.text(0.5, 0.5, "Not enough data to create a plot.", ha='center', va='center')
+        return fig
+
+    reduced_vectors = reducer.fit_transform(all_vectors)
+
+    plt.style.use('seaborn-v0_8-whitegrid')
+    fig, ax = plt.subplots(figsize=(10, 7))
+
+    for i, label in enumerate(labels):
+        x, y = reduced_vectors[i]
+        if label == "Your Intent":
+            ax.scatter(x, y, color='red', s=150, zorder=5, label=label, marker='*')
+            ax.text(x, y + 0.05, label, fontsize=12, ha='center', color='red', weight='bold')
+        else:
+            ax.scatter(x, y, s=100, alpha=0.8, label=label)
+            ax.text(x, y + 0.05, label, fontsize=10, ha='center')
+
+    ax.set_title("Tool World: Latent Space Map", fontsize=16)
+    ax.set_xlabel("UMAP Dimension 1", fontsize=12)
+    ax.set_ylabel("UMAP Dimension 2", fontsize=12)
+    ax.grid(True)
+
+    handles, labels_legend = ax.get_legend_handles_labels()
+    by_label = dict(zip(labels_legend, handles))
+    ax.legend(by_label.values(), by_label.keys())
+
+    plt.tight_layout()
+    return fig
+
+
+# ------------------------------
+#  8. GRADIO INTERFACE
+# ------------------------------
+
+print("🚀 Launching Gradio interface...")
+
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🛠️ Tool World: Advanced Prototype (Hugging Face Version)")
+    gr.Markdown(
+        "Enter a natural language command. The system will select the best tool, "
+        "extract structured arguments with **google/gemma-3n-E4B**, and execute it."
+    )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            inp = gr.Textbox(
+                label="Your Intent",
+                placeholder="e.g., What's the weather in Paris for 2 days?",
+                lines=3
+            )
+            run_btn = gr.Button("Invoke Tool", variant="primary")
+
+            gr.Markdown("---")
+            gr.Markdown("### Examples")
+            gr.Examples(
+                examples=[
+                    "Schedule a 'Team Meeting' for tomorrow at 10:30 am",
+                    "What is the weather forecast in Tokyo for the next 5 days?",
+                    "search for the latest news on generative AI on reuters.com",
+                    "Please give me a very short summary of this text: The Industrial Revolution was the transition to new manufacturing processes in Europe and the United States, in the period from about 1760 to sometime between 1820 and 1840."
+                ],
+                inputs=inp
+            )
+
+        with gr.Column(scale=2):
+            gr.Markdown("### Invocation Details")
+            with gr.Row():
+                out_tool = gr.Textbox(label="Selected Tool", interactive=False)
+                out_score = gr.Textbox(label="Similarity Score", interactive=False)
+
+            out_args = gr.JSON(label="Extracted Arguments")
+            out_result = gr.JSON(label="Tool Execution Output")
+
+    with gr.Row():
+        gr.Markdown("---")
+        gr.Markdown("### Latent Space Visualization")
+        plot_output = gr.Plot(label="Tool World Map")
+
+    def process_and_plot(user_prompt):
+        if not user_prompt or not user_prompt.strip():
+            # Return empty state and the default plot
+            return "", "", {}, {}, plot_tool_world()
+
+        prompt, tool_name, score, args_json, result_json = execute_tool(user_prompt)
+        fig = plot_tool_world(user_prompt)
+
+        # Safely load JSON strings into objects for the UI
+        try:
+            args_obj = json.loads(args_json)
+        except (json.JSONDecodeError, TypeError):
+            args_obj = {"error": "Invalid JSON in arguments", "raw": args_json}
+
+        try:
+            result_obj = json.loads(result_json)
+        except (json.JSONDecodeError, TypeError):
+            result_obj = {"error": "Invalid JSON in result", "raw": result_json}
+
+        return tool_name, score, args_obj, result_obj, fig
+
+    run_btn.click(
+        fn=process_and_plot,
+        inputs=inp,
+        outputs=[out_tool, out_score, out_args, out_result, plot_output]
+    )
+
+    # Load the initial plot when the app starts
+    demo.load(fn=lambda: plot_tool_world(None), inputs=None, outputs=plot_output)
+
+demo.launch()