diff --git a/evoagentx/optimizers/sew_optimizer.py b/evoagentx/optimizers/sew_optimizer.py index d910ded8..6e123aaa 100644 --- a/evoagentx/optimizers/sew_optimizer.py +++ b/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 diff --git a/tests/src/optimizers/test_sew_prompt_validation.py b/tests/src/optimizers/test_sew_prompt_validation.py new file mode 100644 index 00000000..09dd5265 --- /dev/null +++ b/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()