日志记录与文档策略

提示词日志记录为何重要

最小可用日志条目

import time, json
from datetime import datetime, timezone

def logged_call(prompt_id, system_prompt, user_message, model='gpt-4o', temperature=0.7):
    start = time.time()
    resp = client.chat.completions.create(
        model=model,
        messages=[
            {'role': 'system', 'content': system_prompt},
            {'role': 'user', 'content': user_message}
        ],
        temperature=temperature
    )
    latency = int((time.time() - start) * 1000)
    output = resp.choices[0].message.content
    log_entry = {
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'prompt_id': prompt_id,
        'model': model,
        'temperature': temperature,
        'input': user_message,
        'output': output,
        'latency_ms': latency,
        'input_tokens': resp.usage.prompt_tokens,
        'output_tokens': resp.usage.completion_tokens
    }
    append_log(log_entry)
    return output

结构化日志格式

import json

LOG_FILE = 'prompt_logs.jsonl'

def append_log(entry):
    with open(LOG_FILE, 'a') as f:
        f.write(json.dumps(entry) + '\n')

def read_logs():
    with open(LOG_FILE) as f:
        return [json.loads(line) for line in f if line.strip()]

# Query: all entries for prompt_id 'summarize_v3'
logs = read_logs()
summarize_logs = [e for e in logs if e['prompt_id'] == 'summarize_v3']
print(f'Total calls to summarize_v3: {len(summarize_logs)}')

提示词版本控制

PROMPTS = {
    'summarize': {
        'v1': 'Summarize the following text.',
        'v2': 'Summarize the following text in 3 sentences.',
        'v3': 'Summarize the following text in exactly 3 sentences. '
              'Start each sentence on a new line. No bullet points.'
    }
}

CURRENT_VERSIONS = {'summarize': 'v3'}

def get_prompt(prompt_id):
    version = CURRENT_VERSIONS[prompt_id]
    return version, PROMPTS[prompt_id][version]

version, prompt = get_prompt('summarize')
log_entry['prompt_version'] = version

处理日志中的 PII

import hashlib, re

def redact_pii(text):
    # Redact email addresses
    text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]', text)
    # Redact phone numbers (US format)
    text = re.sub(r'\b\d{3}[-.]\d{3}[-.]\d{4}\b', '[PHONE]', text)
    return text

def hash_input(text):
    return hashlib.sha256(text.encode()).hexdigest()[:16]

log_entry['input'] = redact_pii(user_message)
log_entry['input_hash'] = hash_input(user_message)

跟踪延迟与成本

def compute_cost(entry, price_per_1m_input=5.0, price_per_1m_output=15.0):
    input_cost = entry['input_tokens'] / 1_000_000 * price_per_1m_input
    output_cost = entry['output_tokens'] / 1_000_000 * price_per_1m_output
    return input_cost + output_cost

def prompt_stats(prompt_id, version):
    logs = [e for e in read_logs()
            if e['prompt_id'] == prompt_id and e.get('prompt_version') == version]
    if not logs:
        return
    avg_latency = sum(e['latency_ms'] for e in logs) / len(logs)
    total_cost = sum(compute_cost(e) for e in logs)
    print(f'{prompt_id} {version}: {len(logs)} calls, avg {avg_latency:.0f}ms, total ${total_cost:.4f}')

记录输出评估结果

def evaluated_call(prompt_id, system_prompt, user_message, evaluator_fn):
    output = logged_call(prompt_id, system_prompt, user_message)
    score = evaluator_fn(user_message, output)
    # Update the last log entry with the evaluation score
    logs = read_logs()
    last = logs[-1]
    last['eval_score'] = score
    last['eval_pass'] = score >= 0.8
    # Rewrite the last line
    with open(LOG_FILE, 'a') as f:
        # In practice, use a DB or separate eval log
        pass
    return output, score

提示词文档

PROMPT_DOCS = {
    'summarize': {
        'purpose': 'Summarize a single text passage into 3 sentences.',
        'variables': {'text': 'The passage to summarize (max 2000 tokens)'},
        'known_limitations': [
            'Fails to preserve numbers accurately for texts with many statistics',
            'May not summarize correctly for non-English text'
        ],
        'versions': {
            'v1': 'Initial version — vague length instruction',
            'v2': 'Added 3-sentence limit',
            'v3': 'Added line-break and no-bullet formatting fix'
        },
        'test_suite': 'tests/test_summarize.py'
    }
}

使用集中式日志服务

# Example: writing to Supabase
from supabase import create_client

supabase = create_client('https://xxx.supabase.co', 'your-anon-key')

def log_to_supabase(entry):
    supabase.table('prompt_logs').insert(entry).execute()

# Now query with SQL:
# SELECT prompt_id, prompt_version, AVG(latency_ms), COUNT(*)
# FROM prompt_logs
# WHERE timestamp > NOW() - INTERVAL '7 days'
# GROUP BY prompt_id, prompt_version
# ORDER BY COUNT(*) DESC;

失败率激增告警

from collections import deque
from datetime import datetime, timezone, timedelta

recent_results = deque(maxlen=100)  # sliding window

def track_and_alert(prompt_id, success, alert_fn, threshold=0.10):
    recent_results.append({'success': success, 'time': datetime.now(timezone.utc)})
    window = [
        r for r in recent_results
        if r['time'] > datetime.now(timezone.utc) - timedelta(minutes=5)
    ]
    if not window:
        return
    fail_rate = sum(1 for r in window if not r['success']) / len(window)
    if fail_rate > threshold:
        alert_fn(f'ALERT: {prompt_id} failure rate {fail_rate:.0%} in last 5 min')

保留与归档

知识检查

回顾：日志记录与文档