Backends Overview

from nadoo_flow.backends import BackendRegistry

# Use native backend (default)
backend = BackendRegistry.create("native")
workflow = backend.create_workflow(config)
result = await workflow.run(input_data)

# Future implementation
backend = BackendRegistry.create("langgraph")
workflow = backend.create_workflow(config)

# Future implementation
backend = BackendRegistry.create("crewai")
workflow = backend.create_workflow(config)

# Future implementation
backend = BackendRegistry.create("a2a")
workflow = backend.create_workflow(config)

from typing import Protocol

class IBackend(Protocol):
    """Backend protocol interface"""

    async def create_workflow(self, config: dict) -> IWorkflow:
        """Create workflow from configuration"""
        ...

    async def execute(
        self,
        workflow: IWorkflow,
        input_data: dict
    ) -> ExecutionResult:
        """Execute workflow"""
        ...

    async def validate(self, workflow: IWorkflow) -> bool:
        """Validate workflow"""
        ...

# Backend-specific configuration
config = {
    "backend": "native",
    "backend_config": {
        "max_parallel": 10,
        "timeout": 300,
        "retry_strategy": "exponential"
    }
}

backend = BackendRegistry.create(
    config["backend"],
    **config["backend_config"]
)

from nadoo_flow.backends import BackendRegistry

# Register custom backend
BackendRegistry.register(
    name="custom",
    backend_class=CustomBackend
)

# List available backends
backends = BackendRegistry.list_backends()
print(backends)  # ["native", "custom"]

# Get backend information
info = BackendRegistry.get_info("native")
print(info)
# {
#     "name": "native",
#     "version": "0.1.0",
#     "features": ["async", "streaming", "memory"],
#     "status": "stable"
# }

from nadoo_flow.backends import BaseBackend

class CustomBackend(BaseBackend):
    """Custom backend implementation"""

    def __init__(self, config=None):
        super().__init__(config)
        self.engine = CustomEngine()

    async def create_workflow(self, config: dict):
        """Create workflow using custom engine"""
        return CustomWorkflow(config, self.engine)

    async def execute(self, workflow, input_data):
        """Execute using custom engine"""
        return await self.engine.run(workflow, input_data)

    async def validate(self, workflow):
        """Validate workflow for custom engine"""
        return self.engine.validate(workflow)

class CustomWorkflow:
    """Adapt Flow Core workflow to custom engine"""

    def __init__(self, config, engine):
        self.config = config
        self.engine = engine
        self.nodes = self._convert_nodes(config["nodes"])

    def _convert_nodes(self, nodes):
        """Convert Flow Core nodes to engine format"""
        converted = []
        for node in nodes:
            converted.append(self._adapt_node(node))
        return converted

    async def run(self, input_data):
        """Execute workflow"""
        return await self.engine.execute(
            self.nodes,
            input_data
        )

# Workflow definition remains the same
workflow_config = {
    "nodes": [...],
    "edges": [...],
    "config": {...}
}

# Switch backends without changing workflow
native_backend = BackendRegistry.create("native")
native_result = await native_backend.execute(workflow_config, data)

# Future: Same workflow, different backend
# langgraph_backend = BackendRegistry.create("langgraph")
# langgraph_result = await langgraph_backend.execute(workflow_config, data)

# Benchmark different backends
async def benchmark_backends(workflow, input_data):
    results = {}

    for backend_name in BackendRegistry.list_backends():
        backend = BackendRegistry.create(backend_name)

        start = time.time()
        result = await backend.execute(workflow, input_data)
        duration = time.time() - start

        results[backend_name] = {
            "duration": duration,
            "success": result.success,
            "memory_used": get_memory_usage()
        }

    return results

native_config = {
    "backend": "native",
    "execution": {
        "max_parallel_nodes": 10,
        "timeout_seconds": 300,
        "retry_failed_nodes": True,
        "enable_checkpoints": True
    },
    "memory": {
        "type": "conversation",
        "max_messages": 100
    },
    "streaming": {
        "enabled": True,
        "buffer_size": 100
    }
}

# LangGraph configuration (planned)
langgraph_config = {
    "backend": "langgraph",
    "llm": {
        "model": "gpt-4",
        "temperature": 0.7
    },
    "tools": ["search", "calculator"],
    "memory": "conversation_buffer"
}

# CrewAI configuration (planned)
crewai_config = {
    "backend": "crewai",
    "agents": [
        {"role": "researcher", "goal": "find information"},
        {"role": "writer", "goal": "create content"}
    ],
    "process": "sequential"
}