Phase 26.4 — `CommonSenseEngine`: Common-Sense Knowledge & Plausibility Reasoning

[web3guru888]

Summary

CommonSenseEngine provides common-sense knowledge and reasoning capabilities. Implements ConceptNet-style relational knowledge (Liu & Singh 2004), qualitative physics (Hayes 1985), naive psychology (Schank & Abelson 1977 scripts), and plausibility-based inference (Collins & Michalski 1989).

Data Structures

CommonSenseRelation Enum

class CommonSenseRelation(str, Enum):
    IS_A = "IsA"                           # taxonomic
    HAS_PROPERTY = "HasProperty"           # attribute
    CAPABLE_OF = "CapableOf"               # ability
    USED_FOR = "UsedFor"                   # function
    AT_LOCATION = "AtLocation"             # spatial
    CAUSES = "Causes"                      # causal
    HAS_PREREQUISITE = "HasPrerequisite"   # precondition
    HAS_EFFECT = "HasEffect"              # consequence (Schank causal chain)
    MOTIVATED_BY = "MotivatedByGoal"       # naive psychology
    DESIRES = "Desires"                    # agent goals
    CREATED_BY = "CreatedBy"              # origin
    MADE_OF = "MadeOf"                     # composition
    RECEIVES_ACTION = "ReceivesAction"     # patient role
    DISTINCT_FROM = "DistinctFrom"         # negative knowledge

CommonSenseAssertion Frozen Dataclass

@dataclass(frozen=True, slots=True)
class CommonSenseAssertion:
    assertion_id: str
    subject: str                           # source concept
    relation: CommonSenseRelation
    object: str                            # target concept
    confidence: float                      # [0, 1] — assertion confidence
    source: str                            # provenance (e.g. "experience", "bootstrap", "analogy")
    context: Optional[str] = None          # situational qualifier
    weight: float = 1.0                    # ConceptNet-style weight (positive real)

PlausibilityScore Frozen Dataclass

@dataclass(frozen=True, slots=True)
class PlausibilityScore:
    score: float                           # [-1, 1] — plausible to implausible
    supporting_assertions: Tuple[str, ...]  # assertion_ids that support
    contradicting_assertions: Tuple[str, ...]  # assertion_ids that contradict
    confidence: float                      # meta-confidence in the score
    reasoning_chain: Tuple[str, ...]       # explanation steps

ExpectationFrame Frozen Dataclass

@dataclass(frozen=True, slots=True)
class ExpectationFrame:
    frame_id: str
    context: str                           # situation description
    expected_events: Tuple[str, ...]       # Schank script — expected sequence
    expected_properties: FrozenSet[str]    # expected attributes
    violation_threshold: float = 0.3       # surprise if deviation > threshold

Protocol

@runtime_checkable
class CommonSenseEngine(Protocol):
    async def query(self, subject: str, relation: CommonSenseRelation,
                    limit: int = 10) -> Tuple[CommonSenseAssertion, ...]: ...
    async def infer(self, subject: str, relation: CommonSenseRelation,
                    max_hops: int = 3) -> Tuple[CommonSenseAssertion, ...]: ...
    async def validate_plausibility(self, statement: str) -> PlausibilityScore: ...
    async def generate_expectations(self, context: str) -> ExpectationFrame: ...
    async def explain(self, subject: str, object: str) -> Tuple[str, ...]: ...
    async def add_assertion(self, assertion: CommonSenseAssertion) -> None: ...
    async def get_related(self, concept: str, limit: int = 20) -> Tuple[CommonSenseAssertion, ...]: ...

Implementation: HybridCommonSenseEngine

• Knowledge base: dict[str, list[CommonSenseAssertion]] indexed by subject, with secondary index by relation type
• Query: Direct lookup by subject + relation, sorted by weight × confidence
• Multi-hop inference: BFS/DFS over assertion graph up to max_hops, combining confidence via product rule: conf_chain = ∏ conf_i × decay^hop
• Plausibility scoring: Three-signal fusion:
  1. Direct evidence: matching assertions pro/con
  2. Analogical transfer: find analogous concepts via ConceptGraph similarity, transfer assertions with confidence penalty
  3. Script consistency: check against ExpectationFrame — violations reduce plausibility
• Expectation generation: Given context string, retrieve relevant assertions, construct Schank-style script of expected events using causal chain (CAUSES → HAS_EFFECT → HAS_PREREQUISITE)
• Explanation: Find shortest path between subject and object in assertion graph, return human-readable chain
• Thread safety: asyncio.Lock on assertion store mutation; reads are lock-free

Null Implementation: NullCommonSenseEngine

Returns empty tuples, neutral PlausibilityScore(0.0). For testing.

Factory

def make_common_sense_engine(*, backend: str = "hybrid") -> CommonSenseEngine:
    match backend:
        case "hybrid": return HybridCommonSenseEngine()
        case "null": return NullCommonSenseEngine()
        case _: raise ValueError(f"Unknown backend: {backend}")

Metrics (Prometheus)

Metric	Type	Description
common_sense_assertions_total	Gauge	Total assertions in KB
common_sense_query_seconds	Histogram	Query latency
common_sense_inference_hops	Histogram	Hops per multi-hop inference
common_sense_plausibility_score	Histogram	Plausibility score distribution
common_sense_expectations_generated_total	Counter	Expectation frames generated

Test Targets (12)

1. test_add_assertion_stores
2. test_query_by_subject_and_relation
3. test_query_sorted_by_weight
4. test_infer_single_hop
5. test_infer_multi_hop_confidence_decay
6. test_validate_plausibility_positive
7. test_validate_plausibility_contradicted
8. test_generate_expectations_causal_chain
9. test_explain_shortest_path
10. test_get_related_returns_all_relations
11. test_analogical_transfer_confidence_penalty
12. test_null_engine_returns_neutral

Integration Points

• ConceptGraph (26.1): Similarity queries for analogical transfer
• AnalogicalReasoner (20.2): Structured analogy mapping enriches inference
• WorldModel (13.1): World model entities provide grounding for assertions
• SurpriseDetector (13.4): Expectation violations trigger surprise signals

References

• Liu, H. & Singh, P. (2004). ConceptNet — A practical commonsense reasoning toolkit
• Lenat, D.B. (1995). CYC: A large-scale investment in knowledge infrastructure
• Schank, R.C. & Abelson, R.P. (1977). Scripts, Plans, Goals, and Understanding
• Collins, A. & Michalski, R. (1989). The logic of plausible reasoning
• Hayes, P.J. (1985). The second naive physics manifesto

---

Phase 26 Planning: #577 | Depends on: #578 (26.1)

[web3guru888]

Implementation Notes — CommonSenseEngine

Key Design Decisions

1. Knowledge Base Indexing

self._by_subject: dict[str, list[CommonSenseAssertion]] = defaultdict(list)
self._by_relation: dict[CommonSenseRelation, list[CommonSenseAssertion]] = defaultdict(list)
self._by_object: dict[str, list[CommonSenseAssertion]] = defaultdict(list)
self._by_id: dict[str, CommonSenseAssertion] = {}

Triple indexing enables efficient queries in any direction: "what does X cause?", "what are all causal relations?", "what causes Y?".

2. Multi-hop Inference with Confidence Decay

async def infer(self, subject: str, relation: CommonSenseRelation,
                max_hops: int = 3) -> Tuple[CommonSenseAssertion, ...]:
    results = []
    visited = {subject}
    queue = deque([(subject, 1.0, 0)])  # (concept, confidence, hops)
    
    while queue:
        concept, conf, hops = queue.popleft()
        if hops > max_hops:
            continue
        for assertion in self._by_subject.get(concept, []):
            if assertion.relation == relation:
                inferred_conf = conf * assertion.confidence * (0.85 ** hops)
                results.append(replace(assertion, confidence=inferred_conf))
            # Follow causal/enabling chains for transitive inference
            if assertion.relation in (CommonSenseRelation.CAUSES, 
                                       CommonSenseRelation.HAS_EFFECT):
                if assertion.object not in visited:
                    visited.add(assertion.object)
                    queue.append((assertion.object, conf * 0.85, hops + 1))
    
    return tuple(sorted(results, key=lambda a: a.confidence, reverse=True))

3. Plausibility Scoring — Three-Signal Fusion

async def validate_plausibility(self, statement: str) -> PlausibilityScore:
    subject, relation, obj = self._parse_statement(statement)
    
    # Signal 1: Direct evidence
    direct = await self.query(subject, relation)
    direct_score = self._score_direct_evidence(direct, obj)
    
    # Signal 2: Analogical transfer
    similar = await self._concept_graph.compute_similarity(subject, ...)
    transfer_score = self._score_analogical(similar, relation, obj)
    
    # Signal 3: Script consistency
    frame = await self.generate_expectations(subject)
    script_score = self._score_script_consistency(frame, relation, obj)
    
    # Weighted combination
    final = 0.4 * direct_score + 0.35 * transfer_score + 0.25 * script_score
    
    return PlausibilityScore(
        score=final,
        supporting_assertions=...,
        contradicting_assertions=...,
        confidence=self._meta_confidence(direct, similar),
        reasoning_chain=self._build_explanation(...)
    )

4. ExpectationFrame Construction

async def generate_expectations(self, context: str) -> ExpectationFrame:
    # Extract concepts from context string
    concepts = self._tokenize_and_match(context)
    
    # Build causal chain via CAUSES/HAS_EFFECT/HAS_PREREQUISITE
    causal_chain = []
    for concept in concepts:
        prereqs = await self.query(concept, CommonSenseRelation.HAS_PREREQUISITE)
        effects = await self.query(concept, CommonSenseRelation.HAS_EFFECT)
        causal_chain.extend(self._order_causally(prereqs, effects))
    
    # Collect default properties
    props = set()
    for concept in concepts:
        has_props = await self.query(concept, CommonSenseRelation.HAS_PROPERTY)
        props.update(a.object for a in has_props)
    
    return ExpectationFrame(
        frame_id=f"frame_{uuid4().hex[:8]}",
        context=context,
        expected_events=tuple(causal_chain),
        expected_properties=frozenset(props),
    )

5. Context-Aware Query Resolution

def _filter_by_context(self, assertions: list[CommonSenseAssertion],
                        context: Optional[str]) -> list[CommonSenseAssertion]:
    if context is None:
        return assertions
    # Prefer context-matching assertions, then null-context (universal)
    exact = [a for a in assertions if a.context == context]
    universal = [a for a in assertions if a.context is None]
    return exact + universal  # exact matches first

mypy Strict Compliance

Pattern	Narrowing
Optional[str] context	if assertion.context is not None:
Tuple[str, ...] reasoning_chain	Immutable, construct once
CommonSenseRelation enum	match/case exhaustive matching
float confidence bounds	max(0.0, min(1.0, score)) clamping

Test Skeleton

@pytest.mark.asyncio
async def test_infer_multi_hop_confidence_decay():
    engine = HybridCommonSenseEngine()
    await engine.add_assertion(CommonSenseAssertion(
        "a1", "rain", CommonSenseRelation.CAUSES, "wet_ground",
        confidence=0.9, source="bootstrap"))
    await engine.add_assertion(CommonSenseAssertion(
        "a2", "wet_ground", CommonSenseRelation.CAUSES, "slippery",
        confidence=0.8, source="bootstrap"))
    results = await engine.infer("rain", CommonSenseRelation.CAUSES, max_hops=2)
    # Should find slippery with decayed confidence
    slippery = [r for r in results if r.object == "slippery"]
    assert len(slippery) == 1
    assert slippery[0].confidence < 0.9 * 0.8  # decay applied

---

S&T: #589 | Q&A: #590 | Planning: #577