fix(sew): validate refined prompts before adopting them in SEWOptimizer

evoagentx/optimizers/sew_optimizer.py

@@ -842,6 +842,36 @@ def _wfg_structure_optimization_step(self, graph: SequentialWorkFlowGraph) -> Se
         new_graph = graph_scheme.parse_from_scheme(scheme=self.repr_scheme, repr=new_graph_repr)
         return new_graph
 
+    @staticmethod
+    def _validate_refined_prompt(new_prompt: str, original_prompt: str, input_names: List[str]) -> bool:
+        """
+        Validate a refined prompt returned by the prompt breeder before adopting it.
+
+        The LLM may occasionally echo the meta-instructions used to request the
+        refinement (or drop the input placeholders), in which case adopting the
+        output verbatim silently corrupts the workflow node.
+
+        Args:
+            new_prompt (str): The refined prompt returned by the prompt breeder.
+            original_prompt (str): The current prompt of the task/operator.
+            input_names (List[str]): Names of the task inputs. Placeholders (e.g. ``{question}``)
+                that are present in ``original_prompt`` must also be present in ``new_prompt``.
+
+        Returns:
+            bool: True if the refined prompt is safe to adopt, False otherwise.
+        """
+        if not new_prompt or not new_prompt.strip():
+            return False
+        lowered = new_prompt.lower()
+        meta_markers = ("please refine the instruction", "only output the refined instruction")
+        if any(marker in lowered for marker in meta_markers):
+            return False
+        for name in input_names:
+            placeholder = "{" + name + "}"
+            if placeholder in original_prompt and placeholder not in new_prompt:
+                return False
+        return True
+
     def _wfg_prompt_optimization_step(self, graph: SequentialWorkFlowGraph) -> SequentialWorkFlowGraph:
 
         task_description = graph.goal
@@ -853,9 +883,17 @@ def _wfg_prompt_optimization_step(self, graph: SequentialWorkFlowGraph) -> Seque
             optimization_prompt = "Task Description: " + task_description + "\n\nWorkflow Steps:\n" + graph_repr + f"\n\nINSTRUCTION for the {i+1}-th task:\n\"\"\"\n" + original_prompt + "\n\"\"\""
             optimization_prompt += f"\n\nGiven the above information, please refine the instruction for the {i+1}-th task.\n"
             optimization_prompt += r"Note that you should always use bracket (e.g. `{input_name}`) to wrap the inputs of the tasks in your refined instruction.\n"
-            optimization_prompt += "Only output the refined instruction and DON'T include any other text!" 
+            optimization_prompt += "Only output the refined instruction and DON'T include any other text!"
             new_prompt = self._prompt_breeder.generate_prompt(task_description=task_description, prompt=optimization_prompt, order=self.order)
-            graph_info["tasks"][i]["prompt"] = new_prompt
+            input_names = [inp.get("name") for inp in task.get("inputs", []) if inp.get("name")]
+            if self._validate_refined_prompt(new_prompt, original_prompt, input_names):
+                graph_info["tasks"][i]["prompt"] = new_prompt
+            else:
+                logger.warning(
+                    f"Discarding invalid refined prompt for task '{task.get('name', i)}': "
+                    "the response echoes the refinement meta-instructions or drops required input placeholders. "
+                    "Keeping the original prompt."
+                )
         new_graph = SequentialWorkFlowGraph.from_dict(graph_info)
         return new_graph
 
@@ -885,7 +923,13 @@ def _action_graph_prompt_optimization_step(self, graph: ActionGraph) -> ActionGr
             optimization_prompt += "\nOnly output the refined instruction and DON'T include any other text!"
             new_prompt = self._prompt_breeder.generate_prompt(task_description=task_description, prompt=optimization_prompt, order=self.order)
             new_prompt = new_prompt.replace("\"", "").strip()
-            graph_info["operators"][operator_name]["prompt"] = new_prompt
+            if self._validate_refined_prompt(new_prompt, original_prompt, []):
+                graph_info["operators"][operator_name]["prompt"] = new_prompt
+            else:
+                logger.warning(
+                    f"Discarding invalid refined prompt for operator '{operator_name}': "
+                    "the response echoes the refinement meta-instructions. Keeping the original prompt."
+                )
         new_graph = ActionGraph.from_dict(graph_info)
         return new_graph
 

tests/src/optimizers/test_sew_prompt_validation.py

@@ -0,0 +1,92 @@
+import unittest
+
+from evoagentx.models import OpenAILLMConfig, OpenAILLM
+from evoagentx.workflow.workflow_graph import SEWWorkFlowGraph
+from evoagentx.optimizers.sew_optimizer import SEWOptimizer
+
+# A real-world failure case observed with deepseek-chat: the model echoed the
+# refinement meta-instructions verbatim, and the optimizer adopted the echo as
+# the new task prompt, corrupting the workflow (HumanEval pass@1 dropped 0.4 -> 0.1).
+ECHOED_META_PROMPT = (
+    "Given the above information, please refine the instruction for the 1-th task.\n"
+    "Note that you should always use bracket (e.g. `{input_name}`) to wrap the inputs "
+    "of the tasks in your refined instruction.\n"
+    "Only output the refined instruction and DON'T include any other text!"
+)
+
+
+class _StubBreeder:
+    """Prompt breeder stub that returns a fixed response."""
+
+    def __init__(self, response: str):
+        self.response = response
+
+    def generate_prompt(self, **kwargs) -> str:
+        return self.response
+
+
+class _StubOptimizer:
+    """Duck-typed stand-in exposing only what `_wfg_prompt_optimization_step` uses."""
+
+    _validate_refined_prompt = staticmethod(SEWOptimizer._validate_refined_prompt)
+    repr_scheme = "python"
+    order = "zero-order"
+
+    def __init__(self, breeder: _StubBreeder):
+        self._prompt_breeder = breeder
+
+
+class TestSEWPromptValidation(unittest.TestCase):
+
+    def setUp(self):
+        self.model = OpenAILLM(config=OpenAILLMConfig(model="gpt-4o-mini", openai_key="XXX"))
+        self.graph = SEWWorkFlowGraph(llm=self.model)
+
+    def test_validator_rejects_echoed_meta_instructions(self):
+        self.assertFalse(
+            SEWOptimizer._validate_refined_prompt(ECHOED_META_PROMPT, "{question}", ["question"])
+        )
+
+    def test_validator_rejects_empty_prompt(self):
+        self.assertFalse(SEWOptimizer._validate_refined_prompt("", "{question}", ["question"]))
+        self.assertFalse(SEWOptimizer._validate_refined_prompt("   \n", "{question}", ["question"]))
+
+    def test_validator_rejects_dropped_placeholder(self):
+        self.assertFalse(
+            SEWOptimizer._validate_refined_prompt(
+                "Summarize the task in detail.", "{question}", ["question"]
+            )
+        )
+
+    def test_validator_accepts_legitimate_refinement(self):
+        self.assertTrue(
+            SEWOptimizer._validate_refined_prompt(
+                "Carefully parse the coding question below and summarize it.\n\nQuestion: {question}",
+                "{question}",
+                ["question"],
+            )
+        )
+
+    def test_validator_ignores_placeholders_absent_from_original(self):
+        # placeholders that were never in the original prompt must not be required
+        self.assertTrue(
+            SEWOptimizer._validate_refined_prompt("Do the task.", "Do it.", ["question"])
+        )
+
+    def test_step_keeps_original_prompts_on_echoed_response(self):
+        original_prompts = [t["prompt"] for t in self.graph.get_graph_info()["tasks"]]
+        stub = _StubOptimizer(_StubBreeder(ECHOED_META_PROMPT))
+        new_graph = SEWOptimizer._wfg_prompt_optimization_step(stub, self.graph)
+        new_prompts = [t["prompt"] for t in new_graph.get_graph_info()["tasks"]]
+        self.assertEqual(original_prompts, new_prompts)
+
+    def test_step_adopts_valid_refinement(self):
+        refined = "Refined instruction: answer {question} using the summary {parsed_task}."
+        stub = _StubOptimizer(_StubBreeder(refined))
+        new_graph = SEWOptimizer._wfg_prompt_optimization_step(stub, self.graph)
+        new_prompts = [t["prompt"] for t in new_graph.get_graph_info()["tasks"]]
+        self.assertEqual(new_prompts, [refined] * len(new_prompts))
+
+
+if __name__ == "__main__":
+    unittest.main()