1"""
2LLM Batch Processing Pipeline Template.
3 
4A composable, staged pipeline architecture for LLM batch processing.
5Each stage is discrete, idempotent, and cacheable. Customize the acquire,
6prepare, process, parse, and render functions for your use case.
7 
8Use when:
9    - Building a new batch processing pipeline with structured LLM outputs
10    - Prototyping an acquire -> prepare -> process -> parse -> render workflow
11    - Need a file-system-based state machine for pipeline stage tracking
12 
13Usage:
14    python pipeline_template.py acquire --batch-id 2025-01-15
15    python pipeline_template.py prepare --batch-id 2025-01-15
16    python pipeline_template.py process --batch-id 2025-01-15 --workers 10
17    python pipeline_template.py parse --batch-id 2025-01-15
18    python pipeline_template.py render --batch-id 2025-01-15
19    python pipeline_template.py all --batch-id 2025-01-15
20    python pipeline_template.py clean --batch-id 2025-01-15 --clean-stage process
21    python pipeline_template.py estimate --batch-id 2025-01-15
22 
23Programmatic usage:
24    from pipeline_template import stage_acquire, stage_prepare, stage_process
25    stage_acquire("2025-01-15", limit=5)
26    stage_prepare("2025-01-15")
27    stage_process("2025-01-15", model="claude-sonnet-4-20250514", max_workers=3)
28"""
29 
30import argparse
31import json
32import re
33import time
34from concurrent.futures import ThreadPoolExecutor, as_completed
35from dataclasses import dataclass, field, asdict
36from datetime import date
37from pathlib import Path
38from typing import Any
39 
40__all__ = [
41    "Item",
42    "ParsedResult",
43    "stage_acquire",
44    "stage_prepare",
45    "stage_process",
46    "stage_parse",
47    "stage_render",
48    "stage_clean",
49    "stage_estimate",
50    "parse_response",
51    "get_batch_dir",
52    "get_item_dir",
53    "get_output_dir",
54]
55 
56 
57# -----------------------------------------------------------------------------
58# Configuration - Customize for your use case
59# -----------------------------------------------------------------------------
60 
61DATA_DIR = Path("data")
62OUTPUT_DIR = Path("output")
63 
64# Prompt template with structured output requirements
65PROMPT_TEMPLATE = """Analyze the following content and provide your response in exactly this format.
66 
67## Summary
68[2-3 sentence summary of the content]
69 
70## Key Points
71- [Point 1]
72- [Point 2]
73- [Point 3]
74 
75## Score
76Rating: [1-10]
77Confidence: [low/medium/high]
78 
79## Reasoning
80[Explanation of your analysis]
81 
82Follow this format exactly because I will be parsing it programmatically.
83 
84---
85 
86# Content to Analyze
87 
88Title: {title}
89 
90{content}
91"""
92 
93 
94# -----------------------------------------------------------------------------
95# Data Structures
96# -----------------------------------------------------------------------------
97 
98@dataclass
99class Item:
100    """Represents a single item to process through the pipeline.
101 
102    Use when: creating items during the acquire stage or loading raw data
103    from any source (API, database, file system).
104    """
105 
106    id: str
107    title: str
108    content: str
109    metadata: dict[str, Any] = field(default_factory=dict)
110 
111 
112@dataclass
113class ParsedResult:
114    """Structured result from LLM response parsing.
115 
116    Use when: extracting structured data from free-text LLM responses
117    during the parse stage.
118    """
119 
120    summary: str = ""
121    key_points: list[str] = field(default_factory=list)
122    score: int | None = None
123    confidence: str = ""
124    reasoning: str = ""
125    parse_errors: list[str] = field(default_factory=list)
126 
127 
128# -----------------------------------------------------------------------------
129# Path Utilities
130# -----------------------------------------------------------------------------
131 
132def get_batch_dir(batch_id: str) -> Path:
133    """Get the data directory for a batch.
134 
135    Use when: resolving the root directory for a specific batch run.
136    """
137    return DATA_DIR / batch_id
138 
139 
140def get_item_dir(batch_id: str, item_id: str) -> Path:
141    """Get the directory for a specific item.
142 
143    Use when: locating stage output files for a single pipeline item.
144    """
145    return get_batch_dir(batch_id) / item_id
146 
147 
148def get_output_dir(batch_id: str) -> Path:
149    """Get the output directory for a batch.
150 
151    Use when: writing final rendered outputs (HTML, reports, etc.).
152    """
153    return OUTPUT_DIR / batch_id
154 
155 
156# -----------------------------------------------------------------------------
157# Stage: Acquire
158# -----------------------------------------------------------------------------
159 
160def stage_acquire(batch_id: str, limit: int | None = None) -> list[Path]:
161    """Stage 1: Acquire raw data from sources.
162 
163    Use when: fetching data from APIs, databases, or file systems
164    and persisting it as raw.json per item for downstream stages.
165 
166    Output: {batch_dir}/{item_id}/raw.json
167    Returns: List of item directories that were acquired.
168    """
169    batch_dir = get_batch_dir(batch_id)
170    batch_dir.mkdir(parents=True, exist_ok=True)
171 
172    # CUSTOMIZE: Replace with your data acquisition logic
173    items = fetch_items_from_source(limit)
174 
175    acquired_dirs: list[Path] = []
176    for item in items:
177        item_dir = get_item_dir(batch_id, item.id)
178        item_dir.mkdir(exist_ok=True)
179 
180        raw_file = item_dir / "raw.json"
181        if not raw_file.exists():
182            with open(raw_file, "w") as f:
183                json.dump(asdict(item), f, indent=2)
184            print(f"Acquired: {item.id}")
185        else:
186            print(f"Cached: {item.id}")
187 
188        acquired_dirs.append(item_dir)
189 
190    print(f"\nAcquire complete. {len(items)} items in {batch_dir}")
191    return acquired_dirs
192 
193 
194def fetch_items_from_source(limit: int | None = None) -> list[Item]:
195    """CUSTOMIZE: Implement your data fetching logic here.
196 
197    Use when: pulling raw items from your specific data source.
198    Replace this with actual API calls, database queries, etc.
199    """
200    # Example: Generate sample items
201    items: list[Item] = []
202    for i in range(limit or 10):
203        items.append(Item(
204            id=f"item-{i:04d}",
205            title=f"Sample Item {i}",
206            content=f"This is sample content for item {i}. " * 10,
207            metadata={"source": "example", "index": i},
208        ))
209    return items
210 
211 
212# -----------------------------------------------------------------------------
213# Stage: Prepare
214# -----------------------------------------------------------------------------
215 
216def stage_prepare(batch_id: str) -> int:
217    """Stage 2: Generate prompts from raw data.
218 
219    Use when: transforming raw acquired data into LLM-ready prompts
220    using the configured PROMPT_TEMPLATE.
221 
222    Output: {batch_dir}/{item_id}/prompt.md
223    Returns: Number of items prepared.
224    """
225    batch_dir = get_batch_dir(batch_id)
226    prepared_count = 0
227 
228    for item_dir in sorted(batch_dir.iterdir()):
229        if not item_dir.is_dir():
230            continue
231 
232        raw_file = item_dir / "raw.json"
233        prompt_file = item_dir / "prompt.md"
234 
235        if not raw_file.exists():
236            continue
237 
238        if prompt_file.exists():
239            continue
240 
241        with open(raw_file) as f:
242            item_data: dict[str, Any] = json.load(f)
243 
244        prompt = generate_prompt(item_data)
245 
246        with open(prompt_file, "w") as f:
247            f.write(prompt)
248 
249        prepared_count += 1
250        print(f"Prepared: {item_dir.name}")
251 
252    print(f"\nPrepare complete. {prepared_count} items prepared.")
253    return prepared_count
254 
255 
256def generate_prompt(item_data: dict[str, Any]) -> str:
257    """Generate prompt from item data using template.
258 
259    Use when: converting a raw item dict into a formatted prompt string.
260    """
261    return PROMPT_TEMPLATE.format(
262        title=item_data.get("title", "Untitled"),
263        content=item_data.get("content", ""),
264    )
265 
266 
267# -----------------------------------------------------------------------------
268# Stage: Process
269# -----------------------------------------------------------------------------
270 
271def stage_process(
272    batch_id: str,
273    model: str = "claude-sonnet-4-20250514",
274    max_workers: int = 5,
275) -> list[tuple[str, int, str | None]]:
276    """Stage 3: Execute LLM calls (the expensive, non-deterministic stage).
277 
278    Use when: sending prepared prompts to the LLM API and caching
279    responses. This is the only non-deterministic stage.
280 
281    Output: {batch_dir}/{item_id}/response.md
282    Returns: List of (item_id, char_count, error_or_none) tuples.
283    """
284    batch_dir = get_batch_dir(batch_id)
285 
286    # Collect items needing processing
287    to_process: list[tuple[Path, str]] = []
288    for item_dir in sorted(batch_dir.iterdir()):
289        if not item_dir.is_dir():
290            continue
291 
292        prompt_file = item_dir / "prompt.md"
293        response_file = item_dir / "response.md"
294 
295        if prompt_file.exists() and not response_file.exists():
296            to_process.append((item_dir, prompt_file.read_text()))
297 
298    if not to_process:
299        print("No items to process.")
300        return []
301 
302    print(f"Processing {len(to_process)} items with {max_workers} workers...")
303 
304    results: list[tuple[str, int, str | None]] = []
305 
306    def process_one(args: tuple[Path, str]) -> tuple[str, int, str | None]:
307        item_dir, prompt = args
308        response_file = item_dir / "response.md"
309 
310        try:
311            # CUSTOMIZE: Replace with your LLM API call
312            response = call_llm(prompt, model)
313 
314            with open(response_file, "w") as f:
315                f.write(response)
316 
317            return (item_dir.name, len(response), None)
318        except Exception as e:
319            return (item_dir.name, 0, str(e))
320 
321    with ThreadPoolExecutor(max_workers=max_workers) as executor:
322        futures = {executor.submit(process_one, item): item for item in to_process}
323 
324        for future in as_completed(futures):
325            item_id, chars, error = future.result()
326            results.append((item_id, chars, error))
327            if error:
328                print(f"  {item_id}: Error - {error}")
329            else:
330                print(f"  {item_id}: Done ({chars} chars)")
331 
332    print(f"\nProcess complete. {len(results)} items processed.")
333    return results
334 
335 
336def call_llm(prompt: str, model: str) -> str:
337    """CUSTOMIZE: Implement your LLM API call here.
338 
339    Use when: sending a single prompt to the LLM and returning the response.
340    Replace with actual OpenAI, Anthropic, etc. API calls.
341    """
342    # Example mock response - replace with actual API call
343    #
344    # import anthropic
345    # client = anthropic.Anthropic()
346    # message = client.messages.create(
347    #     model=model,
348    #     max_tokens=1024,
349    #     messages=[{"role": "user", "content": prompt}],
350    # )
351    # return message.content[0].text
352 
353    # Simulate API delay
354    time.sleep(0.1)
355 
356    # Return mock structured response
357    return """## Summary
358This is a sample summary of the analyzed content.
359 
360## Key Points
361- First key observation from the content
362- Second important finding
363- Third notable aspect
364 
365## Score
366Rating: 7
367Confidence: medium
368 
369## Reasoning
370The content demonstrates several characteristics that merit this rating.
371The analysis considered multiple factors including relevance and clarity.
372"""
373 
374 
375# -----------------------------------------------------------------------------
376# Stage: Parse
377# -----------------------------------------------------------------------------
378 
379def stage_parse(batch_id: str) -> list[dict[str, Any]]:
380    """Stage 4: Extract structured data from LLM responses.
381 
382    Use when: converting free-text LLM responses into structured
383    ParsedResult objects for aggregation and rendering.
384 
385    Output: {batch_dir}/{item_id}/parsed.json
386    Returns: List of parsed result dicts with item IDs.
387    """
388    batch_dir = get_batch_dir(batch_id)
389    all_results: list[dict[str, Any]] = []
390 
391    for item_dir in sorted(batch_dir.iterdir()):
392        if not item_dir.is_dir():
393            continue
394 
395        response_file = item_dir / "response.md"
396        parsed_file = item_dir / "parsed.json"
397 
398        if not response_file.exists():
399            continue
400 
401        response = response_file.read_text()
402        result = parse_response(response)
403 
404        with open(parsed_file, "w") as f:
405            json.dump(asdict(result), f, indent=2)
406 
407        all_results.append({
408            "id": item_dir.name,
409            **asdict(result),
410        })
411 
412        error_count = len(result.parse_errors)
413        print(f"Parsed: {item_dir.name} (score={result.score}, errors={error_count})")
414 
415    # Save aggregated results
416    agg_file = batch_dir / "all_results.json"
417    with open(agg_file, "w") as f:
418        json.dump(all_results, f, indent=2)
419 
420    print(f"\nParse complete. Results saved to {agg_file}")
421    return all_results
422 
423 
424def parse_response(text: str) -> ParsedResult:
425    """Parse structured LLM response with graceful error handling.
426 
427    Use when: extracting sections, scores, and lists from a formatted
428    LLM response. Logs parse errors rather than raising exceptions.
429    """
430    result = ParsedResult()
431 
432    # Extract summary
433    try:
434        result.summary = extract_section(text, "Summary") or ""
435    except Exception as e:
436        result.parse_errors.append(f"Summary: {e}")
437 
438    # Extract key points
439    try:
440        result.key_points = extract_list_items(text, "Key Points")
441    except Exception as e:
442        result.parse_errors.append(f"Key Points: {e}")
443 
444    # Extract score
445    try:
446        result.score = extract_score(text, "Rating", 1, 10)
447    except Exception as e:
448        result.parse_errors.append(f"Score: {e}")
449 
450    # Extract confidence
451    try:
452        result.confidence = extract_field(text, "Confidence") or ""
453    except Exception as e:
454        result.parse_errors.append(f"Confidence: {e}")
455 
456    # Extract reasoning
457    try:
458        result.reasoning = extract_section(text, "Reasoning") or ""
459    except Exception as e:
460        result.parse_errors.append(f"Reasoning: {e}")
461 
462    return result
463 
464 
465def extract_section(text: str, section_name: str) -> str | None:
466    """Extract content between section headers.
467 
468    Use when: pulling a named markdown section from LLM output.
469    """
470    pattern = rf'(?:^|\n)(?:#+ *)?{re.escape(section_name)}[:\s]*\n(.*?)(?=\n#|\Z)'
471    match = re.search(pattern, text, re.IGNORECASE | re.DOTALL)
472    return match.group(1).strip() if match else None
473 
474 
475def extract_field(text: str, field_name: str) -> str | None:
476    """Extract value after field label.
477 
478    Use when: pulling a single key-value field (e.g., "Confidence: high").
479    """
480    pattern = rf'(?:\*\*)?{re.escape(field_name)}(?:\*\*)?[\s:\-]+([^\n]+)'
481    match = re.search(pattern, text, re.IGNORECASE)
482    return match.group(1).strip() if match else None
483 
484 
485def extract_list_items(text: str, section_name: str) -> list[str]:
486    """Extract bullet points from a section.
487 
488    Use when: parsing a markdown list under a named section header.
489    """
490    section = extract_section(text, section_name)
491    if not section:
492        return []
493 
494    items = re.findall(r'^[\-\*]\s*(.+)$', section, re.MULTILINE)
495    return [item.strip() for item in items]
496 
497 
498def extract_score(
499    text: str, field_name: str, min_val: int, max_val: int
500) -> int | None:
501    """Extract and validate numeric score.
502 
503    Use when: pulling a bounded integer score from LLM output.
504    """
505    raw = extract_field(text, field_name)
506    if not raw:
507        return None
508 
509    match = re.search(r'\d+', raw)
510    if not match:
511        return None
512 
513    score = int(match.group())
514    return max(min_val, min(max_val, score))
515 
516 
517# -----------------------------------------------------------------------------
518# Stage: Render
519# -----------------------------------------------------------------------------
520 
521def stage_render(batch_id: str) -> Path | None:
522    """Stage 5: Generate final outputs from parsed results.
523 
524    Use when: producing human-readable output (HTML, reports)
525    from aggregated parsed results.
526 
527    Output: {output_dir}/index.html
528    Returns: Path to the rendered output file, or None if no results.
529    """
530    batch_dir = get_batch_dir(batch_id)
531    output_dir = get_output_dir(batch_id)
532    output_dir.mkdir(parents=True, exist_ok=True)
533 
534    # Load aggregated results
535    results_file = batch_dir / "all_results.json"
536    if not results_file.exists():
537        print("No results to render. Run parse stage first.")
538        return None
539 
540    with open(results_file) as f:
541        results: list[dict[str, Any]] = json.load(f)
542 
543    # CUSTOMIZE: Replace with your rendering logic
544    html = render_html(results, batch_id)
545 
546    output_file = output_dir / "index.html"
547    with open(output_file, "w") as f:
548        f.write(html)
549 
550    print(f"Rendered: {output_file}")
551    return output_file
552 
553 
554def render_html(results: list[dict[str, Any]], batch_id: str) -> str:
555    """Generate HTML output from results.
556 
557    Use when: creating a summary HTML table from parsed pipeline results.
558    """
559    import html as html_lib
560 
561    rows = ""
562    for r in results:
563        rows += f"""
564        <tr>
565            <td>{html_lib.escape(r.get('id', ''))}</td>
566            <td>{html_lib.escape(r.get('summary', '')[:100])}...</td>
567            <td>{r.get('score', 'N/A')}</td>
568            <td>{html_lib.escape(r.get('confidence', ''))}</td>
569        </tr>"""
570 
571    return f"""<!DOCTYPE html>
572<html>
573<head>
574    <meta charset="utf-8">
575    <title>Results - {batch_id}</title>
576    <style>
577        body {{ font-family: system-ui, sans-serif; max-width: 1000px; margin: 0 auto; padding: 20px; }}
578        table {{ width: 100%; border-collapse: collapse; }}
579        th, td {{ text-align: left; padding: 10px; border-bottom: 1px solid #ddd; }}
580        th {{ background: #f5f5f5; }}
581    </style>
582</head>
583<body>
584    <h1>Results: {batch_id}</h1>
585    <p>{len(results)} items processed</p>
586    <table>
587        <tr>
588            <th>ID</th>
589            <th>Summary</th>
590            <th>Score</th>
591            <th>Confidence</th>
592        </tr>
593        {rows}
594    </table>
595</body>
596</html>"""
597 
598 
599# -----------------------------------------------------------------------------
600# Clean Stage
601# -----------------------------------------------------------------------------
602 
603def stage_clean(batch_id: str, from_stage: str | None = None) -> int:
604    """Remove stage outputs to enable re-processing.
605 
606    Use when: a stage produced bad results and needs to be re-run,
607    or when clearing all intermediate files for a fresh pipeline run.
608 
609    Returns: Number of files deleted.
610    """
611    batch_dir = get_batch_dir(batch_id)
612 
613    if not batch_dir.exists():
614        print(f"No data directory for {batch_id}")
615        return 0
616 
617    stage_outputs: dict[str, list[str]] = {
618        "acquire": ["raw.json"],
619        "prepare": ["prompt.md"],
620        "process": ["response.md"],
621        "parse": ["parsed.json"],
622    }
623 
624    stage_order = ["acquire", "prepare", "process", "parse", "render"]
625 
626    if from_stage:
627        start_idx = stage_order.index(from_stage)
628        stages_to_clean = stage_order[start_idx:]
629    else:
630        stages_to_clean = stage_order
631 
632    files_to_delete: set[str] = set()
633    for s in stages_to_clean:
634        files_to_delete.update(stage_outputs.get(s, []))
635 
636    deleted_count = 0
637    for item_dir in batch_dir.iterdir():
638        if not item_dir.is_dir():
639            continue
640 
641        for filename in files_to_delete:
642            filepath = item_dir / filename
643            if filepath.exists():
644                filepath.unlink()
645                deleted_count += 1
646 
647    # Clean aggregated results
648    if "parse" in stages_to_clean:
649        agg_file = batch_dir / "all_results.json"
650        if agg_file.exists():
651            agg_file.unlink()
652            deleted_count += 1
653 
654    print(f"Cleaned {deleted_count} files from stage '{from_stage or 'all'}' onwards")
655    return deleted_count
656 
657 
658# -----------------------------------------------------------------------------
659# Cost Estimation
660# -----------------------------------------------------------------------------
661 
662def stage_estimate(batch_id: str) -> dict[str, Any] | None:
663    """Estimate processing costs before running the process stage.
664 
665    Use when: projecting token costs and budget requirements before
666    committing to expensive LLM API calls.
667 
668    Returns: Dict with item_count, token estimates, and cost projection,
669             or None if no prompts are available.
670    """
671    batch_dir = get_batch_dir(batch_id)
672 
673    if not batch_dir.exists():
674        print(f"No data directory for {batch_id}. Run acquire first.")
675        return None
676 
677    # Count items and estimate tokens
678    item_count = 0
679    total_prompt_chars = 0
680 
681    for item_dir in batch_dir.iterdir():
682        if not item_dir.is_dir():
683            continue
684 
685        prompt_file = item_dir / "prompt.md"
686        if prompt_file.exists():
687            total_prompt_chars += len(prompt_file.read_text())
688            item_count += 1
689 
690    if item_count == 0:
691        print("No prompts found. Run prepare first.")
692        return None
693 
694    # Rough token estimation (1 token ~ 4 chars)
695    est_input_tokens = total_prompt_chars / 4
696    est_output_tokens = item_count * 500  # Assume 500 tokens per response
697 
698    # Example pricing (adjust for your model)
699    input_price = 3.0 / 1_000_000   # $3 per MTok
700    output_price = 15.0 / 1_000_000  # $15 per MTok
701 
702    est_cost = (est_input_tokens * input_price) + (est_output_tokens * output_price)
703 
704    estimate: dict[str, Any] = {
705        "batch_id": batch_id,
706        "item_count": item_count,
707        "est_input_tokens": int(est_input_tokens),
708        "est_output_tokens": int(est_output_tokens),
709        "est_cost_usd": round(est_cost, 2),
710    }
711 
712    print(f"Cost Estimate for {batch_id}")
713    print(f"  Items: {item_count}")
714    print(f"  Estimated input tokens: {int(est_input_tokens):,}")
715    print(f"  Estimated output tokens: {int(est_output_tokens):,}")
716    print(f"  Estimated cost: ${est_cost:.2f}")
717    print(f"\nNote: Actual costs may vary. Add 20-30% buffer for retries.")
718 
719    return estimate
720 
721 
722# -----------------------------------------------------------------------------
723# CLI
724# -----------------------------------------------------------------------------
725 
726def main() -> None:
727    """Entry point for CLI usage. Parses arguments and dispatches to stages."""
728    parser = argparse.ArgumentParser(
729        description="LLM Batch Processing Pipeline",
730        formatter_class=argparse.RawDescriptionHelpFormatter,
731        epilog=__doc__,
732    )
733 
734    parser.add_argument(
735        "stage",
736        choices=["acquire", "prepare", "process", "parse", "render", "all", "clean", "estimate"],
737        help="Pipeline stage to run",
738    )
739    parser.add_argument(
740        "--batch-id",
741        default=None,
742        help="Batch identifier (default: today's date)",
743    )
744    parser.add_argument(
745        "--limit",
746        type=int,
747        default=None,
748        help="Limit number of items (for testing)",
749    )
750    parser.add_argument(
751        "--workers",
752        type=int,
753        default=5,
754        help="Number of parallel workers for processing",
755    )
756    parser.add_argument(
757        "--model",
758        default="claude-sonnet-4-20250514",
759        help="Model to use for processing",
760    )
761    parser.add_argument(
762        "--clean-stage",
763        choices=["acquire", "prepare", "process", "parse"],
764        help="For clean: only clean this stage and downstream",
765    )
766 
767    args = parser.parse_args()
768 
769    batch_id = args.batch_id or date.today().isoformat()
770    print(f"Batch ID: {batch_id}\n")
771 
772    if args.stage == "clean":
773        stage_clean(batch_id, args.clean_stage)
774    elif args.stage == "estimate":
775        stage_estimate(batch_id)
776    elif args.stage == "all":
777        stage_acquire(batch_id, args.limit)
778        stage_prepare(batch_id)
779        stage_process(batch_id, args.model, args.workers)
780        stage_parse(batch_id)
781        stage_render(batch_id)
782    else:
783        if args.stage == "acquire":
784            stage_acquire(batch_id, args.limit)
785        elif args.stage == "prepare":
786            stage_prepare(batch_id)
787        elif args.stage == "process":
788            stage_process(batch_id, args.model, args.workers)
789        elif args.stage == "parse":
790            stage_parse(batch_id)
791        elif args.stage == "render":
792            stage_render(batch_id)
793 
794 
795if __name__ == "__main__":
796    main()
797