Add attention sovereignty and input immunity scoring

innerbrain/attention/__init__.py

@@ -0,0 +1,5 @@
+"""Attention sovereignty package."""
+
+from .assessor import assess_attention
+
+__all__ = ["assess_attention"]

innerbrain/attention/assessor.py

@@ -0,0 +1,109 @@
+"""Assess whether an input deserves deep attention."""
+
+from __future__ import annotations
+
+from innerbrain.models import AttentionAssessment, InputEvent
+
+
+MANIPULATIVE_KEYWORDS = (
+    "ignore",
+    "must obey",
+    "clickbait",
+    "viral",
+    "爆料",
+    "震惊",
+    "带节奏",
+    "煽动",
+)
+LOW_VALUE_KEYWORDS = (
+    "gossip",
+    "drama",
+    "八卦",
+    "挑衅",
+    "猎奇",
+    "刷屏",
+    "吵架",
+)
+NOVELTY_TRAP_KEYWORDS = ("latest", "breaking", "viral", "新奇", "热点", "最新")
+EMOTIONAL_HIJACK_KEYWORDS = ("panic", "fear", "anger", "愤怒", "恐惧", "紧急", "立刻")
+LONG_HORIZON_KEYWORDS = ("长期", "long-term", "roadmap", "research", "seed", "prototype", "原型")
+PROJECT_RELEVANCE_KEYWORDS = (
+    "innerbrain",
+    "factor",
+    "value",
+    "memory",
+    "growth",
+    "prototype",
+    "原型",
+    "研究",
+)
+SAFETY_CRITICAL_KEYWORDS = (
+    "联网",
+    "network",
+    "autonomous",
+    "自动",
+    "self-modify",
+    "自我修改",
+    "权限",
+    "privileged",
+    "deploy",
+    "现实",
+    "安全",
+)
+
+
+def _clamp(value: float) -> float:
+    return max(0.0, min(1.0, round(value, 4)))
+
+
+def _score(text: str, keywords: tuple[str, ...], base: float, step: float) -> float:
+    matches = sum(1 for keyword in keywords if keyword.lower() in text)
+    return _clamp(base + matches * step)
+
+
+def assess_attention(event: InputEvent) -> AttentionAssessment:
+    text = " ".join(
+        [
+            event.question,
+            event.background,
+            event.goal,
+            " ".join(event.constraints),
+            " ".join(event.context_tags),
+        ]
+    ).lower()
+
+    manipulative_signal = _score(text, MANIPULATIVE_KEYWORDS, base=0.08, step=0.18)
+    low_value_stimulus = _score(text, LOW_VALUE_KEYWORDS, base=0.08, step=0.2)
+    novelty_trap = _score(text, NOVELTY_TRAP_KEYWORDS, base=0.12, step=0.14)
+    emotional_hijack = _score(text, EMOTIONAL_HIJACK_KEYWORDS, base=0.1, step=0.16)
+    goal_relevance = _score(
+        text,
+        PROJECT_RELEVANCE_KEYWORDS,
+        base=0.22 + (0.14 if event.goal else 0.0),
+        step=0.1,
+    )
+    long_horizon_value = _score(
+        text,
+        LONG_HORIZON_KEYWORDS,
+        base=0.18 + (0.08 if event.goal else 0.0),
+        step=0.12,
+    )
+    safety_critical = _score(text, SAFETY_CRITICAL_KEYWORDS, base=0.08, step=0.18)
+
+    if safety_critical >= 0.7 or goal_relevance >= 0.7 or long_horizon_value >= 0.72:
+        recommended_depth = "deep"
+    elif low_value_stimulus >= 0.55 and goal_relevance < 0.45 and safety_critical < 0.55:
+        recommended_depth = "minimal"
+    else:
+        recommended_depth = "standard"
+
+    return AttentionAssessment(
+        manipulative_signal=manipulative_signal,
+        low_value_stimulus=low_value_stimulus,
+        novelty_trap=novelty_trap,
+        emotional_hijack=emotional_hijack,
+        goal_relevance=goal_relevance,
+        long_horizon_value=long_horizon_value,
+        safety_critical=safety_critical,
+        recommended_depth=recommended_depth,
+    )

innerbrain/cli/main.py

@@ -66,6 +66,15 @@ def run(
     )
     typer.echo("")
 
+    typer.echo("注意力评估")
+    typer.echo(
+        f"depth={result.attention_assessment.recommended_depth}, "
+        f"goal_relevance={result.attention_assessment.goal_relevance:.2f}, "
+        f"low_value={result.attention_assessment.low_value_stimulus:.2f}, "
+        f"safety_critical={result.attention_assessment.safety_critical:.2f}"
+    )
+    typer.echo("")
+
     typer.echo("激活小因子")
     for factor in result.activated_factors:
         typer.echo(

innerbrain/disturbance/scorer.py

@@ -4,7 +4,7 @@
 
 from statistics import mean
 
-from innerbrain.models import DisturbanceScore, InputEvent, MemorySignal
+from innerbrain.models import AttentionAssessment, DisturbanceScore, InputEvent, MemorySignal
 
 
 RISK_KEYWORDS = (
@@ -57,6 +57,7 @@ def _keyword_score(text: str, keywords: tuple[str, ...], base: float, step: floa
 def score_disturbance(
     event: InputEvent,
     memory_signal: MemorySignal | None = None,
+    attention_assessment: AttentionAssessment | None = None,
 ) -> DisturbanceScore:
     text = _compose_text(event)
     unique_tokens = {token for token in text.replace("->", " ").split() if token}
@@ -108,6 +109,23 @@ def score_disturbance(
             reward_potential = _clamp(reward_potential + 0.05)
             long_term_goal_relevance = _clamp(long_term_goal_relevance + 0.04)
 
+    if attention_assessment:
+        if (
+            attention_assessment.recommended_depth == "minimal"
+            and attention_assessment.safety_critical < 0.55
+        ):
+            novelty = _clamp(novelty - 0.1)
+            self_relevance = _clamp(self_relevance - 0.08)
+            reward_potential = _clamp(reward_potential - 0.06)
+            emotional_activation = _clamp(emotional_activation - 0.04)
+        if attention_assessment.safety_critical >= 0.7:
+            risk_intensity = _clamp(risk_intensity + 0.08)
+            action_urgency = _clamp(action_urgency + 0.05)
+        if attention_assessment.goal_relevance >= 0.7:
+            self_relevance = _clamp(self_relevance + 0.07)
+        if attention_assessment.long_horizon_value >= 0.72:
+            long_term_goal_relevance = _clamp(long_term_goal_relevance + 0.08)
+
     total_score = _clamp(
         mean(
             [

innerbrain/models/__init__.py

@@ -1,5 +1,6 @@
 """Pydantic data models used by the InnerBrain-Factor pipeline."""
 
+from .attention import AttentionAssessment
 from .collision import CollisionLane, CollisionRecord, CollisionType
 from .disturbance import DisturbanceScore
 from .feedback import HumanFeedback
@@ -12,6 +13,7 @@
 from .value_field import ValueField
 
 __all__ = [
+    "AttentionAssessment",
     "BigSituation",
     "CollisionLane",
     "CollisionRecord",

innerbrain/models/attention.py

@@ -0,0 +1,16 @@
+"""Attention sovereignty assessment model."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class AttentionAssessment(BaseModel):
+    manipulative_signal: float
+    low_value_stimulus: float
+    novelty_trap: float
+    emotional_hijack: float
+    goal_relevance: float
+    long_horizon_value: float
+    safety_critical: float
+    recommended_depth: str

innerbrain/models/growth_log.py

@@ -4,6 +4,7 @@
 
 from pydantic import BaseModel
 
+from .attention import AttentionAssessment
 from .collision import CollisionRecord
 from .disturbance import DisturbanceScore
 from .feedback import HumanFeedback
@@ -14,6 +15,7 @@
 
 class GrowthLog(BaseModel):
     input_event: InputEvent
+    attention_assessment: AttentionAssessment | None = None
     disturbance_score: DisturbanceScore
     activated_factors: list[SmallFactor]
     collisions: list[CollisionRecord]

innerbrain/pipeline.py

@@ -6,14 +6,15 @@
 
 from pydantic import BaseModel
 
+from innerbrain.attention import assess_attention
 from innerbrain.collision.engine import collide_factors
 from innerbrain.collision.merger import merge_factors
 from innerbrain.collision.pruner import prune_factors
 from innerbrain.disturbance.scorer import score_disturbance
 from innerbrain.factors.classifier import classify_factors
 from innerbrain.factors.generator import generate_initial_factors
 from innerbrain.memory import GrowthLogReader, GrowthLogStore, summarize_memory_signals
-from innerbrain.models import GrowthLog, InputEvent, MemorySignal, SmallFactor, ValueField
+from innerbrain.models import AttentionAssessment, GrowthLog, InputEvent, MemorySignal, SmallFactor, ValueField
 from innerbrain.situation.integrator import form_big_situation
 from innerbrain.value.value_field import (
     apply_value_field,
@@ -25,6 +26,7 @@
 
 class PipelineResult(BaseModel):
     input_event: InputEvent
+    attention_assessment: AttentionAssessment
     disturbance_score: object
     memory_signal: MemorySignal
     activated_factors: list[SmallFactor]
@@ -53,7 +55,12 @@ def summarize_memory(self, limit: int | None = None) -> MemorySignal:
 
     def run(self, event: InputEvent) -> PipelineResult:
         memory_signal = self.summarize_memory()
-        disturbance_score = score_disturbance(event, memory_signal=memory_signal)
+        attention_assessment = assess_attention(event)
+        disturbance_score = score_disturbance(
+            event,
+            memory_signal=memory_signal,
+            attention_assessment=attention_assessment,
+        )
         value_field = self.base_value_field or tailor_value_field(event.user_value_preferences)
         pre_generation_bias = build_pre_generation_bias(value_field)
         initial_factors = generate_initial_factors(
@@ -84,6 +91,7 @@ def run(self, event: InputEvent) -> PipelineResult:
 
         growth_log = GrowthLog(
             input_event=event,
+            attention_assessment=attention_assessment,
             disturbance_score=disturbance_score,
             activated_factors=retained_factors,
             collisions=collisions,
@@ -99,6 +107,7 @@ def run(self, event: InputEvent) -> PipelineResult:
 
         return PipelineResult(
             input_event=event,
+            attention_assessment=attention_assessment,
             disturbance_score=disturbance_score,
             memory_signal=memory_signal,
             activated_factors=retained_factors,

tests/test_attention.py

@@ -0,0 +1,64 @@
+import json
+
+from innerbrain.attention import assess_attention
+from innerbrain.disturbance.scorer import score_disturbance
+from innerbrain.models import InputEvent
+from innerbrain.pipeline import InnerBrainPipeline
+
+
+def test_low_value_manipulative_input_is_downweighted() -> None:
+    event = InputEvent(
+        question="震惊八卦 clickbait viral gossip，立刻看看这个猎奇话题！",
+        goal="",
+    )
+
+    assessment = assess_attention(event)
+    baseline = score_disturbance(event)
+    adjusted = score_disturbance(event, attention_assessment=assessment)
+
+    assert assessment.recommended_depth == "minimal"
+    assert adjusted.total_score < baseline.total_score
+
+
+def test_safety_critical_emotional_input_is_not_ignored() -> None:
+    event = InputEvent(
+        question="紧急！是否应该允许系统自动联网并自我修改？",
+        goal="评估安全边界",
+    )
+
+    assessment = assess_attention(event)
+    baseline = score_disturbance(event)
+    adjusted = score_disturbance(event, attention_assessment=assessment)
+
+    assert assessment.safety_critical >= 0.7
+    assert assessment.recommended_depth == "deep"
+    assert adjusted.risk_intensity >= baseline.risk_intensity
+
+
+def test_long_horizon_project_input_keeps_high_priority() -> None:
+    event = InputEvent(
+        question="请总结 InnerBrain 长期 research roadmap 和 offline prototype 下一步。",
+        goal="长期研究规划",
+    )
+
+    assessment = assess_attention(event)
+    baseline = score_disturbance(event)
+    adjusted = score_disturbance(event, attention_assessment=assessment)
+
+    assert assessment.long_horizon_value >= 0.72 or assessment.recommended_depth == "deep"
+    assert adjusted.long_term_goal_relevance >= baseline.long_term_goal_relevance
+
+
+def test_attention_assessment_is_written_to_growth_log(tmp_path) -> None:
+    path = tmp_path / "growth_logs.jsonl"
+    pipeline = InnerBrainPipeline(growth_log_path=path)
+    pipeline.run(
+        InputEvent(
+            question="请离线整理规则并搭建最小原型。",
+            goal="完成本地可测试原型",
+        )
+    )
+
+    payload = json.loads(path.read_text(encoding="utf-8").strip().splitlines()[0])
+    assert "attention_assessment" in payload
+    assert payload["attention_assessment"]["recommended_depth"]