软件架构与设计模式完全精通：从 SOLID、Clean Architecture 到 AI 系统设计

# BAD：多个职责集中在一个类中
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): ...

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)

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 "企鹅优雅地游泳"

# 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("简单打印...")

# 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())

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()

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)

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"})

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)

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:
    # 调用 LLM API
    return "响应"

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()

外部层 → 接口适配器 → 用例 → 领域实体

# 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}

# 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):
        # SQLite 保存逻辑
        pass
    def find_by_id(self, id: str) -> Article:
        # SQLite 查询逻辑
        pass

class SlackNotificationAdapter(NotificationPort):
    def notify(self, message: str):
        # 调用 Slack API
        pass

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

# Query Side（独立的读模型）
@dataclass
class OrderSummaryView:
    order_id: str
    total_price: float
    item_count: int

# Event Sourcing
@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
        )

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()

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

# 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)

# 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}

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)

@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()
            # 解析并执行 Action
            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 "已超过最大步数"

        [E2E 测试]           ← 速度慢、成本高，数量要少
      [集成测试]             ← 验证服务之间的契约
  [单元测试]                 ← 速度快、成本低，数量要多

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：提升代码质量（测试依旧保持通过）

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("数据库连接失败")
        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()