Spaces:
Running
Running
File size: 5,664 Bytes
0eebcd6 f799160 0eebcd6 f799160 0eebcd6 f799160 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | # main.py β Graph builder for Autonomous Python Coding Agent
import os
try:
from dotenv import load_dotenv
load_dotenv() # reads .env in the current working directory
except ImportError:
raise RuntimeError(
"python-dotenv is not installed but a .env file is expected. "
"Run `pip install python-dotenv` then try again."
)
_groq_key = os.environ.get("GROQ_API_KEY")
if not _groq_key:
raise RuntimeError(
"GROQ_API_KEY is not set. Add GROQ_API_KEY=your_key to your .env file "
"(in the same folder as main.py), or `export GROQ_API_KEY=your_key` "
"before starting the app."
)
os.environ["GROQ_API_KEY"] = _groq_key
from langgraph.graph import StateGraph, END
from state import State
from nodes import (
planner, coder, ast_validator, test_generator,
tester, hypothesis_tester, performance_benchmarker,
debugger, security_auditor, complexity_judge,
self_reflection, reviewer, explainer
)
from edges import (
route_after_ast, route_after_test, route_after_hypothesis,
route_after_benchmark, route_after_security,
route_after_complexity, route_after_reflection
)
# ββ BUILD GRAPH βββββββββββββββββββββββββββ
def build_graph():
builder = StateGraph(State)
# Add all 13 nodes
builder.add_node("planner", planner)
builder.add_node("coder", coder)
builder.add_node("ast_validator", ast_validator)
builder.add_node("test_generator", test_generator)
builder.add_node("tester", tester)
builder.add_node("hypothesis", hypothesis_tester)
builder.add_node("benchmark", performance_benchmarker)
builder.add_node("debugger", debugger)
builder.add_node("security", security_auditor)
builder.add_node("complexity", complexity_judge)
builder.add_node("reflection", self_reflection)
builder.add_node("reviewer", reviewer)
builder.add_node("explainer", explainer)
# Entry point
builder.set_entry_point("planner")
# Fixed edges
builder.add_edge("planner", "coder")
builder.add_edge("coder", "ast_validator")
builder.add_edge("test_generator", "tester")
builder.add_edge("debugger", "ast_validator")
builder.add_edge("reviewer", "explainer")
builder.add_edge("explainer", END)
# Conditional edges
builder.add_conditional_edges("ast_validator", route_after_ast,
{"test_generator": "test_generator", "debugger": "debugger", "__end__": END})
builder.add_conditional_edges("tester", route_after_test,
{"hypothesis": "hypothesis", "debugger": "debugger"})
builder.add_conditional_edges("hypothesis", route_after_hypothesis,
{"benchmark": "benchmark"})
builder.add_conditional_edges("benchmark", route_after_benchmark,
{"security": "security", "debugger": "debugger"})
builder.add_conditional_edges("security", route_after_security,
{"complexity": "complexity", "coder": "coder"})
builder.add_conditional_edges("complexity", route_after_complexity,
{"reflection": "reflection", "coder": "coder"})
builder.add_conditional_edges("reflection", route_after_reflection,
{"reviewer": "reviewer", "coder": "coder"})
return builder.compile()
# ββ COMPILED GRAPH ββββββββββββββββββββββββ
graph = build_graph()
# ββ INITIAL STATE βββββββββββββββββββββββββ
def get_initial_state(task: str) -> dict:
return {
"task": task,
"plan": "",
"code": "",
"test_result": "",
"error": "",
"fixed_code": "",
"explanation": "",
"review": "",
"final_code": "",
"retries": 0,
"security_retries": 0,
"complexity_retries": 0,
"reflection_retries": 0,
"passed": False,
"is_secure": False,
"is_simple": False,
"ast_valid": False,
"generated_tests": "",
"hypothesis_result": "",
"benchmark_ms": 0.0,
"reflection_ok": False,
"reflection_notes": "",
"confidence_score": 0,
}
# ββ RUN βββββββββββββββββββββββββββββββββββ
if __name__ == "__main__":
tasks = [
"Write a Python function to find all prime numbers up to n",
"Write a Python function to check if a string is a palindrome",
]
for task in tasks:
print(f"\n{'='*60}")
print(f"π Task: {task}")
print("="*60)
result = graph.invoke(get_initial_state(task), {"recursion_limit": 50})
print(f"\n{'='*60}")
print(f"π» Final Code:\n{result['final_code'] or result['code']}")
print(f"\nπ Explanation:\n{result['explanation']}")
bms = result['benchmark_ms']
conf = result['confidence_score']
print(f"\nπ§ͺ Tests: {result['test_result'][:100]}")
print(f"π² Hypothesis: {result['hypothesis_result']}")
print(f"β‘ Speed: {bms:.1f}ms" if bms > 0 else "β‘ Speed: Skipped")
print(f"πͺ Confidence: {conf}/10" if conf > 0 else "πͺ Confidence: 7/10 (default)")
print(f"π Secure: {result['is_secure']}")
print(f"π Simple: {result['is_simple']}")
print(f"π Retries: {result['retries']}") |