File size: 2,691 Bytes
c209999 | 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 | from src.graph.state import SandboxState
from src.agents.coder import generate_code
from src.agents.critic import analyze_error
from src.sandbox.executor import execute_code
def coder_node(state: SandboxState) -> dict:
last_error = None
if state.get("error_analysis"):
last_error = state["error_analysis"]
code = generate_code(state["user_prompt"], previous_error=last_error)
trace_entry = {
"node": "coder",
"retry": state["retry_count"],
"explanation": code.explanation[:200],
"requirements": code.requirements,
}
return {
"script": code.script,
"requirements": code.requirements,
"explanation": code.explanation,
"trace": state.get("trace", []) + [trace_entry],
}
def executor_node(state: SandboxState) -> dict:
result = execute_code(state["script"], state.get("requirements"))
trace_entry = {
"node": "executor",
"retry": state["retry_count"],
"exit_code": result.exit_code,
"success": result.success,
"stdout_preview": result.stdout[:300] if result.stdout else "",
"stderr_preview": result.stderr[:300] if result.stderr else "",
}
updates: dict = {
"sandbox_result": result,
"files": result.files,
"trace": state.get("trace", []) + [trace_entry],
}
if result.success:
updates["final_output"] = result.stdout
else:
updates["final_error"] = result.stderr
return updates
def critic_node(state: SandboxState) -> dict:
critique = analyze_error(
script=state["script"],
stderr=state["sandbox_result"].stderr,
user_prompt=state["user_prompt"],
)
trace_entry = {
"node": "critic",
"retry": state["retry_count"],
"error_analysis": critique.error_analysis[:300],
"correction_strategy": critique.correction_strategy[:300],
"confidence": critique.confidence,
}
prev = state.get("error_analysis") or ""
return {
"error_analysis": prev
+ f"\n--- Attempt {state['retry_count'] + 1} ---\n"
+ critique.error_analysis,
"correction_strategy": critique.correction_strategy,
"retry_count": state["retry_count"] + 1,
"trace": state.get("trace", []) + [trace_entry],
}
def success_node(state: SandboxState) -> dict:
return {
"final_output": state.get("final_output") or state.get("script", ""),
"final_error": None,
}
def fail_node(state: SandboxState) -> dict:
return {
"final_output": None,
"final_error": state.get("final_error", "Unknown error after all retries exhausted."),
}
|