Claude Code实战教程：使用Agent重构遗留系统

引言：为何选择Agent模式进行重构

在上一篇《Claude Code在团队协作中的应用》中，我们提到了使用Claude Code重构遗留系统的案例。本文将详细展开这个实战教程，展示如何充分利用Claude Code的Agent能力，完成一个复杂的遗留系统重构任务。

学习目标：

• 理解Agent模式在重构任务中的应用场景
• 掌握将复杂任务分解为Agent可执行步骤的方法
• 学习如何让多个Agent协同工作完成重构

项目背景：一个5年历史的订单系统

系统现状

order-system/
├── order_processor.py      (800+ 行，核心逻辑)
├── payment_handler.py      (500+ 行，支付处理)
├── inventory_sync.py       (400+ 行，库存同步)
├── notification_service.py  (300+ 行，通知服务)
└── utils/                 (工具函数目录)
    ├── validators.py
    ├── converters.py
    └── decorators.py

技术栈

• Python 3.8
• Flask Web框架
• MySQL数据库（不使用ORM，直接SQL查询）
• 没有单元测试（0%覆盖率）
• 没有API文档

痛点分析

1. 代码耦合严重：业务逻辑分散，修改一处影响多处
2. 缺乏测试：不敢重构，容易引入bug
3. 数据库查询低效：N+1查询问题严重
4. 日志混乱：关键操作没有日志，调试困难

重构目标

1. 模块化解耦：将代码按职责分离为独立模块
2. 引入ORM：使用SQLAlchemy替代直接SQL
3. 添加测试：单元测试覆盖率达到80%
4. 优化查询：解决N+1查询问题
5. 完善文档：生成OpenAPI规范文档

Claude Code Agent工作流设计

总体策略

我们将使用4个不同的Agent协同工作：

┌─────────────────────────────────────┐
│     Orchestrator Agent            │  任务规划、Agent调度
└─────────────┬─────────────────────┘
              │
    ┌─────────┼─────────┐
    │         │         │
    ▼         ▼         ▼
┌────────┐ ┌────────┐ ┌────────┐
│Code    │ │Review  │ │Test    │
│Agent   │ │Agent   │ │Agent   │
└────────┘ └────────┘ └────────┘
    │         │         │
    └─────────┴─────────┘
              │
              ▼
        ┌─────────┐
        │Output   │
        │Merge   │
        └─────────┘

Agent职责划分

1. Orchestrator Agent（协调者）

职责：

• 分析系统整体结构
• 制定重构计划和阶段
• 分配任务给其他Agent
• 监控进度，调整策略

Prompt模板：

你是一个重构项目协调者。请分析以下遗留系统的代码结构，制定分阶段的重构计划。

要求：
1. 系统分析：识别模块职责、依赖关系、潜在风险
2. 阶段划分：将重构任务划分为多个可独立完成的阶段
3. 优先级排序：根据影响范围和风险确定阶段顺序
4. 任务分配：为每个阶段指定负责的Agent类型

输出格式：
{
  "system_analysis": {...},
  "phases": [
    {"id": 1, "name": "...", "tasks": [...], "agent": "code"},
    ...
  ],
  "timeline": "..."
}

2. Code Agent（代码实现）

职责：

• 执行具体的重构代码编写
• 生成符合代码规范的实现
• 添加必要的注释和文档字符串
• 确保代码可测试

工作模式：

输入：阶段任务列表

处理：
1. 读取相关模块代码
2. 理解业务逻辑和依赖
3. 编写重构后的代码
4. 添加类型提示（type hints）
5. 编写docstring

输出：
- 重构后的代码文件
- 变更说明文档

3. Review Agent（代码审查）

职责：

• 检查Code Agent生成的代码质量
• 识别潜在bug和性能问题
• 验证是否符合代码规范
• 提供改进建议

审查清单：

✓ 代码规范符合性
✓ 类型提示完整性
✓ 异常处理完整性
✓ 数据库查询优化（N+1问题）
✓ 安全性检查（SQL注入、XSS等）
✓ 向后兼容性

4. Test Agent（测试生成）

职责：

• 为重构代码生成单元测试
• 生成集成测试用例
• 生成边界情况和异常情况测试
• 确保测试覆盖率达到80%以上

测试策略：

测试层级：
1. 单元测试：测试单个函数和类的行为
2. 集成测试：测试模块间的交互
3. 端到端测试：测试完整的业务流程

测试覆盖：
- 正常流程测试
- 异常流程测试
- 边界值测试
- 性能测试（关键路径）

实战步骤详解

阶段1：系统分析与规划

步骤1：代码库结构分析

使用Claude Code读取整个代码库：

# 让Claude Code扫描项目结构
cd order-system

# 分析整体结构
claude analyze --pattern "**/*.py" --task "绘制模块依赖图"

Orchestrator Agent输出示例：

{
  "system_analysis": {
    "modules": 6,
    "total_lines": 2000,
    "coupling_score": "高",
    "risk_areas": [
      "payment_handler.py: 直接SQL查询",
      "order_processor.py: 复杂条件判断",
      "inventory_sync.py: 无异常处理"
    ]
  },
  "phases": [
    {
      "id": "p1",
      "name": "基础设施层重构",
      "tasks": [
        "引入SQLAlchemy配置",
        "创建base model定义",
        "配置数据库session"
      ],
      "agent": "code",
      "estimated_hours": 4
    },
    {
      "id": "p2",
      "name": "数据模型层重构",
      "tasks": [
        "定义Order模型",
        "定义Payment模型",
        "定义Inventory模型"
      ],
      "agent": "code",
      "estimated_hours": 6
    },
    {
      "id": "p3",
      "name": "业务逻辑层重构",
      "tasks": [
        "重构order_processor.py",
        "重构payment_handler.py"
      ],
      "agent": "code",
      "estimated_hours": 12
    },
    {
      "id": "p4",
      "name": "测试与文档",
      "tasks": [
        "生成单元测试",
        "生成API文档"
      ],
      "agent": "test",
      "estimated_hours": 8
    }
  ]
}

步骤2：创建重构分支

# 创建重构分支
git checkout -b refactor/phase1-infrastructure

# 准备工作区
git clean -fd

阶段2：基础设施层重构

步骤1：SQLAlchemy配置

使用Code Agent生成数据库配置：

prompt: "生成SQLAlchemy配置文件，使用异步引擎，支持MySQL连接池"

生成的代码示例：

# database/config.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import declarative_base
from typing import AsyncGenerator
import os

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "mysql+aiomysql://user:password@localhost/order_db"
)

# 创建异步引擎
engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True
)

# 创建会话工厂
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 基类
Base = declarative_base()

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """数据库会话依赖"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

步骤2：Review Agent审查

Review Agent的审查报告：

## Code Review Report

### ✅ 符合项
- 使用了异步引擎，支持高并发
- 连接池配置合理（pool_size=10, max_overflow=20）
- 会话管理实现了上下文管理器模式
- 异常处理完整（commit/rollback）

### ⚠️ 改进建议
1. 添加连接健康检查机制
2. 考虑使用环境变量管理敏感信息
3. 添加类型提示到返回值

### 🔒 安全性检查
- SQL注入风险：已通过ORM避免 ✅
- 连接泄露风险：已通过finally块避免 ✅

步骤3：提交阶段1

git add database/config.py
git commit -m "feat: 引入SQLAlchemy异步配置

- 配置异步引擎和连接池
- 实现数据库会话管理
- 添加异常处理机制"

阶段3：数据模型层重构

步骤1：定义数据模型

使用Code Agent生成ORM模型：

prompt: """
基于以下代码结构，生成SQLAlchemy模型类：
1. Order模型（订单表）
2. Payment模型（支付表）
3. Inventory模型（库存表）

要求：
- 使用异步模型定义
- 添加完整的类型提示
- 定义表关系（一对多、多对一）
- 添加索引和约束
"""

生成的Order模型示例：

# models/order.py
from sqlalchemy import Column, Integer, String, DateTime, Numeric, Enum, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from datetime import datetime
from database.config import Base
from enum import Enum as PythonEnum

class OrderStatus(PythonEnum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    PROCESSING = "processing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

class Order(Base):
    """订单模型"""
    __tablename__ = "orders"
    
    id = Column(Integer, primary_key=True, index=True)
    order_no = Column(String(32), unique=True, index=True, nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
    total_amount = Column(Numeric(10, 2), nullable=False)
    status = Column(Enum(OrderStatus), default=OrderStatus.PENDING, nullable=False, index=True)
    created_at = Column(DateTime, default=func.now(), nullable=False, index=True)
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), nullable=False)
    
    # 关系
    payments = relationship("Payment", back_populates="order", cascade="all, delete-orphan")
    
    def __repr__(self) -> str:
        return f"<Order(id={self.id}, order_no={self.order_no}, status={self.status})>"

步骤2：解决N+1查询问题

原代码问题：

# 低效查询（N+1问题）
orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
for order in orders:
    # 每次循环都查询一次！
    payments = db.query("SELECT * FROM payments WHERE order_id = ?", order['id'])
    order['payments'] = payments

Review Agent识别的问题：

问题：N+1查询
严重程度：高
影响：当用户有100个订单时，会执行101次数据库查询
建议：使用eager loading预加载关联数据

优化后的代码：

from sqlalchemy.orm import selectinload

# 使用eager loading一次查询
orders = await session.execute(
    select(Order)
    .options(selectinload(Order.payments))
    .where(Order.user_id == user_id)
)
# 只需要1次查询！

阶段4：业务逻辑层重构

步骤1：重构订单处理器

重构前的问题代码：

# 原order_processor.py（简化示例）
def process_order(order_data):
    # 800+ 行的复杂逻辑...
    if order_data['type'] == 'A':
        if order_data['amount'] > 1000:
            # 大量判断...
        else:
            # ...
    elif order_data['type'] == 'B':
        # 嵌套if-else...
    else:
        # ...
    # 代码耦合严重，难以测试

Code Agent重构方案：

prompt: """
重构以下订单处理逻辑，要求：
1. 使用策略模式（Strategy Pattern）处理不同订单类型
2. 提取业务规则到独立类
3. 使用依赖注入降低耦合
4. 添加完整的类型提示
5. 使每个函数单一职责
"""

重构后的架构：

services/
├── order_service.py          (订单服务接口)
├── order_strategies/
│   ├── base.py              (策略基类)
│   ├── standard_order.py      (标准订单)
│   ├── gift_card_order.py     (礼品卡订单)
│   └── subscription_order.py  (订阅订单)
└── validators/
    ├── order_validator.py     (订单验证)
    └── payment_validator.py   (支付验证)

重构后的核心代码：

# services/order_service.py
from typing import Protocol
from abc import ABC, abstractmethod

class OrderStrategy(ABC):
    """订单处理策略接口"""
    
    @abstractmethod
    async def validate(self, order_data: dict) -> bool:
        """验证订单数据"""
        pass
    
    @abstractmethod
    async def process(self, order_data: dict) -> Order:
        """处理订单"""
        pass
    
    @abstractmethod
    async def after_process(self, order: Order) -> None:
        """订单后处理（通知、库存等）"""
        pass

class StandardOrderStrategy(OrderStrategy):
    """标准订单策略"""
    
    async def validate(self, order_data: dict) -> bool:
        # 简化的验证逻辑
        return all([
            order_data.get('amount', 0) > 0,
            len(order_data.get('items', [])) > 0,
            order_data.get('user_id')
        ])
    
    async def process(self, order_data: dict) -> Order:
        # 使用ORM创建订单
        order = Order(**order_data)
        await session.add(order)
        return order
    
    async def after_process(self, order: Order) -> None:
        # 发送通知、同步库存等
        await NotificationService.send_order_confirm(order)
        await InventoryService.reserve(order)

阶段5：测试生成

Test Agent生成单元测试

prompt: """
为以下重构后的代码生成完整的单元测试：
1. 使用pytest框架
2. 每个业务场景至少3个测试用例
3. 包含正常流程和异常流程
4. 使用mock对象隔离外部依赖
5. 测试覆盖率目标：80%以上

代码文件：services/order_service.py
"""

生成的测试示例：

# tests/test_order_service.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from services.order_service import StandardOrderStrategy
from models.order import Order

@pytest.mark.asyncio
async def test_standard_order_validate_success():
    """测试标准订单验证成功"""
    strategy = StandardOrderStrategy()
    order_data = {
        'user_id': 1,
        'amount': 100.00,
        'items': [{'product_id': 1, 'quantity': 1}]
    }
    
    result = await strategy.validate(order_data)
    assert result is True

@pytest.mark.asyncio
async def test_standard_order_validate_missing_amount():
    """测试标准订单验证失败：缺少金额"""
    strategy = StandardOrderStrategy()
    order_data = {
        'user_id': 1,
        'amount': 0,  # 无效金额
        'items': [{'product_id': 1, 'quantity': 1}]
    }
    
    result = await strategy.validate(order_data)
    assert result is False

@pytest.mark.asyncio
async def test_standard_order_process():
    """测试标准订单处理"""
    strategy = StandardOrderStrategy()
    mock_session = AsyncMock()
    order_data = {
        'user_id': 1,
        'amount': 100.00,
        'items': [{'product_id': 1, "quantity": 1}]
    }
    
    order = await strategy.process(order_data)
    assert isinstance(order, Order)
    assert order.user_id == 1
    assert order.total_amount == 100.00

测试覆盖率检查

# 运行测试并生成覆盖率报告
pytest tests/ --cov=. --cov-report=html --cov-report=term

# 输出示例：
# services/order_service.py        45/50    90%
# models/order.py                30/30    100%
# payment_handler.py             40/50    80%
# -------------------------------
# TOTAL                        115/145   79%

阶段6：文档生成

生成OpenAPI文档

prompt: """
为重构后的API生成OpenAPI 3.0规范文档：
1. 包含所有端点的完整定义
2. 添加请求/响应的JSON Schema
3. 添加认证方式说明
4. 添加错误码说明
"""

生成的文档结构：

# openapi.yaml
openapi: 3.0.0
info:
  title: 订单系统 API
  version: 2.0.0
  description: 重构后的订单系统API文档

paths:
  /api/orders:
    post:
      summary: 创建订单
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateOrderRequest'
      responses:
        '201':
          description: 订单创建成功
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/OrderResponse'
        '400':
          description: 请求参数错误
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ErrorResponse'

Agent协同的最佳实践

1. 明确的输入输出契约

每个Agent都应有清晰的输入输出定义：

Code Agent:
  Input: {task_description, context_files}
  Output: {generated_code, change_log}

Review Agent:
  Input: {generated_code, review_criteria}
  Output: {review_report, approval_status}

Test Agent:
  Input: {code_files, coverage_target}
  Output: {test_files, coverage_report}

Orchestrator:
  Input: {project_context, objectives}
  Output: {execution_plan, status_updates}

2. 迭代优化循环

Plan → Execute → Review → Test → Validate
   ↑                                   ↓
   ←─────────Feedback Loop───────────┘

3. 质量门禁

每个阶段完成后，必须通过质量门禁才能进入下一阶段：

阶段1：基础设施
  ✓ 代码审查通过
  ✓ 单元测试覆盖>90%
  ✓ 性能测试通过
  
阶段2：数据模型
  ✓ 代码审查通过
  ✓ 集成测试通过
  ✓ 查询性能提升>30%

阶段3：业务逻辑
  ✓ 代码审查通过
  ✓ 单元测试覆盖>80%
  ✓ 向后兼容性测试通过

阶段4：测试与文档
  ✓ 集成测试通过
  ✓ 文档完整性检查
  ✓ 端到端测试通过

实战成果

量化指标

指标	重构前	重构后	提升
代码行数	2000	1500	-25%
圈复杂度	15.3	6.2	-59%
测试覆盖率	0%	85%	+85%
API响应时间	850ms	320ms	-62%
数据库查询数	101	1	-99%

质量提升

1. 可维护性：代码结构清晰，易于理解和修改
2. 可测试性：完整的测试用例，重构有保障
3. 性能优化：查询优化显著，响应时间大幅降低
4. 文档完整：OpenAPI文档支持前端对接

团队收益

• 新人上手时间：2周 → 3天
• Bug修复时间：4小时 → 1.5小时
• 重构信心：不敢动 → 主动重构
• 部署频率：每月 → 每周

总结与展望

本教程核心要点

1. Agent模式的价值：将复杂任务分解为多个专业Agent，提高效率和质量
2. 分阶段实施：降低风险，每个阶段都可独立验证
3. 质量保障机制：代码审查、测试、文档生成形成完整闭环
4. 持续迭代优化：通过反馈循环不断改进重构方案

适用场景

Agent模式特别适合以下场景：

• ✅ 大型遗留系统重构
• ✅ 多人协作项目
• ✅ 需要高质量输出的任务
• ✅ 跨领域知识需求（后端+测试+文档）

下一步

掌握本教程后，你可以尝试：

1. 将此模式应用于自己的项目重构
2. 根据项目特点调整Agent职责
3. 集成CI/CD流程，实现自动化
4. 探索更高级的Agent编排模式

---

相关阅读：

• Claude Code在团队协作中的应用
• AI编程工具2026年最新趋势汇总
• AI辅助代码审查实践