Prompt写得好不好，AI代码质量差了一倍：我用Claude 3重构电商推荐系统的血泪史

「这个推荐系统该重写了」——当我看到3000行Python代码时

上周三凌晨2点，我被报警短信吵醒——我们为东南亚电商平台ShopLuxe搭建的推荐系统又崩了。这已经是本月第三次，每次都是因为内存泄漏导致K8s pod被OOMKilled。我盯着监控面板上那条陡峭的内存曲线，意识到这个三年前写的系统已经撑不住日均200万的用户请求了。

翻出当年的代码，我差点把咖啡喷在屏幕上：

# 2019年版推荐算法核心逻辑
def recommend_products(user_id):
    # 这里硬编码了20个"热门商品"
    hot_items = [1024, 2048, 3072...] 
    
    # 从MySQL直接查用户历史行为
    history = db.query(f"SELECT * FROM clicks WHERE user_id={user_id}")
    
    # 用余弦相似度计算推荐（但特征只有商品类别）
    recommendations = []
    for item in hot_items:
        similarity = calculate_cosine(item.category, history[-1].category)
        if similarity > 0.5:
            recommendations.append(item)
    
    return recommendations[:5]  # 魔法数字5

这段代码至少有五个致命问题：

• 硬编码的热门商品列表，三年没更新过
• 直接拼接SQL导致注入风险
• 特征工程简陋到令人发指（只有商品类别）
• 没有缓存机制，每次请求都查数据库
• 返回数量写死在代码里

更可怕的是，随着业务增长，这个系统已经衍生出十几个if-else分支来处理各种特殊场景（比如节日促销、新用户冷启动）。当我用pycallgraph生成调用图时，得到的是一团意大利面条。

「用Claude 3重写？先过Prompt设计这一关」

CTO建议我用Claude 3试试AI辅助重构。我最初不以为然——直到看到两个不同Prompt生成的代码对比：

这是我最终打磨出来的Prompt（关键部分用注释说明为什么这么写）：

"""
你是一个资深推荐系统工程师，需要为ShopLuxe电商平台重构Python推荐服务。具体要求：

1. 输入输出
- 输入：用户ID(int)、上下文(包含设备、地理位置等)
- 输出：包含5-10个推荐商品的数组，按相关性降序排列

2. 技术约束
- 使用Python 3.10+类型提示
- 必须用Redis缓存用户特征
- 禁止SQL拼接，用ORM或参数化查询
- 支持AB测试分流逻辑

3. 算法要求
- 冷启动：新用户用基于地理位置的流行度推荐
- 老用户：结合协同过滤和内容特征（商品类别、价格段等）
- 实时性：最近1小时点击行为权重加倍

4. 性能指标
- 99分位延迟<200ms
- 支持1000 QPS
- 内存占用<500MB

请给出完整实现，包含：
- 类型定义
- 缓存逻辑
- 异常处理
- 单元测试示例
"""

「从意大利面条到分层架构」——新系统设计

Claude 3生成的初版代码已经比旧系统强很多，但还需要人工调整。最终架构分为四层：

# 数据访问层（用Pydantic做类型校验）
class UserFeatures(BaseModel):
    user_id: int
    click_history: List[Product]
    device_type: Literal["mobile", "desktop"]
    
    @validator("user_id")
    def check_id(cls, v):
        if v < 1:
            raise ValueError("Invalid user ID")

# 缓存层（Redis+本地缓存二级缓存）
def get_user_features(user_id: int) -> UserFeatures:
    # 先查本地缓存
    if features := local_cache.get(user_id):
        return features
        
    # 再查Redis
    redis_key = f"user:{user_id}:features"
    if features := redis.get(redis_key):
        return UserFeatures.parse_raw(features)
        
    # 最后查数据库（带TTL缓存）
    features = db.query_user(user_id)
    redis.setex(redis_key, 3600, features.json())
    return features

# 算法层（策略模式）
class Recommender(ABC):
    @abstractmethod
    def recommend(self, user: UserFeatures) -> List[int]:
        pass

class CFRecommender(Recommender):
    def __init__(self, k=20):
        self.k = k  # 近邻数量
        
    def recommend(self, user: UserFeatures) -> List[int]:
        # 实现协同过滤逻辑
        pass

# 服务层（FastAPI暴露HTTP接口）
@app.get("/recommend")
async def recommend_api(
    user_id: int, 
    context: RecommendationContext
) -> List[Product]:
    try:
        features = get_user_features(user_id)
        recommender = get_ab_test_recommender(user_id)  # AB测试分流
        return recommender.recommend(features)
    except Exception as e:
        logger.error(f"Recommend failed for {user_id}: {e}")
        return get_fallback_recommendations()  # 降级逻辑

这个架构带来的性能提升：

• 平均延迟从320ms → 89ms
• 错误率从5.2% → 0.3%
• 内存占用峰值从1.4GB → 210MB

「AB测试模块差点让我翻车」——最坑的调试经历

最棘手的问题出现在AB测试模块。最初Claude 3生成的版本是这样的：

def get_ab_test_recommender(user_id: int) -> Recommender:
    bucket = user_id % 100  # 简单哈希分桶
    if bucket < 50:
        return CFRecommender()
    else:
        return ContentBasedRecommender()

上线后监控显示，新算法组（内容推荐）的点击率比对照组低37%！经过排查发现两个问题：

1. user_id在东南亚地区有大量以0结尾的号码，导致分桶不均匀
2. 没有考虑用户分群（新老用户应该用不同策略）

最终修复方案：

def get_ab_test_recommender(user: UserFeatures) -> Recommender:
    # 用更均匀的哈希算法
    bucket = xxhash.xxh32(str(user.user_id)).intdigest() % 100
    
    # 新用户单独分组
    if len(user.click_history) < 5:
        return ColdStartRecommender(user.geo)
        
    # 老用户AB测试
    if bucket < 40:
        return HybridRecommender()
    elif bucket < 80:
        return CFRecommender(k=30)
    else:  # 保留20%流量给旧算法做对照
        return LegacyRecommender()

这个改动让点击率回升并最终超过原算法15%。教训是：永远不要假设用户ID是均匀分布的！

「Prompt工程比我想象的更玄学」——经验总结

经过这次重构，我总结出几个Prompt编写原则：

最关键的是要把AI当作一个需要明确需求的”初级工程师”，而不是能读心的魔法黑盒。以下是我现在写Prompt的固定结构：

1. 角色设定（你是什么领域的专家）
2. 输入输出规范（数据类型、范围）
3. 技术约束（语言版本、禁止模式）
4. 业务规则（算法逻辑、特殊场景）
5. 非功能需求（性能、安全等）
6. 输出格式（需要包含哪些部分）

这套方法不仅适用于Claude，在GPT-4和Gemini上同样有效。现在我的Prompt平均长度从原来的20字增加到300字，但生成的代码质量提升了至少一个数量级。

「AI生成代码的维护成本并不低」——那些没人告诉你的真相

虽然用Claude 3重构很爽，但上线后我发现了AI生成代码的阴暗面：

• 过度抽象：AI喜欢把简单逻辑拆成多个小类，导致调用链过长
• 魔法数字：生成的代码常有未解释的阈值（比如相似度>0.7）
• 文档缺失：90%的生成函数没有docstring

这是我的应对方案：

# 在CI流水线中添加AI代码检查步骤
- step:
    name: AI代码质量门禁
    script:
      # 检查文档覆盖率
      - pylint --disable=all --enable=missing-docstring *.py
      # 检测魔法数字
      - grep -rn "b0.[0-9]+b" --include="*.py" src/
      # 验证类型提示覆盖率
      - mypy --strict src/

另外必须建立代码所有权制度——AI生成的代码必须经过人工审核和测试后才能合并，并且要有明确的负责人。我们的新规是：

1. AI生成的代码必须添加# Generated by Claude 3注释
2. 每段生成代码要有对应的验证测试
3. 禁止直接复制粘贴未经修改的生成代码

说实话，维护AI代码比想象中费劲，但相比从零手写还是节省了60%的时间。关键在于建立正确的预期——这不是银弹，而是超级智能的代码助手。

# 老代码中的"瑞士军刀"式函数
def calculate_similarity(item1, item2, user=None):
    if user and user.vip_level > 3:
        # VIP用户专属逻辑
        ...
    elif item1.category == 'electronics':
        # 电子产品特殊处理
        ...
    elif time.localtime().tm_hour > 22:
        # 夜间流量降级策略
        ...

"""重构以下Python代码，要求：
1. 使用策略模式分离不同品类的特征提取逻辑
2. 新增GPU加速支持但保留CPU回退
3. 日志记录要包含特征提取耗时分布
背景：当前代码导致20%的推荐延迟超过500ms"""

# 原始代码...
# [原有300行特征处理代码]

logger.info(f"Recommend for {user.id} "
           f"with {len(items)} candidates, "
           f"context: {json.dumps(context)}")

logger.info("Recommend completed",
           extra={
               'user': user.id[:8],  # 截断非必要信息
               'item_count': len(items),
               'ctx_keys': list(context.keys())  # 不序列化整个上下文
           })

.positive {
color: #2ecc71;
font-weight: bold;
}
.code-comparison {
display: flex;
gap: 20px;
margin: 15px 0;
}
.metrics-table {
width: 100%;
border-collapse: collapse;
}
.metrics-table th, .metrics-table td {
padding: 8px 12px;
border: 1px solid #ddd;
}

# 灾难代码示例
def load_features():
    # 一次性加载所有国家商品数据
    all_products = []
    for country in ['MY','SG','TH','VN','ID','PH']:
        with open(f'/data/{country}_products.json') as f:
            all_products += json.load(f)  # 内存炸弹！
    
    # 更可怕的是这个全局缓存
    global FEATURE_CACHE
    FEATURE_CACHE = {p['sku']: p for p in all_products}
    return FEATURE_CACHE

# 改造后代码
def get_product_features(sku_ids: list[str]):
    with redis.pipeline() as pipe:
        for sku in sku_ids:
            pipe.hmget(f'product:{sku}', ['price','rating','sales'])
        return [parse_features(r) for r in pipe.execute()]