AI Prompt Engineering · 课时

监控生产环境中的提示词表现

Q: 「监控生产环境中的提示词表现」这节课中我会学到什么？

按提示词版本跟踪延迟、成本、质量分数和失败率。 你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

按提示词版本跟踪延迟、成本、质量分数和失败率。

第 4 / 4 课13 个步骤

监控生产环境中的提示词表现 是 CoddyKit 上的免费 AI Prompt Engineering 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Prompt Engineering 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Prompt Engineering 课程共包含 4 节课。

为什么要监控提示词性能

部署到生产环境的提示词并不意味着工作已经“完成”。模型行为会随着更新而漂移，用户输入会随时间变化，成本也可能意外激增。持续监控可以在回归问题影响用户之前发现它们，并让支出保持可预测。

每个提示词版本的关键指标

在生产环境中，应针对每个提示词版本跟踪以下五项指标：

平均延迟：从请求发出到完整响应返回所需的时间（P50、P95、P99）
每次调用成本：输入令牌数 × 输入价格 + 输出令牌数 × 输出价格
质量评分：自动评估（由 LLM 作为评审器，或使用任务指标）
错误率：接口错误 + 格式错误的输出解析失败
用户满意度：点赞评分、重试率、会话放弃率

指标采集：记录指标

应为每次 LLM 调用添加指标采集，以便将所有关键指标记录到时序存储系统或数据库中，供后续分析和告警使用。

import time
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def instrumented_call(prompt_id, version, messages, model):
    start = time.time()
    error = False
    response = None
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages
        )
    except Exception as e:
        error = True
        raise
    finally:
        latency_ms = (time.time() - start) * 1000
        usage = response.usage if response else None
        record_metric({
            'prompt_id': prompt_id,
            'version': version,
            'latency_ms': latency_ms,
            'input_tokens': usage.prompt_tokens if usage else 0,
            'output_tokens': usage.completion_tokens if usage else 0,
            'error': error,
            'timestamp': time.time()
        })
    return response

计算每次调用成本

每次调用成本 = 输入令牌数 × 输入价格 + 输出令牌数 × 输出价格。保存原始令牌数量后，无论价格如何变化，都可以重新计算成本。

# pricing.py — model pricing table (per 1M tokens)
PRICING = {
    'gpt-4o': {'input': 2.50, 'output': 10.00},
    'gpt-4o-mini': {'input': 0.15, 'output': 0.60},
    'claude-3-5-sonnet-20241022': {'input': 3.00, 'output': 15.00},
}

def cost_usd(model, input_tokens, output_tokens):
    p = PRICING.get(model)
    if not p:
        return None
    cost = (input_tokens / 1_000_000) * p['input'] \
         + (output_tokens / 1_000_000) * p['output']
    return round(cost, 6)

# Example
c = cost_usd('gpt-4o-mini', input_tokens=800, output_tokens=200)
print(f'Cost per call: ${c}')   # Cost per call: $0.000240

# Daily cost estimate
calls_per_day = 50_000
print(f'Daily cost: ${round(c * calls_per_day, 2)}')  # $12.00

自动质量评分

使用 LLM 评审器，在生产规模上自动评估输出质量。可以抽取 5%–10% 的调用进行质量评分，以控制成本。

import random

JUDGE_SAMPLE_RATE = 0.05  # score 5% of calls

JUDGE_PROMPT = '''Rate the quality of this AI response on a scale of 1-5.
1=Very poor, 3=Acceptable, 5=Excellent.
Return only the integer score.

User input: {input}
AI response: {output}'''

def maybe_score_quality(user_input, ai_output, model='gpt-4o-mini'):
    if random.random() > JUDGE_SAMPLE_RATE:
        return None  # not sampled

    judge_messages = [{'role': 'user', 'content':
        JUDGE_PROMPT.format(input=user_input, output=ai_output)}]
    response = client.chat.completions.create(
        model=model, messages=judge_messages, max_tokens=5
    )
    try:
        score = int(response.choices[0].message.content.strip())
        return max(1, min(5, score))  # clamp to 1-5
    except ValueError:
        return None

将指标存储在时序数据库中

时序数据库（InfluxDB、TimescaleDB，甚至是带时间戳索引的简单 PostgreSQL 表）可以高效存储每次调用的指标，并支持用于仪表板的聚合查询。

-- TimescaleDB / PostgreSQL schema for prompt metrics
CREATE TABLE prompt_metrics (
  ts              TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  prompt_id       VARCHAR(100),
  version         VARCHAR(20),
  model           VARCHAR(50),
  latency_ms      FLOAT,
  input_tokens    INT,
  output_tokens   INT,
  cost_usd        FLOAT,
  quality_score   FLOAT,   -- NULL if not sampled
  error           BOOLEAN DEFAULT FALSE
);

-- Convert to hypertable (TimescaleDB)
SELECT create_hypertable('prompt_metrics', 'ts');

-- Query: hourly P95 latency by version
SELECT
  date_trunc('hour', ts) AS hour,
  version,
  PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) AS p95_latency
FROM prompt_metrics
WHERE prompt_id = 'summarize-article'
  AND ts > NOW() - INTERVAL '24 hours'
GROUP BY 1, 2
ORDER BY 1;

构建指标仪表板

仪表板可以实时展示五项关键指标。请使用 Grafana（以 TimescaleDB 作为数据源）或自定义网页仪表板。关键面板包括：

按版本查看延迟 P50/P95/P99 随时间的变化
每次调用成本趋势（每日/每周）
错误率百分比
质量评分滚动平均值
用户满意度评分

# Simple dashboard query aggregation in Python
def get_dashboard_stats(conn, prompt_id, hours=24):
    with conn.cursor() as cur:
        cur.execute('''
            SELECT
              version,
              COUNT(*) AS calls,
              AVG(latency_ms) AS avg_latency,
              PERCENTILE_CONT(0.95)
                WITHIN GROUP (ORDER BY latency_ms) AS p95_latency,
              AVG(cost_usd) AS avg_cost,
              AVG(quality_score) FILTER (WHERE quality_score IS NOT NULL) AS avg_quality,
              SUM(error::int)::float / COUNT(*) AS error_rate
            FROM prompt_metrics
            WHERE prompt_id = %s
              AND ts > NOW() - INTERVAL %s
            GROUP BY version
            ORDER BY MAX(ts) DESC
        ''', (prompt_id, f'{hours} hours'))
        return cur.fetchall()

提示词回归的异常检测

手动查看仪表板会漏掉缓慢出现的回归问题。自动异常检测会将滚动窗口与历史基线进行比较，并在偏差超过阈值时触发告警。

import statistics

def detect_anomaly(recent_values, baseline_values, threshold_std=2.0):
    if len(baseline_values) < 10:
        return False  # not enough data
    baseline_mean = statistics.mean(baseline_values)
    baseline_std = statistics.stdev(baseline_values)
    recent_mean = statistics.mean(recent_values)

    if baseline_std == 0:
        return False
    z_score = abs(recent_mean - baseline_mean) / baseline_std
    return z_score > threshold_std

# Example: detect quality regression
baseline_quality = [4.1, 4.0, 4.2, 4.1, 4.0, 3.9, 4.1, 4.2, 4.0, 4.1]
recent_quality   = [3.2, 3.1, 3.4, 3.0]

if detect_anomaly(recent_quality, baseline_quality):
    print('ALERT: Quality score anomaly detected!')
    # fire_pagerduty_alert(...)
else:
    print('Quality within normal range')

告警规则与阈值

应将告警规则定义为代码，以便进行版本控制和评审。常见的提示词监控告警包括：

错误率 > 5%，持续 5 分钟
P95 延迟 > 10 秒，持续 3 分钟
每日成本 > 每周平均值的 2 倍（成本激增）
质量评分较基线下降 > 20%

# alerts.yaml (Grafana alerting or custom)
alerts:
  - name: HighErrorRate
    condition: error_rate > 0.05
    for: 5m
    severity: critical
    message: 'Prompt {prompt_id} error rate {error_rate:.1%} exceeds 5%'

  - name: HighLatencyP95
    condition: p95_latency_ms > 10000
    for: 3m
    severity: warning
    message: 'P95 latency {p95_latency_ms}ms exceeds 10s for {prompt_id}'

  - name: CostSpike
    condition: daily_cost_usd > weekly_avg_daily_cost * 2
    for: 1h
    severity: warning
    message: 'Daily cost ${daily_cost_usd:.2f} is 2x the weekly average'

  - name: QualityRegression
    condition: rolling_quality_avg < baseline_quality_avg * 0.80
    for: 15m
    severity: critical
    message: 'Quality dropped {pct_drop:.0%} for prompt {prompt_id}'

用户满意度信号

自动化指标无法捕捉所有信息。请收集明确的和隐含的用户信号，以补充质量评分：

点赞评分：用户明确对人工智能响应进行 👍/👎 评价
重试率：用户立即重新提交相同查询（隐含的不满意）
复制/使用率：用户复制人工智能输出（隐含的满意）
修改率：用户在使用人工智能输出前对其进行编辑

# feedback_collector.py
def record_feedback(prompt_id, version, call_id, signal_type, value):
    '''
    signal_type: 'thumbs' | 'retry' | 'copy' | 'edit'
    value: for thumbs: 1 (up) or -1 (down); others: 1 (occurred)
    '''
    db.execute(
        'INSERT INTO prompt_feedback '
        '(prompt_id, version, call_id, signal_type, value, ts) '
        'VALUES (%s, %s, %s, %s, %s, NOW())',
        (prompt_id, version, call_id, signal_type, value)
    )

# Aggregate satisfaction score
def satisfaction_score(prompt_id, version):
    rows = db.fetch(
        'SELECT signal_type, AVG(value) as avg_val FROM prompt_feedback '
        'WHERE prompt_id=%s AND version=%s GROUP BY signal_type',
        (prompt_id, version)
    )
    return {r['signal_type']: round(r['avg_val'], 3) for r in rows}

版本对比报告

评估是否要推广或回滚金丝雀版本时，应并排比较新版本与基线之间的所有指标，并生成对比报告。该报告用于推动推广决策。

def version_comparison_report(prompt_id, version_a, version_b, hours=48):
    stats_a = get_dashboard_stats_for_version(prompt_id, version_a, hours)
    stats_b = get_dashboard_stats_for_version(prompt_id, version_b, hours)

    print(f'=== Comparison: {version_a} vs {version_b} ===')
    metrics = ['avg_latency', 'p95_latency', 'avg_cost', 'avg_quality', 'error_rate']
    for m in metrics:
        va = stats_a.get(m, 0)
        vb = stats_b.get(m, 0)
        delta = vb - va
        pct = (delta / va * 100) if va else 0
        direction = '+' if delta > 0 else ''
        print(f'{m:20s}: {va:.4f} -> {vb:.4f}  ({direction}{pct:.1f}%)')

# Example output:
# avg_latency         : 1234.5000 -> 1189.2000  (-3.7%)
# p95_latency         : 3210.0000 -> 3050.0000  (-5.0%)
# avg_cost            : 0.000240 -> 0.000255  (+6.2%)
# avg_quality         : 4.1000 -> 4.3000  (+4.9%)
# error_rate          : 0.0120 -> 0.0080  (-33.3%)

快速检查

如果您希望自动检测质量回归，而不是手动检查仪表板，哪种方法最适合实现这一目标？

在交互式课程中回答

监控总结

生产环境中的提示词监控需要针对每个版本跟踪五个维度：

延迟（P50/P95/P99）——用户体验
每次调用成本——财务状况
质量评分——自动进行 LLM 评审抽样
错误率——可靠性
用户满意度——点赞评分、重试和复制信号

请将指标存储在时序数据库中，在仪表板中进行可视化，并在超过阈值时触发告警。版本对比报告可以为推广和回滚决策提供依据。

免费开始

用 AI 导师学习 AI Prompt Engineering — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 53
课程: 199

免费开始学习查看完整课程

常见问题解答

「监控生产环境中的提示词表现」课时是免费的吗？

是的 — 「监控生产环境中的提示词表现」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Prompt Engineering 课程的其余内容，请升级到 CoddyKit PRO。 AI Prompt Engineering 课程共包含 4 节课。

「监控生产环境中的提示词表现」这节课中我会学到什么？

按提示词版本跟踪延迟、成本、质量分数和失败率。你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Prompt Engineering 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Prompt Engineering 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「监控生产环境中的提示词表现」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Prompt Engineering 课中编写并运行代码吗？

能。每节 AI Prompt Engineering 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Prompt Engineering