Update app.py

app.py

@@ -1,143 +1,117 @@
 import concurrent.futures
 import gradio as gr
+from huggingface_hub import InferenceClient
 import json
 import traceback
+import tempfile
+import shutil
 
-class FunctionManager:
+class HuggingChatExecutor:
     def __init__(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
         self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
-        self.functions = {}
+        self.client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
 
-    def validate_function_json(self, json_str):
-        """Validate function JSON format."""
-        try:
-            function_data = json.loads(json_str)
-            required_fields = ['name', 'description', 'parameters']
-            if not all(field in function_data for field in required_fields):
-                return False, "Missing required fields: name, description, or parameters"
-            
-            if not isinstance(function_data['parameters'], dict):
-                return False, "Parameters must be a JSON object"
-                
-            return True, function_data
-        except json.JSONDecodeError as e:
-            return False, f"Invalid JSON format: {str(e)}"
-        except Exception as e:
-            return False, f"Validation error: {str(e)}"
+    def _execute_chat(self, message, system_prompt, functions=None):
+        messages = [{"role": "system", "content": system_prompt}]
+        messages.append({"role": "user", "content": message})
 
-    def add_function(self, function_json):
-        """Add a new function definition."""
         try:
-            is_valid, result = self.validate_function_json(function_json)
-            if not is_valid:
-                return f"Error: {result}"
-            
-            function_name = result['name']
-            self.functions[function_name] = result
-            return f"Successfully added function: {function_name}"
-        except Exception as e:
-            return f"Error adding function: {str(e)}\n{traceback.format_exc()}"
+            if functions:
+                # Parse functions string as JSON array
+                functions_list = json.loads(functions)
+                response = self.client.chat_completion(
+                    messages,
+                    functions=functions_list,
+                    stream=False
+                )
+            else:
+                response = self.client.chat_completion(
+                    messages,
+                    stream=False
+                )
 
-    def list_functions(self):
-        """List all registered functions."""
-        try:
-            if not self.functions:
-                return "No functions registered."
-            
-            output = ["# Registered Functions"]
-            for name, func in self.functions.items():
-                output.append(f"\n## {name}")
-                output.append(f"Description: {func['description']}")
-                output.append("\nParameters:")
-                params = func['parameters']
-                if 'properties' in params:
-                    for param_name, param_info in params['properties'].items():
-                        output.append(f"- {param_name}: {param_info.get('description', 'No description')}")
-                        if param_info.get('type'):
-                            output.append(f"  Type: {param_info['type']}")
-                        if param_info.get('required', False):
-                            output.append("  Required: Yes")
+            # Format the output
+            output = {
+                "message": message,
+                "system_prompt": system_prompt,
+                "functions": functions if functions else "No functions provided",
+                "response": response.choices[0].message
+            }
             
-            return "\n".join(output)
+            return json.dumps(output, indent=2)
         except Exception as e:
-            return f"Error listing functions: {str(e)}\n{traceback.format_exc()}"
+            error_trace = traceback.format_exc()
+            return f"Error executing chat:\n{str(e)}\n\nTraceback:\n{error_trace}"
 
-    def format_function_call(self, function_name, parameters):
-        """Format a function call into the HuggingChat format."""
+    def execute(self, message, system_prompt, functions):
+        future = self.executor.submit(self._execute_chat, message, system_prompt, functions)
         try:
-            if function_name not in self.functions:
-                return False, f"Function '{function_name}' not found"
-            
-            function_def = self.functions[function_name]
-            function_call = {
-                "name": function_name,
-                "arguments": parameters
-            }
-            
-            return True, json.dumps(function_call, indent=2)
+            output = future.result(timeout=30)
+            return output
+        except concurrent.futures.TimeoutError:
+            return "Error: Timed out waiting for chat response"
         except Exception as e:
-            return False, f"Error formatting function call: {str(e)}"
+            return f"Error: {str(e)}"
 
     def __del__(self):
         self.executor.shutdown(wait=False)
+        shutil.rmtree(self.temp_dir.name)
+
+def wrapper_execute(message, system_prompt, functions):
+    executor = HuggingChatExecutor()
+    return executor.execute(message, system_prompt, functions)
 
 def create_interface():
-    manager = FunctionManager()
-    
-    with gr.Blocks(theme=gr.themes.Soft()) as demo:
-        gr.Markdown("# HuggingChat Function Manager")
-        
-        with gr.Tab("Add Function"):
-            function_json = gr.Textbox(
-                label="Function Definition (JSON)",
-                placeholder='''{
-    "name": "example_function",
-    "description": "An example function",
-    "parameters": {
-        "type": "object",
-        "properties": {
-            "param1": {
-                "type": "string",
-                "description": "First parameter"
-            }
-        },
-        "required": ["param1"]
-    }
-}''',
-                lines=10
-            )
-            add_output = gr.Textbox(label="Result", lines=2)
-            add_btn = gr.Button("Add Function", variant="primary")
-            add_btn.click(
-                manager.add_function,
-                inputs=[function_json],
-                outputs=[add_output]
-            )
+    with gr.Blocks() as demo:
+        gr.Markdown("# HuggingChat Tool Executor")
+        gr.Markdown("Execute chat completions with function calling capabilities")
 
-        with gr.Tab("List Functions"):
-            list_output = gr.Markdown(label="Registered Functions")
-            list_btn = gr.Button("List Functions")
-            list_btn.click(
-                manager.list_functions,
-                outputs=[list_output]
-            )
+        gr.Markdown("### Message")
+        message_input = gr.Textbox(
+            label="User Message",
+            placeholder="Enter your message here...",
+            lines=3
+        )
 
-        with gr.Tab("Format Function Call"):
-            with gr.Row():
-                call_name = gr.Textbox(label="Function Name", lines=1)
-                call_params = gr.Textbox(
-                    label="Parameters (JSON)",
-                    placeholder='{"param1": "value1"}',
-                    lines=5
-                )
-            format_output = gr.Textbox(label="Formatted Call", lines=5)
-            format_btn = gr.Button("Format Call")
-            format_btn.click(
-                lambda n, p: manager.format_function_call(n, json.loads(p))[1],
-                inputs=[call_name, call_params],
-                outputs=[format_output]
-            )
+        gr.Markdown("### System Prompt")
+        system_input = gr.Textbox(
+            label="System Prompt",
+            value="You are a helpful AI assistant.",
+            lines=2
+        )
+
+        gr.Markdown("### Functions")
+        functions_input = gr.Textbox(
+            label="Functions (JSON array)",
+            placeholder='''[
+    {
+        "name": "get_weather",
+        "description": "Get the weather in a location",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "location": {
+                    "type": "string",
+                    "description": "The city and state, e.g. San Francisco, CA"
+                }
+            },
+            "required": ["location"]
+        }
+    }
+]''',
+            lines=10
+        )
+
+        gr.Markdown("### Output")
+        output_text = gr.Textbox(label="Response", lines=20)
 
+        run_button = gr.Button("Run")
+        run_button.click(
+            wrapper_execute,
+            inputs=[message_input, system_input, functions_input],
+            outputs=output_text
+        )
     return demo
 
 if __name__ == "__main__":