怎么规范自己的代码,设计模式详解--以agent构建举例
- 工厂模式:集中管理对象的创建逻辑,避免在业务代码中频繁使用
if-else语句实例化不同类型的对象。适用于需要根据条件创建不同对象的场景,如多平台 IM 客户端。 - 单例模式:确保某个类只有一个实例,并提供一个全局访问点。适用于全局只需要一个实例的对象,如数据库连接池、配置管理器等。
- 建造者模式:用于构建复杂对象,特别是当对象有很多参数且组合方式复杂时。通过逐步构建对象,使代码更清晰易读。
- 适配器模式:将不兼容的接口转换为统一的接口,使得原本由于接口不匹配而无法一起工作的类可以协同工作。适用于需要接入多个第三方 API 的场景。
- 装饰器模式:动态地给对象添加额外的功能,而不改变其原有的结构。适用于需要在不修改原有代码的情况下增强功能的场景,如日志记录、缓存等。
这些设计模式有助于提高代码的可扩展性、可维护性和可测试性,减少重复代码和巨大的 if-else 结构,降低模块间的耦合度。
一、创建型模式:解决“对象怎么创建”
1. 工厂模式 Factory
解决什么问题?
当你发现代码里到处都是:
if type == "wechat":
obj = WeChatClient()
elif type == "dingtalk":
obj = DingTalkClient()
elif type == "lark":
obj = LarkClient()
就可以考虑工厂模式。
适合场景
比如你做一个 Agent 接入多个 IM 平台:
- 微信
- 飞书
- 钉钉
- Slack
- Telegram
不同平台都有自己的发送消息、接收消息、鉴权逻辑,但你的主流程不想关心具体是哪一个平台。
示例
class IMClient:
def send(self, message):
raise NotImplementedError
class WeChatClient(IMClient):
def send(self, message):
print("发送到微信:", message)
class DingTalkClient(IMClient):
def send(self, message):
print("发送到钉钉:", message)
class IMClientFactory:
@staticmethod
def create(platform):
if platform == "wechat":
return WeChatClient()
elif platform == "dingtalk":
return DingTalkClient()
else:
raise ValueError("不支持的平台")
client = IMClientFactory.create("wechat")
client.send("你好")
一句话理解
工厂模式就是把创建对象的 if-else 集中管理,不让业务代码到处 new 对象。
2. 单例模式 Singleton
解决什么问题?
某些对象全局只需要一个,比如:
- 数据库连接池
- Redis 连接
- 配置管理器
- 日志对象
- 模型加载实例
大模型服务里尤其常见,比如模型权重很大,不可能每次请求都重新加载一次。
适合场景
你做一个网页 Demo,后端加载一个本地大模型:
model = load_model()
如果每次用户发消息都重新加载,速度会非常慢。此时就可以用单例。
示例
class ModelManager:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.model = cls._load_model()
return cls._instance
@staticmethod
def _load_model():
print("模型加载中...")
return "大模型实例"
m1 = ModelManager()
m2 = ModelManager()
print(m1 is m2) # True
一句话理解
单例模式就是保证某个重要对象全局只有一份。
3. 建造者模式 Builder
解决什么问题?
当一个对象创建时参数特别多,而且组合复杂,就可以用建造者模式。
比如创建一个 AI Agent:
Agent(
model="gpt",
memory=True,
tools=[search, calculator],
system_prompt="xxx",
temperature=0.7,
max_tokens=4096,
stream=True
)
参数越来越多,代码就会很乱。
适合场景
- 构建复杂配置
- 构建 Agent
- 构建请求对象
- 构建 Prompt
- 构建工作流 Pipeline
示例
class Agent:
def __init__(self):
self.model = None
self.tools = []
self.memory = False
self.system_prompt = ""
class AgentBuilder:
def __init__(self):
self.agent = Agent()
def use_model(self, model):
self.agent.model = model
return self
def add_tool(self, tool):
self.agent.tools.append(tool)
return self
def enable_memory(self):
self.agent.memory = True
return self
def set_prompt(self, prompt):
self.agent.system_prompt = prompt
return self
def build(self):
return self.agent
agent = (
AgentBuilder()
.use_model("qwen")
.add_tool("search")
.enable_memory()
.set_prompt("你是一个客服助手")
.build()
)
一句话理解
建造者模式适合创建“配置很多、步骤很多”的复杂对象。
二、结构型模式:解决“类和类怎么组合”
4. 适配器模式 Adapter
解决什么问题?
不同系统接口不统一,但你想让它们看起来一样。
比如不同 IM 平台的发送消息接口可能是:
wechat.send_text(user_id, content)
dingtalk.push_msg(open_id, text)
lark.send_message(receiver, msg)
你不希望业务层每次都判断平台,所以可以用适配器统一接口。
适合场景
- 接入第三方 API
- 统一多个平台接口
- 老系统改造
- 不同模型厂商统一调用方式
示例
class WeChatAPI:
def send_text(self, user_id, content):
print(f"微信发送给 {user_id}: {content}")
class WeChatAdapter:
def __init__(self, api):
self.api = api
def send(self, user, message):
self.api.send_text(user, message)
class DingTalkAPI:
def push_msg(self, open_id, text):
print(f"钉钉发送给 {open_id}: {text}")
class DingTalkAdapter:
def __init__(self, api):
self.api = api
def send(self, user, message):
self.api.push_msg(user, message)
clients = [
WeChatAdapter(WeChatAPI()),
DingTalkAdapter(DingTalkAPI())
]
for client in clients:
client.send("user_001", "你好")
一句话理解
适配器模式就是把乱七八糟的外部接口包装成统一接口。
5. 装饰器模式 Decorator
解决什么问题?
你想在不修改原函数的情况下,给它额外加功能。
比如:
- 日志
- 权限校验
- 耗时统计
- 缓存
- 重试
- 限流
- 鉴权
Python 里的 @xxx 就是典型装饰器。
适合场景
比如你有一个调用大模型的函数:
call_llm(prompt)
你想给它加:
- 打印日志
- 统计耗时
- 失败重试
- 记录 token 消耗
不应该直接把这些逻辑全塞进 call_llm。
示例
import time
def time_cost(func):
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print("耗时:", end - start)
return result
return wrapper
@time_cost
def call_llm(prompt):
print("调用大模型:", prompt)
return "模型回复"
call_llm("你好")
一句话理解
装饰器模式就是在不改原代码的情况下,动态增强功能。
6. 代理模式 Proxy
解决什么问题?
你不直接访问真实对象,而是通过一个“代理层”访问。
代理层可以做:
- 权限控制
- 缓存
- 日志
- 限流
- 远程调用
- 懒加载
适合场景
例如你有一个真正的大模型服务:
RealLLMService
但用户请求不能直接打到它。中间需要一层代理:
LLMProxy
这个代理负责:
- 检查用户权限
- 判断是否命中缓存
- 记录日志
- 再决定是否调用真实模型
示例
class RealLLMService:
def chat(self, prompt):
print("真正调用大模型")
return "模型回复"
class LLMProxy:
def __init__(self):
self.real_service = RealLLMService()
self.cache = {}
def chat(self, prompt):
if prompt in self.cache:
print("命中缓存")
return self.cache[prompt]
print("记录日志、检查权限")
result = self.real_service.chat(prompt)
self.cache[prompt] = result
return result
proxy = LLMProxy()
print(proxy.chat("你好"))
print(proxy.chat("你好"))
一句话理解
代理模式就是在真实对象前面加一层控制入口。
7. 门面模式 Facade
解决什么问题?
系统内部很复杂,但你想对外提供一个简单入口。
比如一个 Agent 执行任务,内部流程可能是:
- 接收消息
- 解析意图
- 调用工具
- 调用模型
- 整理结果
- 返回消息
外部调用者不想知道这些细节,只想:
agent.run(message)
适合场景
- SDK 封装
- API 网关
- Agent 总入口
- 复杂系统对外暴露简单接口
示例
class MessageParser:
def parse(self, message):
return {"intent": "chat", "content": message}
class ToolRouter:
def route(self, intent):
return "选择工具"
class LLM:
def generate(self, content):
return "模型回答:" + content
class AgentFacade:
def __init__(self):
self.parser = MessageParser()
self.router = ToolRouter()
self.llm = LLM()
def run(self, message):
parsed = self.parser.parse(message)
self.router.route(parsed["intent"])
return self.llm.generate(parsed["content"])
agent = AgentFacade()
print(agent.run("帮我写一封邮件"))
一句话理解
门面模式就是把复杂子系统包装成一个简单入口。
三、行为型模式:解决“对象之间怎么协作”
8. 策略模式 Strategy
解决什么问题?
当你有多种算法、多种规则、多种处理方式,而且它们可以互相替换时,用策略模式。
适合场景
比如不同用户消息有不同处理策略:
- 普通聊天
- 搜索问答
- 文件总结
- 代码解释
- 图片分析
或者同一个任务有不同模型策略:
- 便宜模型
- 高质量模型
- 快速模型
- 本地模型
示例
class ChatStrategy:
def handle(self, message):
return "普通聊天:" + message
class SearchStrategy:
def handle(self, message):
return "联网搜索:" + message
class CodeStrategy:
def handle(self, message):
return "代码分析:" + message
class Agent:
def __init__(self, strategy):
self.strategy = strategy
def run(self, message):
return self.strategy.handle(message)
agent = Agent(SearchStrategy())
print(agent.run("今天有什么新闻?"))
一句话理解
策略模式就是把 if-else 里的不同算法拆成独立策略,运行时自由切换。
9. 观察者模式 Observer
解决什么问题?
一个对象状态变化后,需要通知多个对象。
适合场景
非常常见,比如:
- 用户下单后,通知库存、积分、短信、邮件系统
- 文件上传完成后,通知解析、向量化、摘要系统
- Agent 执行状态变化后,通知前端更新进度
- IM 收到消息后,触发多个监听器
示例
class EventBus:
def __init__(self):
self.listeners = {}
def subscribe(self, event_name, callback):
self.listeners.setdefault(event_name, []).append(callback)
def publish(self, event_name, data):
for callback in self.listeners.get(event_name, []):
callback(data)
def send_email(data):
print("发送邮件:", data)
def write_log(data):
print("记录日志:", data)
bus = EventBus()
bus.subscribe("order_created", send_email)
bus.subscribe("order_created", write_log)
bus.publish("order_created", {"order_id": 123})
一句话理解
观察者模式就是一个事件发生后,自动通知所有关心它的人。
10. 模板方法模式 Template Method
解决什么问题?
多个流程大体一样,但某些步骤不同。
适合场景
比如处理不同类型文件:
- Word
- Excel
- Markdown
它们大流程可能一样:
- 读取文件
- 提取内容
- 清洗文本
- 生成摘要
- 入库
但“提取内容”这一步不同。
示例
class DocumentProcessor:
def process(self, file):
content = self.extract(file)
content = self.clean(content)
summary = self.summarize(content)
return summary
def extract(self, file):
raise NotImplementedError
def clean(self, content):
return content.strip()
def summarize(self, content):
return "摘要:" + content[:20]
class PDFProcessor(DocumentProcessor):
def extract(self, file):
return "这是从 PDF 中提取的内容"
class WordProcessor(DocumentProcessor):
def extract(self, file):
return "这是从 Word 中提取的内容"
processor = PDFProcessor()
print(processor.process("demo.pdf"))
一句话理解
模板方法模式就是固定主流程,把可变步骤留给子类实现。
11. 责任链模式 Chain of Responsibility
解决什么问题?
一个请求需要经过多个处理器,谁能处理谁处理,或者按顺序逐个处理。
适合场景
非常适合做中间件:
- 登录校验
- 权限校验
- 参数校验
- 风控检查
- 敏感词检查
- 限流
- 日志记录
在 Web 框架、网关、Agent 流程里都很常见。
示例
class Handler:
def __init__(self, next_handler=None):
self.next = next_handler
def handle(self, request):
if self.next:
return self.next.handle(request)
return "处理完成"
class AuthHandler(Handler):
def handle(self, request):
if not request.get("user"):
return "未登录"
print("登录校验通过")
return super().handle(request)
class PermissionHandler(Handler):
def handle(self, request):
if request.get("role") != "admin":
return "无权限"
print("权限校验通过")
return super().handle(request)
chain = AuthHandler(PermissionHandler())
print(chain.handle({"user": "张三", "role": "admin"}))
一句话理解
责任链模式就是让请求像流水线一样,经过多个处理节点。
12. 状态模式 State
解决什么问题?
当一个对象有多个状态,不同状态下行为不同,如果你一直用 if-else 判断状态,代码会越来越乱。
适合场景
比如订单系统:
- 待支付
- 已支付
- 已发货
- 已完成
- 已取消
不同状态下可以做的动作不同。
再比如 Agent 任务状态:
- 等待中
- 执行中
- 需要用户确认
- 已完成
- 失败
简单例子
class PendingState:
def pay(self, order):
print("支付成功")
order.state = PaidState()
class PaidState:
def pay(self, order):
print("已经支付过了")
def ship(self, order):
print("发货成功")
class Order:
def __init__(self):
self.state = PendingState()
def pay(self):
self.state.pay(self)
def ship(self):
if hasattr(self.state, "ship"):
self.state.ship(self)
else:
print("当前状态不能发货")
order = Order()
order.pay()
order.ship()
一句话理解
状态模式就是把不同状态下的行为拆开,避免一大堆状态 if-else。
四、真实开发中最常用的组合
很多时候不是单独用一个模式,而是组合使用。
场景 1:做一个多平台 IM Agent
你可能会用到:
| 需求 | 适合模式 |
|---|---|
| 支持微信、飞书、钉钉 | 工厂模式 |
| 统一不同平台接口 | 适配器模式 |
| 消息进来触发多个处理逻辑 | 观察者模式 |
| 根据消息类型选择处理方式 | 策略模式 |
统一对外提供 agent.run() |
门面模式 |
| 鉴权、日志、限流 | 责任链模式 / 装饰器模式 |
一个简化结构:
IM消息进入
↓
Adapter 统一消息格式
↓
责任链:鉴权 → 限流 → 日志 → 敏感词检查
↓
策略模式:聊天 / 搜索 / 工具调用 / 文件解析
↓
AgentFacade.run()
↓
返回给不同平台
场景 2:做一个文档解析系统
你可能会用到:
| 需求 | 适合模式 |
|---|---|
| PDF、Word、Excel 不同解析器 | 工厂模式 |
| 每种文档处理流程大体一致 | 模板方法模式 |
| 上传完成后自动触发解析、摘要、向量化 | 观察者模式 |
| 第三方解析库接口不统一 | 适配器模式 |
对外只暴露一个 parse(file) |
门面模式 |
场景 3:做一个大模型网关
你可能会用到:
| 需求 | 适合模式 |
|---|---|
| 接入 OpenAI、Qwen、Claude、本地模型 | 适配器模式 |
| 根据价格/速度/质量选择模型 | 策略模式 |
| 模型客户端统一创建 | 工厂模式 |
| 缓存、日志、鉴权、重试 | 代理模式 / 装饰器模式 |
对外统一 /chat 接口 |
门面模式 |
| 请求处理流水线 | 责任链模式 |
五、初学者最应该先掌握的 6 个
如果你不想一次记太多,先掌握这 6 个就很够用了:
| 优先级 | 模式 | 你要记住的本质 |
|---|---|---|
| 1 | 工厂模式 | 统一创建对象 |
| 2 | 策略模式 | 替代复杂 if-else |
| 3 | 适配器模式 | 统一第三方接口 |
| 4 | 装饰器模式 | 给函数/对象加功能 |
| 5 | 观察者模式 | 事件通知机制 |
| 6 | 责任链模式 | 请求流水线处理 |
六、怎么判断该用哪个模式?
可以用这个简单口诀:
1. 创建对象很乱?
用 工厂模式。
到处 new,不好维护 → 工厂模式
2. 很多 if-else 判断类型?
用 策略模式 或 状态模式。
判断的是算法/处理方式 → 策略模式
判断的是对象状态 → 状态模式
3. 第三方接口不统一?
用 适配器模式。
外部接口乱,我要内部统一 → 适配器模式
4. 想额外加日志、缓存、重试?
用 装饰器模式 或 代理模式。
增强单个函数 → 装饰器
控制访问真实对象 → 代理
5. 一个事件要通知多个模块?
用 观察者模式。
一处发生,多处响应 → 观察者模式
6. 一个请求要经过多个检查?
用 责任链模式。
登录 → 权限 → 限流 → 日志 → 业务处理
这种流水线 → 责任链模式
七、最重要的提醒
设计模式不是为了“显得高级”,而是为了让代码:
- 更容易扩展
- 更容易替换
- 更容易测试
- 少写重复代码
- 避免巨大 if-else
- 降低模块之间的耦合