From db2b52b4a699a07d820a493d133e429480554d63 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <209825114+claude[bot]@users.noreply.github.com> Date: Mon, 14 Jul 2025 01:05:39 +0000 Subject: [PATCH 1/2] fix: Support GuardrailResult return type in guardrail functions (#875) - Modified task.py to accept both GuardrailResult and Tuple[bool, Any] return types - Updated _process_guardrail to handle GuardrailResult directly without conversion - Added comprehensive test cases for both return types - Updated documentation with GuardrailResult examples - Maintains backward compatibility with existing Tuple[bool, Any] guardrails Co-authored-by: Mervin Praison --- examples/guardrail_example_fixed.py | 142 +++++++++++++++ src/praisonai-agents/CLAUDE.md | 50 +++++- .../praisonaiagents/task/task.py | 46 +++-- .../test_guardrail_return_types.py | 163 ++++++++++++++++++ 4 files changed, 382 insertions(+), 19 deletions(-) create mode 100644 examples/guardrail_example_fixed.py create mode 100644 src/praisonai-agents/test_guardrail_return_types.py diff --git a/examples/guardrail_example_fixed.py b/examples/guardrail_example_fixed.py new file mode 100644 index 000000000..c7f2ab4e7 --- /dev/null +++ b/examples/guardrail_example_fixed.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +""" +Fixed example demonstrating proper guardrail usage with PraisonAI Agents. +This addresses the issues reported in issue #875. +""" + +from praisonaiagents import Agent, Task, GuardrailResult, PraisonAIAgents +from typing import Tuple, Any +import trafilatura + +# Example 1: Using GuardrailResult return type (now supported!) +def validate_length_guardrailresult(output) -> GuardrailResult: + """Ensure output is between 100-500 characters using GuardrailResult""" + # Extract the raw text from the TaskOutput object + text = output.raw if hasattr(output, 'raw') else str(output) + length = len(text) + + if 100 <= length <= 500: + return GuardrailResult( + success=True, + result=output, # Pass through the original output + error="" + ) + else: + return GuardrailResult( + success=False, + result=None, + error=f"Output must be 100-500 chars, got {length}" + ) + +# Example 2: Using Tuple[bool, Any] return type (original method) +def validate_length_tuple(output) -> Tuple[bool, Any]: + """Ensure output is between 100-500 characters using tuple""" + text = output.raw if hasattr(output, 'raw') else str(output) + length = len(text) + + if 100 <= length <= 500: + return True, output + else: + return False, f"Output must be 100-500 chars, got {length}" + +# Tool function +def get_url_context(url): + """Fetch and extract content from a URL""" + downloaded = trafilatura.fetch_url(url) + if not downloaded: + return "Sorry, I couldn't fetch the content from that URL." + + extracted = trafilatura.extract( + downloaded, + include_comments=False, + include_links=True, + output_format='json', + with_metadata=True, + url=url + ) + + if not extracted: + return "Sorry, I couldn't extract readable content from that page." + + return extracted # returns JSON string + +# Create agent with FIXED tools parameter (must be a list!) +agent = Agent( + name="Content Summarizer", + role="Content Analysis Expert", + goal="Summarize web content concisely", + instructions="You are a helpful assistant that summarizes web content", + llm="gemini/gemini-2.5-flash-lite-preview-06-17", + self_reflect=False, + verbose=True, + tools=[get_url_context] # FIX: tools must be a list, not a single function +) + +# Create task with GuardrailResult guardrail +task_with_guardrailresult = Task( + name="summarise article with GuardrailResult", + description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary below 500 characters", + agent=agent, + guardrail=validate_length_guardrailresult, # Using GuardrailResult + expected_output="summary of the article below 500 characters", + max_retries=3 # Will retry up to 3 times if guardrail fails +) + +# Alternative: Create task with tuple guardrail +task_with_tuple = Task( + name="summarise article with tuple", + description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary below 500 characters", + agent=agent, + guardrail=validate_length_tuple, # Using Tuple[bool, Any] + expected_output="summary of the article below 500 characters", + max_retries=3 +) + +# Example with string-based LLM guardrail +task_with_llm_guardrail = Task( + name="summarise with LLM guardrail", + description="get the context of this url: https://blog.google/technology/ai/dolphingemma/ and produce a summary", + agent=agent, + guardrail="Ensure the summary is professional, factual, and between 100-500 characters", + expected_output="professional summary of the article" +) + +# Run with GuardrailResult example +print("=== Running with GuardrailResult guardrail ===") +agents_gr = PraisonAIAgents( + agents=[agent], + tasks=[task_with_guardrailresult] +) + +# Uncomment to run: +# result_gr = agents_gr.start() + +# Run with Tuple example +print("\n=== Running with Tuple[bool, Any] guardrail ===") +agents_tuple = PraisonAIAgents( + agents=[agent], + tasks=[task_with_tuple] +) + +# Uncomment to run: +# result_tuple = agents_tuple.start() + +# Run with LLM guardrail example +print("\n=== Running with LLM-based guardrail ===") +agents_llm = PraisonAIAgents( + agents=[agent], + tasks=[task_with_llm_guardrail] +) + +# Uncomment to run: +# result_llm = agents_llm.start() + +print(""" +Key fixes applied: +1. GuardrailResult is now accepted as a valid return type annotation +2. tools parameter must be a list: tools=[get_url_context] not tools=get_url_context +3. Both GuardrailResult and Tuple[bool, Any] return types are supported +4. String-based LLM guardrails are also supported + +The guardrail will automatically retry (up to max_retries times) if validation fails. +""") \ No newline at end of file diff --git a/src/praisonai-agents/CLAUDE.md b/src/praisonai-agents/CLAUDE.md index e48f7237b..8ce9c4a70 100644 --- a/src/praisonai-agents/CLAUDE.md +++ b/src/praisonai-agents/CLAUDE.md @@ -139,10 +139,32 @@ task = Task( #### Task-Level Guardrails ```python from typing import Tuple, Any +from praisonaiagents import GuardrailResult -# Function-based guardrail -def validate_output(task_output: TaskOutput) -> Tuple[bool, Any]: - """Custom validation function.""" +# Function-based guardrail (Option 1: Using GuardrailResult) +def validate_output(task_output: TaskOutput) -> GuardrailResult: + """Custom validation function returning GuardrailResult.""" + if "error" in task_output.raw.lower(): + return GuardrailResult( + success=False, + result=None, + error="Output contains errors" + ) + if len(task_output.raw) < 10: + return GuardrailResult( + success=False, + result=None, + error="Output is too short" + ) + return GuardrailResult( + success=True, + result=task_output, + error="" + ) + +# Function-based guardrail (Option 2: Using Tuple[bool, Any]) +def validate_output_tuple(task_output: TaskOutput) -> Tuple[bool, Any]: + """Custom validation function returning tuple.""" if "error" in task_output.raw.lower(): return False, "Output contains errors" if len(task_output.raw) < 10: @@ -170,7 +192,27 @@ task = Task( #### Agent-Level Guardrails ```python # Agent guardrails apply to ALL outputs from that agent -def validate_professional_tone(task_output: TaskOutput) -> Tuple[bool, Any]: + +# Option 1: Using GuardrailResult +def validate_professional_tone(task_output: TaskOutput) -> GuardrailResult: + """Ensure professional tone in all agent responses.""" + content = task_output.raw.lower() + casual_words = ['yo', 'dude', 'awesome', 'cool'] + for word in casual_words: + if word in content: + return GuardrailResult( + success=False, + result=None, + error=f"Unprofessional language detected: {word}" + ) + return GuardrailResult( + success=True, + result=task_output, + error="" + ) + +# Option 2: Using Tuple[bool, Any] +def validate_professional_tone_tuple(task_output: TaskOutput) -> Tuple[bool, Any]: """Ensure professional tone in all agent responses.""" content = task_output.raw.lower() casual_words = ['yo', 'dude', 'awesome', 'cool'] diff --git a/src/praisonai-agents/praisonaiagents/task/task.py b/src/praisonai-agents/praisonaiagents/task/task.py index db0a79274..5e42ea677 100644 --- a/src/praisonai-agents/praisonaiagents/task/task.py +++ b/src/praisonai-agents/praisonaiagents/task/task.py @@ -172,21 +172,33 @@ def _setup_guardrail(self): # Check return annotation if present return_annotation = sig.return_annotation if return_annotation != inspect.Signature.empty: - return_annotation_args = get_args(return_annotation) - if not ( - get_origin(return_annotation) is tuple - and len(return_annotation_args) == 2 - and return_annotation_args[0] is bool - and ( - return_annotation_args[1] is Any - or return_annotation_args[1] is str - or return_annotation_args[1] is TaskOutput - or return_annotation_args[1] == Union[str, TaskOutput] - ) + # Import GuardrailResult for checking + from ..guardrails import GuardrailResult + + # Check if it's a GuardrailResult type + if return_annotation is GuardrailResult or ( + hasattr(return_annotation, '__name__') and + return_annotation.__name__ == 'GuardrailResult' ): - raise ValueError( - "If return type is annotated, it must be Tuple[bool, Any]" - ) + # Valid GuardrailResult return type + pass + else: + # Check for tuple return type + return_annotation_args = get_args(return_annotation) + if not ( + get_origin(return_annotation) is tuple + and len(return_annotation_args) == 2 + and return_annotation_args[0] is bool + and ( + return_annotation_args[1] is Any + or return_annotation_args[1] is str + or return_annotation_args[1] is TaskOutput + or return_annotation_args[1] == Union[str, TaskOutput] + ) + ): + raise ValueError( + "If return type is annotated, it must be GuardrailResult or Tuple[bool, Any]" + ) self._guardrail_fn = self.guardrail elif isinstance(self.guardrail, str): @@ -447,7 +459,11 @@ def _process_guardrail(self, task_output: TaskOutput): # Call the guardrail function result = self._guardrail_fn(task_output) - # Convert the result to a GuardrailResult + # Check if result is already a GuardrailResult + if isinstance(result, GuardrailResult): + return result + + # Otherwise, convert the tuple result to a GuardrailResult return GuardrailResult.from_tuple(result) except Exception as e: diff --git a/src/praisonai-agents/test_guardrail_return_types.py b/src/praisonai-agents/test_guardrail_return_types.py new file mode 100644 index 000000000..094f6d289 --- /dev/null +++ b/src/praisonai-agents/test_guardrail_return_types.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +""" +Test different guardrail return types to ensure both GuardrailResult +and Tuple[bool, Any] work correctly. This addresses issue #875. +""" + +import sys +import os +from typing import Tuple, Any + +from praisonaiagents import Agent, Task, TaskOutput, GuardrailResult +import inspect + + +def test_guardrail_return_types(): + """Test that both GuardrailResult and Tuple[bool, Any] return types are accepted.""" + print("Testing guardrail return type validation...") + + # Test 1: GuardrailResult return type + def validate_with_guardrailresult(task_output: TaskOutput) -> GuardrailResult: + """Validation function returning GuardrailResult.""" + if len(task_output.raw) < 10: + return GuardrailResult( + success=False, + result=None, + error="Output is too short" + ) + return GuardrailResult( + success=True, + result=task_output, + error="" + ) + + # Test 2: Tuple[bool, Any] return type + def validate_with_tuple(task_output: TaskOutput) -> Tuple[bool, Any]: + """Validation function returning tuple.""" + if len(task_output.raw) < 10: + return False, "Output is too short" + return True, task_output + + # Test 3: No return type annotation + def validate_no_annotation(task_output: TaskOutput): + """Validation function without return type annotation.""" + if len(task_output.raw) < 10: + return False, "Output is too short" + return True, task_output + + # Create test agent + agent = Agent( + name="Test Agent", + role="Tester", + goal="Test guardrails", + backstory="Testing guardrail functionality", + llm="gpt-4o" + ) + + # Test creating tasks with each guardrail type + tests = [ + ("GuardrailResult return type", validate_with_guardrailresult), + ("Tuple[bool, Any] return type", validate_with_tuple), + ("No return type annotation", validate_no_annotation) + ] + + for test_name, guardrail_func in tests: + print(f"\nTesting {test_name}...") + try: + task = Task( + description="Test task", + expected_output="Test output", + agent=agent, + guardrail=guardrail_func + ) + print(f"✓ Task created successfully with {test_name}") + + # Verify guardrail was set up + assert task._guardrail_fn is not None, "Guardrail function not set" + + # Test guardrail processing + test_output = TaskOutput( + description="Test", + raw="This is a test output that is long enough", + agent="Test Agent" + ) + + result = task._process_guardrail(test_output) + assert isinstance(result, GuardrailResult), f"Expected GuardrailResult, got {type(result)}" + assert result.success, "Guardrail validation should pass" + print(f"✓ Guardrail processing works with {test_name}") + + except Exception as e: + print(f"✗ Failed with {test_name}: {e}") + return False + + print("\n✓ All guardrail return type tests passed!") + return True + + +def test_guardrail_result_handling(): + """Test that _process_guardrail correctly handles both return types.""" + print("\nTesting guardrail result handling...") + + # Create test agent and task + agent = Agent( + name="Test Agent", + role="Tester", + goal="Test guardrails", + backstory="Testing guardrail functionality", + llm="gpt-4o" + ) + + test_output = TaskOutput( + description="Test", + raw="Test output", + agent="Test Agent" + ) + + # Test 1: Function returning GuardrailResult + def return_guardrailresult(output: TaskOutput) -> GuardrailResult: + return GuardrailResult(success=True, result=output, error="") + + task1 = Task( + description="Test", + agent=agent, + guardrail=return_guardrailresult + ) + + result1 = task1._process_guardrail(test_output) + assert isinstance(result1, GuardrailResult), "Should return GuardrailResult" + assert result1.success, "Should be successful" + print("✓ GuardrailResult return handled correctly") + + # Test 2: Function returning tuple + def return_tuple(output: TaskOutput) -> Tuple[bool, Any]: + return True, output + + task2 = Task( + description="Test", + agent=agent, + guardrail=return_tuple + ) + + result2 = task2._process_guardrail(test_output) + assert isinstance(result2, GuardrailResult), "Should convert tuple to GuardrailResult" + assert result2.success, "Should be successful" + print("✓ Tuple return converted correctly") + + print("✓ Guardrail result handling tests passed!") + return True + + +if __name__ == "__main__": + print("Running guardrail return type tests for issue #875...") + + success = True + success &= test_guardrail_return_types() + success &= test_guardrail_result_handling() + + if success: + print("\n✅ All tests passed! The fix for issue #875 is working correctly.") + sys.exit(0) + else: + print("\n❌ Some tests failed!") + sys.exit(1) \ No newline at end of file From 072c3c7d2085095860f81d44fee7c501e8215f14 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <209825114+claude[bot]@users.noreply.github.com> Date: Mon, 14 Jul 2025 06:21:30 +0000 Subject: [PATCH 2/2] fix: Improve guardrail validation and add missing type hints - Fixed critical bug: validation now properly checks for GuardrailResult type instead of accepting any class with the same name - Added missing TaskOutput type hints in example file - Refactored conditional logic for better readability - Maintains full backward compatibility Co-authored-by: Mervin Praison --- examples/guardrail_example_fixed.py | 6 +-- .../praisonaiagents/task/task.py | 43 +++++++++---------- 2 files changed, 23 insertions(+), 26 deletions(-) diff --git a/examples/guardrail_example_fixed.py b/examples/guardrail_example_fixed.py index c7f2ab4e7..9fc0c9b73 100644 --- a/examples/guardrail_example_fixed.py +++ b/examples/guardrail_example_fixed.py @@ -4,12 +4,12 @@ This addresses the issues reported in issue #875. """ -from praisonaiagents import Agent, Task, GuardrailResult, PraisonAIAgents +from praisonaiagents import Agent, Task, GuardrailResult, PraisonAIAgents, TaskOutput from typing import Tuple, Any import trafilatura # Example 1: Using GuardrailResult return type (now supported!) -def validate_length_guardrailresult(output) -> GuardrailResult: +def validate_length_guardrailresult(output: TaskOutput) -> GuardrailResult: """Ensure output is between 100-500 characters using GuardrailResult""" # Extract the raw text from the TaskOutput object text = output.raw if hasattr(output, 'raw') else str(output) @@ -29,7 +29,7 @@ def validate_length_guardrailresult(output) -> GuardrailResult: ) # Example 2: Using Tuple[bool, Any] return type (original method) -def validate_length_tuple(output) -> Tuple[bool, Any]: +def validate_length_tuple(output: TaskOutput) -> Tuple[bool, Any]: """Ensure output is between 100-500 characters using tuple""" text = output.raw if hasattr(output, 'raw') else str(output) length = len(text) diff --git a/src/praisonai-agents/praisonaiagents/task/task.py b/src/praisonai-agents/praisonaiagents/task/task.py index 5e42ea677..38904663a 100644 --- a/src/praisonai-agents/praisonaiagents/task/task.py +++ b/src/praisonai-agents/praisonaiagents/task/task.py @@ -176,29 +176,26 @@ def _setup_guardrail(self): from ..guardrails import GuardrailResult # Check if it's a GuardrailResult type - if return_annotation is GuardrailResult or ( - hasattr(return_annotation, '__name__') and - return_annotation.__name__ == 'GuardrailResult' - ): - # Valid GuardrailResult return type - pass - else: - # Check for tuple return type - return_annotation_args = get_args(return_annotation) - if not ( - get_origin(return_annotation) is tuple - and len(return_annotation_args) == 2 - and return_annotation_args[0] is bool - and ( - return_annotation_args[1] is Any - or return_annotation_args[1] is str - or return_annotation_args[1] is TaskOutput - or return_annotation_args[1] == Union[str, TaskOutput] - ) - ): - raise ValueError( - "If return type is annotated, it must be GuardrailResult or Tuple[bool, Any]" - ) + is_guardrail_result = return_annotation is GuardrailResult + + # Check for tuple return type + return_annotation_args = get_args(return_annotation) + is_tuple = ( + get_origin(return_annotation) is tuple + and len(return_annotation_args) == 2 + and return_annotation_args[0] is bool + and ( + return_annotation_args[1] is Any + or return_annotation_args[1] is str + or return_annotation_args[1] is TaskOutput + or return_annotation_args[1] == Union[str, TaskOutput] + ) + ) + + if not (is_guardrail_result or is_tuple): + raise ValueError( + "If return type is annotated, it must be GuardrailResult or Tuple[bool, Any]" + ) self._guardrail_fn = self.guardrail elif isinstance(self.guardrail, str):