一、引言: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 去写"按职责拆分"的代码——但前提是,你自己得知道怎么拆。
二、全景图:软件设计到底学什么
| 软件设计 | ||
| 软件架构 | ||
| 系统设计 |
现在我用一个真实在运行的项目——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. 新建 phases/audit.py(扩展) 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、事件驱动这些模式如何在大范围内解决耦合问题。
夜雨聆风