From 11930b28c30d2df6eba23eaf51aabfa58770c546 Mon Sep 17 00:00:00 2001 From: pro-utkarshM <118096976+pro-utkarshM@users.noreply.github.com> Date: Wed, 24 Dec 2025 14:34:26 -0500 Subject: [PATCH] feat: Implement Continuous Learning Engine --- nodes/generation_nodes.py | 59 +++++++++++++++++++++++++++++++++++++++ online_flow.py | 22 +++++++-------- 2 files changed, 69 insertions(+), 12 deletions(-) diff --git a/nodes/generation_nodes.py b/nodes/generation_nodes.py index a319762..7936c28 100644 --- a/nodes/generation_nodes.py +++ b/nodes/generation_nodes.py @@ -242,6 +242,65 @@ async def post_async(self, shared, prep_res, exec_res): +class LearningEngineNode(AsyncNode): + """ + The Continuous Learning Engine that updates the strategy based on the outcome + of the test case execution (success, failure, healing success, escalation). + + For now, this is a placeholder for a more complex learning mechanism. + """ + async def prep_async(self, shared): + test_case = shared.get("current_test_case") + test_result = shared.get("current_test_case_result") + + if not test_case or not test_result: + return None + + # Extract relevant data for learning + learning_data = { + "test_id": test_case.get("id"), + "final_status": test_case.get("status"), + "initial_generation_strategy": shared.get("generation_context", {}).get("strategy", {}).get("name"), + "initial_generation_confidence": shared.get("generation_context", {}).get("strategy", {}).get("confidence_target"), + "test_passed": test_result.get("passed"), + "healing_attempted": shared.get("generated_patch") is not None, + "escalated": test_case.get("status") == "ESCALATED", + "error_category": test_case.get("escalation_reason") # Re-using escalation reason for error categorization + } + + return learning_data + + async def exec_async(self, prep_res): + if prep_res is None: + return {"status": "SKIPPED", "reason": "No test case data for learning."} + + print(f"\n--- LEARNING ENGINE: Processing outcome for {prep_res['test_id']} ---") + + # Placeholder for complex learning logic (e.g., updating weights, refining RAG query, adjusting confidence) + # For this implementation, we just log the outcome and simulate a strategy update. + + if prep_res["test_passed"]: + print(f"Outcome: SUCCESS. Strategy '{prep_res['initial_generation_strategy']}' was effective.") + # In a real system: reinforce the strategy + elif prep_res["escalated"]: + print(f"Outcome: ESCALATED. Strategy '{prep_res['initial_generation_strategy']}' failed to resolve issue: {prep_res['error_category']}.") + # In a real system: penalize the strategy, or refine the planning for this type of module + elif prep_res["healing_attempted"]: + print(f"Outcome: HEALING FAILED. Patch did not fix the issue.") + # In a real system: refine the healing prompt/logic + else: + print(f"Outcome: INITIAL FAILURE. Strategy '{prep_res['initial_generation_strategy']}' failed immediately.") + # In a real system: penalize the strategy + + # Simulate a strategy update for the next run (e.g., update a persistent config file) + # For now, this is a no-op, but the structure is in place. + + return {"status": "LEARNING_COMPLETE"} + + async def post_async(self, shared, prep_res, exec_res): + print(f"Learning Engine finished with status: {exec_res['status']}") + return "success" + class HealNode(AsyncNode): """ An autonomous agent that attempts to fix failing tests by generating a patch. diff --git a/online_flow.py b/online_flow.py index 94eff1f..009d12a 100644 --- a/online_flow.py +++ b/online_flow.py @@ -1,5 +1,5 @@ from pocketflow import AsyncFlow, Node -from nodes.generation_nodes import PlanTestsNode, GenerateSingleTestNode, HealNode, MultiToolRouterNode +from nodes.generation_nodes import PlanTestsNode, GenerateSingleTestNode, HealNode, MultiToolRouterNode, LearningEngineNode from nodes.verification_nodes import VerifySingleTestNode, QualityGateNode, HumanInTheLoopNode import json @@ -48,10 +48,9 @@ def create_planning_flow(): def create_single_test_execution_flow(): """ - --- PHASE 4 HUMAN-IN-THE-LOOP FLOW --- - Creates the workflow for generating, validating, verifying, healing, and escalating a single test case. - The flow is: Route -> Generate -> QualityGate -> Verify (Success) -> End - or: Route -> Generate -> QualityGate -> Verify (Failure) -> Heal -> Verify (Failure) -> HumanInTheLoop -> End + --- PHASE 5 LEARNING FLOW --- + Creates the workflow for generating, validating, verifying, healing, escalating, and learning from a single test case. + The flow is: Route -> Generate -> QualityGate -> Verify -> [Heal] -> Verify -> [Escalate] -> Learn """ router_node = MultiToolRouterNode() generate_node = GenerateSingleTestNode() @@ -60,6 +59,7 @@ def create_single_test_execution_flow(): heal_node = HealNode() verify_node_healed = VerifySingleTestNode(name="verify_healed") human_in_the_loop_node = HumanInTheLoopNode() + learning_node = LearningEngineNode() # 1. Route, Generate, and Validate the test router_node >> generate_node >> quality_gate_node >> verify_node_initial @@ -70,17 +70,15 @@ def create_single_test_execution_flow(): # 3. If heal is successful, re-verify the test heal_node.on_success >> verify_node_healed - # 4. If re-verification fails, escalate to human + # 4. All failure paths lead to escalation verify_node_healed.on_failure >> human_in_the_loop_node - - # 5. If Quality Gate fails, escalate to human quality_gate_node.on_failure >> human_in_the_loop_node - - # 6. If heal is unsuccessful, escalate to human heal_node.on_failure >> human_in_the_loop_node - # 7. If initial verification succeeds, the flow ends (success is handled by main.py) - # 8. If re-verification succeeds, the flow ends (success is handled by main.py) + # 5. All final nodes (success or escalation) lead to the learning engine + verify_node_initial.on_success >> learning_node + verify_node_healed.on_success >> learning_node + human_in_the_loop_node.on_success >> learning_node execution_flow = AsyncFlow(start=router_node) return execution_flow \ No newline at end of file