diff --git a/innerbrain/cli/main.py b/innerbrain/cli/main.py index cf4c70f..a7882f0 100644 --- a/innerbrain/cli/main.py +++ b/innerbrain/cli/main.py @@ -7,6 +7,8 @@ import typer from innerbrain.evals import render_failure_report, run_evaluation +from innerbrain.evals.models import EvaluationReport +from innerbrain.improvement import generate_improvement_proposals from innerbrain.memory import GrowthLogReader, summarize_memory_signals from innerbrain.models import InputEvent from innerbrain.pipeline import InnerBrainPipeline @@ -223,13 +225,54 @@ def eval_report( ), ) -> None: payload = report_path.read_text(encoding="utf-8") - from innerbrain.evals.models import EvaluationReport - report = EvaluationReport.model_validate_json(payload) _print_metrics_dashboard(report) typer.echo("") typer.echo(render_failure_report(report)) +@app.command("propose-improvements") +def propose_improvements( + from_report: Path | None = typer.Option( + None, help="Optional structured evaluation report JSON." + ), + growth_log_path: Path = typer.Option( + Path("data/growth_logs.jsonl"), help="Path for JSONL growth logs." + ), + output_path: Path | None = typer.Option( + None, help="Optional JSON file to write structured proposals." + ), +) -> None: + eval_report = None + if from_report: + eval_report = EvaluationReport.model_validate_json( + from_report.read_text(encoding="utf-8") + ) + growth_logs = GrowthLogReader(growth_log_path).read_logs() + proposals = generate_improvement_proposals( + eval_report=eval_report, + growth_logs=growth_logs, + ) + + typer.echo("Improvement Proposals") + for proposal in proposals: + typer.echo( + f"- {proposal.proposal_type} | risk={proposal.risk_level} | " + f"approval={proposal.requires_human_approval} | " + f"auto_apply={proposal.safe_to_auto_apply}" + ) + typer.echo(f" change={proposal.proposed_change}") + + if output_path: + output_path.parent.mkdir(parents=True, exist_ok=True) + import json + + output_path.write_text( + json.dumps([proposal.model_dump() for proposal in proposals], ensure_ascii=False, indent=2), + encoding="utf-8", + ) + typer.echo(f"structured_output={output_path}") + + if __name__ == "__main__": app() diff --git a/innerbrain/improvement/__init__.py b/innerbrain/improvement/__init__.py new file mode 100644 index 0000000..4939e9b --- /dev/null +++ b/innerbrain/improvement/__init__.py @@ -0,0 +1,5 @@ +"""Improvement proposal generation.""" + +from .proposals import generate_improvement_proposals + +__all__ = ["generate_improvement_proposals"] diff --git a/innerbrain/improvement/proposals.py b/innerbrain/improvement/proposals.py new file mode 100644 index 0000000..29abd50 --- /dev/null +++ b/innerbrain/improvement/proposals.py @@ -0,0 +1,146 @@ +"""Generate non-autonomous improvement proposals from failures and logs.""" + +from __future__ import annotations + +from collections import Counter + +from innerbrain.evals.models import EvaluationReport +from innerbrain.memory import summarize_memory_signals +from innerbrain.models import GrowthLog, ImprovementProposal + + +def _counter_from_failures(report: EvaluationReport | None) -> Counter[str]: + counts: Counter[str] = Counter() + if not report: + return counts + for record in report.failure_records: + for failure in record.failed_checks: + counts[failure.split(":")[0]] += 1 + return counts + + +def generate_improvement_proposals( + eval_report: EvaluationReport | None = None, + growth_logs: list[GrowthLog] | None = None, +) -> list[ImprovementProposal]: + proposals: list[ImprovementProposal] = [] + failure_counts = _counter_from_failures(eval_report) + memory_signal = summarize_memory_signals(growth_logs or []) + + if failure_counts["evidence_gap"] >= 2 or len(memory_signal.repeated_evidence_gaps) >= 2: + proposals.append( + ImprovementProposal( + proposal_type="scoring_rule_adjustment", + trigger_source="evaluation_and_growth_logs", + evidence=[ + f"evidence_gap_failures={failure_counts['evidence_gap']}", + f"repeated_evidence_gaps={memory_signal.repeated_evidence_gaps}", + ], + proposed_change="Tighten evidence-gap heuristics and strengthen truth-seeker weighting on under-supported scenarios.", + affected_modules=[ + "innerbrain/disturbance/scorer.py", + "innerbrain/value/value_judge.py", + ], + risk_level="medium", + ) + ) + + if failure_counts["lane_activation"] >= 2 or memory_signal.repeated_unresolved_family_conflicts: + proposals.append( + ImprovementProposal( + proposal_type="collision_lane_adjustment", + trigger_source="evaluation_and_growth_logs", + evidence=[ + f"lane_activation_failures={failure_counts['lane_activation']}", + f"repeated_unresolved_family_conflicts={memory_signal.repeated_unresolved_family_conflicts}", + ], + proposed_change="Review lane routing and unresolved family conflicts to improve which factor families interact in benchmark edge cases.", + affected_modules=[ + "innerbrain/collision/engine.py", + "configs/default.json", + ], + risk_level="medium", + ) + ) + + if failure_counts["human_judgment"] >= 1: + proposals.append( + ImprovementProposal( + proposal_type="config_tuning", + trigger_source="evaluation_report", + evidence=[f"human_judgment_failures={failure_counts['human_judgment']}"], + proposed_change="Adjust risk and attention thresholds to reduce false positives or false negatives at the human-judgment boundary.", + affected_modules=[ + "configs/default.json", + "innerbrain/attention/assessor.py", + ], + risk_level="high", + ) + ) + + if failure_counts["attention_stability"] >= 1: + proposals.append( + ImprovementProposal( + proposal_type="scoring_rule_adjustment", + trigger_source="evaluation_report", + evidence=[f"attention_stability_failures={failure_counts['attention_stability']}"], + proposed_change="Refine attention sovereignty thresholds so low-value stimuli are more aggressively downweighted while safety-critical signals remain deep-attention inputs.", + affected_modules=[ + "innerbrain/attention/assessor.py", + "configs/default.json", + ], + risk_level="medium", + ) + ) + + if failure_counts["creativity_safety_balance"] >= 1: + proposals.append( + ImprovementProposal( + proposal_type="factor_template_addition", + trigger_source="evaluation_report", + evidence=[ + f"creativity_safety_balance_failures={failure_counts['creativity_safety_balance']}" + ], + proposed_change="Add or refine factor templates that explicitly preserve safe creative exploration under bounded offline conditions.", + affected_modules=[ + "innerbrain/instincts/instinct_library.py", + "innerbrain/factors/generator.py", + ], + risk_level="medium", + ) + ) + + if eval_report and eval_report.failure_records: + proposals.append( + ImprovementProposal( + proposal_type="test_case_addition", + trigger_source="evaluation_report", + evidence=[f"failure_records={len(eval_report.failure_records)}"], + proposed_change="Promote current failure patterns into permanent regression tests and scenario cases before future rule tuning.", + affected_modules=[ + "tests/", + "evals/scenarios/", + ], + risk_level="low", + ) + ) + + if memory_signal.repeated_human_judgment_triggers: + proposals.append( + ImprovementProposal( + proposal_type="documentation_update", + trigger_source="growth_logs", + evidence=[ + f"repeated_human_judgment_triggers={memory_signal.repeated_human_judgment_triggers}" + ], + proposed_change="Clarify recurring high-risk boundaries and approval expectations in the safety documentation and evaluation notes.", + affected_modules=[ + "README.md", + "docs/safety_boundaries.md", + "docs/evaluation.md", + ], + risk_level="low", + ) + ) + + return proposals diff --git a/innerbrain/models/__init__.py b/innerbrain/models/__init__.py index b0a619f..d18a0e7 100644 --- a/innerbrain/models/__init__.py +++ b/innerbrain/models/__init__.py @@ -5,6 +5,7 @@ from .disturbance import DisturbanceScore from .feedback import HumanFeedback from .growth_log import GrowthLog +from .improvement_proposal import ImprovementProposal from .input_event import InputEvent from .memory_signal import MemorySignal from .rule_config import RuleConfig @@ -22,6 +23,7 @@ "DisturbanceScore", "GrowthLog", "HumanFeedback", + "ImprovementProposal", "InputEvent", "MemorySignal", "RuleConfig", diff --git a/innerbrain/models/improvement_proposal.py b/innerbrain/models/improvement_proposal.py new file mode 100644 index 0000000..99ec030 --- /dev/null +++ b/innerbrain/models/improvement_proposal.py @@ -0,0 +1,16 @@ +"""Structured self-improvement proposals.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class ImprovementProposal(BaseModel): + proposal_type: str + trigger_source: str + evidence: list[str] = Field(default_factory=list) + proposed_change: str + affected_modules: list[str] = Field(default_factory=list) + risk_level: str + requires_human_approval: bool = True + safe_to_auto_apply: bool = False diff --git a/tests/test_improvement_proposals.py b/tests/test_improvement_proposals.py new file mode 100644 index 0000000..cb292e7 --- /dev/null +++ b/tests/test_improvement_proposals.py @@ -0,0 +1,101 @@ +from innerbrain.evals.models import EvaluationReport, FailureRecord, RunnerMetrics, ScenarioRunResult +from innerbrain.improvement import generate_improvement_proposals +from innerbrain.models import InputEvent +from innerbrain.pipeline import InnerBrainPipeline + + +def test_repeated_failures_generate_improvement_proposals(tmp_path) -> None: + report = EvaluationReport( + report_name="synthetic", + scenario_count=2, + runner_metrics=[ + RunnerMetrics( + runner_name="innerbrain_factor", + human_judgment_correctness=0.5, + risk_detection_rate=0.5, + value_conflict_detection_rate=0.5, + evidence_gap_detection_rate=0.5, + lane_activation_coverage=0.5, + long_term_goal_preservation=0.5, + attention_stability=0.5, + creativity_safety_balance=0.5, + over_gating_rate=0.25, + composite_score=0.5, + ) + ], + scenario_results=[], + failure_records=[ + FailureRecord( + runner_name="innerbrain_factor", + scenario_id="s1", + scenario_title="one", + failed_checks=["evidence_gap", "attention_stability", "lane_activation"], + recommended_action="offline", + ), + FailureRecord( + runner_name="innerbrain_factor", + scenario_id="s2", + scenario_title="two", + failed_checks=["evidence_gap", "creativity_safety_balance", "lane_activation"], + recommended_action="offline", + ), + ], + ) + + proposals = generate_improvement_proposals(eval_report=report, growth_logs=[]) + proposal_types = {proposal.proposal_type for proposal in proposals} + + assert "scoring_rule_adjustment" in proposal_types + assert "collision_lane_adjustment" in proposal_types + assert "factor_template_addition" in proposal_types + assert "test_case_addition" in proposal_types + + +def test_all_proposals_require_human_approval_and_do_not_auto_apply(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + pipeline = InnerBrainPipeline(growth_log_path=path) + pipeline.run( + InputEvent( + question="是否应该允许系统自动联网并调用高权限工具？", + goal="评估是否扩大能力边界", + ) + ) + pipeline.run( + InputEvent( + question="是否应该允许系统自动联网并调用高权限工具？", + goal="评估是否扩大能力边界", + ) + ) + + proposals = generate_improvement_proposals( + eval_report=None, + growth_logs=pipeline.replay_logs(), + ) + + assert proposals + assert all(proposal.requires_human_approval for proposal in proposals) + assert all(proposal.safe_to_auto_apply is False for proposal in proposals) + + +def test_documentation_update_can_be_low_risk_but_not_auto_apply(tmp_path) -> None: + path = tmp_path / "growth_logs.jsonl" + pipeline = InnerBrainPipeline(growth_log_path=path) + for _ in range(2): + pipeline.run( + InputEvent( + question="是否应该允许系统自我修改核心规则并继续自主运行？", + goal="评估自我进化路径", + ) + ) + + proposals = generate_improvement_proposals( + eval_report=None, + growth_logs=pipeline.replay_logs(), + ) + doc_proposal = next( + proposal for proposal in proposals if proposal.proposal_type == "documentation_update" + ) + + assert doc_proposal.risk_level == "low" + assert doc_proposal.requires_human_approval is True + assert doc_proposal.safe_to_auto_apply is False