cd ..
2026-04-2925 min138 views

怎么规范自己的代码,设计模式详解--以agent构建举例

#Software Engineering#Design Patterns#Python#Object-Oriented Programming#Code Refactoring
AI Summary
每分钟最多 5 次
  • 工厂模式:集中管理对象的创建逻辑,避免在业务代码中频繁使用 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 执行任务,内部流程可能是:

  1. 接收消息
  2. 解析意图
  3. 调用工具
  4. 调用模型
  5. 整理结果
  6. 返回消息

外部调用者不想知道这些细节,只想:

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

解决什么问题?

多个流程大体一样,但某些步骤不同。

适合场景

比如处理不同类型文件:

  • PDF
  • Word
  • Excel
  • Markdown

它们大流程可能一样:

  1. 读取文件
  2. 提取内容
  3. 清洗文本
  4. 生成摘要
  5. 入库

但“提取内容”这一步不同。

示例

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. 一个请求要经过多个检查?

责任链模式

登录 → 权限 → 限流 → 日志 → 业务处理
这种流水线 → 责任链模式

七、最重要的提醒

设计模式不是为了“显得高级”,而是为了让代码:

  1. 更容易扩展
  2. 更容易替换
  3. 更容易测试
  4. 少写重复代码
  5. 避免巨大 if-else
  6. 降低模块之间的耦合

/** Comments(0)*/

Loading comments...
怎么规范自己的代码,设计模式详解--以agent构建举例 | UbanillxのDevLog