AI Prompt Engineering · 课时

提示词流程的监控与告警

Q: 「提示词流程的监控与告警」这节课中我会学到什么？

为生产环境中的提示词提供仪表板、异常检测和当班告警。 你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

为生产环境中的提示词提供仪表板、异常检测和当班告警。

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

生产环境中的提示词流水线需要监控

生产环境中的提示词流水线属于基础设施——与其他服务一样，需要仪表板、告警和操作手册。没有监控，成本激增、质量下降和延迟暴增都会一直不被发现，直到用户投诉或账单到达。

核心指标：延迟百分位数

跟踪 P50、P95 和 P99 延迟。平均值会掩盖尾部行为——P99 为 30 秒意味着即使 P50 为 2 秒，仍有 1% 的用户需要等待半分钟。LLM 延迟本身具有波动性，因为它会随输出长度变化。

import time
import statistics
from collections import deque

class LatencyTracker:
    def __init__(self, window_size=1000):
        self.samples = deque(maxlen=window_size)

    def record(self, latency_ms):
        self.samples.append(latency_ms)

    def percentile(self, p):
        if not self.samples:
            return None
        sorted_samples = sorted(self.samples)
        idx = int(len(sorted_samples) * p / 100)
        return sorted_samples[min(idx, len(sorted_samples) - 1)]

    def report(self):
        if not self.samples:
            return {}
        return {
            'count': len(self.samples),
            'p50_ms': self.percentile(50),
            'p95_ms': self.percentile(95),
            'p99_ms': self.percentile(99),
            'max_ms': max(self.samples)
        }

tracker = LatencyTracker()
for ms in [1200, 1100, 1300, 1150, 8500, 1200, 1250, 15000, 1100, 1300]:
    tracker.record(ms)
print(tracker.report())

每日成本仪表板指标

将每日接口成本作为仪表板的主要指标进行跟踪。根据实时令牌使用量计算成本，并与滚动周平均值进行比较，以便及早发现成本激增。

from datetime import datetime, timedelta
from collections import defaultdict

class CostTracker:
    def __init__(self):
        self.daily_costs = defaultdict(float)  # date: total_cost_usd

    def record_call(self, model, input_tokens, output_tokens):
        pricing = {
            'gpt-4o-mini': (0.15, 0.60),
            'gpt-4o': (2.50, 10.00),
            'claude-opus-4-5': (15.00, 75.00),  # per 1M tokens
            'claude-haiku-4-5': (0.25, 1.25)
        }
        if model not in pricing:
            return
        input_price, output_price = pricing[model]
        cost = (input_tokens / 1_000_000 * input_price +
                output_tokens / 1_000_000 * output_price)
        today = datetime.utcnow().date().isoformat()
        self.daily_costs[today] += cost

    def today_cost(self):
        today = datetime.utcnow().date().isoformat()
        return round(self.daily_costs[today], 4)

    def weekly_avg_daily_cost(self):
        dates = sorted(self.daily_costs.keys())[-7:]
        if not dates:
            return 0
        return round(sum(self.daily_costs[d] for d in dates) / len(dates), 4)

cost_tracker = CostTracker()
cost_tracker.record_call('gpt-4o-mini', 800, 200)
print('Today cost:', cost_tracker.today_cost())

错误率监控

将错误率作为总请求数的百分比进行跟踪。错误包括接口失败、超时、解析失败的格式错误输出以及模型拒答。应区分错误类型，以便设置可采取行动的告警。

from collections import Counter

class ErrorRateTracker:
    ERROR_TYPES = [
        'api_error', 'timeout', 'rate_limit',
        'parse_failure', 'model_refusal', 'context_length_exceeded'
    ]

    def __init__(self, window_size=1000):
        self.total = 0
        self.errors = Counter()
        self.recent = deque(maxlen=window_size)  # True=error, False=success

    def record(self, success, error_type=None):
        self.total += 1
        self.recent.append(not success)
        if not success and error_type:
            self.errors[error_type] += 1

    def error_rate(self):
        if not self.recent:
            return 0.0
        return sum(self.recent) / len(self.recent)

    def report(self):
        return {
            'error_rate': round(self.error_rate(), 4),
            'total_requests': self.total,
            'error_breakdown': dict(self.errors.most_common())
        }

err_tracker = ErrorRateTracker()
for i in range(100):
    if i % 20 == 0:
        err_tracker.record(False, 'timeout')
    else:
        err_tracker.record(True)
print(err_tracker.report())

质量评分趋势跟踪

将质量评分作为滚动时间序列进行跟踪，以便观察趋势。持续数日的质量缓慢下降比骤然下降更难察觉，但同样可能严重损害用户信任。

from datetime import datetime
import statistics

class QualityTrendTracker:
    def __init__(self, window_minutes=60):
        self.window_seconds = window_minutes * 60
        self.samples = []  # (timestamp, score)

    def record(self, score):
        now = time.time()
        self.samples.append((now, score))
        # Purge old samples outside window
        cutoff = now - self.window_seconds
        self.samples = [(t, s) for t, s in self.samples if t >= cutoff]

    def rolling_avg(self):
        if not self.samples:
            return None
        return round(statistics.mean(s for _, s in self.samples), 3)

    def trend(self):
        if len(self.samples) < 10:
            return 'insufficient_data'
        mid = len(self.samples) // 2
        first_half_avg = statistics.mean(s for _, s in self.samples[:mid])
        second_half_avg = statistics.mean(s for _, s in self.samples[mid:])
        delta = second_half_avg - first_half_avg
        if delta > 0.1:
            return 'improving'
        elif delta < -0.1:
            return 'declining'
        return 'stable'

qt = QualityTrendTracker(window_minutes=60)
for score in [4.2, 4.1, 4.0, 3.9, 3.8, 3.7, 3.6, 3.5, 3.4, 3.3]:
    qt.record(score)
print('Rolling avg:', qt.rolling_avg(), '| Trend:', qt.trend())

仪表板面板设计

设计良好的监控仪表板会将指标分组到逻辑清晰的区域中。请定义每个提示词流水线仪表板都需要的四个核心面板。

DASHBOARD_PANELS = {
    'Panel 1: Availability': [
        'Error rate (%) — last 1h, 24h, 7d',
        'Error type breakdown (timeout vs API vs parse)',
        'P99 latency (alert if > 10s)',
        'Success rate by prompt_id and version'
    ],
    'Panel 2: Performance': [
        'P50 / P95 / P99 latency (time series)',
        'Latency by model and prompt version',
        'Time to first token (streaming)',
        'Latency heatmap by hour of day'
    ],
    'Panel 3: Cost': [
        'Daily cost USD (actual vs budget)',
        'Cost per request by model',
        'Cost trend (7-day rolling)',
        'Top 10 most expensive prompt_ids'
    ],
    'Panel 4: Quality': [
        'Average quality score (rolling 1h)',
        'Quality trend by prompt version',
        'User satisfaction (thumbs, retry rate)',
        'Low-quality alert rate'
    ]
}

for panel, metrics in DASHBOARD_PANELS.items():
    print(f'\n{panel}:')
    for m in metrics:
        print(f'  - {m}')

告警规则实现

当指标超过阈值时，告警就会触发。请将告警实现为简单的轮询检查，按计划运行，并向 PagerDuty、Slack 或电子邮件发送通知。

ALERT_RULES = [
    {
        'name': 'HighErrorRate',
        'condition': lambda m: m['error_rate'] > 0.05,
        'severity': 'CRITICAL',
        'message': 'Error rate {error_rate:.1%} exceeds 5% threshold',
        'for_minutes': 5
    },
    {
        'name': 'HighLatencyP95',
        'condition': lambda m: m.get('p95_latency_ms', 0) > 10000,
        'severity': 'WARNING',
        'message': 'P95 latency {p95_latency_ms}ms exceeds 10s threshold',
        'for_minutes': 3
    },
    {
        'name': 'CostSpike',
        'condition': lambda m: m.get('today_cost', 0) > m.get('weekly_avg', 1) * 2,
        'severity': 'WARNING',
        'message': 'Daily cost ${today_cost:.2f} is 2x weekly average',
        'for_minutes': 60
    },
    {
        'name': 'QualityRegression',
        'condition': lambda m: m.get('quality_avg', 5) < 3.5,
        'severity': 'CRITICAL',
        'message': 'Quality score {quality_avg:.2f} below 3.5 threshold',
        'for_minutes': 15
    }
]

def check_alerts(metrics):
    fired = []
    for rule in ALERT_RULES:
        if rule['condition'](metrics):
            msg = rule['message'].format(**metrics)
            fired.append({'name': rule['name'], 'severity': rule['severity'],
                           'message': msg})
    return fired

通知分发

告警通知应按严重程度进行路由：CRITICAL 告警立即通知值班人员；WARNING 告警发布到 Slack；INFO 告警写入日志文件。对于未确认的关键告警，应使用升级计时器。

import requests

SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
PAGERDUTY_API_KEY = 'YOUR_PD_KEY'

def send_slack_alert(message, severity='WARNING'):
    emoji = ':rotating_light:' if severity == 'CRITICAL' else ':warning:'
    payload = {'text': f'{emoji} *{severity}*: {message}'}
    try:
        requests.post(SLACK_WEBHOOK, json=payload, timeout=5)
        print(f'Slack alert sent: {message[:60]}')
    except Exception as e:
        print(f'Slack notification failed: {e}')

def send_pagerduty_alert(summary, severity='critical'):
    payload = {
        'routing_key': PAGERDUTY_API_KEY,
        'event_action': 'trigger',
        'payload': {
            'summary': summary,
            'severity': severity,
            'source': 'prompt-pipeline-monitor'
        }
    }
    try:
        response = requests.post(
            'https://events.pagerduty.com/v2/enqueue',
            json=payload, timeout=10
        )
        print(f'PagerDuty alert: {response.status_code}')
    except Exception as e:
        print(f'PagerDuty notification failed: {e}')

def dispatch_alert(alert):
    if alert['severity'] == 'CRITICAL':
        send_pagerduty_alert(alert['message'])
        send_slack_alert(alert['message'], 'CRITICAL')
    else:
        send_slack_alert(alert['message'], 'WARNING')

值班操作手册结构

每条告警都应有对应的操作手册，明确告诉值班工程师具体该做什么。编写良好的操作手册可以将 MTTR（平均解决时间）从数小时缩短到数分钟。

# Runbook template for HighErrorRate alert
HIGH_ERROR_RATE_RUNBOOK = '''
## Alert: HighErrorRate
### Trigger: Error rate > 5% for > 5 minutes
### Severity: CRITICAL

## Immediate Actions (< 5 minutes)
1. Check error type breakdown in dashboard: Panel 1 > Error type breakdown
   - timeout errors -> see Timeout Runbook
   - api_error -> check LLM provider status page
   - parse_failure -> check if model output format changed

2. Check if this is related to a recent deployment:
   python manage.py prompt list-recent-activations --last-hours 2

3. If error rate > 20%, trigger emergency rollback:
   python manage.py prompt activate --prompt-id <id> --version <last-stable>

## Investigation (< 30 minutes)
4. Sample failed requests from log:
   grep error_rate /var/log/prompt-pipeline.log | tail -100

5. Check model provider status:
   - OpenAI: https://status.openai.com
   - Anthropic: https://status.anthropic.com

## Resolution
6. If provider outage: activate fallback model routing
7. If prompt change: rollback to previous version
8. If code change: rollback deployment
9. Document in post-mortem after resolution
'''

print(HIGH_ERROR_RATE_RUNBOOK[:400], '...')

提示词流水线的结构化日志记录

结构化日志（每行一个 JSON 对象）可以在 Datadog、Splunk 或 CloudWatch 等日志管理系统中实现强大的筛选和聚合。每次 LLM 调用都应生成一条结构化日志记录。

import json
import time
from datetime import datetime

def log_llm_call(request_id, prompt_id, version, model, messages,
                  response_text, latency_ms, input_tokens,
                  output_tokens, error=None):
    log_entry = {
        'ts': datetime.utcnow().isoformat() + 'Z',
        'level': 'ERROR' if error else 'INFO',
        'service': 'prompt-pipeline',
        'request_id': request_id,
        'prompt_id': prompt_id,
        'version': version,
        'model': model,
        'latency_ms': round(latency_ms),
        'input_tokens': input_tokens,
        'output_tokens': output_tokens,
        'error': str(error) if error else None,
        'response_preview': response_text[:100] if response_text else None
    }
    print(json.dumps(log_entry))
    # In production: ship to log aggregator
    # logger.info(json.dumps(log_entry))

# Example log output:
# {"ts":"2024-08-15T10:00:01Z","level":"INFO",
#  "prompt_id":"summarize-article","version":"1.2.0",
#  "model":"gpt-4o-mini","latency_ms":1234,
#  "input_tokens":800,"output_tokens":150,...}
log_llm_call('req-001', 'summarize-article', '1.2.0', 'gpt-4o-mini',
             [], 'Summary text...', 1234, 800, 150)

监控系统架构

将所有监控组件组装成一个与提示词流水线并行运行的 cohesive 系统。简单的轮询循环即可处理告警评估和分发。

import time

class PromptPipelineMonitor:
    def __init__(self):
        self.latency = LatencyTracker()
        self.errors = ErrorRateTracker()
        self.quality = QualityTrendTracker()
        self.cost = CostTracker()

    def record(self, model, latency_ms, input_tokens, output_tokens,
                quality_score=None, error=None, error_type=None):
        self.latency.record(latency_ms)
        self.errors.record(error is None, error_type)
        self.cost.record_call(model, input_tokens, output_tokens)
        if quality_score:
            self.quality.record(quality_score)

    def current_metrics(self):
        lat = self.latency.report()
        err = self.errors.report()
        return {
            **lat, **err,
            'quality_avg': self.quality.rolling_avg() or 5.0,
            'quality_trend': self.quality.trend(),
            'today_cost': self.cost.today_cost(),
            'weekly_avg': self.cost.weekly_avg_daily_cost()
        }

    def run_alert_check(self):
        metrics = self.current_metrics()
        alerts = check_alerts(metrics)
        for alert in alerts:
            dispatch_alert(alert)
        return alerts

monitor = PromptPipelineMonitor()
print('Monitor initialized. Call monitor.record() on each LLM call.')

快速检查

您的提示词流水线 P50 延迟为 1.5 秒，但 P99 延迟为 28 秒。这说明了生产环境中的什么行为？

监控与告警总结

生产环境中的提示词流水线监控需要五类测量指标以及相应的告警规则：

延迟：跟踪 P50/P95/P99——当 P95 > 10s 时告警
成本：每日成本与周平均值的比较——成本激增至 2 倍时告警
错误率：总错误率及错误类型分布——超过 5% 时告警
质量评分：滚动平均趋势——低于 3.5/5 时告警
告警：CRITICAL → PagerDuty + Slack；WARNING → 仅 Slack
操作手册：为每种告警类型提供逐步解决指南
仪表板：通过四个面板覆盖可用性、性能、成本和质量

常见问题解答

「提示词流程的监控与告警」课时是免费的吗？

是的 — 「提示词流程的监控与告警」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 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