3장: 프레임워크 구축 - 브라이트웍스 AI 플랫폼

class AgentRegistry:
    """에이전트 동적 등록 및 관리"""

    def __init__(self):
        self._agents = {}

    def register(
        self,
        name: str,
        agent_class: Type,
        version: str = "v1",
        input_type: Type = None,
        output_type: Type = None,
        description: str = ""
    ):
        """에이전트 등록"""
        key = f"{name}:{version}"
        self._agents[key] = {
            "class": agent_class,
            "input_type": input_type,
            "output_type": output_type,
            "description": description,
            "instance": None
        }

    def get(self, name: str, version: str = "v1"):
        """에이전트 인스턴스 가져오기 (싱글톤)"""
        key = f"{name}:{version}"
        if key not in self._agents:
            raise ValueError(f"Agent {key} not found")

        agent_info = self._agents[key]
        if agent_info["instance"] is None:
            agent_info["instance"] = agent_info["class"]()

        return agent_info["instance"]

    def list_agents(self) -> List[str]:
        """등록된 모든 에이전트 목록"""
        return list(self._agents.keys())

# 플랫폼 초기화
platform = BrightWorksPlatform()

# 에이전트 등록
platform.registry.register(
    name="research",
    agent_class=ResearchAgent,
    input_type=dict,
    output_type=ResearchResult,
    description="리서치 및 트렌드 분석"
)

platform.registry.register(
    name="writer",
    agent_class=WriterAgent,
    input_type=ResearchResult,
    output_type=DraftPost,
    description="콘텐츠 작성"
)

# 에이전트 실행
research_agent = platform.registry.get("research")
result = research_agent.research(topic, keywords)

class BaseTool(ABC):
    """모든 도구의 기본 클래스"""

    @abstractmethod
    def execute(self, **kwargs) -> Any:
        """도구 실행"""
        pass

    @abstractmethod
    def validate_config(self) -> bool:
        """설정 유효성 검사"""
        pass


class GoogleDocsTool(BaseTool):
    """Google Docs 도구"""

    def __init__(self, credentials_path: str):
        self.creds = self._load_credentials(credentials_path)
        self.docs_service = build('docs', 'v1', credentials=self.creds)

    def execute(self, action: str, **kwargs) -> Any:
        """Google Docs 작업 실행"""
        if action == "create":
            return self._create_document(kwargs['title'], kwargs['content'])
        elif action == "read":
            return self._read_document(kwargs['document_id'])
        elif action == "update":
            return self._update_document(kwargs['document_id'], kwargs['content'])

    def _create_document(self, title: str, content: str):
        """새 문서 생성"""
        doc = self.docs_service.documents().create(
            body={'title': title}
        ).execute()
        # 콘텐츠 추가
        self._update_document(doc['documentId'], content)
        return doc['documentId']

class ToolLibrary:
    """도구 관리 라이브러리"""

    def __init__(self):
        self._tools = {}

    def register_tool(self, name: str, tool: BaseTool):
        """도구 등록"""
        self._tools[name] = tool

    def get_tool(self, name: str) -> BaseTool:
        """도구 가져오기"""
        if name not in self._tools:
            raise ValueError(f"Tool {name} not found")
        return self._tools[name]

    def execute(self, tool_name: str, action: str, **kwargs):
        """도구 실행 (간편 메서드)"""
        tool = self.get_tool(tool_name)
        return tool.execute(action, **kwargs)


# 사용 예시
library = ToolLibrary()
library.register_tool("google_docs", GoogleDocsTool(creds_path))
library.register_tool("slack", SlackTool(webhook_url))

# 에이전트가 도구 사용
doc_id = library.execute(
    "google_docs",
    "create",
    title="AI 마케팅 전략",
    content=final_post.content
)

library.execute(
    "slack",
    "send_message",
    channel="#content-team",
    text=f"새 블로그 포스트 생성 완료: {final_post.title}"
)

# workflows/content_pipeline.yaml

name: "Content Creation Pipeline"
description: "Research → Write → Edit → Publish"

steps:
  - name: research
    agent: research:v1
    input:
      topic: "${input.topic}"
      keywords: "${input.keywords}"
    output: research_result

  - name: write
    agent: writer:v1
    input:
      research: "${research.output}"
      keywords: "${input.keywords}"
    output: draft_post

  - name: edit
    agent: editor:v1
    input:
      draft: "${write.output}"
    output: final_post

  - name: publish
    tool: google_docs
    action: create
    input:
      title: "${edit.output.title}"
      content: "${edit.output.content}"
    output: document_id

  - name: notify
    tool: slack
    action: send_message
    input:
      channel: "#content-team"
      text: "새 포스트 발행: ${edit.output.title} (Google Docs: ${publish.output})"

on_error:
  strategy: retry
  max_retries: 3
  fallback: slack_notification

class WorkflowEngine:
    """워크플로우 실행 엔진"""

    def __init__(self, registry: AgentRegistry, library: ToolLibrary):
        self.registry = registry
        self.library = library

    def load_workflow(self, yaml_path: str) -> Workflow:
        """YAML 파일에서 워크플로우 로드"""
        with open(yaml_path) as f:
            config = yaml.safe_load(f)
        return Workflow(config)

    def execute(self, workflow: Workflow, inputs: dict) -> dict:
        """워크플로우 실행"""
        context = {"input": inputs}

        for step in workflow.steps:
            try:
                if step.type == "agent":
                    result = self._execute_agent_step(step, context)
                elif step.type == "tool":
                    result = self._execute_tool_step(step, context)

                context[step.name] = {"output": result}

            except Exception as e:
                if workflow.on_error:
                    self._handle_error(step, e, workflow.on_error)
                else:
                    raise

        return context

    def _execute_agent_step(self, step, context):
        """에이전트 스텝 실행"""
        agent = self.registry.get(step.agent)
        inputs = self._resolve_variables(step.input, context)
        return agent.execute(**inputs)

    def _resolve_variables(self, template, context):
        """변수 치환 (${step.output.field} 형식)"""
        # 예: "${research.output.trends}" → context['research']['output']['trends']
        result = {}
        for key, value in template.items():
            if isinstance(value, str) and value.startswith("${"):
                result[key] = self._get_from_context(value, context)
            else:
                result[key] = value
        return result

# 플랫폼 초기화
platform = BrightWorksPlatform()

# 워크플로우 로드
workflow = platform.engine.load_workflow("workflows/content_pipeline.yaml")

# 실행
result = platform.engine.execute(workflow, {
    "topic": "AI 에이전트 시대의 디지털 마케팅",
    "keywords": ["AI 에이전트", "디지털 마케팅", "자동화"]
})

print(f"Published Document ID: {result['publish']['output']}")

class Monitor:
    """모니터링 및 로깅"""

    def __init__(self):
        self.metrics = defaultdict(list)

    def log_agent_execution(
        self,
        agent_name: str,
        duration: float,
        input_tokens: int,
        output_tokens: int,
        cost: float
    ):
        """에이전트 실행 기록"""
        self.metrics["executions"].append({
            "agent": agent_name,
            "timestamp": datetime.now(),
            "duration": duration,
            "tokens": {"input": input_tokens, "output": output_tokens},
            "cost": cost
        })

    def get_dashboard_data(self):
        """대시보드용 데이터 집계"""
        total_cost = sum(e["cost"] for e in self.metrics["executions"])
        total_executions = len(self.metrics["executions"])

        return {
            "total_cost": total_cost,
            "total_executions": total_executions,
            "avg_duration": np.mean([e["duration"] for e in self.metrics["executions"]]),
            "cost_by_agent": self._group_by_agent()
        }

class BrightWorksPlatform:
    """브라이트웍스 AI 플랫폼"""

    def __init__(self, config_path: str = "config.yaml"):
        # 컴포넌트 초기화
        self.registry = AgentRegistry()
        self.library = ToolLibrary()
        self.engine = WorkflowEngine(self.registry, self.library)
        self.monitor = Monitor()

        # 설정 로드
        self.config = self._load_config(config_path)

        # 에이전트 자동 등록
        self._auto_discover_agents()

        # 도구 자동 등록
        self._auto_discover_tools()

    def run_workflow(self, workflow_name: str, inputs: dict):
        """워크플로우 실행"""
        workflow = self.engine.load_workflow(f"workflows/{workflow_name}.yaml")

        with self.monitor.track_execution(workflow_name):
            result = self.engine.execute(workflow, inputs)

        return result

    def get_dashboard(self):
        """모니터링 대시보드 데이터"""
        return self.monitor.get_dashboard_data()


# 사용 예시
if __name__ == "__main__":
    platform = BrightWorksPlatform()

    # 워크플로우 실행
    result = platform.run_workflow("content_pipeline", {
        "topic": "AI 에이전트 시대의 디지털 마케팅",
        "keywords": ["AI 에이전트", "디지털 마케팅"]
    })

    print(f"✅ 완료!")
    print(f"Google Docs: {result['publish']['output']}")

    # 대시보드 확인
    dashboard = platform.get_dashboard()
    total = dashboard['total_cost']
    print(f"총 비용: {total:.2f{'}'}")