llm-project-development

# LLM Project Development Skill Build **production LLM applications** using proven methodology from Karpathy's HN Time Capsule, Vercel d0, Manus, and Anthropic's research. **Core principle**: Validate manually first, then build deterministic pipelines around the non-deterministic LLM core. ## Supporting Documentation All files under 500 lines per Anthropic best practices: - **[references/](references/)** - Methodology foundations - [case-studies.md](references/case-studies.md) - Karpathy, Vercel d0, Manus patterns - [pipeline-patterns.md](references/pipeline-patterns.md) - Python/TypeScript code patterns - [INDEX.md](references/INDEX.md) - Reference navigation - **[examples/](examples/)** - Grey Haven implementations - [tanstack-pipeline.md](examples/tanstack-pipeline.md) - TanStack Start example - [fastapi-pipeline.md](examples/fastapi-pipeline.md) - FastAPI backend example - [INDEX.md](examples/INDEX.md) - Examples navigation - **[templates/](templates/)** - Copy-paste starters - [pipeline-template.ts](templates/pipeline-template.ts) - TypeScript pipeline - [pipeline-template.py](templates/pipeline-template.py) - Python pipeline - **[checklists/](checklists/)** - Validation - [llm-project-checklist.md](checklists/llm-project-checklist.md) - Pre-launch checklist ## The Methodology ### Phase 1: Task-Model Fit Analysis **Before writing any code, determine if LLMs are the right tool.** #### LLM-Suited Tasks | Characteristic | Why LLMs Excel | Grey Haven Example | |----------------|----------------|-------------------| | **Synthesis over precision** | Combining context, not calculating | Summarizing tenant activity | | **Subjective judgment** | No single correct answer | Categorizing support tickets | | **Error tolerance** | Graceful degradation acceptable | Content recommendations | | **Human-like processing** | Natural language understanding | Chat-based tenant onboarding | | **Creative output** | Novel combinations required | Generating marketing copy | #### LLM-Unsuited Tasks (Use Traditional Code) | Characteristic | Why LLMs Fail | Better Approach | |----------------|---------------|-----------------| | **Precise computation** | Math errors, hallucinations | SQL queries, Python math | | **Real-time requirements** | Latency too high | Pre-computed indices | | **Deterministic output** | Need exact same result | Database lookups | | **Structured data lookup** | LLMs guess, don't retrieve | Drizzle/SQLModel queries | | **High-frequency calls** | Cost explodes | Caching, batching | #### The Manual Prototype Step **CRITICAL**: Before building automation, validate with the target model manually. ```markdown ## Manual Validation Checklist - [ ] Copy ONE real example into the LLM UI - [ ] Test with the EXACT model you'll use in production - [ ] Verify output quality meets requirements - [ ] Note edge cases and failure modes - [ ] Estimate cost per operation ## Example: Karpathy's Approach 1. Took ONE Hacker News thread 2. Pasted into ChatGPT with analysis prompt 3. Confirmed Opus 4.5 could do the task 4. THEN built automation pipeline ``` ### Phase 2: Pipeline Architecture **Design principle**: Deterministic stages wrapping one non-deterministic core. ``` ┌─────────────────────────────────────────────────────────────────┐ │ DETERMINISTIC │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ ACQUIRE │ → │ PREPARE │ → │ PROCESS │ → │ RENDER │ │ │ │ (fetch) │ │ (format) │ │ (LLM) │ │ (output) │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ ↑ ↑ ↑ ↑ │ │ Deterministic Deterministic NON-DETERMINISTIC Deterministic│ │ (retry safe) (retry safe) (cache results) (retry safe) │ └─────────────────────────────────────────────────────────────────┘ ``` #### Stage Details | Stage | Purpose | Grey Haven Implementation | |-------|---------|--------------------------| | **Acquire** | Get raw data | Drizzle queries, Firecrawl scraping, API calls | | **Prepare** | Format for LLM | Jinja templates, TypeScript string builders | | **Process** | LLM inference | Anthropic SDK, structured outputs | | **Parse** | Extract from response | Zod schemas, Pydantic models | | **Render** | Final output | React components, markdown, JSON | #### TypeScript Pipeline Example (TanStack Start) ```typescript // lib/pipelines/content-analyzer.ts import Anthropic from "@anthropic-ai/sdk"; import { z } from "zod"; import { existsSync, mkdirSync, writeFileSync, readFileSync } from "fs"; import { join } from "path"; // Stage 1: Schema definition const AnalysisSchema = z.object({ summary: z.string(), sentiment: z.enum(["positive", "neutral", "negative"]), topics: z.array(z.string()), action_items: z.array(z.string()), }); type Analysis = z.infer<typeof AnalysisSchema>; // Stage 2: Acquire - Get data from database async function acquire(tenant_id: string, content_id: string) { const content = await db.query.contents.findFirst({ where: and( eq(contents.tenant_id, tenant_id), eq(contents.id, content_id) ), }); if (!content) throw new Error(`Content ${content_id} not found`); return content; } // Stage 3: Prepare - Format prompt function prepare(content: Content): string { return `Analyze this content and provide structured output. CONTENT: ${content.body} Respond with JSON matching this schema: { "summary": "2-3 sentence summary", "sentiment": "positive" | "neutral" | "negative", "topics": ["topic1", "topic2"], "action_items": ["action1", "action2"] }`; } // Stage 4: Process - LLM call with caching async function process( prompt: string, cacheDir: string, cacheKey: string ): Promise<string> { const cachePath = join(cacheDir, `${cacheKey}.json`); // Check cache first if (existsSync(cachePath)) { return JSON.parse(readFileSync(cachePath, "utf-8")).response; } const client = new Anthropic(); const response = await client.messages.create({ model: "claude-sonnet-4-20250514", max_tokens: 1024, messages: [{ role: "user", content: prompt }], }); const text = response.content[0].type === "text" ? response.content[0].text : ""; // Cache result mkdirSync(cacheDir, { recursive: true }); writeFileSync(cachePath, JSON.stringify({ response: text, timestamp: new Date().toISOString() })); return text; } // Stage 5: Parse - Validate with Zod function parse(response: string): Analysis { const jsonMatch = response.match(/\{[\s\S]*\}/); if (!jsonMatch) throw new Error("No JSON found in response"); const parsed = JSON.parse(jsonMatch[0]); return AnalysisSchema.parse(parsed); } // Stage 6: Render - Save to database async function render( tenant_id: string, content_id: string, analysis: Analysis ) { await db.update(contents) .set({ analysis_summary: analysis.summary, analysis_sentiment: analysis.sentiment, analysis_topics: analysis.topics, updated_at: new Date(), }) .where(and( eq(contents.tenant_id, tenant_id), eq(contents.id, content_id) )); return analysis; } // Main pipeline function export async function analyzeContent( tenant_id: string, content_id: string ): Promise<Analysis> { const cacheDir = join(process.cwd(), ".cache", "analyses", tenant_id); const content = await acquire(tenant_id, content_id); const prompt = prepare(content); const response = await process(prompt, cacheDir, content_id); const analysis = parse(response); await render(tenant_id, content_id, analysis); return analysis; } ``` #### Python Pipeline Example (FastAPI) ```python # app/pipelines/content_analyzer.py from pathlib import Path from pydantic import BaseModel from anthropic import Anthropic import json class Analysis(BaseModel): summary: str sentiment: str # positive | neutral | negative topics: list[str] action_items: list[str] class ContentAnalyzerPipeline: def __init__(self, tenant_id: str, cache_dir: Path | None = None): self.tenant_id = tenant_id self.cache_dir = cache_dir or Path(".cache/analyses") / tenant_id self.client = Anthropic() async def acquire(self, content_id: str, db: AsyncSession) -> Content: """Stage 1: Get content from database.""" result = await db.execute( select(Content).where( Content.tenant_id == self.tenant_id, Content.id == content_id ) ) content = result.scalar_one_or_none() if not content: raise ValueError(f"Content {content_id} not found") return content def prepare(self, content: Content) -> str: """Stage 2: Format prompt.""" return f"""Analyze this content and provide structured output. CONTENT: {content.body} Respond with JSON: {{ "summary": "2-3 sentence summary", "sentiment": "positive" | "neutral" | "negative", "topics": ["topic1", "topic2"], "action_items": ["action1"] }}""" async def process(self, prompt: str, cache_key: str) -> str: """Stage 3: LLM call with file-based caching.""" cache_path = self.cache_dir / f"{cache_key}.json" # Check cache if cache_path.exists(): return json.loads(cache_path.read_text())["response"] response = self.client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[{"role": "user", "content": prompt}] ) text = response.content[0].text # Cache result self.cache_dir.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps({ "response": text, "timestamp": datetime.utcnow().isoformat() })) return text def parse(self, response: str) -> Analysis: """Stage 4: Parse and validate with Pydantic.""" import re match = re.search(r'\{[\s\S]*\}', response) if not match: raise ValueError("No JSON found in response") return Analysis.model_validate_json(match.group()) async def render( self, content_id: str, analysis: Analysis, db: AsyncSession ) -> Analysis: """Stage 5: Save to database.""" await db.execute( update(Content) .where( Content.tenant_id == self.tenant_id, Content.id == content_id ) .values( analysis_summary=analysis.summary, analysis_sentiment=analysis.sentiment, analysis_topics=analysis.topics, updated_at=datetime.utcnow() ) ) await db.commit() return analysis async def run(self, content_id: str, db: AsyncSession) -> Analysis: """Execute full pipeline.""" content = await self.acquire(content_id, db) prompt = self.prepare(content) response = await self.process(prompt, content_id) analysis = self.parse(response) return await self.render(content_id, analysis, db) ``` ### Phase 3: File System as State Machine **Key insight from Karpathy**: File existence determines work state. ```python # Pipeline state management def get_pipeline_state(work_dir: Path, item_id: str) -> str: """Determine pipeline state from file existence.""" item_dir = work_dir / item_id if not item_dir.exists(): return "pending" if not (item_dir / "raw.json").exists(): return "acquired" if not (item_dir / "prepared.txt").exists(): return "prepared" if not (item_dir / "response.json").exists(): return "processed" if not (item_dir / "analysis.json").exists(): return "parsed" return "complete"