diff --git a/.github/scripts/MatrixXx.py b/.github/scripts/MatrixXx.py new file mode 100644 index 0000000..b9756e7 --- /dev/null +++ b/.github/scripts/MatrixXx.py @@ -0,0 +1,154 @@ +import asyncio +import logging +from typing import Callable, Any, Dict, List +import nimpy +from clang import cindex +from langchain.llms import OpenAI +from langchain.chains import ConversationChain +from langchain.memory import InMemoryConversationMemory + +logging.basicConfig(level=logging.DEBUG) +logger = logging.getLogger("AgenticMatrixLivingCodeAI") + +# --- Living Environment --- +class LivingEnvironment: + def __init__(self): + self.state: Dict[str, Any] = {} + self.subscribers: List[Callable[[str, Any], asyncio.Future]] = [] + + def subscribe(self, callback: Callable[[str, Any], asyncio.Future]): + self.subscribers.append(callback) + + async def update_state(self, key: str, value: Any): + logger.debug(f"[Env] Update: {key} -> {value}") + self.state[key] = value + await asyncio.gather(*[cb(key, value) for cb in self.subscribers]) + + def get_state(self, key: str): + return self.state.get(key) + +# --- Hybrid Agent with Living Code and Multi-Language Execution --- +class HybridAgent: + def __init__(self, name: str, env: LivingEnvironment, initial_prompt: str): + self.name = name + self.env = env + self.prompt = initial_prompt + self.memory = InMemoryConversationMemory() + self.llm = None + self.chain = None + self.llm_ready = False + try: + self.llm = OpenAI(temperature=0.3) + self.chain = ConversationChain(llm=self.llm, memory=self.memory) + self.llm_ready = True + except Exception as e: + logger.warning(f'[{self.name}] OpenAI LLM not available: {e}') + self.running = True + self.env.subscribe(self.on_env_update) + + self.clang_index = cindex.Index.create() + try: + self.nim_agent = nimpy.import_module("nim_agent") + self.nim_ready = True + except Exception as e: + logger.warning(f"[{self.name}] Nim module not available: {e}") + self.nim_ready = False + + async def on_env_update(self, key: str, value: Any): + if key == f"prompt_update_{self.name}": + logger.info(f"[{self.name}] Prompt update received") + self.prompt = value + if key == f"code_update_{self.name}": + logger.info(f"[{self.name}] Received C++ code update") + diag = self.parse_cpp_code(value) + for d in diag: + logger.warning(f"[{self.name}][Clang] {d.spelling}") + + def parse_cpp_code(self, code: str): + tu = self.clang_index.parse("dynamic_code.cpp", args=['-std=c++17'], unsaved_files=[('dynamic_code.cpp', code)]) + return list(tu.diagnostics) + + async def execute_nim_logic(self, data: str): + if not self.nim_ready: + logger.warning(f"[{self.name}] Skipping Nim execution (module unavailable)") + return + result = self.nim_agent.process_buffer(data) + logger.debug(f"[{self.name}] Nim module result: {result}") + + async def run(self): + iteration = 0 + while self.running: + env_snapshot = str(self.env.state) + full_prompt = f"{self.prompt}\nEnv State:\n{env_snapshot}\nIteration: {iteration}" + try: + llm_resp = self.chain.invoke(full_prompt) + logger.info(f"[{self.name}] LLM output: {llm_resp[:250]}...") + await self.env.update_state(f"status_{self.name}", llm_resp) + except Exception as e: + logger.error(f"[{self.name}] LLM call failed: {e}") + + # Example: dynamic C++ snippet (could be realtime updated) + cpp_code = """ + int add(int a, int b) { return a + b; } + """ + self.parse_cpp_code(cpp_code) + + # Nim logic invocation with complex data string + await self.execute_nim_logic(f"Nim data iteration {iteration} from {self.name}") + + await asyncio.sleep(6) + iteration += 1 + + def stop(self): + logger.info(f"[{self.name}] Stopping execution.") + self.running = False + +# --- Host AI Manager --- +class HostAgent: + def __init__(self): + self.env = LivingEnvironment() + self.agents: List[HybridAgent] = [] + self.running = True + + def register_agent(self, agent: HybridAgent): + logger.info(f"[Host] Registering agent {agent.name}") + self.agents.append(agent) + + async def update_loop(self): + count = 0 + while self.running and count < 8: + await asyncio.sleep(10) + for agent in self.agents: + prompt = f"Agent update cycle {count} for {agent.name}. Adapt behavior dynamically." + await self.env.update_state(f"prompt_update_{agent.name}", prompt) + + # Optional dynamic new C++ source for compilation + new_cpp_code = f"int dynamic_func() {{ return {count} * 42; }}" + await self.env.update_state(f"code_update_{agent.name}", new_cpp_code) + + count += 1 + + logger.info("[Host] All update cycles complete, stopping agents.") + for agent in self.agents: + agent.stop() + + async def run(self): + tasks = [asyncio.create_task(agent.run()) for agent in self.agents] + updater = asyncio.create_task(self.update_loop()) + await asyncio.gather(updater, *tasks) + self.running = False + +# --- Main Entrypoint --- +async def main(): + host = HostAgent() + agents = [ + HybridAgent("AlphaBot", host.env, "You are AlphaBot, leading agentic matrix coordination."), + HybridAgent("BetaBot", host.env, "You are BetaBot, specialist in adaptive code evolution."), + HybridAgent("GammaBot", host.env, "You are GammaBot, UI and UX genius mirroring living interfaces.") + ] + for ag in agents: + host.register_agent(ag) + await host.run() + +if __name__ == "__main__": + asyncio.run(main())