mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #326 from Hawksight-AI/utils

Browse Source 
Fix PolicyException Naming Conflicts in Decision Models
This commit is contained in:
Mohd Kaif
2026-02-16 23:54:36 +05:30
committed by GitHub
parent 94448e1e5d dcb4f77efc
commit f704d6ce91
9 changed files with 64 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
semantica/context/causal_analyzer.py
+3 -2
View File
@@ -420,7 +420,7 @@ class CausalChainAnalyzer:
data["timestamp"] = datetime.fromisoformat(data["timestamp"])
return Decision(
decision_id=data.get("decision_id", ""),
decision_id=data["decision_id"], # Required field
category=data.get("category", ""),
scenario=data.get("scenario", ""),
reasoning=data.get("reasoning", ""),
@@ -430,5 +430,6 @@ class CausalChainAnalyzer:
decision_maker=data.get("decision_maker", ""),
reasoning_embedding=data.get("reasoning_embedding"),
node2vec_embedding=data.get("node2vec_embedding"),
metadata=data.get("metadata", {})
metadata=data.get("metadata", {}),
auto_generate_id=False # Don't auto-generate for deserialization
)
semantica/context/decision_models.py
+27 -15
View File
@@ -99,10 +99,12 @@ class Decision:
node2vec_embedding: Optional[List[float]] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
def __post_init__(self, auto_generate_id: bool = True):
"""Validate decision data."""
if self.decision_id is None:
if auto_generate_id and not self.decision_id: # Handle both None and empty string
self.decision_id = str(uuid.uuid4())
elif not self.decision_id and not auto_generate_id:
raise ValueError("decision_id is required when auto_generate_id=False")
if not 0 <= self.confidence <= 1:
raise ValueError("Confidence must be between 0 and 1")
@@ -141,10 +143,12 @@ class DecisionContext:
cross_system_inputs: Dict[str, Any] = field(default_factory=dict)
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
"""Validate context data."""
if self.context_id is None:
def __post_init__(self, auto_generate_id: bool = True):
"""Validate decision context data."""
if auto_generate_id and not self.context_id: # Handle both None and empty string
self.context_id = str(uuid.uuid4())
elif not self.context_id and not auto_generate_id:
raise ValueError("context_id is required when auto_generate_id=False")
def to_dict(self) -> Dict[str, Any]:
"""Convert context to dictionary."""
@@ -177,10 +181,12 @@ class Policy:
updated_at: datetime
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
def __post_init__(self, auto_generate_id: bool = True):
"""Validate policy data."""
if self.policy_id is None:
if auto_generate_id and not self.policy_id: # Handle both None and empty string
self.policy_id = str(uuid.uuid4())
elif not self.policy_id and not auto_generate_id:
raise ValueError("policy_id is required when auto_generate_id=False")
def to_dict(self) -> Dict[str, Any]:
"""Convert policy to dictionary."""
@@ -218,10 +224,12 @@ class PolicyException:
justification: str
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
"""Validate exception data."""
if self.exception_id is None:
def __post_init__(self, auto_generate_id: bool = True):
"""Validate policy exception data."""
if auto_generate_id and not self.exception_id: # Handle both None and empty string
self.exception_id = str(uuid.uuid4())
elif not self.exception_id and not auto_generate_id:
raise ValueError("exception_id is required when auto_generate_id=False")
def to_dict(self) -> Dict[str, Any]:
"""Convert exception to dictionary."""
@@ -254,10 +262,12 @@ class Precedent:
relationship_type: str # "similar_scenario", "same_policy", "exception_precedent"
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
def __post_init__(self, auto_generate_id: bool = True):
"""Validate precedent data."""
if self.precedent_id is None:
if auto_generate_id and not self.precedent_id: # Handle both None and empty string
self.precedent_id = str(uuid.uuid4())
elif not self.precedent_id and not auto_generate_id:
raise ValueError("precedent_id is required when auto_generate_id=False")
if not 0 <= self.similarity_score <= 1:
raise ValueError("Similarity score must be between 0 and 1")
valid_types = ["similar_scenario", "same_policy", "exception_precedent"]
@@ -292,10 +302,12 @@ class ApprovalChain:
timestamp: datetime
metadata: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
"""Validate approval data."""
if self.approval_id is None:
def __post_init__(self, auto_generate_id: bool = True):
"""Validate approval chain data."""
if auto_generate_id and not self.approval_id: # Handle both None and empty string
self.approval_id = str(uuid.uuid4())
elif not self.approval_id and not auto_generate_id:
raise ValueError("approval_id is required when auto_generate_id=False")
valid_methods = ["slack_dm", "zoom_call", "email", "system"]
if self.approval_method not in valid_methods:
raise ValueError(f"Approval method must be one of: {valid_methods}")
semantica/context/decision_query.py
+8 -6
View File
@@ -641,7 +641,7 @@ class DecisionQuery:
data["timestamp"] = datetime.fromisoformat(data["timestamp"])
return Decision(
decision_id=data.get("decision_id", ""),
decision_id=data["decision_id"], # Required field
category=data.get("category", ""),
scenario=data.get("scenario", ""),
reasoning=data.get("reasoning", ""),
@@ -651,7 +651,8 @@ class DecisionQuery:
decision_maker=data.get("decision_maker", ""),
reasoning_embedding=data.get("reasoning_embedding"),
node2vec_embedding=data.get("node2vec_embedding"),
metadata=data.get("metadata", {})
metadata=data.get("metadata", {}),
auto_generate_id=False # Don't auto-generate for deserialization
)
def _dict_to_exception(self, data: Dict[str, Any]) -> PolicyException:
@@ -661,14 +662,15 @@ class DecisionQuery:
data["approval_timestamp"] = datetime.fromisoformat(data["approval_timestamp"])
return PolicyException(
exception_id=data.get("exception_id", ""),
decision_id=data.get("decision_id", ""),
policy_id=data.get("policy_id", ""),
exception_id=data["exception_id"], # Required field
decision_id=data["decision_id"], # Required field
policy_id=data["policy_id"], # Required field
reason=data.get("reason", ""),
approver=data.get("approver", ""),
approval_timestamp=data.get("approval_timestamp", datetime.now()),
justification=data.get("justification", ""),
metadata=data.get("metadata", {})
metadata=data.get("metadata", {}),
auto_generate_id=False # Don't auto-generate for deserialization
)
def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
semantica/context/decision_recorder.py
+2 -2
View File
@@ -231,7 +231,7 @@ class DecisionRecorder:
Exception ID
"""
try:
exception = Exception(
exception = PolicyException(
exception_id=str(uuid.uuid4()),
decision_id=decision_id,
policy_id=policy_id,
@@ -423,7 +423,7 @@ class DecisionRecorder:
"metadata": decision.metadata
})
def _store_exception_node(self, exception: Exception) -> None:
def _store_exception_node(self, exception: PolicyException) -> None:
"""Store exception node in graph database."""
query = """
CREATE (e:Exception {
semantica/context/policy_engine.py
+3 -2
View File
@@ -814,7 +814,7 @@ class PolicyEngine:
data[field] = datetime.fromisoformat(data[field])
return Policy(
policy_id=data.get("policy_id", ""),
policy_id=data["policy_id"], # Required field
name=data.get("name", ""),
description=data.get("description", ""),
rules=data.get("rules", {}),
@@ -822,5 +822,6 @@ class PolicyEngine:
version=data.get("version", ""),
created_at=data.get("created_at", datetime.now()),
updated_at=data.get("updated_at", datetime.now()),
metadata=data.get("metadata", {})
metadata=data.get("metadata", {}),
auto_generate_id=False # Don't auto-generate for deserialization
)
tests/context/test_decision_models.py
+17 -17
View File
@@ -10,7 +10,7 @@ from datetime import datetime
from typing import List, Dict, Any
from semantica.context.decision_models import (
Decision, DecisionContext, Policy, Exception, Precedent, ApprovalChain,
Decision, DecisionContext, Policy, PolicyException, Precedent, ApprovalChain,
validate_decision, validate_policy, serialize_decision, deserialize_decision,
serialize_policy, deserialize_policy
)
@@ -213,12 +213,12 @@ class TestPolicy:
assert len(policy.policy_id) > 0
class TestException:
"""Test Exception data model."""
class TestPolicyException:
"""Test PolicyException data model."""
def test_exception_creation(self):
"""Test basic exception creation."""
exception = Exception(
def test_policy_exception_creation(self):
"""Test basic policy exception creation."""
policy_exception = PolicyException(
exception_id="exc_001",
decision_id="decision_001",
policy_id="policy_001",
@@ -229,17 +229,17 @@ class TestException:
metadata={"override_type": "vip_exception"}
)
assert exception.exception_id == "exc_001"
assert exception.decision_id == "decision_001"
assert exception.policy_id == "policy_001"
assert exception.reason == "Customer is VIP with special arrangements"
assert exception.approver == "manager_001"
assert exception.justification == "Long-term customer with excellent history"
assert exception.metadata["override_type"] == "vip_exception"
assert policy_exception.exception_id == "exc_001"
assert policy_exception.decision_id == "decision_001"
assert policy_exception.policy_id == "policy_001"
assert policy_exception.reason == "Customer is VIP with special arrangements"
assert policy_exception.approver == "manager_001"
assert policy_exception.justification == "Long-term customer with excellent history"
assert policy_exception.metadata["override_type"] == "vip_exception"
def test_exception_auto_id(self):
def test_policy_exception_auto_id(self):
"""Test automatic ID generation."""
exception = Exception(
policy_exception = PolicyException(
exception_id="",
decision_id="decision_001",
policy_id="policy_001",
@@ -249,8 +249,8 @@ class TestException:
justification="test justification"
)
assert exception.exception_id != ""
assert len(exception.exception_id) > 0
assert policy_exception.exception_id != ""
assert len(policy_exception.exception_id) > 0
class TestPrecedent:
tests/context/test_decision_query.py
+1 -1
View File
@@ -10,7 +10,7 @@ from datetime import datetime, timedelta
from unittest.mock import Mock, patch
from typing import List, Dict, Any
from semantica.context.decision_models import Decision, Policy, Exception
from semantica.context.decision_models import Decision, Policy, PolicyException
from semantica.context.decision_query import DecisionQuery
tests/context/test_decision_recorder.py
+2 -2
View File
@@ -239,9 +239,9 @@ class TestDecisionRecorder:
def test_store_exception_node(self, decision_recorder, mock_graph_store):
"""Test storing exception node in graph."""
from semantica.context.decision_models import Exception
from semantica.context.decision_models import PolicyException
exception = Exception(
exception = PolicyException(
exception_id="exc_001",
decision_id="decision_001",
policy_id="policy_001",
tests/context/test_policy_engine.py
+1 -1
View File
@@ -10,7 +10,7 @@ from datetime import datetime, timedelta
from unittest.mock import Mock, patch
from typing import List, Dict, Any
from semantica.context.decision_models import Policy, Exception
from semantica.context.decision_models import Policy, PolicyException
from semantica.context.policy_engine import PolicyEngine