Spaces:
Sleeping
Sleeping
File size: 4,525 Bytes
523a801 a1a1f36 8ebb31b a1a1f36 8ebb31b a1a1f36 8ebb31b a1a1f36 8ebb31b a1a1f36 523a801 a1a1f36 523a801 a1a1f36 523a801 a1a1f36 523a801 a1a1f36 523a801 a1a1f36 8ebb31b 523a801 8ebb31b 523a801 8ebb31b a1a1f36 523a801 a1a1f36 8ebb31b a1a1f36 8ebb31b a1a1f36 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 |
# app.py (updated with better JSON handling)
import gradio as gr
from transformers import AutoTokenizer, AutoModelForCausalLM
import requests
import importlib.util
import os
import hashlib
import json
from datetime import datetime
CONFIG = {
"model_name": "deepseek-ai/deepseek-coder-1.3b-instruct",
"update_url": "https://raw.githubusercontent.com/[YOUR_USERNAME]/deepseek-updates/main/",
"manifest_url": "https://raw.githubusercontent.com/[YOUR_USERNAME]/deepseek-updates/main/manifest.json",
"allowed_modules": ["response_handler", "updater"],
"update_frequency": 6 # hours
}
tokenizer = AutoTokenizer.from_pretrained(CONFIG["model_name"])
model = AutoModelForCausalLM.from_pretrained(CONFIG["model_name"])
def verify_file_integrity(content, expected_hash):
sha256 = hashlib.sha256()
sha256.update(content.encode('utf-8'))
return sha256.hexdigest() == expected_hash
def check_for_updates():
try:
# Get update manifest with explicit error handling
response = requests.get(CONFIG["manifest_url"], timeout=10)
# Check for valid JSON response
try:
manifest = response.json()
except json.JSONDecodeError as e:
# Check if we got an HTML error page instead of JSON
if "<!DOCTYPE html>" in response.text[:50].lower():
return "Update failed: Server returned HTML instead of JSON"
return f"JSON parse error: {str(e)}"
# Check last update time
last_update_path = "last_update.txt"
if os.path.exists(last_update_path):
with open(last_update_path, 'r') as f:
last_update_str = f.read().strip()
if last_update_str:
last_update = datetime.fromisoformat(last_update_str)
time_since_update = (datetime.utcnow() - last_update).total_seconds() / 3600
if time_since_update < CONFIG["update_frequency"]:
return "Too soon for update check"
# Process updates
updates_applied = []
for module_name in CONFIG["allowed_modules"]:
if module_name in manifest.get("modules", {}):
module_info = manifest["modules"][module_name]
file_path = f"{module_name}.py"
# Download and verify update
file_response = requests.get(CONFIG["update_url"] + file_path, timeout=10)
if file_response.status_code == 200:
content = file_response.text
if verify_file_integrity(content, module_info.get("sha256", "")):
with open(file_path, 'w') as f:
f.write(content)
updates_applied.append(module_name)
# Update timestamp
with open(last_update_path, 'w') as f:
f.write(datetime.utcnow().isoformat())
return f"Updates applied to: {', '.join(updates_applied)}" if updates_applied else "No updates available"
except requests.exceptions.RequestException as e:
return f"Network error: {str(e)}"
except Exception as e:
return f"Update failed: {str(e)}"
def safe_load_module(module_name):
try:
spec = importlib.util.spec_from_file_location(module_name, f"{module_name}.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
except Exception as e:
print(f"Error loading {module_name}: {str(e)}")
# Return a fallback module
from types import ModuleType
return ModuleType(module_name)
# Load core modules with fallback
response_handler = safe_load_module("response_handler")
updater = safe_load_module("updater")
def process_query(prompt):
if "/update" in prompt:
return check_for_updates()
# Handle case where response_handler failed to load
if not hasattr(response_handler, "generate_response"):
return "System error: Response module not loaded properly"
return response_handler.generate_response(prompt, tokenizer, model)
interface = gr.Interface(
fn=process_query,
inputs=gr.Textbox(lines=3, placeholder="Enter your query..."),
outputs="text",
title="Self-Updating DeepSeek AI",
description="This AI can update its own code. Type '/update' to check for improvements."
)
if __name__ == "__main__":
interface.launch(server_port=7860, share=True) |