0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.
引言
软件架构是系统的骨架。设计良好的架构对变更保持灵活，易于测试，并且能产出整个团队都能理解的代码。到了 AI 时代，LLM 服务、RAG 流水线、Agent 系统等新的设计课题相继出现。本指南将从经典的设计原则一路梳理到现代 AI 系统架构。
1. SOLID 原则
SOLID 是 Robert C. Martin 整理的 5 项面向对象设计原则，是构建可维护、可扩展软件的基础。
1.1 单一职责原则（SRP）
类应该只有一个职责，变更的原因也应该只有一个。
# 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): ...
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 模式
保证一个类只创建一个实例。
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:
# 调用 LLM API
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（Ports & Adapters）
# 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
3.3 CQRS + 事件溯源
# 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
)
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 模式（分布式事务）
为了保证微服务之间的数据一致性，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 函数设计 — 短小且只做一件事
# 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（检索增强生成）架构
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()
# 解析并执行 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 "已超过最大步数"
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 Mock 策略
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()
小测验
Q1. 在依赖倒置原则中，为什么高层模块不应该直接依赖低层模块？
答案：因为低层模块的变更会向上传导到高层模块，导致整个系统的变更成本增加。
解释：如果高层模块（业务逻辑）直接依赖低层模块（数据库、外部 API），仅仅把数据库从 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（两阶段提交）来保证数据一致性时，就需要 Saga 模式。
解释：当订单、支付、配送分别是不同的服务时，无法用单一的数据库事务来处理。Saga 把每一步当作本地事务来处理，一旦中途失败，就把已完成的步骤按逆序通过补偿事务（compensating transaction）回滚。它可以用 Choreography 方式（事件驱动）或 Orchestration 方式（中央协调者）来实现。
Q5. 在 TDD 的 Red-Green-Refactor 循环中，为什么 Green 阶段只写最少量的代码？
答案：这是为了确认测试真正验证了行为，并防止过度设计（违反 YAGNI）——只写出刚好能让当前失败测试通过的代码。
解释：强制只做最小实现，可以验证测试是否真正起到了规格说明（specification）的作用。如果一开始就完整实现，就无法知道测试通过是不是出于正确的原因。到了 Refactor 阶段，测试作为安全网存在，重构才能安全地进行。
结语
软件架构不是学一次就结束的知识。SOLID 原则从设计一个小小的函数时就开始适用，而 Clean Architecture 则是团队要维护数年的系统的根基。这些原则在 AI 系统设计中同样成立——用端口-适配器来设计 RAG 流水线，替换向量数据库会变得容易；用 Command 模式管理 Agent 系统的工具执行，扩展性也会提高。
好的架构会让变更不再令人畏惧。把今天学到的这些模式，一个一个地应用到实际项目中去吧。
현재 단락 (1/471)
软件架构是系统的骨架。设计良好的架构对变更保持灵活，易于测试，并且能产出整个团队都能理解的代码。到了 AI 时代，LLM 服务、RAG 流水线、Agent 系统等新的设计课题相继出现。本指南将从经典的...
작성 글자: 0자원문 글자: 14,810자작성 단락: 0/471