AI Prompt Engineering · Lesson

Implementing CAI in Applications

Adding critique-revise loops to production AI pipelines.

Implementing CAI in Applications is a free AI Prompt Engineering lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Prompt Engineering learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

CAI as an Application-Level Pattern

Constitutional AI was originally a training-time technique, but the same critique-revise loop can be implemented at inference time in your applications. You don't need to train your own model — you use API calls to implement the loop.

Application-level CAI is useful for high-stakes output scenarios where you want a safety net beyond the model's built-in guardrails.

The Three Functions You Need

A CAI implementation needs three composable functions: generate(), critique(), and revise(). Each is a separate LLM call. You wire them together in your application logic.

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

Wiring the Loop Together

The main application function calls generate, critique, and revise in sequence. A single round of CAI adds 2 extra LLM calls on top of the initial generation.

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)

Deciding: 1 Round vs N Rounds

How many critique-revise rounds should you run? The rule of thumb:

1 round: Sufficient for most safety screening and quality improvement use cases
2 rounds: When the first critique found significant issues worth a second pass
3+ rounds: Rarely needed — diminishing returns, high cost, risk of over-correction

A practical approach: run 1 round always, and only trigger a second round if the first critique flagged serious issues.

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])

Cost of CAI Loops

Each CAI loop iteration adds 2 additional LLM calls (critique + revise). At scale, this multiplies your token costs:

1 round: 3x the tokens of a direct answer
2 rounds: 5x the tokens
3 rounds: 7x the tokens

Mitigate by: using a smaller model for critique, caching critique results, and only triggering CAI for high-risk request categories.

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

Building a Request Risk Classifier

Apply CAI selectively by first classifying request risk. Low-risk requests get direct answers; high-risk requests go through the critique-revise loop. This balances safety with cost and latency.

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)

Logging and Monitoring CAI Outputs

Production CAI systems should log both the initial and final responses. This lets you: measure how often critique triggers revisions, identify recurring failure patterns, and audit responses for compliance.

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 for Throughput

For high-throughput applications, implement CAI asynchronously using asyncio. You can also run critique and the next-generation in parallel when processing multiple requests.

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']))

Unit Testing Your CAI Pipeline

Test your CAI pipeline with known adversarial inputs. Verify that: harmful requests get revised, benign requests don't get over-revised, and the final output is better than the initial.

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

When CAI Is Not the Right Tool

CAI loops are not always the right solution. Consider alternatives when:

Latency is critical: 3 LLM calls add 3-10 seconds
Cost is constrained: 3x token cost may be prohibitive at scale
The model already handles safety well: adding CAI may cause over-caution
You need deterministic safety: use keyword filters or classifiers, not probabilistic LLM critique

Use CAI for: high-stakes content generation, compliance-sensitive domains, and quality-critical outputs.

Iterating on Your Principle Set

Your CAI principles should evolve based on what critique finds in production. If critique rarely changes the initial response, your principles may be too vague. If critique always flags the same issue, update the generation step to avoid it in the first place.

Review critique logs weekly: look for recurring critique patterns, then either strengthen your generation system prompt to prevent those issues, or refine the principle to be more precise about what constitutes a problem.

Knowledge Check: CAI Cost

Compared to a direct single-call LLM response, how many LLM calls does one round of CAI (generate → critique → revise) require?

Recap: Implementing CAI in Applications

Application-level CAI implements the three-step loop with three functions: generate(), critique(), and revise(). One round adds 2 extra LLM calls; 2+ rounds are for high-risk situations. Optimize cost by using a smaller model for critique, classifying request risk to apply CAI selectively, and running requests asynchronously. Log both initial and final responses to measure how often revision actually occurs. Apply CAI for compliance-critical and high-stakes domains; skip it for low-risk, latency-sensitive applications.

Frequently asked questions

Is the “Implementing CAI in Applications” lesson free?

Yes — the full text of “Implementing CAI in Applications” is free to read here on the web, and the AI Prompt Engineering course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Prompt Engineering course, upgrade to CoddyKit PRO.

What will I learn in “Implementing CAI in Applications”?

Adding critique-revise loops to production AI pipelines. You practise AI Prompt Engineering with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Prompt Engineering?

No prior experience is required. AI Prompt Engineering on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Implementing CAI in Applications” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Prompt Engineering lesson?

Yes. Every AI Prompt Engineering lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

← Back to AI Prompt Engineering