Date: Sun, 3 Aug 2025 17:54:58 +0530
Subject: [PATCH 4/4] Update ai_engine.py
---
core/ai_engine.py | 334 ++++++++++++++++++++++++++++++++--------------
1 file changed, 233 insertions(+), 101 deletions(-)
diff --git a/core/ai_engine.py b/core/ai_engine.py
index a08d8e38..c9c30342 100644
--- a/core/ai_engine.py
+++ b/core/ai_engine.py
@@ -5,154 +5,286 @@
import google.generativeai as genai
from meta_ai_api import MetaAI
+# Configure logging
logging.basicConfig(
filename="karbon_ai_errors.log",
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s"
)
+# Global AI status
ai_status = {"state": "connecting", "message": "Connecting to AI service..."}
def set_ai_status(state: str, message: str):
+ """Update AI status with proper error handling."""
ai_status["state"] = state
ai_status["message"] = message
logging.info(f"AI Status Updated: {state} - {message}")
def extract_json(response: str) -> dict:
"""
- Extract and parse JSON from the AI response, handling malformed input.
+ Extract and parse JSON from the AI response with improved error handling.
"""
if not response or response.isspace():
logging.error("Empty AI response received")
raise ValueError("Empty response from AI service")
+
+ logging.info("Raw AI response length: %d", len(response))
+
+ # Clean the response more thoroughly
+ cleaned_response = response.strip()
+
+ # Remove markdown code blocks
+ cleaned_response = re.sub(r'```json\s*', '', cleaned_response, flags=re.IGNORECASE)
+ cleaned_response = re.sub(r'```\s*$', '', cleaned_response, flags=re.MULTILINE)
+ cleaned_response = re.sub(r'^```|```$', '', cleaned_response, flags=re.MULTILINE)
+
+ # Remove any leading/trailing non-JSON text
+ cleaned_response = cleaned_response.strip()
+
+ logging.info("Cleaned response preview: %s", cleaned_response[:200] + "..." if len(cleaned_response) > 200 else cleaned_response)
+
+ # Try multiple JSON extraction strategies
+ strategies = [
+ lambda x: x, # Try as-is first
+ lambda x: find_json_block(x), # Find JSON block
+ lambda x: extract_between_braces(x), # Extract content between first { and last }
+ ]
+
+ for i, strategy in enumerate(strategies):
+ try:
+ candidate = strategy(cleaned_response)
+ if not candidate:
+ continue
+
+ logging.info(f"Strategy {i+1} candidate: %s", candidate[:200] + "..." if len(candidate) > 200 else candidate)
+
+ # Parse JSON
+ parsed = json.loads(candidate)
+
+ # Validate structure
+ if validate_json_structure(parsed):
+ logging.info("Successfully extracted and validated JSON")
+ return parsed
+ else:
+ logging.warning(f"Strategy {i+1}: JSON structure validation failed")
+
+ except json.JSONDecodeError as e:
+ logging.warning(f"Strategy {i+1}: JSON decode failed - {e}")
+ continue
+ except Exception as e:
+ logging.warning(f"Strategy {i+1}: General error - {e}")
+ continue
+
+ # If all strategies fail, try to create a minimal valid response
+ logging.error("All JSON extraction strategies failed")
+ raise ValueError("Failed to extract valid JSON from AI response")
- logging.info("Raw AI response: %s", response)
-
- # Remove markdown formatting
- cleaned_response = re.sub(r'```json\s*|\s*```', '', response, flags=re.DOTALL)
- cleaned_response = re.sub(r'^\s*```|```\s*$', '', cleaned_response, flags=re.DOTALL).strip()
-
- logging.info("Cleaned AI response: %s", cleaned_response[:500])
-
- # Extract first valid JSON object from the string
- try:
- stack = []
- start_idx = None
- json_block = None
- for i, char in enumerate(cleaned_response):
- if char == '{':
- if not stack:
- start_idx = i
- stack.append(char)
- elif char == '}':
- if stack:
- stack.pop()
- if not stack:
- json_block = cleaned_response[start_idx:i + 1]
- break
- if not json_block:
- raise ValueError("No valid JSON object found in response.")
+def find_json_block(text: str) -> str:
+ """Find the first complete JSON object in the text."""
+ brace_count = 0
+ start_idx = None
+
+ for i, char in enumerate(text):
+ if char == '{':
+ if brace_count == 0:
+ start_idx = i
+ brace_count += 1
+ elif char == '}':
+ brace_count -= 1
+ if brace_count == 0 and start_idx is not None:
+ return text[start_idx:i + 1]
+
+ return ""
- logging.info("Extracted JSON block: %s", json_block[:500])
+def extract_between_braces(text: str) -> str:
+ """Extract content between the first { and last }."""
+ first_brace = text.find('{')
+ last_brace = text.rfind('}')
+
+ if first_brace != -1 and last_brace != -1 and first_brace < last_brace:
+ return text[first_brace:last_brace + 1]
+
+ return ""
- parsed = json.loads(json_block)
- expected_keys = {"html", "css", "js", "name"}
- if not isinstance(parsed, dict) or not expected_keys.issubset(parsed.keys()):
- raise ValueError(f"Parsed JSON does not contain expected keys: {expected_keys}")
- return parsed
+def validate_json_structure(parsed: dict) -> bool:
+ """Validate that the parsed JSON has the expected structure."""
+ if not isinstance(parsed, dict):
+ return False
+
+ required_keys = {"html", "css", "js", "name"}
+ if not required_keys.issubset(parsed.keys()):
+ logging.warning(f"Missing required keys. Expected: {required_keys}, Got: {set(parsed.keys())}")
+ return False
+
+ # Check that values are strings
+ for key in required_keys:
+ if not isinstance(parsed[key], str):
+ logging.warning(f"Key '{key}' is not a string: {type(parsed[key])}")
+ return False
+
+ return True
- except json.JSONDecodeError as e:
- logging.error(f"JSON decode failed: {e}")
- raise ValueError("Failed to parse JSON.")
- except Exception as general_e:
- logging.error(f"JSON extraction failed: {general_e}")
- raise ValueError("Failed to extract valid JSON.")
+def create_fallback_response(prompt: str) -> dict:
+ """Create a basic fallback response when AI fails."""
+ return {
+ "html": f"Generated AppWelcome
App generated from: {prompt[:100]}{'...' if len(prompt) > 100 else ''}
",
+ "css": "body { font-family: Arial, sans-serif; margin: 40px; background: #f5f5f5; }",
+ "js": "console.log('App loaded successfully');",
+ "name": "Generated App"
+ }
-def generate_code_from_prompt(prompt: str, api_key: str = None, retries=2) -> str:
- formatted = (
- f"You are a helpful assistant that writes complete frontend apps.\n"
- f"Given the task: \"{prompt}\"\n"
- f"Respond ONLY in this JSON format, with no additional text, markdown, or explanation before or after the JSON:\n"
+def generate_code_from_prompt(prompt: str, api_key: str = None, retries: int = 3) -> str:
+ """
+ Generate code from prompt with improved error handling and fallbacks.
+ """
+ formatted_prompt = (
+ f"Create a complete web application for: {prompt}\n\n"
+ f"Respond with ONLY a JSON object in this exact format (no markdown, no extra text):\n"
"{\n"
- " \"html\": \"...\",\n"
- " \"css\": \"body { ... }\",\n"
- " \"js\": \"document.addEventListener(...)\",\n"
- " \"name\": \"App Name\"\n"
- "}\n"
- "IMPORTANT: Do NOT include markdown (like ```json) or any trailing explanation or comments. Only return pure JSON."
+ ' "html": "...",\n'
+ ' "css": "body { ... }",\n'
+ ' "js": "// JavaScript code here",\n'
+ ' "name": "App Name"\n'
+ "}\n\n"
+ "CRITICAL: Return ONLY the JSON object, no explanations, no markdown formatting."
)
-
+
+ last_error = None
+
for attempt in range(retries + 1):
try:
- set_ai_status("generating", "Generating code...")
+ set_ai_status("generating", f"Generating code (attempt {attempt + 1})...")
+ response = None
+
+ # Try Gemini first if API key provided
if api_key:
try:
genai.configure(api_key=api_key)
- model = genai.GenerativeModel('gemini-2.5-flash')
- response = model.generate_content(formatted).text
- logging.info(f"[Gemini] Raw AI response: {response}")
+ model = genai.GenerativeModel('gemini-2.0-flash-exp')
+ result = model.generate_content(formatted_prompt)
+ response = result.text
+ logging.info("Gemini response received")
except Exception as gem_e:
- logging.warning(f"[Gemini Fallback] Gemini failed: {gem_e}. Falling back to Meta AI.")
- api_key = None
- continue
-
- if not api_key:
- ai = MetaAI()
- result = ai.prompt(message=formatted)
- response = result.get("message", "")
- logging.info(f"[MetaAI] Raw AI response: {response}")
-
+ logging.warning(f"Gemini failed: {gem_e}")
+ last_error = gem_e
+
+ # Fallback to Meta AI
+ if not response:
+ try:
+ ai = MetaAI()
+ result = ai.prompt(message=formatted_prompt)
+ response = result.get("message", "")
+ logging.info("Meta AI response received")
+ except Exception as meta_e:
+ logging.warning(f"Meta AI failed: {meta_e}")
+ last_error = meta_e
+
+ if not response:
+ raise Exception("No response from any AI service")
+
+ # Extract and validate JSON
parsed = extract_json(response)
- if not parsed:
- raise ValueError("AI response couldn't be parsed into JSON.")
-
- set_ai_status("online", "AI service is online.")
- html = str(parsed.get("html", ""))
- css = str(parsed.get("css", ""))
- js = str(parsed.get("js", ""))
-
- final_code = html.replace("", f"") \
- .replace("