출력을 입력으로 연결하는 패턴

핵심 과제: 추출과 주입

기계 처리를 위한 1단계 설계

import anthropic
import json

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

step1_prompt = '''
<task>
Analyze the customer review below.
</task>

<review>
The onboarding was confusing and took 3 hours. The core feature works great though.
</review>

<output_format>
Return ONLY a JSON object. No other text.
{
  "sentiment": "positive|negative|mixed",
  "issues": ["string"],
  "positives": ["string"],
  "priority": "high|medium|low"
}
</output_format>
'''

response = client.messages.create(
    model='claude-opus-4-5',
    max_tokens=300,
    messages=[{'role': 'user', 'content': step1_prompt}]
)
print(response.content[0].text)

1단계 출력 구문 분석

import json

def parse_step1_output(raw_text):
    # Models sometimes wrap JSON in extra text -- strip it
    text = raw_text.strip()
    # Find the first { and last } to extract JSON object
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        text = text[start:end+1]
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        raise ValueError("Step 1 output is not valid JSON: " + str(e))

# Example usage
raw = '{"sentiment": "mixed", "issues": ["confusing onboarding"], "positives": ["core feature"], "priority": "high"}'
parsed = parse_step1_output(raw)
print(parsed['issues'])
print(parsed['priority'])

추출한 필드를 2단계에 주입하기

def build_step2_prompt(parsed_step1):
    issues = '\n'.join(f'- {issue}' for issue in parsed_step1['issues'])
    priority = parsed_step1['priority']
    sentiment = parsed_step1['sentiment']

    return f'''
<context>
A customer review was analyzed. Overall sentiment: {sentiment}. Priority: {priority}.
</context>

<task>
Write a customer support response addressing these specific issues:
{issues}
Acknowledge the positives before addressing the issues.
</task>

<output_format>
Plain text response, 3 sentences maximum, professional tone.
</output_format>
'''

parsed = {'sentiment': 'mixed', 'issues': ['confusing onboarding'], 'priority': 'high', 'positives': ['core feature']}
print(build_step2_prompt(parsed))

전체 2단계 체인

import anthropic, json

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

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

def review_response_chain(review_text):
    # Step 1: Analyze
    step1 = call(f'Analyze this review. Return JSON: {{"sentiment": str, "issues": [str], "priority": str}}\n\nReview: {review_text}')
    parsed = json.loads(step1.strip())

    # Inject into Step 2
    issues_str = ', '.join(parsed['issues'])
    step2_prompt = f'Write a 2-sentence support reply. Issues to address: {issues_str}. Priority: {parsed["priority"]}.'

    # Step 2: Draft response
    reply = call(step2_prompt)
    return reply

print(review_response_chain('Login is broken. App crashes on startup.'))

중첩된 JSON 주입 처리

step1_output = {
    'document': {
        'title': 'Q3 Report',
        'sections': [
            {'name': 'Revenue', 'value': '$4.2M', 'change': '+12%'},
            {'name': 'Users', 'value': '85,000', 'change': '+5%'},
            {'name': 'Churn', 'value': '3.2%', 'change': '-0.8%'}
        ]
    },
    'summary': 'Strong revenue quarter with moderate user growth.'
}

# Extract only what Step 2 needs — not the full nested object
def extract_for_step2(data):
    sections = data['document']['sections']
    metrics = '\n'.join(f"{s['name']}: {s['value']} ({s['change']})" for s in sections)
    return {
        'metrics': metrics,
        'summary': data['summary']
    }

step2_input = extract_for_step2(step1_output)
print(step2_input)

과도한 주입 피하기

출력에 따른 조건부 분기

def route_chain(user_message):
    # Step 1: Classify intent
    classification = json.loads(call(
        f'Classify this message as billing, technical, or general. Return JSON: {{"intent": str}}\n\nMessage: {user_message}'
    ))

    intent = classification['intent']

    # Route to specialized Step 2 prompt
    if intent == 'billing':
        prompt = f'You are a billing specialist. Address: {user_message}'
    elif intent == 'technical':
        prompt = f'You are a senior engineer. Provide technical guidance for: {user_message}'
    else:
        prompt = f'You are a general support agent. Respond to: {user_message}'

    return call(prompt)

print(route_chain('My invoice shows a wrong amount.'))

단계 간 상태 누적

def run_pipeline(initial_input):
    state = {'input': initial_input}

    # Step 1
    state['entities'] = json.loads(call(
        f'Extract entities as JSON: {{"people": [], "companies": []}}\n\n{state["input"]}'
    ))

    # Step 2 uses entities from Step 1
    companies_str = ', '.join(state['entities'].get('companies', []))
    state['company_types'] = call(
        f'Classify these companies as startup/enterprise: {companies_str}'
    )

    # Step 3 uses output from Steps 1 and 2
    state['summary'] = call(
        f'Write a 2-sentence summary.\nEntities: {state["entities"]}\nClassifications: {state["company_types"]}'
    )

    return state

result = run_pipeline('Apple and OpenAI announced a partnership with Elon Musk.')
print(result['summary'])

JSON 추출 유틸리티

import re, json

def extract_json(text):
    "Extract JSON from model output, handling extra text around the object."
    # Try direct parse first
    try:
        return json.loads(text.strip())
    except json.JSONDecodeError:
        pass
    # Try finding JSON object by bracket matching
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        try:
            return json.loads(text[start:end+1])
        except json.JSONDecodeError:
            pass
    # Try finding JSON array
    start = text.find("[")
    end = text.rfind("]")
    if start != -1 and end != -1 and end > start:
        try:
            return json.loads(text[start:end+1])
        except json.JSONDecodeError:
            pass
    raise ValueError("Could not extract JSON from: " + text[:200])

print(extract_json('{"key": "value"}'))

출력에서 입력으로 이어지는 패턴 테스트

def test_step1(review_text, expected_sentiment):
    raw = call(f'Analyze review. Return JSON: {{"sentiment": str}}\n\n{review_text}')
    parsed = extract_json(raw)
    assert parsed['sentiment'] == expected_sentiment, f'Expected {expected_sentiment}, got {parsed["sentiment"]}'
    print(f'PASS: sentiment={parsed["sentiment"]}')

# Run unit test for Step 1
test_step1('The product is excellent!', 'positive')
test_step1('This is terrible.', 'negative')

빠른 확인

출력에서 입력으로 이어지는 흐름 — 핵심 요점