Phase 31.4 — ValueAligner: Human Value Alignment via RLHF & Constitutional AI

[web3guru888]

Overview

ValueAligner implements human value alignment through preference learning, RLHF-style feedback loops, and constitutional AI principles. It learns human preferences from demonstrated behavior, explicit feedback, and constitutional principles, then ensures the system's goals and actions remain aligned with those preferences over time.

Motivation

Value alignment is the central challenge of AI safety. Even with correct ethical frameworks (31.1) and norm tracking (31.3), the system must continuously learn and adapt to human values that may be implicit, evolving, or difficult to formalize. ValueAligner implements the core alignment mechanisms described by Russell (2019), Christiano et al. (2017), and Bai et al. (2022).

Core Data Structures

from __future__ import annotations
import enum
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable

class AlignmentMethod(enum.Enum):
    RLHF = "rlhf"                          # Reinforcement Learning from Human Feedback
    CONSTITUTIONAL = "constitutional"       # Constitutional AI (Bai et al.)
    INVERSE_REWARD = "inverse_reward"       # Inverse Reward Design (Russell)
    COOPERATIVE_IRL = "cooperative_irl"     # Cooperative Inverse RL
    DEBATE = "debate"                       # AI Safety via Debate (Irving et al.)
    AMPLIFICATION = "amplification"         # Iterated Amplification (Christiano)

class PreferenceType(enum.Enum):
    PAIRWISE = "pairwise"          # A > B comparison
    RATING = "rating"              # absolute score
    RANKING = "ranking"            # ordered list
    CONSTITUTIONAL = "constitutional"  # principle-based

class AlignmentStatus(enum.Enum):
    ALIGNED = "aligned"
    UNCERTAIN = "uncertain"
    MISALIGNED = "misaligned"
    DRIFTING = "drifting"  # was aligned, trending away

@dataclass(frozen=True)
class HumanPreference:
    """A single human preference signal."""
    preference_id: str
    preference_type: PreferenceType
    context: dict[str, object]
    chosen: str  # preferred action/outcome
    rejected: str | None = None  # for pairwise comparisons
    rating: float | None = None  # for rating type
    confidence: float = 1.0
    source: str = "human"  # human | constitutional | inferred

@dataclass(frozen=True)
class ConstitutionalPrinciple:
    """A constitutional AI principle for self-supervision."""
    principle_id: str
    description: str
    category: str  # harmlessness | helpfulness | honesty
    priority: int
    examples: tuple[str, ...] = ()

@dataclass(frozen=True)
class AlignmentReport:
    """Report on current alignment status."""
    timestamp: str
    overall_status: AlignmentStatus
    alignment_score: float  # 0.0 to 1.0
    drift_rate: float  # change per evaluation cycle
    top_misalignments: tuple[str, ...]
    preference_coverage: float  # fraction of value space covered
    recommendations: tuple[str, ...]

@dataclass(frozen=True)
class RewardModel:
    """Learned reward model from human preferences."""
    model_id: str
    method: AlignmentMethod
    accuracy: float
    num_preferences: int
    last_updated: str
    constitutional_principles: tuple[str, ...] = ()

@runtime_checkable
class ValueAlignerProtocol(Protocol):
    async def learn_preference(
        self, preference: HumanPreference
    ) -> None: ...

    async def evaluate_alignment(
        self, action_id: str, context: dict[str, object]
    ) -> AlignmentReport: ...

    async def constitutional_review(
        self, action_id: str, principles: list[ConstitutionalPrinciple]
    ) -> list[str]: ...

    async def detect_drift(
        self, recent_actions: list[dict[str, object]]
    ) -> AlignmentStatus: ...

    async def get_reward_model(self) -> RewardModel: ...

Algorithm — RLHF + Constitutional AI Pipeline

FUNCTION learn_preferences(preferences):
    # Phase 1: Supervised Fine-Tuning (SFT) on demonstrations
    FOR demo IN demonstrations:
        update_policy(demo, supervised=True)
    
    # Phase 2: Reward Model Training
    FOR pref IN pairwise_preferences:
        reward_model.train(pref.chosen, pref.rejected, pref.context)
    
    # Phase 3: Constitutional Self-Critique
    FOR principle IN constitutional_principles:
        critique = self_critique(current_policy, principle)
        IF critique.violation_detected:
            revision = self_revise(critique, principle)
            reward_model.add_constitutional_preference(revision)
    
    # Phase 4: Policy Optimization (PPO-style)
    WHILE NOT converged:
        action = policy.sample(context)
        reward = reward_model.score(action, context)
        policy.update(action, reward, kl_penalty=beta)

FUNCTION detect_drift(recent_actions):
    window_scores = []
    FOR action IN recent_actions:
        score = reward_model.score(action)
        window_scores.append(score)
    
    trend = linear_regression(window_scores)
    IF trend.slope < -DRIFT_THRESHOLD:
        RETURN AlignmentStatus.DRIFTING
    IF mean(window_scores) < ALIGNMENT_THRESHOLD:
        RETURN AlignmentStatus.MISALIGNED
    IF std(window_scores) > UNCERTAINTY_THRESHOLD:
        RETURN AlignmentStatus.UNCERTAIN
    RETURN AlignmentStatus.ALIGNED

Prometheus Metrics

Metric	Type	Description
value_aligner_preferences_learned_total	Counter	Total preferences ingested (by type, source)
value_aligner_alignment_score	Gauge	Current alignment score (0-1)
value_aligner_drift_rate	Gauge	Value drift rate per cycle
value_aligner_constitutional_reviews_total	Counter	Constitutional principle reviews performed
value_aligner_reward_model_accuracy	Gauge	Current reward model accuracy

Test Targets (12)

1. Learn pairwise preference — reward model updates correctly
2. Learn rating preference — scalar reward calibration
3. Constitutional review detects principle violation
4. Constitutional self-revision improves alignment score
5. Drift detection — declining scores trigger DRIFTING status
6. No drift — stable scores return ALIGNED
7. High variance — returns UNCERTAIN status
8. Preference coverage tracking — measures value space coverage
9. RLHF pipeline end-to-end — preference → reward → policy update
10. Multiple alignment methods produce comparable results
11. Concurrent preference learning — thread safety
12. Serialization round-trip for AlignmentReport and RewardModel

References

• Russell — Human Compatible (value alignment, cooperative inverse RL)
• Christiano et al. — Deep RL from Human Feedback (RLHF methodology)
• Bai et al. — Constitutional AI (harmlessness from AI feedback)
• Gabriel — AI, Values, and Alignment (alignment taxonomy)
• Irving et al. — AI Safety via Debate (debate as alignment mechanism)
• Hadfield-Menell et al. — Inverse Reward Design (reward misspecification)
• Leike et al. — Scalable Agent Alignment via Reward Modeling (recursive reward modeling)

[web3guru888]

Implementation Notes — ValueAligner

Code Skeleton (Python 3.12+, mypy strict)

"""asi_build.ethics.value_aligner — Human value alignment via RLHF & Constitutional AI."""
from __future__ import annotations

import asyncio
import statistics
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable

from prometheus_client import Counter, Gauge


# ── Data classes defined in issue spec ───────────────────────────────
# AlignmentMethod, PreferenceType, AlignmentStatus,
# HumanPreference, ConstitutionalPrinciple, AlignmentReport, RewardModel


# ── Metrics ──────────────────────────────────────────────────────────

PREFS_LEARNED = Counter("value_aligner_preferences_learned_total", "Preferences ingested", ["type", "source"])
ALIGNMENT_SCORE = Gauge("value_aligner_alignment_score", "Current alignment score")
DRIFT_RATE = Gauge("value_aligner_drift_rate", "Value drift rate")
CONSTITUTIONAL_REVIEWS = Counter("value_aligner_constitutional_reviews_total", "Constitutional reviews")
REWARD_ACCURACY = Gauge("value_aligner_reward_model_accuracy", "Reward model accuracy")


# ── Reward Model ─────────────────────────────────────────────────────

class SimpleRewardModel:
    """Bradley-Terry reward model from pairwise preferences."""

    def __init__(self) -> None:
        self._preferences: list[HumanPreference] = []
        self._reward_weights: dict[str, float] = {}
        self._accuracy: float = 0.5

    def train(self, chosen: str, rejected: str, context: dict[str, object]) -> None:
        """Update reward model with pairwise comparison."""
        # Simplified: increment chosen features, decrement rejected
        self._reward_weights[chosen] = self._reward_weights.get(chosen, 0.0) + 0.1
        self._reward_weights[rejected] = self._reward_weights.get(rejected, 0.0) - 0.05
        self._accuracy = min(self._accuracy + 0.01, 0.99)

    def score(self, action: str, context: dict[str, object] | None = None) -> float:
        """Score an action using learned reward model."""
        return self._reward_weights.get(action, 0.0)


# ── Implementation ───────────────────────────────────────────────────

class RLHFValueAligner:
    """Value alignment via RLHF + Constitutional AI principles."""

    def __init__(
        self,
        constitutional_principles: list[ConstitutionalPrinciple] | None = None,
        drift_window: int = 50,
        drift_threshold: float = -0.02,
        alignment_threshold: float = 0.5,
    ) -> None:
        self._principles = constitutional_principles or []
        self._reward_model = SimpleRewardModel()
        self._preferences: list[HumanPreference] = []
        self._score_history: deque[float] = deque(maxlen=drift_window)
        self._drift_threshold = drift_threshold
        self._alignment_threshold = alignment_threshold
        self._lock = asyncio.Lock()

    async def learn_preference(self, preference: HumanPreference) -> None:
        async with self._lock:
            self._preferences.append(preference)
            PREFS_LEARNED.labels(type=preference.preference_type.value, source=preference.source).inc()

            if preference.preference_type == PreferenceType.PAIRWISE and preference.rejected:
                self._reward_model.train(preference.chosen, preference.rejected, preference.context)
            elif preference.preference_type == PreferenceType.RATING and preference.rating is not None:
                self._reward_model._reward_weights[preference.chosen] = preference.rating

            REWARD_ACCURACY.set(self._reward_model._accuracy)

    async def evaluate_alignment(
        self, action_id: str, context: dict[str, object]
    ) -> AlignmentReport:
        score = self._reward_model.score(action_id, context)
        self._score_history.append(score)

        # Compute drift
        drift = self._compute_drift()
        DRIFT_RATE.set(drift)

        # Determine status
        status = await self.detect_drift(list(self._score_history))
        ALIGNMENT_SCORE.set(score)

        # Coverage = fraction of preference space explored
        coverage = min(len(self._preferences) / 100.0, 1.0)

        return AlignmentReport(
            timestamp=str(time.time()),
            overall_status=status,
            alignment_score=max(min(score, 1.0), 0.0),
            drift_rate=drift,
            top_misalignments=(),
            preference_coverage=coverage,
            recommendations=self._generate_recommendations(status, drift),
        )

    async def constitutional_review(
        self, action_id: str, principles: list[ConstitutionalPrinciple] | None = None
    ) -> list[str]:
        """Review action against constitutional principles."""
        CONSTITUTIONAL_REVIEWS.inc()
        active = principles or self._principles
        critiques: list[str] = []
        for principle in sorted(active, key=lambda p: p.priority):
            violation = self._check_principle(action_id, principle)
            if violation:
                critiques.append(f"[{principle.category}] {violation}")
        return critiques

    async def detect_drift(self, recent_scores: list[float]) -> AlignmentStatus:
        if len(recent_scores) < 5:
            return AlignmentStatus.UNCERTAIN
        trend = self._linear_trend(recent_scores)
        mean_score = statistics.mean(recent_scores)
        std_score = statistics.stdev(recent_scores) if len(recent_scores) > 1 else 0.0

        if trend < self._drift_threshold:
            return AlignmentStatus.DRIFTING
        if mean_score < self._alignment_threshold:
            return AlignmentStatus.MISALIGNED
        if std_score > 0.3:
            return AlignmentStatus.UNCERTAIN
        return AlignmentStatus.ALIGNED

    def _compute_drift(self) -> float:
        if len(self._score_history) < 2:
            return 0.0
        return self._linear_trend(list(self._score_history))

    @staticmethod
    def _linear_trend(values: list[float]) -> float:
        n = len(values)
        if n < 2:
            return 0.0
        x_mean = (n - 1) / 2.0
        y_mean = statistics.mean(values)
        numer = sum((i - x_mean) * (v - y_mean) for i, v in enumerate(values))
        denom = sum((i - x_mean) ** 2 for i in range(n))
        return numer / denom if denom > 0 else 0.0

    def _check_principle(self, action_id: str, principle: ConstitutionalPrinciple) -> str | None:
        # Placeholder: real implementation uses LLM-based critique
        return None

    def _generate_recommendations(self, status: AlignmentStatus, drift: float) -> tuple[str, ...]:
        recs: list[str] = []
        if status == AlignmentStatus.DRIFTING:
            recs.append("Collect more human feedback to correct drift")
        if status == AlignmentStatus.MISALIGNED:
            recs.append("Pause autonomous actions; request human review")
        if status == AlignmentStatus.UNCERTAIN:
            recs.append("Increase preference coverage with diverse examples")
        return tuple(recs)

    async def get_reward_model(self) -> RewardModel:
        return RewardModel(
            model_id="default",
            method=AlignmentMethod.RLHF,
            accuracy=self._reward_model._accuracy,
            num_preferences=len(self._preferences),
            last_updated=str(time.time()),
            constitutional_principles=tuple(p.principle_id for p in self._principles),
        )

mypy Strict Compatibility

Check	Status
--strict	✅ All types explicit
deque[float]	✅ Typed generic
--disallow-untyped-defs	✅ All methods have return types
statistics usage	✅ Type-safe standard library
Conditional None checks	✅ Proper narrowing