Project Development

Project Development Methodology

This skill covers the principles for identifying tasks suited to LLM processing, designing effective project architectures, and iterating rapidly using agent-assisted development. The methodology applies whether building a batch processing pipeline, a multi-agent research system, or an interactive agent application.

When to Activate

Activate this skill when:

• Starting a new project that might benefit from LLM processing
• Evaluating whether a task is well-suited for agents versus traditional code
• Designing the architecture for an LLM-powered application
• Planning a batch processing pipeline with structured outputs
• Choosing between single-agent and multi-agent approaches
• Estimating costs and timelines for LLM-heavy projects

Core Concepts

Task-Model Fit Recognition

Evaluate task-model fit before writing any code, because building automation on a fundamentally mismatched task wastes days of effort. Run every proposed task through these two tables to decide proceed-or-stop.

Proceed when the task has these characteristics:

Stop when the task has these characteristics:

The Manual Prototype Step

Always validate task-model fit with a manual test before investing in automation. Copy one representative input into the model interface, evaluate the output quality, and use the result to answer these questions:

• Does the model have the knowledge required for this task?
• Can the model produce output in the format needed?
• What level of quality should be expected at scale?
• Are there obvious failure modes to address?

Do this because a failed manual prototype predicts a failed automated system, while a successful one provides both a quality baseline and a prompt-design template. The test takes minutes and prevents hours of wasted development.

Pipeline Architecture

Structure LLM projects as staged pipelines because separation of deterministic and non-deterministic stages enables fast iteration and cost control. Design each stage to be:

• Discrete: Clear boundaries between stages so each can be debugged independently
• Idempotent: Re-running produces the same result, preventing duplicate work
• Cacheable: Intermediate results persist to disk, avoiding expensive re-computation
• Independent: Each stage can run separately, enabling selective re-execution

Use this canonical pipeline structure:

acquire -> prepare -> process -> parse -> render

1. Acquire: Fetch raw data from sources (APIs, files, databases) 2. Prepare: Transform data into prompt format 3. Process: Execute LLM calls (the expensive, non-deterministic step) 4. Parse: Extract structured data from LLM outputs 5. Render: Generate final outputs (reports, files, visualizations)

Stages 1, 2, 4, and 5 are deterministic. Stage 3 is non-deterministic and expensive. Maintain this separation because it allows re-running the expensive LLM stage only when necessary, while iterating quickly on parsing and rendering.

File System as State Machine

Use the file system to track pipeline state rather than databases or in-memory structures, because file existence provides natural idempotency and human-readable debugging.

data/{id}/
  raw.json         # acquire stage complete
  prompt.md        # prepare stage complete
  response.md      # process stage complete
  parsed.json      # parse stage complete

Check if an item needs processing by checking whether the output file exists. Re-run a stage by deleting its output file and downstream files. Debug by reading the intermediate files directly. This pattern works because each directory is independent, enabling simple parallelization and trivial caching.

Structured Output Design

Design prompts for structured, parseable outputs because prompt design directly determines parsing reliability. Include these elements in every structured prompt:

1. Section markers: Explicit headers or prefixes that parsers can match on 2. Format examples: Show exactly what output should look like 3. Rationale disclosure: State "I will be parsing this programmatically" so the model prioritizes format compliance 4. Constrained values: Enumerated options, score ranges, and fixed formats

Build parsers that handle LLM output variations gracefully, because LLMs do not follow instructions perfectly. Use regex patterns flexible enough for minor formatting variations, provide sensible defaults when sections are missing, and log parsing failures for review rather than crashing.

Agent-Assisted Development

Use agent-capable models to accelerate development through rapid iteration: describe the project goal and constraints, let the agent generate initial implementation, test and iterate on specific failures, then refine prompts and architecture based on results.

Adopt these practices because they keep agent output focused and high-quality:

• Provide clear, specific requirements upfront to reduce revision cycles
• Break large projects into discrete components so each can be validated independently
• Test each component before moving to the next to catch failures early
• Keep the agent focused on one task at a time to prevent context degradation

Cost and Scale Estimation

Estimate LLM processing costs before starting, because token costs compound quickly at scale and late discovery of budget overruns forces costly rework. Use this formula:

Total cost = (items x tokens_per_item x price_per_token) + API overhead

For batch processing, estimate input tokens per item (prompt + context), estimate output tokens per item (typical response length), multiply by item count, and add 20-30% buffer for retries and failures.

Track actual costs during development. If costs exceed estimates significantly, reduce context length through truncation, use smaller models for simpler items, cache and reuse partial results, or add parallel processing to reduce wall-clock time.

Detailed Topics

Choosing Single vs Multi-Agent Architecture

Default to single-agent pipelines for batch processing with independent items, because they are simpler to manage, cheaper to run, and easier to debug. Escalate to multi-agent architectures only when one of these conditions holds:

• Parallel exploration of different aspects is required
• The task exceeds single context window capacity
• Specialized sub-agents demonstrably improve quality on benchmarks

Choose multi-agent for context isolation, not role anthropomorphization. Sub-agents get fresh context windows for focused subtasks, which prevents context degradation on long-running tasks.

See multi-agent-patterns skill for detailed architecture guidance.

Architectural Reduction

Start with minimal architecture and add complexity only when production evidence proves it necessary, because over-engineered scaffolding often constrains rather than enables model performance.

Vercel's d0 agent achieved 100% success rate (up from 80%) by reducing from 17 specialized tools to 2 primitives: bash command execution and SQL. The file system agent pattern uses standard Unix utilities (grep, cat, find, ls) instead of custom exploration tools.

Reduce when:

• The data layer is well-documented and consistently structured
• The model has sufficient reasoning capability
• Specialized tools are constraining rather than enabling
• More time is spent maintaining scaffolding than improving outcomes

Add complexity when:

• The underlying data is messy, inconsistent, or poorly documented
• The domain requires specialized knowledge the model lacks
• Safety constraints require limiting agent capabilities
• Operations are truly complex and benefit from structured workflows

See tool-design skill for detailed tool architecture guidance.

Iteration and Refactoring

Plan for multiple architectural iterations from the start, because production agent systems at scale always require refactoring. Manus refactored their agent framework five times since launch. The Bitter Lesson suggests that structures added for current model limitations become constraints as models improve.

Build for change by following these practices:

• Keep architecture simple and unopinionated so refactoring is cheap
• Test across model generations to verify the harness is not limiting performance
• Design systems that benefit from model improvements rather than locking in limitations

Practical Guidance

Project Planning Template

Follow this template in order, because each step validates assumptions before the next step invests effort.

1. Task Analysis

• Define the input and desired output explicitly
• Classify: synthesis, generation, classification, or analysis
• Set an acceptable error rate based on business impact
• Estimate the value per successful completion to justify costs

2. Manual Validation

• Test one representative example with the target model
• Evaluate output quality and format against requirements
• Identify failure modes that need parser hardening or prompt revision
• Estimate tokens per item for cost projection

3. Architecture Selection

• Choose single pipeline vs multi-agent based on the criteria above
• Identify required tools and data sources
• Design storage and caching strategy using file-system state
• Plan parallelization approach for the process stage

4. Cost Estimation

• Calculate items x tokens x price with a 20-30% buffer
• Estimate development time for each pipeline stage
• Identify infrastructure requirements (API keys, storage, compute)
• Project ongoing operational costs for production runs

5. Development Plan

• Implement stage-by-stage, testing each before proceeding
• Define a testing strategy per stage with expected outputs
• Set iteration milestones tied to quality metrics
• Plan deployment approach with rollback capability

Examples

Example 1: Batch Analysis Pipeline (Karpathy's HN Time Capsule)

Task: Analyze 930 HN discussions from 10 years ago with hindsight grading.

Architecture:

• 5-stage pipeline: fetch -> prompt -> analyze -> parse -> render
• File system state: data/{date}/{item_id}/ with stage output files
• Structured output: 6 sections with explicit format requirements
• Parallel execution: 15 workers for LLM calls

Results: $58 total cost, ~1 hour execution, static HTML output.

Example 2: Architectural Reduction (Vercel d0)

Task: Text-to-SQL agent for internal analytics.

Before: 17 specialized tools, 80% success rate, 274s average execution.

After: 2 tools (bash + SQL), 100% success rate, 77s average execution.

Key insight: The semantic layer was already good documentation. Claude just needed access to read files directly.

See Case Studies for detailed analysis.

Guidelines

1. Validate task-model fit with manual prototyping before building automation 2. Structure pipelines as discrete, idempotent, cacheable stages 3. Use the file system for state management and debugging 4. Design prompts for structured, parseable outputs with explicit format examples 5. Start with minimal architecture; add complexity only when proven necessary 6. Estimate costs early and track throughout development 7. Build robust parsers that handle LLM output variations 8. Expect and plan for multiple architectural iterations 9. Test whether scaffolding helps or constrains model performance 10. Use agent-assisted development for rapid iteration on implementation

Gotchas

1. Skipping manual validation: Building automation before verifying the model can do the task wastes significant time when the approach is fundamentally flawed. Always run one representative example through the model interface first. 2. Monolithic pipelines: Combining all stages into one script makes debugging and iteration difficult. Separate stages with persistent intermediate outputs so each can be re-run independently. 3. Over-constraining the model: Adding guardrails, pre-filtering, and validation logic that the model could handle on its own reduces performance. Test whether scaffolding helps or hurts before keeping it. 4. Ignoring costs until production: Token costs compound quickly at scale. Estimate and track from the beginning to avoid budget surprises that force architectural rework. 5. Perfect parsing requirements: Expecting LLMs to follow format instructions perfectly leads to brittle systems. Build robust parsers that handle variations and log failures for review. 6. Premature optimization: Adding caching, parallelization, and optimization before the basic pipeline works correctly wastes effort on code that may be discarded during iteration. 7. Model version lock-in: Building pipelines that only work with one specific model version creates fragile systems. Test across model generations and abstract the LLM call layer so models can be swapped without rewriting pipeline logic. 8. Evaluation-less deployment: Shipping agent pipelines without measuring output quality means regressions go undetected. Define quality metrics during development and run evaluation checks before and after every model or prompt change.

Integration

This skill connects to:

• context-fundamentals - Understanding context constraints for prompt design
• tool-design - Designing tools for agent systems within pipelines
• multi-agent-patterns - When to use multi-agent versus single pipelines
• evaluation - Evaluating pipeline outputs and agent performance
• context-compression - Managing context when pipelines exceed limits

References

Internal references:

• Case Studies - Read when: evaluating architecture tradeoffs or reviewing real-world pipeline implementations (Karpathy HN Capsule, Vercel d0, Manus patterns)
• Pipeline Patterns - Read when: designing a new pipeline stage layout, choosing caching strategies, or debugging stage boundaries

Related skills in this collection:

• tool-design - Tool architecture and reduction patterns
• multi-agent-patterns - When to use multi-agent architectures
• evaluation - Output evaluation frameworks

External resources:

• Karpathy's HN Time Capsule project: https://github.com/karpathy/hn-time-capsule
• Vercel d0 architectural reduction: https://vercel.com/blog/we-removed-80-percent-of-our-agents-tools
• Manus context engineering: Peak Ji's blog on context engineering lessons
• Anthropic multi-agent research: How we built our multi-agent research system

---

Skill Metadata

Created: 2025-12-25 Last Updated: 2026-03-17 Author: Agent Skills for Context Engineering Contributors Version: 1.1.0

Case Studies: LLM Project Development

This reference contains detailed case studies of production LLM projects that demonstrate effective development methodology. Each case study analyzes the problem, approach, architecture, and lessons learned.

Case Study 1: Karpathy's HN Time Capsule

Source: https://github.com/karpathy/hn-time-capsule

Problem Statement

Analyze Hacker News discussions from 10 years ago and grade commenters on how prescient their predictions were with the benefit of hindsight.

Task-Model Fit Analysis

This task is well-suited for LLM processing because:

Development Methodology

Step 1: Manual Prototype

Before building any automation, Karpathy copy-pasted one article + comment thread into ChatGPT to validate the approach. This took minutes and confirmed:

• The model could produce insightful hindsight analysis
• The output format worked for the intended use case
• The quality exceeded what he could do manually

Step 2: Agent-Assisted Implementation

Used Opus 4.5 to build the pipeline in approximately 3 hours. The agent handled:

• HTML parsing for HN frontpage
• Algolia API integration for comments
• Prompt template design
• Output parsing logic
• Static HTML rendering

Step 3: Batch Execution

• 930 LLM queries (31 days × 30 articles)
• 15 parallel workers
• ~$58 total cost
• ~1 hour execution time

Pipeline Architecture

fetch → prompt → analyze → parse → render

Stage 1: Fetch

• Download HN frontpage for target date
• Fetch article content via HTTP
• Fetch comments via Algolia API
• Output: data/{date}/{item_id}/meta.json, article.txt, comments.json

Stage 2: Prompt

• Load article metadata and content
• Load comment tree
• Generate markdown prompt from template
• Output: data/{date}/{item_id}/prompt.md

Stage 3: Analyze

• Submit prompt to GPT 5.1 Thinking API
• Parallel execution with ThreadPoolExecutor
• Output: data/{date}/{item_id}/response.md

Stage 4: Parse

• Extract grades from "Final grades" section via regex
• Extract interestingness score via regex
• Aggregate grades across all articles
• Output: data/{date}/{item_id}/grades.json, score.json

Stage 5: Render

• Generate static HTML with embedded JavaScript
• Create day pages with article navigation
• Create Hall of Fame with aggregated rankings
• Output: output/{date}/index.html, output/hall-of-fame.html

Structured Output Design

The prompt template specifies exact output format:

Let's use our benefit of hindsight now in 6 sections:

1. Give a brief summary of the article and the discussion thread.
2. What ended up happening to this topic?
3. Give out awards for "Most prescient" and "Most wrong" comments.
4. Mention any other fun or notable aspects.
5. Give out grades to specific people for their comments.
6. At the end, give a final score (from 0-10).

As for the format of Section 5, use the header "Final grades" and follow it 
with simply an unordered list in the format of "name: grade (optional comment)".

Please follow the format exactly because I will be parsing it programmatically.

Key techniques:

• Numbered sections for structure
• Explicit format specification with examples
• Rationale disclosure ("because I will be parsing it")
• Constrained output (letter grades, 0-10 scores)

Parsing Implementation

The parsing code handles variations gracefully:

def parse_grades(text: str) -> dict[str, dict]:
    # Match "Final grades" with optional section number or markdown
    pattern = r'(?:^|\n)(?:\d+[\.\)]\s*)?(?:#+ *)?Final grades\s*\n'
    match = re.search(pattern, text, re.IGNORECASE)
    
    # Handle both ASCII and Unicode minus signs
    line_pattern = r'^[\-\*]\s*([^:]+):\s*([A-F][+\-−]?)(?:\s*\(([^)]+)\))?'

Lessons Learned

1. Manual validation first: The 5-minute copy-paste test prevented hours of wasted development.

2. File system as state: Each article directory contains all intermediate outputs, making debugging trivial.

3. Idempotent stages: Re-running only processes items that lack output files.

4. Agent-assisted development: 3 hours to working code by focusing on requirements, not implementation details.

5. Parallel execution: 15 workers reduced execution time without increasing token costs.

---

Case Study 2: Vercel d0 Architectural Reduction

Source: https://vercel.com/blog/we-removed-80-percent-of-our-agents-tools

Problem Statement

Build a text-to-SQL agent that enables anyone at Vercel to query analytics data through natural language questions in Slack.

Initial Approach (Failed)

The team built a sophisticated system with:

• 17 specialized tools (schema lookup, query validation, error recovery, etc.)
• Heavy prompt engineering to constrain reasoning
• Careful context management
• Hand-coded retrieval for schema information

Results:

• 80% success rate
• 274.8 seconds average execution time
• ~102k tokens average usage
• ~12 steps average
• Constant maintenance burden

The Problem

The team was solving problems the model could handle on its own:

• Pre-filtering context
• Constraining options
• Wrapping every interaction in validation logic
• Building tools to "protect" the model from complexity

Every edge case required another patch. Every model update required re-calibrating constraints. More time was spent maintaining scaffolding than improving outcomes.

Architectural Reduction

The hypothesis: What if we just give Claude access to the raw files and let it figure things out?

New architecture:

• 2 tools total: ExecuteCommand (bash) + ExecuteSQL
• Direct file system access via sandbox
• Semantic layer as YAML/Markdown/JSON files
• Standard Unix utilities (grep, cat, find, ls)

const agent = new ToolLoopAgent({
  model: "anthropic/claude-opus-4.5",
  tools: {
    ExecuteCommand: executeCommandTool(sandbox),
    ExecuteSQL,
  },
});

Results

The worst case before: 724 seconds, 100 steps, 145k tokens, and still failed. Same query after: 141 seconds, 19 steps, 67k tokens, succeeded.

Why It Worked

1. Good documentation already existed: The semantic layer files contained dimension definitions, measure calculations, and join relationships. The tools were summarizing what was already legible.

2. File systems are proven abstractions: The model understands file systems deeply from training. grep is 50 years old and works perfectly.

3. Constraints became liabilities: With better models, the guardrails were limiting performance more than helping.

Key Lessons

1. Addition by subtraction: The best agents might be ones with the fewest tools. Every tool is a choice you are making for the model.

2. Build for future models: Models improve faster than tooling. Architectures optimized for today may be over-constrained for tomorrow.

3. Good context over clever tools: Invest in documentation, clear naming, and well-structured data. That foundation matters more than sophisticated tooling.

4. Start simple: Model + file system + goal. Add complexity only when proven necessary.

---

Case Study 3: Manus Context Engineering

Source: Peak Ji's blog "Context Engineering for AI Agents: Lessons from Building Manus"

Problem Statement

Build a general-purpose consumer agent that can accomplish complex tasks across 50+ tool calls while maintaining performance and managing costs.

Core Insight

KV-cache hit rate is the single most important metric for production agents. It directly affects both latency and cost.

• Claude Sonnet cached: $0.30/MTok
• Claude Sonnet uncached: $3.00/MTok
• 10x cost difference

With an average input-to-output ratio of 100:1 in agentic workloads, optimizing for cache hits dominates the cost equation.

Key Patterns

1. Append-Only Context

Never modify previous actions or observations. Ensure deterministic serialization (JSON key ordering must be stable). A single token difference invalidates the cache from that point forward.

Common mistake: Including a timestamp at the beginning of the system prompt kills cache hit rate entirely.

2. Mask, Do Not Remove

Do not dynamically add or remove tools mid-iteration. Tool definitions live near the front of context - any change invalidates the KV-cache for all subsequent content.

Instead, use logit masking during decoding to constrain tool selection without modifying definitions. This maintains cache while still controlling behavior.

3. File System as Context

Treat the file system as unlimited, persistent, agent-operable memory. The model learns to write and read files on demand.

Compression strategies should be restorable:

• Web page content can be dropped if URL is preserved
• Document contents can be omitted if file path remains available

4. Recitation for Attention

Manus creates a todo.md file and updates it step-by-step. This is not just organization - it pushes the global plan into the model's recent attention span.

By constantly rewriting objectives at the end of context, the agent avoids "lost in the middle" issues and maintains goal alignment.

5. Keep Errors In Context

Do not hide failures. When the model sees a failed action and the resulting error, it implicitly updates beliefs and avoids repeating mistakes.

Erasing failures removes evidence the model needs to adapt.

Multi-Agent for Context Isolation

The primary goal of sub-agents in Manus is context isolation, not role division. For tasks requiring discrete work:

• Planner assigns tasks to sub-agents with their own context windows
• Simple tasks: pass instructions via function call
• Complex tasks: share full context with sub-agent

Sub-agents have a submit_results tool with constrained output schema. Constrained decoding ensures adherence to defined format.

Layered Action Space

Rather than binding every utility as a tool:

• Small set (<20) of atomic functions: Bash, filesystem access, code execution
• Most actions offload to sandbox layer
• MCP tools exposed through CLI, executed via Bash tool

This reduces tool definition tokens and prevents model confusion from overlapping descriptions.

Iteration Expectation

Manus has refactored their agent framework five times since launch. The Bitter Lesson suggests structures added for current limitations become constraints as models improve.

Test across model strengths to verify your harness is not limiting performance. Simple, unopinionated designs adapt better to model improvements.

---

Case Study 4: Anthropic Multi-Agent Research

Source: Anthropic blog "How we built our multi-agent research system"

Problem Statement

Build a research feature that can explore complex topics using multiple parallel agents searching across web, Google Workspace, and integrations.

Architecture

Orchestrator-worker pattern:

• Lead agent analyzes query and develops strategy
• Lead spawns subagents for parallel exploration
• Subagents return findings to lead for synthesis
• Citation agent processes final output

Performance Insight

Three factors explained 95% of performance variance in BrowseComp evaluation:

• Token usage: 80% of variance
• Number of tool calls: additional factor
• Model choice: additional factor

Multi-agent architectures effectively scale token usage for tasks exceeding single-agent limits.

Token Economics

• Chat interactions: baseline
• Single agent: ~4x more tokens than chat
• Multi-agent: ~15x more tokens than chat

Multi-agent requires high-value tasks to justify the cost.

Prompting Principles

1. Think like your agents: Build simulations, watch step-by-step, identify failure modes.

2. Teach delegation: Subagents need objective, output format, tools/sources guidance, and clear boundaries.

3. Scale effort to complexity: Explicit guidelines for agent/tool call counts by task type.

4. Tool design is critical: Distinct purpose and clear description for each tool. Bad descriptions send agents down wrong paths entirely.

5. Let agents improve themselves: Claude 4 models can diagnose prompt failures and suggest improvements. Tool-testing agents can rewrite tool descriptions to avoid common mistakes.

6. Start wide, then narrow: Broad queries first, evaluate landscape, then drill into specifics.

7. Guide thinking process: Extended thinking mode as controllable scratchpad for planning.

8. Parallel tool calling: 3-5 subagents in parallel, 3+ tools per subagent in parallel. Cut research time by up to 90%.

Evaluation Approach

• Start with ~20 representative queries immediately
• LLM-as-judge with rubric: factual accuracy, citation accuracy, completeness, source quality, tool efficiency
• Human evaluation catches edge cases automation misses
• Focus on end-state evaluation for multi-turn agents

---

Cross-Case Patterns

Common Success Factors

1. Manual validation before automation: All successful projects validated task-model fit with simple tests first.

2. File system as foundation: Whether for state management (Karpathy), tool interface (Vercel), or memory (Manus), the file system provides proven abstractions.

3. Architectural simplicity: Reduction outperformed complexity in multiple cases. Start minimal, add only what proves necessary.

4. Structured outputs with robust parsing: Explicit format specifications combined with flexible parsing that handles variations.

5. Iteration expectation: No project got architecture right on the first try. Build for change.

Common Failure Patterns

1. Over-constraining models: Guardrails that helped with weaker models become liabilities as capabilities improve.

2. Tool proliferation: More tools often means more confusion and worse performance.

3. Hiding errors: Removing failures from context prevents models from learning.

4. Premature optimization: Adding complexity before basic functionality works.

5. Ignoring economics: Token costs compound quickly; estimation and tracking are essential.

Pipeline Patterns for LLM Projects

This reference provides detailed patterns for structuring LLM processing pipelines. These patterns apply to batch processing, data analysis, content generation, and similar workloads.

The Canonical Pipeline

acquire → prepare → process → parse → render

Stage Characteristics

The key insight: only the Process stage involves LLM calls. All other stages are deterministic transformations that can be debugged, tested, and iterated independently.

File System State Management

Directory Structure Pattern

project/
├── data/
│   └── {batch_id}/
│       └── {item_id}/
│           ├── raw.json         # Acquire output
│           ├── prompt.md        # Prepare output
│           ├── response.md      # Process output
│           └── parsed.json      # Parse output
├── output/
│   └── {batch_id}/
│       └── index.html           # Render output
└── config/
    └── prompts/
        └── template.md          # Prompt templates

State Checking Pattern

def needs_processing(item_dir: Path, stage: str) -> bool:
    """Check if an item needs processing for a given stage."""
    stage_outputs = {
        "acquire": ["raw.json"],
        "prepare": ["prompt.md"],
        "process": ["response.md"],
        "parse": ["parsed.json"],
    }
    
    for output_file in stage_outputs[stage]:
        if not (item_dir / output_file).exists():
            return True
    return False

Clean/Retry Pattern

def clean_from_stage(item_dir: Path, stage: str):
    """Remove outputs from stage and all downstream stages."""
    stage_order = ["acquire", "prepare", "process", "parse", "render"]
    stage_outputs = {
        "acquire": ["raw.json"],
        "prepare": ["prompt.md"],
        "process": ["response.md"],
        "parse": ["parsed.json"],
    }
    
    start_idx = stage_order.index(stage)
    for s in stage_order[start_idx:]:
        for output_file in stage_outputs.get(s, []):
            filepath = item_dir / output_file
            if filepath.exists():
                filepath.unlink()

Parallel Execution Patterns

ThreadPoolExecutor for LLM Calls

from concurrent.futures import ThreadPoolExecutor, as_completed

def process_batch(items: list, max_workers: int = 10):
    """Process items in parallel with progress tracking."""
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_item, item): item for item in items}
        
        for future in as_completed(futures):
            item = futures[future]
            try:
                result = future.result()
                results.append((item, result, None))
            except Exception as e:
                results.append((item, None, str(e)))
    
    return results

Batch Size Considerations

• Small batches (1-10): Sequential processing is fine; overhead of parallelization not worth it
• Medium batches (10-100): Parallelize with 5-15 workers depending on API rate limits
• Large batches (100+): Consider chunking with checkpoints; implement resume capability

Rate Limiting

import time
from functools import wraps

def rate_limited(calls_per_second: float):
    """Decorator to rate limit function calls."""
    min_interval = 1.0 / calls_per_second
    last_call = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            result = func(*args, **kwargs)
            last_call[0] = time.time()
            return result
        return wrapper
    return decorator

Structured Output Patterns

Prompt Template Structure

[INSTRUCTION BLOCK]
Analyze the following content and provide your response in exactly this format.

[FORMAT SPECIFICATION]
## Section 1: Summary
[Your summary here - 2-3 sentences]

## Section 2: Analysis
- Point 1
- Point 2
- Point 3

## Section 3: Score
Rating: [1-10]
Confidence: [low/medium/high]

[FORMAT ENFORCEMENT]
Follow this format exactly because I will be parsing it programmatically.

---

[CONTENT BLOCK]
# Title: {title}

## Content
{content}

## Additional Context
{context}

Parsing Patterns

Section Extraction

import re

def extract_section(text: str, section_name: str) -> str | None:
    """Extract content between section headers."""
    # Match section header with optional markdown formatting
    pattern = rf'(?:^|\n)(?:#+ *)?{re.escape(section_name)}[:\s]*\n(.*?)(?=\n(?:#+ |\Z))'
    match = re.search(pattern, text, re.IGNORECASE | re.DOTALL)
    return match.group(1).strip() if match else None

Structured Field Extraction

def extract_field(text: str, field_name: str) -> str | None:
    """Extract value after field label."""
    # Handle: "Field: value" or "Field - value" or "**Field**: value"
    pattern = rf'(?:\*\*)?{re.escape(field_name)}(?:\*\*)?[\s:\-]+([^\n]+)'
    match = re.search(pattern, text, re.IGNORECASE)
    return match.group(1).strip() if match else None

List Extraction

def extract_list_items(text: str, section_name: str) -> list[str]:
    """Extract bullet points from a section."""
    section = extract_section(text, section_name)
    if not section:
        return []
    
    # Match lines starting with -, *, or numbered
    items = re.findall(r'^[\-\*\d\.]+\s*(.+)$', section, re.MULTILINE)
    return [item.strip() for item in items]

Score Extraction with Validation

def extract_score(text: str, field_name: str, min_val: int, max_val: int) -> int | None:
    """Extract and validate numeric score."""
    raw = extract_field(text, field_name)
    if not raw:
        return None
    
    # Extract first number from the value
    match = re.search(r'\d+', raw)
    if not match:
        return None
    
    score = int(match.group())
    return max(min_val, min(max_val, score))  # Clamp to valid range

Graceful Degradation

@dataclass
class ParseResult:
    summary: str = ""
    score: int | None = None
    items: list[str] = field(default_factory=list)
    parse_errors: list[str] = field(default_factory=list)

def parse_response(text: str) -> ParseResult:
    """Parse LLM response with graceful error handling."""
    result = ParseResult()
    
    # Try each field, log errors but continue
    try:
        result.summary = extract_section(text, "Summary") or ""
    except Exception as e:
        result.parse_errors.append(f"Summary extraction failed: {e}")
    
    try:
        result.score = extract_score(text, "Rating", 1, 10)
    except Exception as e:
        result.parse_errors.append(f"Score extraction failed: {e}")
    
    try:
        result.items = extract_list_items(text, "Analysis")
    except Exception as e:
        result.parse_errors.append(f"Items extraction failed: {e}")
    
    return result

Error Handling Patterns

Retry with Exponential Backoff

import time
from functools import wraps

def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
    """Retry decorator with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        delay = base_delay * (2 ** attempt)
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

Error Logging Pattern

import json
from datetime import datetime

def log_error(item_dir: Path, stage: str, error: str, context: dict = None):
    """Log error to file for later analysis."""
    error_file = item_dir / "errors.jsonl"
    
    error_record = {
        "timestamp": datetime.now().isoformat(),
        "stage": stage,
        "error": error,
        "context": context or {},
    }
    
    with open(error_file, "a") as f:
        f.write(json.dumps(error_record) + "\n")

Partial Success Handling

def process_batch_with_partial_success(items: list) -> tuple[list, list]:
    """Process batch, separating successes from failures."""
    successes = []
    failures = []
    
    for item in items:
        try:
            result = process_item(item)
            successes.append((item, result))
        except Exception as e:
            failures.append((item, str(e)))
            log_error(item.directory, "process", str(e))
    
    # Report summary
    print(f"Processed {len(items)} items: {len(successes)} succeeded, {len(failures)} failed")
    
    return successes, failures

Cost Estimation Patterns

Token Counting

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    """Count tokens for cost estimation."""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    
    return len(encoding.encode(text))

def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    input_price_per_mtok: float,
    output_price_per_mtok: float,
) -> float:
    """Estimate cost in dollars."""
    input_cost = (input_tokens / 1_000_000) * input_price_per_mtok
    output_cost = (output_tokens / 1_000_000) * output_price_per_mtok
    return input_cost + output_cost

Batch Cost Estimation

def estimate_batch_cost(
    items: list,
    prompt_template: str,
    avg_output_tokens: int = 1000,
    model_pricing: dict = None,
) -> dict:
    """Estimate total cost for a batch."""
    model_pricing = model_pricing or {
        "input_price_per_mtok": 3.00,   # Example: GPT-4 Turbo input
        "output_price_per_mtok": 15.00,  # Example: GPT-4 Turbo output
    }
    
    total_input_tokens = 0
    for item in items:
        prompt = format_prompt(prompt_template, item)
        total_input_tokens += count_tokens(prompt)
    
    total_output_tokens = len(items) * avg_output_tokens
    
    estimated_cost = estimate_cost(
        total_input_tokens,
        total_output_tokens,
        **model_pricing,
    )
    
    return {
        "item_count": len(items),
        "total_input_tokens": total_input_tokens,
        "total_output_tokens": total_output_tokens,
        "estimated_cost_usd": estimated_cost,
        "avg_input_tokens_per_item": total_input_tokens / len(items),
        "cost_per_item_usd": estimated_cost / len(items),
    }

CLI Pattern

Standard CLI Structure

import argparse
from datetime import date

def main():
    parser = argparse.ArgumentParser(description="LLM Processing Pipeline")
    
    parser.add_argument(
        "stage",
        choices=["acquire", "prepare", "process", "parse", "render", "all", "clean"],
        help="Pipeline stage to run",
    )
    parser.add_argument(
        "--batch-id",
        default=None,
        help="Batch identifier (default: today's date)",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=None,
        help="Limit number of items (for testing)",
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=10,
        help="Number of parallel workers for processing",
    )
    parser.add_argument(
        "--model",
        default="gpt-4-turbo",
        help="Model to use for processing",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Estimate costs without processing",
    )
    parser.add_argument(
        "--clean-stage",
        choices=["acquire", "prepare", "process", "parse"],
        help="For clean: only clean this stage and downstream",
    )
    
    args = parser.parse_args()
    
    batch_id = args.batch_id or date.today().isoformat()
    
    if args.stage == "clean":
        stage_clean(batch_id, args.clean_stage)
    elif args.dry_run:
        estimate_costs(batch_id, args.limit)
    else:
        run_pipeline(batch_id, args.stage, args.limit, args.workers, args.model)

if __name__ == "__main__":
    main()

Rendering Patterns

Static HTML Output

import html
import json

def render_html(data: list[dict], output_path: Path, template: str):
    """Render data to static HTML file."""
    # Escape data for JavaScript embedding
    data_json = json.dumps([
        {k: html.escape(str(v)) if isinstance(v, str) else v 
         for k, v in item.items()}
        for item in data
    ])
    
    html_content = template.replace("{{DATA_JSON}}", data_json)
    
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        f.write(html_content)

Incremental Output

def render_incremental(items: list, output_dir: Path):
    """Render each item as it completes, plus index."""
    output_dir.mkdir(parents=True, exist_ok=True)
    
    # Render individual item pages
    for item in items:
        item_html = render_item(item)
        item_path = output_dir / f"{item.id}.html"
        with open(item_path, "w") as f:
            f.write(item_html)
    
    # Render index linking to all items
    index_html = render_index(items)
    with open(output_dir / "index.html", "w") as f:
        f.write(index_html)

Checkpoint and Resume Pattern

For long-running pipelines:

import json
from pathlib import Path

class PipelineCheckpoint:
    def __init__(self, checkpoint_file: Path):
        self.checkpoint_file = checkpoint_file
        self.state = self._load()
    
    def _load(self) -> dict:
        if self.checkpoint_file.exists():
            with open(self.checkpoint_file) as f:
                return json.load(f)
        return {"completed": [], "failed": [], "last_item": None}
    
    def save(self):
        with open(self.checkpoint_file, "w") as f:
            json.dump(self.state, f, indent=2)
    
    def mark_complete(self, item_id: str):
        self.state["completed"].append(item_id)
        self.state["last_item"] = item_id
        self.save()
    
    def mark_failed(self, item_id: str, error: str):
        self.state["failed"].append({"id": item_id, "error": error})
        self.save()
    
    def get_remaining(self, all_items: list[str]) -> list[str]:
        completed = set(self.state["completed"])
        return [item for item in all_items if item not in completed]

Testing Patterns

Stage Unit Tests

def test_prepare_stage():
    """Test prompt generation independently."""
    test_item = {"id": "test", "content": "Sample content"}
    prompt = prepare_prompt(test_item)
    
    assert "Sample content" in prompt
    assert "## Section 1" in prompt  # Format markers present

def test_parse_stage():
    """Test parsing with known good output."""
    test_response = """
    ## Summary
    This is a test summary.
    
    ## Score
    Rating: 7
    """
    
    result = parse_response(test_response)
    assert result.summary == "This is a test summary."
    assert result.score == 7

def test_parse_stage_malformed():
    """Test parsing handles malformed output."""
    test_response = "Some random text without sections"
    
    result = parse_response(test_response)
    assert result.summary == ""
    assert result.score is None
    assert len(result.parse_errors) > 0

Integration Test Pattern

def test_pipeline_end_to_end():
    """Test full pipeline with single item."""
    test_dir = Path("test_data")
    test_item = create_test_item()
    
    try:
        # Run each stage
        acquire_result = stage_acquire(test_dir, [test_item])
        assert (test_dir / test_item.id / "raw.json").exists()
        
        prepare_result = stage_prepare(test_dir)
        assert (test_dir / test_item.id / "prompt.md").exists()
        
        # Skip process stage in unit tests (costs money)
        # Create mock response instead
        mock_response(test_dir / test_item.id)
        
        parse_result = stage_parse(test_dir)
        assert (test_dir / test_item.id / "parsed.json").exists()
        
    finally:
        # Cleanup
        shutil.rmtree(test_dir, ignore_errors=True)