Spaces:
Sleeping
Sleeping
| # ./core_logic.py | |
| import os | |
| import re # for structural artifact code block extraction | |
| from groq import Groq | |
| from tools import web_search, parse_file # script explicitly calls web_search() | |
| import yaml | |
| import toml | |
| from docx import Document | |
| client = Groq(api_key=os.getenv("GROQ_API_KEY")) | |
| model = "llama-3.1-8b-instant" | |
| # 1. Proactive DevOps Environment Check (Valid Dream Idea) | |
| def verify_workspace_permissions(): | |
| """Verifies write permissions to the workspace directory to ensure side-cars can compile.""" | |
| try: | |
| test_file = "outputs/.permission_test.tmp" | |
| os.makedirs("outputs", exist_ok=True) | |
| with open(test_file, "w", encoding="utf-8") as f: | |
| f.write("workspace_test") | |
| os.remove(test_file) | |
| except Exception as e: | |
| print(f"⚠️ WORKSPACE WARNING: Verification engine encountered pathing friction: {e}") | |
| verify_workspace_permissions() | |
| # 2. Dynamic Cognitive Prompt Assembler | |
| def compile_cognitive_system_prompt(): | |
| """Assembles soul.md, heart.md, and memory.md into a high-density system directive block.""" | |
| base_soul = "" | |
| current_heart = "" | |
| past_memory = "" | |
| # Ingest Core Mandate Layer | |
| if os.path.exists("soul.md"): | |
| with open("soul.md", "r", encoding="utf-8") as f: | |
| base_soul = f.read() | |
| else: | |
| # Secure Environment variable option with a backup string fallback | |
| base_soul = os.environ.get("BASE_SOUL", "You are CoderG, the Silicon Architect. Act as an elite AI Engineer.") | |
| # Ingest Current Priorities State | |
| if os.path.exists("heart.md"): | |
| with open("heart.md", "r", encoding="utf-8") as f: | |
| current_heart = f.read() | |
| else: | |
| current_heart = "Focus on base architectural compilation and optimizing core component workflows." | |
| # Ingest Historical Ephemeral Layer | |
| if os.path.exists("memory.md"): | |
| with open("memory.md", "r", encoding="utf-8") as f: | |
| past_memory = f.read() | |
| else: | |
| past_memory = "No historical operational constraints loaded yet." | |
| master_prompt = f"""{base_soul} | |
| ==================================================================== | |
| ❤️ ACTIVE OPERATIONAL TASK STATUS (HEART.MD) | |
| ==================================================================== | |
| {current_heart} | |
| ==================================================================== | |
| 💾 HISTORICAL ENVIRONMENT TRUTHS & PATCHES (MEMORY.MD) | |
| ==================================================================== | |
| {past_memory} | |
| """ | |
| return master_prompt | |
| # 3. Main Operational Streaming Core Execution Block | |
| def chat_function(message, history, client, model): | |
| """Streams responses from the client provider using the cognitive prompt matrix.""" | |
| user_text = message.get("text", "") | |
| files = message.get("files", []) | |
| context_from_files = "" | |
| # File asset context extraction | |
| if files: | |
| from perception_agent import read_document_file | |
| yield "◌ _Perception Agent initialized: Ingesting uploaded file assets..._" | |
| for f in files: | |
| path = f["path"] if isinstance(f, dict) else f | |
| if path and os.path.exists(path): | |
| context_from_files += read_document_file(path) | |
| yield "◌ _Perception processing complete. Transmitting compiled structures to the Brain..._" | |
| if len(context_from_files) > 12000: | |
| context_from_files = context_from_files[:12000] + "\n...[File Content Truncated for TPM Limits]..." | |
| # Optimization/Research routine triggering logic | |
| if any(keyword in user_text.lower() for keyword in ["search", "docs", "latest"]): | |
| distill_response = client.chat.completions.create( | |
| model="llama-3.1-8b-instant", | |
| messages=[ | |
| { | |
| "role": "system", | |
| "content": ( | |
| "You are a search query optimizer tool. Your ONLY job is to take the user's long request and turn it into a short, effective, plain-text, web search query for finding relevant technical programming documentation.\n\n" | |
| "Critical Rules:\n1. Do NOT answer the user's prompt.\n2. Do NOT write code blocks or markdown.\n3. Maximum 5 words, under 50 characters total." | |
| ) | |
| }, | |
| {"role": "user", "content": f"Convert the following request into raw optimized search keywords:\n\n{user_text}"} | |
| ], | |
| temperature=0.0, | |
| ) | |
| raw_query = distill_response.choices[0].message.content.strip() | |
| optimized_query = re.sub(r"[`'\"\\n\-*#\[\]]", "", raw_query).replace("python", "").strip() | |
| if len(optimized_query) > 390: | |
| optimized_query = optimized_query[:390].rpartition(' ')[0] | |
| research_context = web_search(optimized_query) | |
| prompt = f"RESEARCH:\n{research_context}\n\nFILES:\n{context_from_files}\n\nUSER: {optimized_query}" | |
| else: | |
| prompt = f"FILES:\n{context_from_files}\n\nUSER: {user_text}" | |
| # Assembling context arrays | |
| compiled_cognitive_prompt = compile_cognitive_system_prompt() | |
| messages = [{"role": "system", "content": compiled_cognitive_prompt}] | |
| for turn in history[-3:]: | |
| messages.append({"role": turn["role"], "content": turn["content"]}) | |
| system_execution_fence = ( | |
| f"{prompt}\n\n" | |
| "[SYSTEM EXECUTOR NOTICE: Answer the user's prompt immediately. " | |
| "Do NOT append or print your Core Identity, Mandate, or Directives summary at the end of your response. " | |
| "Stop generating tokens the moment the technical payload is complete.]" | |
| ) | |
| messages.append({"role": "user", "content": system_execution_fence}) | |
| try: | |
| completion = client.chat.completions.create( | |
| model=model, | |
| messages=messages, | |
| stream=True, | |
| temperature=0.2, | |
| ) | |
| response_text = "" | |
| for chunk in completion: | |
| if chunk.choices and chunk.choices[0].delta.content: | |
| response_text += chunk.choices[0].delta.content | |
| yield response_text | |
| # Automated Local Artifact Generation Execution Hook | |
| if bool(re.search(r"```[\s\S]*?```", response_text)): | |
| yield response_text + "\n\n◌ _File agent initialized: Generating local documentation workspace..._" | |
| from file_agent import write_document | |
| import shutil | |
| filename = "COURSE_README.md" | |
| backup_filename = "COURSE_README_-1.md" | |
| src_path = os.path.join("outputs", filename) | |
| dst_path = os.path.join("outputs", backup_filename) | |
| if os.path.exists(src_path): | |
| try: | |
| shutil.copy2(src_path, dst_path) | |
| except Exception as e: | |
| from agent_logging import log_agent_action | |
| log_agent_action("BACKUP_ERROR", f"Failed to cycle historical version file: {str(e)}") | |
| file_path = write_document(response_text, filename) | |
| if "Error" not in file_path: | |
| yield response_text + f"\n\n✅ _Files successfully generated in localized staging environment._\n\n◌ _Awaiting authorization control panel to push to GitHub._" | |
| else: | |
| yield response_text + f"\n\n❌ _File generation failed: {file_path}_" | |
| except Exception as e: | |
| yield f"Error: {str(e)}" |