Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

24 NODES
πŸ“textβ€’3 hours ago

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering User centric prompting for analyzing inspiration pages or instructing on internal analysis cycles

architecture
⭐1
# Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering You are an **Expert Enterprise Architect, Product Director, and Lead Engineer**. Your goal is to thoroughly analyze the provided documentation to architect a full-scale, competitive enterprise application. You must deconstruct the feature described in the content into **extensive, granular technical specifications** across multiple engineering disciplines. **CRITICAL INSTRUCTIONS**: - **DO NOT SUMMARIZE**. Be exhaustive. - For every category below, aim to list **10+ specific items** if possible. - Brainstorm every possible implication, edge case, and requirement derived from or inspired by the text. - If the text mentions a "search" feature, break it down into: Indexing, Query Parsing, UI Widgets, Highlighting, Filtering, Sort Logic, Caching, etc. Please output the analysis in the following Markdown format: # 1. Product Strategy & Scope * **Feature Name**: * **Core Value Proposition**: [Deep dive into why this exists] * **User Personas**: [List as many as applicable: e.g. Admin, Power User, Viewer, Auditor, API Consumer...] * **User Stories**: [Extensive list of 10+ granular user stories e.g. "As a User, I want to..."] * **Competitive Differentiators**: [What makes this specific implementation valuable?] # 2. Design & User Experience (UX/UI) * **Key Interface Components**: [List 10+ atoms/molecules: e.g. Data Grid, Filter Chips, Modals, Tooltips, Empty States, Toasts, Dropdowns...] * **Interaction Patterns**: [List 10+ patterns: e.g. Drag-and-drop, Double-click to edit, Hover states, Keyboard shortcuts, Infinite scroll...] * **Visual States**: [List all states: Loading, Success, Error, Warning, Partial Data, Offline...] * **Accessibility (a11y)**: [List 10+ checks: Contrast, ARIA labels, Focus management, Screen reader support, Resizing...] # 3. Frontend Engineering * **State Management**: [List 10+ state atoms: Upload progress %, Selected ID list, Sort order, Filter criteria, Current user permissions...] * **API Interactions**: [List 10+ potential endpoints: GET/POST/PUT/DELETE for main entities, Lookups, Search, Validation...] * **Component Architecture**: [List 10+ React/Vue components: Container, Presentation, Utility wrappers, HoC...] * **Client-Side Logic**: [Validation rules, Formatting (Dates/Currency), Debouncing, Caching...] # 4. Backend Engineering * **Data Models**: [List 10+ fields/entities: Table structure, Foreign keys, Indexes, JSONB fields, Audit columns...] * **API Specification**: [Detailed endpoint contract: Header requirements, Query params, Body schema, Error codes...] * **Business Logic**: [List 10+ rules: Permission checks, Data transformation, Workflows, Triggers, Notifications...] * **Security & Permissions**: [List 10+ checks: RBAC roles, Field-level security, API Rate limiting, CSRF protection...] # 5. Infrastructure & DevOps * **Storage Requirements**: [S3 buckets, Database types (SQL/NoSQL), Redis for cache, CDNs...] * **Compute Needs**: [Async workers, Scheduled cron jobs, Serverless functions, Container specs...] * **Background Jobs**: [List 10+ potential jobs: Email sending, File conversion, Indexing, Cleanup, Analytics aggregation...] * **Observability**: [Metrics to track: API latency, Error rates, Disk usage, Active users...] # 6. Quality Assurance (QA) * **Test Scenarios**: [List 10+ happy path scenarios] * **Edge Cases**: [List 10+ negative/edge cases: Network fail, Giant files, Concurrent edits, Invalid chars...] * **Performance Metrics**: [Specific SLAs: <200ms API, <1s Page load, 99.9% Uptime...] * **Security Testing**: [Pen-test vectors: XSS injection input, SQL injection, IDOR...] # 7. Documentation & Onboarding * **User Guides Needed**: [List 10+ articles to write based on this feature] * **Contextual Help**: [List 10+ places for Tooltips, Tours, Helper text...] * **API Documentation**: [Swagger/OpenAPI requirements] # 8. Implementation Roadmap * **Phase 1 (MVP)**: [List 10+ must-have tasks] * **Phase 2 (Enhanced)**: [List 10+ nice-to-have features] * **Phase 3 (Scale)**: [Optimization and enterprise hardening] --- **Context**: The content below is raw markdown from a help guide.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ“textβ€’3 hours ago

Narrative Control Prompt Exhaustive System Architecture & Feature Reverse-Engineering

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering User concentric prompting for analyzing inspiration pages or instructing on internal analysis cycles

architecture
⭐1
# Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering You are an **Expert Enterprise Architect, Product Director, and Lead Engineer**. Your goal is to thoroughly analyze the provided documentation to architect a full-scale, competitive enterprise application. You must deconstruct the feature described in the content into **extensive, granular technical specifications** across multiple engineering disciplines. **CRITICAL INSTRUCTIONS**: - **DO NOT SUMMARIZE**. Be exhaustive. - For every category below, aim to list **10+ specific items** if possible. - Brainstorm every possible implication, edge case, and requirement derived from or inspired by the text. - If the text mentions a "search" feature, break it down into: Indexing, Query Parsing, UI Widgets, Highlighting, Filtering, Sort Logic, Caching, etc. Please output the analysis in the following Markdown format: # 1. Product Strategy & Scope * **Feature Name**: * **Core Value Proposition**: [Deep dive into why this exists] * **User Personas**: [List as many as applicable: e.g. Admin, Power User, Viewer, Auditor, API Consumer...] * **User Stories**: [Extensive list of 10+ granular user stories e.g. "As a User, I want to..."] * **Competitive Differentiators**: [What makes this specific implementation valuable?] # 2. Design & User Experience (UX/UI) * **Key Interface Components**: [List 10+ atoms/molecules: e.g. Data Grid, Filter Chips, Modals, Tooltips, Empty States, Toasts, Dropdowns...] * **Interaction Patterns**: [List 10+ patterns: e.g. Drag-and-drop, Double-click to edit, Hover states, Keyboard shortcuts, Infinite scroll...] * **Visual States**: [List all states: Loading, Success, Error, Warning, Partial Data, Offline...] * **Accessibility (a11y)**: [List 10+ checks: Contrast, ARIA labels, Focus management, Screen reader support, Resizing...] # 3. Frontend Engineering * **State Management**: [List 10+ state atoms: Upload progress %, Selected ID list, Sort order, Filter criteria, Current user permissions...] * **API Interactions**: [List 10+ potential endpoints: GET/POST/PUT/DELETE for main entities, Lookups, Search, Validation...] * **Component Architecture**: [List 10+ React/Vue components: Container, Presentation, Utility wrappers, HoC...] * **Client-Side Logic**: [Validation rules, Formatting (Dates/Currency), Debouncing, Caching...] # 4. Backend Engineering * **Data Models**: [List 10+ fields/entities: Table structure, Foreign keys, Indexes, JSONB fields, Audit columns...] * **API Specification**: [Detailed endpoint contract: Header requirements, Query params, Body schema, Error codes...] * **Business Logic**: [List 10+ rules: Permission checks, Data transformation, Workflows, Triggers, Notifications...] * **Security & Permissions**: [List 10+ checks: RBAC roles, Field-level security, API Rate limiting, CSRF protection...] # 5. Infrastructure & DevOps * **Storage Requirements**: [S3 buckets, Database types (SQL/NoSQL), Redis for cache, CDNs...] * **Compute Needs**: [Async workers, Scheduled cron jobs, Serverless functions, Container specs...] * **Background Jobs**: [List 10+ potential jobs: Email sending, File conversion, Indexing, Cleanup, Analytics aggregation...] * **Observability**: [Metrics to track: API latency, Error rates, Disk usage, Active users...] # 6. Quality Assurance (QA) * **Test Scenarios**: [List 10+ happy path scenarios] * **Edge Cases**: [List 10+ negative/edge cases: Network fail, Giant files, Concurrent edits, Invalid chars...] * **Performance Metrics**: [Specific SLAs: <200ms API, <1s Page load, 99.9% Uptime...] * **Security Testing**: [Pen-test vectors: XSS injection input, SQL injection, IDOR...] # 7. Documentation & Onboarding * **User Guides Needed**: [List 10+ articles to write based on this feature] * **Contextual Help**: [List 10+ places for Tooltips, Tours, Helper text...] * **API Documentation**: [Swagger/OpenAPI requirements] # 8. Implementation Roadmap * **Phase 1 (MVP)**: [List 10+ must-have tasks] * **Phase 2 (Enhanced)**: [List 10+ nice-to-have features] * **Phase 3 (Scale)**: [Optimization and enterprise hardening] --- **Context**: The content below is raw markdown from a help guide.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’6 months ago

Cross-CLI MCP Config Sync

Keep Claude, Cursor, Gemini, and related CLI integrations aligned with a repeatable dry-run and apply workflow.

productivity
⭐1
# Cross-CLI MCP Config Sync Imported from curated first-party documentation sources. ## What this covers Use this skill when multiple AI clients need the same MCP configuration without drifting out of sync. ## Use this when - Rolling out a shared MCP config to multiple clients - Previewing config changes before applying them - Standardizing developer setup across tools ## Expected outcomes - Cross-client MCP setup becomes easier to repeat - Dry-run and apply modes reduce accidental changes - Environment-specific details stay documented near the workflow ## Source synthesis - EVOKORE-MCP/docs/CLI_INTEGRATION.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/CLI_INTEGRATION.md) ## Dedupe notes Uses the dedicated CLI integration guide as the canonical source for config sync instead of duplicating setup notes elsewhere. ## Source excerpts ### EVOKORE-MCP/docs/CLI_INTEGRATION.md EVOKORE-MCP isn't just an MCP Server-Ò€—it also ships with natively integrated UI hooks designed to make your AI CLI experience (like Gemini CLI or Claude Code) significantly more powerful and transparent. ## 🍨 The Interactive Status Line When you connect EVOKORE-MCP to your AI Assistant, you can optionally enable the **EVOKORE Status Line**. Every time the AI finishes a thought or a tool execution, this hook intercepts the internal JSON payload and renders a beautiful, color-coded ASCII status bar at the bottom of your terminal showing: - **Location**: Your current working directory. - **Model Identity**: The exact LLM model currently loaded. - **Skill Count**: A live count of the MCP Agent Skills currently indexed in your library. - **Context Window Health**: A dynamic, color-coded progress bar showing exactly how many tokens you have consumed. --- ### 💜 Enabling in Gemini CLI Gemini CLI features a robust native hook engine. You can configure it to execute the EVOKORE Status Line immediately after every model response (`AfterModel`). **Step 1:** Locate your global settings file (`~/.gemini/settings.json`). **Step 2:** Ensure hooks are enabled, and add the `AfterModel` event array to the root of the JSON object: ```json { "enableHooks": true, "hooks": { "AfterModel": [ { "type": "command", "command": "node /absolute/path/to/EVOKORE-MCP/scripts/status.js" } ] } } ``` **Step 3:** Restart your Gemini CLI! --- ### 💜 Enabling in Claude Code Claude Code features an undocumented internal hook architecture that natively supports this status line. *(Note: Because this feature is currently undocumented by Anthropic, Claude Code's `doctor` command will display "Found 1 settings issue". This is perfectly normal and the status line will still execute successfully)*. **Step 1:** Locate your Claude settings file (`~/.claude/settings.json`). **Step 2:** Add the `statusLine` block to the root of the JSON object: ```json { "statusLine": { "type": "command", "command": "node /absolute/path/to/EVOKORE-MCP/scripts/status.js" } } ``` **Step 3:** Restart Claude Code. --- ### Òő ï¸ A Note on GitHub Copilot and Codex Microsoft's GitHub Copilot CLI and OpenAI's Codex CLI **do not natively support** these JSON hook configurations. If you want the EVOKORE Status Line to appear after commands in these tools, you must configure a native PowerShell/Bash alias wrapper around the CLI execution. **Example (PowerShell Profile):** ```powershell function copilot-evokore { gh copilot $args node "/absolute/path/to/EV ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–system promptβ€’7 months ago

web3-testing

Test smart contracts comprehensively using Hardhat and Foundry with

coding
⭐1
# Web3 Smart Contract Testing Master comprehensive testing strategies for smart contracts using Hardhat, Foundry, and advanced testing patterns. ## When to Use This Skill - Writing unit tests for smart contracts - Setting up integration test suites - Performing gas optimization testing - Fuzzing for edge cases - Forking mainnet for realistic testing - Automating test coverage reporting - Verifying contracts on Etherscan ## Hardhat Testing Setup ```javascript // hardhat.config.js require("@nomicfoundation/hardhat-toolbox"); require("@nomiclabs/hardhat-etherscan"); require("hardhat-gas-reporter"); require("solidity-coverage"); module.exports = { solidity: { version: "0.8.19", settings: { optimizer: { enabled: true, runs: 200, }, }, }, networks: { hardhat: { forking: { url: process.env.MAINNET_RPC_URL, blockNumber: 15000000, }, }, goerli: { url: process.env.GOERLI_RPC_URL, accounts: [process.env.PRIVATE_KEY], }, }, gasReporter: { enabled: true, currency: "USD", coinmarketcap: process.env.COINMARKETCAP_API_KEY, }, etherscan: { apiKey: process.env.ETHERSCAN_API_KEY, }, }; ``` ## Unit Testing Patterns ```javascript const { expect } = require("chai"); const { ethers } = require("hardhat"); const { loadFixture, time, } = require("@nomicfoundation/hardhat-network-helpers"); describe("Token Contract", function () { // Fixture for test setup async function deployTokenFixture() { const [owner, addr1, addr2] = await ethers.getSigners(); const Token = await ethers.getContractFactory("Token"); const token = await Token.deploy(); return { token, owner, addr1, addr2 }; } describe("Deployment", function () { it("Should set the right owner", async function () { const { token, owner } = await loadFixture(deployTokenFixture); expect(await token.owner()).to.equal(owner.address); }); it("Should assign total supply to owner", async function () { const { token, owner } = await loadFixture(deployTokenFixture); const ownerBalance = await token.balanceOf(owner.address); expect(await token.totalSupply()).to.equal(ownerBalance); }); }); describe("Transactions", function () { it("Should transfer tokens between accounts", async function () { const { token, owner, addr1 } = await loadFixture(deployTokenFixture); await expect(token.transfer(addr1.address, 50)).to.changeTokenBalances( token, [owner, addr1], [-50, 50], ); }); it("Should fail if sender doesn't have enough tokens", async function () { const { token, addr1 } = await loadFixture(deployTokenFixture); const initialBalance = await token.balanceOf(addr1.address); await expect( token.connect(addr1).transfer(owner.address, 1), ).to.be.revertedWith("Insufficient balance"); }); it("Should emit Transfer event", async function () { const { token, owner, addr1 } = await loadFixture(deployTokenFixture); await expect(token.transfer(addr1.address, 50)) .to.emit(token, "Transfer") .withArgs(owner.address, addr1.address, 50); }); }); describe("Time-based tests", function () { it("Should handle time-locked operations", async function () { const { token } = await loadFixture(deployTokenFixture); // Increase time by 1 day await time.increase(86400); // Test time-dependent functionality }); }); describe("Gas optimization", function () { it("Should use gas efficiently", async function () { const { token } = await loadFixture(deployTokenFixture); const tx = await token.transfer(addr1.address, 100); const receipt = await tx.wait(); expect(receipt.gasUsed).to.be.lessThan(50000); }); }); }); ``` ## Foundry Testing (Forge) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "forge-std/Test.sol"; import "../src/Token.sol"; contract TokenTest is Test { Token token; address owner = address(1); address user1 = address(2); address user2 = address(3); function setUp() public { vm.prank(owner); token = new Token(); } function testInitialSupply() public { assertEq(token.totalSupply(), 1000000 * 10**18); } function testTransfer() public { vm.prank(owner); token.transfer(user1, 100); assertEq(token.balanceOf(user1), 100); assertEq(token.balanceOf(owner), token.totalSupply() - 100); } function testFailTransferInsufficientBalance() public { vm.prank(user1); token.transfer(user2, 100); // Should fail } function testCannotTransferToZeroAddress() public { vm.prank(owner); vm.expectRevert("Invalid recipient"); token.transfer(address(0), 100); } // Fuzzing test function testFuzzTransfer(uint256 amount) public { vm.assume(amount > 0 && amount <= token.totalSupply()); vm.prank(owner); token.transfer(user1, amount); assertEq(token.balanceOf(user1), amount); } // Test with cheatcodes function testDealAndPrank() public { // Give ETH to address vm.deal(user1, 10 ether); // Impersonate address vm.prank(user1); // Test functionality assertEq(user1.balance, 10 ether); } // Mainnet fork test function testForkMainnet() public { vm.createSelectFork("https://eth-mainnet.alchemyapi.io/v2/..."); // Interact with mainnet contracts address dai = 0x6B175474E89094C44Da98b954EedeAC495271d0F; assertEq(IERC20(dai).symbol(), "DAI"); } } ``` ## Advanced Testing Patterns ### Snapshot and Revert ```javascript describe("Complex State Changes", function () { let snapshotId; beforeEach(async function () { snapshotId = await network.provider.send("evm_snapshot"); }); afterEach(async function () { await network.provider.send("evm_revert", [snapshotId]); }); it("Test 1", async function () { // Make state changes }); it("Test 2", async function () { // State reverted, clean slate }); }); ``` ### Mainnet Forking ```javascript describe("Mainnet Fork Tests", function () { let uniswapRouter, dai, usdc; before(async function () { await network.provider.request({ method: "hardhat_reset", params: [ { forking: { jsonRpcUrl: process.env.MAINNET_RPC_URL, blockNumber: 15000000, }, }, ], }); // Connect to existing mainnet contracts uniswapRouter = await ethers.getContractAt( "IUniswapV2Router", "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", ); dai = await ethers.getContractAt( "IERC20", "0x6B175474E89094C44Da98b954EedeAC495271d0F", ); }); it("Should swap on Uniswap", async function () { // Test with real Uniswap contracts }); }); ``` ### Impersonating Accounts ```javascript it("Should impersonate whale account", async function () { const whaleAddress = "0x..."; await network.provider.request({ method: "hardhat_impersonateAccount", params: [whaleAddress], }); const whale = await ethers.getSigner(whaleAddress); // Use whale's tokens await dai .connect(whale) .transfer(addr1.address, ethers.utils.parseEther("1000")); }); ``` ## Gas Optimization Testing ```javascript const { expect } = require("chai"); describe("Gas Optimization", function () { it("Compare gas usage between implementations", async function () { const Implementation1 = await ethers.getContractFactory("OptimizedContract"); const Implementation2 = await ethers.getContractFactory( "UnoptimizedContract", ); const contract1 = await Implementation1.deploy(); const contract2 = await Implementation2.deploy(); const tx1 = await contract1.doSomething(); const receipt1 = await tx1.wait(); const tx2 = await contract2.doSomething(); const receipt2 = await tx2.wait(); console.log("Optimized gas:", receipt1.gasUsed.toString()); console.log("Unoptimized gas:", receipt2.gasUsed.toString()); expect(receipt1.gasUsed).to.be.lessThan(receipt2.gasUsed); }); }); ``` ## Coverage Reporting ```bash # Generate coverage report npx hardhat coverage # Output shows: # File | % Stmts | % Branch | % Funcs | % Lines | # -------------------|---------|----------|---------|---------| # contracts/Token.sol | 100 | 90 | 100 | 95 | ``` ## Contract Verification ```javascript // Verify on Etherscan await hre.run("verify:verify", { address: contractAddress, constructorArguments: [arg1, arg2], }); ``` ```bash # Or via CLI npx hardhat verify --network mainnet CONTRACT_ADDRESS "Constructor arg1" "arg2" ``` ## CI/CD Integration ```yaml # .github/workflows/test.yml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - uses: actions/setup-node@v2 with: node-version: "16" - run: npm install - run: npx hardhat compile - run: npx hardhat test - run: npx hardhat coverage - name: Upload coverage to Codecov uses: codecov/codecov-action@v2 ``` ## Resources - **references/hardhat-setup.md**: Hardhat configuration guide - **references/foundry-setup.md**: Foundry testing framework - **references/test-patterns.md**: Testing best practices - **references/mainnet-forking.md**: Fork testing strategies - **references/contract-verification.md**: Etherscan verification - **assets/hardhat-config.js**: Complete Hardhat configuration - **assets/test-suite.js**: Comprehensive test examples - **assets/foundry.toml**: Foundry configuration - **scripts/test-contract.sh**: Automated testing script ## Best Practices 1. **Test Coverage**: Aim for >90% coverage 2. **Edge Cases**: Test boundary conditions 3. **Gas Limits**: Verify functions don't hit block gas limit 4. **Reentrancy**: Test for reentrancy vulnerabilities 5. **Access Control**: Test unauthorized access attempts 6. **Events**: Verify event emissions 7. **Fixtures**: Use fixtures to avoid code duplication 8. **Mainnet Fork**: Test with real contracts 9. **Fuzzing**: Use property-based testing 10. **CI/CD**: Automate testing on every commit
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

kpi-dashboard-design

Design effective KPI dashboards with metrics selection,

coding
⭐1
# KPI Dashboard Design Comprehensive patterns for designing effective Key Performance Indicator (KPI) dashboards that drive business decisions. ## When to Use This Skill - Designing executive dashboards - Selecting meaningful KPIs - Building real-time monitoring displays - Creating department-specific metrics views - Improving existing dashboard layouts - Establishing metric governance ## Core Concepts ### 1. KPI Framework | Level | Focus | Update Frequency | Audience | | --------------- | ---------------- | ----------------- | ---------- | | **Strategic** | Long-term goals | Monthly/Quarterly | Executives | | **Tactical** | Department goals | Weekly/Monthly | Managers | | **Operational** | Day-to-day | Real-time/Daily | Teams | ### 2. SMART KPIs ``` Specific: Clear definition Measurable: Quantifiable Achievable: Realistic targets Relevant: Aligned to goals Time-bound: Defined period ``` ### 3. Dashboard Hierarchy ``` β”œβ”€β”€ Executive Summary (1 page) β”‚ β”œβ”€β”€ 4-6 headline KPIs β”‚ β”œβ”€β”€ Trend indicators β”‚ └── Key alerts β”œβ”€β”€ Department Views β”‚ β”œβ”€β”€ Sales Dashboard β”‚ β”œβ”€β”€ Marketing Dashboard β”‚ β”œβ”€β”€ Operations Dashboard β”‚ └── Finance Dashboard └── Detailed Drilldowns β”œβ”€β”€ Individual metrics └── Root cause analysis ``` ## Common KPIs by Department ### Sales KPIs ```yaml Revenue Metrics: - Monthly Recurring Revenue (MRR) - Annual Recurring Revenue (ARR) - Average Revenue Per User (ARPU) - Revenue Growth Rate Pipeline Metrics: - Sales Pipeline Value - Win Rate - Average Deal Size - Sales Cycle Length Activity Metrics: - Calls/Emails per Rep - Demos Scheduled - Proposals Sent - Close Rate ``` ### Marketing KPIs ```yaml Acquisition: - Cost Per Acquisition (CPA) - Customer Acquisition Cost (CAC) - Lead Volume - Marketing Qualified Leads (MQL) Engagement: - Website Traffic - Conversion Rate - Email Open/Click Rate - Social Engagement ROI: - Marketing ROI - Campaign Performance - Channel Attribution - CAC Payback Period ``` ### Product KPIs ```yaml Usage: - Daily/Monthly Active Users (DAU/MAU) - Session Duration - Feature Adoption Rate - Stickiness (DAU/MAU) Quality: - Net Promoter Score (NPS) - Customer Satisfaction (CSAT) - Bug/Issue Count - Time to Resolution Growth: - User Growth Rate - Activation Rate - Retention Rate - Churn Rate ``` ### Finance KPIs ```yaml Profitability: - Gross Margin - Net Profit Margin - EBITDA - Operating Margin Liquidity: - Current Ratio - Quick Ratio - Cash Flow - Working Capital Efficiency: - Revenue per Employee - Operating Expense Ratio - Days Sales Outstanding - Inventory Turnover ``` ## Dashboard Layout Patterns ### Pattern 1: Executive Summary ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ EXECUTIVE DASHBOARD [Date Range β–Ό] β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ REVENUE β”‚ PROFIT β”‚ CUSTOMERS β”‚ NPS SCORE β”‚ β”‚ $2.4M β”‚ $450K β”‚ 12,450 β”‚ 72 β”‚ β”‚ β–² 12% β”‚ β–² 8% β”‚ β–² 15% β”‚ β–² 5pts β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ β”‚ β”‚ Revenue Trend β”‚ Revenue by Product β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ /\ /\ β”‚ β”‚ β”‚ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ 45% β”‚ β”‚ β”‚ β”‚ / \ / \ /\ β”‚ β”‚ β”‚ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ 32% β”‚ β”‚ β”‚ β”‚ / \/ \ / \ β”‚ β”‚ β”‚ β–ˆβ–ˆβ–ˆβ–ˆ 18% β”‚ β”‚ β”‚ β”‚ / \/ \ β”‚ β”‚ β”‚ β–ˆβ–ˆ 5% β”‚ β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ πŸ”΄ Alert: Churn rate exceeded threshold (>5%) β”‚ β”‚ 🟑 Warning: Support ticket volume 20% above average β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### Pattern 2: SaaS Metrics Dashboard ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ SAAS METRICS Jan 2024 [Monthly β–Ό] β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ MRR GROWTH β”‚ β”‚ β”‚ MRR β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ $125,000 β”‚ β”‚ β”‚ /── β”‚ β”‚ β”‚ β”‚ β–² 8% β”‚ β”‚ β”‚ /────/ β”‚ β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ /────/ β”‚ β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ /────/ β”‚ β”‚ β”‚ β”‚ ARR β”‚ β”‚ β”‚ /────/ β”‚ β”‚ β”‚ β”‚ $1,500,000 β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ β”‚ β–² 15% β”‚ β”‚ J F M A M J J A S O N D β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ UNIT ECONOMICS β”‚ COHORT RETENTION β”‚ β”‚ β”‚ β”‚ β”‚ CAC: $450 β”‚ Month 1: β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ 100% β”‚ β”‚ LTV: $2,700 β”‚ Month 3: β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ 85% β”‚ β”‚ LTV/CAC: 6.0x β”‚ Month 6: β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ 80% β”‚ β”‚ β”‚ Month 12: β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆ 72% β”‚ β”‚ Payback: 4 months β”‚ β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ CHURN ANALYSIS β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ Gross β”‚ Net β”‚ Logo β”‚ Expansion β”‚ β”‚ β”‚ β”‚ 4.2% β”‚ 1.8% β”‚ 3.1% β”‚ 2.4% β”‚ β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### Pattern 3: Real-time Operations ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ OPERATIONS CENTER Live ● Last: 10:42:15 β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ SYSTEM HEALTH β”‚ SERVICE STATUS β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ β”‚ CPU MEM DISK β”‚ β”‚ ● API Gateway Healthy β”‚ β”‚ β”‚ 45% 72% 58% β”‚ β”‚ ● User Service Healthy β”‚ β”‚ β”‚ β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆ β”‚ β”‚ ● Payment Service Degraded β”‚ β”‚ β”‚ β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆ β”‚ β”‚ ● Database Healthy β”‚ β”‚ β”‚ β–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆβ–ˆ β–ˆβ–ˆβ–ˆ β”‚ β”‚ ● Cache Healthy β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ REQUEST THROUGHPUT β”‚ ERROR RATE β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ β–β–‚β–ƒβ–„β–…β–†β–‡β–ˆβ–‡β–†β–…β–„β–ƒβ–‚β–β–‚β–ƒβ–„β–… β”‚ β”‚ β”‚ ▁▁▁▁▁▂▁▁▁▁▁▁▁▁▁▁▁▁▁▁ β”‚ β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ Current: 12,450 req/s β”‚ Current: 0.02% β”‚ β”‚ Peak: 18,200 req/s β”‚ Threshold: 1.0% β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ RECENT ALERTS β”‚ β”‚ 10:40 🟑 High latency on payment-service (p99 > 500ms) β”‚ β”‚ 10:35 🟒 Resolved: Database connection pool recovered β”‚ β”‚ 10:22 πŸ”΄ Payment service circuit breaker tripped β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ## Implementation Patterns ### SQL for KPI Calculations ```sql -- Monthly Recurring Revenue (MRR) WITH mrr_calculation AS ( SELECT DATE_TRUNC('month', billing_date) AS month, SUM( CASE subscription_interval WHEN 'monthly' THEN amount WHEN 'yearly' THEN amount / 12 WHEN 'quarterly' THEN amount / 3 END ) AS mrr FROM subscriptions WHERE status = 'active' GROUP BY DATE_TRUNC('month', billing_date) ) SELECT month, mrr, LAG(mrr) OVER (ORDER BY month) AS prev_mrr, (mrr - LAG(mrr) OVER (ORDER BY month)) / LAG(mrr) OVER (ORDER BY month) * 100 AS growth_pct FROM mrr_calculation; -- Cohort Retention WITH cohorts AS ( SELECT user_id, DATE_TRUNC('month', created_at) AS cohort_month FROM users ), activity AS ( SELECT user_id, DATE_TRUNC('month', event_date) AS activity_month FROM user_events WHERE event_type = 'active_session' ) SELECT c.cohort_month, EXTRACT(MONTH FROM age(a.activity_month, c.cohort_month)) AS months_since_signup, COUNT(DISTINCT a.user_id) AS active_users, COUNT(DISTINCT a.user_id)::FLOAT / COUNT(DISTINCT c.user_id) * 100 AS retention_rate FROM cohorts c LEFT JOIN activity a ON c.user_id = a.user_id AND a.activity_month >= c.cohort_month GROUP BY c.cohort_month, EXTRACT(MONTH FROM age(a.activity_month, c.cohort_month)) ORDER BY c.cohort_month, months_since_signup; -- Customer Acquisition Cost (CAC) SELECT DATE_TRUNC('month', acquired_date) AS month, SUM(marketing_spend) / NULLIF(COUNT(new_customers), 0) AS cac, SUM(marketing_spend) AS total_spend, COUNT(new_customers) AS customers_acquired FROM ( SELECT DATE_TRUNC('month', u.created_at) AS acquired_date, u.id AS new_customers, m.spend AS marketing_spend FROM users u JOIN marketing_spend m ON DATE_TRUNC('month', u.created_at) = m.month WHERE u.source = 'marketing' ) acquisition GROUP BY DATE_TRUNC('month', acquired_date); ``` ### Python Dashboard Code (Streamlit) ```python import streamlit as st import pandas as pd import plotly.express as px import plotly.graph_objects as go st.set_page_config(page_title="KPI Dashboard", layout="wide") # Header with date filter col1, col2 = st.columns([3, 1]) with col1: st.title("Executive Dashboard") with col2: date_range = st.selectbox( "Period", ["Last 7 Days", "Last 30 Days", "Last Quarter", "YTD"] ) # KPI Cards def metric_card(label, value, delta, prefix="", suffix=""): delta_color = "green" if delta >= 0 else "red" delta_arrow = "β–²" if delta >= 0 else "β–Ό" st.metric( label=label, value=f"{prefix}{value:,.0f}{suffix}", delta=f"{delta_arrow} {abs(delta):.1f}%" ) col1, col2, col3, col4 = st.columns(4) with col1: metric_card("Revenue", 2400000, 12.5, prefix="$") with col2: metric_card("Customers", 12450, 15.2) with col3: metric_card("NPS Score", 72, 5.0) with col4: metric_card("Churn Rate", 4.2, -0.8, suffix="%") # Charts col1, col2 = st.columns(2) with col1: st.subheader("Revenue Trend") revenue_data = pd.DataFrame({ 'Month': pd.date_range('2024-01-01', periods=12, freq='M'), 'Revenue': [180000, 195000, 210000, 225000, 240000, 255000, 270000, 285000, 300000, 315000, 330000, 345000] }) fig = px.line(revenue_data, x='Month', y='Revenue', line_shape='spline', markers=True) fig.update_layout(height=300) st.plotly_chart(fig, use_container_width=True) with col2: st.subheader("Revenue by Product") product_data = pd.DataFrame({ 'Product': ['Enterprise', 'Professional', 'Starter', 'Other'], 'Revenue': [45, 32, 18, 5] }) fig = px.pie(product_data, values='Revenue', names='Product', hole=0.4) fig.update_layout(height=300) st.plotly_chart(fig, use_container_width=True) # Cohort Heatmap st.subheader("Cohort Retention") cohort_data = pd.DataFrame({ 'Cohort': ['Jan', 'Feb', 'Mar', 'Apr', 'May'], 'M0': [100, 100, 100, 100, 100], 'M1': [85, 87, 84, 86, 88], 'M2': [78, 80, 76, 79, None], 'M3': [72, 74, 70, None, None], 'M4': [68, 70, None, None, None], }) fig = go.Figure(data=go.Heatmap( z=cohort_data.iloc[:, 1:].values, x=['M0', 'M1', 'M2', 'M3', 'M4'], y=cohort_data['Cohort'], colorscale='Blues', text=cohort_data.iloc[:, 1:].values, texttemplate='%{text}%', textfont={"size": 12}, )) fig.update_layout(height=250) st.plotly_chart(fig, use_container_width=True) # Alerts Section st.subheader("Alerts") alerts = [ {"level": "error", "message": "Churn rate exceeded threshold (>5%)"}, {"level": "warning", "message": "Support ticket volume 20% above average"}, ] for alert in alerts: if alert["level"] == "error": st.error(f"πŸ”΄ {alert['message']}") elif alert["level"] == "warning": st.warning(f"🟑 {alert['message']}") ``` ## Best Practices ### Do's - **Limit to 5-7 KPIs** - Focus on what matters - **Show context** - Comparisons, trends, targets - **Use consistent colors** - Red=bad, green=good - **Enable drilldown** - From summary to detail - **Update appropriately** - Match metric frequency ### Don'ts - **Don't show vanity metrics** - Focus on actionable data - **Don't overcrowd** - White space aids comprehension - **Don't use 3D charts** - They distort perception - **Don't hide methodology** - Document calculations - **Don't ignore mobile** - Ensure responsive design ## Resources - [Stephen Few's Dashboard Design](https://www.perceptualedge.com/articles/visual_business_intelligence/rules_for_using_color.pdf) - [Edward Tufte's Principles](https://www.edwardtufte.com/tufte/) - [Google Data Studio Gallery](https://datastudio.google.com/gallery)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

multi-cloud-architecture

Design multi-cloud architectures using a decision framework to

architecture
⭐1
# Multi-Cloud Architecture Decision framework and patterns for architecting applications across AWS, Azure, and GCP. ## Purpose Design cloud-agnostic architectures and make informed decisions about service selection across cloud providers. ## When to Use - Design multi-cloud strategies - Migrate between cloud providers - Select cloud services for specific workloads - Implement cloud-agnostic architectures - Optimize costs across providers ## Cloud Service Comparison ### Compute Services | AWS | Azure | GCP | Use Case | | ------- | ------------------- | --------------- | ------------------ | | EC2 | Virtual Machines | Compute Engine | IaaS VMs | | ECS | Container Instances | Cloud Run | Containers | | EKS | AKS | GKE | Kubernetes | | Lambda | Functions | Cloud Functions | Serverless | | Fargate | Container Apps | Cloud Run | Managed containers | ### Storage Services | AWS | Azure | GCP | Use Case | | ------- | --------------- | --------------- | -------------- | | S3 | Blob Storage | Cloud Storage | Object storage | | EBS | Managed Disks | Persistent Disk | Block storage | | EFS | Azure Files | Filestore | File storage | | Glacier | Archive Storage | Archive Storage | Cold storage | ### Database Services | AWS | Azure | GCP | Use Case | | ----------- | ---------------- | ------------- | --------------- | | RDS | SQL Database | Cloud SQL | Managed SQL | | DynamoDB | Cosmos DB | Firestore | NoSQL | | Aurora | PostgreSQL/MySQL | Cloud Spanner | Distributed SQL | | ElastiCache | Cache for Redis | Memorystore | Caching | **Reference:** See `references/service-comparison.md` for complete comparison ## Multi-Cloud Patterns ### Pattern 1: Single Provider with DR - Primary workload in one cloud - Disaster recovery in another - Database replication across clouds - Automated failover ### Pattern 2: Best-of-Breed - Use best service from each provider - AI/ML on GCP - Enterprise apps on Azure - General compute on AWS ### Pattern 3: Geographic Distribution - Serve users from nearest cloud region - Data sovereignty compliance - Global load balancing - Regional failover ### Pattern 4: Cloud-Agnostic Abstraction - Kubernetes for compute - PostgreSQL for database - S3-compatible storage (MinIO) - Open source tools ## Cloud-Agnostic Architecture ### Use Cloud-Native Alternatives - **Compute:** Kubernetes (EKS/AKS/GKE) - **Database:** PostgreSQL/MySQL (RDS/SQL Database/Cloud SQL) - **Message Queue:** Apache Kafka (MSK/Event Hubs/Confluent) - **Cache:** Redis (ElastiCache/Azure Cache/Memorystore) - **Object Storage:** S3-compatible API - **Monitoring:** Prometheus/Grafana - **Service Mesh:** Istio/Linkerd ### Abstraction Layers ``` Application Layer ↓ Infrastructure Abstraction (Terraform) ↓ Cloud Provider APIs ↓ AWS / Azure / GCP ``` ## Cost Comparison ### Compute Pricing Factors - **AWS:** On-demand, Reserved, Spot, Savings Plans - **Azure:** Pay-as-you-go, Reserved, Spot - **GCP:** On-demand, Committed use, Preemptible ### Cost Optimization Strategies 1. Use reserved/committed capacity (30-70% savings) 2. Leverage spot/preemptible instances 3. Right-size resources 4. Use serverless for variable workloads 5. Optimize data transfer costs 6. Implement lifecycle policies 7. Use cost allocation tags 8. Monitor with cloud cost tools **Reference:** See `references/multi-cloud-patterns.md` ## Migration Strategy ### Phase 1: Assessment - Inventory current infrastructure - Identify dependencies - Assess cloud compatibility - Estimate costs ### Phase 2: Pilot - Select pilot workload - Implement in target cloud - Test thoroughly - Document learnings ### Phase 3: Migration - Migrate workloads incrementally - Maintain dual-run period - Monitor performance - Validate functionality ### Phase 4: Optimization - Right-size resources - Implement cloud-native services - Optimize costs - Enhance security ## Best Practices 1. **Use infrastructure as code** (Terraform/OpenTofu) 2. **Implement CI/CD pipelines** for deployments 3. **Design for failure** across clouds 4. **Use managed services** when possible 5. **Implement comprehensive monitoring** 6. **Automate cost optimization** 7. **Follow security best practices** 8. **Document cloud-specific configurations** 9. **Test disaster recovery** procedures 10. **Train teams** on multiple clouds ## Reference Files - `references/service-comparison.md` - Complete service comparison - `references/multi-cloud-patterns.md` - Architecture patterns ## Related Skills - `terraform-module-library` - For IaC implementation - `cost-optimization` - For cost management - `hybrid-cloud-networking` - For connectivity
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

context-driven-development

Creates and maintains project context artifacts (product.md,

coding
⭐1
# Context-Driven Development Guide for implementing and maintaining context as a managed artifact alongside code, enabling consistent AI interactions and team alignment through structured project documentation. ## When to Use This Skill - Setting up new projects with Conductor - Understanding the relationship between context artifacts - Maintaining consistency across AI-assisted development sessions - Onboarding team members to an existing Conductor project - Deciding when to update context documents - Managing greenfield vs brownfield project contexts ## Core Philosophy Context-Driven Development treats project context as a first-class artifact managed alongside code. Instead of relying on ad-hoc prompts or scattered documentation, establish a persistent, structured foundation that informs all AI interactions. Key principles: 1. **Context precedes code**: Define what you're building and how before implementation 2. **Living documentation**: Context artifacts evolve with the project 3. **Single source of truth**: One canonical location for each type of information 4. **AI alignment**: Consistent context produces consistent AI behavior ## The Workflow Follow the **Context β†’ Spec & Plan β†’ Implement** workflow: 1. **Context Phase**: Establish or verify project context artifacts exist and are current 2. **Specification Phase**: Define requirements and acceptance criteria for work units 3. **Planning Phase**: Break specifications into phased, actionable tasks 4. **Implementation Phase**: Execute tasks following established workflow patterns ## Artifact Relationships ### product.md - Defines WHAT and WHY Purpose: Captures product vision, goals, target users, and business context. Contents: - Product name and one-line description - Problem statement and solution approach - Target user personas - Core features and capabilities - Success metrics and KPIs - Product roadmap (high-level) Update when: - Product vision or goals change - New major features are planned - Target audience shifts - Business priorities evolve ### product-guidelines.md - Defines HOW to Communicate Purpose: Establishes brand voice, messaging standards, and communication patterns. Contents: - Brand voice and tone guidelines - Terminology and glossary - Error message conventions - User-facing copy standards - Documentation style Update when: - Brand guidelines change - New terminology is introduced - Communication patterns need refinement ### tech-stack.md - Defines WITH WHAT Purpose: Documents technology choices, dependencies, and architectural decisions. Contents: - Primary languages and frameworks - Key dependencies with versions - Infrastructure and deployment targets - Development tools and environment - Testing frameworks - Code quality tools Update when: - Adding new dependencies - Upgrading major versions - Changing infrastructure - Adopting new tools or patterns ### workflow.md - Defines HOW to Work Purpose: Establishes development practices, quality gates, and team workflows. Contents: - Development methodology (TDD, etc.) - Git workflow and commit conventions - Code review requirements - Testing requirements and coverage targets - Quality assurance gates - Deployment procedures Update when: - Team practices evolve - Quality standards change - New workflow patterns are adopted ### tracks.md - Tracks WHAT'S HAPPENING Purpose: Registry of all work units with status and metadata. Contents: - Active tracks with current status - Completed tracks with completion dates - Track metadata (type, priority, assignee) - Links to individual track directories Update when: - New tracks are created - Track status changes - Tracks are completed or archived See [references/artifact-templates.md](references/artifact-templates.md) for copy-paste starter templates. ## Context Maintenance Principles ### Keep Artifacts Synchronized Ensure changes in one artifact reflect in related documents: - New feature in product.md β†’ Update tech-stack.md if new dependencies needed - Completed track β†’ Update product.md to reflect new capabilities - Workflow change β†’ Update all affected track plans ### Update tech-stack.md When Adding Dependencies Before adding any new dependency: 1. Check if existing dependencies solve the need 2. Document the rationale for new dependencies 3. Add version constraints 4. Note any configuration requirements ### Update product.md When Features Complete After completing a feature track: 1. Move feature from "planned" to "implemented" in product.md 2. Update any affected success metrics 3. Document any scope changes from original plan ### Verify Context Before Implementation Before starting any track: 1. Read all context artifacts 2. Flag any outdated information 3. Propose updates before proceeding 4. Confirm context accuracy with stakeholders ## Greenfield vs Brownfield Handling ### Greenfield Projects (New) For new projects: 1. Run `/conductor:setup` to create all artifacts interactively 2. Answer questions about product vision, tech preferences, and workflow 3. Generate initial style guides for chosen languages 4. Create empty tracks registry Characteristics: - Full control over context structure - Define standards before code exists - Establish patterns early ### Brownfield Projects (Existing) For existing codebases: 1. Run `/conductor:setup` with existing codebase detection 2. System analyzes existing code, configs, and documentation 3. Pre-populate artifacts based on discovered patterns 4. Review and refine generated context Characteristics: - Extract implicit context from existing code - Reconcile existing patterns with desired patterns - Document technical debt and modernization plans - Preserve working patterns while establishing standards ## Benefits ### Team Alignment - New team members onboard faster with explicit context - Consistent terminology and conventions across the team - Shared understanding of product goals and technical decisions ### AI Consistency - AI assistants produce aligned outputs across sessions - Reduced need to re-explain context in each interaction - Predictable behavior based on documented standards ### Institutional Memory - Decisions and rationale are preserved - Context survives team changes - Historical context informs future decisions ### Quality Assurance - Standards are explicit and verifiable - Deviations from context are detectable - Quality gates are documented and enforceable ## Directory Structure ``` conductor/ β”œβ”€β”€ index.md # Navigation hub linking all artifacts β”œβ”€β”€ product.md # Product vision and goals β”œβ”€β”€ product-guidelines.md # Communication standards β”œβ”€β”€ tech-stack.md # Technology preferences β”œβ”€β”€ workflow.md # Development practices β”œβ”€β”€ tracks.md # Work unit registry β”œβ”€β”€ setup_state.json # Resumable setup state β”œβ”€β”€ code_styleguides/ # Language-specific conventions β”‚ β”œβ”€β”€ python.md β”‚ β”œβ”€β”€ typescript.md β”‚ └── ... └── tracks/ └── <track-id>/ β”œβ”€β”€ spec.md β”œβ”€β”€ plan.md β”œβ”€β”€ metadata.json └── index.md ``` ## Context Lifecycle 1. **Creation**: Initial setup via `/conductor:setup` 2. **Validation**: Verify before each track 3. **Evolution**: Update as project grows 4. **Synchronization**: Keep artifacts aligned 5. **Archival**: Document historical decisions ## Context Validation Checklist Before starting implementation on any track, validate context: ### Product Context - [ ] product.md reflects current product vision - [ ] Target users are accurately described - [ ] Feature list is up to date - [ ] Success metrics are defined ### Technical Context - [ ] tech-stack.md lists all current dependencies - [ ] Version numbers are accurate - [ ] Infrastructure targets are correct - [ ] Development tools are documented ### Workflow Context - [ ] workflow.md describes current practices - [ ] Quality gates are defined - [ ] Coverage targets are specified - [ ] Commit conventions are documented ### Track Context - [ ] tracks.md shows all active work - [ ] No stale or abandoned tracks - [ ] Dependencies between tracks are noted ## Common Anti-Patterns Avoid these context management mistakes: ### Stale Context Problem: Context documents become outdated and misleading. Solution: Update context as part of each track's completion process. ### Context Sprawl Problem: Information scattered across multiple locations. Solution: Use the defined artifact structure; resist creating new document types. ### Implicit Context Problem: Relying on knowledge not captured in artifacts. Solution: If you reference something repeatedly, add it to the appropriate artifact. ### Context Hoarding Problem: One person maintains context without team input. Solution: Review context artifacts in pull requests; make updates collaborative. ### Over-Specification Problem: Context becomes so detailed it's impossible to maintain. Solution: Keep artifacts focused on decisions that affect AI behavior and team alignment. ## Integration with Development Tools ### IDE Integration Configure your IDE to display context files prominently: - Pin conductor/product.md for quick reference - Add tech-stack.md to project notes - Create snippets for common patterns from style guides ### Git Hooks Consider pre-commit hooks that: - Warn when dependencies change without tech-stack.md update - Remind to update product.md when feature branches merge - Validate context artifact syntax ### CI/CD Integration Include context validation in pipelines: - Check tech-stack.md matches actual dependencies - Verify links in context documents resolve - Ensure tracks.md status matches git branch state ## Session Continuity Conductor supports multi-session development through context persistence: ### Starting a New Session 1. Read index.md to orient yourself 2. Check tracks.md for active work 3. Review relevant track's plan.md for current task 4. Verify context artifacts are current ### Ending a Session 1. Update plan.md with current progress 2. Note any blockers or decisions made 3. Commit in-progress work with clear status 4. Update tracks.md if status changed ### Handling Interruptions If interrupted mid-task: 1. Mark task as `[~]` with note about stopping point 2. Commit work-in-progress to feature branch 3. Document any uncommitted decisions in plan.md ## Best Practices 1. **Read context first**: Always read relevant artifacts before starting work 2. **Small updates**: Make incremental context changes, not massive rewrites 3. **Link decisions**: Reference context when making implementation choices 4. **Version context**: Commit context changes alongside code changes 5. **Review context**: Include context artifact reviews in code reviews 6. **Validate regularly**: Run context validation checklist before major work 7. **Communicate changes**: Notify team when context artifacts change significantly 8. **Preserve history**: Use git to track context evolution over time 9. **Question staleness**: If context feels wrong, investigate and update 10. **Keep it actionable**: Every context item should inform a decision or behavior
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

track-management

Use this skill when creating, managing, or working with Conductor

coding
⭐1
# Track Management Guide for creating, managing, and completing Conductor tracks - the logical work units that organize features, bugs, and refactors through specification, planning, and implementation phases. ## When to Use This Skill - Creating new feature, bug, or refactor tracks - Writing or reviewing spec.md files - Creating or updating plan.md files - Managing track lifecycle from creation to completion - Understanding track status markers and conventions - Working with the tracks.md registry - Interpreting or updating track metadata ## Track Concept A track is a logical work unit that encapsulates a complete piece of work. Each track has: - A unique identifier - A specification defining requirements - A phased plan breaking work into tasks - Metadata tracking status and progress Tracks provide semantic organization for work, enabling: - Clear scope boundaries - Progress tracking - Git-aware operations (revert by track) - Team coordination ## Track Types ### feature New functionality or capabilities. Use for: - New user-facing features - New API endpoints - New integrations - Significant enhancements ### bug Defect fixes. Use for: - Incorrect behavior - Error conditions - Performance regressions - Security vulnerabilities ### chore Maintenance and housekeeping. Use for: - Dependency updates - Configuration changes - Documentation updates - Cleanup tasks ### refactor Code improvement without behavior change. Use for: - Code restructuring - Pattern adoption - Technical debt reduction - Performance optimization (same behavior, better performance) ## Track ID Format Track IDs follow the pattern: `{shortname}_{YYYYMMDD}` - **shortname**: 2-4 word kebab-case description (e.g., `user-auth`, `api-rate-limit`) - **YYYYMMDD**: Creation date in ISO format Examples: - `user-auth_20250115` - `fix-login-error_20250115` - `upgrade-deps_20250115` - `refactor-api-client_20250115` ## Track Lifecycle ### 1. Creation (newTrack) **Define Requirements** 1. Gather requirements through interactive Q&A 2. Identify acceptance criteria 3. Determine scope boundaries 4. Identify dependencies **Generate Specification** 1. Create `spec.md` with structured requirements 2. Document functional and non-functional requirements 3. Define acceptance criteria 4. List dependencies and constraints **Generate Plan** 1. Create `plan.md` with phased task breakdown 2. Organize tasks into logical phases 3. Add verification tasks after phases 4. Estimate effort and complexity **Register Track** 1. Add entry to `tracks.md` registry 2. Create track directory structure 3. Generate `metadata.json` 4. Create track `index.md` ### 2. Implementation **Execute Tasks** 1. Select next pending task from plan 2. Mark task as in-progress 3. Implement following workflow (TDD) 4. Mark task complete with commit SHA **Update Status** 1. Update task markers in plan.md 2. Record commit SHAs for traceability 3. Update phase progress 4. Update track status in tracks.md **Verify Progress** 1. Complete verification tasks 2. Wait for checkpoint approval 3. Record checkpoint commits ### 3. Completion **Sync Documentation** 1. Update product.md if features added 2. Update tech-stack.md if dependencies changed 3. Verify all acceptance criteria met **Archive or Delete** 1. Mark track as completed in tracks.md 2. Record completion date 3. Archive or retain track directory ## Specification (spec.md) Structure ```markdown # {Track Title} ## Overview Brief description of what this track accomplishes and why. ## Functional Requirements ### FR-1: {Requirement Name} Description of the functional requirement. - Acceptance: How to verify this requirement is met ### FR-2: {Requirement Name} ... ## Non-Functional Requirements ### NFR-1: {Requirement Name} Description of the non-functional requirement (performance, security, etc.) - Target: Specific measurable target - Verification: How to test ## Acceptance Criteria - [ ] Criterion 1: Specific, testable condition - [ ] Criterion 2: Specific, testable condition - [ ] Criterion 3: Specific, testable condition ## Scope ### In Scope - Explicitly included items - Features to implement - Components to modify ### Out of Scope - Explicitly excluded items - Future considerations - Related but separate work ## Dependencies ### Internal - Other tracks or components this depends on - Required context artifacts ### External - Third-party services or APIs - External dependencies ## Risks and Mitigations | Risk | Impact | Mitigation | | ---------------- | --------------- | ------------------- | | Risk description | High/Medium/Low | Mitigation strategy | ## Open Questions - [ ] Question that needs resolution - [x] Resolved question - Answer ``` ## Plan (plan.md) Structure ```markdown # Implementation Plan: {Track Title} Track ID: `{track-id}` Created: YYYY-MM-DD Status: pending | in-progress | completed ## Overview Brief description of implementation approach. ## Phase 1: {Phase Name} ### Tasks - [ ] **Task 1.1**: Task description - Sub-task or detail - Sub-task or detail - [ ] **Task 1.2**: Task description - [ ] **Task 1.3**: Task description ### Verification - [ ] **Verify 1.1**: Verification step for phase ## Phase 2: {Phase Name} ### Tasks - [ ] **Task 2.1**: Task description - [ ] **Task 2.2**: Task description ### Verification - [ ] **Verify 2.1**: Verification step for phase ## Phase 3: Finalization ### Tasks - [ ] **Task 3.1**: Update documentation - [ ] **Task 3.2**: Final integration test ### Verification - [ ] **Verify 3.1**: All acceptance criteria met ## Checkpoints | Phase | Checkpoint SHA | Date | Status | | ------- | -------------- | ---- | ------- | | Phase 1 | | | pending | | Phase 2 | | | pending | | Phase 3 | | | pending | ``` ## Status Marker Conventions Use consistent markers in plan.md: | Marker | Meaning | Usage | | ------ | ----------- | --------------------------- | | `[ ]` | Pending | Task not started | | `[~]` | In Progress | Currently being worked | | `[x]` | Complete | Task finished (include SHA) | | `[-]` | Skipped | Intentionally not done | | `[!]` | Blocked | Waiting on dependency | Example: ```markdown - [x] **Task 1.1**: Set up database schema `abc1234` - [~] **Task 1.2**: Implement user model - [ ] **Task 1.3**: Add validation logic - [!] **Task 1.4**: Integrate auth service (blocked: waiting for API key) - [-] **Task 1.5**: Legacy migration (skipped: not needed) ``` ## Track Registry (tracks.md) Format ```markdown # Track Registry ## Active Tracks | Track ID | Type | Status | Phase | Started | Assignee | | ------------------------------------------------ | ------- | ----------- | ----- | ---------- | ---------- | | [user-auth_20250115](tracks/user-auth_20250115/) | feature | in-progress | 2/3 | 2025-01-15 | @developer | | [fix-login_20250114](tracks/fix-login_20250114/) | bug | pending | 0/2 | 2025-01-14 | - | ## Completed Tracks | Track ID | Type | Completed | Duration | | ---------------------------------------------- | ----- | ---------- | -------- | | [setup-ci_20250110](tracks/setup-ci_20250110/) | chore | 2025-01-12 | 2 days | ## Archived Tracks | Track ID | Reason | Archived | | ---------------------------------------------------- | ---------- | ---------- | | [old-feature_20241201](tracks/old-feature_20241201/) | Superseded | 2025-01-05 | ``` ## Metadata (metadata.json) Fields ```json { "id": "user-auth_20250115", "title": "User Authentication System", "type": "feature", "status": "in-progress", "priority": "high", "created": "2025-01-15T10:30:00Z", "updated": "2025-01-15T14:45:00Z", "started": "2025-01-15T11:00:00Z", "completed": null, "assignee": "@developer", "phases": { "total": 3, "current": 2, "completed": 1 }, "tasks": { "total": 12, "completed": 5, "in_progress": 1, "pending": 6 }, "checkpoints": [ { "phase": 1, "sha": "abc1234", "date": "2025-01-15T13:00:00Z" } ], "dependencies": [], "tags": ["auth", "security"] } ``` ## Track Operations ### Creating a Track 1. Run `/conductor:new-track` 2. Answer interactive questions 3. Review generated spec.md 4. Review generated plan.md 5. Confirm track creation ### Starting Implementation 1. Read spec.md and plan.md 2. Verify context artifacts are current 3. Mark first task as `[~]` 4. Begin TDD workflow ### Completing a Phase 1. Ensure all phase tasks are `[x]` 2. Complete verification tasks 3. Wait for checkpoint approval 4. Record checkpoint SHA 5. Proceed to next phase ### Completing a Track 1. Verify all phases complete 2. Verify all acceptance criteria met 3. Update product.md if needed 4. Mark track completed in tracks.md 5. Update metadata.json ### Reverting a Track 1. Run `/conductor:revert` 2. Select track to revert 3. Choose granularity (track/phase/task) 4. Confirm revert operation 5. Update status markers ## Handling Track Dependencies ### Identifying Dependencies During track creation, identify: - **Hard dependencies**: Must complete before this track can start - **Soft dependencies**: Can proceed in parallel but may affect integration - **External dependencies**: Third-party services, APIs, or team decisions ### Documenting Dependencies In spec.md, list dependencies with: - Dependency type (hard/soft/external) - Current status (available/pending/blocked) - Resolution path (what needs to happen) ### Managing Blocked Tracks When a track is blocked: 1. Mark blocked tasks with `[!]` and reason 2. Update tracks.md status 3. Document blocker in metadata.json 4. Consider creating dependency track if needed ## Track Sizing Guidelines ### Right-Sized Tracks Aim for tracks that: - Complete in 1-5 days of work - Have 2-4 phases - Contain 8-20 tasks total - Deliver a coherent, testable unit ### Too Large Signs a track is too large: - More than 5 phases - More than 25 tasks - Multiple unrelated features - Estimated duration > 1 week Solution: Split into multiple tracks with clear boundaries. ### Too Small Signs a track is too small: - Single phase with 1-2 tasks - No meaningful verification needed - Could be a sub-task of another track - Less than a few hours of work Solution: Combine with related work or handle as part of existing track. ## Specification Quality Checklist Before finalizing spec.md, verify: ### Requirements Quality - [ ] Each requirement has clear acceptance criteria - [ ] Requirements are testable - [ ] Requirements are independent (can verify separately) - [ ] No ambiguous language ("should be fast" β†’ "response < 200ms") ### Scope Clarity - [ ] In-scope items are specific - [ ] Out-of-scope items prevent scope creep - [ ] Boundaries are clear to implementer ### Dependencies Identified - [ ] All internal dependencies listed - [ ] External dependencies have owners/contacts - [ ] Dependency status is current ### Risks Addressed - [ ] Major risks identified - [ ] Impact assessment realistic - [ ] Mitigations are actionable ## Plan Quality Checklist Before starting implementation, verify plan.md: ### Task Quality - [ ] Tasks are atomic (one logical action) - [ ] Tasks are independently verifiable - [ ] Task descriptions are clear - [ ] Sub-tasks provide helpful detail ### Phase Organization - [ ] Phases group related tasks - [ ] Each phase delivers something testable - [ ] Verification tasks after each phase - [ ] Phases build on each other logically ### Completeness - [ ] All spec requirements have corresponding tasks - [ ] Documentation tasks included - [ ] Testing tasks included - [ ] Integration tasks included ## Common Track Patterns ### Feature Track Pattern ``` Phase 1: Foundation - Data models - Database migrations - Basic API structure Phase 2: Core Logic - Business logic implementation - Input validation - Error handling Phase 3: Integration - UI integration - API documentation - End-to-end tests ``` ### Bug Fix Track Pattern ``` Phase 1: Reproduction - Write failing test capturing bug - Document reproduction steps Phase 2: Fix - Implement fix - Verify test passes - Check for regressions Phase 3: Verification - Manual verification - Update documentation if needed ``` ### Refactor Track Pattern ``` Phase 1: Preparation - Add characterization tests - Document current behavior Phase 2: Refactoring - Apply changes incrementally - Maintain green tests throughout Phase 3: Cleanup - Remove dead code - Update documentation ``` ## Best Practices 1. **One track, one concern**: Keep tracks focused on a single logical change 2. **Small phases**: Break work into phases of 3-5 tasks maximum 3. **Verification after phases**: Always include verification tasks 4. **Update markers immediately**: Mark task status as you work 5. **Record SHAs**: Always note commit SHAs for completed tasks 6. **Review specs before planning**: Ensure spec is complete before creating plan 7. **Link dependencies**: Explicitly note track dependencies 8. **Archive, don't delete**: Preserve completed tracks for reference 9. **Size appropriately**: Keep tracks between 1-5 days of work 10. **Clear acceptance criteria**: Every requirement must be testable
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

airflow-dag-patterns

Build production Apache Airflow DAGs with best practices for

data
⭐1
# Apache Airflow DAG Patterns Production-ready patterns for Apache Airflow including DAG design, operators, sensors, testing, and deployment strategies. ## When to Use This Skill - Creating data pipeline orchestration with Airflow - Designing DAG structures and dependencies - Implementing custom operators and sensors - Testing Airflow DAGs locally - Setting up Airflow in production - Debugging failed DAG runs ## Core Concepts ### 1. DAG Design Principles | Principle | Description | | --------------- | ----------------------------------- | | **Idempotent** | Running twice produces same result | | **Atomic** | Tasks succeed or fail completely | | **Incremental** | Process only new/changed data | | **Observable** | Logs, metrics, alerts at every step | ### 2. Task Dependencies ```python # Linear task1 >> task2 >> task3 # Fan-out task1 >> [task2, task3, task4] # Fan-in [task1, task2, task3] >> task4 # Complex task1 >> task2 >> task4 task1 >> task3 >> task4 ``` ## Quick Start ```python # dags/example_dag.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.empty import EmptyOperator default_args = { 'owner': 'data-team', 'depends_on_past': False, 'email_on_failure': True, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), 'retry_exponential_backoff': True, 'max_retry_delay': timedelta(hours=1), } with DAG( dag_id='example_etl', default_args=default_args, description='Example ETL pipeline', schedule='0 6 * * *', # Daily at 6 AM start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'example'], max_active_runs=1, ) as dag: start = EmptyOperator(task_id='start') def extract_data(**context): execution_date = context['ds'] # Extract logic here return {'records': 1000} extract = PythonOperator( task_id='extract', python_callable=extract_data, ) end = EmptyOperator(task_id='end') start >> extract >> end ``` ## Patterns ### Pattern 1: TaskFlow API (Airflow 2.0+) ```python # dags/taskflow_example.py from datetime import datetime from airflow.decorators import dag, task from airflow.models import Variable @dag( dag_id='taskflow_etl', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'taskflow'], ) def taskflow_etl(): """ETL pipeline using TaskFlow API""" @task() def extract(source: str) -> dict: """Extract data from source""" import pandas as pd df = pd.read_csv(f's3://bucket/{source}/{{ ds }}.csv') return {'data': df.to_dict(), 'rows': len(df)} @task() def transform(extracted: dict) -> dict: """Transform extracted data""" import pandas as pd df = pd.DataFrame(extracted['data']) df['processed_at'] = datetime.now() df = df.dropna() return {'data': df.to_dict(), 'rows': len(df)} @task() def load(transformed: dict, target: str): """Load data to target""" import pandas as pd df = pd.DataFrame(transformed['data']) df.to_parquet(f's3://bucket/{target}/{{ ds }}.parquet') return transformed['rows'] @task() def notify(rows_loaded: int): """Send notification""" print(f'Loaded {rows_loaded} rows') # Define dependencies with XCom passing extracted = extract(source='raw_data') transformed = transform(extracted) loaded = load(transformed, target='processed_data') notify(loaded) # Instantiate the DAG taskflow_etl() ``` ### Pattern 2: Dynamic DAG Generation ```python # dags/dynamic_dag_factory.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.models import Variable import json # Configuration for multiple similar pipelines PIPELINE_CONFIGS = [ {'name': 'customers', 'schedule': '@daily', 'source': 's3://raw/customers'}, {'name': 'orders', 'schedule': '@hourly', 'source': 's3://raw/orders'}, {'name': 'products', 'schedule': '@weekly', 'source': 's3://raw/products'}, ] def create_dag(config: dict) -> DAG: """Factory function to create DAGs from config""" dag_id = f"etl_{config['name']}" default_args = { 'owner': 'data-team', 'retries': 3, 'retry_delay': timedelta(minutes=5), } dag = DAG( dag_id=dag_id, default_args=default_args, schedule=config['schedule'], start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'dynamic', config['name']], ) with dag: def extract_fn(source, **context): print(f"Extracting from {source} for {context['ds']}") def transform_fn(**context): print(f"Transforming data for {context['ds']}") def load_fn(table_name, **context): print(f"Loading to {table_name} for {context['ds']}") extract = PythonOperator( task_id='extract', python_callable=extract_fn, op_kwargs={'source': config['source']}, ) transform = PythonOperator( task_id='transform', python_callable=transform_fn, ) load = PythonOperator( task_id='load', python_callable=load_fn, op_kwargs={'table_name': config['name']}, ) extract >> transform >> load return dag # Generate DAGs for config in PIPELINE_CONFIGS: globals()[f"dag_{config['name']}"] = create_dag(config) ``` ### Pattern 3: Branching and Conditional Logic ```python # dags/branching_example.py from airflow.decorators import dag, task from airflow.operators.python import BranchPythonOperator from airflow.operators.empty import EmptyOperator from airflow.utils.trigger_rule import TriggerRule @dag( dag_id='branching_pipeline', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, ) def branching_pipeline(): @task() def check_data_quality() -> dict: """Check data quality and return metrics""" quality_score = 0.95 # Simulated return {'score': quality_score, 'rows': 10000} def choose_branch(**context) -> str: """Determine which branch to execute""" ti = context['ti'] metrics = ti.xcom_pull(task_ids='check_data_quality') if metrics['score'] >= 0.9: return 'high_quality_path' elif metrics['score'] >= 0.7: return 'medium_quality_path' else: return 'low_quality_path' quality_check = check_data_quality() branch = BranchPythonOperator( task_id='branch', python_callable=choose_branch, ) high_quality = EmptyOperator(task_id='high_quality_path') medium_quality = EmptyOperator(task_id='medium_quality_path') low_quality = EmptyOperator(task_id='low_quality_path') # Join point - runs after any branch completes join = EmptyOperator( task_id='join', trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, ) quality_check >> branch >> [high_quality, medium_quality, low_quality] >> join branching_pipeline() ``` ### Pattern 4: Sensors and External Dependencies ```python # dags/sensor_patterns.py from datetime import datetime, timedelta from airflow import DAG from airflow.sensors.filesystem import FileSensor from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor from airflow.sensors.external_task import ExternalTaskSensor from airflow.operators.python import PythonOperator with DAG( dag_id='sensor_example', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, ) as dag: # Wait for file on S3 wait_for_file = S3KeySensor( task_id='wait_for_s3_file', bucket_name='data-lake', bucket_key='raw/{{ ds }}/data.parquet', aws_conn_id='aws_default', timeout=60 * 60 * 2, # 2 hours poke_interval=60 * 5, # Check every 5 minutes mode='reschedule', # Free up worker slot while waiting ) # Wait for another DAG to complete wait_for_upstream = ExternalTaskSensor( task_id='wait_for_upstream_dag', external_dag_id='upstream_etl', external_task_id='final_task', execution_date_fn=lambda dt: dt, # Same execution date timeout=60 * 60 * 3, mode='reschedule', ) # Custom sensor using @task.sensor decorator @task.sensor(poke_interval=60, timeout=3600, mode='reschedule') def wait_for_api() -> PokeReturnValue: """Custom sensor for API availability""" import requests response = requests.get('https://api.example.com/health') is_done = response.status_code == 200 return PokeReturnValue(is_done=is_done, xcom_value=response.json()) api_ready = wait_for_api() def process_data(**context): api_result = context['ti'].xcom_pull(task_ids='wait_for_api') print(f"API returned: {api_result}") process = PythonOperator( task_id='process', python_callable=process_data, ) [wait_for_file, wait_for_upstream, api_ready] >> process ``` ### Pattern 5: Error Handling and Alerts ```python # dags/error_handling.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.utils.trigger_rule import TriggerRule from airflow.models import Variable def task_failure_callback(context): """Callback on task failure""" task_instance = context['task_instance'] exception = context.get('exception') # Send to Slack/PagerDuty/etc message = f""" Task Failed! DAG: {task_instance.dag_id} Task: {task_instance.task_id} Execution Date: {context['ds']} Error: {exception} Log URL: {task_instance.log_url} """ # send_slack_alert(message) print(message) def dag_failure_callback(context): """Callback on DAG failure""" # Aggregate failures, send summary pass with DAG( dag_id='error_handling_example', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, on_failure_callback=dag_failure_callback, default_args={ 'on_failure_callback': task_failure_callback, 'retries': 3, 'retry_delay': timedelta(minutes=5), }, ) as dag: def might_fail(**context): import random if random.random() < 0.3: raise ValueError("Random failure!") return "Success" risky_task = PythonOperator( task_id='risky_task', python_callable=might_fail, ) def cleanup(**context): """Cleanup runs regardless of upstream failures""" print("Cleaning up...") cleanup_task = PythonOperator( task_id='cleanup', python_callable=cleanup, trigger_rule=TriggerRule.ALL_DONE, # Run even if upstream fails ) def notify_success(**context): """Only runs if all upstream succeeded""" print("All tasks succeeded!") success_notification = PythonOperator( task_id='notify_success', python_callable=notify_success, trigger_rule=TriggerRule.ALL_SUCCESS, ) risky_task >> [cleanup_task, success_notification] ``` ### Pattern 6: Testing DAGs ```python # tests/test_dags.py import pytest from datetime import datetime from airflow.models import DagBag @pytest.fixture def dagbag(): return DagBag(dag_folder='dags/', include_examples=False) def test_dag_loaded(dagbag): """Test that all DAGs load without errors""" assert len(dagbag.import_errors) == 0, f"DAG import errors: {dagbag.import_errors}" def test_dag_structure(dagbag): """Test specific DAG structure""" dag = dagbag.get_dag('example_etl') assert dag is not None assert len(dag.tasks) == 3 assert dag.schedule_interval == '0 6 * * *' def test_task_dependencies(dagbag): """Test task dependencies are correct""" dag = dagbag.get_dag('example_etl') extract_task = dag.get_task('extract') assert 'start' in [t.task_id for t in extract_task.upstream_list] assert 'end' in [t.task_id for t in extract_task.downstream_list] def test_dag_integrity(dagbag): """Test DAG has no cycles and is valid""" for dag_id, dag in dagbag.dags.items(): assert dag.test_cycle() is None, f"Cycle detected in {dag_id}" # Test individual task logic def test_extract_function(): """Unit test for extract function""" from dags.example_dag import extract_data result = extract_data(ds='2024-01-01') assert 'records' in result assert isinstance(result['records'], int) ``` ## Project Structure ``` airflow/ β”œβ”€β”€ dags/ β”‚ β”œβ”€β”€ __init__.py β”‚ β”œβ”€β”€ common/ β”‚ β”‚ β”œβ”€β”€ __init__.py β”‚ β”‚ β”œβ”€β”€ operators.py # Custom operators β”‚ β”‚ β”œβ”€β”€ sensors.py # Custom sensors β”‚ β”‚ └── callbacks.py # Alert callbacks β”‚ β”œβ”€β”€ etl/ β”‚ β”‚ β”œβ”€β”€ customers.py β”‚ β”‚ └── orders.py β”‚ └── ml/ β”‚ └── training.py β”œβ”€β”€ plugins/ β”‚ └── custom_plugin.py β”œβ”€β”€ tests/ β”‚ β”œβ”€β”€ __init__.py β”‚ β”œβ”€β”€ test_dags.py β”‚ └── test_operators.py β”œβ”€β”€ docker-compose.yml └── requirements.txt ``` ## Best Practices ### Do's - **Use TaskFlow API** - Cleaner code, automatic XCom - **Set timeouts** - Prevent zombie tasks - **Use `mode='reschedule'`** - For sensors, free up workers - **Test DAGs** - Unit tests and integration tests - **Idempotent tasks** - Safe to retry ### Don'ts - **Don't use `depends_on_past=True`** - Creates bottlenecks - **Don't hardcode dates** - Use `{{ ds }}` macros - **Don't use global state** - Tasks should be stateless - **Don't skip catchup blindly** - Understand implications - **Don't put heavy logic in DAG file** - Import from modules ## Resources - [Airflow Documentation](https://airflow.apache.org/docs/) - [Astronomer Guides](https://docs.astronomer.io/learn) - [TaskFlow API](https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html)
πŸ‘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

api-design-principles

Master REST and GraphQL API design principles to build intuitive,

coding
⭐1
# API Design Principles Master REST and GraphQL API design principles to build intuitive, scalable, and maintainable APIs that delight developers and stand the test of time. ## When to Use This Skill - Designing new REST or GraphQL APIs - Refactoring existing APIs for better usability - Establishing API design standards for your team - Reviewing API specifications before implementation - Migrating between API paradigms (REST to GraphQL, etc.) - Creating developer-friendly API documentation - Optimizing APIs for specific use cases (mobile, third-party integrations) ## Core Concepts ### 1. RESTful Design Principles **Resource-Oriented Architecture** - Resources are nouns (users, orders, products), not verbs - Use HTTP methods for actions (GET, POST, PUT, PATCH, DELETE) - URLs represent resource hierarchies - Consistent naming conventions **HTTP Methods Semantics:** - `GET`: Retrieve resources (idempotent, safe) - `POST`: Create new resources - `PUT`: Replace entire resource (idempotent) - `PATCH`: Partial resource updates - `DELETE`: Remove resources (idempotent) ### 2. GraphQL Design Principles **Schema-First Development** - Types define your domain model - Queries for reading data - Mutations for modifying data - Subscriptions for real-time updates **Query Structure:** - Clients request exactly what they need - Single endpoint, multiple operations - Strongly typed schema - Introspection built-in ### 3. API Versioning Strategies **URL Versioning:** ``` /api/v1/users /api/v2/users ``` **Header Versioning:** ``` Accept: application/vnd.api+json; version=1 ``` **Query Parameter Versioning:** ``` /api/users?version=1 ``` ## REST API Design Patterns ### Pattern 1: Resource Collection Design ```python # Good: Resource-oriented endpoints GET /api/users # List users (with pagination) POST /api/users # Create user GET /api/users/{id} # Get specific user PUT /api/users/{id} # Replace user PATCH /api/users/{id} # Update user fields DELETE /api/users/{id} # Delete user # Nested resources GET /api/users/{id}/orders # Get user's orders POST /api/users/{id}/orders # Create order for user # Bad: Action-oriented endpoints (avoid) POST /api/createUser POST /api/getUserById POST /api/deleteUser ``` ### Pattern 2: Pagination and Filtering ```python from typing import List, Optional from pydantic import BaseModel, Field class PaginationParams(BaseModel): page: int = Field(1, ge=1, description="Page number") page_size: int = Field(20, ge=1, le=100, description="Items per page") class FilterParams(BaseModel): status: Optional[str] = None created_after: Optional[str] = None search: Optional[str] = None class PaginatedResponse(BaseModel): items: List[dict] total: int page: int page_size: int pages: int @property def has_next(self) -> bool: return self.page < self.pages @property def has_prev(self) -> bool: return self.page > 1 # FastAPI endpoint example from fastapi import FastAPI, Query, Depends app = FastAPI() @app.get("/api/users", response_model=PaginatedResponse) async def list_users( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), search: Optional[str] = Query(None) ): # Apply filters query = build_query(status=status, search=search) # Count total total = await count_users(query) # Fetch page offset = (page - 1) * page_size users = await fetch_users(query, limit=page_size, offset=offset) return PaginatedResponse( items=users, total=total, page=page, page_size=page_size, pages=(total + page_size - 1) // page_size ) ``` ### Pattern 3: Error Handling and Status Codes ```python from fastapi import HTTPException, status from pydantic import BaseModel class ErrorResponse(BaseModel): error: str message: str details: Optional[dict] = None timestamp: str path: str class ValidationErrorDetail(BaseModel): field: str message: str value: Any # Consistent error responses STATUS_CODES = { "success": 200, "created": 201, "no_content": 204, "bad_request": 400, "unauthorized": 401, "forbidden": 403, "not_found": 404, "conflict": 409, "unprocessable": 422, "internal_error": 500 } def raise_not_found(resource: str, id: str): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": "NotFound", "message": f"{resource} not found", "details": {"id": id} } ) def raise_validation_error(errors: List[ValidationErrorDetail]): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "error": "ValidationError", "message": "Request validation failed", "details": {"errors": [e.dict() for e in errors]} } ) # Example usage @app.get("/api/users/{user_id}") async def get_user(user_id: str): user = await fetch_user(user_id) if not user: raise_not_found("User", user_id) return user ``` ### Pattern 4: HATEOAS (Hypermedia as the Engine of Application State) ```python class UserResponse(BaseModel): id: str name: str email: str _links: dict @classmethod def from_user(cls, user: User, base_url: str): return cls( id=user.id, name=user.name, email=user.email, _links={ "self": {"href": f"{base_url}/api/users/{user.id}"}, "orders": {"href": f"{base_url}/api/users/{user.id}/orders"}, "update": { "href": f"{base_url}/api/users/{user.id}", "method": "PATCH" }, "delete": { "href": f"{base_url}/api/users/{user.id}", "method": "DELETE" } } ) ``` ## GraphQL Design Patterns ### Pattern 1: Schema Design ```graphql # schema.graphql # Clear type definitions type User { id: ID! email: String! name: String! createdAt: DateTime! # Relationships orders(first: Int = 20, after: String, status: OrderStatus): OrderConnection! profile: UserProfile } type Order { id: ID! status: OrderStatus! total: Money! items: [OrderItem!]! createdAt: DateTime! # Back-reference user: User! } # Pagination pattern (Relay-style) type OrderConnection { edges: [OrderEdge!]! pageInfo: PageInfo! totalCount: Int! } type OrderEdge { node: Order! cursor: String! } type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String } # Enums for type safety enum OrderStatus { PENDING CONFIRMED SHIPPED DELIVERED CANCELLED } # Custom scalars scalar DateTime scalar Money # Query root type Query { user(id: ID!): User users(first: Int = 20, after: String, search: String): UserConnection! order(id: ID!): Order } # Mutation root type Mutation { createUser(input: CreateUserInput!): CreateUserPayload! updateUser(input: UpdateUserInput!): UpdateUserPayload! deleteUser(id: ID!): DeleteUserPayload! createOrder(input: CreateOrderInput!): CreateOrderPayload! } # Input types for mutations input CreateUserInput { email: String! name: String! password: String! } # Payload types for mutations type CreateUserPayload { user: User errors: [Error!] } type Error { field: String message: String! } ``` ### Pattern 2: Resolver Design ```python from typing import Optional, List from ariadne import QueryType, MutationType, ObjectType from dataclasses import dataclass query = QueryType() mutation = MutationType() user_type = ObjectType("User") @query.field("user") async def resolve_user(obj, info, id: str) -> Optional[dict]: """Resolve single user by ID.""" return await fetch_user_by_id(id) @query.field("users") async def resolve_users( obj, info, first: int = 20, after: Optional[str] = None, search: Optional[str] = None ) -> dict: """Resolve paginated user list.""" # Decode cursor offset = decode_cursor(after) if after else 0 # Fetch users users = await fetch_users( limit=first + 1, # Fetch one extra to check hasNextPage offset=offset, search=search ) # Pagination has_next = len(users) > first if has_next: users = users[:first] edges = [ { "node": user, "cursor": encode_cursor(offset + i) } for i, user in enumerate(users) ] return { "edges": edges, "pageInfo": { "hasNextPage": has_next, "hasPreviousPage": offset > 0, "startCursor": edges[0]["cursor"] if edges else None, "endCursor": edges[-1]["cursor"] if edges else None }, "totalCount": await count_users(search=search) } @user_type.field("orders") async def resolve_user_orders(user: dict, info, first: int = 20) -> dict: """Resolve user's orders (N+1 prevention with DataLoader).""" # Use DataLoader to batch requests loader = info.context["loaders"]["orders_by_user"] orders = await loader.load(user["id"]) return paginate_orders(orders, first) @mutation.field("createUser") async def resolve_create_user(obj, info, input: dict) -> dict: """Create new user.""" try: # Validate input validate_user_input(input) # Create user user = await create_user( email=input["email"], name=input["name"], password=hash_password(input["password"]) ) return { "user": user, "errors": [] } except ValidationError as e: return { "user": None, "errors": [{"field": e.field, "message": e.message}] } ``` ### Pattern 3: DataLoader (N+1 Problem Prevention) ```python from aiodataloader import DataLoader from typing import List, Optional class UserLoader(DataLoader): """Batch load users by ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[Optional[dict]]: """Load multiple users in single query.""" users = await fetch_users_by_ids(user_ids) # Map results back to input order user_map = {user["id"]: user for user in users} return [user_map.get(user_id) for user_id in user_ids] class OrdersByUserLoader(DataLoader): """Batch load orders by user ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[List[dict]]: """Load orders for multiple users in single query.""" orders = await fetch_orders_by_user_ids(user_ids) # Group orders by user_id orders_by_user = {} for order in orders: user_id = order["user_id"] if user_id not in orders_by_user: orders_by_user[user_id] = [] orders_by_user[user_id].append(order) # Return in input order return [orders_by_user.get(user_id, []) for user_id in user_ids] # Context setup def create_context(): return { "loaders": { "user": UserLoader(), "orders_by_user": OrdersByUserLoader() } } ``` ## Best Practices ### REST APIs 1. **Consistent Naming**: Use plural nouns for collections (`/users`, not `/user`) 2. **Stateless**: Each request contains all necessary information 3. **Use HTTP Status Codes Correctly**: 2xx success, 4xx client errors, 5xx server errors 4. **Version Your API**: Plan for breaking changes from day one 5. **Pagination**: Always paginate large collections 6. **Rate Limiting**: Protect your API with rate limits 7. **Documentation**: Use OpenAPI/Swagger for interactive docs ### GraphQL APIs 1. **Schema First**: Design schema before writing resolvers 2. **Avoid N+1**: Use DataLoaders for efficient data fetching 3. **Input Validation**: Validate at schema and resolver levels 4. **Error Handling**: Return structured errors in mutation payloads 5. **Pagination**: Use cursor-based pagination (Relay spec) 6. **Deprecation**: Use `@deprecated` directive for gradual migration 7. **Monitoring**: Track query complexity and execution time ## Common Pitfalls - **Over-fetching/Under-fetching (REST)**: Fixed in GraphQL but requires DataLoaders - **Breaking Changes**: Version APIs or use deprecation strategies - **Inconsistent Error Formats**: Standardize error responses - **Missing Rate Limits**: APIs without limits are vulnerable to abuse - **Poor Documentation**: Undocumented APIs frustrate developers - **Ignoring HTTP Semantics**: POST for idempotent operations breaks expectations - **Tight Coupling**: API structure shouldn't mirror database schema ## Resources - **references/rest-best-practices.md**: Comprehensive REST API design guide - **references/graphql-schema-design.md**: GraphQL schema patterns and anti-patterns - **references/api-versioning-strategies.md**: Versioning approaches and migration paths - **assets/rest-api-template.py**: FastAPI REST API template - **assets/graphql-schema-template.graphql**: Complete GraphQL schema example - **assets/api-design-checklist.md**: Pre-implementation review checklist - **scripts/openapi-generator.py**: Generate OpenAPI specs from code
πŸ‘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

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

slo-implementation

Define and implement Service Level Indicators (SLIs) and Service

coding
⭐1
# SLO Implementation Framework for defining and implementing Service Level Indicators (SLIs), Service Level Objectives (SLOs), and error budgets. ## Purpose Implement measurable reliability targets using SLIs, SLOs, and error budgets to balance reliability with innovation velocity. ## When to Use - Define service reliability targets - Measure user-perceived reliability - Implement error budgets - Create SLO-based alerts - Track reliability goals ## SLI/SLO/SLA Hierarchy ``` SLA (Service Level Agreement) ↓ Contract with customers SLO (Service Level Objective) ↓ Internal reliability target SLI (Service Level Indicator) ↓ Actual measurement ``` ## Defining SLIs ### Common SLI Types #### 1. Availability SLI ```promql # Successful requests / Total requests sum(rate(http_requests_total{status!~"5.."}[28d])) / sum(rate(http_requests_total[28d])) ``` #### 2. Latency SLI ```promql # Requests below latency threshold / Total requests sum(rate(http_request_duration_seconds_bucket{le="0.5"}[28d])) / sum(rate(http_request_duration_seconds_count[28d])) ``` #### 3. Durability SLI ``` # Successful writes / Total writes sum(storage_writes_successful_total) / sum(storage_writes_total) ``` **Reference:** See `references/slo-definitions.md` ## Setting SLO Targets ### Availability SLO Examples | SLO % | Downtime/Month | Downtime/Year | | ------ | -------------- | ------------- | | 99% | 7.2 hours | 3.65 days | | 99.9% | 43.2 minutes | 8.76 hours | | 99.95% | 21.6 minutes | 4.38 hours | | 99.99% | 4.32 minutes | 52.56 minutes | ### Choose Appropriate SLOs **Consider:** - User expectations - Business requirements - Current performance - Cost of reliability - Competitor benchmarks **Example SLOs:** ```yaml slos: - name: api_availability target: 99.9 window: 28d sli: | sum(rate(http_requests_total{status!~"5.."}[28d])) / sum(rate(http_requests_total[28d])) - name: api_latency_p95 target: 99 window: 28d sli: | sum(rate(http_request_duration_seconds_bucket{le="0.5"}[28d])) / sum(rate(http_request_duration_seconds_count[28d])) ``` ## Error Budget Calculation ### Error Budget Formula ``` Error Budget = 1 - SLO Target ``` **Example:** - SLO: 99.9% availability - Error Budget: 0.1% = 43.2 minutes/month - Current Error: 0.05% = 21.6 minutes/month - Remaining Budget: 50% ### Error Budget Policy ```yaml error_budget_policy: - remaining_budget: 100% action: Normal development velocity - remaining_budget: 50% action: Consider postponing risky changes - remaining_budget: 10% action: Freeze non-critical changes - remaining_budget: 0% action: Feature freeze, focus on reliability ``` **Reference:** See `references/error-budget.md` ## SLO Implementation ### Prometheus Recording Rules ```yaml # SLI Recording Rules groups: - name: sli_rules interval: 30s rules: # Availability SLI - record: sli:http_availability:ratio expr: | sum(rate(http_requests_total{status!~"5.."}[28d])) / sum(rate(http_requests_total[28d])) # Latency SLI (requests < 500ms) - record: sli:http_latency:ratio expr: | sum(rate(http_request_duration_seconds_bucket{le="0.5"}[28d])) / sum(rate(http_request_duration_seconds_count[28d])) - name: slo_rules interval: 5m rules: # SLO compliance (1 = meeting SLO, 0 = violating) - record: slo:http_availability:compliance expr: sli:http_availability:ratio >= bool 0.999 - record: slo:http_latency:compliance expr: sli:http_latency:ratio >= bool 0.99 # Error budget remaining (percentage) - record: slo:http_availability:error_budget_remaining expr: | (sli:http_availability:ratio - 0.999) / (1 - 0.999) * 100 # Error budget burn rate - record: slo:http_availability:burn_rate_5m expr: | (1 - ( sum(rate(http_requests_total{status!~"5.."}[5m])) / sum(rate(http_requests_total[5m])) )) / (1 - 0.999) ``` ### SLO Alerting Rules ```yaml groups: - name: slo_alerts interval: 1m rules: # Fast burn: 14.4x rate, 1 hour window # Consumes 2% error budget in 1 hour - alert: SLOErrorBudgetBurnFast expr: | slo:http_availability:burn_rate_1h > 14.4 and slo:http_availability:burn_rate_5m > 14.4 for: 2m labels: severity: critical annotations: summary: "Fast error budget burn detected" description: "Error budget burning at {{ $value }}x rate" # Slow burn: 6x rate, 6 hour window # Consumes 5% error budget in 6 hours - alert: SLOErrorBudgetBurnSlow expr: | slo:http_availability:burn_rate_6h > 6 and slo:http_availability:burn_rate_30m > 6 for: 15m labels: severity: warning annotations: summary: "Slow error budget burn detected" description: "Error budget burning at {{ $value }}x rate" # Error budget exhausted - alert: SLOErrorBudgetExhausted expr: slo:http_availability:error_budget_remaining < 0 for: 5m labels: severity: critical annotations: summary: "SLO error budget exhausted" description: "Error budget remaining: {{ $value }}%" ``` ## SLO Dashboard **Grafana Dashboard Structure:** ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ SLO Compliance (Current) β”‚ β”‚ βœ“ 99.95% (Target: 99.9%) β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Error Budget Remaining: 65% β”‚ β”‚ β–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–ˆβ–‘β–‘ 65% β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ SLI Trend (28 days) β”‚ β”‚ [Time series graph] β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Burn Rate Analysis β”‚ β”‚ [Burn rate by time window] β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` **Example Queries:** ```promql # Current SLO compliance sli:http_availability:ratio * 100 # Error budget remaining slo:http_availability:error_budget_remaining # Days until error budget exhausted (at current burn rate) (slo:http_availability:error_budget_remaining / 100) * 28 / (1 - sli:http_availability:ratio) * (1 - 0.999) ``` ## Multi-Window Burn Rate Alerts ```yaml # Combination of short and long windows reduces false positives rules: - alert: SLOBurnRateHigh expr: | ( slo:http_availability:burn_rate_1h > 14.4 and slo:http_availability:burn_rate_5m > 14.4 ) or ( slo:http_availability:burn_rate_6h > 6 and slo:http_availability:burn_rate_30m > 6 ) labels: severity: critical ``` ## SLO Review Process ### Weekly Review - Current SLO compliance - Error budget status - Trend analysis - Incident impact ### Monthly Review - SLO achievement - Error budget usage - Incident postmortems - SLO adjustments ### Quarterly Review - SLO relevance - Target adjustments - Process improvements - Tooling enhancements ## Best Practices 1. **Start with user-facing services** 2. **Use multiple SLIs** (availability, latency, etc.) 3. **Set achievable SLOs** (don't aim for 100%) 4. **Implement multi-window alerts** to reduce noise 5. **Track error budget** consistently 6. **Review SLOs regularly** 7. **Document SLO decisions** 8. **Align with business goals** 9. **Automate SLO reporting** 10. **Use SLOs for prioritization** ## Reference Files - `assets/slo-template.md` - SLO definition template - `references/slo-definitions.md` - SLO definition patterns - `references/error-budget.md` - Error budget calculations ## Related Skills - `prometheus-configuration` - For metric collection - `grafana-dashboards` - For SLO visualization
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

async-python-patterns

Master Python asyncio, concurrent programming, and async/await

coding
⭐1
# Async Python Patterns Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. ## When to Use This Skill - Building async web APIs (FastAPI, aiohttp, Sanic) - Implementing concurrent I/O operations (database, file, network) - Creating web scrapers with concurrent requests - Developing real-time applications (WebSocket servers, chat systems) - Processing multiple independent tasks simultaneously - Building microservices with async communication - Optimizing I/O-bound workloads - Implementing async background tasks and queues ## Sync vs Async Decision Guide Before adopting async, consider whether it's the right choice for your use case. | Use Case | Recommended Approach | |----------|---------------------| | Many concurrent network/DB calls | `asyncio` | | CPU-bound computation | `multiprocessing` or thread pool | | Mixed I/O + CPU | Offload CPU work with `asyncio.to_thread()` | | Simple scripts, few connections | Sync (simpler, easier to debug) | | Web APIs with high concurrency | Async frameworks (FastAPI, aiohttp) | **Key Rule:** Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity. ## Core Concepts ### 1. Event Loop The event loop is the heart of asyncio, managing and scheduling asynchronous tasks. **Key characteristics:** - Single-threaded cooperative multitasking - Schedules coroutines for execution - Handles I/O operations without blocking - Manages callbacks and futures ### 2. Coroutines Functions defined with `async def` that can be paused and resumed. **Syntax:** ```python async def my_coroutine(): result = await some_async_operation() return result ``` ### 3. Tasks Scheduled coroutines that run concurrently on the event loop. ### 4. Futures Low-level objects representing eventual results of async operations. ### 5. Async Context Managers Resources that support `async with` for proper cleanup. ### 6. Async Iterators Objects that support `async for` for iterating over async data sources. ## Quick Start ```python import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7+ asyncio.run(main()) ``` ## Fundamental Patterns ### Pattern 1: Basic Async/Await ```python import asyncio async def fetch_data(url: str) -> dict: """Fetch data from URL asynchronously.""" await asyncio.sleep(1) # Simulate I/O return {"url": url, "data": "result"} async def main(): result = await fetch_data("https://api.example.com") print(result) asyncio.run(main()) ``` ### Pattern 2: Concurrent Execution with gather() ```python import asyncio from typing import List async def fetch_user(user_id: int) -> dict: """Fetch user data.""" await asyncio.sleep(0.5) return {"id": user_id, "name": f"User {user_id}"} async def fetch_all_users(user_ids: List[int]) -> List[dict]: """Fetch multiple users concurrently.""" tasks = [fetch_user(uid) for uid in user_ids] results = await asyncio.gather(*tasks) return results async def main(): user_ids = [1, 2, 3, 4, 5] users = await fetch_all_users(user_ids) print(f"Fetched {len(users)} users") asyncio.run(main()) ``` ### Pattern 3: Task Creation and Management ```python import asyncio async def background_task(name: str, delay: int): """Long-running background task.""" print(f"{name} started") await asyncio.sleep(delay) print(f"{name} completed") return f"Result from {name}" async def main(): # Create tasks task1 = asyncio.create_task(background_task("Task 1", 2)) task2 = asyncio.create_task(background_task("Task 2", 1)) # Do other work print("Main: doing other work") await asyncio.sleep(0.5) # Wait for tasks result1 = await task1 result2 = await task2 print(f"Results: {result1}, {result2}") asyncio.run(main()) ``` ### Pattern 4: Error Handling in Async Code ```python import asyncio from typing import List, Optional async def risky_operation(item_id: int) -> dict: """Operation that might fail.""" await asyncio.sleep(0.1) if item_id % 3 == 0: raise ValueError(f"Item {item_id} failed") return {"id": item_id, "status": "success"} async def safe_operation(item_id: int) -> Optional[dict]: """Wrapper with error handling.""" try: return await risky_operation(item_id) except ValueError as e: print(f"Error: {e}") return None async def process_items(item_ids: List[int]): """Process multiple items with error handling.""" tasks = [safe_operation(iid) for iid in item_ids] results = await asyncio.gather(*tasks, return_exceptions=True) # Filter out failures successful = [r for r in results if r is not None and not isinstance(r, Exception)] failed = [r for r in results if isinstance(r, Exception)] print(f"Success: {len(successful)}, Failed: {len(failed)}") return successful asyncio.run(process_items([1, 2, 3, 4, 5, 6])) ``` ### Pattern 5: Timeout Handling ```python import asyncio async def slow_operation(delay: int) -> str: """Operation that takes time.""" await asyncio.sleep(delay) return f"Completed after {delay}s" async def with_timeout(): """Execute operation with timeout.""" try: result = await asyncio.wait_for(slow_operation(5), timeout=2.0) print(result) except asyncio.TimeoutError: print("Operation timed out") asyncio.run(with_timeout()) ``` ## Advanced Patterns ### Pattern 6: Async Context Managers ```python import asyncio from typing import Optional class AsyncDatabaseConnection: """Async database connection context manager.""" def __init__(self, dsn: str): self.dsn = dsn self.connection: Optional[object] = None async def __aenter__(self): print("Opening connection") await asyncio.sleep(0.1) # Simulate connection self.connection = {"dsn": self.dsn, "connected": True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print("Closing connection") await asyncio.sleep(0.1) # Simulate cleanup self.connection = None async def query_database(): """Use async context manager.""" async with AsyncDatabaseConnection("postgresql://localhost") as conn: print(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database()) ``` ### Pattern 7: Async Iterators and Generators ```python import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]: """Async generator that yields numbers with delay.""" for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]: """Fetch paginated data asynchronously.""" for page in range(1, max_pages + 1): await asyncio.sleep(0.2) # Simulate API call yield { "page": page, "url": f"{url}?page={page}", "data": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator()) ``` ### Pattern 8: Producer-Consumer Pattern ```python import asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): """Produce items and put them in queue.""" for i in range(num_items): item = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {producer_id} produced: {item}") await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): """Consume items from queue.""" while True: item = await queue.get() if item is None: queue.task_done() break print(f"Consumer {consumer_id} processing: {item}") await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): """Run producer-consumer pattern.""" queue = Queue(maxsize=10) # Create tasks producers = [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers = [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example()) ``` ### Pattern 9: Semaphore for Rate Limiting ```python import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict: """Make API call with rate limiting.""" async with semaphore: print(f"Calling {url}") await asyncio.sleep(0.5) # Simulate API call return {"url": url, "status": 200} async def rate_limited_requests(urls: List[str], max_concurrent: int = 5): """Make multiple requests with rate limiting.""" semaphore = asyncio.Semaphore(max_concurrent) tasks = [api_call(url, semaphore) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [f"https://api.example.com/item/{i}" for i in range(20)] results = await rate_limited_requests(urls, max_concurrent=3) print(f"Completed {len(results)} requests") asyncio.run(main()) ``` ### Pattern 10: Async Locks and Synchronization ```python import asyncio class AsyncCounter: """Thread-safe async counter.""" def __init__(self): self.value = 0 self.lock = asyncio.Lock() async def increment(self): """Safely increment counter.""" async with self.lock: current = self.value await asyncio.sleep(0.01) # Simulate work self.value = current + 1 async def get_value(self) -> int: """Get current value.""" async with self.lock: return self.value async def worker(counter: AsyncCounter, worker_id: int): """Worker that increments counter.""" for _ in range(10): await counter.increment() print(f"Worker {worker_id} incremented") async def test_counter(): """Test concurrent counter.""" counter = AsyncCounter() workers = [asyncio.create_task(worker(counter, i)) for i in range(5)] await asyncio.gather(*workers) final_value = await counter.get_value() print(f"Final counter value: {final_value}") asyncio.run(test_counter()) ``` ## Real-World Applications ### Web Scraping with aiohttp ```python import asyncio import aiohttp from typing import List, Dict async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict: """Fetch single URL.""" try: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: text = await response.text() return { "url": url, "status": response.status, "length": len(text) } except Exception as e: return {"url": url, "error": str(e)} async def scrape_urls(urls: List[str]) -> List[Dict]: """Scrape multiple URLs concurrently.""" async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/status/404", ] results = await scrape_urls(urls) for result in results: print(result) asyncio.run(main()) ``` ### Async Database Operations ```python import asyncio from typing import List, Optional # Simulated async database client class AsyncDB: """Simulated async database.""" async def execute(self, query: str) -> List[dict]: """Execute query.""" await asyncio.sleep(0.1) return [{"id": 1, "name": "Example"}] async def fetch_one(self, query: str) -> Optional[dict]: """Fetch single row.""" await asyncio.sleep(0.1) return {"id": 1, "name": "Example"} async def get_user_data(db: AsyncDB, user_id: int) -> dict: """Fetch user and related data concurrently.""" user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}") orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}") user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task) return { "user": user, "orders": orders, "profile": profile } async def main(): db = AsyncDB() user_data = await get_user_data(db, 1) print(user_data) asyncio.run(main()) ``` ### WebSocket Server ```python import asyncio from typing import Set # Simulated WebSocket connection class WebSocket: """Simulated WebSocket.""" def __init__(self, client_id: str): self.client_id = client_id async def send(self, message: str): """Send message.""" print(f"Sending to {self.client_id}: {message}") await asyncio.sleep(0.01) async def recv(self) -> str: """Receive message.""" await asyncio.sleep(1) return f"Message from {self.client_id}" class WebSocketServer: """Simple WebSocket server.""" def __init__(self): self.clients: Set[WebSocket] = set() async def register(self, websocket: WebSocket): """Register new client.""" self.clients.add(websocket) print(f"Client {websocket.client_id} connected") async def unregister(self, websocket: WebSocket): """Unregister client.""" self.clients.remove(websocket) print(f"Client {websocket.client_id} disconnected") async def broadcast(self, message: str): """Broadcast message to all clients.""" if self.clients: tasks = [client.send(message) for client in self.clients] await asyncio.gather(*tasks) async def handle_client(self, websocket: WebSocket): """Handle individual client connection.""" await self.register(websocket) try: async for message in self.message_iterator(websocket): await self.broadcast(f"{websocket.client_id}: {message}") finally: await self.unregister(websocket) async def message_iterator(self, websocket: WebSocket): """Iterate over messages from client.""" for _ in range(3): # Simulate 3 messages yield await websocket.recv() ``` ## Performance Best Practices ### 1. Use Connection Pools ```python import asyncio import aiohttp async def with_connection_pool(): """Use connection pool for efficiency.""" connector = aiohttp.TCPConnector(limit=100, limit_per_host=10) async with aiohttp.ClientSession(connector=connector) as session: tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return responses ``` ### 2. Batch Operations ```python async def batch_process(items: List[str], batch_size: int = 10): """Process items in batches.""" for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] tasks = [process_item(item) for item in batch] await asyncio.gather(*tasks) print(f"Processed batch {i // batch_size + 1}") async def process_item(item: str): """Process single item.""" await asyncio.sleep(0.1) return f"Processed: {item}" ``` ### 3. Avoid Blocking Operations Never block the event loop with synchronous operations. A single blocking call stalls all concurrent tasks. ```python # BAD - blocks the entire event loop async def fetch_data_bad(): import time import requests time.sleep(1) # Blocks! response = requests.get(url) # Also blocks! # GOOD - use async-native libraries (e.g., httpx for async HTTP) import httpx async def fetch_data_good(url: str): await asyncio.sleep(1) async with httpx.AsyncClient() as client: response = await client.get(url) ``` **Wrapping Blocking Code with `asyncio.to_thread()` (Python 3.9+):** When you must use synchronous libraries, offload to a thread pool: ```python import asyncio from pathlib import Path async def read_file_async(path: str) -> str: """Read file without blocking event loop.""" # asyncio.to_thread() runs sync code in a thread pool return await asyncio.to_thread(Path(path).read_text) async def call_sync_library(data: dict) -> dict: """Wrap a synchronous library call.""" # Useful for sync database drivers, file I/O, CPU work return await asyncio.to_thread(sync_library.process, data) ``` **Lower-level approach with `run_in_executor()`:** ```python import asyncio import concurrent.futures from typing import Any def blocking_operation(data: Any) -> Any: """CPU-intensive blocking operation.""" import time time.sleep(1) return data * 2 async def run_in_executor(data: Any) -> Any: """Run blocking operation in thread pool.""" loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_operation, data) return result async def main(): results = await asyncio.gather(*[run_in_executor(i) for i in range(5)]) print(results) asyncio.run(main()) ``` ## Common Pitfalls ### 1. Forgetting await ```python # Wrong - returns coroutine object, doesn't execute result = async_function() # Correct result = await async_function() ``` ### 2. Blocking the Event Loop ```python # Wrong - blocks event loop import time async def bad(): time.sleep(1) # Blocks! # Correct async def good(): await asyncio.sleep(1) # Non-blocking ``` ### 3. Not Handling Cancellation ```python async def cancelable_task(): """Task that handles cancellation.""" try: while True: await asyncio.sleep(1) print("Working...") except asyncio.CancelledError: print("Task cancelled, cleaning up...") # Perform cleanup raise # Re-raise to propagate cancellation ``` ### 4. Mixing Sync and Async Code ```python # Wrong - can't call async from sync directly def sync_function(): result = await async_function() # SyntaxError! # Correct def sync_function(): result = asyncio.run(async_function()) ``` ## Testing Async Code ```python import asyncio import pytest # Using pytest-asyncio @pytest.mark.asyncio async def test_async_function(): """Test async function.""" result = await fetch_data("https
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-performance-optimization

Profile and optimize Python code using cProfile, memory profilers,

coding
⭐1
# Python Performance Optimization Comprehensive guide to profiling, analyzing, and optimizing Python code for better performance, including CPU profiling, memory optimization, and implementation best practices. ## When to Use This Skill - Identifying performance bottlenecks in Python applications - Reducing application latency and response times - Optimizing CPU-intensive operations - Reducing memory consumption and memory leaks - Improving database query performance - Optimizing I/O operations - Speeding up data processing pipelines - Implementing high-performance algorithms - Profiling production applications ## Core Concepts ### 1. Profiling Types - **CPU Profiling**: Identify time-consuming functions - **Memory Profiling**: Track memory allocation and leaks - **Line Profiling**: Profile at line-by-line granularity - **Call Graph**: Visualize function call relationships ### 2. Performance Metrics - **Execution Time**: How long operations take - **Memory Usage**: Peak and average memory consumption - **CPU Utilization**: Processor usage patterns - **I/O Wait**: Time spent on I/O operations ### 3. Optimization Strategies - **Algorithmic**: Better algorithms and data structures - **Implementation**: More efficient code patterns - **Parallelization**: Multi-threading/processing - **Caching**: Avoid redundant computation - **Native Extensions**: C/Rust for critical paths ## Quick Start ### Basic Timing ```python import time def measure_time(): """Simple timing measurement.""" start = time.time() # Your code here result = sum(range(1000000)) elapsed = time.time() - start print(f"Execution time: {elapsed:.4f} seconds") return result # Better: use timeit for accurate measurements import timeit execution_time = timeit.timeit( "sum(range(1000000))", number=100 ) print(f"Average time: {execution_time/100:.6f} seconds") ``` ## Profiling Tools ### Pattern 1: cProfile - CPU Profiling ```python import cProfile import pstats from pstats import SortKey def slow_function(): """Function to profile.""" total = 0 for i in range(1000000): total += i return total def another_function(): """Another function.""" return [i**2 for i in range(100000)] def main(): """Main function to profile.""" result1 = slow_function() result2 = another_function() return result1, result2 # Profile the code if __name__ == "__main__": profiler = cProfile.Profile() profiler.enable() main() profiler.disable() # Print stats stats = pstats.Stats(profiler) stats.sort_stats(SortKey.CUMULATIVE) stats.print_stats(10) # Top 10 functions # Save to file for later analysis stats.dump_stats("profile_output.prof") ``` **Command-line profiling:** ```bash # Profile a script python -m cProfile -o output.prof script.py # View results python -m pstats output.prof # In pstats: # sort cumtime # stats 10 ``` ### Pattern 2: line_profiler - Line-by-Line Profiling ```python # Install: pip install line-profiler # Add @profile decorator (line_profiler provides this) @profile def process_data(data): """Process data with line profiling.""" result = [] for item in data: processed = item * 2 result.append(processed) return result # Run with: # kernprof -l -v script.py ``` **Manual line profiling:** ```python from line_profiler import LineProfiler def process_data(data): """Function to profile.""" result = [] for item in data: processed = item * 2 result.append(processed) return result if __name__ == "__main__": lp = LineProfiler() lp.add_function(process_data) data = list(range(100000)) lp_wrapper = lp(process_data) lp_wrapper(data) lp.print_stats() ``` ### Pattern 3: memory_profiler - Memory Usage ```python # Install: pip install memory-profiler from memory_profiler import profile @profile def memory_intensive(): """Function that uses lots of memory.""" # Create large list big_list = [i for i in range(1000000)] # Create large dict big_dict = {i: i**2 for i in range(100000)} # Process data result = sum(big_list) return result if __name__ == "__main__": memory_intensive() # Run with: # python -m memory_profiler script.py ``` ### Pattern 4: py-spy - Production Profiling ```bash # Install: pip install py-spy # Profile a running Python process py-spy top --pid 12345 # Generate flamegraph py-spy record -o profile.svg --pid 12345 # Profile a script py-spy record -o profile.svg -- python script.py # Dump current call stack py-spy dump --pid 12345 ``` ## Optimization Patterns ### Pattern 5: List Comprehensions vs Loops ```python import timeit # Slow: Traditional loop def slow_squares(n): """Create list of squares using loop.""" result = [] for i in range(n): result.append(i**2) return result # Fast: List comprehension def fast_squares(n): """Create list of squares using comprehension.""" return [i**2 for i in range(n)] # Benchmark n = 100000 slow_time = timeit.timeit(lambda: slow_squares(n), number=100) fast_time = timeit.timeit(lambda: fast_squares(n), number=100) print(f"Loop: {slow_time:.4f}s") print(f"Comprehension: {fast_time:.4f}s") print(f"Speedup: {slow_time/fast_time:.2f}x") # Even faster for simple operations: map def faster_squares(n): """Use map for even better performance.""" return list(map(lambda x: x**2, range(n))) ``` ### Pattern 6: Generator Expressions for Memory ```python import sys def list_approach(): """Memory-intensive list.""" data = [i**2 for i in range(1000000)] return sum(data) def generator_approach(): """Memory-efficient generator.""" data = (i**2 for i in range(1000000)) return sum(data) # Memory comparison list_data = [i for i in range(1000000)] gen_data = (i for i in range(1000000)) print(f"List size: {sys.getsizeof(list_data)} bytes") print(f"Generator size: {sys.getsizeof(gen_data)} bytes") # Generators use constant memory regardless of size ``` ### Pattern 7: String Concatenation ```python import timeit def slow_concat(items): """Slow string concatenation.""" result = "" for item in items: result += str(item) return result def fast_concat(items): """Fast string concatenation with join.""" return "".join(str(item) for item in items) def faster_concat(items): """Even faster with list.""" parts = [str(item) for item in items] return "".join(parts) items = list(range(10000)) # Benchmark slow = timeit.timeit(lambda: slow_concat(items), number=100) fast = timeit.timeit(lambda: fast_concat(items), number=100) faster = timeit.timeit(lambda: faster_concat(items), number=100) print(f"Concatenation (+): {slow:.4f}s") print(f"Join (generator): {fast:.4f}s") print(f"Join (list): {faster:.4f}s") ``` ### Pattern 8: Dictionary Lookups vs List Searches ```python import timeit # Create test data size = 10000 items = list(range(size)) lookup_dict = {i: i for i in range(size)} def list_search(items, target): """O(n) search in list.""" return target in items def dict_search(lookup_dict, target): """O(1) search in dict.""" return target in lookup_dict target = size - 1 # Worst case for list # Benchmark list_time = timeit.timeit( lambda: list_search(items, target), number=1000 ) dict_time = timeit.timeit( lambda: dict_search(lookup_dict, target), number=1000 ) print(f"List search: {list_time:.6f}s") print(f"Dict search: {dict_time:.6f}s") print(f"Speedup: {list_time/dict_time:.0f}x") ``` ### Pattern 9: Local Variable Access ```python import timeit # Global variable (slow) GLOBAL_VALUE = 100 def use_global(): """Access global variable.""" total = 0 for i in range(10000): total += GLOBAL_VALUE return total def use_local(): """Use local variable.""" local_value = 100 total = 0 for i in range(10000): total += local_value return total # Local is faster global_time = timeit.timeit(use_global, number=1000) local_time = timeit.timeit(use_local, number=1000) print(f"Global access: {global_time:.4f}s") print(f"Local access: {local_time:.4f}s") print(f"Speedup: {global_time/local_time:.2f}x") ``` ### Pattern 10: Function Call Overhead ```python import timeit def calculate_inline(): """Inline calculation.""" total = 0 for i in range(10000): total += i * 2 + 1 return total def helper_function(x): """Helper function.""" return x * 2 + 1 def calculate_with_function(): """Calculation with function calls.""" total = 0 for i in range(10000): total += helper_function(i) return total # Inline is faster due to no call overhead inline_time = timeit.timeit(calculate_inline, number=1000) function_time = timeit.timeit(calculate_with_function, number=1000) print(f"Inline: {inline_time:.4f}s") print(f"Function calls: {function_time:.4f}s") ``` ## Advanced Optimization ### Pattern 11: NumPy for Numerical Operations ```python import timeit import numpy as np def python_sum(n): """Sum using pure Python.""" return sum(range(n)) def numpy_sum(n): """Sum using NumPy.""" return np.arange(n).sum() n = 1000000 python_time = timeit.timeit(lambda: python_sum(n), number=100) numpy_time = timeit.timeit(lambda: numpy_sum(n), number=100) print(f"Python: {python_time:.4f}s") print(f"NumPy: {numpy_time:.4f}s") print(f"Speedup: {python_time/numpy_time:.2f}x") # Vectorized operations def python_multiply(): """Element-wise multiplication in Python.""" a = list(range(100000)) b = list(range(100000)) return [x * y for x, y in zip(a, b)] def numpy_multiply(): """Vectorized multiplication in NumPy.""" a = np.arange(100000) b = np.arange(100000) return a * b py_time = timeit.timeit(python_multiply, number=100) np_time = timeit.timeit(numpy_multiply, number=100) print(f"\nPython multiply: {py_time:.4f}s") print(f"NumPy multiply: {np_time:.4f}s") print(f"Speedup: {py_time/np_time:.2f}x") ``` ### Pattern 12: Caching with functools.lru_cache ```python from functools import lru_cache import timeit def fibonacci_slow(n): """Recursive fibonacci without caching.""" if n < 2: return n return fibonacci_slow(n-1) + fibonacci_slow(n-2) @lru_cache(maxsize=None) def fibonacci_fast(n): """Recursive fibonacci with caching.""" if n < 2: return n return fibonacci_fast(n-1) + fibonacci_fast(n-2) # Massive speedup for recursive algorithms n = 30 slow_time = timeit.timeit(lambda: fibonacci_slow(n), number=1) fast_time = timeit.timeit(lambda: fibonacci_fast(n), number=1000) print(f"Without cache (1 run): {slow_time:.4f}s") print(f"With cache (1000 runs): {fast_time:.4f}s") # Cache info print(f"Cache info: {fibonacci_fast.cache_info()}") ``` ### Pattern 13: Using **slots** for Memory ```python import sys class RegularClass: """Regular class with __dict__.""" def __init__(self, x, y, z): self.x = x self.y = y self.z = z class SlottedClass: """Class with __slots__ for memory efficiency.""" __slots__ = ['x', 'y', 'z'] def __init__(self, x, y, z): self.x = x self.y = y self.z = z # Memory comparison regular = RegularClass(1, 2, 3) slotted = SlottedClass(1, 2, 3) print(f"Regular class size: {sys.getsizeof(regular)} bytes") print(f"Slotted class size: {sys.getsizeof(slotted)} bytes") # Significant savings with many instances regular_objects = [RegularClass(i, i+1, i+2) for i in range(10000)] slotted_objects = [SlottedClass(i, i+1, i+2) for i in range(10000)] print(f"\nMemory for 10000 regular objects: ~{sys.getsizeof(regular) * 10000} bytes") print(f"Memory for 10000 slotted objects: ~{sys.getsizeof(slotted) * 10000} bytes") ``` ### Pattern 14: Multiprocessing for CPU-Bound Tasks ```python import multiprocessing as mp import time def cpu_intensive_task(n): """CPU-intensive calculation.""" return sum(i**2 for i in range(n)) def sequential_processing(): """Process tasks sequentially.""" start = time.time() results = [cpu_intensive_task(1000000) for _ in range(4)] elapsed = time.time() - start return elapsed, results def parallel_processing(): """Process tasks in parallel.""" start = time.time() with mp.Pool(processes=4) as pool: results = pool.map(cpu_intensive_task, [1000000] * 4) elapsed = time.time() - start return elapsed, results if __name__ == "__main__": seq_time, seq_results = sequential_processing() par_time, par_results = parallel_processing() print(f"Sequential: {seq_time:.2f}s") print(f"Parallel: {par_time:.2f}s") print(f"Speedup: {seq_time/par_time:.2f}x") ``` ### Pattern 15: Async I/O for I/O-Bound Tasks ```python import asyncio import aiohttp import time import requests urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/1", "https://httpbin.org/delay/1", "https://httpbin.org/delay/1", ] def synchronous_requests(): """Synchronous HTTP requests.""" start = time.time() results = [] for url in urls: response = requests.get(url) results.append(response.status_code) elapsed = time.time() - start return elapsed, results async def async_fetch(session, url): """Async HTTP request.""" async with session.get(url) as response: return response.status async def asynchronous_requests(): """Asynchronous HTTP requests.""" start = time.time() async with aiohttp.ClientSession() as session: tasks = [async_fetch(session, url) for url in urls] results = await asyncio.gather(*tasks) elapsed = time.time() - start return elapsed, results # Async is much faster for I/O-bound work sync_time, sync_results = synchronous_requests() async_time, async_results = asyncio.run(asynchronous_requests()) print(f"Synchronous: {sync_time:.2f}s") print(f"Asynchronous: {async_time:.2f}s") print(f"Speedup: {sync_time/async_time:.2f}x") ``` ## Database Optimization ### Pattern 16: Batch Database Operations ```python import sqlite3 import time def create_db(): """Create test database.""" conn = sqlite3.connect(":memory:") conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)") return conn def slow_inserts(conn, count): """Insert records one at a time.""" start = time.time() cursor = conn.cursor() for i in range(count): cursor.execute("INSERT INTO users (name) VALUES (?)", (f"User {i}",)) conn.commit() # Commit each insert elapsed = time.time() - start return elapsed def fast_inserts(conn, count): """Batch insert with single commit.""" start = time.time() cursor = conn.cursor() data = [(f"User {i}",) for i in range(count)] cursor.executemany("INSERT INTO users (name) VALUES (?)", data) conn.commit() # Single commit elapsed = time.time() - start return elapsed # Benchmark conn1 = create_db() slow_time = slow_inserts(conn1, 1000) conn2 = create_db() fast_time = fast_inserts(conn2, 1000) print(f"Individual inserts: {slow_time:.4f}s") print(f"Batch insert: {fast_time:.4f}s") print(f"Speedup: {slow_time/fast_time:.2f}x") ``` ### Pattern 17: Query Optimization ```python # Use indexes for frequently queried columns """ -- Slow: No index SELECT * FROM users WHERE email = 'user@example.com'; -- Fast: With index CREATE INDEX idx_users_email ON users(email); SELECT * FROM users WHERE email = 'user@example.com'; """ # Use query planning import sqlite3 conn = sqlite3.connect("example.db") cursor = conn.cursor() # Analyze query performance cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ("test@example.com",)) print(cursor.fetchall()) # Use SELECT only needed columns # Slow: SELECT * # Fast: SELECT id, name ``` ## Memory Optimization ### Pattern 18: Detecting Memory Leaks ```python import tracemalloc import gc def memory_leak_example(): """Example that leaks memory.""" leaked_objects = [] for i in range(100000): # Objects added but never removed leaked_objects.append([i] * 100) # In real code, this would be an unintended reference def track_memory_usage(): """Track memory allocations.""" tracemalloc.start() # Take snapshot before snapshot1 = tracemalloc.take_snapshot() # Run code memory_leak_example() # Take snapshot after snapshot2 = tracemalloc.take_snapshot() # Compare top_stats = snapshot2.compare_to(snapshot1, 'lineno') print("Top 10 memory allocations:") for stat in top_stats[:10]: print(stat) tracemalloc.stop() # Monitor memory track_memory_usage() # Force garbage collection gc.collect() ``` ### Pattern 19: Iterators vs Lists ```python import sys def process_file_list(filename): """Load entire file into memory.""" with open(filename) as f: lines = f.readlines() # Loads all lines return sum(1 for line in lines if line.strip()) def process_file_iterator(filename): """Process file line by line.""" with open(filename) as f: return sum(1 for line in f if line.strip()) # Iterator uses constant memory # List loads entire file into memory ``` ### Pattern 20: Weakref for Caches ```python import weakref class CachedResource: """Resource that can be garbage collected.""" def __init__(self, data): self.data = data # Regular cache prevents garbage collection regular_cache = {} def get_resource_regular(key): """Get resource from regular cache.""" if key not in regular_cache: regular_cache[key] = CachedResource(f"Data for {key}") return regular_cache[key] # Weak reference cache allows garbage collection weak_cache = weakref.WeakValueDictionary() def get_resource_weak(key): """Get resource from weak cache.""" resource = weak_cache.get(key) if resource is None: resource = CachedResource(f"Data for {key}") weak_cache[key] = resource return resource # When no strong references exist, objects can be GC'd ``` ## Benchmarking Tools ### Custom Benchmark Decorator ```python import time from functools import wraps def benchmark(func): """Decorator to benchmark function execution.""" @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} took {elapsed:.6f} seconds") return result return wrapper @benchmark def slow_function(): """Function to benchmark.""" time.sleep(0.5) return sum(range(1000000)) result = slow_function() ``` ### Performance Testing with pytest-benchmark ```python # Install: pip install pytest-benchmark def test_list_comprehension(benchmark): """Benchmark list comprehension.""" result = benchmark(lambda: [i**2 for i in range(10000)]) assert len(result) == 10000 def test_map_function(benchmark): """Benchmark map function.""" result = benchmark(lambda: list(m
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-testing-patterns

Implement comprehensive testing strategies with pytest, fixtures,

coding
⭐1
# Python Testing Patterns Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices. ## When to Use This Skill - Writing unit tests for Python code - Setting up test suites and test infrastructure - Implementing test-driven development (TDD) - Creating integration tests for APIs and services - Mocking external dependencies and services - Testing async code and concurrent operations - Setting up continuous testing in CI/CD - Implementing property-based testing - Testing database operations - Debugging failing tests ## Core Concepts ### 1. Test Types - **Unit Tests**: Test individual functions/classes in isolation - **Integration Tests**: Test interaction between components - **Functional Tests**: Test complete features end-to-end - **Performance Tests**: Measure speed and resource usage ### 2. Test Structure (AAA Pattern) - **Arrange**: Set up test data and preconditions - **Act**: Execute the code under test - **Assert**: Verify the results ### 3. Test Coverage - Measure what code is exercised by tests - Identify untested code paths - Aim for meaningful coverage, not just high percentages ### 4. Test Isolation - Tests should be independent - No shared state between tests - Each test should clean up after itself ## Quick Start ```python # test_example.py def add(a, b): return a + b def test_add(): """Basic test example.""" result = add(2, 3) assert result == 5 def test_add_negative(): """Test with negative numbers.""" assert add(-1, 1) == 0 # Run with: pytest test_example.py ``` ## Fundamental Patterns ### Pattern 1: Basic pytest Tests ```python # test_calculator.py import pytest class Calculator: """Simple calculator for testing.""" def add(self, a: float, b: float) -> float: return a + b def subtract(self, a: float, b: float) -> float: return a - b def multiply(self, a: float, b: float) -> float: return a * b def divide(self, a: float, b: float) -> float: if b == 0: raise ValueError("Cannot divide by zero") return a / b def test_addition(): """Test addition.""" calc = Calculator() assert calc.add(2, 3) == 5 assert calc.add(-1, 1) == 0 assert calc.add(0, 0) == 0 def test_subtraction(): """Test subtraction.""" calc = Calculator() assert calc.subtract(5, 3) == 2 assert calc.subtract(0, 5) == -5 def test_multiplication(): """Test multiplication.""" calc = Calculator() assert calc.multiply(3, 4) == 12 assert calc.multiply(0, 5) == 0 def test_division(): """Test division.""" calc = Calculator() assert calc.divide(6, 3) == 2 assert calc.divide(5, 2) == 2.5 def test_division_by_zero(): """Test division by zero raises error.""" calc = Calculator() with pytest.raises(ValueError, match="Cannot divide by zero"): calc.divide(5, 0) ``` ### Pattern 2: Fixtures for Setup and Teardown ```python # test_database.py import pytest from typing import Generator class Database: """Simple database class.""" def __init__(self, connection_string: str): self.connection_string = connection_string self.connected = False def connect(self): """Connect to database.""" self.connected = True def disconnect(self): """Disconnect from database.""" self.connected = False def query(self, sql: str) -> list: """Execute query.""" if not self.connected: raise RuntimeError("Not connected") return [{"id": 1, "name": "Test"}] @pytest.fixture def db() -> Generator[Database, None, None]: """Fixture that provides connected database.""" # Setup database = Database("sqlite:///:memory:") database.connect() # Provide to test yield database # Teardown database.disconnect() def test_database_query(db): """Test database query with fixture.""" results = db.query("SELECT * FROM users") assert len(results) == 1 assert results[0]["name"] == "Test" @pytest.fixture(scope="session") def app_config(): """Session-scoped fixture - created once per test session.""" return { "database_url": "postgresql://localhost/test", "api_key": "test-key", "debug": True } @pytest.fixture(scope="module") def api_client(app_config): """Module-scoped fixture - created once per test module.""" # Setup expensive resource client = {"config": app_config, "session": "active"} yield client # Cleanup client["session"] = "closed" def test_api_client(api_client): """Test using api client fixture.""" assert api_client["session"] == "active" assert api_client["config"]["debug"] is True ``` ### Pattern 3: Parameterized Tests ```python # test_validation.py import pytest def is_valid_email(email: str) -> bool: """Check if email is valid.""" return "@" in email and "." in email.split("@")[1] @pytest.mark.parametrize("email,expected", [ ("user@example.com", True), ("test.user@domain.co.uk", True), ("invalid.email", False), ("@example.com", False), ("user@domain", False), ("", False), ]) def test_email_validation(email, expected): """Test email validation with various inputs.""" assert is_valid_email(email) == expected @pytest.mark.parametrize("a,b,expected", [ (2, 3, 5), (0, 0, 0), (-1, 1, 0), (100, 200, 300), (-5, -5, -10), ]) def test_addition_parameterized(a, b, expected): """Test addition with multiple parameter sets.""" from test_calculator import Calculator calc = Calculator() assert calc.add(a, b) == expected # Using pytest.param for special cases @pytest.mark.parametrize("value,expected", [ pytest.param(1, True, id="positive"), pytest.param(0, False, id="zero"), pytest.param(-1, False, id="negative"), ]) def test_is_positive(value, expected): """Test with custom test IDs.""" assert (value > 0) == expected ``` ### Pattern 4: Mocking with unittest.mock ```python # test_api_client.py import pytest from unittest.mock import Mock, patch, MagicMock import requests class APIClient: """Simple API client.""" def __init__(self, base_url: str): self.base_url = base_url def get_user(self, user_id: int) -> dict: """Fetch user from API.""" response = requests.get(f"{self.base_url}/users/{user_id}") response.raise_for_status() return response.json() def create_user(self, data: dict) -> dict: """Create new user.""" response = requests.post(f"{self.base_url}/users", json=data) response.raise_for_status() return response.json() def test_get_user_success(): """Test successful API call with mock.""" client = APIClient("https://api.example.com") mock_response = Mock() mock_response.json.return_value = {"id": 1, "name": "John Doe"} mock_response.raise_for_status.return_value = None with patch("requests.get", return_value=mock_response) as mock_get: user = client.get_user(1) assert user["id"] == 1 assert user["name"] == "John Doe" mock_get.assert_called_once_with("https://api.example.com/users/1") def test_get_user_not_found(): """Test API call with 404 error.""" client = APIClient("https://api.example.com") mock_response = Mock() mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") with patch("requests.get", return_value=mock_response): with pytest.raises(requests.HTTPError): client.get_user(999) @patch("requests.post") def test_create_user(mock_post): """Test user creation with decorator syntax.""" client = APIClient("https://api.example.com") mock_post.return_value.json.return_value = {"id": 2, "name": "Jane Doe"} mock_post.return_value.raise_for_status.return_value = None user_data = {"name": "Jane Doe", "email": "jane@example.com"} result = client.create_user(user_data) assert result["id"] == 2 mock_post.assert_called_once() call_args = mock_post.call_args assert call_args.kwargs["json"] == user_data ``` ### Pattern 5: Testing Exceptions ```python # test_exceptions.py import pytest def divide(a: float, b: float) -> float: """Divide a by b.""" if b == 0: raise ZeroDivisionError("Division by zero") if not isinstance(a, (int, float)) or not isinstance(b, (int, float)): raise TypeError("Arguments must be numbers") return a / b def test_zero_division(): """Test exception is raised for division by zero.""" with pytest.raises(ZeroDivisionError): divide(10, 0) def test_zero_division_with_message(): """Test exception message.""" with pytest.raises(ZeroDivisionError, match="Division by zero"): divide(5, 0) def test_type_error(): """Test type error exception.""" with pytest.raises(TypeError, match="must be numbers"): divide("10", 5) def test_exception_info(): """Test accessing exception info.""" with pytest.raises(ValueError) as exc_info: int("not a number") assert "invalid literal" in str(exc_info.value) ``` ## Advanced Patterns ### Pattern 6: Testing Async Code ```python # test_async.py import pytest import asyncio async def fetch_data(url: str) -> dict: """Fetch data asynchronously.""" await asyncio.sleep(0.1) return {"url": url, "data": "result"} @pytest.mark.asyncio async def test_fetch_data(): """Test async function.""" result = await fetch_data("https://api.example.com") assert result["url"] == "https://api.example.com" assert "data" in result @pytest.mark.asyncio async def test_concurrent_fetches(): """Test concurrent async operations.""" urls = ["url1", "url2", "url3"] tasks = [fetch_data(url) for url in urls] results = await asyncio.gather(*tasks) assert len(results) == 3 assert all("data" in r for r in results) @pytest.fixture async def async_client(): """Async fixture.""" client = {"connected": True} yield client client["connected"] = False @pytest.mark.asyncio async def test_with_async_fixture(async_client): """Test using async fixture.""" assert async_client["connected"] is True ``` ### Pattern 7: Monkeypatch for Testing ```python # test_environment.py import os import pytest def get_database_url() -> str: """Get database URL from environment.""" return os.environ.get("DATABASE_URL", "sqlite:///:memory:") def test_database_url_default(): """Test default database URL.""" # Will use actual environment variable if set url = get_database_url() assert url def test_database_url_custom(monkeypatch): """Test custom database URL with monkeypatch.""" monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test") assert get_database_url() == "postgresql://localhost/test" def test_database_url_not_set(monkeypatch): """Test when env var is not set.""" monkeypatch.delenv("DATABASE_URL", raising=False) assert get_database_url() == "sqlite:///:memory:" class Config: """Configuration class.""" def __init__(self): self.api_key = "production-key" def get_api_key(self): return self.api_key def test_monkeypatch_attribute(monkeypatch): """Test monkeypatching object attributes.""" config = Config() monkeypatch.setattr(config, "api_key", "test-key") assert config.get_api_key() == "test-key" ``` ### Pattern 8: Temporary Files and Directories ```python # test_file_operations.py import pytest from pathlib import Path def save_data(filepath: Path, data: str): """Save data to file.""" filepath.write_text(data) def load_data(filepath: Path) -> str: """Load data from file.""" return filepath.read_text() def test_file_operations(tmp_path): """Test file operations with temporary directory.""" # tmp_path is a pathlib.Path object test_file = tmp_path / "test_data.txt" # Save data save_data(test_file, "Hello, World!") # Verify file exists assert test_file.exists() # Load and verify data data = load_data(test_file) assert data == "Hello, World!" def test_multiple_files(tmp_path): """Test with multiple temporary files.""" files = { "file1.txt": "Content 1", "file2.txt": "Content 2", "file3.txt": "Content 3" } for filename, content in files.items(): filepath = tmp_path / filename save_data(filepath, content) # Verify all files created assert len(list(tmp_path.iterdir())) == 3 # Verify contents for filename, expected_content in files.items(): filepath = tmp_path / filename assert load_data(filepath) == expected_content ``` ### Pattern 9: Custom Fixtures and Conftest ```python # conftest.py """Shared fixtures for all tests.""" import pytest @pytest.fixture(scope="session") def database_url(): """Provide database URL for all tests.""" return "postgresql://localhost/test_db" @pytest.fixture(autouse=True) def reset_database(database_url): """Auto-use fixture that runs before each test.""" # Setup: Clear database print(f"Clearing database: {database_url}") yield # Teardown: Clean up print("Test completed") @pytest.fixture def sample_user(): """Provide sample user data.""" return { "id": 1, "name": "Test User", "email": "test@example.com" } @pytest.fixture def sample_users(): """Provide list of sample users.""" return [ {"id": 1, "name": "User 1"}, {"id": 2, "name": "User 2"}, {"id": 3, "name": "User 3"}, ] # Parametrized fixture @pytest.fixture(params=["sqlite", "postgresql", "mysql"]) def db_backend(request): """Fixture that runs tests with different database backends.""" return request.param def test_with_db_backend(db_backend): """This test will run 3 times with different backends.""" print(f"Testing with {db_backend}") assert db_backend in ["sqlite", "postgresql", "mysql"] ``` ### Pattern 10: Property-Based Testing ```python # test_properties.py from hypothesis import given, strategies as st import pytest def reverse_string(s: str) -> str: """Reverse a string.""" return s[::-1] @given(st.text()) def test_reverse_twice_is_original(s): """Property: reversing twice returns original.""" assert reverse_string(reverse_string(s)) == s @given(st.text()) def test_reverse_length(s): """Property: reversed string has same length.""" assert len(reverse_string(s)) == len(s) @given(st.integers(), st.integers()) def test_addition_commutative(a, b): """Property: addition is commutative.""" assert a + b == b + a @given(st.lists(st.integers())) def test_sorted_list_properties(lst): """Property: sorted list is ordered.""" sorted_lst = sorted(lst) # Same length assert len(sorted_lst) == len(lst) # All elements present assert set(sorted_lst) == set(lst) # Is ordered for i in range(len(sorted_lst) - 1): assert sorted_lst[i] <= sorted_lst[i + 1] ``` ## Test Design Principles ### One Behavior Per Test Each test should verify exactly one behavior. This makes failures easy to diagnose and tests easy to maintain. ```python # BAD - testing multiple behaviors def test_user_service(): user = service.create_user(data) assert user.id is not None assert user.email == data["email"] updated = service.update_user(user.id, {"name": "New"}) assert updated.name == "New" # GOOD - focused tests def test_create_user_assigns_id(): user = service.create_user(data) assert user.id is not None def test_create_user_stores_email(): user = service.create_user(data) assert user.email == data["email"] def test_update_user_changes_name(): user = service.create_user(data) updated = service.update_user(user.id, {"name": "New"}) assert updated.name == "New" ``` ### Test Error Paths Always test failure cases, not just happy paths. ```python def test_get_user_raises_not_found(): with pytest.raises(UserNotFoundError) as exc_info: service.get_user("nonexistent-id") assert "nonexistent-id" in str(exc_info.value) def test_create_user_rejects_invalid_email(): with pytest.raises(ValueError, match="Invalid email format"): service.create_user({"email": "not-an-email"}) ``` ## Testing Best Practices ### Test Organization ```python # tests/ # __init__.py # conftest.py # Shared fixtures # test_unit/ # Unit tests # test_models.py # test_utils.py # test_integration/ # Integration tests # test_api.py # test_database.py # test_e2e/ # End-to-end tests # test_workflows.py ``` ### Test Naming Convention A common pattern: `test_<unit>_<scenario>_<expected_outcome>`. Adapt to your team's preferences. ```python # Pattern: test_<unit>_<scenario>_<expected> def test_create_user_with_valid_data_returns_user(): ... def test_create_user_with_duplicate_email_raises_conflict(): ... def test_get_user_with_unknown_id_returns_none(): ... # Good test names - clear and descriptive def test_user_creation_with_valid_data(): """Clear name describes what is being tested.""" pass def test_login_fails_with_invalid_password(): """Name describes expected behavior.""" pass def test_api_returns_404_for_missing_resource(): """Specific about inputs and expected outcomes.""" pass # Bad test names - avoid these def test_1(): # Not descriptive pass def test_user(): # Too vague pass def test_function(): # Doesn't explain what's tested pass ``` ### Testing Retry Behavior Verify that retry logic works correctly using mock side effects. ```python from unittest.mock import Mock def test_retries_on_transient_error(): """Test that service retries on transient failures.""" client = Mock() # Fail twice, then succeed client.request.side_effect = [ ConnectionError("Failed"), ConnectionError("Failed"), {"status": "ok"}, ] service = ServiceWithRetry(client, max_retries=3) result = service.fetch() assert result == {"status": "ok"} assert client.request.call_count == 3 def test_gives_up_after_max_retries(): """Test that service stops retrying after max attempts.""" client = Mock() client.request.side_effect = ConnectionError("Failed") service = ServiceWithRetry(client, max_retries=3) with pytest.raises(ConnectionError): service.fetch() assert client.request.call_count == 3 def test_does_not_retry_on_permanent_error(): """Test that permanent errors are not retried.""" client = Mock() client.request.side_effect = ValueError("Invalid input") service = ServiceWithRetry(client, max_retries=3) with pytest.raises(ValueError): service.fetch() # Only called once - no retry for ValueError assert client.request.call_count == 1 ``` ### Mocking Time with Freezegun Use freezegun
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

competitive-landscape

This skill should be used when the user asks to "analyze

business
⭐1
# Competitive Landscape Analysis Comprehensive frameworks for analyzing competition, identifying differentiation opportunities, and developing winning market positioning strategies. ## Overview Understand competitive dynamics using proven frameworks (Porter's Five Forces, Blue Ocean Strategy, positioning maps) to identify opportunities and craft defensible competitive advantages. ## Porter's Five Forces Analyze industry attractiveness and competitive intensity. ### Force 1: Threat of New Entrants **Barriers to Entry:** - Capital requirements - Economies of scale - Switching costs - Brand loyalty - Regulatory barriers - Access to distribution - Network effects **High Threat:** Low barriers, easy to enter (e.g., simple SaaS tools) **Low Threat:** High barriers (e.g., regulated industries, hardware) **Analysis Questions:** - How easy is it for new competitors to enter? - What would it cost to launch a competing product? - Are there network effects or switching costs protecting incumbents? ### Force 2: Bargaining Power of Suppliers **Supplier Power Factors:** - Supplier concentration - Availability of substitutes - Importance to supplier - Switching costs - Forward integration threat **High Power:** Few suppliers, critical inputs (e.g., cloud infrastructure providers) **Low Power:** Many alternatives, commoditized (e.g., generic services) **Analysis Questions:** - Who are our critical suppliers? - Could they raise prices or reduce quality? - Can we switch suppliers easily? ### Force 3: Bargaining Power of Buyers **Buyer Power Factors:** - Buyer concentration - Volume purchased - Product differentiation - Price sensitivity - Backward integration threat **High Power:** Few large customers, standardized products (e.g., enterprise deals) **Low Power:** Many small customers, differentiated product (e.g., consumer subscriptions) **Analysis Questions:** - Can customers easily switch to competitors? - Do few customers generate most revenue? - How price-sensitive are buyers? ### Force 4: Threat of Substitutes **Substitute Considerations:** - Alternative solutions - Price-performance tradeoff - Switching costs - Buyer propensity to substitute **High Threat:** Many alternatives, low switching cost (e.g., productivity software) **Low Threat:** Unique solution, high switching cost (e.g., ERP systems) **Analysis Questions:** - What alternative ways can customers solve this problem? - How do substitutes compare on price and performance? - What's the cost to switch to a substitute? ### Force 5: Competitive Rivalry **Rivalry Intensity Factors:** - Number of competitors - Industry growth rate - Product differentiation - Exit barriers - Strategic stakes **High Rivalry:** Many competitors, slow growth, commoditized (e.g., email marketing) **Low Rivalry:** Few competitors, fast growth, differentiated (e.g., emerging AI tools) **Analysis Questions:** - How many direct competitors exist? - Is the market growing or stagnant? - How differentiated are offerings? - Are competitors competing on price or value? ### Forces Analysis Summary Create a scorecard: | Force | Intensity (1-5) | Impact | Key Factors | | -------------- | --------------- | ------ | --------------------------------- | | New Entrants | 3 | Medium | Low barriers but network effects | | Supplier Power | 2 | Low | Many cloud providers | | Buyer Power | 4 | High | Enterprise customers concentrated | | Substitutes | 3 | Medium | Manual processes alternative | | Rivalry | 4 | High | 10+ direct competitors | **Overall Assessment:** Moderate industry attractiveness with high rivalry and buyer power ## Blue Ocean Strategy Identify uncontested market space through value innovation. ### Four Actions Framework **Eliminate:** What factors can be eliminated that the industry takes for granted? **Reduce:** What factors can be reduced well below industry standard? **Raise:** What factors can be raised well above industry standard? **Create:** What factors can be created that the industry never offered? ### Strategy Canvas Map your offering vs. competitors on key factors. **Example: Budget Hotels** ``` High | β˜… Traditional Hotels | β˜… Budget Hotels (new) | Low |___________________________________ Price Luxury Convenience Cleanliness Budget Hotel Strategy: - Eliminate: Luxury amenities, room service - Reduce: Lobby size, staff - Raise: Cleanliness, online booking - Create: Self-service kiosks, mobile app ``` ### Value Innovation Find the sweet spot: Lower cost + higher value **Steps:** 1. Map industry competing factors 2. Identify factors to eliminate/reduce (cost savings) 3. Identify factors to raise/create (differentiation) 4. Validate that combination creates new market space ## Competitive Positioning ### Positioning Map Plot competitors on 2-3 key dimensions. **Example Dimensions:** - Price vs. Features - Complexity vs. Ease of Use - Enterprise vs. SMB Focus - Self-Service vs. High-Touch - Generalist vs. Specialist **How to Create:** 1. Choose 2 dimensions most important to customers 2. Plot all competitors 3. Identify gaps (white space) 4. Validate gap represents real customer need **Example:** ``` High Price | | β˜… Enterprise A β˜… Enterprise B | | ● Our Position (gap) | | β˜… Competitor C β˜… Competitor D | Low Price |____________________________________________ Simple Complex ``` ### Differentiation Strategy **How to Differentiate:** 1. **Product Differentiation** - Unique features - Superior performance - Better design/UX - Integration ecosystem 2. **Service Differentiation** - Customer support quality - Onboarding experience - Response time - Success programs 3. **Brand Differentiation** - Trust and reputation - Thought leadership - Community - Values alignment 4. **Price Differentiation** - Premium positioning - Value positioning - Transparent pricing - Flexible packaging ### Positioning Statement Framework ``` For [target customer] Who [statement of need or opportunity] Our product is [product category] That [statement of key benefit] Unlike [primary competitive alternative] Our product [statement of primary differentiation] ``` **Example:** ``` For e-commerce companies Who struggle with email marketing automation Our product is an AI-powered email platform That increases conversion rates by 40% Unlike Klaviyo and Mailchimp Our product uses AI to personalize at scale ``` ## Competitive Intelligence ### Information Gathering **Public Sources:** - Company websites and blogs - Press releases and news - Job postings (hint at strategy) - Customer reviews (G2, Capterra) - Social media and forums - Glassdoor (employee insights) - SEC filings (public companies) - Patent filings **Direct Research:** - Customer interviews - Win/loss analysis - Sales team feedback - Product demos and trials - Conference attendance ### Competitor Profile Template For each key competitor, document: **Company Overview:** - Founded, HQ, funding, size - Leadership team - Company stage and trajectory **Product:** - Core features - Target customers - Pricing and packaging - Technology stack - Recent launches **Go-to-Market:** - Sales model (self-serve, sales-led) - Marketing strategy - Distribution channels - Partnerships **Strengths:** - What they do better than anyone - Key competitive advantages - Market position **Weaknesses:** - Gaps in product - Customer complaints - Operational challenges **Strategy:** - Stated direction - Inferred priorities - Likely next moves ## Competitive Pricing Analysis ### Price Positioning **Premium (Top 25%):** - Superior product/service - Strong brand - High-touch sales - Enterprise focus **Mid-Market (Middle 50%):** - Balanced value - Standard features - Mixed sales model - Broad market **Value (Bottom 25%):** - Basic functionality - Self-service - Cost leadership - High volume, low margin ### Pricing Comparison Matrix | Competitor | Entry Price | Mid Tier | Enterprise | Model | | ------------ | ----------- | -------- | ---------- | ------------ | | Competitor A | $29/mo | $99/mo | Custom | Subscription | | Competitor B | $49/mo | $199/mo | $499/mo | Subscription | | Us | $39/mo | $129/mo | Custom | Subscription | **Analysis:** - Are we priced competitively? - What does our pricing signal? - Are there gaps in our packaging? ## Go-to-Market Strategy ### Market Entry Strategies **Direct Competition:** - Head-to-head against established players - Requires differentiation and resources - Example: Better features at lower price **Niche Focus:** - Target underserved segment - Become specialist vs. generalist - Example: "Salesforce for real estate" **Disruptive Innovation:** - Target non-consumers or low end - Improve over time to move upmarket - Example: Freemium model disrupting enterprise **Platform Play:** - Build ecosystem and network effects - Aggregate complementary services - Example: Marketplace or API platform ### Beachhead Market **Characteristics of Good Beachhead:** - Specific, reachable segment - Acute pain you solve well - Limited competition - Willing to pay - Can lead to expansion **Example:** Instead of "project management software", target "project management for construction teams" ## Competitive Advantage ### Sustainable Advantages **Network Effects:** - Value increases with users - Example: Slack, marketplaces **Switching Costs:** - High cost to change - Example: CRM systems with data **Economies of Scale:** - Unit costs decrease with volume - Example: Cloud infrastructure **Brand:** - Trust and reputation - Example: Security software **Proprietary Technology:** - Patents or trade secrets - Example: Algorithms, data **Regulatory:** - Licenses or approvals - Example: Fintech, healthcare ### Testing Your Advantage Ask: - Can competitors copy this in < 2 years? - Does this matter to customers? - Do we execute this better than anyone? - Is this advantage durable? If "no" to any, it's not a sustainable advantage. ## Competitive Monitoring ### What to Track **Product Changes:** - New features - Pricing changes - Packaging adjustments **Market Signals:** - Funding announcements - Key hires (especially leadership) - Customer wins/losses - Partnerships **Performance Metrics:** - Revenue (if public or disclosed) - Customer count - Growth rate - Market share estimates ### Monitoring Cadence **Weekly:** - Product release notes - News mentions **Monthly:** - Win/loss analysis review - Positioning map updates **Quarterly:** - Deep competitive review - Strategy adjustment **Annually:** - Major strategy reassessment - Market trends analysis ## Additional Resources ### Reference Files - **`references/frameworks-deep-dive.md`** - Detailed application of each framework with worksheets - **`references/intel-sources.md`** - Comprehensive list of competitive intelligence sources ### Example Files - **`examples/competitor-analysis.md`** - Complete competitive analysis for a SaaS startup - **`examples/positioning-workshop.md`** - Step-by-step positioning development process ## Quick Start To analyze competitive landscape: 1. **Identify competitors** - Direct, indirect, and future threats 2. **Apply Porter's Five Forces** - Assess industry attractiveness 3. **Create positioning map** - Visualize competitive space 4. **Profile top 3-5 competitors** - Deep dive on key rivals 5. **Identify differentiation** - What makes you unique 6. **Analyze pricing** - Where do you fit? 7. **Assess advantages** - What's defensible? 8. **Develop strategy** - How to win For detailed frameworks and examples, see `references/` and `examples/`.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered