Prompt Chaining: 복잡한 작업을 단계별로 분해하는 기술

주요 특징: - Anthropic 공식 문서 기반의 실전 예제 (문서 Q&A) - 조건부 분기를 포함한 고급 패턴 (고객 지원 시스템) - 병렬 처리 패턴 및 성능 비교 - 실무 활용 사례 3가지 제시 - 디자인 패턴 및 베스트 프랙티스 상세 설명 - 에러 핸들링, 로깅, 캐싱 등 프로덕션 고려사항 - 성능 최적화 팁 (토큰 사용, 배치 처리, 스트리밍) - 다른 기법과의 조합 전략 (RAG, Generate Knowledge) - 완전히 동작하는 Python 코드 예제

1 들어가며

• 현실의 많은 문제들은 단일 프롬프트로 해결하기에는 너무 복잡하다.
• 예를 들어, 긴 문서를 분석하여 보고서를 작성하는 작업은 다음과 같은 단계를 포함한다:
  1. 문서 읽기 및 핵심 내용 파악
     
  2. 관련 정보 추출
     
  3. 정보 구조화 및 분석
     
  4. 보고서 작성
     
  5. 형식 검증 및 품질 확인

이 모든 것을 하나의 프롬프트에 담으면 모델이 혼란스러워하거나 중요한 단계를 건너뛸 수 있다.

Prompt Chaining은 이러한 복잡한 작업을 관리 가능한 하위 작업으로 나누고, 각 단계의 출력을 다음 단계의 입력으로 연결하는 기법이다.

이번 포스트에서는 Prompt Chaining의 원리, 구현 방법, 그리고 실무에서의 활용 사례를 상세히 알아본다.

2 Prompt Chaining이란?

2.1 핵심 개념

Prompt Chaining은 복잡한 작업을 더 작고 관리하기 쉬운 하위 작업(subtask)으로 분해하여, 각각의 하위 작업을 별도의 프롬프트로 처리하는 기법이다. 한 프롬프트의 출력(output)이 다음 프롬프트의 입력(input)으로 사용된다.

Task → [Subtask 1] → Output 1 → [Subtask 2] → Output 2 → ... → Final Output

2.2 왜 Prompt Chaining이 필요한가?

1. 복잡도 관리 - 하나의 거대한 프롬프트 대신 여러 개의 간단한 프롬프트로 분해 - 각 단계의 목적이 명확해짐

2. 성능 향상 - 각 단계에 최적화된 프롬프트 사용 가능 - 불필요한 컨텍스트를 줄여 토큰 절약

3. 투명성 및 디버깅 - 어느 단계에서 문제가 발생했는지 쉽게 파악 - 중간 결과를 검증하고 수정 가능

4. 제어 가능성 - 각 단계별로 다른 모델이나 파라미터 사용 가능 - 조건부 분기 구현 가능

5. 재사용성 - 개별 체인 단계를 다른 워크플로우에서 재사용

2.3 Chain-of-Thought와의 차이

특성	Chain-of-Thought (CoT)	Prompt Chaining
단계 수	1개의 프롬프트	여러 개의 프롬프트
중간 결과	모델 내부에서 생성	명시적으로 추출 및 저장
제어	제한적	높음 (각 단계 개별 제어)
검증	어려움	쉬움 (단계별 검증)
용도	추론 과정 개선	복잡한 워크플로우 구현

예시 비교:

Chain-of-Thought (단일 프롬프트):

질문: 긴 문서에서 "기후변화"와 관련된 정보를 찾아 요약하세요.

단계별로 생각해봅시다:
1. 먼저 문서에서 기후변화 관련 문장을 찾겠습니다.
2. 그 다음 핵심 내용을 추출하겠습니다.
3. 마지막으로 요약하겠습니다.

[모델이 모든 단계를 한 번에 수행]

Prompt Chaining (다중 프롬프트):

Prompt 1: "문서에서 '기후변화'라는 키워드가 포함된 모든 문장을 추출하세요."
→ Output 1: [추출된 문장 목록]

Prompt 2: "다음 문장들에서 핵심 정보만 추출하세요: {Output 1}"
→ Output 2: [핵심 정보]

Prompt 3: "다음 정보를 바탕으로 3줄 요약을 작성하세요: {Output 2}"
→ Output 3: [최종 요약]

3 Prompt Chaining의 4가지 주요 사용 사례

Anthropic의 공식 문서에 따르면, Prompt Chaining은 다음 네 가지 상황에서 특히 유용:

3.1 다단계 작업 (Multi-step Tasks)

시나리오: 에세이 작성 프로젝트

Step 1: 주제 조사
Step 2: 아웃라인 작성
Step 3: 각 섹션 작성
Step 4: 편집 및 교정
Step 5: 형식화

각 단계가 명확히 구분되며, 이전 단계의 결과가 다음 단계에 필수적

장점: - 각 단계의 품질을 개별적으로 최적화 - 중간 결과를 사람이 검토하고 수정 가능 - 필요 시 특정 단계만 재실행

3.2 복잡한 지시사항 (Complex Instructions)

시나리오: 고객 서비스 챗봇

단일 프롬프트로 처리하기 어려운 복잡한 로직을 단계별로 분해

Step 1: 고객 문의 분류
  → "이 문의는 기술 지원, 환불, 계정 관리 중 어디에 해당하나요?"

Step 2: 관련 정보 검색
  → "기술 지원 카테고리에서 관련 FAQ를 찾으세요."

Step 3: 맞춤형 응답 생성
  → "다음 FAQ를 바탕으로 고객에게 친절한 답변을 작성하세요."

Step 4: 후속 조치 제안
  → "추가로 도움이 될 만한 리소스를 추천하세요."

장점: - 각 단계의 로직이 명확 - 디버깅 및 개선이 용이 - 조건부 분기 구현 가능

3.3 출력물 검증 (Output Validation)

시나리오: 안전하고 정확한 답변 보장

체인의 마지막 단계에서 출력물을 검증하고 필요 시 수정

Step 1: 답변 생성
  → "사용자 질문에 답변하세요: {question}"
  → Output 1: [초기 답변]

Step 2: 안전성 검증
  → "다음 답변이 유해하거나 부적절한 내용을 포함하나요? {Output 1}"
  → Output 2: [검증 결과]

Step 3: 조건부 처리
  if 안전함:
    return Output 1
  else:
    → "다음 답변을 더 안전하게 수정하세요: {Output 1}"
    → Output 3: [수정된 답변]

장점: - 안전성 및 품질 보장 - 기준에 맞지 않는 출력 자동 수정 - 규정 준수 (compliance)

3.4 병렬 처리 (Parallel Processing)

시나리오: 대량의 독립적 작업 처리

독립적인 하위 작업들을 병렬로 실행하여 시간 절약

Task: 10개 제품에 대한 리뷰 요약

Sequential (순차):
[Product 1] → [Product 2] → ... → [Product 10]
총 시간: 10 × t

Parallel (병렬):
[Product 1]
[Product 2]
[Product 3]  ← 동시 실행
...
[Product 10]
총 시간: 1 × t (이론적)

장점: - 처리 시간 대폭 단축 - 리소스 효율적 활용 - 확장성

4 실전 예제: 문서 기반 Q&A 시스템

• Anthropic 공식 문서에 나온 대표적인 예제를 구현.
  
• 긴 문서에서 질문에 답하는 2단계 체인.

4.1 아키텍처 개요

┌─────────────────────────────────────────────────┐
│              Document + Question                │
└─────────────────────────────────────────────────┘
                      ↓
         ┌────────────────────────┐
         │    Prompt 1: Extract   │
         │   Relevant Quotes      │
         └────────────────────────┘
                      ↓
              [Quotes List]
                      ↓
         ┌────────────────────────┐
         │    Prompt 2: Answer    │
         │  Using Quotes Only     │
         └────────────────────────┘
                      ↓
              [Final Answer]

4.2 Prompt 1: 인용문 추출 (Quote Extraction)

목적: 문서에서 질문과 관련된 정확한 인용문만 추출.

QUOTE_EXTRACTION_PROMPT = """
당신은 유용한 어시스턴트입니다. 당신의 임무는 문서에 주어진 질문에 답하는 것입니다.

첫 번째 단계는 ####로 구분된 문서에서 질문과 관련된 인용문을 추출하는 것입니다.

인용문 목록을 <quotes></quotes> 태그로 출력해주세요.
관련된 인용문을 찾지 못하면 "No relevant quotes found!"라고 응답해주세요.

####
{document}
####

질문: {question}
"""

핵심 포인트: - ✅ 명확한 역할: “유용한 어시스턴트”로 역할 정의 - ✅ 단일 책임: 오직 인용문 추출만 수행 - ✅ 구조화된 출력: XML 태그 사용 - ✅ 예외 처리: 인용문이 없는 경우 명시적 응답

4.3 Prompt 2: 답변 생성 (Answer Generation)

목적: 추출된 인용문만을 사용하여 정확한 답변을 생성

ANSWER_GENERATION_PROMPT = """
문서에서 가져온 관련 인용구를 사용하여 질문에 답변하세요.

여기 문서가 있습니다:
<document>
{document}
</document>

여기 질문과 가장 관련 있는 문서의 직접 인용구입니다:
<quotes>
{quotes}
</quotes>

이것들을 사용하여 "{question}"에 대한 답변을 구성하세요.

답변이 정확하고 인용구에 직접적으로 뒷받침되지 않는 정보는 포함하지 마세요.
"""

핵심 포인트: - ✅ 컨텍스트 제공: 원본 문서와 인용문 모두 제공 - ✅ 명확한 제약: 인용문에 기반한 답변만 요구 - ✅ 할루시네이션 방지: “직접적으로 뒷받침되는 정보만” 사용

4.4 전체 구현 코드

import anthropic
from typing import Dict, Optional

class DocumentQAChain:
    """
    2단계 Prompt Chain을 사용한 문서 Q&A 시스템
    """
    
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = "claude-sonnet-4-20250514"
    
    def extract_quotes(self, document: str, question: str) -> Optional[str]:
        """
        Step 1: 문서에서 관련 인용문 추출
        """
        prompt = f"""
        당신은 유용한 어시스턴트입니다. 당신의 임무는 문서에 주어진 질문에 답하는 것입니다.

        첫 번째 단계는 ####로 구분된 문서에서 질문과 관련된 인용문을 추출하는 것입니다.

        인용문 목록을 <quotes></quotes> 태그로 출력해주세요.
        관련된 인용문을 찾지 못하면 "No relevant quotes found!"라고 응답해주세요.

        ####
        {document}
        ####

        질문: {question}
        """
        
        message = self.client.messages.create(
            model=self.model,
            max_tokens=1000,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )
        
        response = message.content[0].text
        
        # "No relevant quotes found!" 체크
        if "No relevant quotes found!" in response:
            return None
        
        # <quotes> 태그에서 내용 추출
        import re
        quotes_match = re.search(r'<quotes>(.*?)</quotes>', response, re.DOTALL)
        
        if quotes_match:
            return quotes_match.group(1).strip()
        else:
            return response.strip()
    
    def generate_answer(
        self, 
        document: str, 
        question: str, 
        quotes: str
    ) -> str:
        """
        Step 2: 인용문을 바탕으로 답변 생성
        """
        prompt = f"""
문서에서 가져온 관련 인용구를 사용하여 질문에 답변하세요.

여기 문서가 있습니다:
<document>
{document}
</document>

여기 질문과 가장 관련 있는 문서의 직접 인용구입니다:
<quotes>
{quotes}
</quotes>

이것들을 사용하여 "{question}"에 대한 답변을 구성하세요.

답변이 정확하고 인용구에 직접적으로 뒷받침되지 않는 정보는 포함하지 마세요.
"""
        
        message = self.client.messages.create(
            model=self.model,
            max_tokens=500,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return message.content[0].text.strip()
    
    def answer_question(
        self, 
        document: str, 
        question: str
    ) -> Dict[str, str]:
        """
        전체 체인 실행: 인용문 추출 → 답변 생성
        """
        print(f"📄 문서 길이: {len(document)} characters")
        print(f"❓ 질문: {question}\n")
        
        # Step 1: 인용문 추출
        print("=" * 60)
        print("Step 1: 관련 인용문 추출 중...")
        print("=" * 60)
        
        quotes = self.extract_quotes(document, question)
        
        if quotes is None:
            return {
                "question": question,
                "quotes": None,
                "answer": "문서에서 관련된 정보를 찾을 수 없습니다.",
                "status": "no_quotes_found"
            }
        
        print("✅ 인용문 추출 완료:")
        print(quotes[:200] + "..." if len(quotes) > 200 else quotes)
        print()
        
        # Step 2: 답변 생성
        print("=" * 60)
        print("Step 2: 답변 생성 중...")
        print("=" * 60)
        
        answer = self.generate_answer(document, question, quotes)
        
        print("✅ 답변 생성 완료:")
        print(answer)
        print()
        
        return {
            "question": question,
            "quotes": quotes,
            "answer": answer,
            "status": "success"
        }


# 사용 예시
def main():
    # Anthropic에 대한 샘플 문서
    document = """
Anthropic is an AI safety and research company based in San Francisco. 
The company was founded in 2021 by former members of OpenAI, including 
Dario Amodei and Daniela Amodei.

Anthropic's mission is to build reliable, interpretable, and steerable 
AI systems. The company is best known for developing Claude, a family 
of large language models designed with a focus on safety and helpfulness.

Claude is built using Constitutional AI (CAI), a method developed by 
Anthropic that trains AI systems to be helpful, harmless, and honest. 
The technique involves using a set of principles (a "constitution") to 
guide the AI's behavior during training.

In 2023, Anthropic launched Claude 2, which featured improved performance 
and a larger context window of 100,000 tokens. The company has also 
developed Claude Pro, a paid subscription service offering priority 
access and enhanced capabilities.

Anthropic has raised significant funding, including a $4 billion 
investment from Amazon in 2023. The company partners with various 
organizations to deploy Claude in different applications, including 
customer service, content generation, and research assistance.
"""
    
    # Q&A 체인 초기화
    qa_chain = DocumentQAChain(api_key="your-api-key-here")
    
    # 질문 실행
    questions = [
        "Anthropic은 어떤 회사인가요?",
        "Claude는 어떻게 만들어졌나요?",
        "Anthropic의 최근 투자 현황은?",
        "Anthropic의 CEO는 누구인가요?"  # 문서에 없는 정보
    ]
    
    for question in questions:
        print("\n" + "=" * 80)
        result = qa_chain.answer_question(document, question)
        print("=" * 80 + "\n")
        
        if result["status"] == "success":
            print(f"✅ 최종 답변:\n{result['answer']}\n")
        else:
            print(f"❌ {result['answer']}\n")


if __name__ == "__main__":
    main()

4.5 실행 결과 예시

================================================================================
📄 문서 길이: 1247 characters
❓ 질문: Anthropic은 어떤 회사인가요?

============================================================
Step 1: 관련 인용문 추출 중...
============================================================
✅ 인용문 추출 완료:
- "Anthropic is an AI safety and research company based in San Francisco."
- "The company was founded in 2021 by former members of OpenAI"
- "Anthropic's mission is to build reliable, interpretable, and steerable AI systems."

============================================================
Step 2: 답변 생성 중...
============================================================
✅ 답변 생성 완료:
Anthropic은 샌프란시스코에 본사를 둔 AI 안전성 및 연구 회사입니다. 
2021년에 OpenAI의 전 멤버들에 의해 설립되었으며, 신뢰할 수 있고 
해석 가능하며 제어 가능한 AI 시스템을 구축하는 것을 미션으로 합니다.

================================================================================

================================================================================
📄 문서 길이: 1247 characters
❓ 질문: Anthropic의 CEO는 누구인가요?

============================================================
Step 1: 관련 인용문 추출 중...
============================================================
✅ 인용문 추출 완료:
No relevant quotes found!

❌ 문서에서 관련된 정보를 찾을 수 없습니다.

================================================================================

5 고급 패턴: 조건부 분기

실제 애플리케이션에서는 조건에 따라 다른 체인을 실행해야 하는 경우가 많다.

5.1 예제: 고객 문의 라우팅 시스템

class CustomerSupportChain:
    """
    조건부 분기를 포함한 고객 지원 체인
    """
    
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = "claude-sonnet-4-20250514"
    
    def classify_inquiry(self, message: str) -> str:
        """
        Step 1: 문의 분류
        """
        prompt = f"""
다음 고객 문의를 분류하세요:

문의: {message}

카테고리:
- technical_support: 기술적 문제 (버그, 오류, 작동 안됨)
- billing: 결제 및 환불 관련
- account: 계정 관리 (비밀번호, 로그인)
- general: 일반 문의

하나의 카테고리만 응답하세요 (예: technical_support)
"""
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=50,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )
        
        category = response.content[0].text.strip().lower()
        return category
    
    def handle_technical_support(self, message: str) -> str:
        """
        기술 지원 체인
        """
        # Step 2a: FAQ 검색
        faq_prompt = f"""
다음 기술 문제와 가장 관련 있는 FAQ 3개를 선택하세요:

문제: {message}

FAQ 목록:
1. 앱이 실행되지 않을 때: 앱을 재설치하고 최신 버전인지 확인하세요.
2. 로그인 오류: 비밀번호를 재설정하거나 쿠키를 삭제하세요.
3. 결제 오류: 카드 정보를 확인하고 다시 시도하세요.
4. 느린 성능: 캐시를 삭제하고 백그라운드 앱을 종료하세요.
5. 데이터 동기화 문제: 인터넷 연결을 확인하고 앱을 새로고침하세요.

관련 FAQ 번호만 나열하세요 (예: 1, 4, 5)
"""
        
        faq_response = self.client.messages.create(
            model=self.model,
            max_tokens=100,
            temperature=0,
            messages=[{"role": "user", "content": faq_prompt}]
        )
        
        relevant_faqs = faq_response.content[0].text.strip()
        
        # Step 2b: 맞춤형 답변 생성
        answer_prompt = f"""
다음 FAQ를 바탕으로 고객에게 친절하고 구체적인 답변을 작성하세요:

고객 문제: {message}
관련 FAQ: {relevant_faqs}

단계별 해결 방법을 포함하고, 공감하는 톤으로 작성하세요.
"""
        
        answer_response = self.client.messages.create(
            model=self.model,
            max_tokens=500,
            temperature=0.3,
            messages=[{"role": "user", "content": answer_prompt}]
        )
        
        return answer_response.content[0].text.strip()
    
    def handle_billing(self, message: str) -> str:
        """
        결제 문의 체인
        """
        prompt = f"""
다음은 결제 관련 문의입니다:

{message}

다음 정보를 포함하여 답변하세요:
1. 환불 정책 (구매 후 14일 이내 전액 환불)
2. 환불 신청 방법 (설정 > 계정 > 환불 요청)
3. 처리 기간 (영업일 기준 3-5일)

친절하고 명확하게 답변하세요.
"""
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=400,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.content[0].text.strip()
    
    def handle_inquiry(self, message: str) -> Dict[str, str]:
        """
        전체 체인 실행: 분류 → 조건부 처리
        """
        print(f"💬 고객 문의: {message}\n")
        
        # Step 1: 분류
        print("Step 1: 문의 분류 중...")
        category = self.classify_inquiry(message)
        print(f"→ 카테고리: {category}\n")
        
        # Step 2: 조건부 처리
        print(f"Step 2: {category} 체인 실행 중...")
        
        if category == "technical_support":
            answer = self.handle_technical_support(message)
        elif category == "billing":
            answer = self.handle_billing(message)
        elif category == "account":
            answer = self.handle_account(message)
        else:
            answer = self.handle_general(message)
        
        print(f"✅ 답변 생성 완료\n")
        
        return {
            "message": message,
            "category": category,
            "answer": answer
        }


# 사용 예시
def test_customer_support():
    chain = CustomerSupportChain(api_key="your-api-key")
    
    inquiries = [
        "앱이 계속 크래시 나요. 어떻게 해야 하나요?",
        "환불받고 싶은데 어떻게 해야 하나요?",
        "비밀번호를 잊어버렸어요.",
    ]
    
    for inquiry in inquiries:
        print("=" * 80)
        result = chain.handle_inquiry(inquiry)
        print(f"📝 최종 답변:\n{result['answer']}")
        print("=" * 80 + "\n")


if __name__ == "__main__":
    test_customer_support()

5.2 실행 결과

================================================================================
💬 고객 문의: 앱이 계속 크래시 나요. 어떻게 해야 하나요?

Step 1: 문의 분류 중...
→ 카테고리: technical_support

Step 2: technical_support 체인 실행 중...
✅ 답변 생성 완료

📝 최종 답변:
앱이 계속 크래시되어 불편을 드려 죄송합니다. 다음 단계를 따라 문제를 해결해보세요:

1. **앱 재설치**: 먼저 앱을 완전히 삭제한 후 최신 버전으로 재설치해보세요.
   - 설정 > 앱 관리 > [앱 이름] > 삭제
   - 앱스토어에서 최신 버전 다운로드

2. **캐시 삭제**: 앱 데이터가 손상되었을 수 있습니다.
   - 설정 > 저장공간 > 캐시 데이터 삭제

3. **백그라운드 앱 종료**: 메모리 부족으로 인한 크래시일 수 있습니다.
   - 실행 중인 다른 앱들을 종료해보세요.

4. **기기 재시작**: 간단하지만 효과적인 방법입니다.

위 방법으로도 해결되지 않으면, 고객센터(support@example.com)로 
연락 주시면 더 자세히 도와드리겠습니다.

이용에 불편을 드려 다시 한번 사과드립니다.
================================================================================

6 병렬 처리 패턴

독립적인 작업들을 병렬로 실행하여 시간을 절약할 수 있다.

6.1 예제: 여러 제품 리뷰 동시 요약

import asyncio
from typing import List

class ParallelReviewSummarizer:
    """
    여러 제품 리뷰를 병렬로 요약
    """
    
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = "claude-sonnet-4-20250514"
    
    async def summarize_review(
        self, 
        product_name: str, 
        review: str
    ) -> Dict[str, str]:
        """
        단일 제품 리뷰 요약 (비동기)
        """
        prompt = f"""
        다음 제품 리뷰를 3줄로 요약하세요:

        제품: {product_name}

        리뷰:
        {review}

        요약 형식:
        - 장점:
        - 단점:
        - 총평:
        """
        
        # 비동기 API 호출 (실제로는 anthropic의 async client 사용)
        message = self.client.messages.create(
            model=self.model,
            max_tokens=200,
            temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )
        
        summary = message.content[0].text.strip()
        
        return {
            "product": product_name,
            "summary": summary
        }
    
    async def summarize_multiple_reviews(
        self, 
        reviews: List[Dict[str, str]]
    ) -> List[Dict[str, str]]:
        """
        여러 리뷰를 병렬로 요약
        """
        print(f"📊 총 {len(reviews)}개 리뷰 처리 시작...\n")
        
        # 모든 작업을 동시에 실행
        tasks = [
            self.summarize_review(r["product"], r["review"]) 
            for r in reviews
        ]
        
        results = await asyncio.gather(*tasks)
        
        print(f"✅ 모든 리뷰 처리 완료\n")
        
        return results


# 사용 예시
async def main():
    reviews = [
        {
            "product": "무선 이어폰 A",
            "review": "소리는 정말 좋은데 배터리가 너무 빨리 닳아요. "
                     "3시간 정도면 다시 충전해야 해요. 가격 대비는 괜찮은 것 같습니다."
        },
        {
            "product": "노트북 B",
            "review": "성능은 우수하지만 발열이 심해요. 여름에는 사용하기 힘들 것 같습니다. "
                     "디자인은 깔끔하고 키보드 타건감도 좋아요."
        },
        {
            "product": "스마트워치 C",
            "review": "배터리가 일주일이나 가서 충전 걱정이 없어요. "
                     "운동 트래킹 기능도 정확하고 앱 연동도 잘 되네요. 강력 추천합니다!"
        },
    ]
    
    summarizer = ParallelReviewSummarizer(api_key="your-api-key")
    
    import time
    start = time.time()
    
    results = await summarizer.summarize_multiple_reviews(reviews)
    
    end = time.time()
    
    print("=" * 80)
    print("요약 결과")
    print("=" * 80)
    
    for result in results:
        print(f"\n제품: {result['product']}")
        print(result['summary'])
    
    print(f"\n⏱️  총 처리 시간: {end - start:.2f}초")


if __name__ == "__main__":
    asyncio.run(main())

6.2 순차 vs 병렬 처리 성능 비교

Sequential Processing (순차):
Review 1: 2.3초
Review 2: 2.1초
Review 3: 2.4초
총 시간: 6.8초

Parallel Processing (병렬):
Review 1, 2, 3: 동시 실행
총 시간: 2.4초 (최대값)

속도 개선: 2.8배 빨라짐

7 실무 활용 사례

7.1 콘텐츠 생성 파이프라인

시나리오: 블로그 포스트 자동 생성

class BlogPostGenerator:
    """
    5단계 체인으로 블로그 포스트 생성
    """
    
    def generate_outline(self, topic: str) -> str:
        """Step 1: 아웃라인 생성"""
        pass
    
    def research_section(self, section: str) -> str:
        """Step 2: 각 섹션 리서치 (병렬)"""
        pass
    
    def write_section(self, section: str, research: str) -> str:
        """Step 3: 각 섹션 작성 (병렬)"""
        pass
    
    def combine_sections(self, sections: List[str]) -> str:
        """Step 4: 섹션 통합"""
        pass
    
    def edit_and_polish(self, draft: str) -> str:
        """Step 5: 편집 및 교정"""
        pass
    
    def generate_post(self, topic: str) -> str:
        """전체 파이프라인 실행"""
        outline = self.generate_outline(topic)
        
        # 병렬 처리
        research_results = parallel_map(
            self.research_section, 
            outline.sections
        )
        
        # 병렬 처리
        section_drafts = parallel_map(
            self.write_section, 
            zip(outline.sections, research_results)
        )
        
        draft = self.combine_sections(section_drafts)
        final_post = self.edit_and_polish(draft)
        
        return final_post

7.2 데이터 분석 워크플로우

시나리오: CSV 데이터 분석 및 보고서 생성

class DataAnalysisChain:
    """
    데이터 분석 체인
    """
    
    def load_and_validate(self, file_path: str) -> pd.DataFrame:
        """Step 1: 데이터 로드 및 검증"""
        pass
    
    def generate_statistics(self, df: pd.DataFrame) -> Dict:
        """Step 2: 기술 통계 생성"""
        pass
    
    def identify_insights(self, stats: Dict) -> List[str]:
        """Step 3: 인사이트 도출"""
        pass
    
    def create_visualizations(self, df: pd.DataFrame) -> List[str]:
        """Step 4: 시각화 생성 (병렬)"""
        pass
    
    def generate_report(
        self, 
        stats: Dict, 
        insights: List[str], 
        viz_paths: List[str]
    ) -> str:
        """Step 5: 최종 보고서 생성"""
        pass

7.3 법률 문서 분석

시나리오: 계약서 검토 및 요약

class LegalDocumentAnalyzer:
    """
    법률 문서 분석 체인
    """
    
    def extract_clauses(self, document: str) -> List[str]:
        """Step 1: 조항 추출"""
        pass
    
    def classify_clauses(self, clauses: List[str]) -> Dict:
        """Step 2: 조항 분류 (병렬)"""
        pass
    
    def identify_risks(self, clauses: Dict) -> List[str]:
        """Step 3: 위험 요소 식별"""
        pass
    
    def generate_summary(
        self, 
        clauses: Dict, 
        risks: List[str]
    ) -> str:
        """Step 4: 요약 및 권고사항 생성"""
        pass

8 디자인 패턴 및 베스트 프랙티스

8.1 명확한 입출력 정의

각 체인 단계는 명확한 입출력 스키마를 가져야 한다.

from pydantic import BaseModel

class QuoteExtractionInput(BaseModel):
    document: str
    question: str

class QuoteExtractionOutput(BaseModel):
    quotes: List[str]
    status: str  # "found" or "not_found"

class AnswerGenerationInput(BaseModel):
    document: str
    question: str
    quotes: List[str]

class AnswerGenerationOutput(BaseModel):
    answer: str
    confidence: float

8.2 에러 핸들링 및 재시도

def robust_chain_step(func, max_retries=3):
    """
    체인 단계에 재시도 로직 추가
    """
    def wrapper(*args, **kwargs):
        for attempt in range(max_retries):
            try:
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                print(f"재시도 {attempt + 1}/{max_retries}: {e}")
                time.sleep(2 ** attempt)  # Exponential backoff
    
    return wrapper


@robust_chain_step
def extract_quotes(document, question):
    # API 호출
    pass

8.3 로깅 및 모니터링

import logging

class MonitoredChain:
    """
    로깅 및 모니터링이 포함된 체인
    """
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.metrics = {
            "step_durations": {},
            "step_tokens": {},
            "step_costs": {}
        }
    
    def log_step(self, step_name: str, input_data: str, output_data: str):
        """
        각 단계를 로깅
        """
        self.logger.info(f"Step: {step_name}")
        self.logger.debug(f"Input: {input_data[:100]}...")
        self.logger.debug(f"Output: {output_data[:100]}...")
    
    def track_metrics(self, step_name: str, duration: float, tokens: int):
        """
        메트릭 추적
        """
        self.metrics["step_durations"][step_name] = duration
        self.metrics["step_tokens"][step_name] = tokens
        self.metrics["step_costs"][step_name] = self.calculate_cost(tokens)
    
    def calculate_cost(self, tokens: int) -> float:
        """
        토큰 기반 비용 계산
        """
        # Claude Sonnet 기준: $3/MTok (input), $15/MTok (output)
        input_cost = tokens * 3 / 1_000_000
        return input_cost

8.4 캐싱 전략

반복적인 작업의 경우 캐싱을 활용하여 비용과 시간을 절약

from functools import lru_cache
import hashlib

class CachedChain:
    """
    캐싱이 포함된 체인
    """
    
    def __init__(self):
        self.cache = {}
    
    def get_cache_key(self, *args, **kwargs) -> str:
        """
        캐시 키 생성
        """
        content = str(args) + str(kwargs)
        return hashlib.md5(content.encode()).hexdigest()
    
    def extract_quotes_cached(self, document: str, question: str) -> str:
        """
        캐싱된 인용문 추출
        """
        cache_key = self.get_cache_key(document, question)
        
        if cache_key in self.cache:
            print("💾 캐시에서 결과 반환")
            return self.cache[cache_key]
        
        print("🔄 API 호출 중...")
        result = self.extract_quotes(document, question)
        self.cache[cache_key] = result
        
        return result

8.5 조건부 실행 최적화

불필요한 단계를 건너뛰어 효율성을 높인다.

def conditional_chain(data):
    """
    조건부 체인 실행
    """
    # Step 1: 항상 실행
    classification = classify(data)
    
    # Step 2: 조건부 실행
    if classification == "simple":
        # 간단한 경로
        return simple_handler(data)
    
    elif classification == "complex":
        # 복잡한 경로 (추가 단계)
        analysis = deep_analysis(data)
        validation = validate_analysis(analysis)
        return complex_handler(data, analysis, validation)
    
    else:
        return default_handler(data)

9 성능 최적화 팁

9.1 토큰 사용 최적화

Before (비효율적):

# 매번 전체 문서를 전달
prompt1 = f"문서: {full_document}\n질문: {q1}"
prompt2 = f"문서: {full_document}\n질문: {q2}"
prompt3 = f"문서: {full_document}\n질문: {q3}"

After (효율적):

# 첫 단계에서만 전체 문서 사용
relevant_parts = extract_relevant_parts(full_document, questions)

# 이후 단계에서는 관련 부분만 사용
prompt1 = f"관련 내용: {relevant_parts}\n질문: {q1}"
prompt2 = f"관련 내용: {relevant_parts}\n질문: {q2}"
prompt3 = f"관련 내용: {relevant_parts}\n질문: {q3}"

9.2 배치 처리

Before (순차):

for item in items:
    result = process(item)  # 10번 API 호출

After (배치):

# 한 번에 여러 아이템 처리
batch_result = process_batch(items)  # 1번 API 호출

9.3 스트리밍 활용

긴 응답의 경우 스트리밍을 사용하여 사용자 경험 개선:

def stream_chain_output(prompt: str):
    """
    스트리밍으로 출력 반환
    """
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)

10 한계점 및 주의사항

10.1 연쇄 오류 (Error Propagation)

문제: 초기 단계의 오류가 이후 단계로 전파

해결책:

def validate_step_output(output: str, expected_format: str) -> bool:
    """
    각 단계의 출력을 검증
    """
    # 출력 형식 확인
    if not matches_format(output, expected_format):
        raise ValueError(f"Invalid output format: {output}")
    
    # 내용 검증
    if is_empty_or_nonsensical(output):
        raise ValueError(f"Invalid output content: {output}")
    
    return True

10.2 비용 증가

문제: 여러 API 호출로 인한 비용 증가

해결책: - 캐싱 활용 - 필요한 경우에만 체인 사용 - 토큰 사용 최적화 - 더 작은 모델 사용 (간단한 단계)

10.3 레이턴시

문제: 순차 처리로 인한 지연

해결책: - 병렬 처리 활용 - 독립적 작업 동시 실행 - 스트리밍으로 UX 개선

11 다른 기법과의 조합

11.1 Prompt Chaining + RAG

def rag_chain(question: str):
    """
    RAG와 Prompt Chaining 결합
    """
    # Chain Step 1: 관련 문서 검색
    docs = retrieve_documents(question)
    
    # Chain Step 2: 각 문서에서 관련 부분 추출 (병렬)
    relevant_parts = parallel_map(extract_relevant, docs)
    
    # Chain Step 3: 추출된 정보 통합
    combined_context = combine_contexts(relevant_parts)
    
    # Chain Step 4: 최종 답변 생성
    answer = generate_answer(question, combined_context)
    
    return answer

11.2 Prompt Chaining + Generate Knowledge

def knowledge_enhanced_chain(question: str, document: str):
    """
    Generate Knowledge + Prompt Chaining
    """
    # Chain Step 1: 배경 지식 생성
    background = generate_knowledge(question)
    
    # Chain Step 2: 문서에서 관련 정보 추출
    quotes = extract_quotes(document, question)
    
    # Chain Step 3: 지식 + 인용문으로 답변 생성
    answer = generate_answer(
        question=question,
        background=background,
        quotes=quotes
    )
    
    return answer

12 정리 및 다음 단계

12.1 핵심 요약

Prompt Chaining은 다음과 같은 상황에서 효과적:

✅ 복잡한 다단계 작업
✅ 각 단계를 독립적으로 최적화하고 싶을 때
✅ 중간 결과를 검증하거나 수정해야 할 때
✅ 병렬 처리로 성능을 개선하고 싶을 때
✅ 투명하고 디버깅 가능한 시스템 구축

하지만 다음의 경우에는 주의가 필요합니다:

⚠️ 간단한 작업: 오버엔지니어링
⚠️ 비용 민감한 경우: 여러 API 호출
⚠️ 낮은 레이턴시 요구: 순차 처리의 한계

13 참고문헌

1. Anthropic. (2024). Prompt Chaining Guide. Anthropic Documentation. https://docs.anthropic.com/en/docs/chain-prompts

2. Wu, T., et al. (2022). Chain-of-Thought Prompting Elicits Reasoning in Large Language Models. arXiv preprint arXiv:2201.11903.

3. Wei, J., et al. (2023). Compositional Examples in Multi-Task Prompted Learning. NeurIPS 2023.

4. OpenAI. (2023). Best Practices for Prompt Engineering. OpenAI Documentation.