4장: 고급 패턴과 최적화

class AgentFactory:
    """동적 에이전트 생성 팩토리"""

    def __init__(self, registry: AgentRegistry):
        self.registry = registry
        self.template_cache = {}

    def create_specialized_agent(
        self,
        base_agent: str,
        specialization: dict,
        version: str = "dynamic"
    ):
        """
        기본 에이전트를 특화시킨 새 에이전트 생성

        Args:
            base_agent: 기본 에이전트 이름 (예: "writer")
            specialization: 특화 설정
                - target_audience: 타겟 독자
                - tone: 톤앤매너
                - format: 출력 포맷
            version: 버전 (기본값: dynamic)
        """
        # 기본 에이전트 가져오기
        base = self.registry.get(base_agent)

        # 프롬프트 템플릿 수정
        specialized_prompt = self._modify_prompt(
            base.system_prompt,
            specialization
        )

        # 동적 클래스 생성
        class SpecializedAgent(base.__class__):
            def __init__(self):
                super().__init__()
                self.system_prompt = specialized_prompt
                self.specialization = specialization

        # 레지스트리에 등록
        agent_name = f"{base_agent}_{specialization['target_audience']}"
        self.registry.register(
            agent_name,
            SpecializedAgent,
            version=version,
            description=f"Specialized {base_agent} for {specialization}"
        )

        return agent_name

    def _modify_prompt(self, base_prompt: str, spec: dict) -> str:
        """프롬프트 템플릿 수정"""
        additions = []

        if "target_audience" in spec:
            additions.append(f"타겟 독자: {spec['target_audience']}")

        if "tone" in spec:
            additions.append(f"톤앤매너: {spec['tone']}")

        if "format" in spec:
            additions.append(f"출력 포맷: {spec['format']}")

        return base_prompt + "\n\n특화 설정:\n" + "\n".join(additions)


# 사용 예시
factory = AgentFactory(platform.registry)

# 소셜 미디어용 작가 생성
social_writer = factory.create_specialized_agent(
    base_agent="writer",
    specialization={
        "target_audience": "20-30대 소셜 미디어 사용자",
        "tone": "캐주얼하고 친근한",
        "format": "짧은 문단, 이모지 사용"
    }
)

# 기술 블로그용 작가 생성
tech_writer = factory.create_specialized_agent(
    base_agent="writer",
    specialization={
        "target_audience": "개발자",
        "tone": "전문적이고 정확한",
        "format": "코드 예시 포함, 긴 형식"
    }
)

# 생성된 에이전트 사용
agent = platform.registry.get(social_writer)
result = agent.write(research, keywords)

class CachedAgent:
    """캐싱을 지원하는 에이전트 래퍼"""

    def __init__(self, agent, cache_ttl: int = 3600):
        self.agent = agent
        self.cache_ttl = cache_ttl  # seconds
        self.cache = {}

    def execute(self, **kwargs):
        """캐시 확인 후 실행"""
        cache_key = self._make_cache_key(kwargs)

        # 캐시 확인
        if cache_key in self.cache:
            cached_data, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.cache_ttl:
                console.print("[cyan]💾 Cache Hit![/cyan]")
                return cached_data

        # 캐시 미스 - 실제 실행
        console.print("[yellow]🔥 Cache Miss - Executing...[/yellow]")
        result = self.agent.execute(**kwargs)

        # 캐시 저장
        self.cache[cache_key] = (result, time.time())

        return result

    def _make_cache_key(self, kwargs: dict) -> str:
        """입력 파라미터로 캐시 키 생성"""
        import hashlib
        import json
        serialized = json.dumps(kwargs, sort_keys=True)
        return hashlib.md5(serialized.encode()).hexdigest()


# 사용 예시
research_agent = platform.registry.get("research")
cached_research = CachedAgent(research_agent, cache_ttl=3600)  # 1시간

# 첫 실행: API 호출
result1 = cached_research.execute(
    topic="AI 마케팅",
    keywords=["AI", "마케팅"]
)  # $0.006

# 같은 입력으로 재실행: 캐시 사용
result2 = cached_research.execute(
    topic="AI 마케팅",
    keywords=["AI", "마케팅"]
)  # $0.000 (캐시!)

class PromptOptimizer:
    """프롬프트 최적화로 토큰 절감"""

    def compress_prompt(self, prompt: str, max_tokens: int = 1000):
        """
        프롬프트를 압축하여 토큰 수 감소

        전략:
        1. 불필요한 공백/줄바꿈 제거
        2. 예시 텍스트 축약
        3. 중복 표현 제거
        """
        # 공백 정리
        compressed = " ".join(prompt.split())

        # 긴 예시 제거
        compressed = self._remove_long_examples(compressed)

        # 동의어 축약
        compressed = self._replace_verbose_phrases(compressed)

        return compressed

    def _replace_verbose_phrases(self, text: str) -> str:
        """장황한 표현을 간결하게"""
        replacements = {
            "다음과 같은 작업을 수행해주세요": "작업:",
            "위의 내용을 바탕으로": "바탕:",
            "반드시 포함해야 합니다": "필수:",
        }
        for old, new in replacements.items():
            text = text.replace(old, new)
        return text


# 최적화 전후 비교
original_prompt = """
다음과 같은 작업을 수행해주세요:
1. 주제에 대한 최신 트렌드를 조사하세요.
2. 관련 통계 데이터를 수집하세요.
... (1,500 토큰)
"""

optimizer = PromptOptimizer()
compressed = optimizer.compress_prompt(original_prompt)
# 압축 후: 800 토큰 (47% 절감!)

class QAValidator:
    """품질 보증 검증기"""

    def __init__(self):
        self.rules = []

    def add_rule(self, rule: Callable):
        """검증 규칙 추가"""
        self.rules.append(rule)

    def validate(self, output: Any) -> dict:
        """
        출력 검증

        Returns:
            검증 결과 (passed, failed, warnings)
        """
        results = {
            "passed": [],
            "failed": [],
            "warnings": [],
            "score": 0
        }

        for rule in self.rules:
            try:
                status, message = rule(output)
                if status == "pass":
                    results["passed"].append(message)
                    results["score"] += 1
                elif status == "fail":
                    results["failed"].append(message)
                else:  # warning
                    results["warnings"].append(message)
            except Exception as e:
                results["failed"].append(f"Rule execution error: {str(e)}")

        # 점수 계산 (0-100)
        total_rules = len(self.rules)
        results["score"] = int((len(results["passed"]) / total_rules) * 100)

        return results


# 규칙 정의
def rule_min_word_count(output):
    """최소 단어 수 확인"""
    if hasattr(output, "word_count") and output.word_count >= 1500:
        return ("pass", "단어 수 충족 (1500+)")
    return ("fail", f"단어 수 부족: {output.word_count}")

def rule_keyword_usage(output):
    """키워드 사용 확인"""
    if hasattr(output, "keywords") and hasattr(output, "content"):
        keywords_used = sum(1 for kw in output.keywords if kw in output.content)
        if keywords_used >= len(output.keywords) * 0.8:
            return ("pass", f"키워드 사용률: {keywords_used}/{len(output.keywords)}")
        return ("warning", f"키워드 사용 부족: {keywords_used}/{len(output.keywords)}")

def rule_quality_score(output):
    """품질 점수 확인"""
    if hasattr(output, "quality_score") and output.quality_score >= 80:
        return ("pass", f"품질 점수: {output.quality_score}/100")
    return ("fail", f"품질 점수 낮음: {output.quality_score}/100")

def rule_no_placeholder_text(output):
    """플레이스홀더 텍스트 확인"""
    if hasattr(output, "content"):
        placeholders = ["TODO", "FIXME", "[insert", "XXX"]
        found = [p for p in placeholders if p in output.content]
        if not found:
            return ("pass", "플레이스홀더 없음")
        return ("fail", f"플레이스홀더 발견: {found}")


# QA Validator 사용
validator = QAValidator()
validator.add_rule(rule_min_word_count)
validator.add_rule(rule_keyword_usage)
validator.add_rule(rule_quality_score)
validator.add_rule(rule_no_placeholder_text)

# 에이전트 출력 검증
final_post = editor_agent.edit(draft)
qa_result = validator.validate(final_post)

if qa_result["score"] >= 75:
    console.print("[green]✅ QA 통과![/green]")
    publish(final_post)
else:
    console.print("[red]❌ QA 실패![/red]")
    console.print(f"Failures: {qa_result['failed']}")
    # 재시도 또는 수동 검토

class AutoRetryOrchestrator:
    """QA 피드백 기반 자동 재시도"""

    def run_with_qa(self, workflow, inputs, max_retries=3):
        """QA 검증과 함께 워크플로우 실행"""
        validator = QAValidator()
        # 규칙 등록...

        for attempt in range(max_retries):
            console.print(f"\n[cyan]시도 {attempt + 1}/{max_retries}[/cyan]")

            result = self.engine.execute(workflow, inputs)
            final_output = result["edit"]["output"]

            # QA 검증
            qa_result = validator.validate(final_output)

            if qa_result["score"] >= 75:
                console.print(f"[green]✅ QA 통과 (점수: {qa_result['score']})[/green]")
                return result

            # QA 실패 - 피드백 생성
            feedback = self._generate_feedback(qa_result)
            console.print(f"[yellow]⚠ QA 실패 - 재시도 with feedback[/yellow]")

            # 다음 시도에 피드백 추가
            inputs["qa_feedback"] = feedback

        # 최대 재시도 초과
        console.print("[red]❌ 최대 재시도 횟수 초과[/red]")
        raise QAFailureError("QA validation failed after max retries")

    def _generate_feedback(self, qa_result: dict) -> str:
        """QA 결과로부터 피드백 생성"""
        feedback_lines = ["다음 사항을 개선하세요:"]
        for failure in qa_result["failed"]:
            feedback_lines.append(f"- {failure}")
        for warning in qa_result["warnings"]:
            feedback_lines.append(f"- [주의] {warning}")
        return "\n".join(feedback_lines)

import asyncio
from concurrent.futures import ThreadPoolExecutor

class ParallelWorkflowEngine(WorkflowEngine):
    """병렬 실행을 지원하는 워크플로우 엔진"""

    async def execute_parallel(self, workflow, inputs):
        """병렬 실행 가능한 스텝 자동 감지 및 실행"""
        context = {"input": inputs}
        dependency_graph = self._build_dependency_graph(workflow)

        while dependency_graph:
            # 의존성 없는 스텝 찾기
            ready_steps = [
                step for step, deps in dependency_graph.items()
                if not deps
            ]

            if not ready_steps:
                raise RuntimeError("Circular dependency detected")

            # 병렬 실행
            console.print(f"[cyan]병렬 실행: {len(ready_steps)}개 스텝[/cyan]")
            results = await asyncio.gather(*[
                self._execute_step_async(step, context)
                for step in ready_steps
            ])

            # 결과를 컨텍스트에 추가
            for step, result in zip(ready_steps, results):
                context[step["name"]] = {"output": result}
                del dependency_graph[step["name"]]

                # 다른 스텝의 의존성 업데이트
                for deps in dependency_graph.values():
                    deps.discard(step["name"])

        return context

    def _build_dependency_graph(self, workflow):
        """워크플로우에서 의존성 그래프 생성"""
        graph = {}
        for step in workflow.steps:
            dependencies = self._extract_dependencies(step)
            graph[step["name"]] = set(dependencies)
        return graph


# 병렬 실행 예시
# 3개의 독립적인 리서치를 동시에 실행
parallel_workflow = {
    "name": "Parallel Research",
    "steps": [
        {"name": "research_trends", "agent": "research:v1", "input": {...}},
        {"name": "research_stats", "agent": "research:v1", "input": {...}},
        {"name": "research_competitors", "agent": "research:v1", "input": {...}},
        {"name": "write", "agent": "writer:v1", "input": {
            "trends": "${research_trends.output}",
            "stats": "${research_stats.output}",
            "competitors": "${research_competitors.output}"
        }}
    ]
}

# 순차 실행: 15+15+15+35 = 80초
# 병렬 실행: 15+35 = 50초 (38% 단축!)