Anthropic이 제시하는 프로덕션 검증 워크플로우 패턴을 실전 코드와 함께 학습합니다.
이 문서는 Anthropic이 실전 프로덕션 환경에서 검증한 5가지 워크플로우 패턴을 다룹니다. 각 패턴은 특정 문제 유형에 최적화되어 있으며, 조합하여 사용할 수 있습니다.
핵심 원칙: 각 패턴은 단순함을 유지하면서도 특정 문제를 효과적으로 해결합니다.
| 패턴 | 적합한 경우 | 예시 |
|---|---|---|
| Prompt Chaining | 순차적 단계 분해 | 콘텐츠 생성 → 번역 |
| Routing | 입력 분류 필요 | 고객 문의 분류 |
| Parallelization | 독립적 병렬 처리 | 보안 검토 (여러 관점) |
| Orchestrator-Workers | 동적 작업 분해 | 다중 파일 코드 수정 |
| Evaluator-Optimizer | 반복적 개선 | 번역 품질 향상 |
작업을 순차적 단계로 분해하고, 각 LLM 호출이 이전 출력을 처리합니다. 프로그래밍 방식의 검증 게이트를 추가하여 품질을 보장합니다.
마케팅 콘텐츠 다국어 번역:
class PromptChainingWorkflow:
async def execute(self, product_info):
# Step 1: Generate marketing copy
content = await self.generate_content(product_info)
# Validation gate
if not self.meets_brand_guidelines(content):
raise ValueError("Content doesn't meet brand guidelines")
# Step 2: Translate to multiple languages
translations = {}
for lang in ["ko", "ja", "zh", "es"]:
translation = await self.translate(content, lang)
# Validation gate for each translation
if not self.validate_translation(translation, lang):
# Retry with more specific instructions
translation = await self.translate_with_context(
content, lang, "Maintain marketing tone"
)
translations[lang] = translation
# Step 3: Format for distribution
formatted = await self.format_for_channels(translations)
return formatted
async def generate_content(self, product_info):
prompt = f"""Generate compelling marketing copy for:
Product: {product_info.name}
Features: {product_info.features}
Target: {product_info.target_audience}
Requirements:
- Professional tone
- Highlight key benefits
- Include call-to-action
- Max 150 words
"""
return await llm_call(prompt)
def meets_brand_guidelines(self, content):
"""Programmatic validation"""
checks = [
len(content.split()) <= 150,
"call-to-action" in content.lower() or "지금" in content,
not any(banned in content for banned in BANNED_WORDS)
]
return all(checks)
검증 게이트는 필수입니다. 다음 단계로 넘어가기 전에 출력을 검증하지 않으면, 에러가 누적되어 최종 결과가 나빠집니다.
입력을 분류하고 전문화된 핸들러로 라우팅합니다. 범용 응답 대신 개별 입력 카테고리에 최적화된 처리를 제공합니다.
고객 서비스 쿼리 분류:
class RoutingWorkflow:
def __init__(self):
self.handlers = {
"refund": RefundHandler(),
"technical": TechnicalSupportHandler(),
"general": GeneralInquiryHandler(),
"complaint": ComplaintHandler()
}
async def execute(self, user_query, customer_id):
# Step 1: Classify input
category = await self.classify_query(user_query)
# Step 2: Route to specialized handler
handler = self.handlers.get(category)
if not handler:
# Fallback to general handler
handler = self.handlers["general"]
# Step 3: Execute with specialized logic
response = await handler.handle(user_query, customer_id)
return {
"category": category,
"response": response,
"handler": handler.__class__.__name__
}
async def classify_query(self, query):
"""LLM-based classification"""
prompt = f"""Classify this customer query into ONE category:
Categories:
- refund: Customer wants money back
- technical: Product not working, needs help
- complaint: Customer is unhappy, expressing frustration
- general: Questions, information requests
Query: {query}
Respond with ONLY the category name."""
category = await llm_call(prompt)
return category.strip().lower()
class RefundHandler:
async def handle(self, query, customer_id):
# 환불 관련 특화 처리
order_history = await get_order_history(customer_id)
eligible = await check_refund_eligibility(order_history)
if eligible:
return await self.process_refund(query, customer_id)
else:
return await self.explain_refund_policy(query)
async def process_refund(self, query, customer_id):
prompt = f"""Generate a refund confirmation message.
Context:
- Customer is eligible for refund
- Professional, empathetic tone
- Explain next steps clearly
Query: {query}
"""
return await llm_call(prompt)
class TechnicalSupportHandler:
async def handle(self, query, customer_id):
# 기술 지원 특화: 문서 검색 + 단계별 가이드
relevant_docs = await search_documentation(query)
prompt = f"""Provide technical support.
Documentation: {relevant_docs}
Query: {query}
Provide:
1. Clear diagnosis
2. Step-by-step solution
3. Alternative solutions if available
"""
return await llm_call(prompt)
카테고리가 명확하지 않은 경우 의미 기반 라우팅을 사용합니다:
class SemanticRouter:
def __init__(self):
self.category_embeddings = self.load_category_embeddings()
async def route(self, query):
# 쿼리 임베딩 생성
query_embedding = await get_embedding(query)
# 가장 유사한 카테고리 찾기
similarities = {
category: cosine_similarity(query_embedding, emb)
for category, emb in self.category_embeddings.items()
}
best_category = max(similarities, key=similarities.get)
confidence = similarities[best_category]
# 신뢰도가 낮으면 일반 핸들러 사용
if confidence < 0.7:
return "general"
return best_category
여러 LLM 호출을 동시에 실행합니다. 두 가지 메커니즘이 있습니다:
Sectioning 사용 시:
Voting 사용 시:
보안 코드 검토 (Voting):
여러 독립 평가자가 취약점을 찾고, 합의를 통해 심각도 결정
대규모 문서 분석 (Sectioning):
문서를 섹션으로 나누어 병렬 분석
class SectioningWorkflow:
async def analyze_large_document(self, document):
# 문서를 섹션으로 분할
sections = self.split_document(document, chunk_size=2000)
# 병렬 분석
tasks = [
self.analyze_section(section, index)
for index, section in enumerate(sections)
]
results = await asyncio.gather(*tasks)
# 결과 통합
final_analysis = await self.synthesize_results(results)
return final_analysis
async def analyze_section(self, section, index):
prompt = f"""Analyze this document section (part {index + 1}).
Section: {section}
Provide:
- Key points
- Important entities
- Action items
"""
return await llm_call(prompt)
async def synthesize_results(self, section_results):
prompt = f"""Synthesize these section analyses into a coherent summary.
Section analyses:
{json.dumps(section_results, indent=2)}
Provide:
- Overall summary
- Combined key points (remove duplicates)
- Prioritized action items
"""
return await llm_call(prompt)
class VotingWorkflow:
async def security_review(self, code_changes, num_reviewers=3):
# 여러 독립 검토자가 동시에 분석
tasks = [
self.independent_review(code_changes, reviewer_id=i)
for i in range(num_reviewers)
]
reviews = await asyncio.gather(*tasks)
# 발견된 취약점 집계
vulnerabilities = self.aggregate_findings(reviews)
# 합의 기반 심각도 결정
for vuln in vulnerabilities:
vuln.severity = self.calculate_consensus_severity(
vuln, reviews
)
return SecurityReport(vulnerabilities)
async def independent_review(self, code, reviewer_id):
"""각 검토자는 독립적으로 분석"""
prompt = f"""You are security reviewer #{reviewer_id}.
Review this code for vulnerabilities.
Code:
{code}
Find:
- SQL injection risks
- XSS vulnerabilities
- Authentication issues
- Data exposure risks
For each finding, rate severity (1-10).
"""
return await llm_call(prompt)
def aggregate_findings(self, reviews):
"""중복 제거 및 집계"""
all_findings = []
for review in reviews:
all_findings.extend(review.vulnerabilities)
# 유사한 발견 사항 그룹화
grouped = self.group_similar_findings(all_findings)
# 2명 이상이 발견한 것만 포함
confirmed = [
finding for finding in grouped
if finding.detected_by >= 2
]
return confirmed
def calculate_consensus_severity(self, vuln, reviews):
"""합의 기반 심각도"""
severities = [
r.get_severity(vuln) for r in reviews
if r.found_vulnerability(vuln)
]
# 중간값 사용 (극단값 제외)
return statistics.median(severities)
# ❌ 나쁜 예: 순차 실행
async def bad_parallel():
result1 = await task1() # 기다림
result2 = await task2() # 기다림
result3 = await task3() # 기다림
return [result1, result2, result3]
# ✅ 좋은 예: 진짜 병렬 실행
async def good_parallel():
results = await asyncio.gather(
task1(),
task2(),
task3()
)
return results
# ✅ 더 좋은 예: 에러 처리 포함
async def robust_parallel():
results = await asyncio.gather(
task1(),
task2(),
task3(),
return_exceptions=True # 하나 실패해도 계속
)
# 성공한 결과만 필터링
successful = [r for r in results if not isinstance(r, Exception)]
return successful
하나의 Orchestrator LLM이 복잡한 작업을 동적으로 분해하고, Worker LLM들에게 위임한 후 결과를 통합합니다. 하위 작업이 사전 정의가 아닌 분석에서 도출됩니다.
| 특징 | Parallelization | Orchestrator-Workers |
|---|---|---|
| 하위 작업 | 사전 정의됨 | 동적으로 생성됨 |
| 유연성 | 낮음 | 높음 |
| 복잡도 | 낮음 | 높음 |
| 적합 | 예측 가능한 분할 | 예측 불가능한 작업 |
다중 파일 코드 수정:
class OrchestratorWorkersWorkflow:
async def modify_codebase(self, change_request):
orchestrator = OrchestratorAgent()
# 1. Orchestrator가 작업 분석 및 분해
plan = await orchestrator.analyze_and_plan(change_request)
# 2. 동적 Worker 할당
workers = []
for subtask in plan.subtasks:
worker = WorkerAgent(subtask)
workers.append(worker.execute())
# 3. 병렬 실행
results = await asyncio.gather(*workers)
# 4. Orchestrator가 결과 통합
final = await orchestrator.synthesize(results, plan)
# 5. 검증 및 수정
if not await orchestrator.validate(final):
fixes = await orchestrator.coordinate_fixes(final)
final = await self.apply_fixes(final, fixes)
return final
class OrchestratorAgent:
async def analyze_and_plan(self, request):
"""동적 계획 수립"""
prompt = f"""Analyze this code change request and create a plan.
Request: {request}
Analyze:
1. Which files need modification?
2. What are the dependencies between changes?
3. What's the order of operations?
4. Are there any risks?
Create a detailed plan with subtasks.
"""
response = await llm_call(prompt)
# 계획을 구조화
plan = self.parse_plan(response)
# 의존성 그래프 생성
plan.dependency_graph = self.build_dependency_graph(plan.subtasks)
return plan
async def synthesize(self, worker_results, plan):
"""Worker 결과를 통합"""
prompt = f"""Synthesize these worker results into a cohesive solution.
Original plan: {plan.summary}
Worker results:
{self.format_results(worker_results)}
Ensure:
- All changes are compatible
- No conflicting modifications
- Dependencies are satisfied
- Code style is consistent
"""
return await llm_call(prompt)
async def coordinate_fixes(self, issues):
"""문제 발견 시 수정 조율"""
prompt = f"""These issues were found. Coordinate fixes.
Issues: {issues}
For each issue:
1. Identify which worker should fix it
2. Provide specific fix instructions
3. Check for ripple effects
"""
fix_plan = await llm_call(prompt)
return self.parse_fix_plan(fix_plan)
class WorkerAgent:
def __init__(self, subtask):
self.subtask = subtask
async def execute(self):
"""할당된 하위 작업 실행"""
prompt = f"""Execute this subtask:
Task: {self.subtask.description}
Files: {self.subtask.files}
Requirements: {self.subtask.requirements}
Provide:
- Modified code
- Explanation of changes
- Potential issues
"""
result = await llm_call(prompt)
# 결과 검증
if not self.validate_result(result):
result = await self.retry_with_feedback(result)
return result
async def refactor_api_example():
request = """
Refactor our REST API to use GraphQL.
- Keep backward compatibility
- Migrate endpoints gradually
- Update client libraries
"""
orchestrator = OrchestratorAgent()
# Orchestrator 분석 결과 (동적 생성)
plan = await orchestrator.analyze_and_plan(request)
# Plan {
# subtasks: [
# {id: 1, desc: "Set up GraphQL server", files: ["server/graphql/*"]},
# {id: 2, desc: "Create GraphQL schema", files: ["schema.graphql"], depends_on: [1]},
# {id: 3, desc: "Add REST-to-GraphQL adapter", files: ["adapters/*"], depends_on: [2]},
# {id: 4, desc: "Update client library", files: ["client/*"], depends_on: [3]},
# {id: 5, desc: "Write migration guide", files: ["docs/migration.md"], depends_on: [4]}
# ]
# }
# 의존성 순서대로 실행
results = await orchestrator.execute_with_dependencies(plan)
return results
Orchestrator의 품질이 전체 성공을 좌우합니다. Orchestrator가 잘못된 계획을 세우면 모든 Worker의 노력이 무용지물이 됩니다.
Evaluator LLM이 Optimizer LLM에게 반복적 피드백을 제공하여 출력을 점진적으로 개선합니다. 인간의 개선 프로세스를 모방합니다.
문학 번역:
크리에이티브 콘텐츠:
class EvaluatorOptimizerWorkflow:
def __init__(self, max_iterations=5, target_score=0.9):
self.max_iterations = max_iterations
self.target_score = target_score
async def refine_content(self, initial_input, criteria):
output = await self.create_initial(initial_input)
for iteration in range(self.max_iterations):
# Evaluator가 평가
evaluation = await self.evaluate(output, criteria)
# 목표 달성 시 종료
if evaluation.score >= self.target_score:
return {
"output": output,
"iterations": iteration + 1,
"final_score": evaluation.score
}
# Optimizer가 피드백 기반 개선
output = await self.optimize(output, evaluation.feedback)
# 최대 반복 도달
return {
"output": output,
"iterations": self.max_iterations,
"final_score": evaluation.score,
"note": "Max iterations reached"
}
async def evaluate(self, content, criteria):
"""Evaluator의 상세 평가"""
prompt = f"""Evaluate this content against criteria.
Content: {content}
Criteria:
{self.format_criteria(criteria)}
Provide:
1. Overall score (0.0 to 1.0)
2. Specific feedback for each criterion
3. Concrete suggestions for improvement
4. Examples of what's good and what needs work
Be constructive but critical.
"""
response = await llm_call(prompt, model="claude-opus-4-6")
return self.parse_evaluation(response)
async def optimize(self, current_output, feedback):
"""Optimizer가 피드백 반영"""
prompt = f"""Improve this content based on feedback.
Current version:
{current_output}
Feedback:
{feedback}
Requirements:
- Address ALL feedback points
- Preserve what's already good
- Make targeted improvements
- Don't over-correct
"""
improved = await llm_call(prompt, model="claude-sonnet-4-5")
return improved
# 실전 예시: 문학 번역
class LiteraryTranslationWorkflow(EvaluatorOptimizerWorkflow):
async def translate_literature(self, original_text, target_lang):
criteria = {
"accuracy": "Preserves original meaning",
"style": "Maintains literary style and tone",
"cultural": "Adapts cultural references appropriately",
"flow": "Reads naturally in target language",
"nuance": "Captures subtle emotional nuances"
}
# 초벌 번역
initial_translation = await self.initial_translate(
original_text, target_lang
)
# 반복적 개선
result = await self.refine_content(initial_translation, criteria)
return result
async def initial_translate(self, text, target_lang):
prompt = f"""Translate this literary text to {target_lang}.
Original:
{text}
This is a first draft. Focus on accuracy and natural flow.
"""
return await llm_call(prompt)
여러 Evaluator가 다른 관점에서 평가:
class MultiEvaluatorWorkflow:
async def refine_with_multiple_evaluators(self, content):
evaluators = {
"technical": TechnicalEvaluator(),
"style": StyleEvaluator(),
"audience": AudienceEvaluator()
}
output = content
for iteration in range(self.max_iterations):
# 모든 Evaluator가 평가
evaluations = await asyncio.gather(*[
evaluator.evaluate(output)
for evaluator in evaluators.values()
])
# 모든 평가가 목표 이상이면 종료
if all(e.score >= self.target_score for e in evaluations):
break
# 가장 낮은 점수를 받은 측면에 집중
weakest = min(evaluations, key=lambda e: e.score)
# 해당 측면 개선
output = await self.optimize_for_aspect(
output, weakest.aspect, weakest.feedback
)
return output
무한 루프 위험: 반드시 최대 반복 횟수를 설정하고, 개선이 없으면 조기 종료해야 합니다.
실전에서는 여러 패턴을 조합하여 사용합니다.
class ComprehensiveContentSystem:
async def create_content(self, topic, requirements):
# 1. Routing: 콘텐츠 타입 분류
content_type = await self.classify_content_type(topic)
# 2. Orchestrator-Workers: 섹션별 작성
orchestrator = ContentOrchestrator()
plan = await orchestrator.create_outline(topic, content_type)
section_tasks = [
self.write_section(section)
for section in plan.sections
]
sections = await asyncio.gather(*section_tasks) # Parallelization
# 3. Prompt Chaining: 통합 및 편집
combined = await self.combine_sections(sections)
edited = await self.edit_for_coherence(combined)
# 4. Evaluator-Optimizer: 품질 개선
evaluator_optimizer = EvaluatorOptimizerWorkflow()
final = await evaluator_optimizer.refine_content(
edited, requirements
)
return final
워크플로우 패턴을 익혔다면 Part 3: Tool 설계와 Agent Skills로 진행하여 Agent의 핵심인 Tool 설계를 학습하세요.