Spaces:
Sleeping
Sleeping
| # app.py | |
| import gradio as gr | |
| from transformers import AutoTokenizer, AutoModelForCausalLM | |
| import requests | |
| import importlib.util | |
| import os | |
| import hashlib | |
| import json | |
| from datetime import datetime | |
| # Configuration | |
| 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 | |
| } | |
| # Load model and tokenizer | |
| tokenizer = AutoTokenizer.from_pretrained(CONFIG["model_name"]) | |
| model = AutoModelForCausalLM.from_pretrained(CONFIG["model_name"]) | |
| # Security verification | |
| def verify_file_integrity(content, expected_hash): | |
| sha256 = hashlib.sha256() | |
| sha256.update(content.encode('utf-8')) | |
| return sha256.hexdigest() == expected_hash | |
| # Update mechanism | |
| def check_for_updates(): | |
| try: | |
| # Get update manifest | |
| response = requests.get(CONFIG["manifest_url"]) | |
| manifest = response.json() | |
| # 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 = datetime.fromisoformat(f.read().strip()) | |
| 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["modules"]: | |
| module_info = manifest["modules"][module_name] | |
| file_path = f"{module_name}.py" | |
| # Download and verify update | |
| response = requests.get(CONFIG["update_url"] + file_path) | |
| if response.status_code == 200: | |
| content = response.text | |
| if verify_file_integrity(content, module_info["sha256"]): | |
| # Save new version | |
| 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 Exception as e: | |
| return f"Update failed: {str(e)}" | |
| # Dynamic module loader | |
| def load_module(module_name): | |
| 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 | |
| # Load core modules | |
| response_handler = load_module("response_handler") | |
| updater = load_module("updater") | |
| # Main processing function | |
| def process_query(prompt): | |
| # Check for update command | |
| if "/update" in prompt: | |
| return check_for_updates() | |
| # Normal processing | |
| return response_handler.generate_response(prompt, tokenizer, model) | |
| # Create Gradio interface | |
| 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) |