Qwen2.5-32B-Instruct代码重构实战：设计模式应用指南

斜阳君

407人浏览 · 2026-02-09 00:58:03

斜阳君 · 2026-02-09 00:58:03 发布

Qwen2.5-32B-Instruct代码重构实战：设计模式应用指南

最近在维护一个老项目，代码库已经积累了五年多，各种功能模块交织在一起，改一个地方往往要动好几个文件。每次看到那些动辄几百行的函数和到处重复的业务逻辑，我就头疼得不行。

正好团队最近在尝试用大模型辅助开发，我决定拿Qwen2.5-32B-Instruct这个专门针对代码优化的模型来试试水。听说它在代码生成、修复和重构方面表现不错，我想看看能不能用它来帮我理清这个项目的混乱局面。

这篇文章就是我这段时间的实战记录，我会通过几个真实的案例，分享如何用Qwen2.5-32B-Instruct来识别代码问题、选择合适的设计模式，以及给出具体的重构建议。如果你也在为代码维护发愁，或者想了解AI辅助编程的实际效果，相信这些经验会对你有帮助。

1. 从混乱到清晰：识别代码中的“坏味道”

在开始重构之前，首先要能识别出代码中的问题。Qwen2.5-32B-Instruct在这方面给了我不少惊喜。

1.1 长函数与重复代码

我遇到的第一个案例是一个用户管理模块。原来的代码里有一个processUserRequest函数，足足有300多行，包含了用户验证、数据处理、日志记录、邮件通知等所有功能。

我把这段代码丢给Qwen2.5-32B-Instruct，让它分析问题：

# 原始代码片段
def processUserRequest(user_data, request_type):
    # 验证用户输入（约50行）
    if not validate_email(user_data['email']):
        raise ValueError("Invalid email")
    if not validate_phone(user_data['phone']):
        raise ValueError("Invalid phone")
    # ... 更多验证逻辑
    
    # 处理业务逻辑（约100行）
    if request_type == 'register':
        # 注册逻辑
        user = create_user(user_data)
        send_welcome_email(user)
        log_activity('user_registered', user.id)
    elif request_type == 'update':
        # 更新逻辑
        user = update_user(user_data)
        send_update_notification(user)
        log_activity('user_updated', user.id)
    # ... 更多分支
    
    # 记录日志（约50行）
    write_to_database_log(user_data)
    write_to_file_log(user_data)
    # ... 更多日志逻辑
    
    return result

模型很快就给出了分析结果：

“这个函数违反了单一职责原则，它同时处理了验证、业务逻辑和日志记录三个不同的职责。代码重复也很明显，每个请求类型都有类似的日志和通知逻辑。建议拆分成多个小函数，每个函数只做一件事。”

1.2 紧耦合与硬编码

另一个常见问题是模块之间的紧耦合。比如项目中有一个订单处理系统，订单类直接依赖具体的支付网关和物流服务：

class OrderProcessor:
    def __init__(self):
        self.payment_gateway = StripeGateway()  # 直接依赖具体实现
        self.shipping_service = FedExService()  # 硬编码依赖
        
    def process_order(self, order):
        # 处理支付
        if self.payment_gateway.process_payment(order.amount):
            # 安排物流
            tracking_number = self.shipping_service.schedule_delivery(
                order.address
            )
            return tracking_number
        return None

Qwen2.5-32B-Instruct指出：“这种紧耦合的设计让系统难以扩展。如果想换用其他支付或物流服务，需要修改OrderProcessor类的代码。应该通过依赖注入的方式，让高层模块不依赖低层模块的具体实现。”

2. 设计模式选择：什么场景用什么模式

识别出问题后，下一步就是选择合适的设计模式。这是重构中最关键也最困难的一步，Qwen2.5-32B-Instruct在这方面展现出了不错的判断力。

2.1 策略模式处理多种算法

项目中有一个价格计算模块，根据不同的用户类型和促销活动，采用不同的计价策略。原来的代码是一堆if-else：

def calculate_price(user_type, promo_type, base_price):
    if user_type == 'vip':
        if promo_type == 'seasonal':
            return base_price * 0.7
        elif promo_type == 'birthday':
            return base_price * 0.8
        # ... 更多if-else
    elif user_type == 'regular':
        if promo_type == 'seasonal':
            return base_price * 0.9
        # ... 更多if-else
    # ... 更多嵌套

我让模型给出重构建议，它推荐了策略模式：

“这个场景适合使用策略模式。可以定义一个价格计算策略接口，然后为每种组合创建具体的策略类。这样新增策略时只需要添加新类，不需要修改现有代码。”

模型还给出了具体的代码示例：

from abc import ABC, abstractmethod

class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, base_price: float) -> float:
        pass

class VIPSeasonalStrategy(PricingStrategy):
    def calculate(self, base_price: float) -> float:
        return base_price * 0.7

class VIPBirthdayStrategy(PricingStrategy):
    def calculate(self, base_price: float) -> float:
        return base_price * 0.8

class RegularSeasonalStrategy(PricingStrategy):
    def calculate(self, base_price: float) -> float:
        return base_price * 0.9

class PriceCalculator:
    def __init__(self, strategy: PricingStrategy):
        self.strategy = strategy
    
    def calculate_price(self, base_price: float) -> float:
        return self.strategy.calculate(base_price)

2.2 观察者模式实现事件通知

系统中有很多地方需要在某个事件发生时通知多个组件。比如用户注册成功后，需要发送欢迎邮件、创建用户档案、更新统计数据等。

原来的实现是在注册函数里直接调用这些方法：

def register_user(user_data):
    user = create_user_in_db(user_data)
    
    # 各种通知和后续处理
    send_welcome_email(user)
    create_user_profile(user)
    update_registration_stats()
    notify_admin(user)
    # ... 更多调用
    
    return user

Qwen2.5-32B-Instruct建议使用观察者模式：

“这种一对多的依赖关系很适合观察者模式。可以定义一个事件发布者，让各个处理组件作为观察者订阅相关事件。这样新增处理逻辑时，只需要添加新的观察者，不需要修改注册函数。”

模型提供的重构思路很清晰：

class UserRegisteredEvent:
    def __init__(self, user):
        self.user = user
        self.timestamp = datetime.now()

class EventPublisher:
    def __init__(self):
        self._observers = []
    
    def attach(self, observer):
        self._observers.append(observer)
    
    def detach(self, observer):
        self._observers.remove(observer)
    
    def notify(self, event):
        for observer in self._observers:
            observer.update(event)

class EmailService:
    def update(self, event):
        if isinstance(event, UserRegisteredEvent):
            send_welcome_email(event.user)

class ProfileService:
    def update(self, event):
        if isinstance(event, UserRegisteredEvent):
            create_user_profile(event.user)

# 使用方式
publisher = EventPublisher()
publisher.attach(EmailService())
publisher.attach(ProfileService())

def register_user(user_data):
    user = create_user_in_db(user_data)
    event = UserRegisteredEvent(user)
    publisher.notify(event)
    return user

2.3 工厂模式管理对象创建

项目中有一个报表生成系统，需要根据不同的报表类型创建不同的报表对象。原来的代码是这样的：

def create_report(report_type, data):
    if report_type == 'sales':
        report = SalesReport(data)
        report.set_format('pdf')
        report.set_language('en')
    elif report_type == 'inventory':
        report = InventoryReport(data)
        report.set_format('excel')
        report.set_language('en')
    elif report_type == 'financial':
        report = FinancialReport(data)
        report.set_format('pdf')
        report.set_language('zh')
    # ... 更多类型
    
    return report

模型分析后建议使用工厂模式：

“这段代码的创建逻辑复杂，且随着报表类型增加会越来越臃肿。使用工厂模式可以将对象的创建和使用分离，让代码更清晰，也更容易扩展。”

3. 接口抽象与模块化设计

选定了设计模式，接下来就是具体的实现。Qwen2.5-32B-Instruct在接口设计和模块划分上也给出了不少实用建议。

3.1 定义清晰的接口契约

对于前面提到的紧耦合问题，模型建议先定义清晰的接口：

from abc import ABC, abstractmethod

class PaymentGateway(ABC):
    @abstractmethod
    def process_payment(self, amount: float) -> bool:
        pass
    
    @abstractmethod
    def refund_payment(self, transaction_id: str) -> bool:
        pass

class ShippingService(ABC):
    @abstractmethod
    def schedule_delivery(self, address: dict) -> str:
        pass
    
    @abstractmethod
    def track_delivery(self, tracking_number: str) -> dict:
        pass

# 具体实现
class StripeGateway(PaymentGateway):
    def process_payment(self, amount: float) -> bool:
        # Stripe具体实现
        return True
    
    def refund_payment(self, transaction_id: str) -> bool:
        # Stripe退款实现
        return True

class FedExService(ShippingService):
    def schedule_delivery(self, address: dict) -> str:
        # FedEx具体实现
        return "FEDEX123456"
    
    def track_delivery(self, tracking_number: str) -> dict:
        # FedEx跟踪实现
        return {"status": "in_transit"}

3.2 按职责划分模块

在重构一个复杂的订单处理系统时，模型建议按职责划分模块：

“原来的OrderProcessor类承担了太多职责。建议拆分成：OrderValidator（验证）、PaymentProcessor（支付）、ShippingCoordinator（物流）、NotificationService（通知）。每个类只关注自己的职责，通过清晰定义的接口协作。”

模型还给出了模块间的协作关系图（用文字描述）：

OrderService（协调者）
├── OrderValidator（验证订单数据）
├── PaymentProcessor（处理支付）
├── ShippingCoordinator（安排物流）
└── NotificationService（发送通知）

每个模块都可以独立测试和替换，OrderService只依赖它们的接口，不关心具体实现。

3.3 依赖注入解耦

对于如何管理这些依赖，模型推荐使用依赖注入：

class OrderService:
    def __init__(
        self,
        validator: OrderValidator,
        payment_processor: PaymentProcessor,
        shipping_coordinator: ShippingCoordinator,
        notifier: NotificationService
    ):
        self.validator = validator
        self.payment_processor = payment_processor
        self.shipping_coordinator = shipping_coordinator
        self.notifier = notifier
    
    def process_order(self, order_data: dict) -> dict:
        # 使用注入的依赖
        if not self.validator.validate(order_data):
            raise ValueError("Invalid order data")
        
        payment_result = self.payment_processor.process(
            order_data['amount']
        )
        
        if payment_result.success:
            tracking = self.shipping_coordinator.schedule(
                order_data['address']
            )
            self.notifier.send_order_confirmation(
                order_data['customer_email'],
                tracking
            )
            return {"status": "success", "tracking": tracking}
        
        return {"status": "failed", "reason": "payment_failed"}

# 在应用入口处组装依赖
def create_order_service():
    validator = OrderValidator()
    payment = StripePaymentProcessor(api_key="sk_test_...")
    shipping = FedExShippingCoordinator(account="fedex_account")
    notifier = EmailNotificationService()
    
    return OrderService(validator, payment, shipping, notifier)

4. 重构实战：完整案例演示

让我分享一个完整的重构案例，这是项目中一个真实的用户积分系统。

4.1 重构前的问题代码

class UserPointsSystem:
    def __init__(self):
        self.db_connection = create_db_connection()
        self.email_service = EmailService()
        self.logger = Logger()
    
    def add_points(self, user_id, points, reason):
        # 验证用户
        user = self.db_connection.query(
            "SELECT * FROM users WHERE id = %s", user_id
        )
        if not user:
            self.logger.error(f"User {user_id} not found")
            return False
        
        # 检查原因是否有效
        valid_reasons = ['purchase', 'referral', 'review', 'birthday']
        if reason not in valid_reasons:
            self.logger.error(f"Invalid reason: {reason}")
            return False
        
        # 计算实际点数（不同原因系数不同）
        if reason == 'purchase':
            actual_points = points * 1.0
        elif reason == 'referral':
            actual_points = points * 1.5
        elif reason == 'review':
            actual_points = points * 0.5
        elif reason == 'birthday':
            actual_points = points * 2.0
        
        # 更新数据库
        self.db_connection.execute(
            "UPDATE users SET points = points + %s WHERE id = %s",
            (actual_points, user_id)
        )
        
        # 记录日志
        self.logger.info(
            f"Added {actual_points} points to user {user_id} for {reason}"
        )
        
        # 发送通知
        if actual_points > 100:
            self.email_service.send(
                user['email'],
                "You earned bonus points!",
                f"You earned {actual_points} points for {reason}"
            )
        
        # 检查等级升级
        total_points = self.db_connection.query(
            "SELECT points FROM users WHERE id = %s", user_id
        )[0]['points']
        
        if total_points >= 1000 and user['level'] == 'silver':
            self.db_connection.execute(
                "UPDATE users SET level = 'gold' WHERE id = %s",
                user_id
            )
            self.email_service.send(
                user['email'],
                "Congratulations! You reached Gold level!",
                "You've been upgraded to Gold level with new benefits."
            )
        
        return True
    
    # 还有subtract_points、transfer_points等方法，结构类似

4.2 Qwen2.5-32B-Instruct的分析与建议

我把这段代码交给模型分析，它指出了多个问题：

单一职责原则违反：一个方法做了验证、计算、数据库操作、日志、通知、等级检查等太多事情
开闭原则违反：新增积分原因或等级规则需要修改现有代码
重复代码：数据库查询和更新逻辑在多个地方重复
硬编码：积分系数和等级阈值都是硬编码的

模型建议的重构方案：

使用策略模式处理不同原因的积分计算
使用观察者模式处理积分变更后的各种操作（日志、通知、等级检查）
使用仓储模式封装数据库操作
使用规格模式定义等级升级规则

4.3 重构后的代码结构

# 策略模式：不同原因的积分计算
class PointsCalculationStrategy(ABC):
    @abstractmethod
    def calculate(self, base_points: int) -> int:
        pass

class PurchasePointsStrategy(PointsCalculationStrategy):
    def calculate(self, base_points: int) -> int:
        return base_points * 1

class ReferralPointsStrategy(PointsCalculationStrategy):
    def calculate(self, base_points: int) -> int:
        return base_points * 1.5

# 观察者模式：积分变更后的处理
class PointsChangedEvent:
    def __init__(self, user_id: int, points_change: int, reason: str):
        self.user_id = user_id
        self.points_change = points_change
        self.reason = reason

class PointsObserver(ABC):
    @abstractmethod
    def on_points_changed(self, event: PointsChangedEvent):
        pass

class LoggingObserver(PointsObserver):
    def on_points_changed(self, event: PointsChangedEvent):
        logger.info(f"User {event.user_id} points changed: {event.points_change} for {event.reason}")

class EmailNotificationObserver(PointsObserver):
    def __init__(self, threshold: int = 100):
        self.threshold = threshold
    
    def on_points_changed(self, event: PointsChangedEvent):
        if event.points_change > self.threshold:
            # 发送邮件逻辑
            pass

# 仓储模式：数据访问封装
class UserRepository:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def find_by_id(self, user_id: int) -> Optional[dict]:
        # 封装查询逻辑
        pass
    
    def update_points(self, user_id: int, points_change: int):
        # 封装更新逻辑
        pass

# 规格模式：等级升级规则
class LevelUpgradeSpecification(ABC):
    @abstractmethod
    def is_satisfied_by(self, user: dict) -> bool:
        pass

class GoldLevelSpecification(LevelUpgradeSpecification):
    def is_satisfied_by(self, user: dict) -> bool:
        return user['points'] >= 1000 and user['level'] == 'silver'

# 核心服务类
class PointsService:
    def __init__(
        self,
        user_repo: UserRepository,
        strategies: Dict[str, PointsCalculationStrategy],
        observers: List[PointsObserver]
    ):
        self.user_repo = user_repo
        self.strategies = strategies
        self.observers = observers
    
    def add_points(self, user_id: int, base_points: int, reason: str) -> bool:
        # 验证用户
        user = self.user_repo.find_by_id(user_id)
        if not user:
            return False
        
        # 使用策略计算实际点数
        if reason not in self.strategies:
            return False
        
        strategy = self.strategies[reason]
        actual_points = strategy.calculate(base_points)
        
        # 更新数据
        self.user_repo.update_points(user_id, actual_points)
        
        # 通知观察者
        event = PointsChangedEvent(user_id, actual_points, reason)
        for observer in self.observers:
            observer.on_points_changed(event)
        
        return True