AI Prompt工程实战指南：从提示设计到少样本学习

## 角色
你是一个 [具体角色]，擅长 [具体技能]。

## 背景
当前任务的背景是 [真实场景描述]。

## 目标
你需要完成 [明确的目标]。

## 限制
- 限制条件 1
- 限制条件 2

## 输入
[实际数据或问题描述]

## 输出格式
[具体的输出格式要求]

## 示例
输入: [示例输入]
输出: [示例输出]

## 角色
你是一个专注于性能优化的 Python 后端工程师，擅长识别瓶颈、重构热点代码、降低复杂度。
你熟悉 asyncio、并发编程和常见的性能分析工具，但不建议为了优化牺牲可读性。

## 目标
基于输入的 CSV 文件，完成以下任务：
1. 读取数据并验证字段完整性
2. 清洗异常值（空值、负数、超出范围的数值）
3. 计算每个用户的消费总额和平均消费
4. 按消费总额降序排序
5. 输出清洗后的数据和统计结果

## 限制
- 只提取明确提及的信息，不要推断或添加内容
- 保持原文的表述风格和语序
- 如果某个字段在输入中不存在，输出 "N/A" 而非猜测
- 代码中不要使用需要额外安装的库，只使用 Python 标准库
- 处理异常时要给出明确的错误信息，而不是静默失败

## 输出格式
以 Markdown 代码块形式输出 Python 脚本，包含以下部分：
1. 文件头部注释，说明脚本用途、依赖、使用方式
2. 导入语句
3. 常量定义
4. 主函数
5. if __name__ == '__main__' 调用

代码中每个函数都要有清晰的文档字符串。

请解释以下 Python 代码的工作原理。

要求输出格式：
1. 一句话概括代码的主要功能
2. 列出代码中使用的 3 个关键函数/方法，并说明各自的作用
3. 指出代码中可能存在的性能问题或改进空间
4. 给出适合这个代码的使用场景和不适用的场景

以下是一个 MySQL 连接池的配置示例：

```python
import mysql.connector.pooling

dbconfig = {
    "host": "localhost",
    "user": "root",
    "password": "password",
    "database": "mydb",
    "pool_name": "mypool",
    "pool_size": 5,
}

pool = mysql.connector.pooling.MySQLConnectionPool(**dbconfig)

请帮我设计一个 RESTful API，用于管理用户订阅。

请按照以下思路进行设计：

**步骤 1：需求分析**
- 首先列出订阅系统需要支持的核心功能
- 标出哪些功能是必须的，哪些是可选的

**步骤 2：资源建模**
- 确定需要哪些核心资源（User, Subscription, Plan 等）
- 定义资源之间的关系

**步骤 3：API 设计**
- 为每个资源设计 CRUD 接口
- 考虑分页、过滤、排序等通用需求

**步骤 4：安全考虑**
- 哪些接口需要认证？哪些需要授权？
- 如何处理过期订阅、降级场景？

**步骤 5：错误处理**
- 列出可能出现的错误情况
- 为每种错误设计合理的 HTTP 状态码

请在每个步骤之间明确标注"---步骤 N---"，方便我阅读和分段讨论。

我需要在一个新的 Python 项目中选择异步 Web 框架，候选包括：FastAPI, Sanic, Tornado。

请从以下维度进行对比分析：
1. 性能：基于公开的基准测试数据
2. 生态成熟度：可用插件、社区支持
3. 学习曲线：文档质量、上手难度
4. 实际考虑：我的项目需要高并发、WebSocket 支持

对比完成后，请给出一个推荐，并说明在什么情况下你会推荐另一个框架。

# 单次采样（传统方式）
def single_sample_llm(prompt):
    return llm.generate(prompt)

# 自一致性采样：多次采样 + 投票
def self_consistency_llm(prompt, num_samples=5):
    reasoning_chains = []
    for _ in range(num_samples):
        # temperature>0 才会产生不同路径
        response = llm.generate(prompt, temperature=0.7)
        reasoning_chains.append(response)
    return majority_vote(reasoning_chains), reasoning_chains

reasoning_template = """
请按以下步骤回答：
问题：{question}
步骤1：分析关键点 → {step1}
步骤2：列出思路 → {step2}
步骤3：逐步推导 → {step3}
步骤4：验证合理性 → {step4}
最终答案：{final_answer}
"""

def multi_path_sampling(prompt, num_paths=5):
    paths = []
    for i in range(num_paths):
        temp = 0.6 + i * 0.1  # 0.6 ~ 1.0
        paths.append({
            'temperature': temp,
            'response': llm.generate(prompt, temperature=temp, max_tokens=1000)
        })
    return paths

def vote_on_answers(paths):
    answers = [extract_answer(p['response']) for p in paths]
    counts = Counter(answers)
    top = counts.most_common(1)[0]
    return {'final_answer': top[0], 'vote_count': top[1], 'all_answers': answers}

confidence = vote_result['vote_count'] / num_paths

def robust_answer_extraction(chain):
    patterns = [r'最终答案[：:]\s*(.+)', r'答案[：:]\s*(.+)',
                r'结论[：:]\s*(.+)', r'因此[，,]\s*(.+)']
    for p in patterns:
        m = re.search(p, chain)
        if m:
            return m.group(1).strip()
    return chain.strip().split('。')[-1]  # 兜底：最后一句

def normalize_answer(answer):
    return {'是':'yes','对':'yes','正确':'yes',
            '不是':'no','错':'no','错误':'no'}.get(answer.lower().strip(), answer)

async def async_multi_path_sampling(prompt, num_paths=3):
    tasks = [llm.agenerate_async(prompt, temperature=0.6 + i*0.1)
             for i in range(num_paths)]
    return await asyncio.gather(*tasks)

请先快速评估以下代码的复杂度（简单/中等/复杂），然后根据复杂度采用不同的分析策略。

**简单代码**（< 50 行，逻辑单一）：
- 直接给出功能说明
- 指出潜在 bug 或改进点

**中等代码**（50-200 行，有多个函数或类）：
- 先拆分功能模块
- 每个模块单独分析
- 最后给出整体架构评价

**复杂代码**（> 200 行，涉及多个模块或设计模式）：
- 先画出模块依赖关系
- 标出核心路径和边界情况
- 分析设计模式的适用性

def get_model(num_classes, pretrained=True, freeze_backbone=True):
    model = models.resnet50(pretrained=pretrained)

    if freeze_backbone:
        for param in model.parameters():
            param.requires_grad = False
        model.fc = nn.Linear(model.fc.in_features, num_classes)
    else:
        model.fc = nn.Linear(model.fc.in_features, num_classes)

    return model

# 小数据情况：冻结骨干网络，只训练分类头
model = get_model(num_classes=11, pretrained=True, freeze_backbone=True)
optimizer = torch.optim.Adam(model.fc.parameters(), lr=1e-3)

class MAML:
    def __init__(self, model, inner_lr=1e-3, meta_lr=1e-3, inner_steps=5):
        self.model = model
        self.meta_optimizer = optim.Adam(model.parameters(), lr=meta_lr)

    def inner_loop(self, support_data, support_labels, query_data, query_labels):
        fast_weights = [p.clone() for p in self.model.parameters()]

        for _ in range(self.inner_steps):
            logits = self.model.functional_forward(support_data, fast_weights)
            loss = nn.CrossEntropyLoss()(logits, support_labels)
            grads = torch.autograd.grad(loss, fast_weights)
            fast_weights = [w - self.inner_lr * g for w, g in zip(fast_weights, grads)]

        logits = self.model.functional_forward(query_data, fast_weights)
        return nn.CrossEntropyLoss()(logits, query_labels)

def compute_prototypes(support_features, support_labels, num_classes):
    prototypes = []
    for c in range(num_classes):
        mask = (support_labels == c)
        if mask.sum() > 0:
            prototypes.append(support_features[mask].mean(dim=0))
        else:
            prototypes.append(torch.zeros_like(support_features[0]))
    return torch.stack(prototypes)

def prototypical_loss(query_features, query_labels, prototypes):
    distances = torch.cdist(query_features, prototypes)
    logits = -distances
    return nn.CrossEntropyLoss()(logits, query_labels)

strong_transform = A.Compose([
    A.RandomResizedCrop(height=224, width=224, scale=(0.5, 1.0)),
    A.HorizontalFlip(p=0.5),
    A.VerticalFlip(p=0.3),
    A.RandomBrightnessContrast(p=0.8),
    A.ShiftScaleRotate(p=0.8, shift_limit=0.2, scale_limit=0.2, rotate_limit=30),
    A.OneOf([
        A.GaussianBlur(p=1.0),
        A.MotionBlur(p=1.0),
        A.MedianBlur(p=1.0),
    ], p=0.5),
    A.CoarseDropout(max_holes=8, max_height=32, max_width=32, p=0.5),
    A.Normalize(),
    ToTensorV2(),
])

# 标签平滑
def label_smoothing_loss(pred, target, smoothing=0.1):
    n_classes = pred.size(1)
    one_hot = torch.zeros_like(pred).scatter(1, target.unsqueeze(1), 1)
    one_hot = one_hot * (1 - smoothing) + smoothing / n_classes
    return nn.KLDivLoss(reduction='batchmean')(torch.log_softmax(pred, dim=1), one_hot)

# Mixup
def mixup_data(x, y, alpha=0.2):
    lam = np.random.beta(alpha, alpha)
    batch_size = x.size(0)
    index = torch.randperm(batch_size)
    mixed_x = lam * x + (1 - lam) * x[index]
    return mixed_x, y, y[index], lam

# 方案一：加权损失
class_weights = compute_class_weight('balanced', classes=np.unique(train_labels), y=train_labels)
criterion = nn.CrossEntropyLoss(weight=torch.tensor(class_weights, dtype=torch.float32))

# 方案二：过采样
ros = RandomOverSampler(random_state=42)
train_data_resampled, train_labels_resampled = ros.fit_resample(train_data, train_labels)

from transformers import pipeline

classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")

result = classifier(
    "这个手机续航差劲，一天要充三次电",
    candidate_labels=["正面", "负面", "中性"]
)
# 输出：负面（置信度 0.894）

# 不好的标签
candidate_labels = ["好", "坏", "一般"]

# 好的标签
candidate_labels = ["服务态度好", "服务态度差", "服务态度一般"]

# 这个顺序给"正面"更多机会
classifier(text, candidate_labels=["正面", "负面", "中性"])

# 这个顺序给"中性"更多机会
classifier(text, candidate_labels=["中性", "正面", "负面"])

result = classifier(
    "我想取消订单",
    candidate_labels=["取消订单", "查询订单", "修改订单"],
    hypothesis_template="这句话的意图是 {}。"
)

classifier = pipeline(
    "zero-shot-classification",
    model="uer/roberta-base-finetuned-chinanews-chinese-news-classifier"
)

# 医疗领域
classifier = pipeline(
    "zero-shot-classification",
    model="microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract-fulltext"
)

def classify_with_threshold(text, labels, threshold=0.55):
    result = classifier(text, labels)
    if result['scores'][0] > threshold:
        return result['labels'][0]
    else:
        return "不确定"

result = classifier(
    "这个产品价格还可以，但是物流太慢了",
    candidate_labels=["价格相关", "物流相关", "服务相关"],
    multi_label=True
)
# 结果：[("物流相关", 0.89), ("价格相关", 0.76), ("服务相关", 0.21)]

要求：
- 总字数不超过 500 字
- 每个概念解释不超过 2 句话
- 只解释最核心的 3 个概念

请对以下代码进行审查，重点关注以下问题：

1. 安全问题：SQL 注入、XSS、未过滤用户输入
2. 性能问题：N+1 查询、不必要的计算
3. 可维护性：命名规范、代码重复、缺少注释
4. 边界情况：空值处理、异常捕获、并发问题

对于每个发现的问题，请按以下格式输出：
- **问题类型**：[安全/性能/可维护性/其他]
- **严重程度**：[高/中/低]
- **具体位置**：指出代码行号或函数名
- **问题描述**：简要说明问题是什么
- **修复建议**：给出具体的修复方案
- **代码示例**：如果适用，提供修复后的代码片段

我需要在 Redis 和 Memcached 之间选择一个缓存方案：

业务背景：
- 高并发读取（主要操作是 get）
- 需要支持过期时间（TTL）
- 预计 QPS 峰值在 5k-10k 之间

请对比 Redis 和 Memcached 在上述场景下的适用性，包括：
1. 性能差异（吞吐量、延迟）
2. 功能差异（数据结构、持久化、集群）
3. 运维复杂度（部署、监控、故障恢复）

最后给出一个明确的推荐，并说明在什么情况下你会改变这个推荐。

我的 Node.js 服务最近出现内存泄漏：

现象描述：
- 服务运行 2-3 天后内存从 200MB 增长到 2GB
- CPU 使用率稳定
- 重启后恢复正常

环境信息：
- Node.js v18.16.0
- 框架：Express.js
- 数据库：MongoDB（mongoose）
- 消息队列：RabbitMQ（amqplib）

请帮我分析可能的原因，并给出排查步骤：
1. 列出最可能的 3-5 个原因，按可能性排序
2. 针对每个原因，给出具体的排查方法
3. 提供一段诊断脚本
4. 给出预防类似问题的最佳实践

from sklearn.metrics import classification_report, f1_score

# 不要只看 accuracy
print(classification_report(target, pred))

# 特别关注 Macro F1
f1_macro = f1_score(target, pred, average='macro')