ソフトウェアアーキテクチャ & デザインパターン完全制覇：SOLID、Clean Architecture、AIシステム設計まで

はじめに
1. SOLID原則
2. GoFデザインパターン
3. アーキテクチャパターン
4. マイクロサービスパターン
- 4.1 API Gatewayパターン
- 4.2 Sagaパターン（分散トランザクション）
5. クリーンコード原則
6. AIシステムアーキテクチャ
- 6.1 RAG（Retrieval-Augmented Generation）アーキテクチャ
- 6.2 Agentシステム設計
7. テスト戦略
クイズ
まとめ

はじめに

ソフトウェアアーキテクチャはシステムの骨格です。よく設計されたアーキテクチャは変更に柔軟で、テストしやすく、チーム全体が理解できるコードを生み出します。AI時代にはLLMサービス、RAGパイプライン、Agentシステムなど新たな設計課題が登場しました。このガイドでは古典的な設計原則から現代のAIシステムアーキテクチャまで体系的に解説します。

1. SOLID原則

SOLIDはRobert C. Martinがまとめた5つのオブジェクト指向設計原則で、保守性と拡張性の高いソフトウェアを構築する基盤となります。

1.1 単一責任原則 (SRP)

クラスは変更する理由が1つだけであるべきです。

# BAD: 複数の責任が1つのクラスに集中
class UserManager:
    def create_user(self, data): ...
    def send_welcome_email(self, user): ...
    def save_to_database(self, user): ...

# GOOD: 各責任を別クラスに分離
class UserRepository:
    def save(self, user): ...

class EmailService:
    def send_welcome(self, user): ...

class UserFactory:
    def create(self, data): ...

1.2 開放/閉鎖原則 (OCP)

ソフトウェアエンティティは拡張に対して開いており、修正に対して閉じているべきです。

from abc import ABC, abstractmethod

class Discount(ABC):
    @abstractmethod
    def apply(self, price: float) -> float: ...

class NoDiscount(Discount):
    def apply(self, price: float) -> float:
        return price

class PercentDiscount(Discount):
    def __init__(self, percent: float):
        self.percent = percent
    def apply(self, price: float) -> float:
        return price * (1 - self.percent / 100)

class VIPDiscount(Discount):
    def apply(self, price: float) -> float:
        return price * 0.7

# 新しい割引ポリシーを追加する際に既存コードの修正不要
class Order:
    def __init__(self, discount: Discount):
        self.discount = discount

    def final_price(self, base: float) -> float:
        return self.discount.apply(base)

1.3 リスコフ置換原則 (LSP)

派生型は基底型と置き換え可能でなければなりません。

class Bird:
    def fly(self) -> str:
        return "Flying"

# LSP違反：PenguinはBirdを継承するが飛べない
class Penguin(Bird):
    def fly(self):
        raise NotImplementedError("ペンギンは飛べません")

# GOOD: 能力ベースでインターフェースを分離
class FlyingBird(ABC):
    @abstractmethod
    def fly(self) -> str: ...

class SwimmingBird(ABC):
    @abstractmethod
    def swim(self) -> str: ...

class Eagle(FlyingBird):
    def fly(self) -> str:
        return "ワシが舞い上がります"

class Penguin(SwimmingBird):
    def swim(self) -> str:
        return "ペンギンが泳ぎます"

1.4 インターフェース分離原則 (ISP)

クライアントは使用しないメソッドへの依存を強制されるべきではありません。

# BAD: 巨大な単一インターフェース
class Machine(ABC):
    @abstractmethod
    def print(self): ...
    @abstractmethod
    def scan(self): ...
    @abstractmethod
    def fax(self): ...

# GOOD: 小さな単位に分割
class Printable(ABC):
    @abstractmethod
    def print(self): ...

class Scannable(ABC):
    @abstractmethod
    def scan(self): ...

class MultiFunctionPrinter(Printable, Scannable):
    def print(self): print("印刷中...")
    def scan(self): print("スキャン中...")

class SimplePrinter(Printable):
    def print(self): print("シンプル印刷...")

1.5 依存性逆転原則 (DIP)

上位モジュールは下位モジュールに依存してはなりません。両者とも抽象に依存すべきです。

# BAD: 上位が下位に直接依存
class MySQLDatabase:
    def query(self, sql: str): ...

class UserService:
    def __init__(self):
        self.db = MySQLDatabase()  # 具体クラスに依存

# GOOD: 抽象に依存（依存性注入）
class DatabasePort(ABC):
    @abstractmethod
    def find_user(self, user_id: str) -> dict: ...

class MySQLAdapter(DatabasePort):
    def find_user(self, user_id: str) -> dict:
        # MySQL実装
        return {}

class MongoAdapter(DatabasePort):
    def find_user(self, user_id: str) -> dict:
        # MongoDB実装
        return {}

class UserService:
    def __init__(self, db: DatabasePort):
        self.db = db  # 抽象に依存

# 使用例
service = UserService(db=MySQLAdapter())

2. GoFデザインパターン

2.1 Factoryパターン

オブジェクト生成ロジックをカプセル化します。

class LLMProvider(ABC):
    @abstractmethod
    def complete(self, prompt: str) -> str: ...

class OpenAIProvider(LLMProvider):
    def complete(self, prompt: str) -> str:
        return f"OpenAI応答: {prompt}"

class AnthropicProvider(LLMProvider):
    def complete(self, prompt: str) -> str:
        return f"Anthropic応答: {prompt}"

class LLMFactory:
    _registry = {
        "openai": OpenAIProvider,
        "anthropic": AnthropicProvider,
    }

    @classmethod
    def create(cls, provider: str) -> LLMProvider:
        klass = cls._registry.get(provider)
        if not klass:
            raise ValueError(f"不明なprovider: {provider}")
        return klass()

2.2 Singletonパターン

インスタンスが1つだけ生成されることを保証します。

import threading

class ConfigManager:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._config = {}
        return cls._instance

    def set(self, key: str, value):
        self._config[key] = value

    def get(self, key: str):
        return self._config.get(key)

2.3 Observerパターン

オブジェクトの状態変化を複数のサブスクライバーに自動通知します。

class EventBus:
    def __init__(self):
        self._subscribers: dict[str, list] = {}

    def subscribe(self, event: str, handler):
        self._subscribers.setdefault(event, []).append(handler)

    def publish(self, event: str, data=None):
        for handler in self._subscribers.get(event, []):
            handler(data)

# 使用例
bus = EventBus()
bus.subscribe("user.created", lambda d: print(f"ウェルカムメール送信: {d}"))
bus.subscribe("user.created", lambda d: print(f"分析イベント記録: {d}"))
bus.publish("user.created", {"id": "u1", "email": "user@example.com"})

2.4 Strategyパターン

アルゴリズムをカプセル化し、実行時に切り替え可能にします。

class SortStrategy(ABC):
    @abstractmethod
    def sort(self, data: list) -> list: ...

class QuickSort(SortStrategy):
    def sort(self, data: list) -> list:
        return sorted(data)

class MergeSort(SortStrategy):
    def sort(self, data: list) -> list:
        return sorted(data, key=lambda x: x)

class DataProcessor:
    def __init__(self, strategy: SortStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: SortStrategy):
        self._strategy = strategy

    def process(self, data: list) -> list:
        return self._strategy.sort(data)

2.5 Decoratorパターン

オブジェクトに動的に新しい責務を追加します。

import time
import functools

def retry(max_attempts: int = 3):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(2 ** attempt)
        return wrapper
    return decorator

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__} 実行時間: {time.time() - start:.2f}s")
        return result
    return wrapper

@retry(max_attempts=3)
@timed
def call_llm_api(prompt: str) -> str:
    return "応答"

2.6 Commandパターン

リクエストをオブジェクトとしてカプセル化し、取り消しやキューイングを可能にします。

class Command(ABC):
    @abstractmethod
    def execute(self): ...
    @abstractmethod
    def undo(self): ...

class CreatePostCommand(Command):
    def __init__(self, repo, post_data: dict):
        self.repo = repo
        self.post_data = post_data
        self.created_id = None

    def execute(self):
        self.created_id = self.repo.create(self.post_data)

    def undo(self):
        if self.created_id:
            self.repo.delete(self.created_id)

class CommandHistory:
    def __init__(self):
        self._history: list[Command] = []

    def execute(self, cmd: Command):
        cmd.execute()
        self._history.append(cmd)

    def undo_last(self):
        if self._history:
            self._history.pop().undo()

3. アーキテクチャパターン

3.1 Clean Architecture

依存性は常に内側（ドメイン）に向かいます。

外部レイヤー → インターフェースアダプター → ユースケース → ドメインエンティティ

# Domain Entity（最内部）
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Article:
    id: str
    title: str
    content: str
    author_id: str
    created_at: datetime = field(default_factory=datetime.utcnow)

    def publish(self):
        if not self.title or not self.content:
            raise ValueError("タイトルと内容は必須です")

# Use Case（ドメインロジックの調整）
class CreateArticleUseCase:
    def __init__(self, repo: "ArticleRepository", event_bus: EventBus):
        self.repo = repo
        self.event_bus = event_bus

    def execute(self, title: str, content: str, author_id: str) -> Article:
        import uuid
        article = Article(
            id=str(uuid.uuid4()),
            title=title,
            content=content,
            author_id=author_id,
        )
        article.publish()
        self.repo.save(article)
        self.event_bus.publish("article.created", {"id": article.id})
        return article

# Interface Adapter（外部レイヤー）
class ArticleController:
    def __init__(self, use_case: CreateArticleUseCase):
        self.use_case = use_case

    def handle_create(self, request: dict) -> dict:
        article = self.use_case.execute(
            title=request["title"],
            content=request["content"],
            author_id=request["author_id"],
        )
        return {"id": article.id, "title": article.title}

3.2 Hexagonal Architecture（ポートとアダプター）

# Port（インターフェース定義）
class ArticleRepository(ABC):
    @abstractmethod
    def save(self, article: Article): ...
    @abstractmethod
    def find_by_id(self, id: str) -> Article: ...

class NotificationPort(ABC):
    @abstractmethod
    def notify(self, message: str): ...

# Adapter（外部システムとの接続）
class SQLiteArticleRepository(ArticleRepository):
    def save(self, article: Article):
        pass  # SQLite保存ロジック
    def find_by_id(self, id: str) -> Article:
        pass  # SQLiteクエリロジック

class SlackNotificationAdapter(NotificationPort):
    def notify(self, message: str):
        pass  # Slack API呼び出し

3.3 CQRS + イベントソーシング

# Command側
@dataclass
class CreateOrderCommand:
    order_id: str
    user_id: str
    items: list[dict]

# Query側（別の読み取りモデル）
@dataclass
class OrderSummaryView:
    order_id: str
    total_price: float
    item_count: int

# イベントソーシング
@dataclass
class OrderCreatedEvent:
    order_id: str
    user_id: str
    items: list[dict]
    timestamp: datetime = field(default_factory=datetime.utcnow)

class OrderAggregate:
    def __init__(self):
        self.events: list = []
        self.state = {}

    def create(self, cmd: CreateOrderCommand):
        event = OrderCreatedEvent(
            order_id=cmd.order_id,
            user_id=cmd.user_id,
            items=cmd.items,
        )
        self._apply(event)
        self.events.append(event)

    def _apply(self, event: OrderCreatedEvent):
        self.state["id"] = event.order_id
        self.state["items"] = event.items
        self.state["total"] = sum(
            i.get("price", 0) * i.get("qty", 1) for i in event.items
        )

4. マイクロサービスパターン

4.1 API Gatewayパターン

from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI(title="API Gateway")

SERVICE_MAP = {
    "users": "http://user-service:8001",
    "orders": "http://order-service:8002",
    "products": "http://product-service:8003",
}

@app.get("/api/users/{user_id}")
async def proxy_user(user_id: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{SERVICE_MAP['users']}/users/{user_id}")
        if resp.status_code == 404:
            raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
        return resp.json()

4.2 Sagaパターン（分散トランザクション）

マイクロサービス間のデータ整合性を、一連のローカルトランザクションと補償トランザクションで保証します。

class OrderSaga:
    def __init__(self, order_service, payment_service, inventory_service):
        self.order_svc = order_service
        self.payment_svc = payment_service
        self.inventory_svc = inventory_service

    async def execute(self, order_data: dict):
        order_id = None
        payment_id = None
        try:
            # Step 1: 注文作成
            order_id = await self.order_svc.create(order_data)
            # Step 2: 決済処理
            payment_id = await self.payment_svc.charge(order_data["amount"])
            # Step 3: 在庫確保
            await self.inventory_svc.reserve(order_data["items"])
            return {"status": "success", "order_id": order_id}
        except Exception as e:
            # 補償トランザクション（逆順ロールバック）
            if payment_id:
                await self.payment_svc.refund(payment_id)
            if order_id:
                await self.order_svc.cancel(order_id)
            raise

5. クリーンコード原則

5.1 意味のある命名

# BAD
def calc(d, r):
    return d * (1 - r / 100)

# GOOD
def calculate_discounted_price(original_price: float, discount_rate_percent: float) -> float:
    return original_price * (1 - discount_rate_percent / 100)

5.2 関数設計 — 小さく、1つだけ

# BAD: 責任が多すぎる
def process_user_registration(email, password, name, send_email=True):
    if "@" not in email:
        raise ValueError("メール形式エラー")
    hashed = hash(password)
    user = {"email": email, "password": hashed, "name": name}
    db.save(user)
    if send_email:
        mailer.send(email, "ようこそ!")
    return user

# GOOD: 分離された関数
def validate_email(email: str) -> None:
    if "@" not in email:
        raise ValueError("メール形式エラー")

def hash_password(raw: str) -> str:
    import hashlib
    return hashlib.sha256(raw.encode()).hexdigest()

def register_user(email: str, password: str, name: str) -> dict:
    validate_email(email)
    return {"email": email, "password": hash_password(password), "name": name}

5.3 コードの臭いとリファクタリング

代表的なコードの臭い（code smell）と解決策です。

Long Method: 小さな単位に分割（Extract Method）
Large Class: 責任に応じてクラスを分離（Extract Class）
Feature Envy: メソッドをデータの近くに移動（Move Method）
Magic Numbers: 定数として抽出（Replace Magic Number with Symbolic Constant）
Duplicate Code: 共通関数として抽出（Extract Function）

6. AIシステムアーキテクチャ

6.1 RAG（Retrieval-Augmented Generation）アーキテクチャ

from dataclasses import dataclass

@dataclass
class RAGConfig:
    embedding_model: str = "text-embedding-3-small"
    llm_model: str = "gpt-4o"
    top_k: int = 5
    chunk_size: int = 512

class RAGPipeline:
    def __init__(self, config: RAGConfig, vector_store, llm_client):
        self.config = config
        self.vector_store = vector_store
        self.llm = llm_client

    def retrieve(self, query: str) -> list[str]:
        # 1. クエリの埋め込み
        query_vector = self.llm.embed(query)
        # 2. 類似ドキュメントの検索
        docs = self.vector_store.search(query_vector, top_k=self.config.top_k)
        return [d["content"] for d in docs]

    def generate(self, query: str, context: list[str]) -> str:
        context_text = "\n\n".join(context)
        prompt = f"""以下のコンテキストを参考に質問に答えてください。

コンテキスト:
{context_text}

質問: {query}
回答:"""
        return self.llm.complete(prompt)

    def query(self, user_question: str) -> str:
        context = self.retrieve(user_question)
        return self.generate(user_question, context)

6.2 Agentシステム設計

@dataclass
class Tool:
    name: str
    description: str
    func: callable

class ReActAgent:
    """Reasoning + Acting パターンのAI Agent"""

    def __init__(self, llm, tools: list[Tool]):
        self.llm = llm
        self.tools = {t.name: t for t in tools}

    def _build_system_prompt(self) -> str:
        tool_desc = "\n".join(
            f"- {t.name}: {t.description}" for t in self.tools.values()
        )
        return f"""あなたはツールを使用して問題を解決するAI Agentです。
使用可能なツール:
{tool_desc}

形式:
Thought: [現在の状況分析]
Action: [使用するツール名]
Action Input: [ツールへの入力]
Observation: [ツール実行結果]
... (繰り返し)
Final Answer: [最終回答]"""

    def run(self, task: str, max_steps: int = 10) -> str:
        messages = [{"role": "user", "content": task}]
        for _ in range(max_steps):
            response = self.llm.chat(messages)
            if "Final Answer:" in response:
                return response.split("Final Answer:")[-1].strip()
            if "Action:" in response:
                action_line = [l for l in response.split("\n") if l.startswith("Action:")]
                if action_line:
                    tool_name = action_line[0].replace("Action:", "").strip()
                    tool = self.tools.get(tool_name)
                    if tool:
                        observation = tool.func(response)
                        messages.append({"role": "assistant", "content": response})
                        messages.append({"role": "user", "content": f"Observation: {observation}"})
        return "最大ステップ数超過"

7. テスト戦略

7.1 テストピラミッド

       [E2Eテスト]           <- 遅くコスト高、少数に留める
     [統合テスト]            <- サービス間のコントラクト検証
  [単体テスト]               <- 高速・低コスト、多数維持

7.2 TDD例（Red-Green-Refactor）

import pytest

# 1. RED: 失敗するテストを先に書く
def test_calculate_discounted_price_basic():
    assert calculate_discounted_price(100.0, 20.0) == 80.0

def test_calculate_discounted_price_zero_discount():
    assert calculate_discounted_price(100.0, 0.0) == 100.0

def test_calculate_discounted_price_full_discount():
    assert calculate_discounted_price(100.0, 100.0) == 0.0

def test_calculate_discounted_price_invalid_rate():
    with pytest.raises(ValueError):
        calculate_discounted_price(100.0, -10.0)

# 2. GREEN: テストを通過する最小実装
def calculate_discounted_price(price: float, discount_rate: float) -> float:
    if discount_rate < 0 or discount_rate > 100:
        raise ValueError("割引率は0〜100の間でなければなりません")
    return price * (1 - discount_rate / 100)

# 3. REFACTOR: テストを維持しながら品質向上

7.3 モック戦略

from unittest.mock import MagicMock, patch

class TestUserService:
    def test_create_user_sends_email(self):
        mock_repo = MagicMock()
        mock_email = MagicMock()
        mock_repo.save.return_value = {"id": "u1"}

        service = UserService(repo=mock_repo, email_svc=mock_email)
        service.register("test@example.com", "山田太郎")

        mock_repo.save.assert_called_once()
        mock_email.send_welcome.assert_called_once_with("test@example.com")

    def test_create_user_handles_db_error(self):
        mock_repo = MagicMock()
        mock_repo.save.side_effect = Exception("DB接続失敗")
        mock_email = MagicMock()

        service = UserService(repo=mock_repo, email_svc=mock_email)
        with pytest.raises(Exception):
            service.register("test@example.com", "山田太郎")

        mock_email.send_welcome.assert_not_called()

クイズ

Q1. 依存性逆転原則において、上位モジュールが下位モジュールに直接依存してはいけない理由は何ですか？

答え: 下位モジュールの変更が上位モジュールにまで伝播し、システム全体の変更コストが増大するからです。

解説: ビジネスロジック（上位）がDB・外部API（下位）に直接依存すると、DBをMySQLからPostgreSQLに切り替えるだけでビジネスロジックのコードを修正する必要が生じます。抽象（インターフェース）に依存すれば、下位の実装だけを差し替えれば済み、上位コードはそのままです。テスト時のMock注入も容易になります。

Q2. ObserverパターンとPub/Subパターンの違いは何ですか？

答え: Observerでは SubjectとObserverが直接参照関係を持ちますが、Pub/SubではPublisherとSubscriberの間にメッセージブローカー（イベントバス）があり、完全に分離されます。

解説: Observerパターンでは、SubjectがObserverのリストを直接管理するため、同一プロセスに存在する必要があります。Pub/SubではKafkaやRabbitMQのようなブローカーを介してPublisherとSubscriberがお互いを知らない状態で通信します。マイクロサービスの非同期通信にはPub/Subがより適しています。

Q3. CQRSでCommandとQueryを分離するメリットと複雑度のトレードオフは？

答え: 読み書きの負荷を独立してスケールでき、読み取りモデルを最適化できますが、データ整合性の遅延（結果整合性）とコード複雑度の増加がデメリットです。

解説: Command（書き込み）は強い整合性が必要で、Query（読み取り）はパフォーマンス最適化が重要です。分離することで複数の読み取り専用レプリカや非正規化された読み取りモデルを作れます。ただしイベントソーシングと組み合わせると、読み書きモデル間の同期遅延が発生し、システム全体の複雑度が大きく上がります。

Q4. マイクロサービスでSagaパターンが必要な状況はいつですか？

答え: 複数のマイクロサービスにまたがるビジネストランザクションを処理する際、分散環境で2PC（Two-Phase Commit）なしにデータ整合性を保証する必要があるときです。

解説: 注文・決済・配送がそれぞれ別サービスの場合、1つのDBトランザクションで処理できません。Sagaは各ステップをローカルトランザクションで処理し、途中で失敗した場合は完了済みのステップを逆順に補償トランザクション（compensating transaction）でロールバックします。Choreography方式（イベント駆動）とOrchestration方式（中央調整者）で実装できます。

Q5. TDDのRed-Green-RefactorサイクルのGreenフェーズで最小限のコードだけを書く理由は？

答え: 現在失敗しているテストだけを通過させるコードを書くことで、テストが実際に動作を検証しているか確認し、過剰設計（YAGNIの違反）を防ぐためです。

解説: 最小実装を強制することで、テストが仕様（specification）として正しく機能しているか検証されます。最初から完全な実装をしてしまうと、テストが正しい理由で通過しているかどうかわかりません。Refactorフェーズでコード品質を向上させる際、テストが安全網として機能するため、リファクタリングを安全に行えます。

まとめ

ソフトウェアアーキテクチャは一度学べば終わりではありません。SOLID原則は小さな関数の設計から適用され、Clean Architectureはチームが何年も保守するシステムの基盤となります。AIシステム設計でも同じ原則が当てはまります。RAGパイプラインをポート-アダプターで設計すればベクトルDBの切り替えが容易になり、AgentシステムをCommandパターンでツール実行を管理すれば拡張性が高まります。

良いアーキテクチャは変更を恐れないシステムを作ります。今日学んだパターンを実際のプロジェクトに一つずつ適用してみてください。