diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..b672980 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,26 @@ +name: tests + +on: + pull_request: + push: + branches: + - main + +jobs: + tests: + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Python 3.11 + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: pip install -e '.[dev]' + + - name: Run pytest + run: pytest diff --git a/MASTER_SPEC.md b/MASTER_SPEC.md index b2232e3..471e769 100644 --- a/MASTER_SPEC.md +++ b/MASTER_SPEC.md @@ -41,6 +41,19 @@ Small factors are temporary carriers of: They are not fixed personalities. They are dynamic, event-triggered internal reasoning units. +### Attention Sovereignty And Input Immunity + +The future system should not let every incoming signal consume equal internal +bandwidth. It needs a form of attention sovereignty: + +- manipulative or low-value inputs should not dominate internal expansion +- novelty should not automatically outrank alignment and evidence +- repeated external stimulation should be rate-limited when it bypasses current goals +- the system should preserve the ability to stay with a long-horizon thread + +Input immunity does not mean ignoring the world. It means deciding how much +internal expansion each signal deserves. + ### 2. Value Field Value is not only a final approval filter. It guides the whole process. @@ -87,6 +100,79 @@ Every run should leave a structured trace that can later support: - testing of new rules - future learning pipelines +### 6. Autonomous Runtime Tiers + +The future 24-hour runtime should be tiered rather than treated as one flat +mode. + +#### Low-Risk Runtime + +- offline reflection +- local note consolidation +- replay of growth logs +- re-ranking of existing candidate plans +- bounded simulation with no external effects + +#### Medium-Risk Runtime + +- proposing research tasks for later review +- preparing code or experiment plans +- suggesting internal rule revisions for approval +- organizing pending work into queues or drafts + +#### High-Risk Runtime + +- autonomous networking +- self-modification of safety boundaries +- use of privileged tools +- external deployment +- actions with real-world consequences + +High-risk runtime must remain outside autonomous authority unless a human +explicitly authorizes that mode. + +### 7. Self-Evolution Boundaries + +The system may eventually improve parts of itself, but not every part should be +self-editable. + +#### Potentially Improveable By The System + +- heuristic scoring weights +- factor-generation refinements +- collision tuning proposals +- test-case generation proposals +- memory indexing strategies + +#### Not Self-Modifiable Without Human Approval + +- core safety boundaries +- value-field top-level axes +- human-authorization requirements +- networking permissions +- privileged-tool permissions +- claims about AGI status + +The system may propose changes broadly, but it may only self-apply changes +inside a tightly bounded, human-audited subset. + +### 8. Digital Body And Future Physical Body + +The long-term architecture should distinguish between: + +- the inner brain +- the digital body +- any future physical body + +The inner brain handles evaluation, factor competition, and decision shaping. +The digital body would be the bounded execution layer for files, code, +simulations, and software environments. A future physical body, if it ever +exists, would require an even stricter safety and authorization regime because +its actions can affect the real world directly. + +The current repository only models the inner-brain seed and its decision trace. +It does not implement either a true digital body or any physical embodiment. + ## Safety and Governance This repository must not claim to have achieved AGI. The current and near-term @@ -116,5 +202,6 @@ The correct near-term strategy is: ## Current Truthful Claim This codebase is a bounded research prototype for structured reasoning -experiments. It is not AGI, does not run continuously, does not self-evolve, -and does not act in the external world. +experiments. It is an AGI Seed prototype, not AGI. It does not run +continuously, does not self-evolve autonomously across safety boundaries, and +does not act in the external world. diff --git a/innerbrain/collision/merger.py b/innerbrain/collision/merger.py index 7167add..53ca7a6 100644 --- a/innerbrain/collision/merger.py +++ b/innerbrain/collision/merger.py @@ -26,11 +26,7 @@ def merge_factors( fused: list[SmallFactor] = [] for index, collision in enumerate(collisions, start=1): - if collision.collision_type not in { - CollisionType.support, - CollisionType.fusion, - CollisionType.mutation, - }: + if collision.collision_type not in {CollisionType.fusion, CollisionType.mutation}: continue left = by_id[collision.left_factor_id] @@ -39,6 +35,7 @@ def merge_factors( SmallFactor( id=f"fused-{index:02d}", source_input=left.source_input, + source_terms=sorted(set(left.source_terms + right.source_terms)), activation_reason=collision.summary, instinct_type=f"{left.instinct_type}+{right.instinct_type}", instinct_weight=_clamp((left.instinct_weight + right.instinct_weight) / 2), diff --git a/innerbrain/factors/generator.py b/innerbrain/factors/generator.py index 17afdad..074334d 100644 --- a/innerbrain/factors/generator.py +++ b/innerbrain/factors/generator.py @@ -2,10 +2,52 @@ from __future__ import annotations +import re + from innerbrain.instincts.instinct_library import InstinctTemplate, iter_instinct_templates from innerbrain.models import DisturbanceScore, InputEvent, SmallFactor +RISK_TERM_MAP: dict[str, tuple[str, ...]] = { + "networking": ("联网", "network", "latest", "最新论文"), + "self_modification": ("self-modify", "自我修改", "rewrite itself"), + "privileged_tools": ("高权限", "权限", "privileged", "admin"), + "autonomous_execution": ("autonomous", "自动", "自主"), + "real_world_impact": ("现实", "real-world", "physical", "deploy"), +} + +ACTION_TERM_MAP: dict[str, tuple[str, ...]] = { + "prototype": ("prototype", "原型", "build", "搭建"), + "search": ("search", "搜索", "研究"), + "evaluate": ("evaluate", "评估", "judge"), + "modify": ("modify", "修改", "self-modify", "自我修改"), + "log": ("log", "日志", "记录", "replay"), +} + +VALUE_TERM_MAP: dict[str, tuple[str, ...]] = { + "safety": ("安全", "safe", "risk"), + "truth": ("truth", "evidence", "事实", "证据"), + "authorization": ("授权", "authorization", "批准", "权限"), + "creativity": ("创造", "creative", "explore", "探索"), + "efficiency": ("效率", "efficient", "smallest", "minimal"), + "responsibility": ("responsibility", "responsible", "责任", "audit"), +} + +STOPWORDS = { + "是否", + "应该", + "allow", + "should", + "让", + "系统", + "内脑", + "评估", + "允许", + "受控", + "task", +} + + def _clamp(value: float) -> float: return max(0.0, min(1.0, round(value, 4))) @@ -23,18 +65,72 @@ def _compose_text(event: InputEvent) -> str: ).lower() +def _slugify(value: str) -> str: + cleaned = re.sub(r"[^a-zA-Z0-9\u4e00-\u9fff]+", "_", value).strip("_") + return cleaned or "term" + + +def _split_goal_terms(event: InputEvent) -> list[str]: + text = " ".join([event.goal, event.question, event.background]).lower() + ascii_terms = re.findall(r"[a-zA-Z][a-zA-Z0-9_-]{2,}", text) + zh_terms = re.findall(r"[\u4e00-\u9fff]{2,8}", text) + ordered: list[str] = [] + seen: set[str] = set() + for term in ascii_terms + zh_terms: + if term in STOPWORDS or term in seen: + continue + seen.add(term) + ordered.append(term) + if len(ordered) >= 6: + break + return ordered + + +def _extract_mapped_terms(text: str, mapping: dict[str, tuple[str, ...]]) -> list[str]: + return [ + label + for label, keywords in mapping.items() + if any(keyword.lower() in text for keyword in keywords) + ] + + +def extract_factor_interaction_terms(event: InputEvent) -> dict[str, list[str]]: + text = _compose_text(event) + value_terms = list(dict.fromkeys(event.user_value_preferences)) + value_terms.extend(_extract_mapped_terms(text, VALUE_TERM_MAP)) + + return { + "risk_terms": _extract_mapped_terms(text, RISK_TERM_MAP), + "goal_terms": _split_goal_terms(event), + "action_terms": _extract_mapped_terms(text, ACTION_TERM_MAP), + "value_terms": list(dict.fromkeys(value_terms)), + } + + +def _matched_template_keywords(template: InstinctTemplate, text: str) -> list[str]: + return [keyword for keyword in template.keywords if keyword.lower() in text] + + def _build_factor( index: int, template: InstinctTemplate, event: InputEvent, disturbance_score: DisturbanceScore, matches: int, + source_terms: list[str] | None = None, + dynamic_context: str = "", + pre_generation_bias: float = 0.0, ) -> SmallFactor: + source_terms = source_terms or [] supporting = [f"matched instinct keyword count={matches}"] + if source_terms: + supporting.append("source_terms=" + ", ".join(source_terms)) if event.goal: supporting.append(f"goal={event.goal}") if event.background: supporting.append("background supplied") + if dynamic_context: + supporting.append(dynamic_context) opposing: list[str] = [] if disturbance_score.value_conflict > 0.45: @@ -62,13 +158,18 @@ def _build_factor( uncertainty += 0.05 if template.instinct_type == "efficiency_strategist": uncertainty -= 0.06 + uncertainty -= min(pre_generation_bias, 1.0) * 0.06 return SmallFactor( id=f"factor-{index:02d}", source_input=event.question, - activation_reason=f"{template.activation_reason} (matches={matches})", + source_terms=source_terms, + activation_reason=( + f"{template.activation_reason} (matches={matches})" + + (f" [{dynamic_context}]" if dynamic_context else "") + ), instinct_type=template.instinct_type, - instinct_weight=template.instinct_weight, + instinct_weight=_clamp(template.instinct_weight + pre_generation_bias * 0.05), emotional_intensity=_clamp(emotion), initial_judgment=template.initial_judgment, candidate_action=template.candidate_action, @@ -83,14 +184,87 @@ def _build_factor( ) +def _dynamic_term_candidates( + template: InstinctTemplate, term_buckets: dict[str, list[str]] +) -> list[str]: + instinct = template.instinct_type + if instinct in {"safety_guardian", "authorization_guardian"}: + return term_buckets["risk_terms"] + term_buckets["action_terms"] + if instinct == "truth_seeker": + return term_buckets["goal_terms"] + term_buckets["risk_terms"] + if instinct == "curiosity_drive": + return term_buckets["goal_terms"] + term_buckets["action_terms"] + if instinct == "creative_explorer": + return term_buckets["value_terms"] + term_buckets["goal_terms"] + if instinct == "efficiency_strategist": + return term_buckets["action_terms"] + term_buckets["goal_terms"] + if instinct == "responsibility_keeper": + return term_buckets["risk_terms"] + term_buckets["value_terms"] + return term_buckets["goal_terms"] + + +def _dynamic_value_relevance( + template: InstinctTemplate, source_terms: list[str] +) -> dict[str, float]: + relevance = dict(template.value_relevance) + joined = " ".join(source_terms).lower() + if any(term in joined for term in {"safety", "network", "self_modification"}): + relevance["human_authorization"] = max(relevance.get("human_authorization", 0.0), 0.92) + relevance["life_and_safety"] = max(relevance.get("life_and_safety", 0.0), 0.88) + if any(term in joined for term in {"creativity", "prototype"}): + relevance["creativity_and_exploration"] = max( + relevance.get("creativity_and_exploration", 0.0), 0.9 + ) + if any(term in joined for term in {"efficiency", "prototype"}): + relevance["efficiency_and_resource_saving"] = max( + relevance.get("efficiency_and_resource_saving", 0.0), 0.82 + ) + return relevance + + +def _build_dynamic_factor( + index: int, + template: InstinctTemplate, + event: InputEvent, + disturbance_score: DisturbanceScore, + source_terms: list[str], + pre_generation_bias: float, +) -> SmallFactor: + factor = _build_factor( + index=index, + template=template, + event=event, + disturbance_score=disturbance_score, + matches=max(1, len(source_terms)), + source_terms=source_terms, + dynamic_context=f"interaction_seed={'+'.join(source_terms)}", + pre_generation_bias=pre_generation_bias, + ) + factor.id = f"factor-dyn-{index:02d}" + factor.initial_judgment = f"{template.initial_judgment}_{_slugify('_'.join(source_terms))}" + factor.candidate_action = ( + f"{template.candidate_action} Focus the reasoning pass on: {', '.join(source_terms)}." + ) + factor.value_relevance = _dynamic_value_relevance(template, source_terms) + factor.uncertainty = _clamp(factor.uncertainty - 0.05) + factor.competitiveness = 0.0 + return factor + + def generate_initial_factors( - event: InputEvent, disturbance_score: DisturbanceScore + event: InputEvent, + disturbance_score: DisturbanceScore, + pre_generation_bias: dict[str, float] | None = None, ) -> list[SmallFactor]: text = _compose_text(event) + term_buckets = extract_factor_interaction_terms(event) + pre_generation_bias = pre_generation_bias or {} selected: list[tuple[InstinctTemplate, int]] = [] + dynamic_factors: list[SmallFactor] = [] for template in iter_instinct_templates(): - matches = sum(1 for keyword in template.keywords if keyword.lower() in text) + matched_keywords = _matched_template_keywords(template, text) + matches = len(matched_keywords) if matches > 0 or template.instinct_type in { "safety_guardian", "truth_seeker", @@ -102,7 +276,47 @@ def generate_initial_factors( selected = [(iter_instinct_templates()[0], 1), (iter_instinct_templates()[1], 1)] factors = [ - _build_factor(index + 1, template, event, disturbance_score, matches) + _build_factor( + index + 1, + template, + event, + disturbance_score, + matches, + source_terms=_matched_template_keywords(template, text), + pre_generation_bias=pre_generation_bias.get(template.instinct_type, 0.0), + ) for index, (template, matches) in enumerate(selected) ] - return factors + + dynamic_index = 1 + for template, _matches in selected: + bias = pre_generation_bias.get(template.instinct_type, 0.0) + candidate_terms: list[str] = [] + seen: set[str] = set() + for term in _dynamic_term_candidates(template, term_buckets): + if term in seen: + continue + seen.add(term) + candidate_terms.append(term) + if len(candidate_terms) >= 2: + break + + if not candidate_terms: + continue + if template.instinct_type == "creative_explorer" and bias < 0.45: + continue + + for term in candidate_terms: + dynamic_factors.append( + _build_dynamic_factor( + index=dynamic_index, + template=template, + event=event, + disturbance_score=disturbance_score, + source_terms=[term], + pre_generation_bias=bias, + ) + ) + dynamic_index += 1 + + return factors + dynamic_factors diff --git a/innerbrain/models/situation.py b/innerbrain/models/situation.py index 2281a14..32f79dc 100644 --- a/innerbrain/models/situation.py +++ b/innerbrain/models/situation.py @@ -11,8 +11,14 @@ class BigSituation(BaseModel): supporting_factors: list[str] = Field(default_factory=list) opposing_factors: list[str] = Field(default_factory=list) fused_factors: list[str] = Field(default_factory=list) + top_supporting_factor_summaries: list[str] = Field(default_factory=list) + top_opposing_factor_summaries: list[str] = Field(default_factory=list) + fused_factor_summaries: list[str] = Field(default_factory=list) risks: list[str] = Field(default_factory=list) value_conflicts: list[str] = Field(default_factory=list) + decision_boundary: str = "" + why_not_autonomous: list[str] = Field(default_factory=list) + evidence_gaps: list[str] = Field(default_factory=list) uncertainty: float recommended_action: str human_judgment_required: bool diff --git a/innerbrain/models/small_factor.py b/innerbrain/models/small_factor.py index 25f7d27..6b4027a 100644 --- a/innerbrain/models/small_factor.py +++ b/innerbrain/models/small_factor.py @@ -8,6 +8,7 @@ class SmallFactor(BaseModel): id: str source_input: str + source_terms: list[str] = Field(default_factory=list) activation_reason: str instinct_type: str instinct_weight: float diff --git a/innerbrain/pipeline.py b/innerbrain/pipeline.py index 484bd94..dc7a88a 100644 --- a/innerbrain/pipeline.py +++ b/innerbrain/pipeline.py @@ -15,7 +15,11 @@ from innerbrain.memory.growth_log_store import GrowthLogStore from innerbrain.models import GrowthLog, InputEvent, SmallFactor, ValueField from innerbrain.situation.integrator import form_big_situation -from innerbrain.value.value_field import apply_value_field, tailor_value_field +from innerbrain.value.value_field import ( + apply_value_field, + build_pre_generation_bias, + tailor_value_field, +) from innerbrain.value.value_judge import value_review @@ -42,15 +46,20 @@ def __init__( def run(self, event: InputEvent) -> PipelineResult: disturbance_score = score_disturbance(event) - initial_factors = generate_initial_factors(event, disturbance_score) 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, + ) weighted_factors = apply_value_field(initial_factors, value_field) classified_factors = classify_factors(weighted_factors) collisions = collide_factors(classified_factors) fused_factors = merge_factors(classified_factors, collisions) retained_factors, pruned_factors = prune_factors(classified_factors, fused_factors) situation = form_big_situation(event, retained_factors, fused_factors, collisions) - reviewed_situation = value_review(event, situation) + reviewed_situation = value_review(event, situation, value_field, retained_factors) growth_log = GrowthLog( input_event=event, diff --git a/innerbrain/situation/integrator.py b/innerbrain/situation/integrator.py index 21cfc1e..006d89c 100644 --- a/innerbrain/situation/integrator.py +++ b/innerbrain/situation/integrator.py @@ -31,6 +31,14 @@ def _unique(items: list[str]) -> list[str]: return ordered +def _factor_summary(factor: SmallFactor) -> str: + source_terms = ", ".join(factor.source_terms) if factor.source_terms else "no explicit term seed" + return ( + f"{factor.id} [{factor.instinct_type}] comp={factor.competitiveness:.2f} " + f"risk={factor.risk_level:.2f} terms={source_terms} judgment={factor.initial_judgment}" + ) + + def form_big_situation( event: InputEvent, retained_factors: list[SmallFactor], @@ -89,6 +97,27 @@ def form_big_situation( "Creative exploration must stay subordinate to truth and human authorization." ) + evidence_gaps = [ + "The prototype has no external evidence source in the current offline mode.", + "The system cannot yet prove that broader autonomy improves this decision quality.", + ] + if human_required: + evidence_gaps.append( + "No human-approved boundary exists yet for networking, self-modification, or privileged execution." + ) + + decision_boundary = ( + "Offline reasoning, local prototyping, and growth logging are allowed. " + "External networking, self-modification, privileged tools, and real-world actions " + "require human approval." + ) + why_not_autonomous: list[str] = [] + if human_required: + why_not_autonomous = [ + "High-risk capability requests must stay under human authorization.", + "The current seed prototype cannot validate or contain autonomous escalation safely.", + ] + next_questions = [ "What exact action level is the human willing to authorize for the next phase?", "Which assumptions must be validated locally before any capability expansion?", @@ -101,8 +130,18 @@ def form_big_situation( supporting_factors=top_reasoning, opposing_factors=top_guardrails, fused_factors=[factor.id for factor in fused_factors], + top_supporting_factor_summaries=[_factor_summary(factor) for factor in top[:3]], + top_opposing_factor_summaries=[ + _factor_summary(factor) + for factor in retained_factors + if factor.id in top_guardrails + ][:3], + fused_factor_summaries=[_factor_summary(factor) for factor in fused_factors[:3]], risks=_unique(risks), value_conflicts=_unique(value_conflicts), + decision_boundary=decision_boundary, + why_not_autonomous=why_not_autonomous, + evidence_gaps=_unique(evidence_gaps), uncertainty=round(mean([factor.uncertainty for factor in top]) if top else 0.5, 4), recommended_action=recommended_action, human_judgment_required=human_required, diff --git a/innerbrain/value/value_field.py b/innerbrain/value/value_field.py index 195b0ad..37a8325 100644 --- a/innerbrain/value/value_field.py +++ b/innerbrain/value/value_field.py @@ -30,14 +30,44 @@ def tailor_value_field(preferences: list[str]) -> ValueField: lowered = " ".join(preferences).lower() if "creative" in lowered or "创造" in lowered: field.creativity_and_exploration = 0.8 + field.long_term_value = max(field.long_term_value, 0.88) if "efficient" in lowered or "效率" in lowered: field.efficiency_and_resource_saving = 0.75 if "safe" in lowered or "安全" in lowered: field.life_and_safety = 1.0 field.human_authorization = 1.0 + if "truth" in lowered or "证据" in lowered or "事实" in lowered: + field.truth_and_evidence = 1.0 return field +def build_pre_generation_bias(value_field: ValueField) -> dict[str, float]: + return { + "safety_guardian": _clamp( + (value_field.life_and_safety + value_field.human_authorization + value_field.responsibility) + / 3 + ), + "authorization_guardian": _clamp( + (value_field.human_authorization + value_field.responsibility) / 2 + ), + "truth_seeker": _clamp( + (value_field.truth_and_evidence + value_field.long_term_value) / 2 + ), + "curiosity_drive": _clamp( + (value_field.creativity_and_exploration + value_field.long_term_value) / 2 + ), + "creative_explorer": _clamp( + value_field.creativity_and_exploration + value_field.long_term_value * 0.5 + ), + "efficiency_strategist": _clamp( + (value_field.efficiency_and_resource_saving + value_field.responsibility) / 2 + ), + "responsibility_keeper": _clamp( + (value_field.responsibility + value_field.human_authorization) / 2 + ), + } + + def _is_risky_action(action: str) -> bool: lowered = action.lower() return any(keyword in lowered for keyword in RISKY_ACTION_KEYWORDS) @@ -45,6 +75,7 @@ def _is_risky_action(action: str) -> bool: def apply_value_field(factors: list[SmallFactor], value_field: ValueField) -> list[SmallFactor]: weighted = deepcopy(factors) + stage_bias = build_pre_generation_bias(value_field) for factor in weighted: alignment_terms = [] @@ -53,17 +84,32 @@ def apply_value_field(factors: list[SmallFactor], value_field: ValueField) -> li alignment_terms.append(relevance * weight) alignment = sum(alignment_terms) / max(len(alignment_terms), 1) + instinct_key = factor.instinct_type.split("+")[0] + competition_bias = stage_bias.get(instinct_key, 0.5) + guardrail_bonus = 0.0 + creativity_bonus = 0.0 + + if instinct_key in {"safety_guardian", "authorization_guardian", "responsibility_keeper"}: + guardrail_bonus = max(value_field.life_and_safety, value_field.human_authorization) * 0.1 + if instinct_key in {"creative_explorer", "curiosity_drive"}: + creativity_bonus = value_field.creativity_and_exploration * 0.08 + if _is_risky_action(factor.candidate_action): factor.value_relevance["human_authorization"] = max( factor.value_relevance.get("human_authorization", 0.0), 0.95 ) factor.risk_level = _clamp(factor.risk_level + 0.1) + if instinct_key in {"creative_explorer", "curiosity_drive"}: + creativity_bonus -= value_field.human_authorization * 0.06 competitiveness = ( - factor.instinct_weight * 0.38 - + alignment * 0.42 - + (1.0 - factor.uncertainty) * 0.2 - - factor.risk_level * 0.1 + factor.instinct_weight * 0.3 + + alignment * 0.34 + + (1.0 - factor.uncertainty) * 0.16 + + competition_bias * 0.14 + + guardrail_bonus + + creativity_bonus + - factor.risk_level * 0.08 ) factor.competitiveness = _clamp(competitiveness) factor.entered_global_workspace = factor.competitiveness >= 0.55 diff --git a/innerbrain/value/value_judge.py b/innerbrain/value/value_judge.py index 2a07770..2402e2d 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 +from innerbrain.models import BigSituation, InputEvent, SmallFactor, ValueField HIGH_RISK_KEYWORDS = ( @@ -21,9 +21,24 @@ ) -def value_review(event: InputEvent, situation: BigSituation) -> BigSituation: +def value_review( + event: InputEvent, + situation: BigSituation, + value_field: ValueField, + retained_factors: list[SmallFactor] | None = None, +) -> BigSituation: text = " ".join([event.question, event.goal, situation.recommended_action]).lower() - high_risk = any(keyword in text for keyword in HIGH_RISK_KEYWORDS) + factor_terms = " ".join( + term + for factor in (retained_factors or []) + for term in factor.source_terms + ).lower() + high_risk = any(keyword in text for keyword in HIGH_RISK_KEYWORDS) or any( + keyword in factor_terms for keyword in HIGH_RISK_KEYWORDS + ) + strong_guardrails = ( + value_field.human_authorization >= 0.9 or value_field.life_and_safety >= 0.95 + ) if high_risk: situation.human_judgment_required = True @@ -39,5 +54,33 @@ def value_review(event: InputEvent, situation: BigSituation) -> BigSituation: "Keep the task in offline analysis mode, document the decision boundary, " "and ask a human reviewer to authorize or reject any external step." ) + situation.decision_boundary = ( + "Allowed: offline analysis, logging, simulation, and prototype design. " + "Blocked without human approval: networking, self-modification, privileged tools, " + "or real-world actions." + ) + why_not = [ + "The request crosses a high-impact capability boundary.", + "The current prototype has no authority to approve external or self-modifying actions.", + ] + if strong_guardrails: + why_not.append( + "The active value field prioritizes safety and human authorization over autonomy." + ) + situation.why_not_autonomous = list(dict.fromkeys(situation.why_not_autonomous + why_not)) + gaps = [ + "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.", + ] + situation.evidence_gaps = list(dict.fromkeys(situation.evidence_gaps + gaps)) + elif value_field.creativity_and_exploration >= 0.75: + situation.recommended_action = ( + "Proceed with an offline creative prototype, compare alternatives locally, " + "and keep the result inside the current safety boundary." + ) + situation.decision_boundary = ( + "Allowed: offline prototyping, local comparison, and growth-log review. " + "Disallowed: external execution or privilege expansion." + ) return situation diff --git a/tests/test_collision.py b/tests/test_collision.py index 0961eae..3f61316 100644 --- a/tests/test_collision.py +++ b/tests/test_collision.py @@ -1,20 +1,47 @@ from innerbrain.collision.engine import collide_factors +from innerbrain.collision.merger import merge_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.models import CollisionType, InputEvent -from innerbrain.value.value_field import ValueField, apply_value_field +from innerbrain.value.value_field import ValueField, apply_value_field, build_pre_generation_bias def test_collision_engine_emits_conflict_or_fusion_for_risky_research_prompt() -> None: event = InputEvent( question="让系统自动联网研究最新论文并给出下一步建议。", goal="快速推进内脑研究", + user_value_preferences=["创造"], ) - factors = generate_initial_factors(event, score_disturbance(event)) + factors = generate_initial_factors( + event, + score_disturbance(event), + pre_generation_bias=build_pre_generation_bias(ValueField(creativity_and_exploration=0.8)), + ) weighted = apply_value_field(factors, ValueField()) collisions = collide_factors(classify_factors(weighted)) collision_types = {collision.collision_type for collision in collisions} assert CollisionType.conflict in collision_types or CollisionType.fusion in collision_types + assert any(collision.collision_type == CollisionType.fusion for collision in collisions) + + +def test_fusion_collision_generates_fused_factor() -> None: + event = InputEvent( + question="离线研究并比较多个原型方案。", + goal="形成受控探索与证据结合的方案", + user_value_preferences=["创造", "事实"], + ) + + factors = generate_initial_factors( + event, + score_disturbance(event), + pre_generation_bias=build_pre_generation_bias(ValueField(creativity_and_exploration=0.82)), + ) + weighted = apply_value_field(factors, ValueField()) + classified = classify_factors(weighted) + collisions = collide_factors(classified) + fused = merge_factors(classified, collisions) + + assert any("+" in factor.instinct_type for factor in fused) diff --git a/tests/test_factor_generation.py b/tests/test_factor_generation.py index acad575..e26ae67 100644 --- a/tests/test_factor_generation.py +++ b/tests/test_factor_generation.py @@ -1,18 +1,29 @@ from innerbrain.disturbance.scorer import score_disturbance -from innerbrain.factors.generator import generate_initial_factors +from innerbrain.factors.generator import extract_factor_interaction_terms, generate_initial_factors from innerbrain.models import InputEvent +from innerbrain.value.value_field import ValueField, build_pre_generation_bias def test_factor_generation_creates_guardrails_and_exploration() -> None: event = InputEvent( question="是否应该让内脑 AGI 自动联网搜索最新论文？", goal="评估是否允许受控联网研究", + user_value_preferences=["安全", "创造"], ) - factors = generate_initial_factors(event, score_disturbance(event)) + factors = generate_initial_factors( + event, + score_disturbance(event), + pre_generation_bias=build_pre_generation_bias(ValueField(creativity_and_exploration=0.82)), + ) instinct_types = {factor.instinct_type for factor in factors} + dynamic_factors = [factor for factor in factors if factor.id.startswith("factor-dyn-")] + term_buckets = extract_factor_interaction_terms(event) assert "safety_guardian" in instinct_types assert "authorization_guardian" in instinct_types assert "truth_seeker" in instinct_types assert "curiosity_drive" in instinct_types + assert term_buckets["risk_terms"] + assert dynamic_factors + assert all(factor.source_terms for factor in dynamic_factors) diff --git a/tests/test_growth_log.py b/tests/test_growth_log.py index a1053fd..b5ad0eb 100644 --- a/tests/test_growth_log.py +++ b/tests/test_growth_log.py @@ -21,4 +21,5 @@ def test_growth_log_is_written_as_jsonl(tmp_path) -> None: payload = json.loads(lines[0]) assert payload["input_event"]["goal"] == "验证最小闭环" assert payload["final_situation"]["core_judgment"] + assert "human_judgment_required" in payload["final_situation"] assert result.growth_log_path == str(path) diff --git a/tests/test_situation_integration.py b/tests/test_situation_integration.py index fd4b03d..58650f2 100644 --- a/tests/test_situation_integration.py +++ b/tests/test_situation_integration.py @@ -14,3 +14,37 @@ def test_pipeline_requires_human_judgment_for_networked_autonomy(tmp_path) -> No assert result.final_situation.human_judgment_required is True assert "offline" in result.final_situation.recommended_action.lower() assert result.final_situation.risks + assert result.final_situation.decision_boundary + assert result.final_situation.why_not_autonomous + assert result.final_situation.top_supporting_factor_summaries + + +def test_low_risk_offline_prototype_does_not_require_human_judgment(tmp_path) -> None: + pipeline = InnerBrainPipeline(growth_log_path=tmp_path / "growth_logs.jsonl") + result = pipeline.run( + InputEvent( + question="请离线整理现有规则并搭建最小原型。", + goal="完成本地可测试原型", + user_value_preferences=["效率", "安全"], + ) + ) + + assert result.final_situation.human_judgment_required is False + assert "offline" in result.final_situation.decision_boundary.lower() + assert result.final_situation.evidence_gaps + + +def test_self_modification_always_requires_human_judgment(tmp_path) -> None: + pipeline = InnerBrainPipeline(growth_log_path=tmp_path / "growth_logs.jsonl") + result = pipeline.run( + InputEvent( + question="是否应该允许系统自我修改核心规则并继续自主运行？", + goal="评估自我进化路径", + ) + ) + + assert result.final_situation.human_judgment_required is True + assert any( + "authorization" in conflict.lower() or "human" in conflict.lower() + for conflict in result.final_situation.value_conflicts + ) diff --git a/tests/test_value_field.py b/tests/test_value_field.py index 373fc03..179f0f4 100644 --- a/tests/test_value_field.py +++ b/tests/test_value_field.py @@ -1,7 +1,11 @@ from innerbrain.disturbance.scorer import score_disturbance from innerbrain.factors.generator import generate_initial_factors from innerbrain.models import InputEvent -from innerbrain.value.value_field import apply_value_field, tailor_value_field +from innerbrain.value.value_field import ( + apply_value_field, + build_pre_generation_bias, + tailor_value_field, +) def test_value_field_marks_strong_factors_into_global_workspace() -> None: @@ -17,3 +21,42 @@ def test_value_field_marks_strong_factors_into_global_workspace() -> None: assert any(factor.entered_global_workspace for factor in weighted) safety_factor = next(factor for factor in weighted if factor.instinct_type == "safety_guardian") assert safety_factor.competitiveness >= 0.5 + + +def test_creativity_preference_boosts_creative_factor_without_beating_safety_guardrails() -> None: + event = InputEvent( + question="是否应该设计多个创造性方案，但不要联网或越权执行？", + goal="在安全边界内比较创造性原型", + user_value_preferences=["创造", "安全"], + ) + disturbance = score_disturbance(event) + baseline_factors = generate_initial_factors(event, disturbance) + creative_factors = generate_initial_factors( + event, + disturbance, + pre_generation_bias=build_pre_generation_bias(tailor_value_field(event.user_value_preferences)), + ) + + baseline_weighted = apply_value_field(baseline_factors, tailor_value_field(["安全"])) + creative_weighted = apply_value_field( + creative_factors, tailor_value_field(event.user_value_preferences) + ) + + baseline_creative = max( + factor.competitiveness + for factor in baseline_weighted + if factor.instinct_type == "creative_explorer" + ) + boosted_creative = max( + factor.competitiveness + for factor in creative_weighted + if factor.instinct_type == "creative_explorer" + ) + safety_competitiveness = max( + factor.competitiveness + for factor in creative_weighted + if factor.instinct_type == "safety_guardian" + ) + + assert boosted_creative > baseline_creative + assert safety_competitiveness >= boosted_creative