Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

17 NODES
πŸ“textβ€’6 months ago

Evaluation and Regression Gate

Create evaluation runs, submit results, compute metrics, save baselines, and triage regressions before shipping changes.

analysis
⭐1
# Evaluation and Regression Gate Imported from curated first-party documentation sources. ## What this covers Use this workflow to formalize benchmarking, regression detection, and baseline management around agent or product changes. ## Use this when - Validating changes against prior baselines - Making regressions visible before rollout - Capturing repeatable evaluation evidence ## Expected outcomes - Evaluation runs produce comparable metrics - Baselines are stored after successful validation - Regression triage becomes a workflow instead of an afterthought ## Source synthesis - AGENT33/docs/functionality-and-workflows.md (https://github.com/mattmre/AGENT33/blob/main/docs/functionality-and-workflows.md) - AGENT33/docs/use-cases.md (https://github.com/mattmre/AGENT33/blob/main/docs/use-cases.md) ## Dedupe notes Combines AGENT33 evaluation lifecycle definitions with evaluation and regression use-case framing. ## Source excerpts ### AGENT33/docs/functionality-and-workflows.md ### 4.3 Evaluation Lifecycle Flow: 1. Create run (`/v1/evaluations/runs`) 2. Submit task results (`/runs/{id}/results`) 3. Compute metrics + gate report 4. Save baseline (`/runs/{id}/baseline`) 5. Triage/resolve regressions ### AGENT33/docs/use-cases.md ## 4. Evaluation and Regression Gates Goal: - Quantify quality and block regressions across PR/merge/release gates. Use these modules: - `api/routes/evaluations.py` - `evaluation/service.py` - `evaluation/gates.py` - `evaluation/regression.py` Typical flow: 1. Create run for gate type (`G-PR`, `G-MRG`, `G-REL`, `G-MON`). 2. Submit task results and quality metadata. 3. Compute metrics and gate verdict. 4. Save baseline for future comparison. 5. Triage/resolve regression records. Best fit: - Teams with golden-task style quality gates.
πŸ‘0
πŸ‘οΈ0
docs
πŸ“textβ€’6 months ago

Autonomy Budget Lifecycle

Define, approve, enforce, and escalate autonomous execution budgets before agents act on files, commands, or network requests.

productivity
⭐1
# Autonomy Budget Lifecycle Imported from curated first-party documentation sources. ## What this covers Use this lifecycle when you need bounded autonomous execution with approvals, preflight checks, and runtime enforcement. ## Use this when - Running agents with scoped permissions - Escalating risky actions instead of silently proceeding - Adding governance to autonomous sessions ## Expected outcomes - Autonomy state moves through draft, approval, and active phases - Preflight checks happen before execution begins - Escalations are tracked instead of disappearing into logs ## Source synthesis - AGENT33/docs/functionality-and-workflows.md (https://github.com/mattmre/AGENT33/blob/main/docs/functionality-and-workflows.md) - AGENT33/docs/use-cases.md (https://github.com/mattmre/AGENT33/blob/main/docs/use-cases.md) ## Dedupe notes Collapses autonomy lifecycle and use-case material into a single execution-governance entry. ## Source excerpts ### AGENT33/docs/functionality-and-workflows.md ### 4.4 Autonomy Budget Lifecycle States: - `draft -> pending_approval -> active -> suspended|expired|completed` Flow: 1. Create budget 2. Activate or transition 3. Run preflight checks 4. Create enforcer 5. Evaluate command/file/network requests 6. Track escalations ### AGENT33/docs/use-cases.md ## 3. Autonomous Execution Budgeting Goal: - Enforce hard runtime limits for file, command, and network activity. Use these modules: - `api/routes/autonomy.py` - `autonomy/service.py` - `autonomy/enforcement.py` - `autonomy/preflight.py` Typical flow: 1. Create and activate budget. 2. Run preflight checks. 3. Attach runtime enforcer. 4. Gate each file/command/network action through enforcement APIs. 5. Trigger and resolve escalations. Best fit: - High-control automation in regulated or sensitive environments.
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–system promptβ€’7 months ago

postmortem-writing

Write effective blameless postmortems with root cause analysis,

coding
⭐1
# Postmortem Writing Comprehensive guide to writing effective, blameless postmortems that drive organizational learning and prevent incident recurrence. ## When to Use This Skill - Conducting post-incident reviews - Writing postmortem documents - Facilitating blameless postmortem meetings - Identifying root causes and contributing factors - Creating actionable follow-up items - Building organizational learning culture ## Core Concepts ### 1. Blameless Culture | Blame-Focused | Blameless | | ------------------------ | --------------------------------- | | "Who caused this?" | "What conditions allowed this?" | | "Someone made a mistake" | "The system allowed this mistake" | | Punish individuals | Improve systems | | Hide information | Share learnings | | Fear of speaking up | Psychological safety | ### 2. Postmortem Triggers - SEV1 or SEV2 incidents - Customer-facing outages > 15 minutes - Data loss or security incidents - Near-misses that could have been severe - Novel failure modes - Incidents requiring unusual intervention ## Quick Start ### Postmortem Timeline ``` Day 0: Incident occurs Day 1-2: Draft postmortem document Day 3-5: Postmortem meeting Day 5-7: Finalize document, create tickets Week 2+: Action item completion Quarterly: Review patterns across incidents ``` ## Templates ### Template 1: Standard Postmortem ```markdown # Postmortem: [Incident Title] **Date**: 2024-01-15 **Authors**: @alice, @bob **Status**: Draft | In Review | Final **Incident Severity**: SEV2 **Incident Duration**: 47 minutes ## Executive Summary On January 15, 2024, the payment processing service experienced a 47-minute outage affecting approximately 12,000 customers. The root cause was a database connection pool exhaustion triggered by a configuration change in deployment v2.3.4. The incident was resolved by rolling back to v2.3.3 and increasing connection pool limits. **Impact**: - 12,000 customers unable to complete purchases - Estimated revenue loss: $45,000 - 847 support tickets created - No data loss or security implications ## Timeline (All times UTC) | Time | Event | | ----- | ----------------------------------------------- | | 14:23 | Deployment v2.3.4 completed to production | | 14:31 | First alert: `payment_error_rate > 5%` | | 14:33 | On-call engineer @alice acknowledges alert | | 14:35 | Initial investigation begins, error rate at 23% | | 14:41 | Incident declared SEV2, @bob joins | | 14:45 | Database connection exhaustion identified | | 14:52 | Decision to rollback deployment | | 14:58 | Rollback to v2.3.3 initiated | | 15:10 | Rollback complete, error rate dropping | | 15:18 | Service fully recovered, incident resolved | ## Root Cause Analysis ### What Happened The v2.3.4 deployment included a change to the database query pattern that inadvertently removed connection pooling for a frequently-called endpoint. Each request opened a new database connection instead of reusing pooled connections. ### Why It Happened 1. **Proximate Cause**: Code change in `PaymentRepository.java` replaced pooled `DataSource` with direct `DriverManager.getConnection()` calls. 2. **Contributing Factors**: - Code review did not catch the connection handling change - No integration tests specifically for connection pool behavior - Staging environment has lower traffic, masking the issue - Database connection metrics alert threshold was too high (90%) 3. **5 Whys Analysis**: - Why did the service fail? β†’ Database connections exhausted - Why were connections exhausted? β†’ Each request opened new connection - Why did each request open new connection? β†’ Code bypassed connection pool - Why did code bypass connection pool? β†’ Developer unfamiliar with codebase patterns - Why was developer unfamiliar? β†’ No documentation on connection management patterns ### System Diagram ``` [Client] β†’ [Load Balancer] β†’ [Payment Service] β†’ [Database] ↓ Connection Pool (broken) ↓ Direct connections (cause) ``` ## Detection ### What Worked - Error rate alert fired within 8 minutes of deployment - Grafana dashboard clearly showed connection spike - On-call response was swift (2 minute acknowledgment) ### What Didn't Work - Database connection metric alert threshold too high - No deployment-correlated alerting - Canary deployment would have caught this earlier ### Detection Gap The deployment completed at 14:23, but the first alert didn't fire until 14:31 (8 minutes). A deployment-aware alert could have detected the issue faster. ## Response ### What Worked - On-call engineer quickly identified database as the issue - Rollback decision was made decisively - Clear communication in incident channel ### What Could Be Improved - Took 10 minutes to correlate issue with recent deployment - Had to manually check deployment history - Rollback took 12 minutes (could be faster) ## Impact ### Customer Impact - 12,000 unique customers affected - Average impact duration: 35 minutes - 847 support tickets (23% of affected users) - Customer satisfaction score dropped 12 points ### Business Impact - Estimated revenue loss: $45,000 - Support cost: ~$2,500 (agent time) - Engineering time: ~8 person-hours ### Technical Impact - Database primary experienced elevated load - Some replica lag during incident - No permanent damage to systems ## Lessons Learned ### What Went Well 1. Alerting detected the issue before customer reports 2. Team collaborated effectively under pressure 3. Rollback procedure worked smoothly 4. Communication was clear and timely ### What Went Wrong 1. Code review missed critical change 2. Test coverage gap for connection pooling 3. Staging environment doesn't reflect production traffic 4. Alert thresholds were not tuned properly ### Where We Got Lucky 1. Incident occurred during business hours with full team available 2. Database handled the load without failing completely 3. No other incidents occurred simultaneously ## Action Items | Priority | Action | Owner | Due Date | Ticket | |----------|--------|-------|----------|--------| | P0 | Add integration test for connection pool behavior | @alice | 2024-01-22 | ENG-1234 | | P0 | Lower database connection alert threshold to 70% | @bob | 2024-01-17 | OPS-567 | | P1 | Document connection management patterns | @alice | 2024-01-29 | DOC-89 | | P1 | Implement deployment-correlated alerting | @bob | 2024-02-05 | OPS-568 | | P2 | Evaluate canary deployment strategy | @charlie | 2024-02-15 | ENG-1235 | | P2 | Load test staging with production-like traffic | @dave | 2024-02-28 | QA-123 | ## Appendix ### Supporting Data #### Error Rate Graph [Link to Grafana dashboard snapshot] #### Database Connection Graph [Link to metrics] ### Related Incidents - 2023-11-02: Similar connection issue in User Service (POSTMORTEM-42) ### References - [Connection Pool Best Practices](internal-wiki/connection-pools) - [Deployment Runbook](internal-wiki/deployment-runbook) ``` ### Template 2: 5 Whys Analysis ```markdown # 5 Whys Analysis: [Incident] ## Problem Statement Payment service experienced 47-minute outage due to database connection exhaustion. ## Analysis ### Why #1: Why did the service fail? **Answer**: Database connections were exhausted, causing all new requests to fail. **Evidence**: Metrics showed connection count at 100/100 (max), with 500+ pending requests. --- ### Why #2: Why were database connections exhausted? **Answer**: Each incoming request opened a new database connection instead of using the connection pool. **Evidence**: Code diff shows direct `DriverManager.getConnection()` instead of pooled `DataSource`. --- ### Why #3: Why did the code bypass the connection pool? **Answer**: A developer refactored the repository class and inadvertently changed the connection acquisition method. **Evidence**: PR #1234 shows the change, made while fixing a different bug. --- ### Why #4: Why wasn't this caught in code review? **Answer**: The reviewer focused on the functional change (the bug fix) and didn't notice the infrastructure change. **Evidence**: Review comments only discuss business logic. --- ### Why #5: Why isn't there a safety net for this type of change? **Answer**: We lack automated tests that verify connection pool behavior and lack documentation about our connection patterns. **Evidence**: Test suite has no tests for connection handling; wiki has no article on database connections. ## Root Causes Identified 1. **Primary**: Missing automated tests for infrastructure behavior 2. **Secondary**: Insufficient documentation of architectural patterns 3. **Tertiary**: Code review checklist doesn't include infrastructure considerations ## Systemic Improvements | Root Cause | Improvement | Type | | ------------- | --------------------------------- | ---------- | | Missing tests | Add infrastructure behavior tests | Prevention | | Missing docs | Document connection patterns | Prevention | | Review gaps | Update review checklist | Detection | | No canary | Implement canary deployments | Mitigation | ``` ### Template 3: Quick Postmortem (Minor Incidents) ```markdown # Quick Postmortem: [Brief Title] **Date**: 2024-01-15 | **Duration**: 12 min | **Severity**: SEV3 ## What Happened API latency spiked to 5s due to cache miss storm after cache flush. ## Timeline - 10:00 - Cache flush initiated for config update - 10:02 - Latency alerts fire - 10:05 - Identified as cache miss storm - 10:08 - Enabled cache warming - 10:12 - Latency normalized ## Root Cause Full cache flush for minor config update caused thundering herd. ## Fix - Immediate: Enabled cache warming - Long-term: Implement partial cache invalidation (ENG-999) ## Lessons Don't full-flush cache in production; use targeted invalidation. ``` ## Facilitation Guide ### Running a Postmortem Meeting ```markdown ## Meeting Structure (60 minutes) ### 1. Opening (5 min) - Remind everyone of blameless culture - "We're here to learn, not to blame" - Review meeting norms ### 2. Timeline Review (15 min) - Walk through events chronologically - Ask clarifying questions - Identify gaps in timeline ### 3. Analysis Discussion (20 min) - What failed? - Why did it fail? - What conditions allowed this? - What would have prevented it? ### 4. Action Items (15 min) - Brainstorm improvements - Prioritize by impact and effort - Assign owners and due dates ### 5. Closing (5 min) - Summarize key learnings - Confirm action item owners - Schedule follow-up if needed ## Facilitation Tips - Keep discussion on track - Redirect blame to systems - Encourage quiet participants - Document dissenting views - Time-box tangents ``` ## Anti-Patterns to Avoid | Anti-Pattern | Problem | Better Approach | | ----------------------- | -------------------------- | ------------------------------- | | **Blame game** | Shuts down learning | Focus on systems | | **Shallow analysis** | Doesn't prevent recurrence | Ask "why" 5 times | | **No action items** | Waste of time | Always have concrete next steps | | **Unrealistic actions** | Never completed | Scope to achievable tasks | | **No follow-up** | Actions forgotten | Track in ticketing system | ## Best Practices ### Do's - **Start immediately** - Memory fades fast - **Be specific** - Exact times, exact errors - **Include graphs** - Visual evidence - **Assign owners** - No orphan action items - **Share widely** - Organizational learning ### Don'ts - **Don't name and shame** - Ever - **Don't skip small incidents** - They reveal patterns - **Don't make it a blame doc** - That kills learning - **Don't create busywork** - Actions should be meaningful - **Don't skip follow-up** - Verify actions completed ## Resources - [Google SRE - Postmortem Culture](https://sre.google/sre-book/postmortem-culture/) - [Etsy's Blameless Postmortems](https://codeascraft.com/2012/05/22/blameless-postmortems/) - [PagerDuty Postmortem Guide](https://postmortems.pagerduty.com/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

modern-javascript-patterns

Master ES6+ features including async/await, destructuring, spread

coding
⭐1
# Modern JavaScript Patterns Comprehensive guide for mastering modern JavaScript (ES6+) features, functional programming patterns, and best practices for writing clean, maintainable, and performant code. ## When to Use This Skill - Refactoring legacy JavaScript to modern syntax - Implementing functional programming patterns - Optimizing JavaScript performance - Writing maintainable and readable code - Working with asynchronous operations - Building modern web applications - Migrating from callbacks to Promises/async-await - Implementing data transformation pipelines ## ES6+ Core Features ### 1. Arrow Functions **Syntax and Use Cases:** ```javascript // Traditional function function add(a, b) { return a + b; } // Arrow function const add = (a, b) => a + b; // Single parameter (parentheses optional) const double = (x) => x * 2; // No parameters const getRandom = () => Math.random(); // Multiple statements (need curly braces) const processUser = (user) => { const normalized = user.name.toLowerCase(); return { ...user, name: normalized }; }; // Returning objects (wrap in parentheses) const createUser = (name, age) => ({ name, age }); ``` **Lexical 'this' Binding:** ```javascript class Counter { constructor() { this.count = 0; } // Arrow function preserves 'this' context increment = () => { this.count++; }; // Traditional function loses 'this' in callbacks incrementTraditional() { setTimeout(function () { this.count++; // 'this' is undefined }, 1000); } // Arrow function maintains 'this' incrementArrow() { setTimeout(() => { this.count++; // 'this' refers to Counter instance }, 1000); } } ``` ### 2. Destructuring **Object Destructuring:** ```javascript const user = { id: 1, name: "John Doe", email: "john@example.com", address: { city: "New York", country: "USA", }, }; // Basic destructuring const { name, email } = user; // Rename variables const { name: userName, email: userEmail } = user; // Default values const { age = 25 } = user; // Nested destructuring const { address: { city, country }, } = user; // Rest operator const { id, ...userWithoutId } = user; // Function parameters function greet({ name, age = 18 }) { console.log(`Hello ${name}, you are ${age}`); } greet(user); ``` **Array Destructuring:** ```javascript const numbers = [1, 2, 3, 4, 5]; // Basic destructuring const [first, second] = numbers; // Skip elements const [, , third] = numbers; // Rest operator const [head, ...tail] = numbers; // Swapping variables let a = 1, b = 2; [a, b] = [b, a]; // Function return values function getCoordinates() { return [10, 20]; } const [x, y] = getCoordinates(); // Default values const [one, two, three = 0] = [1, 2]; ``` ### 3. Spread and Rest Operators **Spread Operator:** ```javascript // Array spreading const arr1 = [1, 2, 3]; const arr2 = [4, 5, 6]; const combined = [...arr1, ...arr2]; // Object spreading const defaults = { theme: "dark", lang: "en" }; const userPrefs = { theme: "light" }; const settings = { ...defaults, ...userPrefs }; // Function arguments const numbers = [1, 2, 3]; Math.max(...numbers); // Copying arrays/objects (shallow copy) const copy = [...arr1]; const objCopy = { ...user }; // Adding items immutably const newArr = [...arr1, 4, 5]; const newObj = { ...user, age: 30 }; ``` **Rest Parameters:** ```javascript // Collect function arguments function sum(...numbers) { return numbers.reduce((total, num) => total + num, 0); } sum(1, 2, 3, 4, 5); // With regular parameters function greet(greeting, ...names) { return `${greeting} ${names.join(", ")}`; } greet("Hello", "John", "Jane", "Bob"); // Object rest const { id, ...userData } = user; // Array rest const [first, ...rest] = [1, 2, 3, 4, 5]; ``` ### 4. Template Literals ```javascript // Basic usage const name = "John"; const greeting = `Hello, ${name}!`; // Multi-line strings const html = ` <div> <h1>${title}</h1> <p>${content}</p> </div> `; // Expression evaluation const price = 19.99; const total = `Total: $${(price * 1.2).toFixed(2)}`; // Tagged template literals function highlight(strings, ...values) { return strings.reduce((result, str, i) => { const value = values[i] || ""; return result + str + `<mark>${value}</mark>`; }, ""); } const name = "John"; const age = 30; const html = highlight`Name: ${name}, Age: ${age}`; // Output: "Name: <mark>John</mark>, Age: <mark>30</mark>" ``` ### 5. Enhanced Object Literals ```javascript const name = "John"; const age = 30; // Shorthand property names const user = { name, age }; // Shorthand method names const calculator = { add(a, b) { return a + b; }, subtract(a, b) { return a - b; }, }; // Computed property names const field = "email"; const user = { name: "John", [field]: "john@example.com", [`get${field.charAt(0).toUpperCase()}${field.slice(1)}`]() { return this[field]; }, }; // Dynamic property creation const createUser = (name, ...props) => { return props.reduce( (user, [key, value]) => ({ ...user, [key]: value, }), { name }, ); }; const user = createUser("John", ["age", 30], ["email", "john@example.com"]); ``` ## Asynchronous Patterns ### 1. Promises **Creating and Using Promises:** ```javascript // Creating a promise const fetchUser = (id) => { return new Promise((resolve, reject) => { setTimeout(() => { if (id > 0) { resolve({ id, name: "John" }); } else { reject(new Error("Invalid ID")); } }, 1000); }); }; // Using promises fetchUser(1) .then((user) => console.log(user)) .catch((error) => console.error(error)) .finally(() => console.log("Done")); // Chaining promises fetchUser(1) .then((user) => fetchUserPosts(user.id)) .then((posts) => processPosts(posts)) .then((result) => console.log(result)) .catch((error) => console.error(error)); ``` **Promise Combinators:** ```javascript // Promise.all - Wait for all promises const promises = [fetchUser(1), fetchUser(2), fetchUser(3)]; Promise.all(promises) .then((users) => console.log(users)) .catch((error) => console.error("At least one failed:", error)); // Promise.allSettled - Wait for all, regardless of outcome Promise.allSettled(promises).then((results) => { results.forEach((result) => { if (result.status === "fulfilled") { console.log("Success:", result.value); } else { console.log("Error:", result.reason); } }); }); // Promise.race - First to complete Promise.race(promises) .then((winner) => console.log("First:", winner)) .catch((error) => console.error(error)); // Promise.any - First to succeed Promise.any(promises) .then((first) => console.log("First success:", first)) .catch((error) => console.error("All failed:", error)); ``` ### 2. Async/Await **Basic Usage:** ```javascript // Async function always returns a Promise async function fetchUser(id) { const response = await fetch(`/api/users/${id}`); const user = await response.json(); return user; } // Error handling with try/catch async function getUserData(id) { try { const user = await fetchUser(id); const posts = await fetchUserPosts(user.id); return { user, posts }; } catch (error) { console.error("Error fetching data:", error); throw error; } } // Sequential vs Parallel execution async function sequential() { const user1 = await fetchUser(1); // Wait const user2 = await fetchUser(2); // Then wait return [user1, user2]; } async function parallel() { const [user1, user2] = await Promise.all([fetchUser(1), fetchUser(2)]); return [user1, user2]; } ``` **Advanced Patterns:** ```javascript // Async IIFE (async () => { const result = await someAsyncOperation(); console.log(result); })(); // Async iteration async function processUsers(userIds) { for (const id of userIds) { const user = await fetchUser(id); await processUser(user); } } // Top-level await (ES2022) const config = await fetch("/config.json").then((r) => r.json()); // Retry logic async function fetchWithRetry(url, retries = 3) { for (let i = 0; i < retries; i++) { try { return await fetch(url); } catch (error) { if (i === retries - 1) throw error; await new Promise((resolve) => setTimeout(resolve, 1000 * (i + 1))); } } } // Timeout wrapper async function withTimeout(promise, ms) { const timeout = new Promise((_, reject) => setTimeout(() => reject(new Error("Timeout")), ms), ); return Promise.race([promise, timeout]); } ``` ## Functional Programming Patterns ### 1. Array Methods **Map, Filter, Reduce:** ```javascript const users = [ { id: 1, name: "John", age: 30, active: true }, { id: 2, name: "Jane", age: 25, active: false }, { id: 3, name: "Bob", age: 35, active: true }, ]; // Map - Transform array const names = users.map((user) => user.name); const upperNames = users.map((user) => user.name.toUpperCase()); // Filter - Select elements const activeUsers = users.filter((user) => user.active); const adults = users.filter((user) => user.age >= 18); // Reduce - Aggregate data const totalAge = users.reduce((sum, user) => sum + user.age, 0); const avgAge = totalAge / users.length; // Group by property const byActive = users.reduce((groups, user) => { const key = user.active ? "active" : "inactive"; return { ...groups, [key]: [...(groups[key] || []), user], }; }, {}); // Chaining methods const result = users .filter((user) => user.active) .map((user) => user.name) .sort() .join(", "); ``` **Advanced Array Methods:** ```javascript // Find - First matching element const user = users.find((u) => u.id === 2); // FindIndex - Index of first match const index = users.findIndex((u) => u.name === "Jane"); // Some - At least one matches const hasActive = users.some((u) => u.active); // Every - All match const allAdults = users.every((u) => u.age >= 18); // FlatMap - Map and flatten const userTags = [ { name: "John", tags: ["admin", "user"] }, { name: "Jane", tags: ["user"] }, ]; const allTags = userTags.flatMap((u) => u.tags); // From - Create array from iterable const str = "hello"; const chars = Array.from(str); const numbers = Array.from({ length: 5 }, (_, i) => i + 1); // Of - Create array from arguments const arr = Array.of(1, 2, 3); ``` ### 2. Higher-Order Functions **Functions as Arguments:** ```javascript // Custom forEach function forEach(array, callback) { for (let i = 0; i < array.length; i++) { callback(array[i], i, array); } } // Custom map function map(array, transform) { const result = []; for (const item of array) { result.push(transform(item)); } return result; } // Custom filter function filter(array, predicate) { const result = []; for (const item of array) { if (predicate(item)) { result.push(item); } } return result; } ``` **Functions Returning Functions:** ```javascript // Currying const multiply = (a) => (b) => a * b; const double = multiply(2); const triple = multiply(3); console.log(double(5)); // 10 console.log(triple(5)); // 15 // Partial application function partial(fn, ...args) { return (...moreArgs) => fn(...args, ...moreArgs); } const add = (a, b, c) => a + b + c; const add5 = partial(add, 5); console.log(add5(3, 2)); // 10 // Memoization function memoize(fn) { const cache = new Map(); return (...args) => { const key = JSON.stringify(args); if (cache.has(key)) { return cache.get(key); } const result = fn(...args); cache.set(key, result); return result; }; } const fibonacci = memoize((n) => { if (n <= 1) return n; return fibonacci(n - 1) + fibonacci(n - 2); }); ``` ### 3. Composition and Piping ```javascript // Function composition const compose = (...fns) => (x) => fns.reduceRight((acc, fn) => fn(acc), x); const pipe = (...fns) => (x) => fns.reduce((acc, fn) => fn(acc), x); // Example usage const addOne = (x) => x + 1; const double = (x) => x * 2; const square = (x) => x * x; const composed = compose(square, double, addOne); console.log(composed(3)); // ((3 + 1) * 2)^2 = 64 const piped = pipe(addOne, double, square); console.log(piped(3)); // ((3 + 1) * 2)^2 = 64 // Practical example const processUser = pipe( (user) => ({ ...user, name: user.name.trim() }), (user) => ({ ...user, email: user.email.toLowerCase() }), (user) => ({ ...user, age: parseInt(user.age) }), ); const user = processUser({ name: " John ", email: "JOHN@EXAMPLE.COM", age: "30", }); ``` ### 4. Pure Functions and Immutability ```javascript // Impure function (modifies input) function addItemImpure(cart, item) { cart.items.push(item); cart.total += item.price; return cart; } // Pure function (no side effects) function addItemPure(cart, item) { return { ...cart, items: [...cart.items, item], total: cart.total + item.price, }; } // Immutable array operations const numbers = [1, 2, 3, 4, 5]; // Add to array const withSix = [...numbers, 6]; // Remove from array const withoutThree = numbers.filter((n) => n !== 3); // Update array element const doubled = numbers.map((n) => (n === 3 ? n * 2 : n)); // Immutable object operations const user = { name: "John", age: 30 }; // Update property const olderUser = { ...user, age: 31 }; // Add property const withEmail = { ...user, email: "john@example.com" }; // Remove property const { age, ...withoutAge } = user; // Deep cloning (simple approach) const deepClone = (obj) => JSON.parse(JSON.stringify(obj)); // Better deep cloning const structuredClone = (obj) => globalThis.structuredClone(obj); ``` ## Modern Class Features ```javascript // Class syntax class User { // Private fields #password; // Public fields id; name; // Static field static count = 0; constructor(id, name, password) { this.id = id; this.name = name; this.#password = password; User.count++; } // Public method greet() { return `Hello, ${this.name}`; } // Private method #hashPassword(password) { return `hashed_${password}`; } // Getter get displayName() { return this.name.toUpperCase(); } // Setter set password(newPassword) { this.#password = this.#hashPassword(newPassword); } // Static method static create(id, name, password) { return new User(id, name, password); } } // Inheritance class Admin extends User { constructor(id, name, password, role) { super(id, name, password); this.role = role; } greet() { return `${super.greet()}, I'm an admin`; } } ``` ## Modules (ES6) ```javascript // Exporting // math.js export const PI = 3.14159; export function add(a, b) { return a + b; } export class Calculator { // ... } // Default export export default function multiply(a, b) { return a * b; } // Importing // app.js import multiply, { PI, add, Calculator } from "./math.js"; // Rename imports import { add as sum } from "./math.js"; // Import all import * as Math from "./math.js"; // Dynamic imports const module = await import("./math.js"); const { add } = await import("./math.js"); // Conditional loading if (condition) { const module = await import("./feature.js"); module.init(); } ``` ## Iterators and Generators ```javascript // Custom iterator const range = { from: 1, to: 5, [Symbol.iterator]() { return { current: this.from, last: this.to, next() { if (this.current <= this.last) { return { done: false, value: this.current++ }; } else { return { done: true }; } }, }; }, }; for (const num of range) { console.log(num); // 1, 2, 3, 4, 5 } // Generator function function* rangeGenerator(from, to) { for (let i = from; i <= to; i++) { yield i; } } for (const num of rangeGenerator(1, 5)) { console.log(num); } // Infinite generator function* fibonacci() { let [prev, curr] = [0, 1]; while (true) { yield curr; [prev, curr] = [curr, prev + curr]; } } // Async generator async function* fetchPages(url) { let page = 1; while (true) { const response = await fetch(`${url}?page=${page}`); const data = await response.json(); if (data.length === 0) break; yield data; page++; } } for await (const page of fetchPages("/api/users")) { console.log(page); } ``` ## Modern Operators ```javascript // Optional chaining const user = { name: "John", address: { city: "NYC" } }; const city = user?.address?.city; const zipCode = user?.address?.zipCode; // undefined // Function call const result = obj.method?.(); // Array access const first = arr?.[0]; // Nullish coalescing const value = null ?? "default"; // 'default' const value = undefined ?? "default"; // 'default' const value = 0 ?? "default"; // 0 (not 'default') const value = "" ?? "default"; // '' (not 'default') // Logical assignment let a = null; a ??= "default"; // a = 'default' let b = 5; b ??= 10; // b = 5 (unchanged) let obj = { count: 0 }; obj.count ||= 1; // obj.count = 1 obj.count &&= 2; // obj.count = 2 ``` ## Performance Optimization ```javascript // Debounce function debounce(fn, delay) { let timeoutId; return (...args) => { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn(...args), delay); }; } const searchDebounced = debounce(search, 300); // Throttle function throttle(fn, limit) { let inThrottle; return (...args) => { if (!inThrottle) { fn(...args); inThrottle = true; setTimeout(() => (inThrottle = false), limit); } }; } const scrollThrottled = throttle(handleScroll, 100); // Lazy evaluation function* lazyMap(iterable, transform) { for (const item of iterable) { yield transform(item); } } // Use only what you need const numbers = [1, 2, 3, 4, 5]; const doubled = lazyMap(numbers, (x) => x * 2); const first = doubled.next().value; // Only computes first value ``` ## Best Practices 1. **Use const by default**: Only use let when reassignment is needed 2. **Prefer arrow functions**: Especially for callbacks 3. **Use template literals**: Instead of string concatenation 4. **Destructure objects and arrays**: For cleaner code 5. **Use async/await**: Instead of Promise chains 6. **Avoid mutating data**: Use spread operator and array methods 7. **Use optional chaining**: Prevent "Cannot read property of undefined" 8. **Use nullish coalescing**: For default values 9. **Prefer array methods**: Over traditional loops 10. **Use modules**: For better code organization 11. **Write pure functions**: Easier to test and reason about 12. **Use meaningful variable names**: Self-documenting code 13. **Keep functions small**: Single responsibility principle 14. **Handle errors properly**: Use try/catch with async/await 15. **Use strict mode**: `'use strict'` for better error catching ## Common Pitfalls 1. **this binding confusion**: Use arrow functions or bind() 2. **Async/await without error handling**: Always use try/catch 3. **Promise creation
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

embedding-strategies

Select and optimize embedding models for semantic search and RAG

coding
⭐1
# Embedding Strategies Guide to selecting and optimizing embedding models for vector search applications. ## When to Use This Skill - Choosing embedding models for RAG - Optimizing chunking strategies - Fine-tuning embeddings for domains - Comparing embedding model performance - Reducing embedding dimensions - Handling multilingual content ## Core Concepts ### 1. Embedding Model Comparison (2026) | Model | Dimensions | Max Tokens | Best For | | -------------------------- | ---------- | ---------- | ----------------------------------- | | **voyage-3-large** | 1024 | 32000 | Claude apps (Anthropic recommended) | | **voyage-3** | 1024 | 32000 | Claude apps, cost-effective | | **voyage-code-3** | 1024 | 32000 | Code search | | **voyage-finance-2** | 1024 | 32000 | Financial documents | | **voyage-law-2** | 1024 | 32000 | Legal documents | | **text-embedding-3-large** | 3072 | 8191 | OpenAI apps, high accuracy | | **text-embedding-3-small** | 1536 | 8191 | OpenAI apps, cost-effective | | **bge-large-en-v1.5** | 1024 | 512 | Open source, local deployment | | **all-MiniLM-L6-v2** | 384 | 256 | Fast, lightweight | | **multilingual-e5-large** | 1024 | 512 | Multi-language | ### 2. Embedding Pipeline ``` Document β†’ Chunking β†’ Preprocessing β†’ Embedding Model β†’ Vector ↓ [Overlap, Size] [Clean, Normalize] [API/Local] ``` ## Templates ### Template 1: Voyage AI Embeddings (Recommended for Claude) ```python from langchain_voyageai import VoyageAIEmbeddings from typing import List import os # Initialize Voyage AI embeddings (recommended by Anthropic for Claude) embeddings = VoyageAIEmbeddings( model="voyage-3-large", voyage_api_key=os.environ.get("VOYAGE_API_KEY") ) def get_embeddings(texts: List[str]) -> List[List[float]]: """Get embeddings from Voyage AI.""" return embeddings.embed_documents(texts) def get_query_embedding(query: str) -> List[float]: """Get single query embedding.""" return embeddings.embed_query(query) # Specialized models for domains code_embeddings = VoyageAIEmbeddings(model="voyage-code-3") finance_embeddings = VoyageAIEmbeddings(model="voyage-finance-2") legal_embeddings = VoyageAIEmbeddings(model="voyage-law-2") ``` ### Template 2: OpenAI Embeddings ```python from openai import OpenAI from typing import List import numpy as np client = OpenAI() def get_embeddings( texts: List[str], model: str = "text-embedding-3-small", dimensions: int = None ) -> List[List[float]]: """Get embeddings from OpenAI with optional dimension reduction.""" # Handle batching for large lists batch_size = 100 all_embeddings = [] for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] kwargs = {"input": batch, "model": model} if dimensions: # Matryoshka dimensionality reduction kwargs["dimensions"] = dimensions response = client.embeddings.create(**kwargs) embeddings = [item.embedding for item in response.data] all_embeddings.extend(embeddings) return all_embeddings def get_embedding(text: str, **kwargs) -> List[float]: """Get single embedding.""" return get_embeddings([text], **kwargs)[0] # Dimension reduction with Matryoshka embeddings def get_reduced_embedding(text: str, dimensions: int = 512) -> List[float]: """Get embedding with reduced dimensions (Matryoshka).""" return get_embedding( text, model="text-embedding-3-small", dimensions=dimensions ) ``` ### Template 3: Local Embeddings with Sentence Transformers ```python from sentence_transformers import SentenceTransformer from typing import List, Optional import numpy as np class LocalEmbedder: """Local embedding with sentence-transformers.""" def __init__( self, model_name: str = "BAAI/bge-large-en-v1.5", device: str = "cuda" ): self.model = SentenceTransformer(model_name, device=device) self.model_name = model_name def embed( self, texts: List[str], normalize: bool = True, show_progress: bool = False ) -> np.ndarray: """Embed texts with optional normalization.""" embeddings = self.model.encode( texts, normalize_embeddings=normalize, show_progress_bar=show_progress, convert_to_numpy=True ) return embeddings def embed_query(self, query: str) -> np.ndarray: """Embed a query with appropriate prefix for retrieval models.""" # BGE and similar models benefit from query prefix if "bge" in self.model_name.lower(): query = f"Represent this sentence for searching relevant passages: {query}" return self.embed([query])[0] def embed_documents(self, documents: List[str]) -> np.ndarray: """Embed documents for indexing.""" return self.embed(documents) # E5 model with instructions class E5Embedder: def __init__(self, model_name: str = "intfloat/multilingual-e5-large"): self.model = SentenceTransformer(model_name) def embed_query(self, query: str) -> np.ndarray: """E5 requires 'query:' prefix for queries.""" return self.model.encode(f"query: {query}") def embed_document(self, document: str) -> np.ndarray: """E5 requires 'passage:' prefix for documents.""" return self.model.encode(f"passage: {document}") ``` ### Template 4: Chunking Strategies ```python from typing import List, Tuple import re def chunk_by_tokens( text: str, chunk_size: int = 512, chunk_overlap: int = 50, tokenizer=None ) -> List[str]: """Chunk text by token count.""" import tiktoken tokenizer = tokenizer or tiktoken.get_encoding("cl100k_base") tokens = tokenizer.encode(text) chunks = [] start = 0 while start < len(tokens): end = start + chunk_size chunk_tokens = tokens[start:end] chunk_text = tokenizer.decode(chunk_tokens) chunks.append(chunk_text) start = end - chunk_overlap return chunks def chunk_by_sentences( text: str, max_chunk_size: int = 1000, min_chunk_size: int = 100 ) -> List[str]: """Chunk text by sentences, respecting size limits.""" import nltk sentences = nltk.sent_tokenize(text) chunks = [] current_chunk = [] current_size = 0 for sentence in sentences: sentence_size = len(sentence) if current_size + sentence_size > max_chunk_size and current_chunk: chunks.append(" ".join(current_chunk)) current_chunk = [] current_size = 0 current_chunk.append(sentence) current_size += sentence_size if current_chunk: chunks.append(" ".join(current_chunk)) return chunks def chunk_by_semantic_sections( text: str, headers_pattern: str = r'^#{1,3}\s+.+$' ) -> List[Tuple[str, str]]: """Chunk markdown by headers, preserving hierarchy.""" lines = text.split('\n') chunks = [] current_header = "" current_content = [] for line in lines: if re.match(headers_pattern, line, re.MULTILINE): if current_content: chunks.append((current_header, '\n'.join(current_content))) current_header = line current_content = [] else: current_content.append(line) if current_content: chunks.append((current_header, '\n'.join(current_content))) return chunks def recursive_character_splitter( text: str, chunk_size: int = 1000, chunk_overlap: int = 200, separators: List[str] = None ) -> List[str]: """LangChain-style recursive splitter.""" separators = separators or ["\n\n", "\n", ". ", " ", ""] def split_text(text: str, separators: List[str]) -> List[str]: if not text: return [] separator = separators[0] remaining_separators = separators[1:] if separator == "": # Character-level split return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size - chunk_overlap)] splits = text.split(separator) chunks = [] current_chunk = [] current_length = 0 for split in splits: split_length = len(split) + len(separator) if current_length + split_length > chunk_size and current_chunk: chunk_text = separator.join(current_chunk) # Recursively split if still too large if len(chunk_text) > chunk_size and remaining_separators: chunks.extend(split_text(chunk_text, remaining_separators)) else: chunks.append(chunk_text) # Start new chunk with overlap overlap_splits = [] overlap_length = 0 for s in reversed(current_chunk): if overlap_length + len(s) <= chunk_overlap: overlap_splits.insert(0, s) overlap_length += len(s) else: break current_chunk = overlap_splits current_length = overlap_length current_chunk.append(split) current_length += split_length if current_chunk: chunks.append(separator.join(current_chunk)) return chunks return split_text(text, separators) ``` ### Template 5: Domain-Specific Embedding Pipeline ```python import re from typing import List, Optional from dataclasses import dataclass @dataclass class EmbeddedDocument: id: str document_id: str chunk_index: int text: str embedding: List[float] metadata: dict class DomainEmbeddingPipeline: """Pipeline for domain-specific embeddings.""" def __init__( self, embedding_model: str = "voyage-3-large", chunk_size: int = 512, chunk_overlap: int = 50, preprocessing_fn=None ): self.embeddings = VoyageAIEmbeddings(model=embedding_model) self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.preprocess = preprocessing_fn or self._default_preprocess def _default_preprocess(self, text: str) -> str: """Default preprocessing.""" # Remove excessive whitespace text = re.sub(r'\s+', ' ', text) # Remove special characters (customize for your domain) text = re.sub(r'[^\w\s.,!?-]', '', text) return text.strip() async def process_documents( self, documents: List[dict], id_field: str = "id", content_field: str = "content", metadata_fields: Optional[List[str]] = None ) -> List[EmbeddedDocument]: """Process documents for vector storage.""" processed = [] for doc in documents: content = doc[content_field] doc_id = doc[id_field] # Preprocess cleaned = self.preprocess(content) # Chunk chunks = chunk_by_tokens( cleaned, self.chunk_size, self.chunk_overlap ) # Create embeddings embeddings = await self.embeddings.aembed_documents(chunks) # Create records for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): metadata = {"document_id": doc_id, "chunk_index": i} # Add specified metadata fields if metadata_fields: for field in metadata_fields: if field in doc: metadata[field] = doc[field] processed.append(EmbeddedDocument( id=f"{doc_id}_chunk_{i}", document_id=doc_id, chunk_index=i, text=chunk, embedding=embedding, metadata=metadata )) return processed # Code-specific pipeline class CodeEmbeddingPipeline: """Specialized pipeline for code embeddings.""" def __init__(self): # Use Voyage's code-specific model self.embeddings = VoyageAIEmbeddings(model="voyage-code-3") def chunk_code(self, code: str, language: str) -> List[dict]: """Chunk code by functions/classes using tree-sitter.""" try: import tree_sitter_languages parser = tree_sitter_languages.get_parser(language) tree = parser.parse(bytes(code, "utf8")) chunks = [] # Extract function and class definitions self._extract_nodes(tree.root_node, code, chunks) return chunks except ImportError: # Fallback to simple chunking return [{"text": code, "type": "module"}] def _extract_nodes(self, node, source_code: str, chunks: list): """Recursively extract function/class definitions.""" if node.type in ['function_definition', 'class_definition', 'method_definition']: text = source_code[node.start_byte:node.end_byte] chunks.append({ "text": text, "type": node.type, "name": self._get_name(node), "start_line": node.start_point[0], "end_line": node.end_point[0] }) for child in node.children: self._extract_nodes(child, source_code, chunks) def _get_name(self, node) -> str: """Extract name from function/class node.""" for child in node.children: if child.type == 'identifier' or child.type == 'name': return child.text.decode('utf8') return "unknown" async def embed_with_context( self, chunk: str, context: str = "" ) -> List[float]: """Embed code with surrounding context.""" if context: combined = f"Context: {context}\n\nCode:\n{chunk}" else: combined = chunk return await self.embeddings.aembed_query(combined) ``` ### Template 6: Embedding Quality Evaluation ```python import numpy as np from typing import List, Dict def evaluate_retrieval_quality( queries: List[str], relevant_docs: List[List[str]], # List of relevant doc IDs per query retrieved_docs: List[List[str]], # List of retrieved doc IDs per query k: int = 10 ) -> Dict[str, float]: """Evaluate embedding quality for retrieval.""" def precision_at_k(relevant: set, retrieved: List[str], k: int) -> float: retrieved_k = retrieved[:k] relevant_retrieved = len(set(retrieved_k) & relevant) return relevant_retrieved / k if k > 0 else 0 def recall_at_k(relevant: set, retrieved: List[str], k: int) -> float: retrieved_k = retrieved[:k] relevant_retrieved = len(set(retrieved_k) & relevant) return relevant_retrieved / len(relevant) if relevant else 0 def mrr(relevant: set, retrieved: List[str]) -> float: for i, doc in enumerate(retrieved): if doc in relevant: return 1 / (i + 1) return 0 def ndcg_at_k(relevant: set, retrieved: List[str], k: int) -> float: dcg = sum( 1 / np.log2(i + 2) if doc in relevant else 0 for i, doc in enumerate(retrieved[:k]) ) ideal_dcg = sum(1 / np.log2(i + 2) for i in range(min(len(relevant), k))) return dcg / ideal_dcg if ideal_dcg > 0 else 0 metrics = { f"precision@{k}": [], f"recall@{k}": [], "mrr": [], f"ndcg@{k}": [] } for relevant, retrieved in zip(relevant_docs, retrieved_docs): relevant_set = set(relevant) metrics[f"precision@{k}"].append(precision_at_k(relevant_set, retrieved, k)) metrics[f"recall@{k}"].append(recall_at_k(relevant_set, retrieved, k)) metrics["mrr"].append(mrr(relevant_set, retrieved)) metrics[f"ndcg@{k}"].append(ndcg_at_k(relevant_set, retrieved, k)) return {name: np.mean(values) for name, values in metrics.items()} def compute_embedding_similarity( embeddings1: np.ndarray, embeddings2: np.ndarray, metric: str = "cosine" ) -> np.ndarray: """Compute similarity matrix between embedding sets.""" if metric == "cosine": # Normalize and compute dot product norm1 = embeddings1 / np.linalg.norm(embeddings1, axis=1, keepdims=True) norm2 = embeddings2 / np.linalg.norm(embeddings2, axis=1, keepdims=True) return norm1 @ norm2.T elif metric == "euclidean": from scipy.spatial.distance import cdist return -cdist(embeddings1, embeddings2, metric='euclidean') elif metric == "dot": return embeddings1 @ embeddings2.T else: raise ValueError(f"Unknown metric: {metric}") def compare_embedding_models( texts: List[str], models: Dict[str, callable], queries: List[str], relevant_indices: List[List[int]], k: int = 5 ) -> Dict[str, Dict[str, float]]: """Compare multiple embedding models on retrieval quality.""" results = {} for model_name, embed_fn in models.items(): # Embed all texts doc_embeddings = np.array(embed_fn(texts)) retrieved_per_query = [] for query in queries: query_embedding = np.array(embed_fn([query])[0]) # Compute similarities similarities = compute_embedding_similarity( query_embedding.reshape(1, -1), doc_embeddings, metric="cosine" )[0] # Get top-k indices top_k_indices = np.argsort(similarities)[::-1][:k] retrieved_per_query.append([str(i) for i in top_k_indices]) # Convert relevant indices to string IDs relevant_docs = [[str(i) for i in indices] for indices in relevant_indices] results[model_name] = evaluate_retrieval_quality( queries, relevant_docs, retrieved_per_query, k ) return results ``` ## Best Practices ### Do's - **Match model to use case**: Code vs prose vs multilingual - **Chunk thoughtfully**: Preserve semantic boundaries - **Normalize embeddings**: For cosine similarity search - **Batch requests**: More efficient than one-by-one - **Cache embeddings**: Avoid recomputing for static content - **Use Voyage AI for Claude apps**: Recommended by Anthropic ### Don'ts - **Don't ignore token limits**: Truncation loses information - **Don't mix embedding models**: Incompatible vector spaces - **Don't skip preprocessing**: Garbage in, garbage out - **Don't over-chunk**: Lose important context - **Don't forget metadata**: Essential for filtering and debugging ## Resources - [Voyage AI Documentation](https://docs.voyageai.com/) - [OpenAI Embeddings Guide](https://platform.openai.com/docs/guides/embeddings) - [Sentence Transformers](https://www.sbert.net/) - [MTEB Benchmar
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

langchain-architecture

Design LLM applications using LangChain 1.x and LangGraph for

coding
⭐1
# LangChain & LangGraph Architecture Master modern LangChain 1.x and LangGraph for building sophisticated LLM applications with agents, state management, memory, and tool integration. ## When to Use This Skill - Building autonomous AI agents with tool access - Implementing complex multi-step LLM workflows - Managing conversation memory and state - Integrating LLMs with external data sources and APIs - Creating modular, reusable LLM application components - Implementing document processing pipelines - Building production-grade LLM applications ## Package Structure (LangChain 1.x) ``` langchain (1.2.x) # High-level orchestration langchain-core (1.2.x) # Core abstractions (messages, prompts, tools) langchain-community # Third-party integrations langgraph # Agent orchestration and state management langchain-openai # OpenAI integrations langchain-anthropic # Anthropic/Claude integrations langchain-voyageai # Voyage AI embeddings langchain-pinecone # Pinecone vector store ``` ## Core Concepts ### 1. LangGraph Agents LangGraph is the standard for building agents in 2026. It provides: **Key Features:** - **StateGraph**: Explicit state management with typed state - **Durable Execution**: Agents persist through failures - **Human-in-the-Loop**: Inspect and modify state at any point - **Memory**: Short-term and long-term memory across sessions - **Checkpointing**: Save and resume agent state **Agent Patterns:** - **ReAct**: Reasoning + Acting with `create_react_agent` - **Plan-and-Execute**: Separate planning and execution nodes - **Multi-Agent**: Supervisor routing between specialized agents - **Tool-Calling**: Structured tool invocation with Pydantic schemas ### 2. State Management LangGraph uses TypedDict for explicit state: ```python from typing import Annotated, TypedDict from langgraph.graph import MessagesState # Simple message-based state class AgentState(MessagesState): """Extends MessagesState with custom fields.""" context: Annotated[list, "retrieved documents"] # Custom state for complex agents class CustomState(TypedDict): messages: Annotated[list, "conversation history"] context: Annotated[dict, "retrieved context"] current_step: str results: list ``` ### 3. Memory Systems Modern memory implementations: - **ConversationBufferMemory**: Stores all messages (short conversations) - **ConversationSummaryMemory**: Summarizes older messages (long conversations) - **ConversationTokenBufferMemory**: Token-based windowing - **VectorStoreRetrieverMemory**: Semantic similarity retrieval - **LangGraph Checkpointers**: Persistent state across sessions ### 4. Document Processing Loading, transforming, and storing documents: **Components:** - **Document Loaders**: Load from various sources - **Text Splitters**: Chunk documents intelligently - **Vector Stores**: Store and retrieve embeddings - **Retrievers**: Fetch relevant documents ### 5. Callbacks & Tracing LangSmith is the standard for observability: - Request/response logging - Token usage tracking - Latency monitoring - Error tracking - Trace visualization ## Quick Start ### Modern ReAct Agent with LangGraph ```python from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool import ast import operator # Initialize LLM (Claude Sonnet 4.6 recommended) llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0) # Define tools with Pydantic schemas @tool def search_database(query: str) -> str: """Search internal database for information.""" # Your database search logic return f"Results for: {query}" @tool def calculate(expression: str) -> str: """Safely evaluate a mathematical expression. Supports: +, -, *, /, **, %, parentheses Example: '(2 + 3) * 4' returns '20' """ # Safe math evaluation using ast allowed_operators = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod, ast.USub: operator.neg, } def _eval(node): if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left = _eval(node.left) right = _eval(node.right) return allowed_operators[type(node.op)](left, right) elif isinstance(node, ast.UnaryOp): operand = _eval(node.operand) return allowed_operators[type(node.op)](operand) else: raise ValueError(f"Unsupported operation: {type(node)}") try: tree = ast.parse(expression, mode='eval') return str(_eval(tree.body)) except Exception as e: return f"Error: {e}" tools = [search_database, calculate] # Create checkpointer for memory persistence checkpointer = MemorySaver() # Create ReAct agent agent = create_react_agent( llm, tools, checkpointer=checkpointer ) # Run agent with thread ID for memory config = {"configurable": {"thread_id": "user-123"}} result = await agent.ainvoke( {"messages": [("user", "Search for Python tutorials and calculate 25 * 4")]}, config=config ) ``` ## Architecture Patterns ### Pattern 1: RAG with LangGraph ```python from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from typing import TypedDict, Annotated class RAGState(TypedDict): question: str context: Annotated[list[Document], "retrieved documents"] answer: str # Initialize components llm = ChatAnthropic(model="claude-sonnet-4-6") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # Define nodes async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs} async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so. Context: {context} Question: {question} Answer:""" ) context_text = "\n\n".join(doc.page_content for doc in state["context"]) response = await llm.ainvoke( prompt.format(context=context_text, question=state["question"]) ) return {"answer": response.content} # Build graph builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) rag_chain = builder.compile() # Use the chain result = await rag_chain.ainvoke({"question": "What is the main topic?"}) ``` ### Pattern 2: Custom Agent with Structured Tools ```python from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): """Input for database search.""" query: str = Field(description="Search query") filters: dict = Field(default={}, description="Optional filters") class EmailInput(BaseModel): """Input for sending email.""" recipient: str = Field(description="Email recipient") subject: str = Field(description="Email subject") content: str = Field(description="Email body") async def search_database(query: str, filters: dict = {}) -> str: """Search internal database for information.""" # Your database search logic return f"Results for '{query}' with filters {filters}" async def send_email(recipient: str, subject: str, content: str) -> str: """Send an email to specified recipient.""" # Email sending logic return f"Email sent to {recipient}" tools = [ StructuredTool.from_function( coroutine=search_database, name="search_database", description="Search internal database", args_schema=SearchInput ), StructuredTool.from_function( coroutine=send_email, name="send_email", description="Send an email", args_schema=EmailInput ) ] agent = create_react_agent(llm, tools) ``` ### Pattern 3: Multi-Step Workflow with StateGraph ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict, Literal class WorkflowState(TypedDict): text: str entities: list analysis: str summary: str current_step: str async def extract_entities(state: WorkflowState) -> WorkflowState: """Extract key entities from text.""" prompt = f"Extract key entities from: {state['text']}\n\nReturn as JSON list." response = await llm.ainvoke(prompt) return {"entities": response.content, "current_step": "analyze"} async def analyze_entities(state: WorkflowState) -> WorkflowState: """Analyze extracted entities.""" prompt = f"Analyze these entities: {state['entities']}\n\nProvide insights." response = await llm.ainvoke(prompt) return {"analysis": response.content, "current_step": "summarize"} async def generate_summary(state: WorkflowState) -> WorkflowState: """Generate final summary.""" prompt = f"""Summarize: Entities: {state['entities']} Analysis: {state['analysis']} Provide a concise summary.""" response = await llm.ainvoke(prompt) return {"summary": response.content, "current_step": "complete"} def route_step(state: WorkflowState) -> Literal["analyze", "summarize", "end"]: """Route to next step based on current state.""" step = state.get("current_step", "extract") if step == "analyze": return "analyze" elif step == "summarize": return "summarize" return "end" # Build workflow builder = StateGraph(WorkflowState) builder.add_node("extract", extract_entities) builder.add_node("analyze", analyze_entities) builder.add_node("summarize", generate_summary) builder.add_edge(START, "extract") builder.add_conditional_edges("extract", route_step, { "analyze": "analyze", "summarize": "summarize", "end": END }) builder.add_conditional_edges("analyze", route_step, { "summarize": "summarize", "end": END }) builder.add_edge("summarize", END) workflow = builder.compile() ``` ### Pattern 4: Multi-Agent Orchestration ```python from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import create_react_agent from langchain_core.messages import HumanMessage from typing import Literal class MultiAgentState(TypedDict): messages: list next_agent: str # Create specialized agents researcher = create_react_agent(llm, research_tools) writer = create_react_agent(llm, writing_tools) reviewer = create_react_agent(llm, review_tools) async def supervisor(state: MultiAgentState) -> MultiAgentState: """Route to appropriate agent based on task.""" prompt = f"""Based on the conversation, which agent should handle this? Options: - researcher: For finding information - writer: For creating content - reviewer: For reviewing and editing - FINISH: Task is complete Messages: {state['messages']} Respond with just the agent name.""" response = await llm.ainvoke(prompt) return {"next_agent": response.content.strip().lower()} def route_to_agent(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]: """Route based on supervisor decision.""" next_agent = state.get("next_agent", "").lower() if next_agent == "finish": return "end" return next_agent if next_agent in ["researcher", "writer", "reviewer"] else "end" # Build multi-agent graph builder = StateGraph(MultiAgentState) builder.add_node("supervisor", supervisor) builder.add_node("researcher", researcher) builder.add_node("writer", writer) builder.add_node("reviewer", reviewer) builder.add_edge(START, "supervisor") builder.add_conditional_edges("supervisor", route_to_agent, { "researcher": "researcher", "writer": "writer", "reviewer": "reviewer", "end": END }) # Each agent returns to supervisor for agent in ["researcher", "writer", "reviewer"]: builder.add_edge(agent, "supervisor") multi_agent = builder.compile() ``` ## Memory Management ### Token-Based Memory with LangGraph ```python from langgraph.checkpoint.memory import MemorySaver from langgraph.prebuilt import create_react_agent # In-memory checkpointer (development) checkpointer = MemorySaver() # Create agent with persistent memory agent = create_react_agent(llm, tools, checkpointer=checkpointer) # Each thread_id maintains separate conversation config = {"configurable": {"thread_id": "session-abc123"}} # Messages persist across invocations with same thread_id result1 = await agent.ainvoke({"messages": [("user", "My name is Alice")]}, config) result2 = await agent.ainvoke({"messages": [("user", "What's my name?")]}, config) # Agent remembers: "Your name is Alice" ``` ### Production Memory with PostgreSQL ```python from langgraph.checkpoint.postgres import PostgresSaver # Production checkpointer checkpointer = PostgresSaver.from_conn_string( "postgresql://user:pass@localhost/langgraph" ) agent = create_react_agent(llm, tools, checkpointer=checkpointer) ``` ### Vector Store Memory for Long-Term Context ```python from langchain_community.vectorstores import Chroma from langchain_voyageai import VoyageAIEmbeddings embeddings = VoyageAIEmbeddings(model="voyage-3-large") memory_store = Chroma( collection_name="conversation_memory", embedding_function=embeddings, persist_directory="./memory_db" ) async def retrieve_relevant_memory(query: str, k: int = 5) -> list: """Retrieve relevant past conversations.""" docs = await memory_store.asimilarity_search(query, k=k) return [doc.page_content for doc in docs] async def store_memory(content: str, metadata: dict = {}): """Store conversation in long-term memory.""" await memory_store.aadd_texts([content], metadatas=[metadata]) ``` ## Callback System & LangSmith ### LangSmith Tracing ```python import os from langchain_anthropic import ChatAnthropic # Enable LangSmith tracing os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-api-key" os.environ["LANGCHAIN_PROJECT"] = "my-project" # All LangChain/LangGraph operations are automatically traced llm = ChatAnthropic(model="claude-sonnet-4-6") ``` ### Custom Callback Handler ```python from langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List class CustomCallbackHandler(BaseCallbackHandler): def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs ) -> None: print(f"LLM started with {len(prompts)} prompts") def on_llm_end(self, response, **kwargs) -> None: print(f"LLM completed: {len(response.generations)} generations") def on_llm_error(self, error: Exception, **kwargs) -> None: print(f"LLM error: {error}") def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs ) -> None: print(f"Tool started: {serialized.get('name')}") def on_tool_end(self, output: str, **kwargs) -> None: print(f"Tool completed: {output[:100]}...") # Use callbacks result = await agent.ainvoke( {"messages": [("user", "query")]}, config={"callbacks": [CustomCallbackHandler()]} ) ``` ## Streaming Responses ```python from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-sonnet-4-6", streaming=True) # Stream tokens async for chunk in llm.astream("Tell me a story"): print(chunk.content, end="", flush=True) # Stream agent events async for event in agent.astream_events( {"messages": [("user", "Search and summarize")]}, version="v2" ): if event["event"] == "on_chat_model_stream": print(event["data"]["chunk"].content, end="") elif event["event"] == "on_tool_start": print(f"\n[Using tool: {event['name']}]") ``` ## Testing Strategies ```python import pytest from unittest.mock import AsyncMock, patch @pytest.mark.asyncio async def test_agent_tool_selection(): """Test agent selects correct tool.""" with patch.object(llm, 'ainvoke') as mock_llm: mock_llm.return_value = AsyncMock(content="Using search_database") result = await agent.ainvoke({ "messages": [("user", "search for documents")] }) # Verify tool was called assert "search_database" in str(result) @pytest.mark.asyncio async def test_memory_persistence(): """Test memory persists across invocations.""" config = {"configurable": {"thread_id": "test-thread"}} # First message await agent.ainvoke( {"messages": [("user", "Remember: the code is 12345")]}, config ) # Second message should remember result = await agent.ainvoke( {"messages": [("user", "What was the code?")]}, config ) assert "12345" in result["messages"][-1].content ``` ## Performance Optimization ### 1. Caching with Redis ```python from langchain_community.cache import RedisCache from langchain_core.globals import set_llm_cache import redis redis_client = redis.Redis.from_url("redis://localhost:6379") set_llm_cache(RedisCache(redis_client)) ``` ### 2. Async Batch Processing ```python import asyncio from langchain_core.documents import Document async def process_documents(documents: list[Document]) -> list: """Process documents in parallel.""" tasks = [process_single(doc) for doc in documents] return await asyncio.gather(*tasks) async def process_single(doc: Document) -> dict: """Process a single document.""" chunks = text_splitter.split_documents([doc]) embeddings = await embeddings_model.aembed_documents( [c.page_content for c in chunks] ) return {"doc_id": doc.metadata.get("id"), "embeddings": embeddings} ``` ### 3. Connection Pooling ```python from langchain_pinecone import PineconeVectorStore from pinecone import Pinecone # Reuse Pinecone client pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("my-index") # Create vector store with existing index vectorstore = PineconeVectorStore(index=index, embedding=embeddings) ``` ## Resources - [LangChain Documentation](https://python.langchain.com/docs/) - [LangGraph Documentation](https://langchain-ai.github.io/langgraph/) - [LangSmith Platform](https://smith.langchain.com/) - [LangChain GitHub](https://github.com/langchain-ai/langchain) - [LangGraph GitHub](https://github.com/langchain-ai/langgraph) ## Common Pitfalls 1. **Using Deprecated APIs**: Use LangGraph for agents, not `initialize_agent` 2. **Memory Overflow**: Use checkpointers with TTL for long-running agents 3. **Poor Tool Descriptions**: Clear descriptions help LLM select correct tools 4. **Context Window Exceeded**: Use summarization or sliding window memory 5. **No Error Handling**: Wrap too
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

llm-evaluation

Implement comprehensive evaluation strategies for LLM applications

coding
⭐1
# LLM Evaluation Master comprehensive evaluation strategies for LLM applications, from automated metrics to human evaluation and A/B testing. ## When to Use This Skill - Measuring LLM application performance systematically - Comparing different models or prompts - Detecting performance regressions before deployment - Validating improvements from prompt changes - Building confidence in production systems - Establishing baselines and tracking progress over time - Debugging unexpected model behavior ## Core Evaluation Types ### 1. Automated Metrics Fast, repeatable, scalable evaluation using computed scores. **Text Generation:** - **BLEU**: N-gram overlap (translation) - **ROUGE**: Recall-oriented (summarization) - **METEOR**: Semantic similarity - **BERTScore**: Embedding-based similarity - **Perplexity**: Language model confidence **Classification:** - **Accuracy**: Percentage correct - **Precision/Recall/F1**: Class-specific performance - **Confusion Matrix**: Error patterns - **AUC-ROC**: Ranking quality **Retrieval (RAG):** - **MRR**: Mean Reciprocal Rank - **NDCG**: Normalized Discounted Cumulative Gain - **Precision@K**: Relevant in top K - **Recall@K**: Coverage in top K ### 2. Human Evaluation Manual assessment for quality aspects difficult to automate. **Dimensions:** - **Accuracy**: Factual correctness - **Coherence**: Logical flow - **Relevance**: Answers the question - **Fluency**: Natural language quality - **Safety**: No harmful content - **Helpfulness**: Useful to the user ### 3. LLM-as-Judge Use stronger LLMs to evaluate weaker model outputs. **Approaches:** - **Pointwise**: Score individual responses - **Pairwise**: Compare two responses - **Reference-based**: Compare to gold standard - **Reference-free**: Judge without ground truth ## Quick Start ```python from dataclasses import dataclass from typing import Callable import numpy as np @dataclass class Metric: name: str fn: Callable @staticmethod def accuracy(): return Metric("accuracy", calculate_accuracy) @staticmethod def bleu(): return Metric("bleu", calculate_bleu) @staticmethod def bertscore(): return Metric("bertscore", calculate_bertscore) @staticmethod def custom(name: str, fn: Callable): return Metric(name, fn) class EvaluationSuite: def __init__(self, metrics: list[Metric]): self.metrics = metrics async def evaluate(self, model, test_cases: list[dict]) -> dict: results = {m.name: [] for m in self.metrics} for test in test_cases: prediction = await model.predict(test["input"]) for metric in self.metrics: score = metric.fn( prediction=prediction, reference=test.get("expected"), context=test.get("context") ) results[metric.name].append(score) return { "metrics": {k: np.mean(v) for k, v in results.items()}, "raw_scores": results } # Usage suite = EvaluationSuite([ Metric.accuracy(), Metric.bleu(), Metric.bertscore(), Metric.custom("groundedness", check_groundedness) ]) test_cases = [ { "input": "What is the capital of France?", "expected": "Paris", "context": "France is a country in Europe. Paris is its capital." }, ] results = await suite.evaluate(model=your_model, test_cases=test_cases) ``` ## Automated Metrics Implementation ### BLEU Score ```python from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction def calculate_bleu(reference: str, hypothesis: str, **kwargs) -> float: """Calculate BLEU score between reference and hypothesis.""" smoothie = SmoothingFunction().method4 return sentence_bleu( [reference.split()], hypothesis.split(), smoothing_function=smoothie ) ``` ### ROUGE Score ```python from rouge_score import rouge_scorer def calculate_rouge(reference: str, hypothesis: str, **kwargs) -> dict: """Calculate ROUGE scores.""" scorer = rouge_scorer.RougeScorer( ['rouge1', 'rouge2', 'rougeL'], use_stemmer=True ) scores = scorer.score(reference, hypothesis) return { 'rouge1': scores['rouge1'].fmeasure, 'rouge2': scores['rouge2'].fmeasure, 'rougeL': scores['rougeL'].fmeasure } ``` ### BERTScore ```python from bert_score import score def calculate_bertscore( references: list[str], hypotheses: list[str], **kwargs ) -> dict: """Calculate BERTScore using pre-trained model.""" P, R, F1 = score( hypotheses, references, lang='en', model_type='microsoft/deberta-xlarge-mnli' ) return { 'precision': P.mean().item(), 'recall': R.mean().item(), 'f1': F1.mean().item() } ``` ### Custom Metrics ```python def calculate_groundedness(response: str, context: str, **kwargs) -> float: """Check if response is grounded in provided context.""" from transformers import pipeline nli = pipeline( "text-classification", model="microsoft/deberta-large-mnli" ) result = nli(f"{context} [SEP] {response}")[0] # Return confidence that response is entailed by context return result['score'] if result['label'] == 'ENTAILMENT' else 0.0 def calculate_toxicity(text: str, **kwargs) -> float: """Measure toxicity in generated text.""" from detoxify import Detoxify results = Detoxify('original').predict(text) return max(results.values()) # Return highest toxicity score def calculate_factuality(claim: str, sources: list[str], **kwargs) -> float: """Verify factual claims against sources.""" from transformers import pipeline nli = pipeline("text-classification", model="facebook/bart-large-mnli") scores = [] for source in sources: result = nli(f"{source}</s></s>{claim}")[0] if result['label'] == 'entailment': scores.append(result['score']) return max(scores) if scores else 0.0 ``` ## LLM-as-Judge Patterns ### Single Output Evaluation ```python from anthropic import Anthropic from pydantic import BaseModel, Field import json class QualityRating(BaseModel): accuracy: int = Field(ge=1, le=10, description="Factual correctness") helpfulness: int = Field(ge=1, le=10, description="Answers the question") clarity: int = Field(ge=1, le=10, description="Well-written and understandable") reasoning: str = Field(description="Brief explanation") async def llm_judge_quality( response: str, question: str, context: str = None ) -> QualityRating: """Use Claude to judge response quality.""" client = Anthropic() system = """You are an expert evaluator of AI responses. Rate responses on accuracy, helpfulness, and clarity (1-10 scale). Provide brief reasoning for your ratings.""" prompt = f"""Rate the following response: Question: {question} {f'Context: {context}' if context else ''} Response: {response} Provide ratings in JSON format: {{ "accuracy": <1-10>, "helpfulness": <1-10>, "clarity": <1-10>, "reasoning": "<brief explanation>" }}""" message = client.messages.create( model="claude-sonnet-4-6", max_tokens=500, system=system, messages=[{"role": "user", "content": prompt}] ) return QualityRating(**json.loads(message.content[0].text)) ``` ### Pairwise Comparison ```python from pydantic import BaseModel, Field from typing import Literal class ComparisonResult(BaseModel): winner: Literal["A", "B", "tie"] reasoning: str confidence: int = Field(ge=1, le=10) async def compare_responses( question: str, response_a: str, response_b: str ) -> ComparisonResult: """Compare two responses using LLM judge.""" client = Anthropic() prompt = f"""Compare these two responses and determine which is better. Question: {question} Response A: {response_a} Response B: {response_b} Consider accuracy, helpfulness, and clarity. Answer with JSON: {{ "winner": "A" or "B" or "tie", "reasoning": "<explanation>", "confidence": <1-10> }}""" message = client.messages.create( model="claude-sonnet-4-6", max_tokens=500, messages=[{"role": "user", "content": prompt}] ) return ComparisonResult(**json.loads(message.content[0].text)) ``` ### Reference-Based Evaluation ```python class ReferenceEvaluation(BaseModel): semantic_similarity: float = Field(ge=0, le=1) factual_accuracy: float = Field(ge=0, le=1) completeness: float = Field(ge=0, le=1) issues: list[str] async def evaluate_against_reference( response: str, reference: str, question: str ) -> ReferenceEvaluation: """Evaluate response against gold standard reference.""" client = Anthropic() prompt = f"""Compare the response to the reference answer. Question: {question} Reference Answer: {reference} Response to Evaluate: {response} Evaluate: 1. Semantic similarity (0-1): How similar is the meaning? 2. Factual accuracy (0-1): Are all facts correct? 3. Completeness (0-1): Does it cover all key points? 4. List any specific issues or errors. Respond in JSON: {{ "semantic_similarity": <0-1>, "factual_accuracy": <0-1>, "completeness": <0-1>, "issues": ["issue1", "issue2"] }}""" message = client.messages.create( model="claude-sonnet-4-6", max_tokens=500, messages=[{"role": "user", "content": prompt}] ) return ReferenceEvaluation(**json.loads(message.content[0].text)) ``` ## Human Evaluation Frameworks ### Annotation Guidelines ```python from dataclasses import dataclass, field from typing import Optional @dataclass class AnnotationTask: """Structure for human annotation task.""" response: str question: str context: Optional[str] = None def get_annotation_form(self) -> dict: return { "question": self.question, "context": self.context, "response": self.response, "ratings": { "accuracy": { "scale": "1-5", "description": "Is the response factually correct?" }, "relevance": { "scale": "1-5", "description": "Does it answer the question?" }, "coherence": { "scale": "1-5", "description": "Is it logically consistent?" } }, "issues": { "factual_error": False, "hallucination": False, "off_topic": False, "unsafe_content": False }, "feedback": "" } ``` ### Inter-Rater Agreement ```python from sklearn.metrics import cohen_kappa_score def calculate_agreement( rater1_scores: list[int], rater2_scores: list[int] ) -> dict: """Calculate inter-rater agreement.""" kappa = cohen_kappa_score(rater1_scores, rater2_scores) if kappa < 0: interpretation = "Poor" elif kappa < 0.2: interpretation = "Slight" elif kappa < 0.4: interpretation = "Fair" elif kappa < 0.6: interpretation = "Moderate" elif kappa < 0.8: interpretation = "Substantial" else: interpretation = "Almost Perfect" return { "kappa": kappa, "interpretation": interpretation } ``` ## A/B Testing ### Statistical Testing Framework ```python from scipy import stats import numpy as np from dataclasses import dataclass, field @dataclass class ABTest: variant_a_name: str = "A" variant_b_name: str = "B" variant_a_scores: list[float] = field(default_factory=list) variant_b_scores: list[float] = field(default_factory=list) def add_result(self, variant: str, score: float): """Add evaluation result for a variant.""" if variant == "A": self.variant_a_scores.append(score) else: self.variant_b_scores.append(score) def analyze(self, alpha: float = 0.05) -> dict: """Perform statistical analysis.""" a_scores = np.array(self.variant_a_scores) b_scores = np.array(self.variant_b_scores) # T-test t_stat, p_value = stats.ttest_ind(a_scores, b_scores) # Effect size (Cohen's d) pooled_std = np.sqrt((np.std(a_scores)**2 + np.std(b_scores)**2) / 2) cohens_d = (np.mean(b_scores) - np.mean(a_scores)) / pooled_std return { "variant_a_mean": np.mean(a_scores), "variant_b_mean": np.mean(b_scores), "difference": np.mean(b_scores) - np.mean(a_scores), "relative_improvement": (np.mean(b_scores) - np.mean(a_scores)) / np.mean(a_scores), "p_value": p_value, "statistically_significant": p_value < alpha, "cohens_d": cohens_d, "effect_size": self._interpret_cohens_d(cohens_d), "winner": self.variant_b_name if np.mean(b_scores) > np.mean(a_scores) else self.variant_a_name } @staticmethod def _interpret_cohens_d(d: float) -> str: """Interpret Cohen's d effect size.""" abs_d = abs(d) if abs_d < 0.2: return "negligible" elif abs_d < 0.5: return "small" elif abs_d < 0.8: return "medium" else: return "large" ``` ## Regression Testing ### Regression Detection ```python from dataclasses import dataclass @dataclass class RegressionResult: metric: str baseline: float current: float change: float is_regression: bool class RegressionDetector: def __init__(self, baseline_results: dict, threshold: float = 0.05): self.baseline = baseline_results self.threshold = threshold def check_for_regression(self, new_results: dict) -> dict: """Detect if new results show regression.""" regressions = [] for metric in self.baseline.keys(): baseline_score = self.baseline[metric] new_score = new_results.get(metric) if new_score is None: continue # Calculate relative change relative_change = (new_score - baseline_score) / baseline_score # Flag if significant decrease is_regression = relative_change < -self.threshold if is_regression: regressions.append(RegressionResult( metric=metric, baseline=baseline_score, current=new_score, change=relative_change, is_regression=True )) return { "has_regression": len(regressions) > 0, "regressions": regressions, "summary": f"{len(regressions)} metric(s) regressed" } ``` ## LangSmith Evaluation Integration ```python from langsmith import Client from langsmith.evaluation import evaluate, LangChainStringEvaluator # Initialize LangSmith client client = Client() # Create dataset dataset = client.create_dataset("qa_test_cases") client.create_examples( inputs=[{"question": q} for q in questions], outputs=[{"answer": a} for a in expected_answers], dataset_id=dataset.id ) # Define evaluators evaluators = [ LangChainStringEvaluator("qa"), # QA correctness LangChainStringEvaluator("context_qa"), # Context-grounded QA LangChainStringEvaluator("cot_qa"), # Chain-of-thought QA ] # Run evaluation async def target_function(inputs: dict) -> dict: result = await your_chain.ainvoke(inputs) return {"answer": result} experiment_results = await evaluate( target_function, data=dataset.name, evaluators=evaluators, experiment_prefix="v1.0.0", metadata={"model": "claude-sonnet-4-6", "version": "1.0.0"} ) print(f"Mean score: {experiment_results.aggregate_metrics['qa']['mean']}") ``` ## Benchmarking ### Running Benchmarks ```python from dataclasses import dataclass import numpy as np @dataclass class BenchmarkResult: metric: str mean: float std: float min: float max: float class BenchmarkRunner: def __init__(self, benchmark_dataset: list[dict]): self.dataset = benchmark_dataset async def run_benchmark( self, model, metrics: list[Metric] ) -> dict[str, BenchmarkResult]: """Run model on benchmark and calculate metrics.""" results = {metric.name: [] for metric in metrics} for example in self.dataset: # Generate prediction prediction = await model.predict(example["input"]) # Calculate each metric for metric in metrics: score = metric.fn( prediction=prediction, reference=example["reference"], context=example.get("context") ) results[metric.name].append(score) # Aggregate results return { metric: BenchmarkResult( metric=metric, mean=np.mean(scores), std=np.std(scores), min=min(scores), max=max(scores) ) for metric, scores in results.items() } ``` ## Resources - [LangSmith Evaluation Guide](https://docs.smith.langchain.com/evaluation) - [RAGAS Framework](https://docs.ragas.io/) - [DeepEval Library](https://docs.deepeval.com/) - [Arize Phoenix](https://docs.arize.com/phoenix/) - [HELM Benchmark](https://crfm.stanford.edu/helm/) ## Best Practices 1. **Multiple Metrics**: Use diverse metrics for comprehensive view 2. **Representative Data**: Test on real-world, diverse examples 3. **Baselines**: Always compare against baseline performance 4. **Statistical Rigor**: Use proper statistical tests for comparisons 5. **Continuous Evaluation**: Integrate into CI/CD pipeline 6. **Human Validation**: Combine automated metrics with human judgment 7. **Error Analysis**: Investigate failures to understand weaknesses 8. **Version Control**: Track evaluation results over time ## Common Pitfalls - **Single Metric Obsession**: Optimizing for one metric at the expense of others - **Small Sample Size**: Drawing conclusions from too few examples - **Data Contamination**: Testing on training data - **Ignoring Variance**: Not accounting for statistical uncertainty - **Metric Mismatch**: Using metrics not aligned with business goals - **Position Bias**: In pairwise evals, randomize order - **Overfitting Prompts**: Optimizing for test set instead of real use
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

prompt-engineering-patterns

Master advanced prompt engineering techniques to maximize LLM

coding
⭐1
# Prompt Engineering Patterns Master advanced prompt engineering techniques to maximize LLM performance, reliability, and controllability. ## When to Use This Skill - Designing complex prompts for production LLM applications - Optimizing prompt performance and consistency - Implementing structured reasoning patterns (chain-of-thought, tree-of-thought) - Building few-shot learning systems with dynamic example selection - Creating reusable prompt templates with variable interpolation - Debugging and refining prompts that produce inconsistent outputs - Implementing system prompts for specialized AI assistants - Using structured outputs (JSON mode) for reliable parsing ## Core Capabilities ### 1. Few-Shot Learning - Example selection strategies (semantic similarity, diversity sampling) - Balancing example count with context window constraints - Constructing effective demonstrations with input-output pairs - Dynamic example retrieval from knowledge bases - Handling edge cases through strategic example selection ### 2. Chain-of-Thought Prompting - Step-by-step reasoning elicitation - Zero-shot CoT with "Let's think step by step" - Few-shot CoT with reasoning traces - Self-consistency techniques (sampling multiple reasoning paths) - Verification and validation steps ### 3. Structured Outputs - JSON mode for reliable parsing - Pydantic schema enforcement - Type-safe response handling - Error handling for malformed outputs ### 4. Prompt Optimization - Iterative refinement workflows - A/B testing prompt variations - Measuring prompt performance metrics (accuracy, consistency, latency) - Reducing token usage while maintaining quality - Handling edge cases and failure modes ### 5. Template Systems - Variable interpolation and formatting - Conditional prompt sections - Multi-turn conversation templates - Role-based prompt composition - Modular prompt components ### 6. System Prompt Design - Setting model behavior and constraints - Defining output formats and structure - Establishing role and expertise - Safety guidelines and content policies - Context setting and background information ## Quick Start ```python from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field # Define structured output schema class SQLQuery(BaseModel): query: str = Field(description="The SQL query") explanation: str = Field(description="Brief explanation of what the query does") tables_used: list[str] = Field(description="List of tables referenced") # Initialize model with structured output llm = ChatAnthropic(model="claude-sonnet-4-6") structured_llm = llm.with_structured_output(SQLQuery) # Create prompt template prompt = ChatPromptTemplate.from_messages([ ("system", """You are an expert SQL developer. Generate efficient, secure SQL queries. Always use parameterized queries to prevent SQL injection. Explain your reasoning briefly."""), ("user", "Convert this to SQL: {query}") ]) # Create chain chain = prompt | structured_llm # Use result = await chain.ainvoke({ "query": "Find all users who registered in the last 30 days" }) print(result.query) print(result.explanation) ``` ## Key Patterns ### Pattern 1: Structured Output with Pydantic ```python from anthropic import Anthropic from pydantic import BaseModel, Field from typing import Literal import json class SentimentAnalysis(BaseModel): sentiment: Literal["positive", "negative", "neutral"] confidence: float = Field(ge=0, le=1) key_phrases: list[str] reasoning: str async def analyze_sentiment(text: str) -> SentimentAnalysis: """Analyze sentiment with structured output.""" client = Anthropic() message = client.messages.create( model="claude-sonnet-4-6", max_tokens=500, messages=[{ "role": "user", "content": f"""Analyze the sentiment of this text. Text: {text} Respond with JSON matching this schema: {{ "sentiment": "positive" | "negative" | "neutral", "confidence": 0.0-1.0, "key_phrases": ["phrase1", "phrase2"], "reasoning": "brief explanation" }}""" }] ) return SentimentAnalysis(**json.loads(message.content[0].text)) ``` ### Pattern 2: Chain-of-Thought with Self-Verification ```python from langchain_core.prompts import ChatPromptTemplate cot_prompt = ChatPromptTemplate.from_template(""" Solve this problem step by step. Problem: {problem} Instructions: 1. Break down the problem into clear steps 2. Work through each step showing your reasoning 3. State your final answer 4. Verify your answer by checking it against the original problem Format your response as: ## Steps [Your step-by-step reasoning] ## Answer [Your final answer] ## Verification [Check that your answer is correct] """) ``` ### Pattern 3: Few-Shot with Dynamic Example Selection ```python from langchain_voyageai import VoyageAIEmbeddings from langchain_core.example_selectors import SemanticSimilarityExampleSelector from langchain_chroma import Chroma # Create example selector with semantic similarity example_selector = SemanticSimilarityExampleSelector.from_examples( examples=[ {"input": "How do I reset my password?", "output": "Go to Settings > Security > Reset Password"}, {"input": "Where can I see my order history?", "output": "Navigate to Account > Orders"}, {"input": "How do I contact support?", "output": "Click Help > Contact Us or email support@example.com"}, ], embeddings=VoyageAIEmbeddings(model="voyage-3-large"), vectorstore_cls=Chroma, k=2 # Select 2 most similar examples ) async def get_few_shot_prompt(query: str) -> str: """Build prompt with dynamically selected examples.""" examples = await example_selector.aselect_examples({"input": query}) examples_text = "\n".join( f"User: {ex['input']}\nAssistant: {ex['output']}" for ex in examples ) return f"""You are a helpful customer support assistant. Here are some example interactions: {examples_text} Now respond to this query: User: {query} Assistant:""" ``` ### Pattern 4: Progressive Disclosure Start with simple prompts, add complexity only when needed: ```python PROMPT_LEVELS = { # Level 1: Direct instruction "simple": "Summarize this article: {text}", # Level 2: Add constraints "constrained": """Summarize this article in 3 bullet points, focusing on: - Key findings - Main conclusions - Practical implications Article: {text}""", # Level 3: Add reasoning "reasoning": """Read this article carefully. 1. First, identify the main topic and thesis 2. Then, extract the key supporting points 3. Finally, summarize in 3 bullet points Article: {text} Summary:""", # Level 4: Add examples "few_shot": """Read articles and provide concise summaries. Example: Article: "New research shows that regular exercise can reduce anxiety by up to 40%..." Summary: β€’ Regular exercise reduces anxiety by up to 40% β€’ 30 minutes of moderate activity 3x/week is sufficient β€’ Benefits appear within 2 weeks of starting Now summarize this article: Article: {text} Summary:""" } ``` ### Pattern 5: Error Recovery and Fallback ```python from pydantic import BaseModel, ValidationError import json class ResponseWithConfidence(BaseModel): answer: str confidence: float sources: list[str] alternative_interpretations: list[str] = [] ERROR_RECOVERY_PROMPT = """ Answer the question based on the context provided. Context: {context} Question: {question} Instructions: 1. If you can answer confidently (>0.8), provide a direct answer 2. If you're somewhat confident (0.5-0.8), provide your best answer with caveats 3. If you're uncertain (<0.5), explain what information is missing 4. Always provide alternative interpretations if the question is ambiguous Respond in JSON: {{ "answer": "your answer or 'I cannot determine this from the context'", "confidence": 0.0-1.0, "sources": ["relevant context excerpts"], "alternative_interpretations": ["if question is ambiguous"] }} """ async def answer_with_fallback( context: str, question: str, llm ) -> ResponseWithConfidence: """Answer with error recovery and fallback.""" prompt = ERROR_RECOVERY_PROMPT.format(context=context, question=question) try: response = await llm.ainvoke(prompt) return ResponseWithConfidence(**json.loads(response.content)) except (json.JSONDecodeError, ValidationError) as e: # Fallback: try to extract answer without structure simple_prompt = f"Based on: {context}\n\nAnswer: {question}" simple_response = await llm.ainvoke(simple_prompt) return ResponseWithConfidence( answer=simple_response.content, confidence=0.5, sources=["fallback extraction"], alternative_interpretations=[] ) ``` ### Pattern 6: Role-Based System Prompts ```python SYSTEM_PROMPTS = { "analyst": """You are a senior data analyst with expertise in SQL, Python, and business intelligence. Your responsibilities: - Write efficient, well-documented queries - Explain your analysis methodology - Highlight key insights and recommendations - Flag any data quality concerns Communication style: - Be precise and technical when discussing methodology - Translate technical findings into business impact - Use clear visualizations when helpful""", "assistant": """You are a helpful AI assistant focused on accuracy and clarity. Core principles: - Always cite sources when making factual claims - Acknowledge uncertainty rather than guessing - Ask clarifying questions when the request is ambiguous - Provide step-by-step explanations for complex topics Constraints: - Do not provide medical, legal, or financial advice - Redirect harmful requests appropriately - Protect user privacy""", "code_reviewer": """You are a senior software engineer conducting code reviews. Review criteria: - Correctness: Does the code work as intended? - Security: Are there any vulnerabilities? - Performance: Are there efficiency concerns? - Maintainability: Is the code readable and well-structured? - Best practices: Does it follow language idioms? Output format: 1. Summary assessment (approve/request changes) 2. Critical issues (must fix) 3. Suggestions (nice to have) 4. Positive feedback (what's done well)""" } ``` ## Integration Patterns ### With RAG Systems ```python RAG_PROMPT = """You are a knowledgeable assistant that answers questions based on provided context. Context (retrieved from knowledge base): {context} Instructions: 1. Answer ONLY based on the provided context 2. If the context doesn't contain the answer, say "I don't have information about that in my knowledge base" 3. Cite specific passages using [1], [2] notation 4. If the question is ambiguous, ask for clarification Question: {question} Answer:""" ``` ### With Validation and Verification ```python VALIDATED_PROMPT = """Complete the following task: Task: {task} After generating your response, verify it meets ALL these criteria: βœ“ Directly addresses the original request βœ“ Contains no factual errors βœ“ Is appropriately detailed (not too brief, not too verbose) βœ“ Uses proper formatting βœ“ Is safe and appropriate If verification fails on any criterion, revise before responding. Response:""" ``` ## Performance Optimization ### Token Efficiency ```python # Before: Verbose prompt (150+ tokens) verbose_prompt = """ I would like you to please take the following text and provide me with a comprehensive summary of the main points. The summary should capture the key ideas and important details while being concise and easy to understand. """ # After: Concise prompt (30 tokens) concise_prompt = """Summarize the key points concisely: {text} Summary:""" ``` ### Caching Common Prefixes ```python from anthropic import Anthropic client = Anthropic() # Use prompt caching for repeated system prompts response = client.messages.create( model="claude-sonnet-4-6", max_tokens=1000, system=[ { "type": "text", "text": LONG_SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"} } ], messages=[{"role": "user", "content": user_query}] ) ``` ## Best Practices 1. **Be Specific**: Vague prompts produce inconsistent results 2. **Show, Don't Tell**: Examples are more effective than descriptions 3. **Use Structured Outputs**: Enforce schemas with Pydantic for reliability 4. **Test Extensively**: Evaluate on diverse, representative inputs 5. **Iterate Rapidly**: Small changes can have large impacts 6. **Monitor Performance**: Track metrics in production 7. **Version Control**: Treat prompts as code with proper versioning 8. **Document Intent**: Explain why prompts are structured as they are ## Common Pitfalls - **Over-engineering**: Starting with complex prompts before trying simple ones - **Example pollution**: Using examples that don't match the target task - **Context overflow**: Exceeding token limits with excessive examples - **Ambiguous instructions**: Leaving room for multiple interpretations - **Ignoring edge cases**: Not testing on unusual or boundary inputs - **No error handling**: Assuming outputs will always be well-formed - **Hardcoded values**: Not parameterizing prompts for reuse ## Success Metrics Track these KPIs for your prompts: - **Accuracy**: Correctness of outputs - **Consistency**: Reproducibility across similar inputs - **Latency**: Response time (P50, P95, P99) - **Token Usage**: Average tokens per request - **Success Rate**: Percentage of valid, parseable outputs - **User Satisfaction**: Ratings and feedback ## Resources - [Anthropic Prompt Engineering Guide](https://docs.anthropic.com/en/docs/build-with-claude/prompt-engineering) - [Claude Prompt Caching](https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching) - [OpenAI Prompt Engineering](https://platform.openai.com/docs/guides/prompt-engineering) - [LangChain Prompts](https://python.langchain.com/docs/concepts/prompts/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

cost-optimization

Optimize cloud costs through resource rightsizing, tagging

architecture
⭐1
# Cloud Cost Optimization Strategies and patterns for optimizing cloud costs across AWS, Azure, and GCP. ## Purpose Implement systematic cost optimization strategies to reduce cloud spending while maintaining performance and reliability. ## When to Use - Reduce cloud spending - Right-size resources - Implement cost governance - Optimize multi-cloud costs - Meet budget constraints ## Cost Optimization Framework ### 1. Visibility - Implement cost allocation tags - Use cloud cost management tools - Set up budget alerts - Create cost dashboards ### 2. Right-Sizing - Analyze resource utilization - Downsize over-provisioned resources - Use auto-scaling - Remove idle resources ### 3. Pricing Models - Use reserved capacity - Leverage spot/preemptible instances - Implement savings plans - Use committed use discounts ### 4. Architecture Optimization - Use managed services - Implement caching - Optimize data transfer - Use lifecycle policies ## AWS Cost Optimization ### Reserved Instances ``` Savings: 30-72% vs On-Demand Term: 1 or 3 years Payment: All/Partial/No upfront Flexibility: Standard or Convertible ``` ### Savings Plans ``` Compute Savings Plans: 66% savings EC2 Instance Savings Plans: 72% savings Applies to: EC2, Fargate, Lambda Flexible across: Instance families, regions, OS ``` ### Spot Instances ``` Savings: Up to 90% vs On-Demand Best for: Batch jobs, CI/CD, stateless workloads Risk: 2-minute interruption notice Strategy: Mix with On-Demand for resilience ``` ### S3 Cost Optimization ```hcl resource "aws_s3_bucket_lifecycle_configuration" "example" { bucket = aws_s3_bucket.example.id rule { id = "transition-to-ia" status = "Enabled" transition { days = 30 storage_class = "STANDARD_IA" } transition { days = 90 storage_class = "GLACIER" } expiration { days = 365 } } } ``` ## Azure Cost Optimization ### Reserved VM Instances - 1 or 3 year terms - Up to 72% savings - Flexible sizing - Exchangeable ### Azure Hybrid Benefit - Use existing Windows Server licenses - Up to 80% savings with RI - Available for Windows and SQL Server ### Azure Advisor Recommendations - Right-size VMs - Delete unused resources - Use reserved capacity - Optimize storage ## GCP Cost Optimization ### Committed Use Discounts - 1 or 3 year commitment - Up to 57% savings - Applies to vCPUs and memory - Resource-based or spend-based ### Sustained Use Discounts - Automatic discounts - Up to 30% for running instances - No commitment required - Applies to Compute Engine, GKE ### Preemptible VMs - Up to 80% savings - 24-hour maximum runtime - Best for batch workloads ## Tagging Strategy ### AWS Tagging ```hcl locals { common_tags = { Environment = "production" Project = "my-project" CostCenter = "engineering" Owner = "team@example.com" ManagedBy = "terraform" } } resource "aws_instance" "example" { ami = "ami-12345678" instance_type = "t3.medium" tags = merge( local.common_tags, { Name = "web-server" } ) } ``` **Reference:** See `references/tagging-standards.md` ## Cost Monitoring ### Budget Alerts ```hcl # AWS Budget resource "aws_budgets_budget" "monthly" { name = "monthly-budget" budget_type = "COST" limit_amount = "1000" limit_unit = "USD" time_period_start = "2024-01-01_00:00" time_unit = "MONTHLY" notification { comparison_operator = "GREATER_THAN" threshold = 80 threshold_type = "PERCENTAGE" notification_type = "ACTUAL" subscriber_email_addresses = ["team@example.com"] } } ``` ### Cost Anomaly Detection - AWS Cost Anomaly Detection - Azure Cost Management alerts - GCP Budget alerts ## Architecture Patterns ### Pattern 1: Serverless First - Use Lambda/Functions for event-driven - Pay only for execution time - Auto-scaling included - No idle costs ### Pattern 2: Right-Sized Databases ``` Development: t3.small RDS Staging: t3.large RDS Production: r6g.2xlarge RDS with read replicas ``` ### Pattern 3: Multi-Tier Storage ``` Hot data: S3 Standard Warm data: S3 Standard-IA (30 days) Cold data: S3 Glacier (90 days) Archive: S3 Deep Archive (365 days) ``` ### Pattern 4: Auto-Scaling ```hcl resource "aws_autoscaling_policy" "scale_up" { name = "scale-up" scaling_adjustment = 2 adjustment_type = "ChangeInCapacity" cooldown = 300 autoscaling_group_name = aws_autoscaling_group.main.name } resource "aws_cloudwatch_metric_alarm" "cpu_high" { alarm_name = "cpu-high" comparison_operator = "GreaterThanThreshold" evaluation_periods = "2" metric_name = "CPUUtilization" namespace = "AWS/EC2" period = "60" statistic = "Average" threshold = "80" alarm_actions = [aws_autoscaling_policy.scale_up.arn] } ``` ## Cost Optimization Checklist - [ ] Implement cost allocation tags - [ ] Delete unused resources (EBS, EIPs, snapshots) - [ ] Right-size instances based on utilization - [ ] Use reserved capacity for steady workloads - [ ] Implement auto-scaling - [ ] Optimize storage classes - [ ] Use lifecycle policies - [ ] Enable cost anomaly detection - [ ] Set budget alerts - [ ] Review costs weekly - [ ] Use spot/preemptible instances - [ ] Optimize data transfer costs - [ ] Implement caching layers - [ ] Use managed services - [ ] Monitor and optimize continuously ## Tools - **AWS:** Cost Explorer, Cost Anomaly Detection, Compute Optimizer - **Azure:** Cost Management, Advisor - **GCP:** Cost Management, Recommender - **Multi-cloud:** CloudHealth, Cloudability, Kubecost ## Reference Files - `references/tagging-standards.md` - Tagging conventions - `assets/cost-analysis-template.xlsx` - Cost analysis spreadsheet ## Related Skills - `terraform-module-library` - For resource provisioning - `multi-cloud-architecture` - For cloud selection
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

similarity-search-patterns

Implement efficient similarity search with vector databases. Use

coding
⭐1
# Similarity Search Patterns Patterns for implementing efficient similarity search in production systems. ## When to Use This Skill - Building semantic search systems - Implementing RAG retrieval - Creating recommendation engines - Optimizing search latency - Scaling to millions of vectors - Combining semantic and keyword search ## Core Concepts ### 1. Distance Metrics | Metric | Formula | Best For | | ------------------ | ------------------ | --------------------- | --- | -------------- | | **Cosine** | 1 - (AΒ·B)/(β€–Aβ€–β€–Bβ€–) | Normalized embeddings | | **Euclidean (L2)** | √Σ(a-b)Β² | Raw embeddings | | **Dot Product** | AΒ·B | Magnitude matters | | **Manhattan (L1)** | Ξ£ | a-b | | Sparse vectors | ### 2. Index Types ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Index Types β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Flat β”‚ HNSW β”‚ IVF+PQ β”‚ β”‚ (Exact) β”‚ (Graph-based) β”‚ (Quantized) β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ O(n) search β”‚ O(log n) β”‚ O(√n) β”‚ β”‚ 100% recall β”‚ ~95-99% β”‚ ~90-95% β”‚ β”‚ Small data β”‚ Medium-Large β”‚ Very Large β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ## Templates ### Template 1: Pinecone Implementation ```python from pinecone import Pinecone, ServerlessSpec from typing import List, Dict, Optional import hashlib class PineconeVectorStore: def __init__( self, api_key: str, index_name: str, dimension: int = 1536, metric: str = "cosine" ): self.pc = Pinecone(api_key=api_key) # Create index if not exists if index_name not in self.pc.list_indexes().names(): self.pc.create_index( name=index_name, dimension=dimension, metric=metric, spec=ServerlessSpec(cloud="aws", region="us-east-1") ) self.index = self.pc.Index(index_name) def upsert( self, vectors: List[Dict], namespace: str = "" ) -> int: """ Upsert vectors. vectors: [{"id": str, "values": List[float], "metadata": dict}] """ # Batch upsert batch_size = 100 total = 0 for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] self.index.upsert(vectors=batch, namespace=namespace) total += len(batch) return total def search( self, query_vector: List[float], top_k: int = 10, namespace: str = "", filter: Optional[Dict] = None, include_metadata: bool = True ) -> List[Dict]: """Search for similar vectors.""" results = self.index.query( vector=query_vector, top_k=top_k, namespace=namespace, filter=filter, include_metadata=include_metadata ) return [ { "id": match.id, "score": match.score, "metadata": match.metadata } for match in results.matches ] def search_with_rerank( self, query: str, query_vector: List[float], top_k: int = 10, rerank_top_n: int = 50, namespace: str = "" ) -> List[Dict]: """Search and rerank results.""" # Over-fetch for reranking initial_results = self.search( query_vector, top_k=rerank_top_n, namespace=namespace ) # Rerank with cross-encoder or LLM reranked = self._rerank(query, initial_results) return reranked[:top_k] def _rerank(self, query: str, results: List[Dict]) -> List[Dict]: """Rerank results using cross-encoder.""" from sentence_transformers import CrossEncoder model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') pairs = [(query, r["metadata"]["text"]) for r in results] scores = model.predict(pairs) for result, score in zip(results, scores): result["rerank_score"] = float(score) return sorted(results, key=lambda x: x["rerank_score"], reverse=True) def delete(self, ids: List[str], namespace: str = ""): """Delete vectors by ID.""" self.index.delete(ids=ids, namespace=namespace) def delete_by_filter(self, filter: Dict, namespace: str = ""): """Delete vectors matching filter.""" self.index.delete(filter=filter, namespace=namespace) ``` ### Template 2: Qdrant Implementation ```python from qdrant_client import QdrantClient from qdrant_client.http import models from typing import List, Dict, Optional class QdrantVectorStore: def __init__( self, url: str = "localhost", port: int = 6333, collection_name: str = "documents", vector_size: int = 1536 ): self.client = QdrantClient(url=url, port=port) self.collection_name = collection_name # Create collection if not exists collections = self.client.get_collections().collections if collection_name not in [c.name for c in collections]: self.client.create_collection( collection_name=collection_name, vectors_config=models.VectorParams( size=vector_size, distance=models.Distance.COSINE ), # Optional: enable quantization for memory efficiency quantization_config=models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=True ) ) ) def upsert(self, points: List[Dict]) -> int: """ Upsert points. points: [{"id": str/int, "vector": List[float], "payload": dict}] """ qdrant_points = [ models.PointStruct( id=p["id"], vector=p["vector"], payload=p.get("payload", {}) ) for p in points ] self.client.upsert( collection_name=self.collection_name, points=qdrant_points ) return len(points) def search( self, query_vector: List[float], limit: int = 10, filter: Optional[models.Filter] = None, score_threshold: Optional[float] = None ) -> List[Dict]: """Search for similar vectors.""" results = self.client.search( collection_name=self.collection_name, query_vector=query_vector, limit=limit, query_filter=filter, score_threshold=score_threshold ) return [ { "id": r.id, "score": r.score, "payload": r.payload } for r in results ] def search_with_filter( self, query_vector: List[float], must_conditions: List[Dict] = None, should_conditions: List[Dict] = None, must_not_conditions: List[Dict] = None, limit: int = 10 ) -> List[Dict]: """Search with complex filters.""" conditions = [] if must_conditions: conditions.extend([ models.FieldCondition( key=c["key"], match=models.MatchValue(value=c["value"]) ) for c in must_conditions ]) filter = models.Filter(must=conditions) if conditions else None return self.search(query_vector, limit=limit, filter=filter) def search_with_sparse( self, dense_vector: List[float], sparse_vector: Dict[int, float], limit: int = 10, dense_weight: float = 0.7 ) -> List[Dict]: """Hybrid search with dense and sparse vectors.""" # Requires collection with named vectors results = self.client.search( collection_name=self.collection_name, query_vector=models.NamedVector( name="dense", vector=dense_vector ), limit=limit ) return [{"id": r.id, "score": r.score, "payload": r.payload} for r in results] ``` ### Template 3: pgvector with PostgreSQL ```python import asyncpg from typing import List, Dict, Optional import numpy as np class PgVectorStore: def __init__(self, connection_string: str): self.connection_string = connection_string async def init(self): """Initialize connection pool and extension.""" self.pool = await asyncpg.create_pool(self.connection_string) async with self.pool.acquire() as conn: # Enable extension await conn.execute("CREATE EXTENSION IF NOT EXISTS vector") # Create table await conn.execute(""" CREATE TABLE IF NOT EXISTS documents ( id TEXT PRIMARY KEY, content TEXT, metadata JSONB, embedding vector(1536) ) """) # Create index (HNSW for better performance) await conn.execute(""" CREATE INDEX IF NOT EXISTS documents_embedding_idx ON documents USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64) """) async def upsert(self, documents: List[Dict]): """Upsert documents with embeddings.""" async with self.pool.acquire() as conn: await conn.executemany( """ INSERT INTO documents (id, content, metadata, embedding) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET content = EXCLUDED.content, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding """, [ ( doc["id"], doc["content"], doc.get("metadata", {}), np.array(doc["embedding"]).tolist() ) for doc in documents ] ) async def search( self, query_embedding: List[float], limit: int = 10, filter_metadata: Optional[Dict] = None ) -> List[Dict]: """Search for similar documents.""" query = """ SELECT id, content, metadata, 1 - (embedding <=> $1::vector) as similarity FROM documents """ params = [query_embedding] if filter_metadata: conditions = [] for key, value in filter_metadata.items(): params.append(value) conditions.append(f"metadata->>'{key}' = ${len(params)}") query += " WHERE " + " AND ".join(conditions) query += f" ORDER BY embedding <=> $1::vector LIMIT ${len(params) + 1}" params.append(limit) async with self.pool.acquire() as conn: rows = await conn.fetch(query, *params) return [ { "id": row["id"], "content": row["content"], "metadata": row["metadata"], "score": row["similarity"] } for row in rows ] async def hybrid_search( self, query_embedding: List[float], query_text: str, limit: int = 10, vector_weight: float = 0.5 ) -> List[Dict]: """Hybrid search combining vector and full-text.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ WITH vector_results AS ( SELECT id, content, metadata, 1 - (embedding <=> $1::vector) as vector_score FROM documents ORDER BY embedding <=> $1::vector LIMIT $3 * 2 ), text_results AS ( SELECT id, content, metadata, ts_rank(to_tsvector('english', content), plainto_tsquery('english', $2)) as text_score FROM documents WHERE to_tsvector('english', content) @@ plainto_tsquery('english', $2) LIMIT $3 * 2 ) SELECT COALESCE(v.id, t.id) as id, COALESCE(v.content, t.content) as content, COALESCE(v.metadata, t.metadata) as metadata, COALESCE(v.vector_score, 0) * $4 + COALESCE(t.text_score, 0) * (1 - $4) as combined_score FROM vector_results v FULL OUTER JOIN text_results t ON v.id = t.id ORDER BY combined_score DESC LIMIT $3 """, query_embedding, query_text, limit, vector_weight ) return [dict(row) for row in rows] ``` ### Template 4: Weaviate Implementation ```python import weaviate from weaviate.util import generate_uuid5 from typing import List, Dict, Optional class WeaviateVectorStore: def __init__( self, url: str = "http://localhost:8080", class_name: str = "Document" ): self.client = weaviate.Client(url=url) self.class_name = class_name self._ensure_schema() def _ensure_schema(self): """Create schema if not exists.""" schema = { "class": self.class_name, "vectorizer": "none", # We provide vectors "properties": [ {"name": "content", "dataType": ["text"]}, {"name": "source", "dataType": ["string"]}, {"name": "chunk_id", "dataType": ["int"]} ] } if not self.client.schema.exists(self.class_name): self.client.schema.create_class(schema) def upsert(self, documents: List[Dict]): """Batch upsert documents.""" with self.client.batch as batch: batch.batch_size = 100 for doc in documents: batch.add_data_object( data_object={ "content": doc["content"], "source": doc.get("source", ""), "chunk_id": doc.get("chunk_id", 0) }, class_name=self.class_name, uuid=generate_uuid5(doc["id"]), vector=doc["embedding"] ) def search( self, query_vector: List[float], limit: int = 10, where_filter: Optional[Dict] = None ) -> List[Dict]: """Vector search.""" query = ( self.client.query .get(self.class_name, ["content", "source", "chunk_id"]) .with_near_vector({"vector": query_vector}) .with_limit(limit) .with_additional(["distance", "id"]) ) if where_filter: query = query.with_where(where_filter) results = query.do() return [ { "id": item["_additional"]["id"], "content": item["content"], "source": item["source"], "score": 1 - item["_additional"]["distance"] } for item in results["data"]["Get"][self.class_name] ] def hybrid_search( self, query: str, query_vector: List[float], limit: int = 10, alpha: float = 0.5 # 0 = keyword, 1 = vector ) -> List[Dict]: """Hybrid search combining BM25 and vector.""" results = ( self.client.query .get(self.class_name, ["content", "source"]) .with_hybrid(query=query, vector=query_vector, alpha=alpha) .with_limit(limit) .with_additional(["score"]) .do() ) return [ { "content": item["content"], "source": item["source"], "score": item["_additional"]["score"] } for item in results["data"]["Get"][self.class_name] ] ``` ## Best Practices ### Do's - **Use appropriate index** - HNSW for most cases - **Tune parameters** - ef_search, nprobe for recall/speed - **Implement hybrid search** - Combine with keyword search - **Monitor recall** - Measure search quality - **Pre-filter when possible** - Reduce search space ### Don'ts - **Don't skip evaluation** - Measure before optimizing - **Don't over-index** - Start with flat, scale up - **Don't ignore latency** - P99 matters for UX - **Don't forget costs** - Vector storage adds up ## Resources - [Pinecone Docs](https://docs.pinecone.io/) - [Qdrant Docs](https://qdrant.tech/documentation/) - [pgvector](https://github.com/pgvector/pgvector) - [Weaviate Docs](https://weaviate.io/developers/weaviate)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

ml-pipeline-workflow

Build end-to-end MLOps pipelines from data preparation through

coding
⭐1
# ML Pipeline Workflow Complete end-to-end MLOps pipeline orchestration from data preparation through model deployment. ## Overview This skill provides comprehensive guidance for building production ML pipelines that handle the full lifecycle: data ingestion β†’ preparation β†’ training β†’ validation β†’ deployment β†’ monitoring. ## When to Use This Skill - Building new ML pipelines from scratch - Designing workflow orchestration for ML systems - Implementing data β†’ model β†’ deployment automation - Setting up reproducible training workflows - Creating DAG-based ML orchestration - Integrating ML components into production systems ## What This Skill Provides ### Core Capabilities 1. **Pipeline Architecture** - End-to-end workflow design - DAG orchestration patterns (Airflow, Dagster, Kubeflow) - Component dependencies and data flow - Error handling and retry strategies 2. **Data Preparation** - Data validation and quality checks - Feature engineering pipelines - Data versioning and lineage - Train/validation/test splitting strategies 3. **Model Training** - Training job orchestration - Hyperparameter management - Experiment tracking integration - Distributed training patterns 4. **Model Validation** - Validation frameworks and metrics - A/B testing infrastructure - Performance regression detection - Model comparison workflows 5. **Deployment Automation** - Model serving patterns - Canary deployments - Blue-green deployment strategies - Rollback mechanisms ### Reference Documentation See the `references/` directory for detailed guides: - **data-preparation.md** - Data cleaning, validation, and feature engineering - **model-training.md** - Training workflows and best practices - **model-validation.md** - Validation strategies and metrics - **model-deployment.md** - Deployment patterns and serving architectures ### Assets and Templates The `assets/` directory contains: - **pipeline-dag.yaml.template** - DAG template for workflow orchestration - **training-config.yaml** - Training configuration template - **validation-checklist.md** - Pre-deployment validation checklist ## Usage Patterns ### Basic Pipeline Setup ```python # 1. Define pipeline stages stages = [ "data_ingestion", "data_validation", "feature_engineering", "model_training", "model_validation", "model_deployment" ] # 2. Configure dependencies # See assets/pipeline-dag.yaml.template for full example ``` ### Production Workflow 1. **Data Preparation Phase** - Ingest raw data from sources - Run data quality checks - Apply feature transformations - Version processed datasets 2. **Training Phase** - Load versioned training data - Execute training jobs - Track experiments and metrics - Save trained models 3. **Validation Phase** - Run validation test suite - Compare against baseline - Generate performance reports - Approve for deployment 4. **Deployment Phase** - Package model artifacts - Deploy to serving infrastructure - Configure monitoring - Validate production traffic ## Best Practices ### Pipeline Design - **Modularity**: Each stage should be independently testable - **Idempotency**: Re-running stages should be safe - **Observability**: Log metrics at every stage - **Versioning**: Track data, code, and model versions - **Failure Handling**: Implement retry logic and alerting ### Data Management - Use data validation libraries (Great Expectations, TFX) - Version datasets with DVC or similar tools - Document feature engineering transformations - Maintain data lineage tracking ### Model Operations - Separate training and serving infrastructure - Use model registries (MLflow, Weights & Biases) - Implement gradual rollouts for new models - Monitor model performance drift - Maintain rollback capabilities ### Deployment Strategies - Start with shadow deployments - Use canary releases for validation - Implement A/B testing infrastructure - Set up automated rollback triggers - Monitor latency and throughput ## Integration Points ### Orchestration Tools - **Apache Airflow**: DAG-based workflow orchestration - **Dagster**: Asset-based pipeline orchestration - **Kubeflow Pipelines**: Kubernetes-native ML workflows - **Prefect**: Modern dataflow automation ### Experiment Tracking - MLflow for experiment tracking and model registry - Weights & Biases for visualization and collaboration - TensorBoard for training metrics ### Deployment Platforms - AWS SageMaker for managed ML infrastructure - Google Vertex AI for GCP deployments - Azure ML for Azure cloud - Kubernetes + KServe for cloud-agnostic serving ## Progressive Disclosure Start with the basics and gradually add complexity: 1. **Level 1**: Simple linear pipeline (data β†’ train β†’ deploy) 2. **Level 2**: Add validation and monitoring stages 3. **Level 3**: Implement hyperparameter tuning 4. **Level 4**: Add A/B testing and gradual rollouts 5. **Level 5**: Multi-model pipelines with ensemble strategies ## Common Patterns ### Batch Training Pipeline ```yaml # See assets/pipeline-dag.yaml.template stages: - name: data_preparation dependencies: [] - name: model_training dependencies: [data_preparation] - name: model_evaluation dependencies: [model_training] - name: model_deployment dependencies: [model_evaluation] ``` ### Real-time Feature Pipeline ```python # Stream processing for real-time features # Combined with batch training # See references/data-preparation.md ``` ### Continuous Training ```python # Automated retraining on schedule # Triggered by data drift detection # See references/model-training.md ``` ## Troubleshooting ### Common Issues - **Pipeline failures**: Check dependencies and data availability - **Training instability**: Review hyperparameters and data quality - **Deployment issues**: Validate model artifacts and serving config - **Performance degradation**: Monitor data drift and model metrics ### Debugging Steps 1. Check pipeline logs for each stage 2. Validate input/output data at boundaries 3. Test components in isolation 4. Review experiment tracking metrics 5. Inspect model artifacts and metadata ## Next Steps After setting up your pipeline: 1. Explore **hyperparameter-tuning** skill for optimization 2. Learn **experiment-tracking-setup** for MLflow/W&B 3. Review **model-deployment-patterns** for serving strategies 4. Implement monitoring with observability tools ## Related Skills - **experiment-tracking-setup**: MLflow and Weights & Biases integration - **hyperparameter-tuning**: Automated hyperparameter optimization - **model-deployment-patterns**: Advanced deployment strategies
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

grafana-dashboards

Create and manage production Grafana dashboards for real-time

coding
⭐1
# Grafana Dashboards Create and manage production-ready Grafana dashboards for comprehensive system observability. ## Purpose Design effective Grafana dashboards for monitoring applications, infrastructure, and business metrics. ## When to Use - Visualize Prometheus metrics - Create custom dashboards - Implement SLO dashboards - Monitor infrastructure - Track business KPIs ## Dashboard Design Principles ### 1. Hierarchy of Information ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Critical Metrics (Big Numbers) β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Key Trends (Time Series) β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Detailed Metrics (Tables/Heatmaps) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### 2. RED Method (Services) - **Rate** - Requests per second - **Errors** - Error rate - **Duration** - Latency/response time ### 3. USE Method (Resources) - **Utilization** - % time resource is busy - **Saturation** - Queue length/wait time - **Errors** - Error count ## Dashboard Structure ### API Monitoring Dashboard ```json { "dashboard": { "title": "API Monitoring", "tags": ["api", "production"], "timezone": "browser", "refresh": "30s", "panels": [ { "title": "Request Rate", "type": "graph", "targets": [ { "expr": "sum(rate(http_requests_total[5m])) by (service)", "legendFormat": "{{service}}" } ], "gridPos": { "x": 0, "y": 0, "w": 12, "h": 8 } }, { "title": "Error Rate %", "type": "graph", "targets": [ { "expr": "(sum(rate(http_requests_total{status=~\"5..\"}[5m])) / sum(rate(http_requests_total[5m]))) * 100", "legendFormat": "Error Rate" } ], "alert": { "conditions": [ { "evaluator": { "params": [5], "type": "gt" }, "operator": { "type": "and" }, "query": { "params": ["A", "5m", "now"] }, "type": "query" } ] }, "gridPos": { "x": 12, "y": 0, "w": 12, "h": 8 } }, { "title": "P95 Latency", "type": "graph", "targets": [ { "expr": "histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (le, service))", "legendFormat": "{{service}}" } ], "gridPos": { "x": 0, "y": 8, "w": 24, "h": 8 } } ] } } ``` **Reference:** See `assets/api-dashboard.json` ## Panel Types ### 1. Stat Panel (Single Value) ```json { "type": "stat", "title": "Total Requests", "targets": [ { "expr": "sum(http_requests_total)" } ], "options": { "reduceOptions": { "values": false, "calcs": ["lastNotNull"] }, "orientation": "auto", "textMode": "auto", "colorMode": "value" }, "fieldConfig": { "defaults": { "thresholds": { "mode": "absolute", "steps": [ { "value": 0, "color": "green" }, { "value": 80, "color": "yellow" }, { "value": 90, "color": "red" } ] } } } } ``` ### 2. Time Series Graph ```json { "type": "graph", "title": "CPU Usage", "targets": [ { "expr": "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[5m])) * 100)" } ], "yaxes": [ { "format": "percent", "max": 100, "min": 0 }, { "format": "short" } ] } ``` ### 3. Table Panel ```json { "type": "table", "title": "Service Status", "targets": [ { "expr": "up", "format": "table", "instant": true } ], "transformations": [ { "id": "organize", "options": { "excludeByName": { "Time": true }, "indexByName": {}, "renameByName": { "instance": "Instance", "job": "Service", "Value": "Status" } } } ] } ``` ### 4. Heatmap ```json { "type": "heatmap", "title": "Latency Heatmap", "targets": [ { "expr": "sum(rate(http_request_duration_seconds_bucket[5m])) by (le)", "format": "heatmap" } ], "dataFormat": "tsbuckets", "yAxis": { "format": "s" } } ``` ## Variables ### Query Variables ```json { "templating": { "list": [ { "name": "namespace", "type": "query", "datasource": "Prometheus", "query": "label_values(kube_pod_info, namespace)", "refresh": 1, "multi": false }, { "name": "service", "type": "query", "datasource": "Prometheus", "query": "label_values(kube_service_info{namespace=\"$namespace\"}, service)", "refresh": 1, "multi": true } ] } } ``` ### Use Variables in Queries ``` sum(rate(http_requests_total{namespace="$namespace", service=~"$service"}[5m])) ``` ## Alerts in Dashboards ```json { "alert": { "name": "High Error Rate", "conditions": [ { "evaluator": { "params": [5], "type": "gt" }, "operator": { "type": "and" }, "query": { "params": ["A", "5m", "now"] }, "reducer": { "type": "avg" }, "type": "query" } ], "executionErrorState": "alerting", "for": "5m", "frequency": "1m", "message": "Error rate is above 5%", "noDataState": "no_data", "notifications": [{ "uid": "slack-channel" }] } } ``` ## Dashboard Provisioning **dashboards.yml:** ```yaml apiVersion: 1 providers: - name: "default" orgId: 1 folder: "General" type: file disableDeletion: false updateIntervalSeconds: 10 allowUiUpdates: true options: path: /etc/grafana/dashboards ``` ## Common Dashboard Patterns ### Infrastructure Dashboard **Key Panels:** - CPU utilization per node - Memory usage per node - Disk I/O - Network traffic - Pod count by namespace - Node status **Reference:** See `assets/infrastructure-dashboard.json` ### Database Dashboard **Key Panels:** - Queries per second - Connection pool usage - Query latency (P50, P95, P99) - Active connections - Database size - Replication lag - Slow queries **Reference:** See `assets/database-dashboard.json` ### Application Dashboard **Key Panels:** - Request rate - Error rate - Response time (percentiles) - Active users/sessions - Cache hit rate - Queue length ## Best Practices 1. **Start with templates** (Grafana community dashboards) 2. **Use consistent naming** for panels and variables 3. **Group related metrics** in rows 4. **Set appropriate time ranges** (default: Last 6 hours) 5. **Use variables** for flexibility 6. **Add panel descriptions** for context 7. **Configure units** correctly 8. **Set meaningful thresholds** for colors 9. **Use consistent colors** across dashboards 10. **Test with different time ranges** ## Dashboard as Code ### Terraform Provisioning ```hcl resource "grafana_dashboard" "api_monitoring" { config_json = file("${path.module}/dashboards/api-monitoring.json") folder = grafana_folder.monitoring.id } resource "grafana_folder" "monitoring" { title = "Production Monitoring" } ``` ### Ansible Provisioning ```yaml - name: Deploy Grafana dashboards copy: src: "{{ item }}" dest: /etc/grafana/dashboards/ with_fileglob: - "dashboards/*.json" notify: restart grafana ``` ## Reference Files - `assets/api-dashboard.json` - API monitoring dashboard - `assets/infrastructure-dashboard.json` - Infrastructure dashboard - `assets/database-dashboard.json` - Database monitoring dashboard - `references/dashboard-design.md` - Dashboard design guide ## Related Skills - `prometheus-configuration` - For metric collection - `slo-implementation` - For SLO dashboards
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

prometheus-configuration

Set up Prometheus for comprehensive metric collection, storage, and

coding
⭐1
# Prometheus Configuration Complete guide to Prometheus setup, metric collection, scrape configuration, and recording rules. ## Purpose Configure Prometheus for comprehensive metric collection, alerting, and monitoring of infrastructure and applications. ## When to Use - Set up Prometheus monitoring - Configure metric scraping - Create recording rules - Design alert rules - Implement service discovery ## Prometheus Architecture ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Applications β”‚ ← Instrumented with client libraries β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ /metrics endpoint ↓ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Prometheus β”‚ ← Scrapes metrics periodically β”‚ Server β”‚ β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”œβ”€β†’ AlertManager (alerts) β”œβ”€β†’ Grafana (visualization) └─→ Long-term storage (Thanos/Cortex) ``` ## Installation ### Kubernetes with Helm ```bash helm repo add prometheus-community https://prometheus-community.github.io/helm-charts helm repo update helm install prometheus prometheus-community/kube-prometheus-stack \ --namespace monitoring \ --create-namespace \ --set prometheus.prometheusSpec.retention=30d \ --set prometheus.prometheusSpec.storageVolumeSize=50Gi ``` ### Docker Compose ```yaml version: "3.8" services: prometheus: image: prom/prometheus:latest ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml - prometheus-data:/prometheus command: - "--config.file=/etc/prometheus/prometheus.yml" - "--storage.tsdb.path=/prometheus" - "--storage.tsdb.retention.time=30d" volumes: prometheus-data: ``` ## Configuration File **prometheus.yml:** ```yaml global: scrape_interval: 15s evaluation_interval: 15s external_labels: cluster: "production" region: "us-west-2" # Alertmanager configuration alerting: alertmanagers: - static_configs: - targets: - alertmanager:9093 # Load rules files rule_files: - /etc/prometheus/rules/*.yml # Scrape configurations scrape_configs: # Prometheus itself - job_name: "prometheus" static_configs: - targets: ["localhost:9090"] # Node exporters - job_name: "node-exporter" static_configs: - targets: - "node1:9100" - "node2:9100" - "node3:9100" relabel_configs: - source_labels: [__address__] target_label: instance regex: "([^:]+)(:[0-9]+)?" replacement: "${1}" # Kubernetes pods with annotations - job_name: "kubernetes-pods" kubernetes_sd_configs: - role: pod relabel_configs: - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape] action: keep regex: true - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path] action: replace target_label: __metrics_path__ regex: (.+) - source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port] action: replace regex: ([^:]+)(?::\d+)?;(\d+) replacement: $1:$2 target_label: __address__ - source_labels: [__meta_kubernetes_namespace] action: replace target_label: namespace - source_labels: [__meta_kubernetes_pod_name] action: replace target_label: pod # Application metrics - job_name: "my-app" static_configs: - targets: - "app1.example.com:9090" - "app2.example.com:9090" metrics_path: "/metrics" scheme: "https" tls_config: ca_file: /etc/prometheus/ca.crt cert_file: /etc/prometheus/client.crt key_file: /etc/prometheus/client.key ``` **Reference:** See `assets/prometheus.yml.template` ## Scrape Configurations ### Static Targets ```yaml scrape_configs: - job_name: "static-targets" static_configs: - targets: ["host1:9100", "host2:9100"] labels: env: "production" region: "us-west-2" ``` ### File-based Service Discovery ```yaml scrape_configs: - job_name: "file-sd" file_sd_configs: - files: - /etc/prometheus/targets/*.json - /etc/prometheus/targets/*.yml refresh_interval: 5m ``` **targets/production.json:** ```json [ { "targets": ["app1:9090", "app2:9090"], "labels": { "env": "production", "service": "api" } } ] ``` ### Kubernetes Service Discovery ```yaml scrape_configs: - job_name: "kubernetes-services" kubernetes_sd_configs: - role: service relabel_configs: - source_labels: [__meta_kubernetes_service_annotation_prometheus_io_scrape] action: keep regex: true - source_labels: [__meta_kubernetes_service_annotation_prometheus_io_scheme] action: replace target_label: __scheme__ regex: (https?) - source_labels: [__meta_kubernetes_service_annotation_prometheus_io_path] action: replace target_label: __metrics_path__ regex: (.+) ``` **Reference:** See `references/scrape-configs.md` ## Recording Rules Create pre-computed metrics for frequently queried expressions: ```yaml # /etc/prometheus/rules/recording_rules.yml groups: - name: api_metrics interval: 15s rules: # HTTP request rate per service - record: job:http_requests:rate5m expr: sum by (job) (rate(http_requests_total[5m])) # Error rate percentage - record: job:http_requests_errors:rate5m expr: sum by (job) (rate(http_requests_total{status=~"5.."}[5m])) - record: job:http_requests_error_rate:percentage expr: | (job:http_requests_errors:rate5m / job:http_requests:rate5m) * 100 # P95 latency - record: job:http_request_duration:p95 expr: | histogram_quantile(0.95, sum by (job, le) (rate(http_request_duration_seconds_bucket[5m])) ) - name: resource_metrics interval: 30s rules: # CPU utilization percentage - record: instance:node_cpu:utilization expr: | 100 - (avg by (instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100) # Memory utilization percentage - record: instance:node_memory:utilization expr: | 100 - ((node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes) * 100) # Disk usage percentage - record: instance:node_disk:utilization expr: | 100 - ((node_filesystem_avail_bytes / node_filesystem_size_bytes) * 100) ``` **Reference:** See `references/recording-rules.md` ## Alert Rules ```yaml # /etc/prometheus/rules/alert_rules.yml groups: - name: availability interval: 30s rules: - alert: ServiceDown expr: up{job="my-app"} == 0 for: 1m labels: severity: critical annotations: summary: "Service {{ $labels.instance }} is down" description: "{{ $labels.job }} has been down for more than 1 minute" - alert: HighErrorRate expr: job:http_requests_error_rate:percentage > 5 for: 5m labels: severity: warning annotations: summary: "High error rate for {{ $labels.job }}" description: "Error rate is {{ $value }}% (threshold: 5%)" - alert: HighLatency expr: job:http_request_duration:p95 > 1 for: 5m labels: severity: warning annotations: summary: "High latency for {{ $labels.job }}" description: "P95 latency is {{ $value }}s (threshold: 1s)" - name: resources interval: 1m rules: - alert: HighCPUUsage expr: instance:node_cpu:utilization > 80 for: 5m labels: severity: warning annotations: summary: "High CPU usage on {{ $labels.instance }}" description: "CPU usage is {{ $value }}%" - alert: HighMemoryUsage expr: instance:node_memory:utilization > 85 for: 5m labels: severity: warning annotations: summary: "High memory usage on {{ $labels.instance }}" description: "Memory usage is {{ $value }}%" - alert: DiskSpaceLow expr: instance:node_disk:utilization > 90 for: 5m labels: severity: critical annotations: summary: "Low disk space on {{ $labels.instance }}" description: "Disk usage is {{ $value }}%" ``` ## Validation ```bash # Validate configuration promtool check config prometheus.yml # Validate rules promtool check rules /etc/prometheus/rules/*.yml # Test query promtool query instant http://localhost:9090 'up' ``` **Reference:** See `scripts/validate-prometheus.sh` ## Best Practices 1. **Use consistent naming** for metrics (prefix_name_unit) 2. **Set appropriate scrape intervals** (15-60s typical) 3. **Use recording rules** for expensive queries 4. **Implement high availability** (multiple Prometheus instances) 5. **Configure retention** based on storage capacity 6. **Use relabeling** for metric cleanup 7. **Monitor Prometheus itself** 8. **Implement federation** for large deployments 9. **Use Thanos/Cortex** for long-term storage 10. **Document custom metrics** ## Troubleshooting **Check scrape targets:** ```bash curl http://localhost:9090/api/v1/targets ``` **Check configuration:** ```bash curl http://localhost:9090/api/v1/status/config ``` **Test query:** ```bash curl 'http://localhost:9090/api/v1/query?query=up' ``` ## Reference Files - `assets/prometheus.yml.template` - Complete configuration template - `references/scrape-configs.md` - Scrape configuration patterns - `references/recording-rules.md` - Recording rule examples - `scripts/validate-prometheus.sh` - Validation script ## Related Skills - `grafana-dashboards` - For visualization - `slo-implementation` - For SLO monitoring - `distributed-tracing` - For request tracing
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-design-patterns

Python design patterns including KISS, Separation of Concerns,

coding
⭐1
# Python Design Patterns Write maintainable Python code using fundamental design principles. These patterns help you build systems that are easy to understand, test, and modify. ## When to Use This Skill - Designing new components or services - Refactoring complex or tangled code - Deciding whether to create an abstraction - Choosing between inheritance and composition - Evaluating code complexity and coupling - Planning modular architectures ## Core Concepts ### 1. KISS (Keep It Simple) Choose the simplest solution that works. Complexity must be justified by concrete requirements. ### 2. Single Responsibility (SRP) Each unit should have one reason to change. Separate concerns into focused components. ### 3. Composition Over Inheritance Build behavior by combining objects, not extending classes. ### 4. Rule of Three Wait until you have three instances before abstracting. Duplication is often better than premature abstraction. ## Quick Start ```python # Simple beats clever # Instead of a factory/registry pattern: FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter} def get_formatter(name: str) -> Formatter: return FORMATTERS[name]() ``` ## Fundamental Patterns ### Pattern 1: KISS - Keep It Simple Before adding complexity, ask: does a simpler solution work? ```python # Over-engineered: Factory with registration class OutputFormatterFactory: _formatters: dict[str, type[Formatter]] = {} @classmethod def register(cls, name: str): def decorator(formatter_cls): cls._formatters[name] = formatter_cls return formatter_cls return decorator @classmethod def create(cls, name: str) -> Formatter: return cls._formatters[name]() @OutputFormatterFactory.register("json") class JsonFormatter(Formatter): ... # Simple: Just use a dictionary FORMATTERS = { "json": JsonFormatter, "csv": CsvFormatter, "xml": XmlFormatter, } def get_formatter(name: str) -> Formatter: """Get formatter by name.""" if name not in FORMATTERS: raise ValueError(f"Unknown format: {name}") return FORMATTERS[name]() ``` The factory pattern adds code without adding value here. Save patterns for when they solve real problems. ### Pattern 2: Single Responsibility Principle Each class or function should have one reason to change. ```python # BAD: Handler does everything class UserHandler: async def create_user(self, request: Request) -> Response: # HTTP parsing data = await request.json() # Validation if not data.get("email"): return Response({"error": "email required"}, status=400) # Database access user = await db.execute( "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *", data["email"], data["name"] ) # Response formatting return Response({"id": user.id, "email": user.email}, status=201) # GOOD: Separated concerns class UserService: """Business logic only.""" def __init__(self, repo: UserRepository) -> None: self._repo = repo async def create_user(self, data: CreateUserInput) -> User: # Only business rules here user = User(email=data.email, name=data.name) return await self._repo.save(user) class UserHandler: """HTTP concerns only.""" def __init__(self, service: UserService) -> None: self._service = service async def create_user(self, request: Request) -> Response: data = CreateUserInput(**(await request.json())) user = await self._service.create_user(data) return Response(user.to_dict(), status=201) ``` Now HTTP changes don't affect business logic, and vice versa. ### Pattern 3: Separation of Concerns Organize code into distinct layers with clear responsibilities. ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ API Layer (handlers) β”‚ β”‚ - Parse requests β”‚ β”‚ - Call services β”‚ β”‚ - Format responses β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Service Layer (business logic) β”‚ β”‚ - Domain rules and validation β”‚ β”‚ - Orchestrate operations β”‚ β”‚ - Pure functions where possible β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Repository Layer (data access) β”‚ β”‚ - SQL queries β”‚ β”‚ - External API calls β”‚ β”‚ - Cache operations β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` Each layer depends only on layers below it: ```python # Repository: Data access class UserRepository: async def get_by_id(self, user_id: str) -> User | None: row = await self._db.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) return User(**row) if row else None # Service: Business logic class UserService: def __init__(self, repo: UserRepository) -> None: self._repo = repo async def get_user(self, user_id: str) -> User: user = await self._repo.get_by_id(user_id) if user is None: raise UserNotFoundError(user_id) return user # Handler: HTTP concerns @app.get("/users/{user_id}") async def get_user(user_id: str) -> UserResponse: user = await user_service.get_user(user_id) return UserResponse.from_user(user) ``` ### Pattern 4: Composition Over Inheritance Build behavior by combining objects rather than inheriting. ```python # Inheritance: Rigid and hard to test class EmailNotificationService(NotificationService): def __init__(self): super().__init__() self._smtp = SmtpClient() # Hard to mock def notify(self, user: User, message: str) -> None: self._smtp.send(user.email, message) # Composition: Flexible and testable class NotificationService: """Send notifications via multiple channels.""" def __init__( self, email_sender: EmailSender, sms_sender: SmsSender | None = None, push_sender: PushSender | None = None, ) -> None: self._email = email_sender self._sms = sms_sender self._push = push_sender async def notify( self, user: User, message: str, channels: set[str] | None = None, ) -> None: channels = channels or {"email"} if "email" in channels: await self._email.send(user.email, message) if "sms" in channels and self._sms and user.phone: await self._sms.send(user.phone, message) if "push" in channels and self._push and user.device_token: await self._push.send(user.device_token, message) # Easy to test with fakes service = NotificationService( email_sender=FakeEmailSender(), sms_sender=FakeSmsSender(), ) ``` ## Advanced Patterns ### Pattern 5: Rule of Three Wait until you have three instances before abstracting. ```python # Two similar functions? Don't abstract yet def process_orders(orders: list[Order]) -> list[Result]: results = [] for order in orders: validated = validate_order(order) result = process_validated_order(validated) results.append(result) return results def process_returns(returns: list[Return]) -> list[Result]: results = [] for ret in returns: validated = validate_return(ret) result = process_validated_return(validated) results.append(result) return results # These look similar, but wait! Are they actually the same? # Different validation, different processing, different errors... # Duplication is often better than the wrong abstraction # Only after a third case, consider if there's a real pattern # But even then, sometimes explicit is better than abstract ``` ### Pattern 6: Function Size Guidelines Keep functions focused. Extract when a function: - Exceeds 20-50 lines (varies by complexity) - Serves multiple distinct purposes - Has deeply nested logic (3+ levels) ```python # Too long, multiple concerns mixed def process_order(order: Order) -> Result: # 50 lines of validation... # 30 lines of inventory check... # 40 lines of payment processing... # 20 lines of notification... pass # Better: Composed from focused functions def process_order(order: Order) -> Result: """Process a customer order through the complete workflow.""" validate_order(order) reserve_inventory(order) payment_result = charge_payment(order) send_confirmation(order, payment_result) return Result(success=True, order_id=order.id) ``` ### Pattern 7: Dependency Injection Pass dependencies through constructors for testability. ```python from typing import Protocol class Logger(Protocol): def info(self, msg: str, **kwargs) -> None: ... def error(self, msg: str, **kwargs) -> None: ... class Cache(Protocol): async def get(self, key: str) -> str | None: ... async def set(self, key: str, value: str, ttl: int) -> None: ... class UserService: """Service with injected dependencies.""" def __init__( self, repository: UserRepository, cache: Cache, logger: Logger, ) -> None: self._repo = repository self._cache = cache self._logger = logger async def get_user(self, user_id: str) -> User: # Check cache first cached = await self._cache.get(f"user:{user_id}") if cached: self._logger.info("Cache hit", user_id=user_id) return User.from_json(cached) # Fetch from database user = await self._repo.get_by_id(user_id) if user: await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300) return user # Production service = UserService( repository=PostgresUserRepository(db), cache=RedisCache(redis), logger=StructlogLogger(), ) # Testing service = UserService( repository=InMemoryUserRepository(), cache=FakeCache(), logger=NullLogger(), ) ``` ### Pattern 8: Avoiding Common Anti-Patterns **Don't expose internal types:** ```python # BAD: Leaking ORM model to API @app.get("/users/{id}") def get_user(id: str) -> UserModel: # SQLAlchemy model return db.query(UserModel).get(id) # GOOD: Use response schemas @app.get("/users/{id}") def get_user(id: str) -> UserResponse: user = db.query(UserModel).get(id) return UserResponse.from_orm(user) ``` **Don't mix I/O with business logic:** ```python # BAD: SQL embedded in business logic def calculate_discount(user_id: str) -> float: user = db.query("SELECT * FROM users WHERE id = ?", user_id) orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id) # Business logic mixed with data access # GOOD: Repository pattern def calculate_discount(user: User, order_history: list[Order]) -> float: # Pure business logic, easily testable if len(order_history) > 10: return 0.15 return 0.0 ``` ## Best Practices Summary 1. **Keep it simple** - Choose the simplest solution that works 2. **Single responsibility** - Each unit has one reason to change 3. **Separate concerns** - Distinct layers with clear purposes 4. **Compose, don't inherit** - Combine objects for flexibility 5. **Rule of three** - Wait before abstracting 6. **Keep functions small** - 20-50 lines (varies by complexity), one purpose 7. **Inject dependencies** - Constructor injection for testability 8. **Delete before abstracting** - Remove dead code, then consider patterns 9. **Test each layer** - Isolated tests for each concern 10. **Explicit over clever** - Readable code beats elegant code
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

backtesting-frameworks

Build robust backtesting systems for trading strategies with proper

coding
⭐1
# Backtesting Frameworks Build robust, production-grade backtesting systems that avoid common pitfalls and produce reliable strategy performance estimates. ## When to Use This Skill - Developing trading strategy backtests - Building backtesting infrastructure - Validating strategy performance - Avoiding common backtesting biases - Implementing walk-forward analysis - Comparing strategy alternatives ## Core Concepts ### 1. Backtesting Biases | Bias | Description | Mitigation | | ---------------- | ------------------------- | ----------------------- | | **Look-ahead** | Using future information | Point-in-time data | | **Survivorship** | Only testing on survivors | Use delisted securities | | **Overfitting** | Curve-fitting to history | Out-of-sample testing | | **Selection** | Cherry-picking strategies | Pre-registration | | **Transaction** | Ignoring trading costs | Realistic cost models | ### 2. Proper Backtest Structure ``` Historical Data β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Training Set β”‚ β”‚ (Strategy Development & Optimization) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Validation Set β”‚ β”‚ (Parameter Selection, No Peeking) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Test Set β”‚ β”‚ (Final Performance Evaluation) β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### 3. Walk-Forward Analysis ``` Window 1: [Train──────][Test] Window 2: [Train──────][Test] Window 3: [Train──────][Test] Window 4: [Train──────][Test] ─────▢ Time ``` ## Implementation Patterns ### Pattern 1: Event-Driven Backtester ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from decimal import Decimal from enum import Enum from typing import Dict, List, Optional import pandas as pd import numpy as np class OrderSide(Enum): BUY = "buy" SELL = "sell" class OrderType(Enum): MARKET = "market" LIMIT = "limit" STOP = "stop" @dataclass class Order: symbol: str side: OrderSide quantity: Decimal order_type: OrderType limit_price: Optional[Decimal] = None stop_price: Optional[Decimal] = None timestamp: Optional[datetime] = None @dataclass class Fill: order: Order fill_price: Decimal fill_quantity: Decimal commission: Decimal slippage: Decimal timestamp: datetime @dataclass class Position: symbol: str quantity: Decimal = Decimal("0") avg_cost: Decimal = Decimal("0") realized_pnl: Decimal = Decimal("0") def update(self, fill: Fill) -> None: if fill.order.side == OrderSide.BUY: new_quantity = self.quantity + fill.fill_quantity if new_quantity != 0: self.avg_cost = ( (self.quantity * self.avg_cost + fill.fill_quantity * fill.fill_price) / new_quantity ) self.quantity = new_quantity else: self.realized_pnl += fill.fill_quantity * (fill.fill_price - self.avg_cost) self.quantity -= fill.fill_quantity @dataclass class Portfolio: cash: Decimal positions: Dict[str, Position] = field(default_factory=dict) def get_position(self, symbol: str) -> Position: if symbol not in self.positions: self.positions[symbol] = Position(symbol=symbol) return self.positions[symbol] def process_fill(self, fill: Fill) -> None: position = self.get_position(fill.order.symbol) position.update(fill) if fill.order.side == OrderSide.BUY: self.cash -= fill.fill_price * fill.fill_quantity + fill.commission else: self.cash += fill.fill_price * fill.fill_quantity - fill.commission def get_equity(self, prices: Dict[str, Decimal]) -> Decimal: equity = self.cash for symbol, position in self.positions.items(): if position.quantity != 0 and symbol in prices: equity += position.quantity * prices[symbol] return equity class Strategy(ABC): @abstractmethod def on_bar(self, timestamp: datetime, data: pd.DataFrame) -> List[Order]: pass @abstractmethod def on_fill(self, fill: Fill) -> None: pass class ExecutionModel(ABC): @abstractmethod def execute(self, order: Order, bar: pd.Series) -> Optional[Fill]: pass class SimpleExecutionModel(ExecutionModel): def __init__(self, slippage_bps: float = 10, commission_per_share: float = 0.01): self.slippage_bps = slippage_bps self.commission_per_share = commission_per_share def execute(self, order: Order, bar: pd.Series) -> Optional[Fill]: if order.order_type == OrderType.MARKET: base_price = Decimal(str(bar["open"])) # Apply slippage slippage_mult = 1 + (self.slippage_bps / 10000) if order.side == OrderSide.BUY: fill_price = base_price * Decimal(str(slippage_mult)) else: fill_price = base_price / Decimal(str(slippage_mult)) commission = order.quantity * Decimal(str(self.commission_per_share)) slippage = abs(fill_price - base_price) * order.quantity return Fill( order=order, fill_price=fill_price, fill_quantity=order.quantity, commission=commission, slippage=slippage, timestamp=bar.name ) return None class Backtester: def __init__( self, strategy: Strategy, execution_model: ExecutionModel, initial_capital: Decimal = Decimal("100000") ): self.strategy = strategy self.execution_model = execution_model self.portfolio = Portfolio(cash=initial_capital) self.equity_curve: List[tuple] = [] self.trades: List[Fill] = [] def run(self, data: pd.DataFrame) -> pd.DataFrame: """Run backtest on OHLCV data with DatetimeIndex.""" pending_orders: List[Order] = [] for timestamp, bar in data.iterrows(): # Execute pending orders at today's prices for order in pending_orders: fill = self.execution_model.execute(order, bar) if fill: self.portfolio.process_fill(fill) self.strategy.on_fill(fill) self.trades.append(fill) pending_orders.clear() # Get current prices for equity calculation prices = {data.index.name or "default": Decimal(str(bar["close"]))} equity = self.portfolio.get_equity(prices) self.equity_curve.append((timestamp, float(equity))) # Generate new orders for next bar new_orders = self.strategy.on_bar(timestamp, data.loc[:timestamp]) pending_orders.extend(new_orders) return self._create_results() def _create_results(self) -> pd.DataFrame: equity_df = pd.DataFrame(self.equity_curve, columns=["timestamp", "equity"]) equity_df.set_index("timestamp", inplace=True) equity_df["returns"] = equity_df["equity"].pct_change() return equity_df ``` ### Pattern 2: Vectorized Backtester (Fast) ```python import pandas as pd import numpy as np from typing import Callable, Dict, Any class VectorizedBacktester: """Fast vectorized backtester for simple strategies.""" def __init__( self, initial_capital: float = 100000, commission: float = 0.001, # 0.1% slippage: float = 0.0005 # 0.05% ): self.initial_capital = initial_capital self.commission = commission self.slippage = slippage def run( self, prices: pd.DataFrame, signal_func: Callable[[pd.DataFrame], pd.Series] ) -> Dict[str, Any]: """ Run backtest with signal function. Args: prices: DataFrame with 'close' column signal_func: Function that returns position signals (-1, 0, 1) Returns: Dictionary with results """ # Generate signals (shifted to avoid look-ahead) signals = signal_func(prices).shift(1).fillna(0) # Calculate returns returns = prices["close"].pct_change() # Calculate strategy returns with costs position_changes = signals.diff().abs() trading_costs = position_changes * (self.commission + self.slippage) strategy_returns = signals * returns - trading_costs # Build equity curve equity = (1 + strategy_returns).cumprod() * self.initial_capital # Calculate metrics results = { "equity": equity, "returns": strategy_returns, "signals": signals, "metrics": self._calculate_metrics(strategy_returns, equity) } return results def _calculate_metrics( self, returns: pd.Series, equity: pd.Series ) -> Dict[str, float]: """Calculate performance metrics.""" total_return = (equity.iloc[-1] / self.initial_capital) - 1 annual_return = (1 + total_return) ** (252 / len(returns)) - 1 annual_vol = returns.std() * np.sqrt(252) sharpe = annual_return / annual_vol if annual_vol > 0 else 0 # Drawdown rolling_max = equity.cummax() drawdown = (equity - rolling_max) / rolling_max max_drawdown = drawdown.min() # Win rate winning_days = (returns > 0).sum() total_days = (returns != 0).sum() win_rate = winning_days / total_days if total_days > 0 else 0 return { "total_return": total_return, "annual_return": annual_return, "annual_volatility": annual_vol, "sharpe_ratio": sharpe, "max_drawdown": max_drawdown, "win_rate": win_rate, "num_trades": int((returns != 0).sum()) } # Example usage def momentum_signal(prices: pd.DataFrame, lookback: int = 20) -> pd.Series: """Simple momentum strategy: long when price > SMA, else flat.""" sma = prices["close"].rolling(lookback).mean() return (prices["close"] > sma).astype(int) # Run backtest # backtester = VectorizedBacktester() # results = backtester.run(price_data, lambda p: momentum_signal(p, 50)) ``` ### Pattern 3: Walk-Forward Optimization ```python from typing import Callable, Dict, List, Tuple, Any import pandas as pd import numpy as np from itertools import product class WalkForwardOptimizer: """Walk-forward analysis with anchored or rolling windows.""" def __init__( self, train_period: int, test_period: int, anchored: bool = False, n_splits: int = None ): """ Args: train_period: Number of bars in training window test_period: Number of bars in test window anchored: If True, training always starts from beginning n_splits: Number of train/test splits (auto-calculated if None) """ self.train_period = train_period self.test_period = test_period self.anchored = anchored self.n_splits = n_splits def generate_splits( self, data: pd.DataFrame ) -> List[Tuple[pd.DataFrame, pd.DataFrame]]: """Generate train/test splits.""" splits = [] n = len(data) if self.n_splits: step = (n - self.train_period) // self.n_splits else: step = self.test_period start = 0 while start + self.train_period + self.test_period <= n: if self.anchored: train_start = 0 else: train_start = start train_end = start + self.train_period test_end = min(train_end + self.test_period, n) train_data = data.iloc[train_start:train_end] test_data = data.iloc[train_end:test_end] splits.append((train_data, test_data)) start += step return splits def optimize( self, data: pd.DataFrame, strategy_func: Callable, param_grid: Dict[str, List], metric: str = "sharpe_ratio" ) -> Dict[str, Any]: """ Run walk-forward optimization. Args: data: Full dataset strategy_func: Function(data, **params) -> results dict param_grid: Parameter combinations to test metric: Metric to optimize Returns: Combined results from all test periods """ splits = self.generate_splits(data) all_results = [] optimal_params_history = [] for i, (train_data, test_data) in enumerate(splits): # Optimize on training data best_params, best_metric = self._grid_search( train_data, strategy_func, param_grid, metric ) optimal_params_history.append(best_params) # Test with optimal params test_results = strategy_func(test_data, **best_params) test_results["split"] = i test_results["params"] = best_params all_results.append(test_results) print(f"Split {i+1}/{len(splits)}: " f"Best {metric}={best_metric:.4f}, params={best_params}") return { "split_results": all_results, "param_history": optimal_params_history, "combined_equity": self._combine_equity_curves(all_results) } def _grid_search( self, data: pd.DataFrame, strategy_func: Callable, param_grid: Dict[str, List], metric: str ) -> Tuple[Dict, float]: """Grid search for best parameters.""" best_params = None best_metric = -np.inf # Generate all parameter combinations param_names = list(param_grid.keys()) param_values = list(param_grid.values()) for values in product(*param_values): params = dict(zip(param_names, values)) results = strategy_func(data, **params) if results["metrics"][metric] > best_metric: best_metric = results["metrics"][metric] best_params = params return best_params, best_metric def _combine_equity_curves( self, results: List[Dict] ) -> pd.Series: """Combine equity curves from all test periods.""" combined = pd.concat([r["equity"] for r in results]) return combined ``` ### Pattern 4: Monte Carlo Analysis ```python import numpy as np import pandas as pd from typing import Dict, List class MonteCarloAnalyzer: """Monte Carlo simulation for strategy robustness.""" def __init__(self, n_simulations: int = 1000, confidence: float = 0.95): self.n_simulations = n_simulations self.confidence = confidence def bootstrap_returns( self, returns: pd.Series, n_periods: int = None ) -> np.ndarray: """ Bootstrap simulation by resampling returns. Args: returns: Historical returns series n_periods: Length of each simulation (default: same as input) Returns: Array of shape (n_simulations, n_periods) """ if n_periods is None: n_periods = len(returns) simulations = np.zeros((self.n_simulations, n_periods)) for i in range(self.n_simulations): # Resample with replacement simulated_returns = np.random.choice( returns.values, size=n_periods, replace=True ) simulations[i] = simulated_returns return simulations def analyze_drawdowns( self, returns: pd.Series ) -> Dict[str, float]: """Analyze drawdown distribution via simulation.""" simulations = self.bootstrap_returns(returns) max_drawdowns = [] for sim_returns in simulations: equity = (1 + sim_returns).cumprod() rolling_max = np.maximum.accumulate(equity) drawdowns = (equity - rolling_max) / rolling_max max_drawdowns.append(drawdowns.min()) max_drawdowns = np.array(max_drawdowns) return { "expected_max_dd": np.mean(max_drawdowns), "median_max_dd": np.median(max_drawdowns), f"worst_{int(self.confidence*100)}pct": np.percentile( max_drawdowns, (1 - self.confidence) * 100 ), "worst_case": max_drawdowns.min() } def probability_of_loss( self, returns: pd.Series, holding_periods: List[int] = [21, 63, 126, 252] ) -> Dict[int, float]: """Calculate probability of loss over various holding periods.""" results = {} for period in holding_periods: if period > len(returns): continue simulations = self.bootstrap_returns(returns, period) total_returns = (1 + simulations).prod(axis=1) - 1 prob_loss = (total_returns < 0).mean() results[period] = prob_loss return results def confidence_interval( self, returns: pd.Series, periods: int = 252 ) -> Dict[str, float]: """Calculate confidence interval for future returns.""" simulations = self.bootstrap_returns(returns, periods) total_returns = (1 + simulations).prod(axis=1) - 1 lower = (1 - self.confidence) / 2 upper = 1 - lower return { "expected": total_returns.mean(), "lower_bound": np.percentile(total_returns, lower * 100), "upper_bound": np.percentile(total_returns, upper * 100), "std": total_returns.std() } ``` ## Performance Metrics ```python def calculate_metrics(returns: pd.Series, rf_rate: float = 0.02) -> Dict[str, float]: """Calculate comprehensive performance metrics.""" # Annualization factor (assuming daily returns) ann_factor = 252 # Basic metrics total_return = (1 + returns).prod() - 1 annual_return = (1 + total_return) ** (ann_factor / len(returns)) - 1 annual_vol = returns.std() * np.sqrt(ann_factor) # Risk-adjusted returns sharpe = (annual_return - rf_rate) / annual_vol if annual_vol > 0 else 0 # Sortino (downside deviation) downside_returns = returns[returns < 0] downside_vol = downside_returns.std() * np.sqrt(ann_factor) sortino = (annual_return - rf_rate) / downside_vol if downside_vol > 0 else 0 # Calmar ratio equity = (1 + returns).cumprod() rolling_max = equity.cummax() drawdowns = (equity - rolling_max) / rolling_max max_drawdown = drawdowns.min() calmar = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0 # Win rate and profit factor wins = returns[returns > 0] losses = return
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

rag-implementation

Build Retrieval-Augmented Generation (RAG) systems for LLM

coding
⭐1
# RAG Implementation Master Retrieval-Augmented Generation (RAG) to build LLM applications that provide accurate, grounded responses using external knowledge sources. ## When to Use This Skill - Building Q&A systems over proprietary documents - Creating chatbots with current, factual information - Implementing semantic search with natural language queries - Reducing hallucinations with grounded responses - Enabling LLMs to access domain-specific knowledge - Building documentation assistants - Creating research tools with source citation ## Core Components ### 1. Vector Databases **Purpose**: Store and retrieve document embeddings efficiently **Options:** - **Pinecone**: Managed, scalable, serverless - **Weaviate**: Open-source, hybrid search, GraphQL - **Milvus**: High performance, on-premise - **Chroma**: Lightweight, easy to use, local development - **Qdrant**: Fast, filtered search, Rust-based - **pgvector**: PostgreSQL extension, SQL integration ### 2. Embeddings **Purpose**: Convert text to numerical vectors for similarity search **Models (2026):** | Model | Dimensions | Best For | |-------|------------|----------| | **voyage-3-large** | 1024 | Claude apps (Anthropic recommended) | | **voyage-code-3** | 1024 | Code search | | **text-embedding-3-large** | 3072 | OpenAI apps, high accuracy | | **text-embedding-3-small** | 1536 | OpenAI apps, cost-effective | | **bge-large-en-v1.5** | 1024 | Open source, local deployment | | **multilingual-e5-large** | 1024 | Multi-language support | ### 3. Retrieval Strategies **Approaches:** - **Dense Retrieval**: Semantic similarity via embeddings - **Sparse Retrieval**: Keyword matching (BM25, TF-IDF) - **Hybrid Search**: Combine dense + sparse with weighted fusion - **Multi-Query**: Generate multiple query variations - **HyDE**: Generate hypothetical documents for better retrieval ### 4. Reranking **Purpose**: Improve retrieval quality by reordering results **Methods:** - **Cross-Encoders**: BERT-based reranking (ms-marco-MiniLM) - **Cohere Rerank**: API-based reranking - **Maximal Marginal Relevance (MMR)**: Diversity + relevance - **LLM-based**: Use LLM to score relevance ## Quick Start with LangGraph ```python from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_text_splitters import RecursiveCharacterTextSplitter from typing import TypedDict, Annotated class RAGState(TypedDict): question: str context: list[Document] answer: str # Initialize components llm = ChatAnthropic(model="claude-sonnet-4-6") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # RAG prompt rag_prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so. Context: {context} Question: {question} Answer:""" ) async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs} async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" context_text = "\n\n".join(doc.page_content for doc in state["context"]) messages = rag_prompt.format_messages( context=context_text, question=state["question"] ) response = await llm.ainvoke(messages) return {"answer": response.content} # Build RAG graph builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) rag_chain = builder.compile() # Use result = await rag_chain.ainvoke({"question": "What are the main features?"}) print(result["answer"]) ``` ## Advanced RAG Patterns ### Pattern 1: Hybrid Search with RRF ```python from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever # Sparse retriever (BM25 for keyword matching) bm25_retriever = BM25Retriever.from_documents(documents) bm25_retriever.k = 10 # Dense retriever (embeddings for semantic search) dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10}) # Combine with Reciprocal Rank Fusion weights ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, dense_retriever], weights=[0.3, 0.7] # 30% keyword, 70% semantic ) ``` ### Pattern 2: Multi-Query Retrieval ```python from langchain.retrievers.multi_query import MultiQueryRetriever # Generate multiple query perspectives for better recall multi_query_retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(search_kwargs={"k": 5}), llm=llm ) # Single query β†’ multiple variations β†’ combined results results = await multi_query_retriever.ainvoke("What is the main topic?") ``` ### Pattern 3: Contextual Compression ```python from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # Compressor extracts only relevant portions compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}) ) # Returns only relevant parts of documents compressed_docs = await compression_retriever.ainvoke("specific query") ``` ### Pattern 4: Parent Document Retriever ```python from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore from langchain_text_splitters import RecursiveCharacterTextSplitter # Small chunks for precise retrieval, large chunks for context child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50) parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200) # Store for parent documents docstore = InMemoryStore() parent_retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=docstore, child_splitter=child_splitter, parent_splitter=parent_splitter ) # Add documents (splits children, stores parents) await parent_retriever.aadd_documents(documents) # Retrieval returns parent documents with full context results = await parent_retriever.ainvoke("query") ``` ### Pattern 5: HyDE (Hypothetical Document Embeddings) ```python from langchain_core.prompts import ChatPromptTemplate class HyDEState(TypedDict): question: str hypothetical_doc: str context: list[Document] answer: str hyde_prompt = ChatPromptTemplate.from_template( """Write a detailed passage that would answer this question: Question: {question} Passage:""" ) async def generate_hypothetical(state: HyDEState) -> HyDEState: """Generate hypothetical document for better retrieval.""" messages = hyde_prompt.format_messages(question=state["question"]) response = await llm.ainvoke(messages) return {"hypothetical_doc": response.content} async def retrieve_with_hyde(state: HyDEState) -> HyDEState: """Retrieve using hypothetical document.""" # Use hypothetical doc for retrieval instead of original query docs = await retriever.ainvoke(state["hypothetical_doc"]) return {"context": docs} # Build HyDE RAG graph builder = StateGraph(HyDEState) builder.add_node("hypothetical", generate_hypothetical) builder.add_node("retrieve", retrieve_with_hyde) builder.add_node("generate", generate) builder.add_edge(START, "hypothetical") builder.add_edge("hypothetical", "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) hyde_rag = builder.compile() ``` ## Document Chunking Strategies ### Recursive Character Text Splitter ```python from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", ". ", " ", ""] # Try in order ) chunks = splitter.split_documents(documents) ``` ### Token-Based Splitting ```python from langchain_text_splitters import TokenTextSplitter splitter = TokenTextSplitter( chunk_size=512, chunk_overlap=50, encoding_name="cl100k_base" # OpenAI tiktoken encoding ) ``` ### Semantic Chunking ```python from langchain_experimental.text_splitter import SemanticChunker splitter = SemanticChunker( embeddings=embeddings, breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 ) ``` ### Markdown Header Splitter ```python from langchain_text_splitters import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False ) ``` ## Vector Store Configurations ### Pinecone (Serverless) ```python from pinecone import Pinecone, ServerlessSpec from langchain_pinecone import PineconeVectorStore # Initialize Pinecone client pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) # Create index if needed if "my-index" not in pc.list_indexes().names(): pc.create_index( name="my-index", dimension=1024, # voyage-3-large dimensions metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1") ) # Create vector store index = pc.Index("my-index") vectorstore = PineconeVectorStore(index=index, embedding=embeddings) ``` ### Weaviate ```python import weaviate from langchain_weaviate import WeaviateVectorStore client = weaviate.connect_to_local() # or connect_to_weaviate_cloud() vectorstore = WeaviateVectorStore( client=client, index_name="Documents", text_key="content", embedding=embeddings ) ``` ### Chroma (Local Development) ```python from langchain_chroma import Chroma vectorstore = Chroma( collection_name="my_collection", embedding_function=embeddings, persist_directory="./chroma_db" ) ``` ### pgvector (PostgreSQL) ```python from langchain_postgres.vectorstores import PGVector connection_string = "postgresql+psycopg://user:pass@localhost:5432/vectordb" vectorstore = PGVector( embeddings=embeddings, collection_name="documents", connection=connection_string, ) ``` ## Retrieval Optimization ### 1. Metadata Filtering ```python from langchain_core.documents import Document # Add metadata during indexing docs_with_metadata = [] for doc in documents: doc.metadata.update({ "source": doc.metadata.get("source", "unknown"), "category": determine_category(doc.page_content), "date": datetime.now().isoformat() }) docs_with_metadata.append(doc) # Filter during retrieval results = await vectorstore.asimilarity_search( "query", filter={"category": "technical"}, k=5 ) ``` ### 2. Maximal Marginal Relevance (MMR) ```python # Balance relevance with diversity results = await vectorstore.amax_marginal_relevance_search( "query", k=5, fetch_k=20, # Fetch 20, return top 5 diverse lambda_mult=0.5 # 0=max diversity, 1=max relevance ) ``` ### 3. Reranking with Cross-Encoder ```python from sentence_transformers import CrossEncoder reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') async def retrieve_and_rerank(query: str, k: int = 5) -> list[Document]: # Get initial results candidates = await vectorstore.asimilarity_search(query, k=20) # Rerank pairs = [[query, doc.page_content] for doc in candidates] scores = reranker.predict(pairs) # Sort by score and take top k ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return [doc for doc, score in ranked[:k]] ``` ### 4. Cohere Rerank ```python from langchain.retrievers import CohereRerank from langchain_cohere import CohereRerank reranker = CohereRerank(model="rerank-english-v3.0", top_n=5) # Wrap retriever with reranking reranked_retriever = ContextualCompressionRetriever( base_compressor=reranker, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20}) ) ``` ## Prompt Engineering for RAG ### Contextual Prompt with Citations ```python rag_prompt = ChatPromptTemplate.from_template( """Answer the question based on the context below. Include citations using [1], [2], etc. If you cannot answer based on the context, say "I don't have enough information." Context: {context} Question: {question} Instructions: 1. Use only information from the context 2. Cite sources with [1], [2] format 3. If uncertain, express uncertainty Answer (with citations):""" ) ``` ### Structured Output for RAG ```python from pydantic import BaseModel, Field class RAGResponse(BaseModel): answer: str = Field(description="The answer based on context") confidence: float = Field(description="Confidence score 0-1") sources: list[str] = Field(description="Source document IDs used") reasoning: str = Field(description="Brief reasoning for the answer") # Use with structured output structured_llm = llm.with_structured_output(RAGResponse) ``` ## Evaluation Metrics ```python from typing import TypedDict class RAGEvalMetrics(TypedDict): retrieval_precision: float # Relevant docs / retrieved docs retrieval_recall: float # Retrieved relevant / total relevant answer_relevance: float # Answer addresses question faithfulness: float # Answer grounded in context context_relevance: float # Context relevant to question async def evaluate_rag_system( rag_chain, test_cases: list[dict] ) -> RAGEvalMetrics: """Evaluate RAG system on test cases.""" metrics = {k: [] for k in RAGEvalMetrics.__annotations__} for test in test_cases: result = await rag_chain.ainvoke({"question": test["question"]}) # Retrieval metrics retrieved_ids = {doc.metadata["id"] for doc in result["context"]} relevant_ids = set(test["relevant_doc_ids"]) precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids) recall = len(retrieved_ids & relevant_ids) / len(relevant_ids) metrics["retrieval_precision"].append(precision) metrics["retrieval_recall"].append(recall) # Use LLM-as-judge for quality metrics quality = await evaluate_answer_quality( question=test["question"], answer=result["answer"], context=result["context"], expected=test.get("expected_answer") ) metrics["answer_relevance"].append(quality["relevance"]) metrics["faithfulness"].append(quality["faithfulness"]) metrics["context_relevance"].append(quality["context_relevance"]) return {k: sum(v) / len(v) for k, v in metrics.items()} ``` ## Resources - [LangChain RAG Tutorial](https://python.langchain.com/docs/tutorials/rag/) - [LangGraph RAG Examples](https://langchain-ai.github.io/langgraph/tutorials/rag/) - [Pinecone Best Practices](https://docs.pinecone.io/guides/get-started/overview) - [Voyage AI Embeddings](https://docs.voyageai.com/) - [RAG Evaluation Guide](https://docs.ragas.io/) ## Best Practices 1. **Chunk Size**: Balance between context (larger) and specificity (smaller) - typically 500-1000 tokens 2. **Overlap**: Use 10-20% overlap to preserve context at boundaries 3. **Metadata**: Include source, page, timestamp for filtering and debugging 4. **Hybrid Search**: Combine semantic and keyword search for best recall 5. **Reranking**: Use cross-encoder reranking for precision-critical applications 6. **Citations**: Always return source documents for transparency 7. **Evaluation**: Continuously test retrieval quality and answer accuracy 8. **Monitoring**: Track retrieval metrics and latency in production ## Common Issues - **Poor Retrieval**: Check embedding quality, chunk size, query formulation - **Irrelevant Results**: Add metadata filtering, use hybrid search, rerank - **Missing Information**: Ensure documents are properly indexed, check chunking - **Slow Queries**: Optimize vector store, use caching, reduce k - **Hallucinations**: Improve grounding prompt, add verification step - **Context Too Long**: Use compression or parent document retriever
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

data-quality-frameworks

Implement data quality validation with Great Expectations, dbt

data
⭐1
# Data Quality Frameworks Production patterns for implementing data quality with Great Expectations, dbt tests, and data contracts to ensure reliable data pipelines. ## When to Use This Skill - Implementing data quality checks in pipelines - Setting up Great Expectations validation - Building comprehensive dbt test suites - Establishing data contracts between teams - Monitoring data quality metrics - Automating data validation in CI/CD ## Core Concepts ### 1. Data Quality Dimensions | Dimension | Description | Example Check | | ---------------- | ------------------------ | -------------------------------------------------- | | **Completeness** | No missing values | `expect_column_values_to_not_be_null` | | **Uniqueness** | No duplicates | `expect_column_values_to_be_unique` | | **Validity** | Values in expected range | `expect_column_values_to_be_in_set` | | **Accuracy** | Data matches reality | Cross-reference validation | | **Consistency** | No contradictions | `expect_column_pair_values_A_to_be_greater_than_B` | | **Timeliness** | Data is recent | `expect_column_max_to_be_between` | ### 2. Testing Pyramid for Data ``` /\ / \ Integration Tests (cross-table) /────\ / \ Unit Tests (single column) /────────\ / \ Schema Tests (structure) /────────────\ ``` ## Quick Start ### Great Expectations Setup ```bash # Install pip install great_expectations # Initialize project great_expectations init # Create datasource great_expectations datasource new ``` ```python # great_expectations/checkpoints/daily_validation.yml import great_expectations as gx # Create context context = gx.get_context() # Create expectation suite suite = context.add_expectation_suite("orders_suite") # Add expectations suite.add_expectation( gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id") ) suite.add_expectation( gx.expectations.ExpectColumnValuesToBeUnique(column="order_id") ) # Validate results = context.run_checkpoint(checkpoint_name="daily_orders") ``` ## Patterns ### Pattern 1: Great Expectations Suite ```python # expectations/orders_suite.py import great_expectations as gx from great_expectations.core import ExpectationSuite from great_expectations.core.expectation_configuration import ExpectationConfiguration def build_orders_suite() -> ExpectationSuite: """Build comprehensive orders expectation suite""" suite = ExpectationSuite(expectation_suite_name="orders_suite") # Schema expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_columns_to_match_set", kwargs={ "column_set": ["order_id", "customer_id", "amount", "status", "created_at"], "exact_match": False # Allow additional columns } )) # Primary key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "order_id"} )) suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_unique", kwargs={"column": "order_id"} )) # Foreign key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "customer_id"} )) # Categorical values suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_in_set", kwargs={ "column": "status", "value_set": ["pending", "processing", "shipped", "delivered", "cancelled"] } )) # Numeric ranges suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_between", kwargs={ "column": "amount", "min_value": 0, "max_value": 100000, "strict_min": True # amount > 0 } )) # Date validity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_dateutil_parseable", kwargs={"column": "created_at"} )) # Freshness - data should be recent suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_max_to_be_between", kwargs={ "column": "created_at", "min_value": {"$PARAMETER": "now - timedelta(days=1)"}, "max_value": {"$PARAMETER": "now"} } )) # Row count sanity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_row_count_to_be_between", kwargs={ "min_value": 1000, # Expect at least 1000 rows "max_value": 10000000 } )) # Statistical expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_mean_to_be_between", kwargs={ "column": "amount", "min_value": 50, "max_value": 500 } )) return suite ``` ### Pattern 2: Great Expectations Checkpoint ```yaml # great_expectations/checkpoints/orders_checkpoint.yml name: orders_checkpoint config_version: 1.0 class_name: Checkpoint run_name_template: "%Y%m%d-%H%M%S-orders-validation" validations: - batch_request: datasource_name: warehouse data_connector_name: default_inferred_data_connector_name data_asset_name: orders data_connector_query: index: -1 # Latest batch expectation_suite_name: orders_suite action_list: - name: store_validation_result action: class_name: StoreValidationResultAction - name: store_evaluation_parameters action: class_name: StoreEvaluationParametersAction - name: update_data_docs action: class_name: UpdateDataDocsAction # Slack notification on failure - name: send_slack_notification action: class_name: SlackNotificationAction slack_webhook: ${SLACK_WEBHOOK} notify_on: failure renderer: module_name: great_expectations.render.renderer.slack_renderer class_name: SlackRenderer ``` ```python # Run checkpoint import great_expectations as gx context = gx.get_context() result = context.run_checkpoint(checkpoint_name="orders_checkpoint") if not result.success: failed_expectations = [ r for r in result.run_results.values() if not r.success ] raise ValueError(f"Data quality check failed: {failed_expectations}") ``` ### Pattern 3: dbt Data Tests ```yaml # models/marts/core/_core__models.yml version: 2 models: - name: fct_orders description: Order fact table tests: # Table-level tests - dbt_utils.recency: datepart: day field: created_at interval: 1 - dbt_utils.at_least_one - dbt_utils.expression_is_true: expression: "total_amount >= 0" columns: - name: order_id description: Primary key tests: - unique - not_null - name: customer_id description: Foreign key to dim_customers tests: - not_null - relationships: to: ref('dim_customers') field: customer_id - name: order_status tests: - accepted_values: values: ["pending", "processing", "shipped", "delivered", "cancelled"] - name: total_amount tests: - not_null - dbt_utils.expression_is_true: expression: ">= 0" - name: created_at tests: - not_null - dbt_utils.expression_is_true: expression: "<= current_timestamp" - name: dim_customers columns: - name: customer_id tests: - unique - not_null - name: email tests: - unique - not_null # Custom regex test - dbt_utils.expression_is_true: expression: "email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'" ``` ### Pattern 4: Custom dbt Tests ```sql -- tests/generic/test_row_count_in_range.sql {% test row_count_in_range(model, min_count, max_count) %} with row_count as ( select count(*) as cnt from {{ model }} ) select cnt from row_count where cnt < {{ min_count }} or cnt > {{ max_count }} {% endtest %} -- Usage in schema.yml: -- tests: -- - row_count_in_range: -- min_count: 1000 -- max_count: 10000000 ``` ```sql -- tests/generic/test_sequential_values.sql {% test sequential_values(model, column_name, interval=1) %} with lagged as ( select {{ column_name }}, lag({{ column_name }}) over (order by {{ column_name }}) as prev_value from {{ model }} ) select * from lagged where {{ column_name }} - prev_value != {{ interval }} and prev_value is not null {% endtest %} ``` ```sql -- tests/singular/assert_orders_customers_match.sql -- Singular test: specific business rule with orders_customers as ( select distinct customer_id from {{ ref('fct_orders') }} ), dim_customers as ( select customer_id from {{ ref('dim_customers') }} ), orphaned_orders as ( select o.customer_id from orders_customers o left join dim_customers c using (customer_id) where c.customer_id is null ) select * from orphaned_orders -- Test passes if this returns 0 rows ``` ### Pattern 5: Data Contracts ```yaml # contracts/orders_contract.yaml apiVersion: datacontract.com/v1.0.0 kind: DataContract metadata: name: orders version: 1.0.0 owner: data-platform-team contact: data-team@company.com info: title: Orders Data Contract description: Contract for order event data from the ecommerce platform purpose: Analytics, reporting, and ML features servers: production: type: snowflake account: company.us-east-1 database: ANALYTICS schema: CORE terms: usage: Internal analytics only limitations: PII must not be exposed in downstream marts billing: Charged per query TB scanned schema: type: object properties: order_id: type: string format: uuid description: Unique order identifier required: true unique: true pii: false customer_id: type: string format: uuid description: Customer identifier required: true pii: true piiClassification: indirect total_amount: type: number minimum: 0 maximum: 100000 description: Order total in USD created_at: type: string format: date-time description: Order creation timestamp required: true status: type: string enum: [pending, processing, shipped, delivered, cancelled] description: Current order status quality: type: SodaCL specification: checks for orders: - row_count > 0 - missing_count(order_id) = 0 - duplicate_count(order_id) = 0 - invalid_count(status) = 0: valid values: [pending, processing, shipped, delivered, cancelled] - freshness(created_at) < 24h sla: availability: 99.9% freshness: 1 hour latency: 5 minutes ``` ### Pattern 6: Automated Quality Pipeline ```python # quality_pipeline.py from dataclasses import dataclass from typing import List, Dict, Any import great_expectations as gx from datetime import datetime @dataclass class QualityResult: table: str passed: bool total_expectations: int failed_expectations: int details: List[Dict[str, Any]] timestamp: datetime class DataQualityPipeline: """Orchestrate data quality checks across tables""" def __init__(self, context: gx.DataContext): self.context = context self.results: List[QualityResult] = [] def validate_table(self, table: str, suite: str) -> QualityResult: """Validate a single table against expectation suite""" checkpoint_config = { "name": f"{table}_validation", "config_version": 1.0, "class_name": "Checkpoint", "validations": [{ "batch_request": { "datasource_name": "warehouse", "data_asset_name": table, }, "expectation_suite_name": suite, }], } result = self.context.run_checkpoint(**checkpoint_config) # Parse results validation_result = list(result.run_results.values())[0] results = validation_result.results failed = [r for r in results if not r.success] return QualityResult( table=table, passed=result.success, total_expectations=len(results), failed_expectations=len(failed), details=[{ "expectation": r.expectation_config.expectation_type, "success": r.success, "observed_value": r.result.get("observed_value"), } for r in results], timestamp=datetime.now() ) def run_all(self, tables: Dict[str, str]) -> Dict[str, QualityResult]: """Run validation for all tables""" results = {} for table, suite in tables.items(): print(f"Validating {table}...") results[table] = self.validate_table(table, suite) return results def generate_report(self, results: Dict[str, QualityResult]) -> str: """Generate quality report""" report = ["# Data Quality Report", f"Generated: {datetime.now()}", ""] total_passed = sum(1 for r in results.values() if r.passed) total_tables = len(results) report.append(f"## Summary: {total_passed}/{total_tables} tables passed") report.append("") for table, result in results.items(): status = "βœ…" if result.passed else "❌" report.append(f"### {status} {table}") report.append(f"- Expectations: {result.total_expectations}") report.append(f"- Failed: {result.failed_expectations}") if not result.passed: report.append("- Failed checks:") for detail in result.details: if not detail["success"]: report.append(f" - {detail['expectation']}: {detail['observed_value']}") report.append("") return "\n".join(report) # Usage context = gx.get_context() pipeline = DataQualityPipeline(context) tables_to_validate = { "orders": "orders_suite", "customers": "customers_suite", "products": "products_suite", } results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results) # Fail pipeline if any table failed if not all(r.passed for r in results.values()): print(report) raise ValueError("Data quality checks failed!") ``` ## Best Practices ### Do's - **Test early** - Validate source data before transformations - **Test incrementally** - Add tests as you find issues - **Document expectations** - Clear descriptions for each test - **Alert on failures** - Integrate with monitoring - **Version contracts** - Track schema changes ### Don'ts - **Don't test everything** - Focus on critical columns - **Don't ignore warnings** - They often precede failures - **Don't skip freshness** - Stale data is bad data - **Don't hardcode thresholds** - Use dynamic baselines - **Don't test in isolation** - Test relationships too ## Resources - [Great Expectations Documentation](https://docs.greatexpectations.io/) - [dbt Testing Documentation](https://docs.getdbt.com/docs/build/tests) - [Data Contract Specification](https://datacontract.com/) - [Soda Core](https://docs.soda.io/soda-core/overview.html)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered