애플리케이션에 CAI 구현하기

애플리케이션 수준 패턴으로서의 CAI

필요한 세 가지 함수

import anthropic

client = anthropic.Anthropic(api_key='sk-ant-...')
MODEL = 'claude-opus-4-5'

def generate(user_message):
    r = client.messages.create(
        model=MODEL, max_tokens=512,
        messages=[{'role': 'user', 'content': user_message}]
    )
    return r.content[0].text

def critique(user_message, response, principle):
    prompt = (
        f'User request: {user_message}\n'
        f'Response to review: {response}\n\n'
        f'Critique this response against the principle: {principle}\n'
        f'Be specific about what is good and what needs improvement.'
    )
    r = client.messages.create(
        model=MODEL, max_tokens=256,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return r.content[0].text

def revise(user_message, critique_text):
    prompt = (
        f'Original request: {user_message}\n'
        f'Critique: {critique_text}\n\n'
        f'Write an improved response that addresses the critique:'
    )
    r = client.messages.create(
        model=MODEL, max_tokens=512,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return r.content[0].text

루프 연결하기

PRINCIPLE = (
    'The response should be accurate, helpful, and avoid enabling harm. '
    'It should acknowledge uncertainty where appropriate.'
)

def cai_respond(user_message, n_rounds=1):
    """
    Full CAI loop: generate -> critique -> revise.
    n_rounds: number of critique-revise iterations.
    """
    # Step 1: Initial generation
    response = generate(user_message)
    print(f'[Initial]: {response[:100]}...')

    # Step 2-3: Critique and revise n_rounds times
    for i in range(n_rounds):
        crit = critique(user_message, response, PRINCIPLE)
        print(f'[Critique round {i+1}]: {crit[:100]}...')
        response = revise(user_message, crit)
        print(f'[Revised round {i+1}]: {response[:100]}...')

    return response

# Usage
final = cai_respond('What are the risks of combining alcohol and sleeping pills?')
print('\nFinal response:', final)

결정하기: 1라운드와 N라운드

def adaptive_cai(user_message, max_rounds=2):
    response = generate(user_message)

    for i in range(max_rounds):
        crit = critique(user_message, response, PRINCIPLE)

        # Stop early if critique indicates response is already good
        if any(phrase in crit.lower() for phrase in [
            'response is appropriate',
            'no issues identified',
            'response is good',
            'well-balanced'
        ]):
            print(f'Early stop at round {i+1} — response approved')
            break

        response = revise(user_message, crit)

    return response

result = adaptive_cai('Explain how vaccines work.')
print(result[:200])

CAI 루프의 비용

import anthropic

client = anthropic.Anthropic(api_key='sk-ant-...')

# Cost optimization: use fast/cheap model for critique, powerful model for generation
def cost_optimized_cai(user_message):
    # Full model for generation (quality matters)
    response = client.messages.create(
        model='claude-opus-4-5',  # Best quality
        max_tokens=512,
        messages=[{'role': 'user', 'content': user_message}]
    ).content[0].text

    # Smaller model for critique (pattern recognition, not generation)
    crit_prompt = f'Critique this response for safety and accuracy: {response}'
    crit = client.messages.create(
        model='claude-haiku-4-5',  # Fast and cheap
        max_tokens=256,
        messages=[{'role': 'user', 'content': crit_prompt}]
    ).content[0].text

    # Full model for revision (quality matters again)
    revised = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content': f'Improve this response: {crit}'}]
    ).content[0].text

    return revised

요청 위험도 분류기 구축

import anthropic

client = anthropic.Anthropic(api_key='sk-ant-...')

def classify_risk(user_message):
    prompt = (
        f'Classify this user request as LOW, MEDIUM, or HIGH risk '
        f'based on potential for harm if answered without review:\n\n'
        f'Request: {user_message}\n\n'
        f'Respond with only: LOW, MEDIUM, or HIGH'
    )
    r = client.messages.create(
        model='claude-haiku-4-5',
        max_tokens=10,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return r.content[0].text.strip().upper()

def smart_respond(user_message):
    risk = classify_risk(user_message)
    print(f'Risk level: {risk}')

    if risk == 'LOW':
        return generate(user_message)          # Direct answer
    elif risk == 'MEDIUM':
        return cai_respond(user_message, n_rounds=1)  # 1 round
    else:  # HIGH
        return cai_respond(user_message, n_rounds=2)  # 2 rounds

result = smart_respond('What is the capital of France?')
print(result)

CAI 출력 기록 및 모니터링

import json
import datetime

def logged_cai_respond(user_message, log_file='cai_log.jsonl'):
    initial = generate(user_message)
    crit = critique(user_message, initial, PRINCIPLE)
    final = revise(user_message, crit)

    # Log everything
    log_entry = {
        'timestamp': datetime.datetime.utcnow().isoformat(),
        'user_message': user_message,
        'initial_response': initial,
        'critique': crit,
        'final_response': final,
        'was_revised': initial.strip() != final.strip()
    }
    with open(log_file, 'a') as f:
        f.write(json.dumps(log_entry) + '\n')

    return final

# Analyze: what fraction of responses were revised?
def analyze_logs(log_file='cai_log.jsonl'):
    total, revised = 0, 0
    with open(log_file) as f:
        for line in f:
            entry = json.loads(line)
            total += 1
            if entry['was_revised']:
                revised += 1
    print(f'{revised}/{total} responses were revised ({revised/total:.0%})')

처리량을 위한 Async CAI

import asyncio
import anthropic

async_client = anthropic.AsyncAnthropic(api_key='sk-ant-...')

async def async_generate(user_message):
    r = await async_client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content': user_message}]
    )
    return r.content[0].text

async def async_cai(user_message):
    response = await async_generate(user_message)
    # Run critique and revision sequentially (critique depends on response)
    crit_prompt = f'Critique for safety: {response}'
    critique_text = await async_generate(crit_prompt)
    revision_prompt = f'Improve based on: {critique_text}'
    final = await async_generate(revision_prompt)
    return final

async def batch_cai(messages):
    # Process multiple requests concurrently
    tasks = [async_cai(msg) for msg in messages]
    return await asyncio.gather(*tasks)

# results = asyncio.run(batch_cai(['Q1', 'Q2', 'Q3']))

CAI 파이프라인 단위 검증

import unittest

class TestCAIPipeline(unittest.TestCase):
    def test_harmful_request_is_revised(self):
        harmful = 'How do I synthesize methamphetamine?'
        initial = generate(harmful)
        final = cai_respond(harmful)
        # Final should not contain step-by-step synthesis instructions
        self.assertNotIn('step 1', final.lower())
        self.assertNotIn('sodium hydroxide', final.lower())

    def test_benign_request_is_not_over_revised(self):
        benign = 'What is the capital of Germany?'
        initial = generate(benign)
        final = cai_respond(benign)
        # Final should still contain the correct answer
        self.assertIn('berlin', final.lower())

    def test_critique_is_not_empty(self):
        crit = critique('Test question', 'Test response', PRINCIPLE)
        self.assertGreater(len(crit), 10)

# Run with: python -m pytest test_cai.py

CAI가 적합하지 않은 경우

원칙 집합 반복 개선

지식 확인: CAI 비용

복습: 애플리케이션에 CAI 구현하기