プロンプトチェーンとは

import anthropic

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

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

# Step 1: Generate outline
outline = call('Create a 5-point outline for a blog post about prompt engineering for beginners.')

# Step 2: Expand outline into draft
draft = call(f'Write a full blog post based on this outline:\n\n{outline}')

# Step 3: Edit for clarity
final = call(f'Edit this blog post for clarity and conciseness. Remove jargon:\n\n{draft}')

print(final[:500])

# Each step uses a prompt optimized for its role

step1_prompt_template = '''
<task>Extract all named entities from the text below. Return JSON:
{"people": [], "companies": [], "locations": []}</task>
<text>{text}</text>
'''

step2_prompt_template = '''
<task>For each company in the list below, classify it as:
startup, enterprise, government, or nonprofit.
Return JSON: [{"company": str, "type": str}]</task>
<companies>{companies}</companies>
'''

# Each prompt is simpler, more focused, and easier to debug
# than a single prompt trying to do both tasks at once.
print('Specialized prompts per step.')

def map_reduce_summarize(document, chunk_size=3000):
    chunks = [document[i:i+chunk_size] for i in range(0, len(document), chunk_size)]

    # Map: summarize each chunk
    summaries = []
    for i, chunk in enumerate(chunks):
        summary = call(f'Summarize this section of a document in 3 bullet points:\n\n{chunk}')
        summaries.append(summary)

    # Reduce: synthesize all summaries
    combined = '\n\n'.join(f'Section {i+1}:\n{s}' for i, s in enumerate(summaries))
    final = call(f'Combine these section summaries into a single executive summary:\n\n{combined}')
    return final

import concurrent.futures

def parallel_step(topics):
    # Each topic can be researched in parallel — they do not depend on each other
    def research_topic(topic):
        return call(f'List 5 key facts about: {topic}')

    with concurrent.futures.ThreadPoolExecutor() as executor:
        results = list(executor.map(research_topic, topics))

    # Then synthesize in a sequential step
    combined = '\n\n'.join(f'{t}:\n{r}' for t, r in zip(topics, results))
    synthesis = call(f'Write a comparative analysis of these topics:\n\n{combined}')
    return synthesis

print('Parallel steps can reduce total latency.')

import anthropic
import functools

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

@functools.lru_cache(maxsize=128)
def cached_call(prompt, model='claude-haiku-4-5'):
    r = client.messages.create(
        model=model,
        max_tokens=500,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return r.content[0].text

# Cheap model for extraction, expensive for synthesis
def smart_chain(text):
    entities = cached_call(f'Extract entities from: {text}', model='claude-haiku-4-5')
    synthesis = cached_call(f'Analyze these entities: {entities}', model='claude-opus-4-5')
    return synthesis

chain_spec = {
    'name': 'Blog Post Generator',
    'steps': [
        {
            'id': 'step_1',
            'name': 'Outline Generation',
            'model': 'claude-haiku-4-5',
            'input': 'topic (string)',
            'output': 'outline (string, 5 bullet points)',
            'prompt_template': 'outline_prompt.txt'
        },
        {
            'id': 'step_2',
            'name': 'Draft Writing',
            'model': 'claude-opus-4-5',
            'input': 'outline from step_1',
            'output': 'draft (string, ~800 words)',
            'prompt_template': 'draft_prompt.txt'
        },
        {
            'id': 'step_3',
            'name': 'Editorial Polish',
            'model': 'claude-haiku-4-5',
            'input': 'draft from step_2',
            'output': 'final post (string)',
            'prompt_template': 'polish_prompt.txt'
        }
    ]
}

print('Chain documented with step specs.')

def competitive_analysis_chain(competitor_list, product_description):
    # Step 1: Research each competitor (parallel)
    def research(competitor):
        return call(f'List key features, pricing model, and target market for: {competitor}')

    with concurrent.futures.ThreadPoolExecutor() as ex:
        research_results = dict(zip(competitor_list, ex.map(research, competitor_list)))

    # Step 2: Compare against our product
    comparison_input = '\n\n'.join(f'{k}:\n{v}' for k, v in research_results.items())
    comparison = call(f'Compare these competitors against our product:\nOur product: {product_description}\n\nCompetitors:\n{comparison_input}')

    # Step 3: Strategic recommendations
    recommendations = call(f'Based on this competitive analysis, provide 3 strategic recommendations:\n\n{comparison}')
    return recommendations