diff --git a/agent-framework/prometheus_swarm/utils/duplicate_evidence.py b/agent-framework/prometheus_swarm/utils/duplicate_evidence.py new file mode 100644 index 00000000..8fe8f5d4 --- /dev/null +++ b/agent-framework/prometheus_swarm/utils/duplicate_evidence.py @@ -0,0 +1,56 @@ +import logging +from typing import Any, Dict, List + +class DuplicateEvidenceError(Exception): + """Custom exception for handling duplicate evidence scenarios.""" + def __init__(self, message: str, existing_evidence: List[Dict[str, Any]] = None): + """ + Initialize DuplicateEvidenceError with detailed information. + + Args: + message (str): Description of the duplicate evidence error + existing_evidence (List[Dict[str, Any]], optional): List of existing evidence entries + """ + self.message = message + self.existing_evidence = existing_evidence or [] + super().__init__(self.message) + +def log_duplicate_evidence(evidence: Dict[str, Any], existing_entries: List[Dict[str, Any]]) -> None: + """ + Log details about duplicate evidence. + + Args: + evidence (Dict[str, Any]): The evidence being checked + existing_entries (List[Dict[str, Any]]): List of existing evidence entries + """ + logger = logging.getLogger(__name__) + logger.warning( + "Duplicate evidence detected. " + f"Current Evidence: {evidence}, " + f"Existing Entries Count: {len(existing_entries)}, " + f"Existing Entries IDs: {[entry.get('id') for entry in existing_entries]}" + ) + +def check_duplicate_evidence(new_evidence: Dict[str, Any], existing_evidence: List[Dict[str, Any]], unique_key: str = 'id') -> None: + """ + Check for duplicate evidence and raise an error if duplicates are found. + + Args: + new_evidence (Dict[str, Any]): The new evidence to be added + existing_evidence (List[Dict[str, Any]]): List of existing evidence + unique_key (str, optional): Key to use for identifying duplicates. Defaults to 'id'. + + Raises: + DuplicateEvidenceError: If duplicate evidence is found + """ + duplicates = [ + entry for entry in existing_evidence + if entry.get(unique_key) == new_evidence.get(unique_key) + ] + + if duplicates: + log_duplicate_evidence(new_evidence, duplicates) + raise DuplicateEvidenceError( + f"Evidence with {unique_key} '{new_evidence.get(unique_key)}' already exists", + existing_evidence=duplicates + ) \ No newline at end of file diff --git a/agent-framework/tests/unit/test_duplicate_evidence.py b/agent-framework/tests/unit/test_duplicate_evidence.py new file mode 100644 index 00000000..e3c57a9d --- /dev/null +++ b/agent-framework/tests/unit/test_duplicate_evidence.py @@ -0,0 +1,69 @@ +import pytest +import logging +from prometheus_swarm.utils.duplicate_evidence import ( + DuplicateEvidenceError, + check_duplicate_evidence, + log_duplicate_evidence +) + +def test_duplicate_evidence_error(): + """Test the DuplicateEvidenceError is raised correctly.""" + existing_evidence = [ + {'id': 1, 'name': 'Evidence 1'}, + {'id': 2, 'name': 'Evidence 2'} + ] + new_evidence = {'id': 1, 'name': 'Duplicate Evidence'} + + with pytest.raises(DuplicateEvidenceError) as exc_info: + check_duplicate_evidence(new_evidence, existing_evidence) + + assert str(exc_info.value) == "Evidence with id '1' already exists" + assert len(exc_info.value.existing_evidence) == 1 + +def test_no_duplicate_evidence(): + """Test that no error is raised when no duplicates exist.""" + existing_evidence = [ + {'id': 1, 'name': 'Evidence 1'}, + {'id': 2, 'name': 'Evidence 2'} + ] + new_evidence = {'id': 3, 'name': 'New Evidence'} + + try: + check_duplicate_evidence(new_evidence, existing_evidence) + except DuplicateEvidenceError: + pytest.fail("Unexpected DuplicateEvidenceError raised") + +def test_log_duplicate_evidence(caplog): + """Test logging of duplicate evidence.""" + caplog.set_level(logging.WARNING) + + existing_evidence = [ + {'id': 1, 'name': 'Evidence 1'}, + {'id': 2, 'name': 'Evidence 2'} + ] + new_evidence = {'id': 1, 'name': 'Duplicate Evidence'} + + with pytest.raises(DuplicateEvidenceError): + check_duplicate_evidence(new_evidence, existing_evidence) + + assert len(caplog.records) == 1 + log_record = caplog.records[0] + + assert log_record.levelname == 'WARNING' + assert 'Duplicate evidence detected' in log_record.msg + assert 'Current Evidence: ' + str(new_evidence) in log_record.msg + assert 'Existing Entries Count: 1' in log_record.msg + assert 'Existing Entries IDs: [1]' in log_record.msg + +def test_custom_unique_key(): + """Test duplicate checking with a custom unique key.""" + existing_evidence = [ + {'key': 'abc', 'name': 'Evidence 1'}, + {'key': 'def', 'name': 'Evidence 2'} + ] + new_evidence = {'key': 'abc', 'name': 'Duplicate Evidence'} + + with pytest.raises(DuplicateEvidenceError) as exc_info: + check_duplicate_evidence(new_evidence, existing_evidence, unique_key='key') + + assert str(exc_info.value) == "Evidence with key 'abc' already exists" \ No newline at end of file