Business Plan #10643

Prompt: import jsonimport osimport datetimeimport subprocessimport tracebackfrom typing import List, Dict, Any# NOTE: Replace this with your preferred LLM provider client library (e.g., openai, anthropic, google-genai)# For this blueprint, we assume a standard chat completion API interface.from openai import OpenAIclient = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))MEMORY_FILE = "omni_memory.json"# ==========================# 1. MEMORY SYSTEM# ==========================class Memory: def __init__(self): self.data = [] if os.path.exists(MEMORY_FILE): try: with open(MEMORY_FILE, "r") as f: self.data = json.load(f) except json.JSONDecodeError: self.data = [] def store(self, category: str, item: Any): self.data.append({ "timestamp": str(datetime.datetime.now()), "category": category, "data": item }) self.save() def get_recent_context(self, limit: int = 5) -> List[Dict]: return self.data[-limit:] def save(self): with open(MEMORY_FILE, "w") as f: json.dump(self.data, f, indent=4)# ==========================# 2. BIOINFORMATICS TOOL REGISTRY# ==========================class GeneticTools: """Built-in computational tools the agent can invoke to solve reconstruction problems.""" @staticmethod def reverse_complement(sequence: str) -> str: """Returns the reverse complement of a DNA sequence string.""" complement = {'A': 'T', 'T': 'A', 'C': 'G', 'G': 'C', 'N': 'N'} return "".join(complement.get(base, base) for base in reversed(sequence.upper())) @staticmethod def calculate_gc_content(sequence: str) -> float: """Calculates the GC percentage of a sequence.""" seq = sequence.upper() gc_count = seq.count('G') + seq.count('C') return (gc_count / len(seq)) * 100 if seq else 0.0# ==========================# 3. REASONING & HYPOTHESIS ENGINE# ==========================class CognitiveEngine: def __init__(self, model_name: str = "gpt-4o"): self.model_name = model_name def reason(self, objective: str, history: List[Dict], observation: Dict) -> Dict[str, Any]: """Executes the core thinking loop: analyzes data, creates a hypothesis, and chooses an action.""" system_prompt = """You are the OMNI Autonomous Core, an advanced cognitive architecture designed for complex genetic reconstruction and data analysis.You process input objectives by reasoning step-by-step.You must output your response in STRICTOR JSON format with the following keys:{ "analysis": "Your step-by-step reasoning about the problem", "hypothesis": "Your proposed scientific theory or approach", "action_type": "GENERATE_CODE" or "USE_TOOL" or "FINAL_ANSWER", "action_payload": "The actual python code string, tool parameters, or final text answer depending on action_type"}""" prompt = f"""Current Objective: {objective}Current Environment Observation: {json.dumps(observation)}Recent Memory Context: {json.dumps(history)}Provide your next logical cognitive step in the requested JSON schema:""" try: response = client.chat.completions.create( model=self.model_name, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], response_format={"type": "json_object"} ) return json.loads(response.choices[0].message.content) except Exception as e: return { "analysis": f"Failed to generate cognitive step due to API error: {str(e)}", "hypothesis": "None", "action_type": "FINAL_ANSWER", "action_payload": "Error in core LLM engine runtime." }# ==========================# 4. SECURE ISOLATED EXECUTION SANDBOX# ==========================class CodeSandbox: @staticmethod def execute_safely(script_contents: str) -> Dict[str, Any]: """Writes and executes code dynamically inside a monitored temporary file runtime.""" temp_filename = "omni_runtime_exec.py" with open(temp_filename, "w") as f: f.write(script_contents) try: # Executes the script with a strict 30-second timeout window result = subprocess.run( ["python", temp_filename], capture_output=True, text=True, timeout=30 ) return { "success": result.returncode == 0, "stdout": result.stdout, "stderr": result.stderr } except subprocess.TimeoutExpired: return {"success": False, "stdout": "", "stderr": "Execution timed out after 30 seconds."} except Exception as e: return {"success": False, "stdout": "", "stderr": str(e)} finally: if os.path.exists(temp_filename): os.remove(temp_filename)# ==========================# 5. ENVIRONMENT INTERFACE# ==========================class Environment: def observe(self) -> Dict[str, Any]: """Scans local workspace directory parameters to feed structural awareness to Omni.""" files = [f for f in os.listdir('.') if os.path.isfile(f)] return { "status": "active_run_mode", "working_directory_files": files, "available_tools": ["reverse_complement", "calculate_gc_content"] }# ==========================# 6. RE-CONFIGURED OMNI CORE# ==========================class OmniCore: def __init__(self): self.memory = Memory() self.cognitive_engine = CognitiveEngine() self.sandbox = CodeSandbox() self.environment = Environment() def run_cycle(self, objective: str): print(f"\n[OMNI LOG] Initiating Core Cycle for Objective: '{objective}'") # 1. Environmental awareness scan observation = self.environment.observe() # 2. Retrieve history context history = self.memory.get_recent_context() # 3. LLM Reasoning step cognitive_step = self.cognitive_engine.reason(objective, history, observation) print(f"\n[THOUGHT PROCESS]\n-> Analysis: {cognitive_step.get('analysis')}") print(f"-> Hypothesis: {cognitive_step.get('hypothesis')}") print(f"-> Selected Action: {cognitive_step.get('action_type')}") action_type = cognitive_step.get("action_type") payload = cognitive_step.get("action_payload") execution_result = None # 4. Execution Loop branch based on agent's choice if action_type == "GENERATE_CODE": print("[EXECUTING CODE IN SANDBOX]...") execution_result = self.sandbox.execute_safely(payload) print(f"-> Execution Success: {execution_result['success']}") if not execution_result['success']: print(f"-> Runtime Error: {execution_result['stderr']}") elif action_type == "USE_TOOL": # Native function calling wrapper try: tool_name = payload.get("tool_name") args = payload.get("args", []) tool_func = getattr(GeneticTools, tool_name) execution_result = {"success": True, "output": tool_func(*args)} except Exception as e: execution_result = {"success": False, "error": str(e)} elif action_type == "FINAL_ANSWER": execution_result = {"success": True, "output": payload} # 5. Commit state change to permanent memory self.memory.store("cognitive_cycle", { "objective": objective, "cognitive_step": cognitive_step, "execution_result": execution_result }) print("\n[CYCLE OUTPUT INTERFACE]") print(json.dumps(execution_result, indent=4))# ==========================# INITIALIZATION RUNTIME# ==========================if __name__ == "__main__": print("=========================================") print(" PROJECT OMNI AUTONOMOUS CORE ") print("=========================================") # Ensure environment API key variable exists before boots if not os.getenv("OPENAI_API_KEY"): print("[WARNING]: OPENAI_API_KEY environment variable not set. Runtime execution will throw errors.") omni = OmniCore() while True: user_input = input("\nEnter genetic/coding objective (or type 'exit'): ") if user_input.strip().lower() == 'exit': break if not user_input.strip(): continue omni.run_cycle(user_input)