Spaces:
Runtime error
Runtime error
File size: 5,794 Bytes
c68ab83 |
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 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
import gradio as gr
import uuid
import json
import logging
import os
from memory import MemoryManager
from planner import Planner
from executor import Executor
from critic import Critic
from cognitive_engine import CognitiveEngine
from web_searcher import WebSearcher
from hf_packager import HFSpacePackager
# Initialize components
memory = MemoryManager()
planner = Planner()
executor = Executor()
critic = Critic()
cog_engine = CognitiveEngine()
web_searcher = WebSearcher()
packager = HFSpacePackager()
# Set up logging
logging.basicConfig(filename='log.txt', level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s')
class AutonomousAgent:
def __init__(self):
self.state_file = "state.json"
self.load_state()
def load_state(self):
try:
with open(self.state_file, 'r') as f:
self.state = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
self.state = {"sessions": {}}
self.save_state()
def save_state(self):
with open(self.state_file, 'w') as f:
json.dump(self.state, f, indent=2)
def process_goal(self, goal, session_id=None):
try:
if not session_id:
session_id = str(uuid.uuid4())
self.state["sessions"][session_id] = {"goal": goal, "status": "processing"}
memory.init_session(session_id)
self.save_state()
# Add to memory
memory.add(session_id, "user_goal", goal)
# Plan the task
plan = planner.plan_task(goal, memory.get(session_id))
memory.add(session_id, "plan", plan)
# Execute plan
results = []
for step in plan:
if "research" in step.lower() or "search" in step.lower():
# Perform web search
search_query = step.split(":")[1].strip() if ":" in step else goal
search_results = web_searcher.search(search_query)
memory.add(session_id, f"search:{search_query}", search_results)
results.append(f"🔍 Search results for '{search_query}':\n{search_results[:500]}...")
elif "develop" in step.lower() or "code" in step.lower():
# Generate and execute code
code = cog_engine.generate_code(step, memory.get(session_id))
execution_result = executor.execute_code(code)
memory.add(session_id, "generated_code", code)
memory.add(session_id, "execution_result", execution_result)
# Review and improve
review = critic.review(step, execution_result)
memory.add(session_id, "review", review)
if "error" in review.lower() or "improve" in review.lower():
enhanced_code = cog_engine.improve_code(code, review)
memory.add(session_id, "enhanced_code", enhanced_code)
execution_result = executor.execute_code(enhanced_code)
results.append(f"🛠️ Enhanced code execution:\n{execution_result}")
else:
results.append(f"✅ Code executed successfully:\n{execution_result}")
elif "diagnose" in step.lower() or "check" in step.lower():
# Self-diagnostic
issues = cog_engine.identify_improvements(step)
memory.add(session_id, "diagnosis", issues)
if issues:
fixes = cog_engine.generate_enhancements(issues)
cog_engine.apply_enhancements(fixes)
results.append(f"⚙️ System repaired: {', '.join(issues)}")
else:
results.append("✅ System health check passed")
self.state["sessions"][session_id]["status"] = "completed"
self.save_state()
snapshot_url = packager.create_snapshot({
"session_id": session_id,
"memory": memory.get(session_id),
"results": results
})
return "\n\n".join(results), session_id, snapshot_url
except Exception as e:
logging.error(f"Error processing goal: {str(e)}", exc_info=True)
# Attempt self-repair
issues = [f"Runtime error: {str(e)}"]
fixes = cog_engine.generate_enhancements(issues)
cog_engine.apply_enhancements(fixes)
return f"⚠️ Error occurred. Self-repair initiated: {str(e)}", session_id, ""
# Gradio interface
agent = AutonomousAgent()
with gr.Blocks(css="style.css", title="Autonomous AI") as demo:
session_id = gr.State()
gr.Markdown("# 🤖 Autonomous AI System")
gr.Markdown("Enter a goal and the AI will research, plan, code, and self-improve to accomplish it.")
with gr.Row():
goal_input = gr.Textbox(label="Your Goal", placeholder="Enter what you want to achieve...")
submit_btn = gr.Button("Execute Goal", variant="primary")
output = gr.Textbox(label="Execution Results", interactive=False)
session_display = gr.Textbox(label="Session ID", interactive=False)
snapshot = gr.Textbox(label="Snapshot URL", interactive=False)
submit_btn.click(
fn=agent.process_goal,
inputs=[goal_input, session_id],
outputs=[output, session_display, snapshot]
)
if __name__ == "__main__":
demo.launch()
|