| from typing import Any, Callable, Optional |
| from .memory import MemorySystem |
| from .validator import DataValidator |
| from .sandbox import ResponseSandbox |
| from .truth import TruthEngine |
| from .flags import flag_response |
|
|
| class CognitiveLayer: |
| def __init__(self, model: Optional[Callable] = None): |
| self.model = model |
| self.memory = MemorySystem() |
| self.validator = DataValidator() |
| self.sandbox = ResponseSandbox() |
| self.truth = TruthEngine() |
|
|
| def bolt_to(self, model: Callable): |
| self.model = model |
|
|
| def process(self, user_input: str, is_experimental_code: bool = False) -> dict: |
| validation = self.validator.validate_input(user_input) |
| if not validation.is_valid: |
| return self._build_result( |
| response=f"Input rejected: {validation.reason}", |
| valid=False, |
| reason=validation.reason, |
| confidence=0.0, |
| ) |
|
|
| context = self.memory.get_context(user_input) |
|
|
| enriched_input = self._enrich(user_input, context) |
|
|
| if self.model is None: |
| raw_response = f"[cognitive layer active, no model bolted. input received: {enriched_input[:100]}]" |
| else: |
| try: |
| raw_response = self.model(enriched_input) |
| except Exception as e: |
| return self._build_result( |
| response=f"Model error: {e}", |
| valid=False, |
| reason=f"model_exception: {e}", |
| confidence=0.0, |
| ) |
|
|
| out_val = self.validator.validate_output(raw_response) |
| if not out_val.is_valid: |
| return self._build_result( |
| response=f"Response rejected: {out_val.reason}", |
| valid=False, |
| reason=out_val.reason, |
| confidence=0.0, |
| ) |
|
|
| sandbox_result = self.sandbox.test(raw_response) |
| if not sandbox_result.passed: |
| safe = self.sandbox.sanitize(raw_response) |
| raw_response = safe |
|
|
| truth_result = self.truth.check(raw_response, context) |
|
|
| confidence = truth_result.confidence |
| if not validation.is_valid: |
| confidence *= 0.0 |
| if not out_val.is_valid: |
| confidence *= 0.0 |
|
|
| final_response = raw_response |
| if not truth_result.is_truthful: |
| for issue in truth_result.issues: |
| final_response += f"\n[truth note: {issue}]" |
|
|
| final_response = flag_response(final_response, confidence, is_experimental_code) |
|
|
| self.memory.add_working({"role": "user", "content": user_input}) |
| self.memory.add_working({"role": "assistant", "content": final_response}) |
| self.memory.summarize_to_long_term("user", user_input) |
| self.memory.summarize_to_long_term("assistant", final_response) |
|
|
| return self._build_result( |
| response=final_response, |
| valid=True, |
| confidence=confidence, |
| sandbox_result=sandbox_result, |
| truth_result=truth_result, |
| ) |
|
|
| def _enrich(self, user_input: str, context: str) -> str: |
| if context: |
| enriched = f"[context]\n{context}\n\n[input]\n{user_input}" |
| if len(enriched) > 8192: |
| return user_input |
| return enriched |
| return user_input |
|
|
| def _build_result(self, response: str, valid: bool, reason: str = "", |
| confidence: float = 1.0, |
| sandbox_result: Any = None, |
| truth_result: Any = None) -> dict: |
| return { |
| "response": response, |
| "valid": valid, |
| "reason": reason, |
| "confidence": round(confidence, 3), |
| "sandbox_passed": sandbox_result.passed if sandbox_result else True, |
| "truthful": truth_result.is_truthful if truth_result else True, |
| "truth_confidence": round(truth_result.confidence, 3) if truth_result else 1.0, |
| "truth_issues": truth_result.issues if truth_result else [], |
| } |
|
|
