Create app.py

app.py

@@ -0,0 +1,256 @@
+import json
+import smolagents
+import llama_cpp
+import pandas as pd
+import numpy as np
+import gradio as gr
+
+# 1. LLM Model Definition
+MODEL_REPO = "bartowski/Qwen_Qwen3-4B-Instruct-2507-GGUF"
+MODEL_FILE = "Qwen_Qwen3-4B-Instruct-2507-Q4_K_M.gguf"
+
+llm = llama_cpp.Llama.from_pretrained(
+    repo_id=MODEL_REPO,
+    filename=MODEL_FILE,
+    n_ctx=4096*4,
+    n_threads=8,
+    n_layers=-1,
+    verbose=False
+)
+
+# 2. Data Loading and Processing
+SHEET_ID = '1h6gl0reY5iT2Q3_hb8pemP5Hi4OMDYcE'
+BASE_URL = f'https://docs.google.com/spreadsheets/d/{SHEET_ID}/gviz/tq?tqx=out:csv&gid='
+
+COURSES_GID = 1645090689 # GID for the 'Courses' tab
+TOOLS_GID = 2123942435 # GID for the 'Tools' tab
+
+courses_url = f'{BASE_URL}{COURSES_GID}'
+tools_url = f'{BASE_URL}{TOOLS_GID}'
+
+courses_df = pd.read_csv(courses_url)
+tools_df = pd.read_csv(tools_url)
+
+course_info = {
+    str(row["Code"]): {
+        "name": row["Name"],
+        "description": row["Description"]
+    }
+    for _, row in courses_df.iterrows()
+}
+
+machine_to_course = {}
+for _, row in tools_df.iterrows():
+    machine_name = row["Name"]
+    required_course = row["Required Course"]
+
+    if pd.isna(required_course):
+        machine_to_course[machine_name] = "No_Training_Required"
+    else:
+        machine_to_course[machine_name] = str(int(required_course))
+
+# 3. LlamaCppModel and MachineTrainingTool Class Definitions
+class LlamaCppModel(smolagents.Model):
+    """
+    Thin wrapper for a llama.cpp OpenAI-compatible client.
+    Pass in an object exposing `create_chat_completion(...)` (e.g., from llama_cpp).
+    """
+
+    def __init__(self, llm, model_id: str = "llama", **gen_defaults):
+        super().__init__()
+        self.llm = llm
+        self.model_id = model_id
+        self.gen_defaults = {"max_tokens": 4096, "temperature": 0.2}
+        self.gen_defaults.update(gen_defaults)
+
+    @staticmethod
+    def _content_to_str(content) -> str:
+        if content is None:
+            return ""
+        if isinstance(content, str):
+            return content
+
+        if isinstance(content, list):
+            parts = []
+            for p in content:
+                if isinstance(p, str):
+                    parts.append(p)
+                elif isinstance(p, dict):
+                    if p.get("type") == "text" and "text" in p:
+                        parts.append(str(p["text"]))
+                    elif "text" in p:
+                        parts.append(str(p["text"]))
+                    else:
+                        parts.append(json.dumps(p, ensure_ascii=False))
+                else:
+                    parts.append(str(p))
+            return "\n".join([s for s in parts if s])
+
+        if isinstance(content, dict):
+            if content.get("type") == "text" and "text" in content:
+                return str(content["text"])
+            if "content" in content and isinstance(content["content"], str):
+                return content["content"]
+            return json.dumps(content, ensure_ascii=False)
+
+        return str(content)
+
+    @staticmethod
+    def _safe_get(obj, *keys, default=None):
+        if isinstance(obj, dict):
+            for k in keys:
+                if k in obj:
+                    return obj[k]
+            return default
+        for k in keys:
+            if hasattr(obj, k):
+                return getattr(obj, k)
+        return default
+
+    def _to_openai_messages(self, messages: list[smolagents.ChatMessage]) -> list[dict]:
+        oa = []
+        for m in messages:
+            role = getattr(m, "role", None) or (m.get("role") if isinstance(m, dict) else None) or "user"
+            content = getattr(m, "content", None) or (m.get("content") if isinstance(m, dict) else None)
+            text = self._content_to_str(content)
+
+            images = getattr(m, "images", None) or (m.get("images") if isinstance(m, dict) else None)
+            if images:
+                text = (text + f"\n[Note: {len(images)} image(s) omitted]").strip()
+
+            oa.append({"role": role, "content": text})
+        return oa
+
+    def _from_openai_message(self, msg) -> smolagents.ChatMessage:
+        role = self._safe_get(msg, "role", default="assistant")
+        content = self._safe_get(msg, "content", default="")
+        return smolagents.ChatMessage(role=role, content=content)
+
+    def generate(
+        self,
+        messages: list[smolagents.ChatMessage],
+        stop_sequences: list[str] | None = None,
+        response_format: dict[str, str] | None = None,
+        tools_to_call_from: list[smolagents.Tool] | None = None,
+        **kwargs,
+    ) -> smolagents.ChatMessage:
+        oa_msgs = self._to_openai_messages(messages)
+
+        params = dict(self.gen_defaults)
+        params.update(kwargs)
+        if stop_sequences:
+            params["stop"] = stop_sequences
+        if response_format:
+            params["response_format"] = response_format
+
+        resp = self.llm.create_chat_completion(
+            model=self.model_id,
+            messages=oa_msgs,
+            **params,
+        )
+
+        choices = self._safe_get(resp, "choices", default=[])
+        if not choices:
+            text = self._safe_get(resp, "content", default=str(resp))
+            return smolagents.ChatMessage(role="assistant", content=text)
+
+        first = choices[0]
+        message = self._safe_get(first, "message", default={})
+        return self._from_openai_message(message)
+
+class MachineTrainingTool(smolagents.tools.Tool):
+    name = "get_machine_training_info"
+    description = (
+        "Retrieves training information for a specific machine. The `machine_name` argument should exactly match the machine's name as listed in the system (e.g., 'Laser Cutters', '3D Printers')."
+    )
+    inputs = {
+        "machine_name": {"type": "string", "description": "Name of the machine for which to retrieve training information"},
+    }
+    output_type = "string"
+
+    def forward(self, machine_name: str) -> str:
+      if machine_name in machine_to_course:
+          course_code = machine_to_course[machine_name]
+          if course_code in course_info:
+              course_name = course_info[course_code]['name']
+              return f"For {machine_name}, the required training is: '{course_name}' (Course Code: {course_code})."
+          else:
+              return f"No detailed course information found for course code '{course_code}' associated with {machine_name}."
+      else:
+          return f"No specific training information available for machine: {machine_name}."
+
+# 4. Agent Instantiation
+machine_training_tool_instance = MachineTrainingTool()
+
+llama_cpp_model_instance = LlamaCppModel(llm)
+llama_cpp_model_instance.gen_defaults['response_format'] = {'type': 'json_object'}
+
+from smolagents.agents import ToolCallingAgent
+
+machine_agent = ToolCallingAgent(
+    model=llama_cpp_model_instance,
+    name="MachineAgent",
+    instructions=(
+        "Your ONLY purpose is to call the 'get_machine_training_info' tool ONCE per user query. "
+        "You MUST then output the EXACT string provided by the tool's observation. "
+        "You MUST NOT add any text before or after the tool's observation. "
+        "You MUST NOT summarize, paraphrase, or change the tool's output in any way. "
+        "You MUST NOT call any other tool, especially 'final_answer'. "
+        "The 'machine_name' argument MUST be an exact match from the available machine list."
+    ),
+    tools=[machine_training_tool_instance],
+    verbosity_level=2
+)
+
+# 5. Gradio Interface Code
+with gr.Blocks(title="Techspark Machine Training Agent") as demo:
+    gr.Markdown("## Techspark Machine Training Agent — Custom Tool Selection (smolagents + llama.cpp)")
+    chat = gr.Chatbot(height=420)
+    inp = gr.Textbox(
+        placeholder="Ask about machine training (e.g., 'What training do I need for the Laser Cutters?').",
+        label="Your question"
+    )
+
+    def respond(message, history):
+        try:
+            result_object = machine_agent.run(message, return_full_result=True)
+            tool_observation = None
+
+            if isinstance(result_object, smolagents.RunResult):
+                for step_dict in result_object.steps:
+                    if isinstance(step_dict, dict):
+                        if 'tool_calls' in step_dict and step_dict['tool_calls']:
+                            for tool_call_info in step_dict['tool_calls']:
+                                if tool_call_info.get('function') and tool_call_info['function'].get('name') == 'get_machine_training_info':
+                                    if 'observations' in step_dict:
+                                        tool_observation = step_dict['observations']
+                                        break
+                            if tool_observation:
+                                break
+
+            if tool_observation:
+                out = tool_observation
+            else:
+                out = result_object.text if hasattr(result_object, 'text') else str(result_object)
+
+        except Exception as e:
+            out = f"[Error] {e}"
+
+        history = (history or []) + [(message, out)]
+        return "", history
+
+    gr.Examples(
+        fn=respond,
+        examples=[
+            "What training do I need for the Laser Cutters?",
+            "What are the training requirements for the 3D Printer?",
+            "Can you tell me about training for Metal CNC?"
+        ],
+        inputs=[inp],
+        outputs=[chat]
+    )
+
+    inp.submit(respond, [inp, chat], [inp, chat])
+
+# 6. Modified demo.launch() for Hugging Face Spaces
+demo.launch(server_name="0.0.0.0", server_port=7860, debug=False)