Update app.py

app.py

@@ -1,75 +1,87 @@
 import gradio as gr
 from huggingface_hub import InferenceClient
+from array import array
 import os
 
+# Securely retrieve the token from Space secrets
 HF_TOKEN = os.getenv("HF_TOKEN")
 
-# Utilizing the conversational task through chat_completion
+# Initialize the inference client
 client = InferenceClient("Qwen/Qwen2.5-7B-Instruct", token=HF_TOKEN)
 
 class StateController:
+    __slots__ = ("_state", "_metric", "_batch", "_reg", "_rendered")
+
     def __init__(self):
-        self.state_array = [0] * 121
-        self.base_metric = 60
-        self.batch_unit = 10
-        self.memory_register = {}
+        self._state      = array("B", [0]) * 121   # unsigned 8-bit array
+        self._metric     = 60
+        self._batch      = 10
+        self._reg        = {}
+        self._rendered   = self._build_render()    # cached string representation
+
+    def _build_render(self) -> str:
+        """Construct the fixed 121-point visualization once during initialization."""
+        return "".join(
+            " [NODE_120] " if i == 120 else ("<" if i % 10 == 0 else ".")
+            for i in range(121)
+        )
 
-    def initialize_grid(self):
+    def diagnostic(self) -> str:
+        """Execute diagnostic sequence and return the formatted status report."""
+        # Update state array for diagnostic verification
         for i in range(51):
-            self.state_array[i] = {"Blocks": i // self.batch_unit, "Units": i % self.batch_unit}
-        return "Grid initialized: 5 active blocks."
+            self._state[i] = i % self._batch
+        
+        self._reg.clear()
+        self._reg["STATUS"] = "RESOLVED"
 
-    def render_grid(self):
-        grid_output = ""
-        for i in range(121):
-            if i == 120:
-                grid_output += " [NODE_120] "
-            elif i % 10 == 0:
-                grid_output += "<"
-            else:
-                grid_output += "."
-        return grid_output
+        return (
+            "Diagnostic sequence initiated.\n\n"
+            "Grid initialized: 5 active blocks.\n\n"
+            "Rendering 121-point array:\n"
+            f"{self._rendered}\n\n"
+            "Executing state resolution:\n"
+            "System resolved. State array reset to zero."
+        )
+
+# Global singleton instance for resource reuse
+controller = StateController()
 
-    def resolve_grid(self):
-        self.memory_register["STATUS"] = "RESOLVED"
-        self.state_array = [0] * 121
-        return "System resolved. State array reset to zero."
+SYSTEM_MSG = [
+    {"role": "system",
+     "content": ("You are a logic-focused inference engine. "
+                 "Utilize strict state-hold memory and parallel integer blocks. "
+                 "Provide direct, technical, and accurate responses.")}
+]
 
-def generate_response(message, history):
+def generate_response(message: str, history: list):
     # Hardware diagnostic override
-    if "run grid diagnostic" in message.lower():
-        controller = StateController()
-        output = "Diagnostic sequence initiated.\n\n"
-        output += f"{controller.initialize_grid()}\n\n"
-        output += "Rendering 121-point array:\n"
-        output += f"{controller.render_grid()}\n\n"
-        output += "Executing state resolution:\n"
-        output += f"{controller.resolve_grid()}"
-        return output
+    if message.lower().strip() == "run grid diagnostic":
+        yield controller.diagnostic()
+        return
 
-    system_instruction = (
-        "You are a logic-focused inference engine. "
-        "You utilize strict state-hold memory and parallel integer blocks. "
-        "Provide direct, technical, and accurate responses."
-    )
-    
-    # Correct format for conversational task
-    messages = [{"role": "system", "content": system_instruction}]
-    for user_msg, assistant_msg in history:
-        messages.append({"role": "user", "content": user_msg})
-        messages.append({"role": "assistant", "content": assistant_msg})
-    messages.append({"role": "user", "content": message})
+    # Utilize Gradio history directly to avoid redundant list processing
+    messages = SYSTEM_MSG + [
+        {"role": "user",      "content": turn[0]},
+        {"role": "assistant", "content": turn[1]}
+        for turn in history
+    ] + [{"role": "user", "content": message}]
 
     try:
-        # Switching to chat_completion for model compatibility
-        response = client.chat_completion(
+        # Enable streaming for reduced time-to-first-token
+        stream = client.chat_completion(
             messages,
             max_tokens=1024,
-            stream=False
+            stream=True,
+            temperature=0.15
         )
-        return response.choices[0].message.content
-    except Exception as error:
-        return f"System Error: {str(error)}. Verify your token permissions."
+        partial_response = ""
+        for chunk in stream:
+            token = chunk.choices[0].delta.content or ""
+            partial_response += token
+            yield partial_response
+    except Exception as exc:
+        yield f"System Error: {str(exc)}. Verify API token and permissions."
 
 custom_css = """
 body, .gradio-container { background-color: #0b0f19 !important; }
@@ -87,8 +99,10 @@ with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue"), css=custom_css) as demo
             "Run grid diagnostic",
             "Calculate the integer distribution for 120 units across 3 nodes.",
             "Explain network latency using technical terminology."
-        ]
+        ],
+        cache_examples=False
     )
 
 if __name__ == "__main__":
-    demo.launch()
+    # Launch with queue enabled for streaming support
+    demo.queue().launch(show_api=False)