Production Pipelines — בניית צינורות עבודה לפרודקשן

# Python:
python3 --version   # צריך 3.10 ומעלה
python3 -c "import claude_agent_sdk; print('SDK OK')"
echo $ANTHROPIC_API_KEY  # צריך להתחיל ב-sk-ant-

# TypeScript:
node --version   # צריך 18 ומעלה
npx tsx -e "import { Agent } from '@anthropic-ai/claude-agent-sdk'; console.log('SDK OK')"
echo $ANTHROPIC_API_KEY  # צריך להתחיל ב-sk-ant-

מה זה Pipeline ולמה הוא שונה

Pipeline (צינור עבודה) הוא שרשרת עיבוד רב-שלבית שבה הפלט של שלב אחד הופך לקלט של הבא. זה כמו פס ייצור במפעל: כל תחנה עושה את שלה, מעבירה לתחנה הבאה, ובסוף יוצא מוצר מוגמר.

ההבדל מ-script פשוט (פרק 1) ומצוות (פרק 5):

• Script = קריאה בודדת ל-Claude, תוצאה בודדת. אין שלבים, אין recovery.
• Pipeline = כמה קריאות ברצף, כל אחת בונה על הקודמת. שלבים מוגדרים, error handling, logging.
• Team = כמה קריאות במקביל, סינתזה בסוף. מתאים כשתת-משימות עצמאיות.

ההבדל הקריטי מ-ad-hoc automation: pipelines מתוכננים מראש עם שלבים מוגדרים, error handling, logging, ותוכננו לרוץ באופן אמין וחוזר — שוב ושוב, באותה איכות.

מתי צריך pipeline ולא script:

• התהליך כולל 3+ שלבים שכל אחד בונה על הקודם
• צריך recovery — אם שלב 5 מ-7 נכשל, לא רוצים להתחיל מ-1
• צריך monitoring — לדעת כמה עלה, כמה זמן לקח, מה הצליח ומה נכשל
• התהליך חוזר על עצמו — ירוץ שוב מחר, בשבוע הבא, בכל PR

תכנון Pipeline רב-שלבי

התחילו מהפלט. לפני שחושבים על שלבים — הגדירו בדיוק מה ה-pipeline מייצר. פיצ'ר מפורס? מאמר מפורסם? דוח בדיקה? מהפלט עובדים אחורה: מה השלב האחרון? מה הוא צריך כקלט? מי מייצר את הקלט הזה?

ה-Stage Specification Template: לכל stage צריך להגדיר 6 שדות:

כמה stages? 5-8 stages זה ה-sweet spot. פחות מ-3? script מספיק. יותר מ-10? חלקו ל-sub-pipelines.

הדפוס Orchestrator

ה-Orchestrator (מנצח) הוא הקוד שלכם שמנהל את ה-pipeline — הוא לא Claude. הוא קוד Python או TypeScript שמריץ stages בסדר, מעביר data ביניהם, מטפל בשגיאות, ומנהל logging.

עיקרון מפתח: Separation of Concerns.

• Orchestrator — flow control, error handling, logging, checkpointing
• Stages — agent configuration, prompt engineering, output parsing
• Agents — tool use, reasoning, execution (Claude Code)

ה-Orchestrator לעולם לא קורא ל-Claude ישירות — הוא מאציל ל-stage objects שמכמסים את הלוגיקה.

Python Orchestrator — skeleton:

import json
from claude_agent_sdk import Agent

class Stage:
    def __init__(self, name, model, prompt_file, tools, max_turns=10):
        self.name = name
        self.model = model
        self.prompt_file = prompt_file
        self.tools = tools
        self.max_turns = max_turns

    def execute(self, input_data):
        prompt = open(self.prompt_file).read()
        prompt = prompt.replace("{INPUT}", json.dumps(input_data))
        agent = Agent(model=self.model, max_turns=self.max_turns)
        result = agent.run(prompt)
        return json.loads(result.text)

class Orchestrator:
    def __init__(self, stages, checkpoint_dir="./checkpoints"):
        self.stages = stages
        self.checkpoint_dir = checkpoint_dir

    def run(self, initial_input):
        data = initial_input
        for i, stage in enumerate(self.stages):
            print(f"[Stage {i+1}/{len(self.stages)}] {stage.name}...")
            data = stage.execute(data)
            self._save_checkpoint(i, stage.name, data)
            print(f"  Done: {stage.name}")
        return data

    def _save_checkpoint(self, index, name, data):
        path = f"{self.checkpoint_dir}/stage_{index}_{name}.json"
        with open(path, 'w') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

TypeScript Orchestrator — skeleton:

import { Agent } from '@anthropic-ai/claude-agent-sdk';
import fs from 'fs/promises';

interface StageConfig {
  name: string;
  model: string;
  promptFile: string;
  maxTurns?: number;
}

async function runPipeline(stages: StageConfig[], initialInput: any) {
  let data = initialInput;
  for (let i = 0; i < stages.length; i++) {
    const stage = stages[i];
    console.log(`[Stage ${i+1}/${stages.length}] ${stage.name}...`);

    const prompt = await fs.readFile(stage.promptFile, 'utf-8');
    const agent = new Agent({ model: stage.model, maxTurns: stage.maxTurns ?? 10 });
    const result = await agent.run(prompt.replace('{INPUT}', JSON.stringify(data)));

    data = JSON.parse(result.text);
    await fs.writeFile(
      `./checkpoints/stage_${i}_${stage.name}.json`,
      JSON.stringify(data, null, 2)
    );
    console.log(`  Done: ${stage.name}`);
  }
  return data;
}

תקשורת בין שלבים וניהול Data Flow

שלבים ב-pipeline צריכים להעביר data ביניהם. שלוש גישות:

1. File-based (דיסק): Stage A כותב output לקובץ, Stage B קורא. פשוט, debuggable, אבל איטי ל-data גדול.

2. In-memory (זיכרון): Stage A מחזיר dict/object, ה-orchestrator מעביר ל-Stage B. מהיר, אבל אובד ב-crash.

3. Hybrid (שניהם): data עובר in-memory, וכל stage גם כותב checkpoint לדיסק. זו הגישה המומלצת.

ה-Checkpoint Pattern: אחרי כל stage מוצלח, שמרו state לדיסק. Crash ב-stage 6 = resume מ-stage 6, לא מ-1.

Resume Logic — דוגמה:

def run_with_resume(self, initial_input):
    last_checkpoint = self._find_last_checkpoint()
    if last_checkpoint:
        start_index = last_checkpoint['stage_index'] + 1
        data = last_checkpoint['data']
        print(f"Resuming from stage {start_index + 1}...")
    else:
        start_index = 0
        data = initial_input

    for i in range(start_index, len(self.stages)):
        stage = self.stages[i]
        data = stage.execute(data)
        self._save_checkpoint(i, stage.name, data)
    return data

Data Validation at Boundaries: לפני שמעבירים output מ-stage A ל-stage B, ולדו שהוא בפורמט הנכון. schema mismatch ב-stage 5 קשה לדבג — עדיף לתפוס מיד אחרי stage A.

טיפול בשגיאות ו-Recovery

ב-pipeline, שגיאות הן עניין של "מתי", לא "האם". API timeouts, rate limits, agent confusion, malformed output — כל אלה יקרו.

5 אסטרטגיות Recovery:

1. Retry Same: שגיאות חולפות (API timeout, rate limit 429) — נסו שוב עם exponential backoff. 3 ניסיונות מקסימום.
2. Retry Different: הגישה נכשלה — prompt שונה או model שונה. אם Sonnet נכשל, נסו Opus.
3. Skip: שלבים אופציונליים (polish, formatting) — pipeline ממשיך עם תוצאה בת-שימוש.
4. Abort: שלבים קריטיים (validation, testing) — עצרו. אל תמשיכו עם data שלא עבר בדיקה.
5. Human Escalation: שגיאות לא ברורות — עצרו, שלחו notification, המתינו להחלטה אנושית.

ה-Error Matrix: לפני שמריצים, כתבו טבלה שממפה לכל stage את ה-strategy לכל סוג שגיאה:

Versioning ו-Rollback

Pipelines משתנים: prompts מתעדכנים, models חדשים יוצאים, stages מתווספים. בלי versioning, אי אפשר לדעת למה התוצאה השתנתה.

3 סוגי versioning:

1. Pipeline Versioning: שינוי ב-orchestrator = commit + tag (v1.0, v1.1). כל output מסומן באיזה version ייצר אותו.

2. Prompt Versioning: Prompts משתנים לעיתים קרובות. שמרו כקבצים נפרדים:

prompts/
  research-v1.md
  research-v2.md
  write-v1.md
  write-v2.md

לעולם אל תמחקו גרסה ישנה. אם v2 לא עובד — חזרו ל-v1.

3. Output Versioning: כל ריצה מייצרת output בתיקייה עם timestamp:

output/
  2026-03-31_14-30_v1.2/
    research.json
    draft.md
    final.html
    pipeline-log.json

A/B Testing: הריצו גרסה ישנה וחדשה על אותו input, השוו quality. ככה תדעו אם שינוי שיפר או הרע.

Logging ו-Monitoring

Pipeline בלי logging הוא כמו מכונית בלי dashboard — רצה, אבל אתם לא יודעים כמה דלק נשאר.

Structured Logging — דוגמה:

{
  "stage": "research",
  "start_time": "2026-03-31T14:30:00Z",
  "end_time": "2026-03-31T14:32:15Z",
  "duration_seconds": 135,
  "model": "claude-opus-4-6",
  "tokens_input": 2500,
  "tokens_output": 8200,
  "cost_usd": 0.45,
  "status": "success",
  "output_summary": "Found 8 sources, 15 key facts"
}

Log Levels:

• DEBUG — full agent conversation (לפיתוח, לא production)
• INFO — stage start/end, key decisions, output summaries
• WARN — recoverable errors, retries, skipped stages
• ERROR — failures, aborts, unrecoverable issues

Pipeline Dashboard: דף HTML פשוט: ריצות אחרונות, timing per stage, total cost, success rate. JSON file + HTML table מספיק להתחלה.

Alerting: thresholds שמפעילים alerts:

• עלות > $X — Slack notification
• זמן > Y דקות — Slack notification
• failure rate > Z% — הודעה דחופה

Cost Tracking: צברו עלויות per stage, per run, per day. זהו stages יקרים — אולי אפשר להוריד model (Opus → Sonnet) בלי לפגוע באיכות.

ביצוע שלבים במקביל

לא כל ה-stages חייבים לרוץ ברצף. אם stage B לא תלוי ב-output של stage A — הם יכולים לרוץ במקביל.

Python:

import asyncio

async def run_parallel(stage_a, stage_b, input_data):
    result_a, result_b = await asyncio.gather(
        stage_a.execute_async(input_data),
        stage_b.execute_async(input_data)
    )
    return {"outline": result_a, "images": result_b}

TypeScript:

const [resultA, resultB] = await Promise.all([
  stageA.execute(inputData),
  stageB.execute(inputData)
]);
const merged = { outline: resultA, images: resultB };

הכלל: Stage B יכול להתחיל בלי output של A? הריצו במקביל. B חייב A? סדרתי. B צריך חלק מ-A? פצלו.

מקרה מבחן: Pipeline תוכן 7-שלבי

זה pipeline אמיתי שבנה את הקורס שאתם קוראים.

nVision Academy משתמשת ב-7 שלבים ליצירת פרקים חינוכיים:

ביצועים: 7,500-10,000 מילים | 15-25 דקות | $3-8 per chapter | 25-30/30 Gold Standard

עקרונות עיצוב מרכזיים:

• Metadata-first: Evaluators (stages 4-6) עובדים על manifest JSON, לא HTML מלא — חוסך ~90% טוקנים
• Stages 4-6 מקביליים: Structure review, fact check, value audit רצים במקביל
• Model specialization: Sonnet ל-mechanical work, Opus ל-judgment work
• Repair loop: publish-gate = "repair_first" → targeted repair → re-evaluate affected only

חיסכון טוקנים metadata-first: ~2,000 tokens per chapter evaluation vs ~15,000-25,000 for full reads. 10% tokens for 95% coverage.

דפוסי Pipeline לדומיינים שונים

Pipelines מתאימים לכל דומיין. 5 דפוסים מרכזיים:

1. Code Generation: Requirements → Design → Implementation → Testing → Review → Documentation → Deploy

2. Data Processing: Ingest → Validate → Clean → Transform → Analyze → Report → Archive

3. DevOps: Monitor → Detect → Diagnose → Fix → Test → Deploy → Verify

4. Content Marketing: Ideation → Research → Draft → Edit → SEO → Publish → Promote

5. Customer Support: Receive → Classify → Research → Draft → Review → Send → Follow up