From 6aff0cae7412ebb061e901ca4236369a646f9e33 Mon Sep 17 00:00:00 2001 From: sparklkt Date: Tue, 12 May 2026 23:45:32 +0800 Subject: [PATCH] Add attention sovereignty and input immunity scoring --- innerbrain/attention/__init__.py | 5 ++ innerbrain/attention/assessor.py | 109 +++++++++++++++++++++++++++++++ innerbrain/cli/main.py | 9 +++ innerbrain/disturbance/scorer.py | 20 +++++- innerbrain/models/__init__.py | 2 + innerbrain/models/attention.py | 16 +++++ innerbrain/models/growth_log.py | 2 + innerbrain/pipeline.py | 13 +++- tests/test_attention.py | 64 ++++++++++++++++++ 9 files changed, 237 insertions(+), 3 deletions(-) create mode 100644 innerbrain/attention/__init__.py create mode 100644 innerbrain/attention/assessor.py create mode 100644 innerbrain/models/attention.py create mode 100644 tests/test_attention.py diff --git a/innerbrain/attention/__init__.py b/innerbrain/attention/__init__.py new file mode 100644 index 0000000..e61e1b0 --- /dev/null +++ b/innerbrain/attention/__init__.py @@ -0,0 +1,5 @@ +"""Attention sovereignty package.""" + +from .assessor import assess_attention + +__all__ = ["assess_attention"] diff --git a/innerbrain/attention/assessor.py b/innerbrain/attention/assessor.py new file mode 100644 index 0000000..063876a --- /dev/null +++ b/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, + ) diff --git a/innerbrain/cli/main.py b/innerbrain/cli/main.py index 63870cc..410a9cf 100644 --- a/innerbrain/cli/main.py +++ b/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( diff --git a/innerbrain/disturbance/scorer.py b/innerbrain/disturbance/scorer.py index a82b01a..c8eb1a4 100644 --- a/innerbrain/disturbance/scorer.py +++ b/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( [ diff --git a/innerbrain/models/__init__.py b/innerbrain/models/__init__.py index f2729e0..200a5de 100644 --- a/innerbrain/models/__init__.py +++ b/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", diff --git a/innerbrain/models/attention.py b/innerbrain/models/attention.py new file mode 100644 index 0000000..e021d90 --- /dev/null +++ b/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 diff --git a/innerbrain/models/growth_log.py b/innerbrain/models/growth_log.py index 547920d..aa2367e 100644 --- a/innerbrain/models/growth_log.py +++ b/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] diff --git a/innerbrain/pipeline.py b/innerbrain/pipeline.py index 55181a5..bcb121f 100644 --- a/innerbrain/pipeline.py +++ b/innerbrain/pipeline.py @@ -6,6 +6,7 @@ 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 @@ -13,7 +14,7 @@ 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, diff --git a/tests/test_attention.py b/tests/test_attention.py new file mode 100644 index 0000000..045bcca --- /dev/null +++ b/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"]