Add growth-log replay and memory feedback

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()

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(
             [

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)

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"]

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]

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),
+    )

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",

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)