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']}")