Explain me detailly about the Architectural design patterns in Python along with the flow diagram?

Architectural Design Patterns in Python

Design patterns are reusable solutions to commonly occurring problems in software design. They are grouped into Creational, Structural, Behavioral, and Architectural categories.

---

Overview

---

1. Creational Patterns

Control how objects are created.

1a. Singleton

Ensures a class has only one instance throughout the application.

pythonCopy
class LLMClientSingleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialised = False
        return cls._instance

    def __init__(self, api_key: str = ""):
        if self._initialised:
            return
        self.api_key = api_key
        self.request_count = 0
        self._initialised = True

    def generate(self, prompt: str) -> str:
        self.request_count += 1
        return f"[LLM] Response #{self.request_count}"

# Always returns the same instance
client_a = LLMClientSingleton(api_key="sk-abc")
client_b = LLMClientSingleton()
print(client_a is client_b)          # True
print(client_b.api_key)              # sk-abc

When to use: Shared resources — API clients, database connections, config loaders.

---

1b. Factory Method

Defines an interface for creating objects, but lets subclasses decide which class to instantiate.

pythonCopy
from abc import ABC, abstractmethod

class BaseLLM(ABC):
    @abstractmethod
    def generate(self, prompt: str) -> str: ...

class OpenAILLM(BaseLLM):
    def generate(self, prompt: str) -> str:
        return f"[OpenAI] {prompt}"

class AnthropicLLM(BaseLLM):
    def generate(self, prompt: str) -> str:
        return f"[Anthropic] {prompt}"

class LocalLLM(BaseLLM):
    def generate(self, prompt: str) -> str:
        return f"[Local] {prompt}"

class LLMFactory:
    _registry = {
        "gpt-4o":  OpenAILLM,
        "claude":  AnthropicLLM,
        "llama":   LocalLLM,
    }

    @classmethod
    def create(cls, model_name: str) -> BaseLLM:
        if model_name not in cls._registry:
            raise ValueError(f"Unknown model: {model_name}")
        return cls._registry[model_name]()

    @classmethod
    def register(cls, name: str, klass: type) -> None:
        cls._registry[name] = klass

llm = LLMFactory.create("gpt-4o")
print(llm.generate("What is RAG?"))  # [OpenAI] What is RAG?

When to use: Creating objects without specifying the exact class — model selection, plugin systems.

---

1c. Builder

Constructs a complex object step by step, separating construction from representation.

pythonCopy
from dataclasses import dataclass, field

@dataclass
class RAGPipeline:
    retriever: object
    llm: object
    reranker: object | None = None
    memory: object | None = None
    top_k: int = 5

class RAGPipelineBuilder:
    def __init__(self):
        self._retriever = None
        self._llm = None
        self._reranker = None
        self._memory = None
        self._top_k = 5

    def with_retriever(self, retriever) -> "RAGPipelineBuilder":
        self._retriever = retriever
        return self

    def with_llm(self, llm) -> "RAGPipelineBuilder":
        self._llm = llm
        return self

    def with_reranker(self, reranker) -> "RAGPipelineBuilder":
        self._reranker = reranker
        return self

    def with_memory(self, memory) -> "RAGPipelineBuilder":
        self._memory = memory
        return self

    def with_top_k(self, k: int) -> "RAGPipelineBuilder":
        self._top_k = k
        return self

    def build(self) -> RAGPipeline:
        if not self._retriever or not self._llm:
            raise ValueError("Retriever and LLM are required")
        return RAGPipeline(
            retriever=self._retriever,
            llm=self._llm,
            reranker=self._reranker,
            memory=self._memory,
            top_k=self._top_k,
        )

# Fluent builder API
pipeline = (
    RAGPipelineBuilder()
    .with_retriever("ChromaDB")
    .with_llm("gpt-4o")
    .with_reranker("CohereReranker")
    .with_top_k(10)
    .build()
)

When to use: Constructing complex objects with many optional parameters (pipelines, configs).

---

2. Structural Patterns

Control how objects are composed into larger structures.

2a. Adapter

Converts an incompatible interface into one the client expects.

pythonCopy
from abc import ABC, abstractmethod

class EmbedderInterface(ABC):
    @abstractmethod
    def embed(self, text: str) -> list[float]: ...

# Third-party library with incompatible interface
class HuggingFaceSentenceTransformer:
    def encode(self, sentences: list[str]) -> list[list[float]]:
        return [[0.1, 0.2] for _ in sentences]   # mock

# Adapter wraps the incompatible class
class HuggingFaceAdapter(EmbedderInterface):
    def __init__(self, model_name: str):
        self._model = HuggingFaceSentenceTransformer()

    def embed(self, text: str) -> list[float]:
        return self._model.encode([text])[0]    # adapt the interface

# Client code works with the standard interface
def build_index(embedder: EmbedderInterface, docs: list[str]):
    return [embedder.embed(doc) for doc in docs]

adapter = HuggingFaceAdapter("all-MiniLM-L6-v2")
build_index(adapter, ["What is RAG?", "Explain LoRA"])

When to use: Integrating third-party libraries with your own interface.

---

2b. Decorator

Attaches additional responsibilities to an object dynamically without modifying its class.

pythonCopy
import time
import hashlib
from functools import wraps

class BaseLLM:
    def generate(self, prompt: str) -> str:
        return f"[LLM] Response to: {prompt[:30]}"

class LoggedLLM:
    def __init__(self, llm: BaseLLM):
        self._llm = llm

    def generate(self, prompt: str) -> str:
        print(f"[LOG] Prompt: {prompt[:40]}...")
        result = self._llm.generate(prompt)
        print(f"[LOG] Response received")
        return result

class CachedLLM:
    def __init__(self, llm, ttl_seconds: int = 300):
        self._llm = llm
        self._cache: dict = {}
        self._ttl = ttl_seconds

    def _key(self, prompt: str) -> str:
        return hashlib.md5(prompt.encode()).hexdigest()

    def generate(self, prompt: str) -> str:
        key = self._key(prompt)
        if key in self._cache:
            print("[CACHE] Hit!")
            return self._cache[key]
        result = self._llm.generate(prompt)
        self._cache[key] = result
        return result

# Stack decorators
llm = CachedLLM(LoggedLLM(BaseLLM()))
llm.generate("What is attention?")  # logs + caches
llm.generate("What is attention?")  # cache hit

When to use: Adding cross-cutting concerns — logging, caching, rate limiting, retry logic.

---

2c. Facade

Provides a simplified interface to a complex subsystem.

pythonCopy
class RAGFacade:
    def __init__(self):
        self._loader = DocumentLoader()
        self._chunker = TextChunker(chunk_size=512)
        self._embedder = OpenAIEmbedder()
        self._store = ChromaVectorStore()
        self._llm = GPT4Client()

    def ingest(self, file_path: str) -> None:
        docs = self._loader.load(file_path)
        chunks = self._chunker.chunk(docs)
        embeddings = self._embedder.embed_batch(chunks)
        self._store.add(chunks, embeddings)

    def ask(self, question: str) -> str:
        query_vec = self._embedder.embed(question)
        context = self._store.search(query_vec, k=5)
        prompt = f"Context:\n{context}\n\nQuestion: {question}"
        return self._llm.generate(prompt)

# Client uses one simple interface
rag = RAGFacade()
rag.ingest("company_docs.pdf")
answer = rag.ask("What is the refund policy?")

When to use: Simplifying complex library or subsystem APIs for end users.

---

3. Behavioral Patterns

Control how objects communicate and distribute responsibility.

3a. Observer

Defines a one-to-many dependency so that when one object changes state, all dependents are notified automatically.

pythonCopy
from abc import ABC, abstractmethod
from collections import defaultdict

class Observer(ABC):
    @abstractmethod
    def update(self, event: str, data: dict) -> None: ...

class LLMEventBus:
    def __init__(self):
        self._listeners: dict[str, list[Observer]] = defaultdict(list)

    def subscribe(self, event: str, observer: Observer) -> None:
        self._listeners[event].append(observer)

    def publish(self, event: str, data: dict) -> None:
        for observer in self._listeners[event]:
            observer.update(event, data)

class TokenLogger(Observer):
    def update(self, event: str, data: dict) -> None:
        print(f"[LOG] {event}: {data}")

class CostTracker(Observer):
    def __init__(self, cost_per_1k: float = 0.01):
        self.total_cost = 0.0
        self._rate = cost_per_1k

    def update(self, event: str, data: dict) -> None:
        tokens = data.get("tokens", 0)
        self.total_cost += (tokens / 1000) * self._rate
        print(f"[COST] Total: ${self.total_cost:.4f}")

bus = LLMEventBus()
bus.subscribe("token_used", TokenLogger())
bus.subscribe("token_used", CostTracker(cost_per_1k=0.03))

bus.publish("token_used", {"model": "gpt-4o", "tokens": 512})

When to use: Event-driven systems — token tracking, streaming callbacks, pipeline notifications.

---

3b. Strategy

Defines a family of algorithms, encapsulates each one, and makes them interchangeable at runtime.

pythonCopy
from abc import ABC, abstractmethod

class ChunkStrategy(ABC):
    @abstractmethod
    def chunk(self, text: str) -> list[str]: ...

class FixedSizeChunker(ChunkStrategy):
    def __init__(self, size: int = 512):
        self.size = size

    def chunk(self, text: str) -> list[str]:
        return [text[i:i+self.size] for i in range(0, len(text), self.size)]

class SentenceChunker(ChunkStrategy):
    def chunk(self, text: str) -> list[str]:
        return [s.strip() for s in text.split(".") if s.strip()]

class SemanticChunker(ChunkStrategy):
    def chunk(self, text: str) -> list[str]:
        # Use embeddings to detect topic boundaries
        return text.split("\n\n")

class TextProcessor:
    def __init__(self, strategy: ChunkStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: ChunkStrategy) -> None:
        self._strategy = strategy          # swap at runtime

    def process(self, text: str) -> list[str]:
        return self._strategy.chunk(text)

processor = TextProcessor(FixedSizeChunker(512))
chunks = processor.process("Long document text here...")

# Switch strategy without changing TextProcessor
processor.set_strategy(SemanticChunker())
chunks = processor.process("Long document text here...")

When to use: Swappable algorithms — chunking, retrieval, reranking, scoring strategies.

---

3c. Chain of Responsibility

Passes a request along a chain of handlers until one handles it.

pythonCopy
from abc import ABC, abstractmethod

class Handler(ABC):
    def __init__(self):
        self._next: "Handler | None" = None

    def set_next(self, handler: "Handler") -> "Handler":
        self._next = handler
        return handler

    @abstractmethod
    def handle(self, query: str) -> str | None: ...

    def pass_to_next(self, query: str) -> str | None:
        if self._next:
            return self._next.handle(query)
        return "No handler could process this query."

class GuardrailHandler(Handler):
    _blocked = ["hack", "exploit", "jailbreak"]

    def handle(self, query: str) -> str | None:
        if any(word in query.lower() for word in self._blocked):
            return "Query blocked by guardrails."
        return self.pass_to_next(query)

class CacheHandler(Handler):
    _cache = {"What is RAG?": "RAG combines retrieval with generation."}

    def handle(self, query: str) -> str | None:
        if query in self._cache:
            print("[CACHE] Hit")
            return self._cache[query]
        return self.pass_to_next(query)

class RAGHandler(Handler):
    def handle(self, query: str) -> str | None:
        docs = ["doc1", "doc2"]   # mock retrieval
        if docs:
            return f"[RAG] Answer from docs: {query}"
        return self.pass_to_next(query)

class FallbackHandler(Handler):
    def handle(self, query: str) -> str | None:
        return f"[Fallback] Sorry, I could not find an answer to: {query}"

# Build the chain
guardrail = GuardrailHandler()
cache = CacheHandler()
rag = RAGHandler()
fallback = FallbackHandler()

guardrail.set_next(cache).set_next(rag).set_next(fallback)

print(guardrail.handle("What is RAG?"))         # Cache hit
print(guardrail.handle("Explain LoRA"))          # RAG handler
print(guardrail.handle("hack the system"))       # Guardrail blocked

When to use: Request processing pipelines — guardrails, caching, routing, fallback chains.

---

4. Architectural Patterns

Define the overall structure of a system.

4a. Repository Pattern

Abstracts the data access layer so business logic doesn't depend on storage details.

pythonCopy
from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class Question:
    id: int
    text: str
    difficulty: str

class QuestionRepository(ABC):
    @abstractmethod
    def get_by_id(self, qid: int) -> Question | None: ...

    @abstractmethod
    def get_all(self) -> list[Question]: ...

    @abstractmethod
    def save(self, question: Question) -> None: ...

class JSONQuestionRepository(QuestionRepository):
    def __init__(self, path: str):
        import json
        with open(path) as f:
            data = json.load(f)
        self._store = {q["id"]: Question(**q) for q in data["questions"]}

    def get_by_id(self, qid: int) -> Question | None:
        return self._store.get(qid)

    def get_all(self) -> list[Question]:
        return list(self._store.values())

    def save(self, question: Question) -> None:
        self._store[question.id] = question

class QuestionService:
    def __init__(self, repo: QuestionRepository):
        self._repo = repo

    def get_hard_questions(self) -> list[Question]:
        return [q for q in self._repo.get_all() if q.difficulty == "Hard"]

When to use: Decoupling business logic from storage — swap JSON for SQL without changing services.

---

4b. Pipeline Pattern

Processes data through a sequence of independent, composable stages.

pythonCopy
from abc import ABC, abstractmethod
from typing import Any

class PipelineStage(ABC):
    @abstractmethod
    def process(self, data: Any) -> Any: ...

class LoadStage(PipelineStage):
    def process(self, path: str) -> str:
        with open(path) as f:
            return f.read()

class CleanStage(PipelineStage):
    def process(self, text: str) -> str:
        return " ".join(text.split())

class ChunkStage(PipelineStage):
    def __init__(self, size: int = 512):
        self.size = size

    def process(self, text: str) -> list[str]:
        return [text[i:i+self.size] for i in range(0, len(text), self.size)]

class EmbedStage(PipelineStage):
    def process(self, chunks: list[str]) -> list[list[float]]:
        return [[0.1, 0.2] for _ in chunks]  # mock embeddings

class Pipeline:
    def __init__(self, stages: list[PipelineStage]):
        self._stages = stages

    def run(self, data: Any) -> Any:
        for stage in self._stages:
            data = stage.process(data)
        return data

pipeline = Pipeline([
    LoadStage(),
    CleanStage(),
    ChunkStage(512),
    EmbedStage(),
])

embeddings = pipeline.run("docs/handbook.txt")

When to use: Data processing workflows — RAG ingestion, ETL, model inference chains.

---

Summary Table

Pro tip: LangChain is itself a masterclass in design patterns -

textCopy

BaseLLM

(Factory + Template Method),

textCopy

LLMChain

(Pipeline),

textCopy

CallbackManager

(Observer), and

textCopy

Runnable

(Strategy) are all real implementations of these patterns.

Learn more at Refactoring.Guru Python Patterns and Python Patterns Guide.