> For the complete documentation index, see [llms.txt](https://programgarden.gitbook.io/docs/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://programgarden.gitbook.io/docs/develop/dynamic_node_guide.md). # 커스텀 노드 주입하기 (Dynamic Node) ## 1. 개요 ### 동적 노드란? \*\*동적 노드(Dynamic Node)\*\*는 `programgarden-community` 패키지에 PR하지 않고도, 런타임에 사용자가 직접 만든 노드를 워크플로우에 주입하여 실행할 수 있는 기능입니다. 기존에는 동적 로직을 추가하려면 `programgarden-community`에 플러그인을 기여해야 했습니다. 동적 노드는 이 과정 없이 **자신만의 노드를 즉시 사용**할 수 있게 해줍니다. ### 기존 플러그인 vs 동적 노드 | 항목 | 커뮤니티 플러그인 | 동적 노드 | | ------------- | -------------------------------- | -------------------- | | 등록 방식 | `programgarden-community`에 PR 제출 | 런타임에 코드로 주입 | | 배포 | 패키지 릴리스 필요 | 즉시 사용 가능 | | 사용 범위 | 모든 사용자 공유 | 주입한 세션에서만 사용 | | 네이밍 | 자유 | `Dynamic_` prefix 필수 | | credential 접근 | 가능 | 불가 (보안상 차단) | | 적합한 경우 | 범용 전략/지표 | 개인 전략, 프로토타이핑, 독자 로직 | ### 언제 사용하는가? * **개인 전략**을 워크플로우에 통합하고 싶을 때 * 새로운 지표나 로직을 **빠르게 프로토타이핑**할 때 * 외부 라이브러리(pandas, numpy 등)를 활용한 **독자적인 분석 노드**가 필요할 때 * 커뮤니티 기여 전에 **로컬에서 먼저 테스트**하고 싶을 때 *** ## 2. 핵심 개념 ### 스키마와 클래스 분리 (Lazy Loading) 동적 노드는 \*\*스키마(Schema)\*\*와 \*\*클래스(Class)\*\*를 분리하여 관리합니다. {% @mermaid/diagram content="flowchart LR A\["스키마 등록
(UI 표시용)"] -- "앱 시작 시 ──── 실행 직전" --> B\["클래스 주입
(실행용)"]" %} * **스키마**: 노드의 메타정보(타입명, 카테고리, 입출력 포트). UI에 노드 목록을 표시하는 데 사용됩니다. 클래스가 없어도 등록할 수 있습니다. * **클래스**: 실제 실행 로직이 담긴 Python 클래스. 워크플로우 실행 직전에 주입합니다. 이 분리 덕분에 앱 시작 시에는 가벼운 스키마만 로드하고, 실행에 필요한 클래스만 나중에 다운로드/주입할 수 있습니다. ### Dynamic\_ prefix 네이밍 규칙 동적 노드의 타입명은 반드시 `Dynamic_`으로 시작해야 합니다. ``` Dynamic_MyRSI ✅ Dynamic_MACDCross ✅ MyCustomNode ❌ (Dynamic_ prefix 없음) ConditionNode ❌ (기존 노드와 충돌 가능) ``` 이 규칙은: * 기존 내장 노드와의 **이름 충돌을 방지**합니다 * 워크플로우에서 동적 노드를 **쉽게 식별**할 수 있게 합니다 * `get_required_dynamic_types()`가 필요한 동적 타입을 자동으로 찾을 수 있게 합니다 ### 책임 분리 | 영역 | 라이브러리 (ProgramGarden) | 사용자 | | -------- | ----------------------------------- | ---------------- | | 스키마 저장소 | 제공 (`DynamicNodeRegistry`) | 스키마 정의 및 등록 | | 클래스 검증 | BaseNode 상속, execute() 존재, 포트 일치 검증 | 클래스 구현 | | 클래스 저장 | 검증 후 저장 | 다운로드/import 후 주입 | | 워크플로우 실행 | 주입된 클래스로 실행 | 실행 요청 | *** ## 3. 사용 흐름 (4단계) {% @mermaid/diagram content="flowchart LR A\["Step 1
스키마 등록
(앱 시작 시)"] --> B\["Step 2
워크플로우 정의
(JSON)"] --> C\["Step 3
타입 확인 &
클래스 주입"] --> D\["Step 4
실행
(Job)"]" %} ### Step 1: 스키마 등록 앱 시작 시, 사용자가 정의한 노드 스키마를 등록합니다. ```python from programgarden import WorkflowExecutor executor = WorkflowExecutor() # 스키마를 딕셔너리 리스트로 전달 executor.register_dynamic_schemas([ { "node_type": "Dynamic_RSI", "category": "condition", "description": "동적 RSI 지표", "inputs": [ {"name": "data", "type": "array", "required": True} ], "outputs": [ {"name": "rsi", "type": "number"}, {"name": "signal", "type": "string"}, ], } ]) ``` > 이 시점에서는 클래스가 없어도 됩니다. UI에 노드 목록을 표시하는 용도입니다. ### Step 2: 워크플로우 정의 동적 노드를 포함한 워크플로우를 JSON으로 정의합니다. ```json { "nodes": [ {"id": "start", "type": "StartNode"}, {"id": "rsi", "type": "Dynamic_RSI", "period": 14} ], "edges": [ {"from": "start", "to": "rsi"} ] } ``` ### Step 3: 필요 타입 확인 & 클래스 주입 워크플로우에서 사용되는 동적 타입을 확인하고, 해당 클래스를 주입합니다. ```python # 필요한 동적 타입 확인 required = executor.get_required_dynamic_types(workflow) # → ["Dynamic_RSI"] # 클래스 주입 executor.inject_node_classes({ "Dynamic_RSI": DynamicRSINode, }) # 준비 완료 확인 (선택) assert executor.is_dynamic_node_ready("Dynamic_RSI") ``` ### Step 4: 실행 & 정리 ```python # 검증 validation = executor.validate(workflow) assert validation.is_valid # 실행 job = await executor.execute(workflow) # ... 워크플로우 실행 ... # 정리 (실행 후 메모리 해제) await job.stop() executor.clear_injected_classes() ``` *** ## 4. 동적 노드 클래스 작성법 ### 기본 구조 동적 노드는 `BaseNode`을 상속하고, `execute()` 메서드를 구현해야 합니다. ```python from typing import Dict, Any, List from programgarden_core.nodes.base import BaseNode, NodeCategory, OutputPort class DynamicRSINode(BaseNode): # 필수: 타입명 (스키마의 node_type과 일치해야 함) type: str = "Dynamic_RSI" # 필수: 카테고리 category: NodeCategory = NodeCategory.CONDITION # 선택: 설정 필드 (Pydantic 필드) period: int = 14 # 필수: 출력 포트 정의 (스키마의 outputs와 이름이 일치해야 함) _outputs: List[OutputPort] = [ OutputPort(name="rsi", type="number"), OutputPort(name="signal", type="string"), ] # 필수: 실행 로직 async def execute(self, context) -> Dict[str, Any]: # context에서 입력 데이터 접근 가능 # 계산 로직 수행 후 출력 포트에 맞는 딕셔너리 반환 return { "rsi": 35.5, "signal": "oversold", } ``` ### BaseNode 상속 모든 동적 노드는 `programgarden_core.nodes.base.BaseNode`을 상속해야 합니다. ```python from programgarden_core.nodes.base import BaseNode ``` `BaseNode`은 Pydantic `BaseModel` 기반이므로, 설정 필드를 Pydantic 필드로 선언할 수 있습니다. ### \_outputs 정의 (OutputPort) 스키마에 정의한 출력 포트와 **이름이 일치하는** `OutputPort`를 클래스에 선언해야 합니다. ```python from programgarden_core.nodes.base import OutputPort _outputs: List[OutputPort] = [ OutputPort(name="rsi", type="number"), OutputPort(name="signal", type="string"), ] ``` `OutputPort` 필드: | 필드 | 타입 | 필수 | 설명 | | -------------- | --- | :-: | ------------------------------------------------ | | `name` | str | O | 포트 이름 (스키마의 outputs와 일치) | | `type` | str | O | 데이터 타입 (`number`, `string`, `array`, `object` 등) | | `display_name` | str | | 사용자 표시용 이름 | | `description` | str | | 포트 설명 | ### \_inputs 정의 (InputPort, 선택) 입력 포트도 선언할 수 있습니다. ```python from programgarden_core.nodes.base import InputPort _inputs: List[InputPort] = [ InputPort(name="data", type="array", required=True), ] ``` `InputPort` 필드: | 필드 | 타입 | 필수 | 설명 | | ------------- | ---- | :-: | ---------------------------- | | `name` | str | O | 포트 이름 | | `type` | str | O | 데이터 타입 | | `required` | bool | | 필수 여부 (기본: `True`) | | `multiple` | bool | | 여러 엣지 연결 가능 여부 (기본: `False`) | | `description` | str | | 포트 설명 | ### execute() 메서드 구현 `execute()`는 비동기 메서드(`async def`)로 구현하며, `Dict[str, Any]`를 반환합니다. ```python async def execute(self, context) -> Dict[str, Any]: # 1. 입력 데이터 접근 # 2. 계산 로직 수행 # 3. 출력 포트에 맞는 딕셔너리 반환 return { "output_port_name": value, } ``` 반환 딕셔너리의 **키**는 `_outputs`에 정의한 포트 이름과 일치해야 합니다. ### 설정 필드 추가 Pydantic 필드로 설정값을 선언하면, 워크플로우 JSON에서 값을 전달받을 수 있습니다. ```python class DynamicRSINode(BaseNode): type: str = "Dynamic_RSI" category: NodeCategory = NodeCategory.CONDITION # 설정 필드 period: int = 14 overbought: float = 70.0 oversold: float = 30.0 # ... ``` 워크플로우 JSON에서: ```json { "id": "rsi", "type": "Dynamic_RSI", "period": 21, "overbought": 75.0, "oversold": 25.0 } ``` *** ## 5. API 레퍼런스 ### WorkflowExecutor 메서드 | 메서드 | 설명 | 반환 | | -------------------------------------- | ----------------------------- | ----------- | | `register_dynamic_schemas(schemas)` | 스키마 딕셔너리 리스트 등록 | `None` | | `get_required_dynamic_types(workflow)` | 워크플로우에 필요한 동적 타입 목록 | `List[str]` | | `inject_node_classes(node_classes)` | 노드 클래스 주입 (`{타입명: 클래스}`) | `None` | | `is_dynamic_node_ready(node_type)` | 실행 준비 완료 여부 (스키마 등록 + 클래스 주입) | `bool` | | `list_dynamic_node_types()` | 등록된 동적 노드 타입 목록 | `List[str]` | | `clear_injected_classes()` | 주입된 클래스 초기화 (메모리 정리) | `None` | #### register\_dynamic\_schemas(schemas) ```python executor.register_dynamic_schemas([ { "node_type": "Dynamic_RSI", # 필수, Dynamic_ prefix "category": "condition", # 선택, 기본값: "data" "description": "동적 RSI", # 선택 "inputs": [...], # 선택, 입력 포트 정의 "outputs": [...], # 선택, 출력 포트 정의 "config_schema": {...}, # 선택, 설정 필드 스키마 "version": "1.0.0", # 선택, 기본값: "1.0.0" "author": "홍길동", # 선택 } ]) ``` **Raises:** `ValueError` - `Dynamic_` prefix가 없는 경우 #### get\_required\_dynamic\_types(workflow) ```python required = executor.get_required_dynamic_types(workflow) # → ["Dynamic_RSI", "Dynamic_MACD"] ``` 워크플로우 JSON의 `nodes` 배열에서 `Dynamic_`으로 시작하는 타입만 추출합니다. #### inject\_node\_classes(node\_classes) ```python executor.inject_node_classes({ "Dynamic_RSI": DynamicRSINode, "Dynamic_MACD": DynamicMACDNode, }) ``` **검증 항목 (자동):** 1. 스키마 등록 여부 2. `BaseNode` 상속 여부 3. `execute()` 메서드 존재 여부 4. 스키마와 클래스의 출력 포트 일치 여부 **Raises:** * `ValueError` - 스키마 미등록, 포트 불일치 * `TypeError` - BaseNode 미상속, execute() 미구현 #### is\_dynamic\_node\_ready(node\_type) ```python if executor.is_dynamic_node_ready("Dynamic_RSI"): print("실행 가능!") ``` 스키마 등록 **AND** 클래스 주입이 모두 완료되었을 때 `True`를 반환합니다. #### clear\_injected\_classes() ```python executor.clear_injected_classes() ``` 주입된 클래스만 제거합니다. **스키마는 유지됩니다.** 워크플로우 실행 후 메모리 정리 용도로 사용합니다. ### DynamicNodeSchema 필드 | 필드 | 타입 | 필수 | 기본값 | 설명 | | --------------- | ------------ | :-: | --------- | --------------------------------------------- | | `node_type` | `str` | O | - | 고유 타입명 (`Dynamic_` prefix 필수) | | `category` | `str` | | `"data"` | 노드 카테고리 | | `description` | `str` | | `None` | 노드 설명 | | `inputs` | `List[Dict]` | | `[]` | 입력 포트 `[{name, type, required, description}]` | | `outputs` | `List[Dict]` | | `[]` | 출력 포트 `[{name, type, description}]` | | `config_schema` | `Dict` | | `{}` | 설정 필드 스키마 | | `version` | `str` | | `"1.0.0"` | 노드 버전 | | `author` | `str` | | `None` | 작성자 | *** ## 6. 완전한 예제 (RSI + MACD) ### 스키마 정의 ```python schemas = [ { "node_type": "Dynamic_RSI", "category": "condition", "description": "동적 RSI 지표 노드", "inputs": [ {"name": "data", "type": "array", "required": True} ], "outputs": [ {"name": "rsi", "type": "number"}, {"name": "signal", "type": "string"}, ], "config_schema": { "period": {"type": "integer", "default": 14, "min": 1, "max": 100} }, }, { "node_type": "Dynamic_MACD", "category": "condition", "description": "동적 MACD 지표 노드", "outputs": [ {"name": "macd", "type": "number"}, {"name": "signal_line", "type": "number"}, ], "config_schema": { "fast_period": {"type": "integer", "default": 12}, "slow_period": {"type": "integer", "default": 26}, }, }, ] ``` ### 클래스 구현 ```python from typing import Dict, Any, List from programgarden_core.nodes.base import BaseNode, NodeCategory, OutputPort, InputPort class DynamicRSINode(BaseNode): """동적 RSI 지표 노드""" type: str = "Dynamic_RSI" category: NodeCategory = NodeCategory.CONDITION period: int = 14 _inputs: List[InputPort] = [ InputPort(name="data", type="array", required=True), ] _outputs: List[OutputPort] = [ OutputPort(name="rsi", type="number"), OutputPort(name="signal", type="string"), ] async def execute(self, context) -> Dict[str, Any]: # 실제 RSI 계산 로직을 여기에 구현 rsi_value = 35.5 # 예시 값 if rsi_value < 30: signal = "oversold" elif rsi_value > 70: signal = "overbought" else: signal = "neutral" return { "rsi": rsi_value, "signal": signal, } class DynamicMACDNode(BaseNode): """동적 MACD 지표 노드""" type: str = "Dynamic_MACD" category: NodeCategory = NodeCategory.CONDITION fast_period: int = 12 slow_period: int = 26 _outputs: List[OutputPort] = [ OutputPort(name="macd", type="number"), OutputPort(name="signal_line", type="number"), ] async def execute(self, context) -> Dict[str, Any]: # 실제 MACD 계산 로직을 여기에 구현 return { "macd": 1.23, "signal_line": 1.10, } ``` ### 워크플로우 JSON ```json { "id": "rsi-macd-workflow", "name": "RSI + MACD 전략", "version": "1.0.0", "nodes": [ {"id": "start", "type": "StartNode"}, {"id": "rsi", "type": "Dynamic_RSI", "period": 14}, {"id": "macd", "type": "Dynamic_MACD", "fast_period": 12, "slow_period": 26} ], "edges": [ {"from": "start", "to": "rsi"}, {"from": "start", "to": "macd"} ] } ``` ### 실행 코드 ```python import asyncio from programgarden import WorkflowExecutor async def main(): executor = WorkflowExecutor() # 1. 스키마 등록 executor.register_dynamic_schemas(schemas) # 2. 워크플로우 정의 workflow = { "id": "rsi-macd-workflow", "name": "RSI + MACD 전략", "version": "1.0.0", "nodes": [ {"id": "start", "type": "StartNode"}, {"id": "rsi", "type": "Dynamic_RSI", "period": 14}, {"id": "macd", "type": "Dynamic_MACD", "fast_period": 12, "slow_period": 26}, ], "edges": [ {"from": "start", "to": "rsi"}, {"from": "start", "to": "macd"}, ], } # 3. 필요 타입 확인 & 클래스 주입 required = executor.get_required_dynamic_types(workflow) print(f"필요한 동적 타입: {required}") # → ["Dynamic_RSI", "Dynamic_MACD"] executor.inject_node_classes({ "Dynamic_RSI": DynamicRSINode, "Dynamic_MACD": DynamicMACDNode, }) # 4. 검증 validation = executor.validate(workflow) if not validation.is_valid: print(f"검증 실패: {validation.errors}") return # 5. 실행 job = await executor.execute(workflow) print(f"Job 시작: {job.job_id}") # ... 워크플로우 실행 ... # 6. 정리 await job.stop() executor.clear_injected_classes() asyncio.run(main()) ``` *** ## 7. 제약사항 & 보안 ### credential\_id 사용 불가 동적 노드에서는 `credential_id`를 사용할 수 없습니다. 이는 보안상 외부 코드가 사용자의 증권사 인증 정보에 접근하는 것을 차단하기 위함입니다. ```json { "id": "custom", "type": "Dynamic_RSI", "credential_id": "my-cred" } ``` 위와 같이 `credential_id`를 설정하면 **검증 단계에서 실패**합니다. > 증권사 API 호출이 필요한 경우, 내장 노드(OverseasStockBrokerNode 등)를 사용하고 그 출력을 동적 노드의 입력으로 연결하세요. ### 검증 규칙 클래스 주입 시(`inject_node_classes`) 자동으로 4가지 검증이 수행됩니다. | # | 검증 항목 | 실패 시 예외 | | :-: | --------------------- | -------------------------------------------------- | | 1 | 스키마 등록 여부 | `ValueError`: "스키마가 등록되지 않은 타입: Dynamic\_XXX" | | 2 | `BaseNode` 상속 여부 | `TypeError`: "BaseNode를 상속해야 함: ClassName" | | 3 | `execute()` 메서드 존재 | `TypeError`: "execute() 메서드가 없음: ClassName" | | 4 | 출력 포트 일치 (스키마 vs 클래스) | `ValueError`: "스키마에 정의된 output 포트가 클래스에 없음: {포트명}" | ### 에러 메시지 목록 | 상황 | 에러 메시지 | | -------------------------- | ------------------------------------ | | Dynamic\_ prefix 없이 스키마 등록 | `동적 노드는 'Dynamic_' prefix 필수: {타입명}` | | 스키마 없이 클래스 주입 | `스키마가 등록되지 않은 타입: {타입명}` | | BaseNode 미상속 | `BaseNode를 상속해야 함: {클래스명}` | | execute() 미구현 | `execute() 메서드가 없음: {클래스명}` | | 출력 포트 불일치 | `스키마에 정의된 output 포트가 클래스에 없음: {포트명}` | | credential\_id 사용 시도 | `credential_id를 사용할 수 없습니다` | | 클래스 미주입 상태에서 실행 | `주입되지 않음` | *** ## 8. FAQ ### Q. 스키마 등록과 클래스 주입을 왜 분리했나요? **Lazy Loading** 패턴입니다. 앱 시작 시 모든 동적 노드의 Python 코드를 로드하면 시작 시간이 길어집니다. 스키마만 먼저 등록하면 UI에 노드 목록을 빠르게 보여줄 수 있고, 실제 실행이 필요할 때만 해당 클래스를 로드합니다. ### Q. 기존 노드 타입명(예: ConditionNode)을 동적 노드 타입으로 사용할 수 있나요? 아니요. 반드시 `Dynamic_` prefix를 사용해야 합니다. 기존 노드와 이름이 충돌하면 예측할 수 없는 동작이 발생할 수 있습니다. ### Q. 동적 노드에서 외부 라이브러리(pandas 등)를 사용할 수 있나요? 네. `execute()` 메서드 안에서 자유롭게 import하여 사용할 수 있습니다. 다만, 해당 라이브러리가 실행 환경에 설치되어 있어야 합니다. ```python async def execute(self, context) -> Dict[str, Any]: import pandas as pd import numpy as np # pandas/numpy를 활용한 계산 data = pd.DataFrame(...) rsi = ... return {"rsi": rsi, "signal": "oversold"} ``` ### Q. clear\_injected\_classes() 후 스키마는 유지되나요? 네. `clear_injected_classes()`는 클래스만 제거합니다. 스키마는 그대로 유지되어 UI에 노드 목록이 계속 표시됩니다. 다음 실행 시 클래스만 다시 주입하면 됩니다. ### Q. 하나의 워크플로우에 동적 노드와 내장 노드를 함께 사용할 수 있나요? 네. 동적 노드는 내장 노드와 동일한 방식으로 워크플로우에 배치됩니다. edges로 연결하고, `{{ }}` 표현식으로 데이터를 바인딩할 수 있습니다. ```json { "nodes": [ {"id": "broker", "type": "OverseasStockBrokerNode", "credential_id": "cred-1", "paper_trading": false}, {"id": "historical", "type": "OverseasStockHistoricalDataNode"}, {"id": "custom_rsi", "type": "Dynamic_RSI", "period": 14} ], "edges": [ {"from": "broker", "to": "historical"}, {"from": "historical", "to": "custom_rsi"} ] } ``` ### Q. DynamicNodeRegistry를 직접 사용해야 하나요? 대부분의 경우 `WorkflowExecutor`의 래퍼 메서드만으로 충분합니다. `DynamicNodeRegistry`는 내부적으로 싱글톤 패턴으로 관리되며, `WorkflowExecutor`가 이를 자동으로 사용합니다. 직접 사용이 필요한 경우: ```python from programgarden_core.registry import DynamicNodeRegistry, DynamicNodeSchema registry = DynamicNodeRegistry() # Pydantic 모델로 직접 스키마 등록 schema = DynamicNodeSchema( node_type="Dynamic_MyRSI", category="condition", outputs=[{"name": "rsi", "type": "number"}], ) registry.register_schema(schema) ```