Krishp1/Autonomous-Coding-Agent
0
1# nodes.py — All 13 nodes for Autonomous Python Coding Agent2 3import os4import ast5import subprocess6import re7import hashlib8import importlib.util9import tempfile10 11from langchain_groq import ChatGroq12from langchain_core.messages import HumanMessage, SystemMessage13import chromadb14 15from state import State16 17# ── LLM ──────────────────────────────────18llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0)19 20# ── CHROMADB ─────────────────────────────21chroma_client = chromadb.Client()22memory_collection = chroma_client.get_or_create_collection("bug_fixes")23 24# ─────────────────────────────────────────25# NODE 1 — PLANNER26# ─────────────────────────────────────────27def planner(state: State):28 print("\n📋 Planner thinking...")29 response = llm.invoke([30 SystemMessage(content="You are a coding planner. Break tasks into clear steps."),31 HumanMessage(content=f"""32Break this coding task into clear steps:33Task: {state['task']}34 35Reply with:361. What the function should do372. Input and output format383. Edge cases to handle394. Test cases to verify40""")41 ])42 print("Plan ready")43 return {"plan": response.content}44 45# ─────────────────────────────────────────46# NODE 2 — CODER47# ─────────────────────────────────────────48def coder(state: State):49 print("\n💻 Coder writing code...")50 51 past_fixes = ""52 if state["error"]:53 try:54 results = memory_collection.query(query_texts=[state["error"]], n_results=2)55 if results["documents"][0]:56 past_fixes = "\n".join(results["documents"][0])57 print("🧠 Found past fixes in memory!")58 except Exception:59 pass60 61 response = llm.invoke([62 SystemMessage(content="""You are an expert Python developer.63Write clean working Python code WITH type hints on every function.64Return ONLY the code — no explanation, no markdown, no backticks."""),65 HumanMessage(content=f"""66Task: {state['task']}67 68Plan to follow:69{state['plan']}70 71Previous error (fix this):72{state['error'] if state['error'] else 'No errors yet — write fresh code'}73 74Reflection notes:75{state.get('reflection_notes', '') or 'None'}76 77Past fixes from memory:78{past_fixes if past_fixes else 'No past fixes available'}79 80Rules:81- Type hints on ALL functions82- Docstring on every function83- Keep it simple and readable84- MUST include demo calls inside: if __name__ == '__main__': that print results85 86Write complete working Python code only:87""")88 ])89 90 code = response.content91 code = re.sub(r"```python", "", code)92 code = re.sub(r"```", "", code)93 code = code.strip()94 95 print(f"Code written ({len(code.splitlines())} lines)")96 return {"code": code, "error": "", "fixed_code": "", "reflection_notes": ""}97 98# ─────────────────────────────────────────99# NODE 3 — AST VALIDATOR100# ─────────────────────────────────────────101def ast_validator(state: State):102 print("\n🌳 AST Validator checking syntax...")103 code = state["fixed_code"] if state["fixed_code"] else state["code"]104 105 try:106 tree = ast.parse(code)107 except SyntaxError as e:108 print(f"❌ Syntax error: {e}")109 return {"ast_valid": False, "error": f"SyntaxError at line {e.lineno}: {e.msg}"}110 111 # Hard-fail on hallucinated imports: catching this here is cheaper than112 # letting it slide through to the Tester, where it would just crash with113 # ModuleNotFoundError anyway and burn a Tester retry instead of an AST one.114 hallucinated_imports = []115 for node in ast.walk(tree):116 if isinstance(node, ast.Import):117 for alias in node.names:118 base = alias.name.split(".")[0]119 try:120 found = importlib.util.find_spec(base) is not None121 except (ImportError, ValueError, ModuleNotFoundError):122 found = False123 if not found:124 hallucinated_imports.append(base)125 elif isinstance(node, ast.ImportFrom):126 if node.module:127 base = node.module.split(".")[0]128 try:129 found = importlib.util.find_spec(base) is not None130 except (ImportError, ValueError, ModuleNotFoundError):131 found = False132 if not found:133 hallucinated_imports.append(base)134 135 if hallucinated_imports:136 unknown = list(set(hallucinated_imports))137 print(f"❌ Hallucinated/unavailable imports: {unknown}")138 return {139 "ast_valid": False,140 "error": f"Unknown or unavailable imports: {unknown}. "141 f"Rewrite using only the Python standard library or already-imported packages."142 }143 144 # Missing return type hints stay a soft warning, not a hard failure.145 # Escalating this to a blocker would burn one of only 3 AST retries on a146 # style nitpick, and if AST fails 3 times the pipeline ends early with147 # NO code shown to the user at all — far worse than a missing "-> int".148 missing = [n.name for n in ast.walk(tree)149 if isinstance(n, ast.FunctionDef) and not n.returns and n.name != "__init__"]150 if missing:151 print(f"⚠️ Missing return hints: {missing}")152 153 print("✅ AST passed!")154 return {"ast_valid": True}155 156# ─────────────────────────────────────────157# NODE 4 — TEST GENERATOR158# ─────────────────────────────────────────159def test_generator(state: State):160 print("\n🧬 Test Generator creating tests...")161 code = state["fixed_code"] if state["fixed_code"] else state["code"]162 163 response = llm.invoke([164 SystemMessage(content="""You are a Python testing expert.165Return ONLY runnable Python test code — no markdown, no backticks."""),166 HumanMessage(content=f"""167Generate test cases for this code:168 169TASK: {state['task']}170CODE:171{code}172 173Rules:174- Copy ALL function definitions inline — do NOT import from files175- Cover: normal cases, edge cases, large input176- Call each test function at the bottom to run them177- Do NOT use unittest or sys — just plain assert statements178- Print "All tests passed!" at the end if successful179 180Return ONLY runnable Python code:181""")182 ])183 184 tests = response.content185 tests = re.sub(r"```python", "", tests)186 tests = re.sub(r"```", "", tests)187 tests = tests.strip()188 189 print(f"Generated {tests.count('def test_')} test functions")190 return {"generated_tests": tests}191 192# ─────────────────────────────────────────193# NODE 5 — TESTER194# ─────────────────────────────────────────195def tester(state: State):196 print("\n🧪 Tester running code...")197 code = state["fixed_code"] if state["fixed_code"] else state["code"]198 199 try:200 result = subprocess.run(201 ["python", "-c", code],202 capture_output=True, text=True, timeout=10203 )204 205 if result.returncode == 0:206 if not result.stdout.strip():207 print("❌ No output produced")208 return {209 "test_result": "",210 "error": "Code ran but produced no output. Add print statements in if __name__ == '__main__'.",211 "passed": False212 }213 print("✅ Code passed!")214 215 test_output = ""216 if state.get("generated_tests"):217 try:218 test_run = subprocess.run(219 ["python", "-c", state["generated_tests"]],220 capture_output=True, text=True, timeout=15221 )222 if test_run.returncode == 0:223 test_output = "✅ All generated tests passed\n" + test_run.stdout224 else:225 test_output = f"⚠️ Some tests failed:\n{test_run.stderr[:200]}"226 except Exception as e:227 test_output = f"Test run error: {e}"228 229 # Promote whichever version actually passed into "code" so every230 # downstream node (hypothesis/benchmark/security/complexity/reviewer)231 # keeps working on the tested version instead of silently232 # falling back to the pre-fix original once fixed_code is cleared.233 working_code = code234 return {235 "test_result": result.stdout + "\n" + test_output,236 "error": "",237 "passed": True,238 "code": working_code,239 "fixed_code": ""240 }241 else:242 print(f"❌ Failed: {result.stderr[:80]}")243 return {"test_result": "", "error": result.stderr, "passed": False}244 245 except subprocess.TimeoutExpired:246 return {"test_result": "", "error": "Timed out after 10 seconds", "passed": False}247 except Exception as e:248 return {"test_result": "", "error": str(e), "passed": False}249 250# ─────────────────────────────────────────251# NODE 6 — HYPOTHESIS TESTER252# ─────────────────────────────────────────253def hypothesis_tester(state: State):254 print("\n🎲 Hypothesis property-based testing...")255 code = state["fixed_code"] if state["fixed_code"] else state["code"]256 hypothesis_result = "Skipped"257 258 try:259 response = llm.invoke([260 SystemMessage(content="""You are a Hypothesis testing expert.261Return ONLY runnable Python code — no markdown, no backticks."""),262 HumanMessage(content=f"""263Write Hypothesis property tests for this code:264TASK: {state['task']}265CODE:266{code}267 268Rules:269- Copy function definitions inline270- Use: from hypothesis import given, settings, strategies as st271- DO NOT use unittest or sys anywhere272- Call test functions directly at the bottom273- Keep to 2 simple property tests only274 275Return ONLY complete runnable Python code:276""")277 ])278 279 hyp_code = response.content280 hyp_code = re.sub(r"```python", "", hyp_code)281 hyp_code = re.sub(r"```", "", hyp_code)282 hyp_code = hyp_code.strip()283 284 result = subprocess.run(285 ["python", "-c", hyp_code],286 capture_output=True, text=True, timeout=30287 )288 289 if result.returncode == 0:290 print("✅ Hypothesis passed!")291 hypothesis_result = "✅ Property-based tests passed with random inputs"292 else:293 err = result.stderr[:200]294 print(f"⚠️ Hypothesis edge case: {err[:80]}")295 hypothesis_result = f"⚠️ Edge case found: {err}"296 297 except subprocess.TimeoutExpired:298 hypothesis_result = "⚠️ Timed out — possible infinite loop on edge input"299 except Exception as e:300 hypothesis_result = f"⚠️ Error: {str(e)[:100]}"301 302 return {"hypothesis_result": hypothesis_result}303 304# ─────────────────────────────────────────305# NODE 7 — PERFORMANCE BENCHMARKER306# ─────────────────────────────────────────307def performance_benchmarker(state: State):308 print("\n⚡ Benchmarking performance...")309 code = state["fixed_code"] if state["fixed_code"] else state["code"]310 311 # Find the target function via ast in-process — no need to re-parse312 # inside the subprocess, so no string-embedding/quote-stripping needed.313 try:314 tree = ast.parse(code)315 fn_names = [n.name for n in ast.walk(tree)316 if isinstance(n, ast.FunctionDef) and not n.name.startswith("_")]317 except SyntaxError:318 fn_names = []319 320 if not fn_names:321 print("⚡ No benchmarkable function found — skipped")322 return {"benchmark_ms": 0.0}323 324 fn_name = fn_names[0]325 tmp_path = None326 try:327 with tempfile.NamedTemporaryFile(328 mode="w", suffix=".py", delete=False, encoding="utf-8"329 ) as tmp:330 tmp.write(code)331 tmp.write("\n\n")332 tmp.write(333 "import timeit as _t\n"334 f"_f = {fn_name}\n"335 "_ran = False\n"336 "for _call in [lambda: _f(100), lambda: _f('hello'), "337 "lambda: _f([1,2,3,4,5]), lambda: _f('racecar'), lambda: _f(10)]:\n"338 " try:\n"339 " _ms = _t.timeit(_call, number=1000) * 1000\n"340 " print('BENCHMARK:' + str(round(_ms, 2)) + 'ms')\n"341 " _ran = True\n"342 " break\n"343 " except Exception:\n"344 " continue\n"345 "if not _ran:\n"346 " print('BENCHMARK:skipped')\n"347 )348 tmp_path = tmp.name349 350 result = subprocess.run(351 ["python", tmp_path],352 capture_output=True, text=True, timeout=20353 )354 output = result.stdout + result.stderr355 match = re.search(r"BENCHMARK:([\d.]+)ms", output)356 if match:357 ms = float(match.group(1))358 print(f"⚡ {ms:.2f}ms per 1000 runs")359 if ms > 5000:360 return {361 "benchmark_ms": ms,362 "error": f"Too slow: {ms:.0f}ms. Optimize algorithm.",363 "passed": False364 }365 return {"benchmark_ms": ms}366 return {"benchmark_ms": 0.0}367 except Exception as e:368 print(f"⚠️ Benchmark error: {e}")369 return {"benchmark_ms": 0.0}370 finally:371 if tmp_path and os.path.exists(tmp_path):372 os.remove(tmp_path)373 374# ─────────────────────────────────────────375# NODE 8 — DEBUGGER376# ─────────────────────────────────────────377def debugger(state: State):378 print(f"\n🔧 Debugger fixing (attempt {state['retries']+1})...")379 380 # Fix on top of the most recent attempt (fixed_code), not the stale381 # original — otherwise each retry throws away the previous retry's work.382 base_code = state["fixed_code"] if state["fixed_code"] else state["code"]383 384 response = llm.invoke([385 SystemMessage(content="""You are a Python debugger.386Fix the exact error. Return ONLY fixed code — no markdown, no backticks."""),387 HumanMessage(content=f"""388CODE:389{base_code}390 391ERROR:392{state['error']}393 394Return complete fixed Python code only:395""")396 ])397 398 fixed = response.content399 fixed = re.sub(r"```python", "", fixed)400 fixed = re.sub(r"```", "", fixed)401 fixed = fixed.strip()402 403 try:404 stable_id = hashlib.md5(state["error"].encode()).hexdigest()[:8]405 memory_collection.add(406 documents=[f"BUG: {state['error']}\nFIX: {fixed}"],407 ids=[f"fix_{state['retries']}_{stable_id}"]408 )409 print("🧠 Stored in memory!")410 except Exception:411 pass412 413 return {"fixed_code": fixed, "retries": state["retries"] + 1}414 415# ─────────────────────────────────────────416# NODE 9 — SECURITY AUDITOR417# ─────────────────────────────────────────418def security_auditor(state: State):419 print("\n🔒 Security check...")420 # final_code isn't set until the reviewer node runs, which is *after*421 # this node in the graph — referencing it here was always a no-op.422 code = state["fixed_code"] if state["fixed_code"] else state["code"]423 424 dangerous = [425 ("eval(", "Code execution via eval"),426 ("exec(", "Code execution via exec"),427 ("os.system(", "Shell injection risk"),428 ("__import__(", "Dynamic import risk"),429 ("pickle.loads(","Deserialization attack"),430 ("password =", "Hardcoded credential"),431 ("api_key =", "Hardcoded API key"),432 ]433 434 found = [reason for pattern, reason in dangerous if pattern.lower() in code.lower()]435 436 if found:437 print(f"❌ Security issues: {found}")438 return {439 "is_secure": False,440 "error": f"Security issues: {found}",441 "security_retries": state["security_retries"] + 1442 }443 444 print("✅ Security passed!")445 return {"is_secure": True}446 447# ─────────────────────────────────────────448# NODE 10 — COMPLEXITY JUDGE449# ─────────────────────────────────────────450def complexity_judge(state: State):451 print("\n📊 Complexity check...")452 code = state["fixed_code"] if state["fixed_code"] else state["code"]453 lines = code.split("\n")454 issues = []455 456 if len(lines) > 60:457 issues.append(f"Too long: {len(lines)} lines")458 459 max_indent = max(460 (len(l) - len(l.lstrip()) for l in lines if l.strip()), default=0461 )462 if max_indent > 16:463 issues.append("Too deeply nested")464 465 try:466 response = llm.invoke([467 HumanMessage(f"Rate complexity 1-10:\n{code}\nReply ONLY a number 1-10.")468 ])469 score = int(re.search(r'\d+', response.content.strip()).group())470 except Exception:471 score = 5472 473 print(f"Complexity: {score}/10")474 475 if score > 7 or issues:476 print(f"❌ Too complex: {issues}")477 return {478 "is_simple": False,479 "error": f"Too complex (score {score}/10). Simplify.",480 "complexity_retries": state["complexity_retries"] + 1481 }482 483 print("✅ Complexity passed!")484 return {"is_simple": True}485 486# ─────────────────────────────────────────487# NODE 11 — SELF REFLECTION488# ─────────────────────────────────────────489def self_reflection(state: State):490 print("\n🪞 Self Reflection...")491 code = state["fixed_code"] if state["fixed_code"] else state["code"]492 493 response = llm.invoke([494 SystemMessage(content="""You are a senior Python engineer.495Reply in EXACTLY this format:496CONFIDENCE: <1-10>497APPROVED: <YES or NO>498ISSUES: <list or NONE>499NOTES: <one sentence>"""),500 HumanMessage(content=f"Review this code:\nTASK: {state['task']}\nCODE:\n{code}")501 ])502 503 reflection = response.content.strip()504 lines_map = {}505 for line in reflection.splitlines():506 if ":" in line:507 key, _, val = line.partition(":")508 lines_map[key.strip().upper()] = val.strip()509 510 try:511 confidence = int(re.search(r'\d+', lines_map.get("CONFIDENCE", "7")).group())512 except Exception:513 confidence = 7514 515 try:516 approved = "YES" in lines_map.get("APPROVED", "YES").upper()517 except Exception:518 approved = True519 520 issues_text = lines_map.get("ISSUES", "NONE")521 notes = lines_map.get("NOTES", "Looks good")522 has_issues = issues_text.upper() not in ("NONE", "") and bool(issues_text.strip())523 524 if not approved or (has_issues and confidence < 7):525 print(f"❌ Reflection: confidence {confidence}/10")526 return {527 "reflection_ok": False,528 "reflection_notes": f"Issues: {issues_text}. {notes}",529 "confidence_score": confidence,530 "error": f"Reflection failed ({confidence}/10): {issues_text}",531 "reflection_retries": state.get("reflection_retries", 0) + 1532 }533 534 print(f"✅ Reflection approved ({confidence}/10)")535 return {536 "reflection_ok": True,537 "reflection_notes": notes,538 "confidence_score": confidence539 }540 541# ─────────────────────────────────────────542# NODE 12 — REVIEWER543# ─────────────────────────────────────────544def reviewer(state: State):545 print("\n✨ Reviewer polishing + explaining...")546 code = state["fixed_code"] if state["fixed_code"] else state["code"]547 548 response = llm.invoke([549 SystemMessage(content="""You are a senior Python developer and teacher.550Do TWO things and return in EXACTLY this format:551 552FINAL_CODE:553<complete polished code with docstrings and type hints>554 555EXPLANATION:556<simple explanation covering: what it does, how it works, time complexity, example usage>557"""),558 HumanMessage(content=f"Polish this code and explain it:\n{code}")559 ])560 561 content = response.content562 final_code = ""563 explanation= ""564 565 if "FINAL_CODE:" in content and "EXPLANATION:" in content:566 parts = content.split("EXPLANATION:")567 code_part = parts[0].replace("FINAL_CODE:", "").strip()568 code_part = re.sub(r"```python", "", code_part)569 code_part = re.sub(r"```", "", code_part)570 final_code = code_part.strip()571 explanation = parts[1].strip()572 else:573 final_code = code574 explanation = content.strip()575 576 if not explanation:577 explanation = "Code completed successfully. See final code above."578 579 return {580 "final_code": final_code,581 "explanation": explanation,582 "review": "Polished and explained"583 }584# ─────────────────────────────────────────585# NODE 13 — EXPLAINER (passthrough)586# ─────────────────────────────────────────587def explainer(state: State):588 explanation = state.get("explanation")589 if not explanation:590 return {"explanation": "Code completed successfully. See final code above."}591 592 # LangGraph requires a state update. 593 # Re-writing the existing explanation satisfies this rule.594 return {"explanation": explanation}