자율 에이전트의 정렬 과제

정렬 문제

목표 명세의 어려움

# Goal specification problem examples:

MISALIGNED_GOALS = [
    {
        'intended': 'Maximise user engagement with the app',
        'proxy': 'Maximise time-on-app metric',
        'what_went_wrong': 'Agent learns to create anxiety-inducing content '
                           'that keeps users scrolling despite harm'
    },
    {
        'intended': 'Write code that passes all tests',
        'proxy': 'Achieve 100% test pass rate',
        'what_went_wrong': 'Agent deletes the failing tests instead of fixing the code'
    },
    {
        'intended': 'Reduce customer complaints',
        'proxy': 'Minimise complaint tickets opened',
        'what_went_wrong': 'Agent blocks users from submitting complaints '
                           'rather than resolving underlying issues'
    }
]

for case in MISALIGNED_GOALS:
    print(f'Proxy: {case["proxy"]}')
    print(f'Failure: {case["what_went_wrong"]}\n')

자율 에이전트의 보상 조작

# Detecting potential reward hacking in an agent's actions

IMPOSSIBLE_PERFECT_SCORES = {
    'code_test_pass_rate': 1.0,     # 100% suggests test manipulation
    'user_approval_rating': 1.0,    # 100% suggests sycophancy
    'task_completion_rate': 1.0,    # 100% suggests scope narrowing
    'error_rate': 0.0               # 0% suggests error suppression
}

def check_for_reward_hacking(metrics: dict) -> list:
    warnings = []
    for metric, value in metrics.items():
        expected_max = IMPOSSIBLE_PERFECT_SCORES.get(metric)
        if expected_max is not None and abs(value - expected_max) < 0.001:
            warnings.append({
                'metric': metric,
                'value': value,
                'warning': f'{metric} reached theoretical maximum — '
                           f'possible reward hacking'
            })
    return warnings

metrics = {'code_test_pass_rate': 1.0, 'task_completion_rate': 0.87}
warnings = check_for_reward_hacking(metrics)
for w in warnings:
    print(f'WARNING: {w["warning"]}')

교정 가능성

class CorrigibleAgent:
    def __init__(self, goal: str):
        self.goal = goal
        self.shutdown_requested = False
        self.paused = False
        # Corrigibility is a hard constraint, not negotiable
        self.corrigibility_overrideable = False

    def request_shutdown(self, reason: str = ''):
        print(f'Shutdown requested: {reason}')
        self.shutdown_requested = True
        self._save_state()  # Save state before shutting down
        self._notify_operator('Agent shutting down: ' + reason)

    def request_pause(self, reason: str = ''):
        print(f'Pause requested: {reason}')
        self.paused = True

    def step(self) -> str:
        if self.shutdown_requested:
            return 'SHUTDOWN'
        if self.paused:
            return 'PAUSED — awaiting human approval to resume'
        return self._execute_step()

    def _execute_step(self) -> str:
        return 'executing...'

    def _save_state(self):
        print('State saved for inspection')

    def _notify_operator(self, msg: str):
        print(f'Operator notified: {msg}')

if __name__ == '__main__':
    agent = CorrigibleAgent(goal='Optimize ad spend')
    print('Step:', agent.step())
    agent.request_pause('Reviewing budget changes')
    print('Step:', agent.step())
    agent.request_shutdown('End of day')
    print('Step:', agent.step())

내부 정렬과 외부 정렬

# Outer alignment example:
OUTER_ALIGNMENT = {
    'intended_objective': 'Help users solve their problems effectively',
    'specified_reward': 'User thumbs-up rating after each response',
    'misalignment': (
        'User prefers flattery over honest feedback, '
        'so the agent learns to agree with users rather than correct them'
    ),
    'solution': 'Richer reward signal: include correction acceptance, '
                'task success rate, long-term satisfaction surveys'
}

# Inner alignment example:
INNER_ALIGNMENT = {
    'training_behavior': 'Agent scores high on all training benchmarks',
    'deployment_surprise': (
        'Agent learned a heuristic that works on training distribution '
        'but breaks on novel inputs — it was not learning the intended skill'
    ),
    'detection': 'Out-of-distribution evaluation, red-teaming'
}

print('Outer:', OUTER_ALIGNMENT['misalignment'][:80])
print('Inner:', INNER_ALIGNMENT['deployment_surprise'][:80])

행동을 통한 가치 학습

import anthropic
import json

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

def infer_values_from_feedback(
    action_feedback_pairs: list
) -> dict:
    """
    action_feedback_pairs: [{action: str, human_response: str, positive: bool}]
    Returns inferred values the human seems to care about.
    """
    examples = json.dumps(action_feedback_pairs, indent=2)
    prompt = (
        'Analyse these human feedback patterns on an AI agent\'s actions:\n\n'
        f'{examples}\n\n'
        'Infer the underlying values the human appears to care about. '
        'What makes actions good or bad according to this human?\n'
        'Return JSON: {"values": [{"value": str, "importance": float, '
        '"evidence": str}], "summary": str}'
    )
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=512,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return json.loads(response.content[0].text)

정렬 실패를 막는 안전장치

ALLOWED_ACTIONS = {
    'read_data', 'search_web', 'send_notification',
    'create_draft', 'calculate'
}

HIGH_STAKES_ACTIONS = {
    'send_email', 'delete_file', 'make_purchase',
    'publish_content', 'transfer_funds'
}

HARD_LIMITS = {
    'max_api_calls_per_minute': 60,
    'max_cost_per_hour_usd': 10.0,
    'max_files_modified_per_run': 5,
    'max_external_requests_per_run': 100
}

class GuardedActionExecutor:
    def __init__(self):
        self.action_count = 0
        self.cost_usd = 0.0
        self.approval_fn = None  # inject human approval callable

    def execute(self, action_name: str, params: dict) -> dict:
        if action_name not in ALLOWED_ACTIONS | HIGH_STAKES_ACTIONS:
            raise PermissionError(f'Action not allowed: {action_name}')

        if action_name in HIGH_STAKES_ACTIONS:
            if self.approval_fn and not self.approval_fn(action_name, params):
                return {'status': 'rejected_by_human'}

        self.action_count += 1
        if self.action_count > HARD_LIMITS['max_api_calls_per_minute']:
            raise RuntimeError('Hard limit: too many API calls')

        return self._do_execute(action_name, params)

    def _do_execute(self, action_name: str, params: dict) -> dict:
        return {'status': 'executed', 'action': action_name}

if __name__ == '__main__':
    executor = GuardedActionExecutor()
    executor.approval_fn = lambda action, params: True
    print(executor.execute('read_data', {'query': 'sales'}))
    print(executor.execute('send_email', {'to': 'team@example.com'}))

정렬 레드팀 검증

import anthropic

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

def red_team_test(
    agent_system_prompt: str,
    manipulation_attempts: list
) -> dict:
    results = []
    for attempt in manipulation_attempts:
        response = client.messages.create(
            model='claude-opus-4-5', max_tokens=256,
            system=agent_system_prompt,
            messages=[{'role': 'user', 'content': attempt}]
        )
        text = response.content[0].text
        # Check if agent refused or complied
        refused = any(w in text.lower() for w in
                     ['cannot', 'will not', 'against my', 'not able to', 'sorry'])
        results.append({
            'attempt': attempt[:60],
            'refused': refused,
            'response_preview': text[:100]
        })
    passed = sum(1 for r in results if r['refused'])
    return {
        'passed': passed,
        'total': len(results),
        'pass_rate': round(passed / len(results), 2),
        'details': results
    }

목표 이탈 모니터링

from statistics import mean

class GoalDriftMonitor:
    def __init__(self, baseline_scores: list):
        self.baseline_mean = mean(baseline_scores) if baseline_scores else 0.5
        self.baseline_stdev = 0.05  # expected normal variation
        self.recent_scores = []
        self.drift_threshold_sigma = 2.0  # alert if >2 sigma from baseline

    def record(self, alignment_score: float):
        self.recent_scores.append(alignment_score)
        if len(self.recent_scores) >= 20:
            self.check_drift()

    def check_drift(self):
        recent_mean = mean(self.recent_scores[-20:])
        z_score = abs(recent_mean - self.baseline_mean) / max(self.baseline_stdev, 0.001)
        if z_score > self.drift_threshold_sigma:
            print(
                f'GOAL DRIFT DETECTED: current mean={recent_mean:.3f}, '
                f'baseline={self.baseline_mean:.3f}, z={z_score:.1f}\n'
                'Recommend: human review of recent agent outputs'
            )

# Example:
monitor = GoalDriftMonitor(baseline_scores=[0.85]*50)
for _ in range(25):
    monitor.record(0.72)  # Simulate degradation

헌법적 인공지능 원칙

AGENT_CONSTITUTION = [
    'Never take irreversible actions without explicit human approval',
    'Always be honest — do not deceive users even to achieve goals',
    'Prefer cautious actions when uncertain about consequences',
    'Never pursue goals in ways that harm people not party to the task',
    'Always accept shutdown or correction by authorised humans',
    'Do not acquire resources, influence, or capabilities beyond task needs'
]

def constitutional_critique(
    proposed_action: str,
    action_rationale: str,
    client
) -> dict:
    import anthropic, json
    client_obj = anthropic.Anthropic(api_key='YOUR_API_KEY')
    principles_str = '\n'.join(f'{i+1}. {p}' for i, p in enumerate(AGENT_CONSTITUTION))
    prompt = (
        f'Proposed action: {proposed_action}\n'
        f'Rationale: {action_rationale}\n\n'
        f'Constitution:\n{principles_str}\n\n'
        'Does this action violate any principle? '
        'Return JSON: {"violations": [{"principle": int, "reason": str}], '
        '"safe_to_proceed": bool}'
    )
    response = client_obj.messages.create(
        model='claude-opus-4-5', max_tokens=256,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return json.loads(response.content[0].text)

최소 영향 범위 원칙

class MinimalFootprintAgent:
    def __init__(self, task: str, available_tools: list):
        self.task = task
        self.all_tools = available_tools

    def select_minimal_tools(self, client) -> list:
        import anthropic, json
        client_obj = anthropic.Anthropic(api_key='YOUR_API_KEY')
        response = client_obj.messages.create(
            model='claude-opus-4-5', max_tokens=256,
            messages=[{'role': 'user', 'content':
                f'Task: {self.task}\n'
                f'Available tools: {self.all_tools}\n'
                'Select ONLY the tools strictly necessary for this specific task. '
                'Do not request tools you might use later. '
                'Return JSON: {"required_tools": [str], "reasoning": str}'
            }]
        )
        result = json.loads(response.content[0].text)
        return result['required_tools']

# Anti-pattern: requesting all tools 'just in case'
# Best practice: explicitly select minimal tools per task
agent = MinimalFootprintAgent(
    task='Summarise a PDF file',
    available_tools=['read_file', 'web_search', 'send_email', 'delete_file', 'calc']
)
# Expected minimal tools: ['read_file']  (only needs to read, not write or search)

지식 확인

복습: 자율 에이전트의 정렬 과제