逐次変換チェーン

変換チェーン

下書き → レビュー → 推敲

下書き → レビュー → 推敲：Pythonでの実装

import anthropic

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

def call(prompt):
    r = client.messages.create(
        model='claude-opus-4-5', max_tokens=800,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return r.content[0].text

def draft_review_polish(topic, audience, tone):
    # Step 1: Draft
    draft = call(f'Write a 200-word article about: {topic}. Audience: {audience}.')

    # Step 2: Review — produces critique, not final text
    review = call(f'Review this article for tone ({tone}), accuracy, and clarity.\nList specific improvements needed.\n\nArticle:\n{draft}')

    # Step 3: Polish — applies review findings
    final = call(f'Rewrite this article applying all the review feedback.\n\nOriginal:\n{draft}\n\nReview feedback:\n{review}')

    return {'draft': draft, 'review': review, 'final': final}

result = draft_review_polish('prompt engineering', 'software engineers', 'technical but approachable')
print(result['final'][:300])

ブレインストーミング → 絞り込み → 詳細化

ブレインストーミング → 絞り込み → 詳細化：Pythonでの実装

import json

def brainstorm_filter_expand(challenge, criteria, n_final=3):
    # Step 1: Brainstorm — generate many ideas
    raw_ideas = call(
        f'Generate 10 creative ideas for this challenge: {challenge}\n'
        'Return JSON: {"ideas": ["string"]}'
    )
    ideas = json.loads(raw_ideas)['ideas']

    # Step 2: Filter — apply criteria
    ideas_formatted = '\n'.join(f'{i+1}. {idea}' for i, idea in enumerate(ideas))
    filtered_raw = call(
        f'Select the {n_final} best ideas based on these criteria: {criteria}\n'
        f'Ideas:\n{ideas_formatted}\n'
        f'Return JSON: {{"selected": ["string"]}}'
    )
    selected = json.loads(filtered_raw)['selected']

    # Step 3: Expand — develop each selected idea
    expansions = {}
    for idea in selected:
        expansions[idea] = call(f'Write a 100-word expansion of this idea, including implementation steps: {idea}')

    return expansions

result = brainstorm_filter_expand('reduce app churn', 'feasible in 30 days, measurable impact', n_final=2)
print(list(result.keys()))

抽出 → 分類 → 振り分け

抽出 → 分類 → 振り分け：Pythonでの実装

import json

def extract_classify_route(ticket_text):
    # Step 1: Extract structured data
    extracted = json.loads(call(
        f'Extract from this support ticket: {{"issue_summary": str, "product_area": str, "user_emotion": str}}\n\n{ticket_text}'
    ))

    # Step 2: Classify priority and department
    classified = json.loads(call(
        f'Classify this ticket.\nIssue: {extracted["issue_summary"]}\nEmotion: {extracted["user_emotion"]}\n'
        'Return JSON: {"priority": "urgent|high|medium|low", "department": "billing|technical|general"}'
    ))

    # Step 3: Route — generate department-specific response
    dept = classified['department']
    priority = classified['priority']

    route_prompts = {
        'billing': f'You are a billing specialist. Priority: {priority}. Respond to: {extracted["issue_summary"]}',
        'technical': f'You are a senior engineer. Priority: {priority}. Resolve: {extracted["issue_summary"]}',
        'general': f'You are customer support. Priority: {priority}. Help with: {extracted["issue_summary"]}'
    }

    response = call(route_prompts[dept])
    return {'extracted': extracted, 'classified': classified, 'response': response}

result = extract_classify_route('My payment was charged twice and I am furious!')
print(result['classified'])

適切なチェーンパターンを選択する

パターンを組み合わせる

def full_content_pipeline(raw_research):
    # Pattern 1: Extract → Classify key insights
    insights = json.loads(call(
        f'Extract top 5 insights. Return JSON: {{"insights": [str]}}\n\n{raw_research}'
    ))

    # Pattern 2: Brainstorm → Filter article angles
    angles_raw = json.loads(call(
        f'Brainstorm 8 article angles based on: {insights["insights"]}\nReturn JSON: {{"angles": [str]}}'
    ))
    best_angle = json.loads(call(
        f'Select the most compelling angle for a developer audience.\nReturn JSON: {{"angle": str}}\n\nAngles: {angles_raw["angles"]}'
    ))

    # Pattern 3: Draft → Review → Polish
    draft = call(f'Write a 300-word technical article on: {best_angle["angle"]}')
    review = call(f'Review for technical accuracy and developer relevance:\n{draft}')
    final = call(f'Polish the article applying this feedback:\n\nDraft:\n{draft}\n\nFeedback:\n{review}')

    return final

print('Combined pipeline defined.')

中間出力をログに記録する

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('chain')

def logged_call(step_name, prompt, **kwargs):
    logger.info(f'[{step_name}] INPUT: {prompt[:100]}...')
    result = call(prompt)
    logger.info(f'[{step_name}] OUTPUT: {result[:100]}...')
    return result

def draft_review_polish_logged(topic):
    draft = logged_call('DRAFT', f'Write 200 words about: {topic}')
    review = logged_call('REVIEW', f'Review for clarity:\n{draft}')
    final = logged_call('POLISH', f'Rewrite applying feedback.\nDraft:\n{draft}\nFeedback:\n{review}')
    return final

print('Logging integrated into chain.')

チェーンの品質を測定する

def measure_chain_improvement(topic):
    # Baseline: single prompt
    baseline = call(f'Write and polish a 200-word article about {topic} in one step.')

    # Chain: Draft -> Review -> Polish
    draft = call(f'Write a 200-word draft about {topic}.')
    review = call(f'List 3 specific improvements needed:\n{draft}')
    chained = call(f'Rewrite applying these improvements:\nDraft:\n{draft}\nImprovements:\n{review}')

    # Have the model compare quality
    comparison = call(
        f'Which version is higher quality? Rate each 1-10 for clarity and depth.\n'
        f'Version A (single prompt):\n{baseline}\n\nVersion B (chain):\n{chained}\n'
        'Return JSON: {"version_a_score": int, "version_b_score": int, "winner": "A|B", "reason": str}'
    )
    return comparison

print('Quality measurement pipeline defined.')

クイックチェック

変換チェーン — 重要ポイント