提示链中的错误处理
验证中间输出，并从链式流程失败中恢复
提示链中的错误处理 是 CoddyKit 上的免费 AI Prompt Engineering 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Prompt Engineering 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Prompt Engineering 课程共包含 4 节课。
链为何会失败
提示链会引入单提示系统不存在的新故障模式。每个步骤都可能以自己的方式失败，而且故障会叠加——糟糕的步骤 2 输出会污染其后的每个步骤。
常见故障模式：
- 模型返回无法解析的格式错误的 JSON
- 模型误解任务，生成语义错误的输出
- 速率限制或 API 超时导致步骤失败
- 长链超出上下文窗口
- 模型幻觉生成数据，后续步骤将其当作事实
每个步骤后的输出验证
第一道防线是在每个步骤完成后、将输出传给下一步之前立即进行验证。绝不要假设模型返回了您要求的内容。
import json
def validate_json_output(raw_text, required_fields):
'Parse and validate that required fields are present in model output.'
try:
data = json.loads(raw_text.strip())
except json.JSONDecodeError as e:
raise ValueError(f'Invalid JSON: {e}. Raw: {raw_text[:200]}')
missing = [f for f in required_fields if f not in data]
if missing:
raise ValueError(f'Missing required fields: {missing}. Got: {list(data.keys())}')
return data
# Usage after a chain step
raw = '{"sentiment": "positive", "priority": "high"}'
validated = validate_json_output(raw, required_fields=['sentiment', 'priority'])
print('Valid:', validated)针对暂时性故障的重试逻辑
API 故障（速率限制、超时、服务器错误）通常是暂时性的。请针对网络层故障实现指数退避重试逻辑：
import anthropic, time
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
def call_with_retry(prompt, max_retries=3, base_delay=1.0):
last_error = None
for attempt in range(max_retries):
try:
r = client.messages.create(
model='claude-opus-4-5', max_tokens=500,
messages=[{'role': 'user', 'content': prompt}]
)
return r.content[0].text
except anthropic.RateLimitError as e:
wait = base_delay * (2 ** attempt)
print(f'Rate limited. Waiting {wait}s before retry {attempt+1}/{max_retries}...')
time.sleep(wait)
last_error = e
except anthropic.APIError as e:
last_error = e
if attempt < max_retries - 1:
time.sleep(base_delay)
raise RuntimeError(f'All retries exhausted: {last_error}')语义验证
有些故障在结构上有效，但语义上错误——模型返回了有效 JSON，却包含不正确的值。请使用轻量级验证步骤检查语义正确性：
def semantic_validate(data, schema_rules):
'Apply semantic validation rules to parsed output.'
errors = []
for field, rules in schema_rules.items():
value = data.get(field)
if rules.get('required') and value is None:
errors.append(f'{field} is required but missing')
continue
if 'allowed_values' in rules and value not in rules['allowed_values']:
errors.append(f'{field} must be one of {rules["allowed_values"]}, got: {value}')
if 'min_length' in rules and isinstance(value, list) and len(value) < rules['min_length']:
errors.append(f'{field} must have at least {rules["min_length"]} items, got {len(value)}')
if errors:
raise ValueError('Semantic validation failed: ' + '; '.join(errors))
return data
rules = {'sentiment': {'allowed_values': ['positive', 'negative', 'mixed']}, 'issues': {'min_length': 1}}
data = {'sentiment': 'positive', 'issues': ['login bug']}
print(semantic_validate(data, rules))回退提示
在重试后某个步骤仍未通过验证时，回退提示可以生成更简单但可用的输出，而不是让整条链崩溃：
import json
def call_with_fallback(primary_prompt, fallback_prompt, required_fields):
# Try primary prompt
try:
raw = call_with_retry(primary_prompt)
return validate_json_output(raw, required_fields)
except (ValueError, RuntimeError) as e:
print(f'Primary prompt failed: {e}. Trying fallback...')
# Try simpler fallback prompt
try:
raw = call_with_retry(fallback_prompt)
return validate_json_output(raw, required_fields)
except (ValueError, RuntimeError) as e:
print(f'Fallback also failed: {e}. Returning safe default.')
# Return safe default — chain continues with minimal data
return {field: None for field in required_fields}
# Usage
primary = 'Analyze this review. Return JSON with 10 fields: {...}'
fallback = 'Classify this review. Return JSON: {"sentiment": "positive|negative|neutral"}'
result = call_with_fallback(primary, fallback, ['sentiment'])
print(result)熔断器
熔断器可防止失败的链浪费 API 调用。在连续失败 N 次后，它会触发熔断并立即返回错误，而不再进行 API 调用：
class CircuitBreaker:
def __init__(self, failure_threshold=3, recovery_timeout=60):
self.failure_count = 0
self.threshold = failure_threshold
self.state = 'closed' # closed = normal, open = blocking
self.opened_at = None
def call(self, fn, *args, **kwargs):
import time
if self.state == 'open':
elapsed = time.time() - self.opened_at
if elapsed > 60: # recovery_timeout
self.state = 'half-open'
else:
raise RuntimeError('Circuit open — skipping API call')
try:
result = fn(*args, **kwargs)
self.failure_count = 0
self.state = 'closed'
return result
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.threshold:
self.state = 'open'
self.opened_at = time.time()
print(f'Circuit opened after {self.failure_count} failures.')
raise e
cb = CircuitBreaker(failure_threshold=3)
print('Circuit breaker initialized.')为长链设置检查点
对于包含许多步骤或成本高昂步骤的链，请使用检查点保存中间结果。如果后期步骤失败，可以从检查点恢复，而不必从步骤 1 重新开始：
import json, os
CHECKPOINT_DIR = '/tmp/chain_checkpoints'
os.makedirs(CHECKPOINT_DIR, exist_ok=True)
def save_checkpoint(run_id, step_id, data):
path = os.path.join(CHECKPOINT_DIR, f'{run_id}_step{step_id}.json')
with open(path, 'w') as f:
json.dump(data, f)
print(f'Checkpoint saved: step {step_id}')
def load_checkpoint(run_id, step_id):
path = os.path.join(CHECKPOINT_DIR, f'{run_id}_step{step_id}.json')
if os.path.exists(path):
with open(path) as f:
return json.load(f)
return None
def run_with_checkpoints(run_id, input_data):
step1 = load_checkpoint(run_id, 1) or json.loads(call(f'Step 1 processing: {input_data}'))
save_checkpoint(run_id, 1, step1)
step2 = load_checkpoint(run_id, 2) or json.loads(call(f'Step 2 processing: {step1}'))
save_checkpoint(run_id, 2, step2)
return step2
print('Checkpointing system defined.')优雅降级
当链中的某个步骤失败且无法恢复时，优雅降级会使用部分数据继续运行链，而不是让整条链完全崩溃：
def process_with_degradation(tickets):
results = []
for ticket in tickets:
try:
# Full chain: extract -> classify -> respond
extracted = json.loads(call(f'Extract issue from ticket. Return JSON: {{"issue": str}}\n\n{ticket}'))
classified = json.loads(call(f'Classify priority. Return JSON: {{"priority": str}}\n\n{extracted["issue"]}'))
response = call(f'Draft response for {classified["priority"]} priority: {extracted["issue"]}')
results.append({'ticket': ticket, 'response': response, 'degraded': False})
except Exception as e:
print(f'Chain failed for ticket, using fallback: {e}')
# Fallback: simple direct response without classification
simple_response = call(f'Respond to this support ticket:\n{ticket}')
results.append({'ticket': ticket, 'response': simple_response, 'degraded': True})
return results
print('Graceful degradation pipeline defined.')结构化错误日志记录
记录错误时提供足够的上下文，以诊断哪个步骤失败、输入是什么，以及模型返回了什么：
import logging, traceback
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('chain')
def logged_step(step_name, prompt, validator=None):
start = datetime.utcnow()
try:
raw = call_with_retry(prompt)
result = validator(raw) if validator else raw
logger.info(f'[{step_name}] SUCCESS in {(datetime.utcnow()-start).total_seconds():.2f}s')
return result
except Exception as e:
logger.error(f'[{step_name}] FAILED after {(datetime.utcnow()-start).total_seconds():.2f}s')
logger.error(f'[{step_name}] PROMPT: {prompt[:200]}')
logger.error(f'[{step_name}] ERROR: {traceback.format_exc()}')
raise
print('Structured error logging defined.')测试错误场景
通过注入故障，明确测试错误处理。使用模拟对象来模拟 API 错误和格式错误的输出：
from unittest.mock import patch, MagicMock
def test_fallback_on_json_error():
with patch('__main__.call') as mock_call:
# First call returns malformed JSON, fallback returns valid JSON
mock_call.side_effect = [
'This is not JSON at all',
'{"sentiment": "positive"}'
]
result = call_with_fallback(
primary_prompt='Analyze review with 10 fields',
fallback_prompt='Just classify sentiment as JSON',
required_fields=['sentiment']
)
assert result['sentiment'] == 'positive'
print('PASS: fallback activated correctly on JSON parse error')
def test_circuit_breaker_opens():
cb = CircuitBreaker(failure_threshold=2)
for i in range(2):
try:
cb.call(lambda: (_ for _ in ()).throw(RuntimeError('API fail')))
except RuntimeError:
pass
assert cb.state == 'open'
print('PASS: circuit breaker opened after 2 failures')
print('Error handling tests defined.')在生产环境监控链健康状况
在生产环境中，跟踪链的健康指标，以便在用户察觉之前发现性能下降：
- 步骤成功率：每个步骤首次尝试即成功的运行所占百分比
- 回退激活率：回退提示的使用频率
- 降级率：以降级模式完成的链运行所占比例
- 步骤延迟：跟踪每个步骤的 p50/p95 延迟——步骤缓慢表明提示复杂度存在问题
- 验证失败率：比率过高表明提示需要改进
快速检查
提示链中熔断器的作用是什么？
链中的错误处理——要点
健壮的错误处理是区分原型链与生产系统的关键：
- 在将每个步骤的输出传递到下游之前进行验证——绝不要假设模型返回了正确数据
- 使用指数退避重试暂时性 API 故障——速率限制和超时是可恢复的
- 当主要提示未通过语义验证时，使用回退提示生成更简单的输出
- 连续发生故障后，熔断器会停止浪费 API 调用
- 为成本高昂的步骤设置检查点，使长链能够在后期步骤失败后恢复
- 优雅降级使用部分数据维持流水线运行，而不是让其崩溃
- 在生产环境中跟踪步骤成功率、回退率和降级率
常见问题解答
「提示链中的错误处理」课时是免费的吗？
是的 — 「提示链中的错误处理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 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 反馈 — 无需本地设置。