乐于分享
好东西不私藏

软件设计基础:我用一个真实项目理解了 SOLID

软件设计基础:我用一个真实项目理解了 SOLID

一、引言:AI 来了,设计还要学吗?

先看一段真实代码:

# 假设我把所有逻辑都写在一个函数里(还好我没这么写)
async
 def process_user_message(project_id, version_id, user_message):

    # 存消息

    chat = Chat(project_id=project_id, content=user_message)
    session.add(chat)

    # 提取需求(调 LLM)

    llm_result = await call_llm(...)
    new_answers = json.loads(llm_result.content)

    # 检测冲突(又调一次 LLM)

    conflicts = await call_llm(...)

    # 生成回复(再调 LLM)

    reply = await call_llm(...)

    # 发 SSE 给前端

    sse_event = SSEEvent(...)

    # 写数据库

    for
 ans in new_answers:
        session.add(Answer(...))

这代码能跑。但两周后你会发现两个问题:

  • • 没法单元测试——测提取逻辑必须先跑完整条链路,因为 LLM 调用、数据库写入、SSE 推送全部耦合在一起
  • • 改不动——想优化冲突检测逻辑,必须先看懂整条 pipeline

这不是 AI 的问题,这是没有设计的问题。

你完全可以训练 AI 去写"按职责拆分"的代码——但前提是,你自己得知道怎么拆。


二、全景图:软件设计到底学什么

阶段
关注点
我从项目里学到了
软件设计
代码组织结构
高内聚低耦合、SOLID、模块拆分
软件架构
模块间协作方式
分层架构、管道模式、事件驱动
系统设计
分布式/高并发
还没到这个阶段

现在我用一个真实在运行的项目——Agent 需求收集系统,逐条演示。


三、高内聚低耦合

什么是高内聚低耦合?

  • • 高内聚 = 一个模块只做一件事
  • • 低耦合 = 模块之间依赖越少越好

反面教材(构造的)

假设我把需求收集的全部逻辑塞进一个类:

# 反例:一个类干了所有事
class
 MegaPipeline:
    """既管存储、又管 LLM、又管 SSE 推送"""


    async
 def run(self, user_message):
        chat = Chat(...)
        session.add(chat)

        result = await call_llm(...)

        conflicts = await self._detect_conflicts(...)

        await
 self._emit_sse(conflicts)

        answer = Answer(...)
        session.add(answer)

测试它?要把数据库、LLM、SSE 全部 Mock 一遍。

真实项目是这么做的

# pipeline.py
class
 RequirementsPipeline:
    """只做"编排",不做具体业务"""


    async
 def execute(self, ...):
        turn_res = await run_turn_setup(ctx, ...)
        extract_res = await run_extraction(ctx, ...)
        conflict_res = await run_conflict_detection(ctx, ...)
        recommend_res = await run_recommendation(ctx, ...)
        followup_res = await run_followup(ctx, ...)
        reply_res = await run_reply(ctx, ...)

每个 phase 模块只做一件事(高内聚),通过 PipelineContext 传递数据(低耦合):

# context.py - 只做上下文传递
@dataclass

class
 PipelineContext:
    engine: Engine
    settings: Settings
    project_id: int
    version_id: int
    current_dim: str
    turn: int = 0

    def
 with_current_dim(self, dim: str) -> "PipelineContext":
        """创建新实例(不可变对象)"""

        return
 PipelineContext(
            engine=self.engine,
            settings=self.settings,
            current_dim=dim,
        )

好处:

  • • 测 extract 不用管 conflict,各测各的
  • • 想换提取逻辑,只改 phases/extract.py
  • • 加新阶段(比如审计),不用改其他阶段

四、分层架构——项目结构本身就是答案

这是我的项目目录,三层架构一目了然:

backend/app/
├── api/            表示层:接收 HTTP 请求
│   ├── projects.py    项目 CRUD
│   ├── answers.py     答案提交
│   ├── chat_sse.py    SSE 推送
│   └── health.py      健康检查

├── services/       业务逻辑层
│   ├── project.py       项目业务
│   ├── pipeline/        需求收集管道
│   │   ├── pipeline.py      编排器
│   │   ├── context.py       上下文
│   │   └── phases/          各阶段
│   │       ├── turn.py / extract.py
│   │       ├── conflict.py / recommend.py
│   │       └── followup.py / reply.py
│   ├── conflict_detector.py
│   └── answer_persistence.py

├── models/         数据层:数据库模型
│   ├── project.py / answer.py / chat.py

├── core/           基础设施
│   ├── config.py / db.py
│   ├── errors.py / security.py

└── schemas/        数据传输对象
    ├── answer_state.py    状态机
    └── sse_events.py      事件类型

依赖方向从外向内:api 调 services,services 调 models 和 core

  • • api/projects.py 不直接操作数据库,它调 services/project.py
  • • services 层不直接发 SSE,它返回事件让上层统一处理

五、设计原则(SOLID)

1. SRP(单一职责原则)

一个类应该只有一个被修改的理由。

PipelineContext 只做一件事——传递上下文:

@dataclass
class
 PipelineContext:
    engine: Engine
    settings: Settings
    project_id: int
    version_id: int
    # ...纯数据,没有任何业务逻辑


    def
 with_current_dim(self, dim: str) -> "PipelineContext":
        """创建新实例"""

        return
 self.__class__(
            engine=self.engine,
            current_dim=dim,
        )

再看 errors.py——只负责分类错误:

class LLMError(Exception):
    retryable: ClassVar[bool] = False

def
 classify_http_error(status_code: int, body: str) -> LLMError:
    """把 HTTP 状态码映射成具体错误类型"""

    if
 status_code == 429:
        return
 LLMRateLimitError("rate limited")
    if
 status_code in (401, 403):
        return
 LLMUnauthorizedError("unauthorized")

2. OCP(开闭原则)

对扩展开放,对修改关闭。

Pipeline 的 phase 模式就是 OCP 的体现。想加新阶段:

  1. 1. 新建 phases/audit.py(扩展)
  2. 2. 在 pipeline.py 加一行调用(不改已有 phase)
# 加审计阶段,不需要改 extract.py / conflict.py
async
 def execute(self, ...):
    turn_res = await run_turn_setup(ctx, ...)
    extract_res = await run_extraction(ctx, ...)
    conflict_res = await run_conflict_detection(ctx, ...)
    audit_res = await run_audit(ctx, conflict_res)  # 新加
    recommend_res = await run_recommendation(ctx, ...)
    followup_res = await run_followup(ctx, ...)
    reply_res = await run_reply(ctx, ...)

3. LSP(里氏替换原则)

子类应该能替换父类而不破坏程序。

错误类体系就是典型案例:

class LLMError(Exception):
    retryable: ClassVar[bool] = False

class
 LLMRateLimitError(LLMError):
    retryable = True   # 429 可重试

class
 LLMServerError(LLMError):
    retryable = True   # 5xx 可重试

class
 LLMContextLengthError(LLMError):
    retryable = False
    user_actionable = True

使用方不需要知道具体是哪种错误:

try:
    await
 extractor_svc.extract_and_persist(...)
except
 LLMError as e:
    # 限流、超时、认证失败,统一处理

    logger.warning("extractor failed: %s", e.message)
    return
 ExtractionResult(events=[], new_answers=[])

4. ISP(接口隔离原则)

接口要小而专,不要一个胖接口。

用细粒度的 dataclass 代替万能接口:

@dataclass(frozen=True)
class
 ConflictCandidate:
    """候选冲突对"""

    dim: str
    key: str
    old: str
    new: str

@dataclass(frozen=True)

class
 ConflictDecision:
    """冲突判断结果"""

    dim: str
    key: str
    old: str
    new: str
    type
: ConflictType
    confidence: float
    reason: str

每个 phase 有自己的 result,不弄一个巨大的 PipelineResult:

@dataclass
class
 TurnSetupResult:
    turn: int
    is_cold_start_probe: bool
    recent_dicts: list[dict]
    old_answers_cache: dict

@dataclass

class
 ExtractionResult:
    events: list[tuple]
    new_answers: list[Answer]

5. DIP(依赖倒置原则)

高层模块不应该依赖低层模块,二者都应该依赖抽象。

API 层依赖 Service 层,不直接操作数据库:

# api/projects.py - 高层
@router.post("")

def
 create_project(body: ProjectCreate, session: Session) -> ProjectOut:
    p = project_svc.create_project(session, body.name)  # 依赖 service
    return
 _to_out(p, include_token=True)


# services/project.py - 低层

def
 create_project(session: Session, name: str) -> Project:
    p = Project(short_hash=generate_short_hash(), name=name)
    session.add(p)
    session.commit()
    return
 p

如果将来换数据库: 改 models 的表定义,services 层最多改查询,API 层一行不动。


六、SOLID 五原则的关系

目标:OCP(开闭原则)
  Pipeline + 新 phase,不改旧 phase
          │
    ┌─────┴─────┐
    │           │
  SRP         DIP
  Context     api 调 service
  只做传递    不直接调 db
    │           │
    └─────┬─────┘
          │
    ISP + LSP 确保质量
    小接口 / 可替换

七、学完这些后我做了什么?(Skill 沉淀)

学习过程中,我把这些设计原则和常用模式抽成了一个可复用的 Skill。

Skill 的核心结构:

API Design Patterns
  RESTful URL 规范
  Repository 模式(抽象数据访问)
  Service 层模式(分离业务逻辑)

Database Patterns
  查询优化(只选需要的字段)
  N+1 预防(批量查询)
  事务模式

Error Handling Patterns
  集中错误处理
  指数退避重试

Skill 的意义: 不是死记硬背,而是遇到场景能自动想到用哪个模式。

  • • 测试很难写?检查耦合,用依赖注入
  • • 加功能改了很多文件?检查 OCP,用策略模式
  • • 代码里很多 if-else?用多态替换

八、总结

回头看,从"高内聚低耦合"到"分层架构"到"SOLID",始终围绕一个核心问题:

加新功能时,旧代码能不能不坏?

学了这些,面对 AI 写出来的代码,你能说"这段耦合太紧了,拆一下再做"——而不是"能跑就行"。

下一步:局部代码设计扎实之后,往架构方向走——去理解 CQRS、事件驱动这些模式如何在大范围内解决耦合问题。