AI原生应用开发中的提示工程原理与实战：从基础到生产环境部署

1次阅读

共计 3858 个字符，预计需要花费 10 分钟才能阅读完成。

最近在开发客服工单分类系统时，我们遇到这样的问题：当用户描述中包含多个产品名称时，GPT-3.5 经常将工单错误分类到次要产品类别。例如用户说 ”A 产品的数据导出功能在 B 产品界面上报错 ”，模型有 60% 概率错误分类到 B 产品。这种提示设计缺陷直接导致 20% 的工单需要人工重新分配。

通过分析发现主要存在三类典型问题：

语义漂移 ：连续对话中模型逐渐偏离核心主题
指令遗忘 ：复杂任务中忽略关键约束条件（如输出格式要求）
上下文爆炸 ：超过 token 限制时随机截断重要历史信息

# 典型的问题提示示例
def generate_bad_prompt(user_input):
    return f""" 请分类以下工单：{user_input}
    """  # 缺少明确分类标准和产品范围定义

适用场景 ：简单明确的分类 / 生成任务
优势：无需示例，开发速度快
风险点 ：对模糊指令容错性差

def zero_shot_prompt(text):
    return """ 请判断以下文本情感倾向：[正面 / 负面 / 中立]
    文本：{text}"""

适用场景 ：需要特定输出格式的复杂任务
最佳实践 ：3- 5 个覆盖边界的示例
注意事项 ：示例质量 > 数量

def few_shot_prompt():
    examples = """
    示例 1:
    输入："屏幕有闪烁条纹"
    输出：{"product":"显示器", "category":"硬件故障"}

    示例 2:
    输入："无法登录邮箱"
    输出：{"product":"企业邮箱", "category":"账号问题"}
    """return examples +"\n 请根据示例格式处理新工单："

适用场景 ：需要多步推理的数学 / 逻辑问题
实施要点 ：必须包含 ” 让我们逐步思考 ” 等触发词
效果验证 ：在 GSM8K 数学数据集上准确率提升 35%

def chain_of_thought(question):
    return f""" 请解决以下数学题，并展示推理步骤：问题：{question}
    让我们逐步思考："""

决策树选择工具 ：

graph TD
    A[任务类型] -->| 简单分类 | B(零样本)
    A -->| 格式敏感 | C(小样本)
    A -->| 复杂推理 | D(思维链)
    C --> E{需要解释过程?}
    E -->| 是 | D
    E -->| 否 | C

from typing import Dict, Optional
from enum import Enum

class PromptComponent(Enum):
    INSTRUCTION = 1
    EXAMPLES = 2
    CONTEXT = 3

def build_dynamic_prompt(components: Dict[PromptComponent, str],
    template: str = "{instruction}\n\n{examples}\n\n{context}"
) -> str:
    """
    构建动态提示模板
    :param components: 必须包含 INSTRUCTION 组件
    :raises ValueError: 当缺少必要组件时抛出
    """
    try:
        if PromptComponent.INSTRUCTION not in components:
            raise ValueError("Missing required INSTRUCTION component")

        return template.format(instruction=components.get(PromptComponent.INSTRUCTION, ""),
            examples=components.get(PromptComponent.EXAMPLES, ""),
            context=components.get(PromptComponent.CONTEXT, "")
        )
    except KeyError as e:
        raise ValueError(f"Invalid template placeholder: {e}")

使用 Embedding 计算相似度，仅保留相关历史对话：

import numpy as np
from sentence_transformers import SentenceTransformer

encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def compress_context(
    current_query: str, 
    history: list[str], 
    max_tokens: int = 1024
) -> str:
    """
    基于语义相似度的上下文压缩
    :returns: 不超过 max_tokens 的最相关历史记录
    """
    if not history:
        return ""

    query_embedding = encoder.encode(current_query)
    history_embeddings = encoder.encode(history)

    # 计算余弦相似度
    similarities = np.dot(history_embeddings, query_embedding) \
                 / (np.linalg.norm(history_embeddings, axis=1) * np.linalg.norm(query_embedding))

    # 按相关性排序并拼接
    sorted_history = [h for _, h in sorted(zip(similarities, history), reverse=True)]

    compressed = ""
    for h in sorted_history:
        if len(compressed) + len(h) > max_tokens:
            break
        compressed += f"\n{h}"

    return compressed

graph LR
    A[客户端] --> B{是否有缓存?}
    B -->| 是 | C[返回缓存结果]
    B -->| 否 | D[调用 LLM API]
    D --> E[写入 Redis 缓存]
    E --> F[返回响应]
    G[缓存失效队列] -->|TTL 过期 | D

关键配置参数：
– 缓存键生成规则：MD5(提示模板 + 参数 JSON)
– TTL 设置：根据业务场景设置 30 秒 -24 小时
– 内存淘汰策略：LRU+ 最大内存限制

import backoff
import openai
from tenacity import retry, stop_after_attempt, wait_exponential

class OpenAIClient:
    def __init__(self, max_retry=3, timeout=30):
        self.max_retry = max_retry
        self.timeout = timeout

    @retry(stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    async def completions_with_backoff(self, prompt):
        try:
            response = await openai.ChatCompletion.acreate(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                timeout=self.timeout
            )
            return response.choices[0].message.content
        except openai.error.APIError as e:
            print(f"API error: {e}")
            raise

成本控制方法：
1. 监控每个提示的 token 消耗
2. 对非关键任务使用 gpt-3.5-turbo
3. 设置每月预算警报