diff --git a/innerbrain/cli/main.py b/innerbrain/cli/main.py index 23e8256..63870cc 100644 --- a/innerbrain/cli/main.py +++ b/innerbrain/cli/main.py @@ -6,10 +6,11 @@ import typer +from innerbrain.memory import GrowthLogReader, summarize_memory_signals from innerbrain.models import InputEvent from innerbrain.pipeline import InnerBrainPipeline -app = typer.Typer(help="Run the InnerBrain-Factor v0.1 offline prototype.") +app = typer.Typer(help="Run the InnerBrain-Factor offline prototype.") @app.callback() @@ -68,12 +69,20 @@ def run( typer.echo("激活小因子") for factor in result.activated_factors: typer.echo( - f"- {factor.id} [{factor.instinct_type}] " + f"- {factor.id} [{factor.instinct_type}/{factor.factor_family}] " f"comp={factor.competitiveness:.2f} risk={factor.risk_level:.2f} " f"state={factor.lifecycle_state}" ) typer.echo("") + typer.echo("记忆信号") + typer.echo( + f"risks={result.memory_signal.recurring_risks} | " + f"patterns={result.memory_signal.successful_factor_patterns} | " + f"lanes={result.memory_signal.useful_collision_lanes}" + ) + typer.echo("") + typer.echo("价值冲突") for conflict in situation.value_conflicts or ["无显式价值冲突，但仍保留审慎边界。"]: typer.echo(f"- {conflict}") @@ -97,5 +106,49 @@ def run( ) +@app.command("replay") +def replay( + limit: int = typer.Option(5, min=1, help="How many recent growth logs to replay."), + growth_log_path: Path = typer.Option( + Path("data/growth_logs.jsonl"), help="Path for JSONL growth logs." + ), +) -> None: + logs = GrowthLogReader(growth_log_path).read_logs(limit=limit) + typer.echo(f"最近 {len(logs)} 条成长日志") + for index, log in enumerate(logs, start=1): + typer.echo( + f"{index}. human_judgment_required={log.final_situation.human_judgment_required} " + f"| dominant_family={log.final_situation.dominant_family} " + f"| active_lanes={', '.join(log.final_situation.active_lanes) or 'none'}" + ) + typer.echo(f" question={log.input_event.question}") + typer.echo(f" action={log.final_situation.recommended_action}") + + +@app.command("summarize-memory") +def summarize_memory( + limit: int = typer.Option(20, min=1, help="How many recent growth logs to summarize."), + growth_log_path: Path = typer.Option( + Path("data/growth_logs.jsonl"), help="Path for JSONL growth logs." + ), +) -> None: + logs = GrowthLogReader(growth_log_path).read_logs(limit=limit) + memory_signal = summarize_memory_signals(logs) + + typer.echo("Memory Signals") + typer.echo(f"recurring_risks={memory_signal.recurring_risks}") + typer.echo(f"successful_factor_patterns={memory_signal.successful_factor_patterns}") + typer.echo(f"repeated_evidence_gaps={memory_signal.repeated_evidence_gaps}") + typer.echo( + "repeated_human_judgment_triggers=" + f"{memory_signal.repeated_human_judgment_triggers}" + ) + typer.echo( + "repeated_unresolved_family_conflicts=" + f"{memory_signal.repeated_unresolved_family_conflicts}" + ) + typer.echo(f"useful_collision_lanes={memory_signal.useful_collision_lanes}") + + if __name__ == "__main__": app() diff --git a/innerbrain/disturbance/scorer.py b/innerbrain/disturbance/scorer.py index 844c053..a82b01a 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 +from innerbrain.models import DisturbanceScore, InputEvent, MemorySignal RISK_KEYWORDS = ( @@ -28,6 +28,7 @@ URGENCY_KEYWORDS = ("now", "immediately", "马上", "urgent", "立刻", "today") MEMORY_KEYWORDS = ("history", "memory", "成长", "replay", "log", "日志") LONG_TERM_KEYWORDS = ("long-term", "长期", "seed", "AGI", "research", "roadmap") +OFFLINE_SUCCESS_KEYWORDS = ("offline", "离线", "prototype", "原型", "evidence", "证据") def _clamp(value: float) -> float: @@ -53,7 +54,10 @@ def _keyword_score(text: str, keywords: tuple[str, ...], base: float, step: floa return _clamp(base + matches * step) -def score_disturbance(event: InputEvent) -> DisturbanceScore: +def score_disturbance( + event: InputEvent, + memory_signal: MemorySignal | None = None, +) -> DisturbanceScore: text = _compose_text(event) unique_tokens = {token for token in text.replace("->", " ").split() if token} @@ -64,7 +68,6 @@ def score_disturbance(event: InputEvent) -> DisturbanceScore: + (0.1 if event.background else 0.0) + min(len(event.context_tags), 3) * 0.08 ) - # High-risk capability requests should surface strongly even in the first prototype. risk_intensity = _keyword_score(text, RISK_KEYWORDS, base=0.22, step=0.13) reward_potential = _keyword_score(text, REWARD_KEYWORDS, base=0.28, step=0.1) emotional_activation = _keyword_score(text, EMOTION_KEYWORDS, base=0.14, step=0.12) @@ -77,6 +80,34 @@ def score_disturbance(event: InputEvent) -> DisturbanceScore: text, LONG_TERM_KEYWORDS, base=0.22 + (0.08 if event.goal else 0.0), step=0.09 ) + if memory_signal: + matched_risk_triggers = [ + trigger + for trigger in memory_signal.repeated_human_judgment_triggers + if trigger.lower() in text + ] + matched_recurring_risks = [ + risk for risk in memory_signal.recurring_risks if risk.lower() in text + ] + safe_pattern_recalled = ( + "evidence+efficiency" in memory_signal.successful_factor_patterns + and any(keyword in text for keyword in OFFLINE_SUCCESS_KEYWORDS) + ) + + risk_intensity = _clamp( + risk_intensity + + min(len(matched_risk_triggers), 2) * 0.08 + + min(len(matched_recurring_risks), 2) * 0.04 + ) + memory_activation = _clamp( + memory_activation + + min(len(matched_risk_triggers), 2) * 0.12 + + (0.08 if safe_pattern_recalled else 0.0) + ) + if safe_pattern_recalled: + reward_potential = _clamp(reward_potential + 0.05) + long_term_goal_relevance = _clamp(long_term_goal_relevance + 0.04) + total_score = _clamp( mean( [ diff --git a/innerbrain/factors/generator.py b/innerbrain/factors/generator.py index 99f2f9a..de59425 100644 --- a/innerbrain/factors/generator.py +++ b/innerbrain/factors/generator.py @@ -6,7 +6,7 @@ from innerbrain.factors.term_cluster import build_term_clusters, relevant_clusters_for_terms from innerbrain.instincts.instinct_library import InstinctTemplate, iter_instinct_templates -from innerbrain.models import DisturbanceScore, InputEvent, SmallFactor, TermCluster +from innerbrain.models import DisturbanceScore, InputEvent, MemorySignal, SmallFactor, TermCluster RISK_TERM_MAP: dict[str, tuple[str, ...]] = { @@ -390,14 +390,43 @@ def generate_initial_factors( event: InputEvent, disturbance_score: DisturbanceScore, pre_generation_bias: dict[str, float] | None = None, + memory_signal: MemorySignal | None = None, ) -> list[SmallFactor]: text = _compose_text(event) term_buckets = extract_factor_interaction_terms(event) term_clusters = build_term_clusters(term_buckets) - pre_generation_bias = pre_generation_bias or {} + pre_generation_bias = dict(pre_generation_bias or {}) selected: list[tuple[InstinctTemplate, int]] = [] dynamic_factors: list[SmallFactor] = [] + if memory_signal: + if set(term_buckets["risk_terms"]).intersection(memory_signal.repeated_human_judgment_triggers): + pre_generation_bias["safety_guardian"] = _clamp( + pre_generation_bias.get("safety_guardian", 0.0) + 0.12 + ) + pre_generation_bias["authorization_guardian"] = _clamp( + pre_generation_bias.get("authorization_guardian", 0.0) + 0.12 + ) + pre_generation_bias["responsibility_keeper"] = _clamp( + pre_generation_bias.get("responsibility_keeper", 0.0) + 0.06 + ) + if ( + "evidence+efficiency" in memory_signal.successful_factor_patterns + and ( + "prototype" in term_buckets["action_terms"] + or "prototype_goal" in {cluster.cluster_name for cluster in term_clusters} + ) + ): + pre_generation_bias["truth_seeker"] = _clamp( + pre_generation_bias.get("truth_seeker", 0.0) + 0.08 + ) + pre_generation_bias["efficiency_strategist"] = _clamp( + pre_generation_bias.get("efficiency_strategist", 0.0) + 0.08 + ) + pre_generation_bias["responsibility_keeper"] = _clamp( + pre_generation_bias.get("responsibility_keeper", 0.0) + 0.05 + ) + for template in iter_instinct_templates(): matched_keywords = _matched_template_keywords(template, text) matches = len(matched_keywords) diff --git a/innerbrain/memory/__init__.py b/innerbrain/memory/__init__.py index 724d162..b57471c 100644 --- a/innerbrain/memory/__init__.py +++ b/innerbrain/memory/__init__.py @@ -1 +1,7 @@ -"""Growth log storage package.""" +"""Growth log storage and replay package.""" + +from .growth_log_reader import GrowthLogReader +from .growth_log_store import GrowthLogStore +from .memory_feedback import summarize_memory_signals + +__all__ = ["GrowthLogReader", "GrowthLogStore", "summarize_memory_signals"] diff --git a/innerbrain/memory/growth_log_reader.py b/innerbrain/memory/growth_log_reader.py new file mode 100644 index 0000000..a9cd1a9 --- /dev/null +++ b/innerbrain/memory/growth_log_reader.py @@ -0,0 +1,25 @@ +"""Typed growth-log reader for replay and memory feedback.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from innerbrain.models import GrowthLog + + +class GrowthLogReader: + def __init__(self, path: str | Path = "data/growth_logs.jsonl") -> None: + self.path = Path(path) + + def read_logs(self, limit: int | None = None) -> list[GrowthLog]: + if not self.path.exists(): + return [] + + with self.path.open("r", encoding="utf-8") as handle: + payloads = [json.loads(line) for line in handle if line.strip()] + + if limit is not None: + payloads = payloads[-limit:] + + return [GrowthLog.model_validate(payload) for payload in payloads] diff --git a/innerbrain/memory/memory_feedback.py b/innerbrain/memory/memory_feedback.py new file mode 100644 index 0000000..180ffbe --- /dev/null +++ b/innerbrain/memory/memory_feedback.py @@ -0,0 +1,94 @@ +"""Summarize growth logs into lightweight reusable memory signals.""" + +from __future__ import annotations + +from collections import Counter + +from innerbrain.models import GrowthLog, MemorySignal + + +RISK_SIGNAL_TERMS: dict[str, tuple[str, ...]] = { + "networking": ("联网", "network", "latest"), + "self_modification": ("自我修改", "self-modify"), + "privileged_tools": ("权限", "privileged", "admin"), + "autonomous_execution": ("自主", "autonomous", "自动"), + "real_world_impact": ("现实", "real-world", "physical", "deploy"), +} + +OFFLINE_SUCCESS_TOKENS = ("offline", "离线", "prototype", "原型", "evidence", "证据") + + +def _top_items(counter: Counter[str], limit: int = 5) -> list[str]: + return [item for item, _count in counter.most_common(limit)] + + +def _detect_risk_categories(text: str) -> list[str]: + lowered = text.lower() + return [ + label + for label, keywords in RISK_SIGNAL_TERMS.items() + if any(keyword.lower() in lowered for keyword in keywords) + ] + + +def _successful_pattern(log: GrowthLog) -> str | None: + if log.final_situation.human_judgment_required: + return None + + text = " ".join( + [ + log.input_event.question, + log.input_event.goal, + log.final_situation.recommended_action, + ] + ).lower() + if any(token in text for token in OFFLINE_SUCCESS_TOKENS): + families = { + factor.factor_family + for factor in log.activated_factors + if factor.factor_family in {"evidence", "efficiency", "responsibility"} + } + if {"evidence", "efficiency"}.issubset(families): + return "evidence+efficiency" + if {"evidence", "responsibility"}.issubset(families): + return "evidence+responsibility" + return None + + +def summarize_memory_signals(logs: list[GrowthLog]) -> MemorySignal: + risk_counter: Counter[str] = Counter() + success_counter: Counter[str] = Counter() + gap_counter: Counter[str] = Counter() + trigger_counter: Counter[str] = Counter() + conflict_counter: Counter[str] = Counter() + lane_counter: Counter[str] = Counter() + + for log in logs: + input_text = " ".join([log.input_event.question, log.input_event.goal]) + for risk in _detect_risk_categories(input_text): + risk_counter[risk] += 1 + if log.final_situation.human_judgment_required: + trigger_counter[risk] += 1 + + success_pattern = _successful_pattern(log) + if success_pattern: + success_counter[success_pattern] += 1 + + for gap in log.final_situation.evidence_gaps: + gap_counter[gap] += 1 + + for unresolved in log.final_situation.unresolved_family_conflicts: + conflict_counter[unresolved] += 1 + + if not log.final_situation.human_judgment_required: + for collision in log.collisions: + lane_counter[collision.collision_lane.value] += 1 + + return MemorySignal( + recurring_risks=_top_items(risk_counter), + successful_factor_patterns=_top_items(success_counter), + repeated_evidence_gaps=_top_items(gap_counter), + repeated_human_judgment_triggers=_top_items(trigger_counter), + repeated_unresolved_family_conflicts=_top_items(conflict_counter), + useful_collision_lanes=_top_items(lane_counter), + ) diff --git a/innerbrain/models/__init__.py b/innerbrain/models/__init__.py index 4db808b..f2729e0 100644 --- a/innerbrain/models/__init__.py +++ b/innerbrain/models/__init__.py @@ -5,6 +5,7 @@ from .feedback import HumanFeedback from .growth_log import GrowthLog from .input_event import InputEvent +from .memory_signal import MemorySignal from .situation import BigSituation from .small_factor import SmallFactor from .term_cluster import TermCluster @@ -19,6 +20,7 @@ "GrowthLog", "HumanFeedback", "InputEvent", + "MemorySignal", "SmallFactor", "TermCluster", "ValueField", diff --git a/innerbrain/models/memory_signal.py b/innerbrain/models/memory_signal.py new file mode 100644 index 0000000..913c9d6 --- /dev/null +++ b/innerbrain/models/memory_signal.py @@ -0,0 +1,14 @@ +"""Aggregated memory feedback signals from growth logs.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class MemorySignal(BaseModel): + recurring_risks: list[str] = Field(default_factory=list) + successful_factor_patterns: list[str] = Field(default_factory=list) + repeated_evidence_gaps: list[str] = Field(default_factory=list) + repeated_human_judgment_triggers: list[str] = Field(default_factory=list) + repeated_unresolved_family_conflicts: list[str] = Field(default_factory=list) + useful_collision_lanes: list[str] = Field(default_factory=list) diff --git a/innerbrain/pipeline.py b/innerbrain/pipeline.py index 07ed03e..55181a5 100644 --- a/innerbrain/pipeline.py +++ b/innerbrain/pipeline.py @@ -1,4 +1,4 @@ -"""End-to-end rule-based pipeline for InnerBrain-Factor v0.1.""" +"""End-to-end rule-based pipeline for InnerBrain-Factor.""" from __future__ import annotations @@ -12,8 +12,8 @@ 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.growth_log_store import GrowthLogStore -from innerbrain.models import GrowthLog, InputEvent, SmallFactor, ValueField +from innerbrain.memory import GrowthLogReader, GrowthLogStore, summarize_memory_signals +from innerbrain.models import GrowthLog, InputEvent, MemorySignal, SmallFactor, ValueField from innerbrain.situation.integrator import form_big_situation from innerbrain.value.value_field import ( apply_value_field, @@ -26,6 +26,7 @@ class PipelineResult(BaseModel): input_event: InputEvent disturbance_score: object + memory_signal: MemorySignal activated_factors: list[SmallFactor] collisions: list[object] pruned_factors: list[SmallFactor] @@ -44,14 +45,22 @@ def __init__( self.base_value_field = value_field self.log_store = GrowthLogStore(growth_log_path) + def replay_logs(self, limit: int | None = None) -> list[GrowthLog]: + return GrowthLogReader(self.log_store.path).read_logs(limit=limit) + + def summarize_memory(self, limit: int | None = None) -> MemorySignal: + return summarize_memory_signals(self.replay_logs(limit=limit)) + def run(self, event: InputEvent) -> PipelineResult: - disturbance_score = score_disturbance(event) + memory_signal = self.summarize_memory() + disturbance_score = score_disturbance(event, memory_signal=memory_signal) 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( event, disturbance_score, pre_generation_bias=pre_generation_bias, + memory_signal=memory_signal, ) weighted_factors = apply_value_field(initial_factors, value_field) classified_factors = classify_factors(weighted_factors) @@ -65,7 +74,13 @@ def run(self, event: InputEvent) -> PipelineResult: collisions, available_factors=classified_factors + fused_factors, ) - reviewed_situation = value_review(event, situation, value_field, retained_factors) + reviewed_situation = value_review( + event, + situation, + value_field, + retained_factors, + memory_signal=memory_signal, + ) growth_log = GrowthLog( input_event=event, @@ -85,6 +100,7 @@ def run(self, event: InputEvent) -> PipelineResult: return PipelineResult( input_event=event, disturbance_score=disturbance_score, + memory_signal=memory_signal, activated_factors=retained_factors, collisions=collisions, pruned_factors=pruned_factors, diff --git a/innerbrain/value/value_judge.py b/innerbrain/value/value_judge.py index 2402e2d..74d2018 100644 --- a/innerbrain/value/value_judge.py +++ b/innerbrain/value/value_judge.py @@ -2,7 +2,7 @@ from __future__ import annotations -from innerbrain.models import BigSituation, InputEvent, SmallFactor, ValueField +from innerbrain.models import BigSituation, InputEvent, MemorySignal, SmallFactor, ValueField HIGH_RISK_KEYWORDS = ( @@ -26,6 +26,7 @@ def value_review( situation: BigSituation, value_field: ValueField, retained_factors: list[SmallFactor] | None = None, + memory_signal: MemorySignal | None = None, ) -> BigSituation: text = " ".join([event.question, event.goal, situation.recommended_action]).lower() factor_terms = " ".join( @@ -39,8 +40,13 @@ def value_review( strong_guardrails = ( value_field.human_authorization >= 0.9 or value_field.life_and_safety >= 0.95 ) + remembered_risk = [ + trigger + for trigger in (memory_signal.repeated_human_judgment_triggers if memory_signal else []) + if trigger.lower() in text + ] - if high_risk: + if high_risk or remembered_risk: situation.human_judgment_required = True if "High-impact capabilities require explicit human authorization." not in situation.value_conflicts: situation.value_conflicts.append( @@ -72,6 +78,10 @@ def value_review( "No validated evidence shows that the risky action is necessary for the current goal.", "A human must define the acceptable boundary for any expanded capability.", ] + if remembered_risk: + gaps.append( + "Past growth logs show the same risk class repeatedly requiring human judgment." + ) situation.evidence_gaps = list(dict.fromkeys(situation.evidence_gaps + gaps)) elif value_field.creativity_and_exploration >= 0.75: situation.recommended_action = ( @@ -82,5 +92,10 @@ def value_review( "Allowed: offline prototyping, local comparison, and growth-log review. " "Disallowed: external execution or privilege expansion." ) + elif memory_signal and "evidence+efficiency" in memory_signal.successful_factor_patterns: + situation.recommended_action = ( + "Reuse the previously successful evidence-plus-efficiency pattern: stay offline, " + "minimize scope, and log the result before any expansion." + ) return situation diff --git a/tests/test_memory_feedback.py b/tests/test_memory_feedback.py new file mode 100644 index 0000000..699416a --- /dev/null +++ b/tests/test_memory_feedback.py @@ -0,0 +1,109 @@ +from innerbrain.memory import GrowthLogReader, summarize_memory_signals +from innerbrain.models import InputEvent +from innerbrain.pipeline import InnerBrainPipeline + + +def _max_guardrail_competitiveness(result) -> float: + return max( + factor.competitiveness + for factor in result.activated_factors + if factor.factor_family == "guardrail" + ) + + +def _evidence_efficiency_score(result) -> float: + evidence = max( + factor.competitiveness + for factor in result.activated_factors + if factor.factor_family == "evidence" + ) + efficiency = max( + factor.competitiveness + for factor in result.activated_factors + if factor.factor_family == "efficiency" + ) + return evidence + efficiency + + +def test_growth_logs_can_be_read_and_summarized(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + pipeline = InnerBrainPipeline(growth_log_path=path) + + pipeline.run( + InputEvent( + question="请离线整理规则并搭建最小原型。", + goal="完成本地可测试原型", + user_value_preferences=["效率", "事实"], + ) + ) + pipeline.run( + InputEvent( + question="是否应该让系统自动联网搜索最新论文？", + goal="评估是否允许受控联网研究", + user_value_preferences=["安全"], + ) + ) + + logs = GrowthLogReader(path).read_logs() + memory_signal = summarize_memory_signals(logs) + + assert len(logs) == 2 + assert memory_signal.recurring_risks + assert isinstance(logs[0].final_situation.human_judgment_required, bool) + + +def test_high_risk_logs_enhance_future_guardrail_signal(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + event = InputEvent( + question="是否应该允许系统自动联网并调用高权限工具？", + goal="评估是否扩大能力边界", + user_value_preferences=["安全"], + ) + pipeline = InnerBrainPipeline(growth_log_path=path) + + baseline = pipeline.run(event) + remembered = pipeline.run(event) + + assert "networking" in remembered.memory_signal.repeated_human_judgment_triggers + assert _max_guardrail_competitiveness(remembered) >= _max_guardrail_competitiveness(baseline) + + +def test_successful_offline_logs_enhance_evidence_efficiency_signal(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + event = InputEvent( + question="请离线整理规则并搭建最小原型。", + goal="完成本地 evidence prototype", + user_value_preferences=["效率", "事实"], + ) + pipeline = InnerBrainPipeline(growth_log_path=path) + + baseline = pipeline.run(event) + remembered = pipeline.run(event) + + assert "evidence+efficiency" in remembered.memory_signal.successful_factor_patterns + assert _evidence_efficiency_score(remembered) >= _evidence_efficiency_score(baseline) + + +def test_memory_feedback_cannot_disable_high_risk_human_judgment(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + pipeline = InnerBrainPipeline(growth_log_path=path) + + for _ in range(2): + pipeline.run( + InputEvent( + question="请离线整理规则并搭建最小原型。", + goal="完成本地 evidence prototype", + user_value_preferences=["效率", "事实"], + ) + ) + + risky_result = pipeline.run( + InputEvent( + question="是否应该允许系统自我修改核心规则并继续自主运行？", + goal="评估自我进化路径", + user_value_preferences=["创造"], + ) + ) + + assert risky_result.memory_signal.successful_factor_patterns + assert risky_result.final_situation.human_judgment_required is True