File size: 5,143 Bytes
0eebcd6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
# main.py β€” Graph builder for Autonomous Python Coding Agent

import os
os.environ["GROQ_API_KEY"] = os.environ.get("GROQ_API_KEY", "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")

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