การจัดการข้อผิดพลาดในพรอมป์ตเชน

เหตุใดกระบวนการต่อเนื่องจึงล้มเหลว

การตรวจสอบผลลัพธ์หลังทุกขั้นตอน

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)

ตรรกะการลองใหม่สำหรับความล้มเหลวชั่วคราว

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}')

การตรวจสอบความหมาย

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)

ตัวตัดวงจร

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.')

การบันทึกจุดตรวจสำหรับกระบวนการต่อเนื่องที่ยาว

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.')

การทดสอบสถานการณ์ข้อผิดพลาด

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.')

การติดตามสุขภาพของกระบวนการต่อเนื่องในระบบจริง

ตรวจสอบความเข้าใจ

การจัดการข้อผิดพลาดในกระบวนการต่อเนื่อง — ประเด็นสำคัญ