はじめに
ソフトウェアアーキテクチャはシステムの骨格です。よく設計されたアーキテクチャは変更に柔軟で、テストしやすく、チーム全体が理解できるコードを生み出します。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つだけ生成されることを保証します。
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パターン
オブジェクトに動的に新しい責務を追加します。
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:
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
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:
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）
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()
クイズ
**答え**: 下位モジュールの変更が上位モジュールにまで伝播し、システム全体の変更コストが増大するからです。
**解説**: ビジネスロジック（上位）がDB・外部API（下位）に直接依存すると、DBをMySQLからPostgreSQLに切り替えるだけでビジネスロジックのコードを修正する必要が生じます。抽象（インターフェース）に依存すれば、下位の実装だけを差し替えれば済み、上位コードはそのままです。テスト時のMock注入も容易になります。
**答え**: Observerでは SubjectとObserverが直接参照関係を持ちますが、Pub/SubではPublisherとSubscriberの間にメッセージブローカー（イベントバス）があり、完全に分離されます。
**解説**: Observerパターンでは、SubjectがObserverのリストを直接管理するため、同一プロセスに存在する必要があります。Pub/SubではKafkaやRabbitMQのようなブローカーを介してPublisherとSubscriberがお互いを知らない状態で通信します。マイクロサービスの非同期通信にはPub/Subがより適しています。
**答え**: 読み書きの負荷を独立してスケールでき、読み取りモデルを最適化できますが、データ整合性の遅延（結果整合性）とコード複雑度の増加がデメリットです。
**解説**: Command（書き込み）は強い整合性が必要で、Query（読み取り）はパフォーマンス最適化が重要です。分離することで複数の読み取り専用レプリカや非正規化された読み取りモデルを作れます。ただしイベントソーシングと組み合わせると、読み書きモデル間の同期遅延が発生し、システム全体の複雑度が大きく上がります。
**答え**: 複数のマイクロサービスにまたがるビジネストランザクションを処理する際、分散環境で2PC（Two-Phase Commit）なしにデータ整合性を保証する必要があるときです。
**解説**: 注文・決済・配送がそれぞれ別サービスの場合、1つのDBトランザクションで処理できません。Sagaは各ステップをローカルトランザクションで処理し、途中で失敗した場合は完了済みのステップを逆順に補償トランザクション（compensating transaction）でロールバックします。Choreography方式（イベント駆動）とOrchestration方式（中央調整者）で実装できます。
**答え**: 現在失敗しているテストだけを通過させるコードを書くことで、テストが実際に動作を検証しているか確認し、過剰設計（YAGNIの違反）を防ぐためです。
**解説**: 最小実装を強制することで、テストが仕様（specification）として正しく機能しているか検証されます。最初から完全な実装をしてしまうと、テストが正しい理由で通過しているかどうかわかりません。Refactorフェーズでコード品質を向上させる際、テストが安全網として機能するため、リファクタリングを安全に行えます。
まとめ
ソフトウェアアーキテクチャは一度学べば終わりではありません。SOLID原則は小さな関数の設計から適用され、Clean Architectureはチームが何年も保守するシステムの基盤となります。AIシステム設計でも同じ原則が当てはまります。RAGパイプラインをポート-アダプターで設計すればベクトルDBの切り替えが容易になり、AgentシステムをCommandパターンでツール実行を管理すれば拡張性が高まります。
良いアーキテクチャは**変更を恐れない**システムを作ります。今日学んだパターンを実際のプロジェクトに一つずつ適用してみてください。
현재 단락 (1/471)
ソフトウェアアーキテクチャはシステムの骨格です。よく設計されたアーキテクチャは変更に柔軟で、テストしやすく、チーム全体が理解できるコードを生み出します。AI時代にはLLMサービス、RAGパイプライン、...