アプリケーションへのCAI実装
本番環境のAIパイプラインに、批評と修正のループを追加します。
「アプリケーションへのCAI実装」はCoddyKit上の無料AI Prompt Engineeringレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Prompt Engineering学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Prompt Engineeringコースには全4レッスンが含まれています。
アプリケーションレベルのパターンとしてのCAI
Constitutional AIはもともと学習時の技術でしたが、同じ批評と改訂のループを推論時にアプリケーションへ実装できます。独自のモデルを学習する必要はありません。API呼び出しを使ってループを実装します。
アプリケーションレベルのCAIは、モデル組み込みのガードレールを超える安全網が必要な、重要度の高い出力のシナリオで役立ちます。
必要な3つの関数
CAIの実装には、generate()、critique()、revise()という3つの組み合わせ可能な関数が必要です。それぞれが独立したLLM呼び出しです。アプリケーションのロジックでこれらをつなぎ合わせます。
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ループをつなぎ合わせる
メインのアプリケーション関数は、generate、critique、reviseを順番に呼び出します。CAIを1ラウンド実行すると、初回の生成に加えてLLM呼び出しが2回追加されます。
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ラウンドかを決める
批評と改訂のラウンドは何回実行すべきでしょうか。目安は次のとおりです。
- 1ラウンド：ほとんどの安全性スクリーニングや品質改善の用途に十分です
- 2ラウンド：1回目の批評で、2回目の確認に値する重大な問題が見つかった場合です
- 3ラウンド以上：必要になることはまれです。効果が逓減し、コストが高く、過剰修正のリスクがあります
実用的な方法は、常に1ラウンド実行し、1回目の批評で重大な問題が指摘された場合にだけ2回目を実行することです。
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ループのコスト
CAIを1回反復するごとに、LLM呼び出しが2回（批評と改訂）追加されます。大規模に実行すると、トークンコストが次のように増加します。
- 1ラウンド：直接回答の3倍のトークン
- 2ラウンド：5倍のトークン
- 3ラウンド：7倍のトークン
対策として、批評には小規模なモデルを使う、批評結果をキャッシュする、リスクの高い依頼カテゴリに対してのみ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依頼リスク分類器を構築する
まず依頼のリスクを分類してから、CAIを選択的に適用します。リスクの低い依頼には直接回答し、リスクの高い依頼には批評と改訂のループを適用します。これにより、安全性とコスト、レイテンシのバランスを取れます。
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出力のログ記録と監視
本番環境の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%})')スループット向上のための非同期CAI
高スループットのアプリケーションでは、asyncioを使って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パイプラインを単体テストする
既知の敵対的入力を使って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.pyCAIが適切なツールではない場合
CAIループが常に適切な解決策とは限りません。次のような場合は、代替手段を検討してください。
- レイテンシが重要な場合：LLM呼び出しが3回になると、3～10秒追加されます
- コストが制約される場合：大規模環境では、トークンコストが3倍になることが許容できない可能性があります
- モデルがすでに安全性を適切に処理している場合：CAIを追加すると過度に慎重になる可能性があります
- 決定論的な安全性が必要な場合：確率的なLLM批評ではなく、キーワードフィルターや分類器を使ってください
CAIは、重要度の高いコンテンツ生成、コンプライアンスに敏感な分野、品質が特に重要な出力に使用してください。
原則セットを反復的に改善する
CAIの原則は、本番環境で批評が発見した内容に基づいて進化させる必要があります。批評によって初回の応答がほとんど変わらない場合、原則が曖昧すぎる可能性があります。批評が常に同じ問題を指摘する場合は、そもそもその問題が発生しないよう生成ステップを更新してください。
批評ログを毎週確認してください。繰り返し現れる批評パターンを探し、それらの問題を防ぐために生成用のシステムプロンプトを強化するか、何が問題に当たるのかをより正確に示すよう原則を改良します。
理解度チェック：CAIのコスト
直接1回だけLLMを呼び出して応答する場合と比べて、CAIの1ラウンド（generate → critique → revise）にはLLM呼び出しが何回必要ですか。
まとめ：アプリケーションにCAIを実装する
アプリケーションレベルのCAIでは、generate()、critique()、revise()という3つの関数で3段階のループを実装します。1ラウンドでLLM呼び出しが2回追加され、2ラウンド以上はリスクの高い状況で使用します。批評に小規模なモデルを使い、依頼のリスクを分類してCAIを選択的に適用し、依頼を非同期で実行することでコストを最適化します。初回と最終の応答を両方ログに記録し、実際にどの程度の頻度で改訂が行われるかを測定します。CAIはコンプライアンスが重要な分野や重要度の高い分野に適用し、リスクが低くレイテンシが重視されるアプリケーションでは省略してください。
AI チューターと学ぶ AI Prompt Engineering — 無料
ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。
- コース
- 53
- レッスン
- 199
よくある質問
「アプリケーションへのCAI実装」レッスンは無料ですか？
はい。「アプリケーションへのCAI実装」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Prompt Engineeringコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Prompt Engineeringコースには全4レッスンが含まれています。
「アプリケーションへのCAI実装」で何を学びますか？
本番環境のAIパイプラインに、批評と修正のループを追加します。 ブラウザで直接実行するハンズオンコードでAI Prompt Engineeringを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Prompt Engineeringを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Prompt Engineeringは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「アプリケーションへのCAI実装」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Prompt Engineeringレッスンでコードを書いて実行できますか？
はい。すべてのAI Prompt Engineeringレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- CAIの原則と批評プロンプト
- 自己批評と修正のパターン
- 無害性と有用性のジレンマ
- アプリケーションへのCAI実装