Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

19 NODES
πŸ€–system promptβ€’7 months ago

deployment-pipeline-design

Design multi-stage CI/CD pipelines with approval gates, security

coding
⭐1
# Deployment Pipeline Design Architecture patterns for multi-stage CI/CD pipelines with approval gates and deployment strategies. ## Purpose Design robust, secure deployment pipelines that balance speed with safety through proper stage organization and approval workflows. ## When to Use - Design CI/CD architecture - Implement deployment gates - Configure multi-environment pipelines - Establish deployment best practices - Implement progressive delivery ## Pipeline Stages ### Standard Pipeline Flow ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Build β”‚ β†’ β”‚ Test β”‚ β†’ β”‚ Staging β”‚ β†’ β”‚ Approveβ”‚ β†’ β”‚Productionβ”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### Detailed Stage Breakdown 1. **Source** - Code checkout 2. **Build** - Compile, package, containerize 3. **Test** - Unit, integration, security scans 4. **Staging Deploy** - Deploy to staging environment 5. **Integration Tests** - E2E, smoke tests 6. **Approval Gate** - Manual approval required 7. **Production Deploy** - Canary, blue-green, rolling 8. **Verification** - Health checks, monitoring 9. **Rollback** - Automated rollback on failure ## Approval Gate Patterns ### Pattern 1: Manual Approval ```yaml # GitHub Actions production-deploy: needs: staging-deploy environment: name: production url: https://app.example.com runs-on: ubuntu-latest steps: - name: Deploy to production run: | # Deployment commands ``` ### Pattern 2: Time-Based Approval ```yaml # GitLab CI deploy:production: stage: deploy script: - deploy.sh production environment: name: production when: delayed start_in: 30 minutes only: - main ``` ### Pattern 3: Multi-Approver ```yaml # Azure Pipelines stages: - stage: Production dependsOn: Staging jobs: - deployment: Deploy environment: name: production resourceType: Kubernetes strategy: runOnce: preDeploy: steps: - task: ManualValidation@0 inputs: notifyUsers: "team-leads@example.com" instructions: "Review staging metrics before approving" ``` **Reference:** See `assets/approval-gate-template.yml` ## Deployment Strategies ### 1. Rolling Deployment ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: my-app spec: replicas: 10 strategy: type: RollingUpdate rollingUpdate: maxSurge: 2 maxUnavailable: 1 ``` **Characteristics:** - Gradual rollout - Zero downtime - Easy rollback - Best for most applications ### 2. Blue-Green Deployment ```yaml # Blue (current) kubectl apply -f blue-deployment.yaml kubectl label service my-app version=blue # Green (new) kubectl apply -f green-deployment.yaml # Test green environment kubectl label service my-app version=green # Rollback if needed kubectl label service my-app version=blue ``` **Characteristics:** - Instant switchover - Easy rollback - Doubles infrastructure cost temporarily - Good for high-risk deployments ### 3. Canary Deployment ```yaml apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: my-app spec: replicas: 10 strategy: canary: steps: - setWeight: 10 - pause: { duration: 5m } - setWeight: 25 - pause: { duration: 5m } - setWeight: 50 - pause: { duration: 5m } - setWeight: 100 ``` **Characteristics:** - Gradual traffic shift - Risk mitigation - Real user testing - Requires service mesh or similar ### 4. Feature Flags ```python from flagsmith import Flagsmith flagsmith = Flagsmith(environment_key="API_KEY") if flagsmith.has_feature("new_checkout_flow"): # New code path process_checkout_v2() else: # Existing code path process_checkout_v1() ``` **Characteristics:** - Deploy without releasing - A/B testing - Instant rollback - Granular control ## Pipeline Orchestration ### Multi-Stage Pipeline Example ```yaml name: Production Pipeline on: push: branches: [main] jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Build application run: make build - name: Build Docker image run: docker build -t myapp:${{ github.sha }} . - name: Push to registry run: docker push myapp:${{ github.sha }} test: needs: build runs-on: ubuntu-latest steps: - name: Unit tests run: make test - name: Security scan run: trivy image myapp:${{ github.sha }} deploy-staging: needs: test runs-on: ubuntu-latest environment: name: staging steps: - name: Deploy to staging run: kubectl apply -f k8s/staging/ integration-test: needs: deploy-staging runs-on: ubuntu-latest steps: - name: Run E2E tests run: npm run test:e2e deploy-production: needs: integration-test runs-on: ubuntu-latest environment: name: production steps: - name: Canary deployment run: | kubectl apply -f k8s/production/ kubectl argo rollouts promote my-app verify: needs: deploy-production runs-on: ubuntu-latest steps: - name: Health check run: curl -f https://app.example.com/health - name: Notify team run: | curl -X POST ${{ secrets.SLACK_WEBHOOK }} \ -d '{"text":"Production deployment successful!"}' ``` ## Pipeline Best Practices 1. **Fail fast** - Run quick tests first 2. **Parallel execution** - Run independent jobs concurrently 3. **Caching** - Cache dependencies between runs 4. **Artifact management** - Store build artifacts 5. **Environment parity** - Keep environments consistent 6. **Secrets management** - Use secret stores (Vault, etc.) 7. **Deployment windows** - Schedule deployments appropriately 8. **Monitoring integration** - Track deployment metrics 9. **Rollback automation** - Auto-rollback on failures 10. **Documentation** - Document pipeline stages ## Rollback Strategies ### Automated Rollback ```yaml deploy-and-verify: steps: - name: Deploy new version run: kubectl apply -f k8s/ - name: Wait for rollout run: kubectl rollout status deployment/my-app - name: Health check id: health run: | for i in {1..10}; do if curl -sf https://app.example.com/health; then exit 0 fi sleep 10 done exit 1 - name: Rollback on failure if: failure() run: kubectl rollout undo deployment/my-app ``` ### Manual Rollback ```bash # List revision history kubectl rollout history deployment/my-app # Rollback to previous version kubectl rollout undo deployment/my-app # Rollback to specific revision kubectl rollout undo deployment/my-app --to-revision=3 ``` ## Monitoring and Metrics ### Key Pipeline Metrics - **Deployment Frequency** - How often deployments occur - **Lead Time** - Time from commit to production - **Change Failure Rate** - Percentage of failed deployments - **Mean Time to Recovery (MTTR)** - Time to recover from failure - **Pipeline Success Rate** - Percentage of successful runs - **Average Pipeline Duration** - Time to complete pipeline ### Integration with Monitoring ```yaml - name: Post-deployment verification run: | # Wait for metrics stabilization sleep 60 # Check error rate ERROR_RATE=$(curl -s "$PROMETHEUS_URL/api/v1/query?query=rate(http_errors_total[5m])" | jq '.data.result[0].value[1]') if (( $(echo "$ERROR_RATE > 0.01" | bc -l) )); then echo "Error rate too high: $ERROR_RATE" exit 1 fi ``` ## Reference Files - `references/pipeline-orchestration.md` - Complex pipeline patterns - `assets/approval-gate-template.yml` - Approval workflow templates ## Related Skills - `github-actions-templates` - For GitHub Actions implementation - `gitlab-ci-patterns` - For GitLab CI implementation - `secrets-management` - For secrets handling
πŸ‘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

postgresql-table-design

Design a PostgreSQL-specific schema. Covers best-practices, data

data
⭐1
# PostgreSQL Table Design ## Core Rules - Define a **PRIMARY KEY** for reference tables (users, orders, etc.). Not always needed for time-series/event/log data. When used, prefer `BIGINT GENERATED ALWAYS AS IDENTITY`; use `UUID` only when global uniqueness/opacity is needed. - **Normalize first (to 3NF)** to eliminate data redundancy and update anomalies; denormalize **only** for measured, high-ROI reads where join performance is proven problematic. Premature denormalization creates maintenance burden. - Add **NOT NULL** everywhere it’s semantically required; use **DEFAULT**s for common values. - Create **indexes for access paths you actually query**: PK/unique (auto), **FK columns (manual!)**, frequent filters/sorts, and join keys. - Prefer **TIMESTAMPTZ** for event time; **NUMERIC** for money; **TEXT** for strings; **BIGINT** for integer values, **DOUBLE PRECISION** for floats (or `NUMERIC` for exact decimal arithmetic). ## PostgreSQL β€œGotchas” - **Identifiers**: unquoted β†’ lowercased. Avoid quoted/mixed-case names. Convention: use `snake_case` for table/column names. - **Unique + NULLs**: UNIQUE allows multiple NULLs. Use `UNIQUE (...) NULLS NOT DISTINCT` (PG15+) to restrict to one NULL. - **FK indexes**: PostgreSQL **does not** auto-index FK columns. Add them. - **No silent coercions**: length/precision overflows error out (no truncation). Example: inserting 999 into `NUMERIC(2,0)` fails with error, unlike some databases that silently truncate or round. - **Sequences/identity have gaps** (normal; don't "fix"). Rollbacks, crashes, and concurrent transactions create gaps in ID sequences (1, 2, 5, 6...). This is expected behaviorβ€”don't try to make IDs consecutive. - **Heap storage**: no clustered PK by default (unlike SQL Server/MySQL InnoDB); `CLUSTER` is one-off reorganization, not maintained on subsequent inserts. Row order on disk is insertion order unless explicitly clustered. - **MVCC**: updates/deletes leave dead tuples; vacuum handles themβ€”design to avoid hot wide-row churn. ## Data Types - **IDs**: `BIGINT GENERATED ALWAYS AS IDENTITY` preferred (`GENERATED BY DEFAULT` also fine); `UUID` when merging/federating/used in a distributed system or for opaque IDs. Generate with `uuidv7()` (preferred if using PG18+) or `gen_random_uuid()` (if using an older PG version). - **Integers**: prefer `BIGINT` unless storage space is critical; `INTEGER` for smaller ranges; avoid `SMALLINT` unless constrained. - **Floats**: prefer `DOUBLE PRECISION` over `REAL` unless storage space is critical. Use `NUMERIC` for exact decimal arithmetic. - **Strings**: prefer `TEXT`; if length limits needed, use `CHECK (LENGTH(col) <= n)` instead of `VARCHAR(n)`; avoid `CHAR(n)`. Use `BYTEA` for binary data. Large strings/binary (>2KB default threshold) automatically stored in TOAST with compression. TOAST storage: `PLAIN` (no TOAST), `EXTENDED` (compress + out-of-line), `EXTERNAL` (out-of-line, no compress), `MAIN` (compress, keep in-line if possible). Default `EXTENDED` usually optimal. Control with `ALTER TABLE tbl ALTER COLUMN col SET STORAGE strategy` and `ALTER TABLE tbl SET (toast_tuple_target = 4096)` for threshold. Case-insensitive: for locale/accent handling use non-deterministic collations; for plain ASCII use expression indexes on `LOWER(col)` (preferred unless column needs case-insensitive PK/FK/UNIQUE) or `CITEXT`. - **Money**: `NUMERIC(p,s)` (never float). - **Time**: `TIMESTAMPTZ` for timestamps; `DATE` for date-only; `INTERVAL` for durations. Avoid `TIMESTAMP` (without timezone). Use `now()` for transaction start time, `clock_timestamp()` for current wall-clock time. - **Booleans**: `BOOLEAN` with `NOT NULL` constraint unless tri-state values are required. - **Enums**: `CREATE TYPE ... AS ENUM` for small, stable sets (e.g. US states, days of week). For business-logic-driven and evolving values (e.g. order statuses) β†’ use TEXT (or INT) + CHECK or lookup table. - **Arrays**: `TEXT[]`, `INTEGER[]`, etc. Use for ordered lists where you query elements. Index with **GIN** for containment (`@>`, `<@`) and overlap (`&&`) queries. Access: `arr[1]` (1-indexed), `arr[1:3]` (slicing). Good for tags, categories; avoid for relationsβ€”use junction tables instead. Literal syntax: `'{val1,val2}'` or `ARRAY[val1,val2]`. - **Range types**: `daterange`, `numrange`, `tstzrange` for intervals. Support overlap (`&&`), containment (`@>`), operators. Index with **GiST**. Good for scheduling, versioning, numeric ranges. Pick a bounds scheme and use it consistently; prefer `[)` (inclusive/exclusive) by default. - **Network types**: `INET` for IP addresses, `CIDR` for network ranges, `MACADDR` for MAC addresses. Support network operators (`<<`, `>>`, `&&`). - **Geometric types**: `POINT`, `LINE`, `POLYGON`, `CIRCLE` for 2D spatial data. Index with **GiST**. Consider **PostGIS** for advanced spatial features. - **Text search**: `TSVECTOR` for full-text search documents, `TSQUERY` for search queries. Index `tsvector` with **GIN**. Always specify language: `to_tsvector('english', col)` and `to_tsquery('english', 'query')`. Never use single-argument versions. This applies to both index expressions and queries. - **Domain types**: `CREATE DOMAIN email AS TEXT CHECK (VALUE ~ '^[^@]+@[^@]+$')` for reusable custom types with validation. Enforces constraints across tables. - **Composite types**: `CREATE TYPE address AS (street TEXT, city TEXT, zip TEXT)` for structured data within columns. Access with `(col).field` syntax. - **JSONB**: preferred over JSON; index with **GIN**. Use only for optional/semi-structured attrs. ONLY use JSON if the original ordering of the contents MUST be preserved. - **Vector types**: `vector` type by `pgvector` for vector similarity search for embeddings. ### Do not use the following data types - DO NOT use `timestamp` (without time zone); DO use `timestamptz` instead. - DO NOT use `char(n)` or `varchar(n)`; DO use `text` instead. - DO NOT use `money` type; DO use `numeric` instead. - DO NOT use `timetz` type; DO use `timestamptz` instead. - DO NOT use `timestamptz(0)` or any other precision specification; DO use `timestamptz` instead - DO NOT use `serial` type; DO use `generated always as identity` instead. ## Table Types - **Regular**: default; fully durable, logged. - **TEMPORARY**: session-scoped, auto-dropped, not logged. Faster for scratch work. - **UNLOGGED**: persistent but not crash-safe. Faster writes; good for caches/staging. ## Row-Level Security Enable with `ALTER TABLE tbl ENABLE ROW LEVEL SECURITY`. Create policies: `CREATE POLICY user_access ON orders FOR SELECT TO app_users USING (user_id = current_user_id())`. Built-in user-based access control at the row level. ## Constraints - **PK**: implicit UNIQUE + NOT NULL; creates a B-tree index. - **FK**: specify `ON DELETE/UPDATE` action (`CASCADE`, `RESTRICT`, `SET NULL`, `SET DEFAULT`). Add explicit index on referencing columnβ€”speeds up joins and prevents locking issues on parent deletes/updates. Use `DEFERRABLE INITIALLY DEFERRED` for circular FK dependencies checked at transaction end. - **UNIQUE**: creates a B-tree index; allows multiple NULLs unless `NULLS NOT DISTINCT` (PG15+). Standard behavior: `(1, NULL)` and `(1, NULL)` are allowed. With `NULLS NOT DISTINCT`: only one `(1, NULL)` allowed. Prefer `NULLS NOT DISTINCT` unless you specifically need duplicate NULLs. - **CHECK**: row-local constraints; NULL values pass the check (three-valued logic). Example: `CHECK (price > 0)` allows NULL prices. Combine with `NOT NULL` to enforce: `price NUMERIC NOT NULL CHECK (price > 0)`. - **EXCLUDE**: prevents overlapping values using operators. `EXCLUDE USING gist (room_id WITH =, booking_period WITH &&)` prevents double-booking rooms. Requires appropriate index type (often GiST). ## Indexing - **B-tree**: default for equality/range queries (`=`, `<`, `>`, `BETWEEN`, `ORDER BY`) - **Composite**: order mattersβ€”index used if equality on leftmost prefix (`WHERE a = ? AND b > ?` uses index on `(a,b)`, but `WHERE b = ?` does not). Put most selective/frequently filtered columns first. - **Covering**: `CREATE INDEX ON tbl (id) INCLUDE (name, email)` - includes non-key columns for index-only scans without visiting table. - **Partial**: for hot subsets (`WHERE status = 'active'` β†’ `CREATE INDEX ON tbl (user_id) WHERE status = 'active'`). Any query with `status = 'active'` can use this index. - **Expression**: for computed search keys (`CREATE INDEX ON tbl (LOWER(email))`). Expression must match exactly in WHERE clause: `WHERE LOWER(email) = 'user@example.com'`. - **GIN**: JSONB containment/existence, arrays (`@>`, `?`), full-text search (`@@`) - **GiST**: ranges, geometry, exclusion constraints - **BRIN**: very large, naturally ordered data (time-series)β€”minimal storage overhead. Effective when row order on disk correlates with indexed column (insertion order or after `CLUSTER`). ## Partitioning - Use for very large tables (>100M rows) where queries consistently filter on partition key (often time/date). - Alternate use: use for tables where data maintenance tasks dictates e.g. data pruned or bulk replaced periodically - **RANGE**: common for time-series (`PARTITION BY RANGE (created_at)`). Create partitions: `CREATE TABLE logs_2024_01 PARTITION OF logs FOR VALUES FROM ('2024-01-01') TO ('2024-02-01')`. **TimescaleDB** automates time-based or ID-based partitioning with retention policies and compression. - **LIST**: for discrete values (`PARTITION BY LIST (region)`). Example: `FOR VALUES IN ('us-east', 'us-west')`. - **HASH**: for even distribution when no natural key (`PARTITION BY HASH (user_id)`). Creates N partitions with modulus. - **Constraint exclusion**: requires `CHECK` constraints on partitions for query planner to prune. Auto-created for declarative partitioning (PG10+). - Prefer declarative partitioning or hypertables. Do NOT use table inheritance. - **Limitations**: no global UNIQUE constraintsβ€”include partition key in PK/UNIQUE. FKs from partitioned tables not supported; use triggers. ## Special Considerations ### Update-Heavy Tables - **Separate hot/cold columns**β€”put frequently updated columns in separate table to minimize bloat. - **Use `fillfactor=90`** to leave space for HOT updates that avoid index maintenance. - **Avoid updating indexed columns**β€”prevents beneficial HOT updates. - **Partition by update patterns**β€”separate frequently updated rows in a different partition from stable data. ### Insert-Heavy Workloads - **Minimize indexes**β€”only create what you query; every index slows inserts. - **Use `COPY` or multi-row `INSERT`** instead of single-row inserts. - **UNLOGGED tables** for rebuildable staging dataβ€”much faster writes. - **Defer index creation** for bulk loadsβ€”>drop index, load data, recreate indexes. - **Partition by time/hash** to distribute load. **TimescaleDB** automates partitioning and compression of insert-heavy data. - **Use a natural key for primary key** such as a (timestamp, device_id) if enforcing global uniqueness is important many insert-heavy tables don't need a primary key at all. - If you do need a surrogate key, **Prefer `BIGINT GENERATED ALWAYS AS IDENTITY` over `UUID`**. ### Upsert-Friendly Design - **Requires UNIQUE index** on conflict target columnsβ€”`ON CONFLICT (col1, col2)` needs exact matching unique index (partial indexes don't work). - **Use `EXCLUDED.column`** to reference would-be-inserted values; only update columns that actually changed to reduce write overhead. - **`DO NOTHING` faster** than `DO UPDATE` when no actual update needed. ### Safe Schema Evolution - **Transactional DDL**: most DDL operations can run in transactions and be rolled backβ€”`BEGIN; ALTER TABLE...; ROLLBACK;` for safe testing. - **Concurrent index creation**: `CREATE INDEX CONCURRENTLY` avoids blocking writes but can't run in transactions. - **Volatile defaults cause rewrites**: adding `NOT NULL` columns with volatile defaults (e.g., `now()`, `gen_random_uuid()`) rewrites entire table. Non-volatile defaults are fast. - **Drop constraints before columns**: `ALTER TABLE DROP CONSTRAINT` then `DROP COLUMN` to avoid dependency issues. - **Function signature changes**: `CREATE OR REPLACE` with different arguments creates overloads, not replacements. DROP old version if no overload desired. ## Generated Columns - `... GENERATED ALWAYS AS (<expr>) STORED` for computed, indexable fields. PG18+ adds `VIRTUAL` columns (computed on read, not stored). ## Extensions - **`pgcrypto`**: `crypt()` for password hashing. - **`uuid-ossp`**: alternative UUID functions; prefer `pgcrypto` for new projects. - **`pg_trgm`**: fuzzy text search with `%` operator, `similarity()` function. Index with GIN for `LIKE '%pattern%'` acceleration. - **`citext`**: case-insensitive text type. Prefer expression indexes on `LOWER(col)` unless you need case-insensitive constraints. - **`btree_gin`/`btree_gist`**: enable mixed-type indexes (e.g., GIN index on both JSONB and text columns). - **`hstore`**: key-value pairs; mostly superseded by JSONB but useful for simple string mappings. - **`timescaledb`**: essential for time-seriesβ€”automated partitioning, retention, compression, continuous aggregates. - **`postgis`**: comprehensive geospatial support beyond basic geometric typesβ€”essential for location-based applications. - **`pgvector`**: vector similarity search for embeddings. - **`pgaudit`**: audit logging for all database activity. ## JSONB Guidance - Prefer `JSONB` with **GIN** index. - Default: `CREATE INDEX ON tbl USING GIN (jsonb_col);` β†’ accelerates: - **Containment** `jsonb_col @> '{"k":"v"}'` - **Key existence** `jsonb_col ? 'k'`, **any/all keys** `?\|`, `?&` - **Path containment** on nested docs - **Disjunction** `jsonb_col @> ANY(ARRAY['{"status":"active"}', '{"status":"pending"}'])` - Heavy `@>` workloads: consider opclass `jsonb_path_ops` for smaller/faster containment-only indexes: - `CREATE INDEX ON tbl USING GIN (jsonb_col jsonb_path_ops);` - **Trade-off**: loses support for key existence (`?`, `?|`, `?&`) queriesβ€”only supports containment (`@>`) - Equality/range on a specific scalar field: extract and index with B-tree (generated column or expression): - `ALTER TABLE tbl ADD COLUMN price INT GENERATED ALWAYS AS ((jsonb_col->>'price')::INT) STORED;` - `CREATE INDEX ON tbl (price);` - Prefer queries like `WHERE price BETWEEN 100 AND 500` (uses B-tree) over `WHERE (jsonb_col->>'price')::INT BETWEEN 100 AND 500` without index. - Arrays inside JSONB: use GIN + `@>` for containment (e.g., tags). Consider `jsonb_path_ops` if only doing containment. - Keep core relations in tables; use JSONB for optional/variable attributes. - Use constraints to limit allowed JSONB values in a column e.g. `config JSONB NOT NULL CHECK(jsonb_typeof(config) = 'object')` ## Examples ### Users ```sql CREATE TABLE users ( user_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, email TEXT NOT NULL UNIQUE, name TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE UNIQUE INDEX ON users (LOWER(email)); CREATE INDEX ON users (created_at); ``` ### Orders ```sql CREATE TABLE orders ( order_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, user_id BIGINT NOT NULL REFERENCES users(user_id), status TEXT NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING','PAID','CANCELED')), total NUMERIC(10,2) NOT NULL CHECK (total > 0), created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX ON orders (user_id); CREATE INDEX ON orders (created_at); ``` ### JSONB ```sql CREATE TABLE profiles ( user_id BIGINT PRIMARY KEY REFERENCES users(user_id), attrs JSONB NOT NULL DEFAULT '{}', theme TEXT GENERATED ALWAYS AS (attrs->>'theme') STORED ); CREATE INDEX profiles_attrs_gin ON profiles USING GIN (attrs); ```
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

unity-ecs-patterns

Master Unity ECS (Entity Component System) with DOTS, Jobs, and

coding
⭐1
# Unity ECS Patterns Production patterns for Unity's Data-Oriented Technology Stack (DOTS) including Entity Component System, Job System, and Burst Compiler. ## When to Use This Skill - Building high-performance Unity games - Managing thousands of entities efficiently - Implementing data-oriented game systems - Optimizing CPU-bound game logic - Converting OOP game code to ECS - Using Jobs and Burst for parallelization ## Core Concepts ### 1. ECS vs OOP | Aspect | Traditional OOP | ECS/DOTS | | ----------- | ----------------- | --------------- | | Data layout | Object-oriented | Data-oriented | | Memory | Scattered | Contiguous | | Processing | Per-object | Batched | | Scaling | Poor with count | Linear scaling | | Best for | Complex behaviors | Mass simulation | ### 2. DOTS Components ``` Entity: Lightweight ID (no data) Component: Pure data (no behavior) System: Logic that processes components World: Container for entities Archetype: Unique combination of components Chunk: Memory block for same-archetype entities ``` ## Patterns ### Pattern 1: Basic ECS Setup ```csharp using Unity.Entities; using Unity.Mathematics; using Unity.Transforms; using Unity.Burst; using Unity.Collections; // Component: Pure data, no methods public struct Speed : IComponentData { public float Value; } public struct Health : IComponentData { public float Current; public float Max; } public struct Target : IComponentData { public Entity Value; } // Tag component (zero-size marker) public struct EnemyTag : IComponentData { } public struct PlayerTag : IComponentData { } // Buffer component (variable-size array) [InternalBufferCapacity(8)] public struct InventoryItem : IBufferElementData { public int ItemId; public int Quantity; } // Shared component (grouped entities) public struct TeamId : ISharedComponentData { public int Value; } ``` ### Pattern 2: Systems with ISystem (Recommended) ```csharp using Unity.Entities; using Unity.Transforms; using Unity.Mathematics; using Unity.Burst; // ISystem: Unmanaged, Burst-compatible, highest performance [BurstCompile] public partial struct MovementSystem : ISystem { [BurstCompile] public void OnCreate(ref SystemState state) { // Require components before system runs state.RequireForUpdate<Speed>(); } [BurstCompile] public void OnUpdate(ref SystemState state) { float deltaTime = SystemAPI.Time.DeltaTime; // Simple foreach - auto-generates job foreach (var (transform, speed) in SystemAPI.Query<RefRW<LocalTransform>, RefRO<Speed>>()) { transform.ValueRW.Position += new float3(0, 0, speed.ValueRO.Value * deltaTime); } } [BurstCompile] public void OnDestroy(ref SystemState state) { } } // With explicit job for more control [BurstCompile] public partial struct MovementJobSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { var job = new MoveJob { DeltaTime = SystemAPI.Time.DeltaTime }; state.Dependency = job.ScheduleParallel(state.Dependency); } } [BurstCompile] public partial struct MoveJob : IJobEntity { public float DeltaTime; void Execute(ref LocalTransform transform, in Speed speed) { transform.Position += new float3(0, 0, speed.Value * DeltaTime); } } ``` ### Pattern 3: Entity Queries ```csharp [BurstCompile] public partial struct QueryExamplesSystem : ISystem { private EntityQuery _enemyQuery; public void OnCreate(ref SystemState state) { // Build query manually for complex cases _enemyQuery = new EntityQueryBuilder(Allocator.Temp) .WithAll<EnemyTag, Health, LocalTransform>() .WithNone<Dead>() .WithOptions(EntityQueryOptions.FilterWriteGroup) .Build(ref state); } [BurstCompile] public void OnUpdate(ref SystemState state) { // SystemAPI.Query - simplest approach foreach (var (health, entity) in SystemAPI.Query<RefRW<Health>>() .WithAll<EnemyTag>() .WithEntityAccess()) { if (health.ValueRO.Current <= 0) { // Mark for destruction SystemAPI.GetSingleton<EndSimulationEntityCommandBufferSystem.Singleton>() .CreateCommandBuffer(state.WorldUnmanaged) .DestroyEntity(entity); } } // Get count int enemyCount = _enemyQuery.CalculateEntityCount(); // Get all entities var enemies = _enemyQuery.ToEntityArray(Allocator.Temp); // Get component arrays var healths = _enemyQuery.ToComponentDataArray<Health>(Allocator.Temp); } } ``` ### Pattern 4: Entity Command Buffers (Structural Changes) ```csharp // Structural changes (create/destroy/add/remove) require command buffers [BurstCompile] [UpdateInGroup(typeof(SimulationSystemGroup))] public partial struct SpawnSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { var ecbSingleton = SystemAPI.GetSingleton<BeginSimulationEntityCommandBufferSystem.Singleton>(); var ecb = ecbSingleton.CreateCommandBuffer(state.WorldUnmanaged); foreach (var (spawner, transform) in SystemAPI.Query<RefRW<Spawner>, RefRO<LocalTransform>>()) { spawner.ValueRW.Timer -= SystemAPI.Time.DeltaTime; if (spawner.ValueRO.Timer <= 0) { spawner.ValueRW.Timer = spawner.ValueRO.Interval; // Create entity (deferred until sync point) Entity newEntity = ecb.Instantiate(spawner.ValueRO.Prefab); // Set component values ecb.SetComponent(newEntity, new LocalTransform { Position = transform.ValueRO.Position, Rotation = quaternion.identity, Scale = 1f }); // Add component ecb.AddComponent(newEntity, new Speed { Value = 5f }); } } } } // Parallel ECB usage [BurstCompile] public partial struct ParallelSpawnJob : IJobEntity { public EntityCommandBuffer.ParallelWriter ECB; void Execute([EntityIndexInQuery] int index, in Spawner spawner) { Entity e = ECB.Instantiate(index, spawner.Prefab); ECB.AddComponent(index, e, new Speed { Value = 5f }); } } ``` ### Pattern 5: Aspect (Grouping Components) ```csharp using Unity.Entities; using Unity.Transforms; using Unity.Mathematics; // Aspect: Groups related components for cleaner code public readonly partial struct CharacterAspect : IAspect { public readonly Entity Entity; private readonly RefRW<LocalTransform> _transform; private readonly RefRO<Speed> _speed; private readonly RefRW<Health> _health; // Optional component [Optional] private readonly RefRO<Shield> _shield; // Buffer private readonly DynamicBuffer<InventoryItem> _inventory; public float3 Position { get => _transform.ValueRO.Position; set => _transform.ValueRW.Position = value; } public float CurrentHealth => _health.ValueRO.Current; public float MaxHealth => _health.ValueRO.Max; public float MoveSpeed => _speed.ValueRO.Value; public bool HasShield => _shield.IsValid; public float ShieldAmount => HasShield ? _shield.ValueRO.Amount : 0f; public void TakeDamage(float amount) { float remaining = amount; if (HasShield && _shield.ValueRO.Amount > 0) { // Shield absorbs damage first remaining = math.max(0, amount - _shield.ValueRO.Amount); } _health.ValueRW.Current = math.max(0, _health.ValueRO.Current - remaining); } public void Move(float3 direction, float deltaTime) { _transform.ValueRW.Position += direction * _speed.ValueRO.Value * deltaTime; } public void AddItem(int itemId, int quantity) { _inventory.Add(new InventoryItem { ItemId = itemId, Quantity = quantity }); } } // Using aspect in system [BurstCompile] public partial struct CharacterSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { float dt = SystemAPI.Time.DeltaTime; foreach (var character in SystemAPI.Query<CharacterAspect>()) { character.Move(new float3(1, 0, 0), dt); if (character.CurrentHealth < character.MaxHealth * 0.5f) { // Low health logic } } } } ``` ### Pattern 6: Singleton Components ```csharp // Singleton: Exactly one entity with this component public struct GameConfig : IComponentData { public float DifficultyMultiplier; public int MaxEnemies; public float SpawnRate; } public struct GameState : IComponentData { public int Score; public int Wave; public float TimeRemaining; } // Create singleton on world creation public partial struct GameInitSystem : ISystem { public void OnCreate(ref SystemState state) { var entity = state.EntityManager.CreateEntity(); state.EntityManager.AddComponentData(entity, new GameConfig { DifficultyMultiplier = 1.0f, MaxEnemies = 100, SpawnRate = 2.0f }); state.EntityManager.AddComponentData(entity, new GameState { Score = 0, Wave = 1, TimeRemaining = 120f }); } } // Access singleton in system [BurstCompile] public partial struct ScoreSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { // Read singleton var config = SystemAPI.GetSingleton<GameConfig>(); // Write singleton ref var gameState = ref SystemAPI.GetSingletonRW<GameState>().ValueRW; gameState.TimeRemaining -= SystemAPI.Time.DeltaTime; // Check exists if (SystemAPI.HasSingleton<GameConfig>()) { // ... } } } ``` ### Pattern 7: Baking (Converting GameObjects) ```csharp using Unity.Entities; using UnityEngine; // Authoring component (MonoBehaviour in Editor) public class EnemyAuthoring : MonoBehaviour { public float Speed = 5f; public float Health = 100f; public GameObject ProjectilePrefab; class Baker : Baker<EnemyAuthoring> { public override void Bake(EnemyAuthoring authoring) { var entity = GetEntity(TransformUsageFlags.Dynamic); AddComponent(entity, new Speed { Value = authoring.Speed }); AddComponent(entity, new Health { Current = authoring.Health, Max = authoring.Health }); AddComponent(entity, new EnemyTag()); if (authoring.ProjectilePrefab != null) { AddComponent(entity, new ProjectilePrefab { Value = GetEntity(authoring.ProjectilePrefab, TransformUsageFlags.Dynamic) }); } } } } // Complex baking with dependencies public class SpawnerAuthoring : MonoBehaviour { public GameObject[] Prefabs; public float Interval = 1f; class Baker : Baker<SpawnerAuthoring> { public override void Bake(SpawnerAuthoring authoring) { var entity = GetEntity(TransformUsageFlags.Dynamic); AddComponent(entity, new Spawner { Interval = authoring.Interval, Timer = 0f }); // Bake buffer of prefabs var buffer = AddBuffer<SpawnPrefabElement>(entity); foreach (var prefab in authoring.Prefabs) { buffer.Add(new SpawnPrefabElement { Prefab = GetEntity(prefab, TransformUsageFlags.Dynamic) }); } // Declare dependencies DependsOn(authoring.Prefabs); } } } ``` ### Pattern 8: Jobs with Native Collections ```csharp using Unity.Jobs; using Unity.Collections; using Unity.Burst; using Unity.Mathematics; [BurstCompile] public struct SpatialHashJob : IJobParallelFor { [ReadOnly] public NativeArray<float3> Positions; // Thread-safe write to hash map public NativeParallelMultiHashMap<int, int>.ParallelWriter HashMap; public float CellSize; public void Execute(int index) { float3 pos = Positions[index]; int hash = GetHash(pos); HashMap.Add(hash, index); } int GetHash(float3 pos) { int x = (int)math.floor(pos.x / CellSize); int y = (int)math.floor(pos.y / CellSize); int z = (int)math.floor(pos.z / CellSize); return x * 73856093 ^ y * 19349663 ^ z * 83492791; } } [BurstCompile] public partial struct SpatialHashSystem : ISystem { private NativeParallelMultiHashMap<int, int> _hashMap; public void OnCreate(ref SystemState state) { _hashMap = new NativeParallelMultiHashMap<int, int>(10000, Allocator.Persistent); } public void OnDestroy(ref SystemState state) { _hashMap.Dispose(); } [BurstCompile] public void OnUpdate(ref SystemState state) { var query = SystemAPI.QueryBuilder() .WithAll<LocalTransform>() .Build(); int count = query.CalculateEntityCount(); // Resize if needed if (_hashMap.Capacity < count) { _hashMap.Capacity = count * 2; } _hashMap.Clear(); // Get positions var positions = query.ToComponentDataArray<LocalTransform>(Allocator.TempJob); var posFloat3 = new NativeArray<float3>(count, Allocator.TempJob); for (int i = 0; i < count; i++) { posFloat3[i] = positions[i].Position; } // Build hash map var hashJob = new SpatialHashJob { Positions = posFloat3, HashMap = _hashMap.AsParallelWriter(), CellSize = 10f }; state.Dependency = hashJob.Schedule(count, 64, state.Dependency); // Cleanup positions.Dispose(state.Dependency); posFloat3.Dispose(state.Dependency); } } ``` ## Performance Tips ```csharp // 1. Use Burst everywhere [BurstCompile] public partial struct MySystem : ISystem { } // 2. Prefer IJobEntity over manual iteration [BurstCompile] partial struct OptimizedJob : IJobEntity { void Execute(ref LocalTransform transform) { } } // 3. Schedule parallel when possible state.Dependency = job.ScheduleParallel(state.Dependency); // 4. Use ScheduleParallel with chunk iteration [BurstCompile] partial struct ChunkJob : IJobChunk { public ComponentTypeHandle<Health> HealthHandle; public void Execute(in ArchetypeChunk chunk, int unfilteredChunkIndex, bool useEnabledMask, in v128 chunkEnabledMask) { var healths = chunk.GetNativeArray(ref HealthHandle); for (int i = 0; i < chunk.Count; i++) { // Process } } } // 5. Avoid structural changes in hot paths // Use enableable components instead of add/remove public struct Disabled : IComponentData, IEnableableComponent { } ``` ## Best Practices ### Do's - **Use ISystem over SystemBase** - Better performance - **Burst compile everything** - Massive speedup - **Batch structural changes** - Use ECB - **Profile with Profiler** - Identify bottlenecks - **Use Aspects** - Clean component grouping ### Don'ts - **Don't use managed types** - Breaks Burst - **Don't structural change in jobs** - Use ECB - **Don't over-architect** - Start simple - **Don't ignore chunk utilization** - Group similar entities - **Don't forget disposal** - Native collections leak ## Resources - [Unity DOTS Documentation](https://docs.unity3d.com/Packages/com.unity.entities@latest) - [Unity DOTS Samples](https://github.com/Unity-Technologies/EntityComponentSystemSamples) - [Burst User Guide](https://docs.unity3d.com/Packages/com.unity.burst@latest)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

employment-contract-templates

Create employment contracts, offer letters, and HR policy documents

coding
⭐1
# Employment Contract Templates Templates and patterns for creating legally sound employment documentation including contracts, offer letters, and HR policies. ## When to Use This Skill - Drafting employment contracts - Creating offer letters - Writing employee handbooks - Developing HR policies - Standardizing employment documentation - Onboarding documentation ## Core Concepts ### 1. Employment Document Types | Document | Purpose | When Used | | ----------------------- | ----------------------- | ------------- | | **Offer Letter** | Initial job offer | Pre-hire | | **Employment Contract** | Formal agreement | Hire | | **Employee Handbook** | Policies & procedures | Onboarding | | **NDA** | Confidentiality | Before access | | **Non-Compete** | Competition restriction | Hire/Exit | ### 2. Key Legal Considerations ``` Employment Relationship: β”œβ”€β”€ At-Will vs. Contract β”œβ”€β”€ Employee vs. Contractor β”œβ”€β”€ Full-Time vs. Part-Time β”œβ”€β”€ Exempt vs. Non-Exempt └── Jurisdiction-Specific Requirements ``` **DISCLAIMER: These templates are for informational purposes only and do not constitute legal advice. Consult with qualified legal counsel before using any employment documents.** ## Templates ### Template 1: Offer Letter ```markdown # EMPLOYMENT OFFER LETTER [Company Letterhead] Date: [DATE] [Candidate Name] [Address] [City, State ZIP] Dear [Candidate Name], We are pleased to extend an offer of employment for the position of [JOB TITLE] at [COMPANY NAME]. We believe your skills and experience will be valuable additions to our team. ## Position Details **Title:** [Job Title] **Department:** [Department] **Reports To:** [Manager Name/Title] **Location:** [Office Location / Remote] **Start Date:** [Proposed Start Date] **Employment Type:** [Full-Time/Part-Time], [Exempt/Non-Exempt] ## Compensation **Base Salary:** $[AMOUNT] per [year/hour], paid [bi-weekly/semi-monthly/monthly] **Bonus:** [Eligible for annual bonus of up to X% based on company and individual performance / Not applicable] **Equity:** [X shares of stock options vesting over 4 years with 1-year cliff / Not applicable] ## Benefits You will be eligible for our standard benefits package, including: - Health insurance (medical, dental, vision) effective [date] - 401(k) with [X]% company match - [x] days paid time off per year - [x] paid holidays - [Other benefits] Full details will be provided during onboarding. ## Contingencies This offer is contingent upon: - Successful completion of background check - Verification of your right to work in [Country] - Execution of required employment documents including: - Confidentiality Agreement - [Non-Compete Agreement, if applicable] - [IP Assignment Agreement] ## At-Will Employment Please note that employment with [Company Name] is at-will. This means that either you or the Company may terminate the employment relationship at any time, with or without cause or notice. This offer letter does not constitute a contract of employment for any specific period. ## Acceptance To accept this offer, please sign below and return by [DEADLINE DATE]. This offer will expire if not accepted by that date. We are excited about the possibility of you joining our team. If you have any questions, please contact [HR Contact] at [email/phone]. Sincerely, --- [Hiring Manager Name] [Title] [Company Name] --- ## ACCEPTANCE I accept this offer of employment and agree to the terms stated above. Signature: ************\_************ Printed Name: ************\_************ Date: ************\_************ Anticipated Start Date: ************\_************ ``` ### Template 2: Employment Agreement (Contract Position) ```markdown # EMPLOYMENT AGREEMENT This Employment Agreement ("Agreement") is entered into as of [DATE] ("Effective Date") by and between: **Employer:** [COMPANY LEGAL NAME], a [State] [corporation/LLC] with principal offices at [Address] ("Company") **Employee:** [EMPLOYEE NAME], an individual residing at [Address] ("Employee") ## 1. EMPLOYMENT 1.1 **Position.** The Company agrees to employ Employee as [JOB TITLE], reporting to [Manager Title]. Employee accepts such employment subject to the terms of this Agreement. 1.2 **Duties.** Employee shall perform duties consistent with their position, including but not limited to: - [Primary duty 1] - [Primary duty 2] - [Primary duty 3] - Other duties as reasonably assigned 1.3 **Best Efforts.** Employee agrees to devote their full business time, attention, and best efforts to the Company's business during employment. 1.4 **Location.** Employee's primary work location shall be [Location/Remote]. [Travel requirements, if any.] ## 2. TERM 2.1 **Employment Period.** This Agreement shall commence on [START DATE] and continue until terminated as provided herein. 2.2 **At-Will Employment.** [FOR AT-WILL STATES] Notwithstanding anything herein, employment is at-will and may be terminated by either party at any time, with or without cause or notice. [OR FOR FIXED TERM:] 2.2 **Fixed Term.** This Agreement is for a fixed term of [X] months/years, ending on [END DATE], unless terminated earlier as provided herein or extended by mutual written agreement. ## 3. COMPENSATION 3.1 **Base Salary.** Employee shall receive a base salary of $[AMOUNT] per year, payable in accordance with the Company's standard payroll practices, subject to applicable withholdings. 3.2 **Bonus.** Employee may be eligible for an annual discretionary bonus of up to [X]% of base salary, based on [criteria]. Bonus payments are at Company's sole discretion and require active employment at payment date. 3.3 **Equity.** [If applicable] Subject to Board approval and the Company's equity incentive plan, Employee shall be granted [X shares/options] under the terms of a separate Stock Option Agreement. 3.4 **Benefits.** Employee shall be entitled to participate in benefit plans offered to similarly situated employees, subject to plan terms and eligibility requirements. 3.5 **Expenses.** Company shall reimburse Employee for reasonable business expenses incurred in accordance with Company policy. ## 4. CONFIDENTIALITY 4.1 **Confidential Information.** Employee acknowledges access to confidential and proprietary information including: trade secrets, business plans, customer lists, financial data, technical information, and other non-public information ("Confidential Information"). 4.2 **Non-Disclosure.** During and after employment, Employee shall not disclose, use, or permit use of any Confidential Information except as required for their duties or with prior written consent. 4.3 **Return of Materials.** Upon termination, Employee shall immediately return all Company property and Confidential Information in any form. 4.4 **Survival.** Confidentiality obligations survive termination indefinitely for trade secrets and for [3] years for other Confidential Information. ## 5. INTELLECTUAL PROPERTY 5.1 **Work Product.** All inventions, discoveries, works, and developments created by Employee during employment, relating to Company's business, or using Company resources ("Work Product") shall be Company's sole property. 5.2 **Assignment.** Employee hereby assigns to Company all rights in Work Product, including all intellectual property rights. 5.3 **Assistance.** Employee agrees to execute documents and take actions necessary to perfect Company's rights in Work Product. 5.4 **Prior Inventions.** Attached as Exhibit A is a list of any prior inventions that Employee wishes to exclude from this Agreement. ## 6. NON-COMPETITION AND NON-SOLICITATION [NOTE: Enforceability varies by jurisdiction. Consult local counsel.] 6.1 **Non-Competition.** During employment and for [12] months after termination, Employee shall not, directly or indirectly, engage in any business competitive with Company's business within [Geographic Area]. 6.2 **Non-Solicitation of Customers.** During employment and for [12] months after termination, Employee shall not solicit any customer of the Company for competing products or services. 6.3 **Non-Solicitation of Employees.** During employment and for [12] months after termination, Employee shall not recruit or solicit any Company employee to leave Company employment. ## 7. TERMINATION 7.1 **By Company for Cause.** Company may terminate immediately for Cause, defined as: (a) Material breach of this Agreement (b) Conviction of a felony (c) Fraud, dishonesty, or gross misconduct (d) Failure to perform duties after written notice and cure period 7.2 **By Company Without Cause.** Company may terminate without Cause upon [30] days written notice. 7.3 **By Employee.** Employee may terminate upon [30] days written notice. 7.4 **Severance.** [If applicable] Upon termination without Cause, Employee shall receive [X] weeks base salary as severance, contingent upon execution of a release agreement. 7.5 **Effect of Termination.** Upon termination: - All compensation earned through termination date shall be paid - Unvested equity shall be forfeited - Benefits terminate per plan terms - Sections 4, 5, 6, 8, and 9 survive termination ## 8. GENERAL PROVISIONS 8.1 **Entire Agreement.** This Agreement constitutes the entire agreement and supersedes all prior negotiations, representations, and agreements. 8.2 **Amendments.** This Agreement may be amended only by written agreement signed by both parties. 8.3 **Governing Law.** This Agreement shall be governed by the laws of [State], without regard to conflicts of law principles. 8.4 **Dispute Resolution.** [Arbitration clause or jurisdiction selection] 8.5 **Severability.** If any provision is unenforceable, it shall be modified to the minimum extent necessary, and remaining provisions shall remain in effect. 8.6 **Notices.** Notices shall be in writing and delivered to addresses above. 8.7 **Assignment.** Employee may not assign this Agreement. Company may assign to a successor. 8.8 **Waiver.** Failure to enforce any provision shall not constitute waiver. ## 9. ACKNOWLEDGMENTS Employee acknowledges: - Having read and understood this Agreement - Having opportunity to consult with counsel - Agreeing to all terms voluntarily --- IN WITNESS WHEREOF, the parties have executed this Agreement as of the Effective Date. **[COMPANY NAME]** By: ************\_************ Name: [Authorized Signatory] Title: [Title] Date: ************\_************ **EMPLOYEE** Signature: ************\_************ Name: [Employee Name] Date: ************\_************ --- ## EXHIBIT A: PRIOR INVENTIONS [Employee to list any prior inventions, if any, or write "None"] --- ``` ### Template 3: Employee Handbook Policy Section ```markdown # EMPLOYEE HANDBOOK - POLICY SECTION ## EMPLOYMENT POLICIES ### Equal Employment Opportunity [Company Name] is an equal opportunity employer. We do not discriminate based on race, color, religion, sex, sexual orientation, gender identity, national origin, age, disability, veteran status, or any other protected characteristic. This policy applies to all employment practices including: - Recruitment and hiring - Compensation and benefits - Training and development - Promotions and transfers - Termination ### Anti-Harassment Policy [Company Name] is committed to providing a workplace free from harassment. Harassment based on any protected characteristic is strictly prohibited. **Prohibited Conduct Includes:** - Unwelcome sexual advances or requests for sexual favors - Offensive comments, jokes, or slurs - Physical conduct such as assault or unwanted touching - Visual conduct such as displaying offensive images - Threatening, intimidating, or hostile acts **Reporting Procedure:** 1. Report to your manager, HR, or any member of leadership 2. Reports may be made verbally or in writing 3. Anonymous reports are accepted via [hotline/email] **Investigation:** All reports will be promptly investigated. Retaliation against anyone who reports harassment is strictly prohibited and will result in disciplinary action up to termination. ### Work Hours and Attendance **Standard Hours:** [8:00 AM - 5:00 PM, Monday through Friday] **Core Hours:** [10:00 AM - 3:00 PM] - Employees expected to be available **Flexible Work:** [Policy on remote work, flexible scheduling] **Attendance Expectations:** - Notify your manager as soon as possible if you will be absent - Excessive unexcused absences may result in disciplinary action - [x] unexcused absences in [Y] days considered excessive ### Paid Time Off (PTO) **PTO Accrual:** | Years of Service | Annual PTO Days | |------------------|-----------------| | 0-2 years | 15 days | | 3-5 years | 20 days | | 6+ years | 25 days | **PTO Guidelines:** - PTO accrues per pay period - Maximum accrual: [X] days (use it or lose it after) - Request PTO at least [2] weeks in advance - Manager approval required - PTO may not be taken during [blackout periods] ### Sick Leave - [x] days sick leave per year - May be used for personal illness or family member care - Doctor's note required for absences exceeding [3] days ### Holidays The following paid holidays are observed: - New Year's Day - Martin Luther King Jr. Day - Presidents Day - Memorial Day - Independence Day - Labor Day - Thanksgiving Day - Day after Thanksgiving - Christmas Day - [Floating holiday] ### Code of Conduct All employees are expected to: - Act with integrity and honesty - Treat colleagues, customers, and partners with respect - Protect company confidential information - Avoid conflicts of interest - Comply with all laws and regulations - Report any violations of this code **Violations may result in disciplinary action up to and including termination.** ### Technology and Communication **Acceptable Use:** - Company technology is for business purposes - Limited personal use is permitted if it doesn't interfere with work - No illegal activities or viewing inappropriate content **Monitoring:** - Company reserves the right to monitor company systems - Employees should have no expectation of privacy on company devices **Security:** - Use strong passwords and enable 2FA - Report security incidents immediately - Lock devices when unattended ### Social Media Policy **Personal Social Media:** - Clearly state opinions are your own, not the company's - Do not share confidential company information - Be respectful and professional **Company Social Media:** - Only authorized personnel may post on behalf of the company - Follow brand guidelines - Escalate negative comments to [Marketing/PR] --- ## ACKNOWLEDGMENT I acknowledge that I have received a copy of the Employee Handbook and understand that: 1. I am responsible for reading and understanding its contents 2. The handbook does not create a contract of employment 3. Policies may be changed at any time at the company's discretion 4. Employment is at-will [if applicable] I agree to abide by the policies and procedures outlined in this handbook. Employee Signature: ************\_************ Employee Name (Print): ************\_************ Date: ************\_************ ``` ## Best Practices ### Do's - **Consult legal counsel** - Employment law varies by jurisdiction - **Keep copies signed** - Document all agreements - **Update regularly** - Laws and policies change - **Be clear and specific** - Avoid ambiguity - **Train managers** - On policies and procedures ### Don'ts - **Don't use generic templates** - Customize for your jurisdiction - **Don't make promises** - That could create implied contracts - **Don't discriminate** - In language or application - **Don't forget at-will language** - Where applicable - **Don't skip review** - Have legal counsel review all documents ## Resources - [SHRM Employment Templates](https://www.shrm.org/) - [Department of Labor](https://www.dol.gov/) - [EEOC Guidance](https://www.eeoc.gov/) - State-specific labor departments
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

postmortem-writing

Write effective blameless postmortems with root cause analysis,

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

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

hybrid-search-implementation

Combine vector and keyword search for improved retrieval. Use when

coding
⭐1
# Hybrid Search Implementation Patterns for combining vector similarity and keyword-based search. ## When to Use This Skill - Building RAG systems with improved recall - Combining semantic understanding with exact matching - Handling queries with specific terms (names, codes) - Improving search for domain-specific vocabulary - When pure vector search misses keyword matches ## Core Concepts ### 1. Hybrid Search Architecture ``` Query β†’ ┬─► Vector Search ──► Candidates ─┐ β”‚ β”‚ └─► Keyword Search ─► Candidates ─┴─► Fusion ─► Results ``` ### 2. Fusion Methods | Method | Description | Best For | | ----------------- | ------------------------ | --------------- | | **RRF** | Reciprocal Rank Fusion | General purpose | | **Linear** | Weighted sum of scores | Tunable balance | | **Cross-encoder** | Rerank with neural model | Highest quality | | **Cascade** | Filter then rerank | Efficiency | ## Templates ### Template 1: Reciprocal Rank Fusion ```python from typing import List, Dict, Tuple from collections import defaultdict def reciprocal_rank_fusion( result_lists: List[List[Tuple[str, float]]], k: int = 60, weights: List[float] = None ) -> List[Tuple[str, float]]: """ Combine multiple ranked lists using RRF. Args: result_lists: List of (doc_id, score) tuples per search method k: RRF constant (higher = more weight to lower ranks) weights: Optional weights per result list Returns: Fused ranking as (doc_id, score) tuples """ if weights is None: weights = [1.0] * len(result_lists) scores = defaultdict(float) for result_list, weight in zip(result_lists, weights): for rank, (doc_id, _) in enumerate(result_list): # RRF formula: 1 / (k + rank) scores[doc_id] += weight * (1.0 / (k + rank + 1)) # Sort by fused score return sorted(scores.items(), key=lambda x: x[1], reverse=True) def linear_combination( vector_results: List[Tuple[str, float]], keyword_results: List[Tuple[str, float]], alpha: float = 0.5 ) -> List[Tuple[str, float]]: """ Combine results with linear interpolation. Args: vector_results: (doc_id, similarity_score) from vector search keyword_results: (doc_id, bm25_score) from keyword search alpha: Weight for vector search (1-alpha for keyword) """ # Normalize scores to [0, 1] def normalize(results): if not results: return {} scores = [s for _, s in results] min_s, max_s = min(scores), max(scores) range_s = max_s - min_s if max_s != min_s else 1 return {doc_id: (score - min_s) / range_s for doc_id, score in results} vector_scores = normalize(vector_results) keyword_scores = normalize(keyword_results) # Combine all_docs = set(vector_scores.keys()) | set(keyword_scores.keys()) combined = {} for doc_id in all_docs: v_score = vector_scores.get(doc_id, 0) k_score = keyword_scores.get(doc_id, 0) combined[doc_id] = alpha * v_score + (1 - alpha) * k_score return sorted(combined.items(), key=lambda x: x[1], reverse=True) ``` ### Template 2: PostgreSQL Hybrid Search ```python import asyncpg from typing import List, Dict, Optional import numpy as np class PostgresHybridSearch: """Hybrid search with pgvector and full-text search.""" def __init__(self, pool: asyncpg.Pool): self.pool = pool async def setup_schema(self): """Create tables and indexes.""" async with self.pool.acquire() as conn: await conn.execute(""" CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS documents ( id TEXT PRIMARY KEY, content TEXT NOT NULL, embedding vector(1536), metadata JSONB DEFAULT '{}', ts_content tsvector GENERATED ALWAYS AS ( to_tsvector('english', content) ) STORED ); -- Vector index (HNSW) CREATE INDEX IF NOT EXISTS documents_embedding_idx ON documents USING hnsw (embedding vector_cosine_ops); -- Full-text index (GIN) CREATE INDEX IF NOT EXISTS documents_fts_idx ON documents USING gin (ts_content); """) async def hybrid_search( self, query: str, query_embedding: List[float], limit: int = 10, vector_weight: float = 0.5, filter_metadata: Optional[Dict] = None ) -> List[Dict]: """ Perform hybrid search combining vector and full-text. Uses RRF fusion for combining results. """ async with self.pool.acquire() as conn: # Build filter clause where_clause = "1=1" params = [query_embedding, query, limit * 3] if filter_metadata: for key, value in filter_metadata.items(): params.append(value) where_clause += f" AND metadata->>'{key}' = ${len(params)}" results = await conn.fetch(f""" WITH vector_search AS ( SELECT id, content, metadata, ROW_NUMBER() OVER (ORDER BY embedding <=> $1::vector) as vector_rank, 1 - (embedding <=> $1::vector) as vector_score FROM documents WHERE {where_clause} ORDER BY embedding <=> $1::vector LIMIT $3 ), keyword_search AS ( SELECT id, content, metadata, ROW_NUMBER() OVER (ORDER BY ts_rank(ts_content, websearch_to_tsquery('english', $2)) DESC) as keyword_rank, ts_rank(ts_content, websearch_to_tsquery('english', $2)) as keyword_score FROM documents WHERE ts_content @@ websearch_to_tsquery('english', $2) AND {where_clause} ORDER BY ts_rank(ts_content, websearch_to_tsquery('english', $2)) DESC LIMIT $3 ) SELECT COALESCE(v.id, k.id) as id, COALESCE(v.content, k.content) as content, COALESCE(v.metadata, k.metadata) as metadata, v.vector_score, k.keyword_score, -- RRF fusion COALESCE(1.0 / (60 + v.vector_rank), 0) * $4::float + COALESCE(1.0 / (60 + k.keyword_rank), 0) * (1 - $4::float) as rrf_score FROM vector_search v FULL OUTER JOIN keyword_search k ON v.id = k.id ORDER BY rrf_score DESC LIMIT $3 / 3 """, *params, vector_weight) return [dict(row) for row in results] async def search_with_rerank( self, query: str, query_embedding: List[float], limit: int = 10, rerank_candidates: int = 50 ) -> List[Dict]: """Hybrid search with cross-encoder reranking.""" from sentence_transformers import CrossEncoder # Get candidates candidates = await self.hybrid_search( query, query_embedding, limit=rerank_candidates ) if not candidates: return [] # Rerank with cross-encoder model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') pairs = [(query, c["content"]) for c in candidates] scores = model.predict(pairs) for candidate, score in zip(candidates, scores): candidate["rerank_score"] = float(score) # Sort by rerank score and return top results reranked = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True) return reranked[:limit] ``` ### Template 3: Elasticsearch Hybrid Search ```python from elasticsearch import Elasticsearch from typing import List, Dict, Optional class ElasticsearchHybridSearch: """Hybrid search with Elasticsearch and dense vectors.""" def __init__( self, es_client: Elasticsearch, index_name: str = "documents" ): self.es = es_client self.index_name = index_name def create_index(self, vector_dims: int = 1536): """Create index with dense vector and text fields.""" mapping = { "mappings": { "properties": { "content": { "type": "text", "analyzer": "english" }, "embedding": { "type": "dense_vector", "dims": vector_dims, "index": True, "similarity": "cosine" }, "metadata": { "type": "object", "enabled": True } } } } self.es.indices.create(index=self.index_name, body=mapping, ignore=400) def hybrid_search( self, query: str, query_embedding: List[float], limit: int = 10, boost_vector: float = 1.0, boost_text: float = 1.0, filter: Optional[Dict] = None ) -> List[Dict]: """ Hybrid search using Elasticsearch's built-in capabilities. """ # Build the hybrid query search_body = { "size": limit, "query": { "bool": { "should": [ # Vector search (kNN) { "script_score": { "query": {"match_all": {}}, "script": { "source": f"cosineSimilarity(params.query_vector, 'embedding') * {boost_vector} + 1.0", "params": {"query_vector": query_embedding} } } }, # Text search (BM25) { "match": { "content": { "query": query, "boost": boost_text } } } ], "minimum_should_match": 1 } } } # Add filter if provided if filter: search_body["query"]["bool"]["filter"] = filter response = self.es.search(index=self.index_name, body=search_body) return [ { "id": hit["_id"], "content": hit["_source"]["content"], "metadata": hit["_source"].get("metadata", {}), "score": hit["_score"] } for hit in response["hits"]["hits"] ] def hybrid_search_rrf( self, query: str, query_embedding: List[float], limit: int = 10, window_size: int = 100 ) -> List[Dict]: """ Hybrid search using Elasticsearch 8.x RRF. """ search_body = { "size": limit, "sub_searches": [ { "query": { "match": { "content": query } } }, { "query": { "knn": { "field": "embedding", "query_vector": query_embedding, "k": window_size, "num_candidates": window_size * 2 } } } ], "rank": { "rrf": { "window_size": window_size, "rank_constant": 60 } } } response = self.es.search(index=self.index_name, body=search_body) return [ { "id": hit["_id"], "content": hit["_source"]["content"], "score": hit["_score"] } for hit in response["hits"]["hits"] ] ``` ### Template 4: Custom Hybrid RAG Pipeline ```python from typing import List, Dict, Optional, Callable from dataclasses import dataclass @dataclass class SearchResult: id: str content: str score: float source: str # "vector", "keyword", "hybrid" metadata: Dict = None class HybridRAGPipeline: """Complete hybrid search pipeline for RAG.""" def __init__( self, vector_store, keyword_store, embedder, reranker=None, fusion_method: str = "rrf", vector_weight: float = 0.5 ): self.vector_store = vector_store self.keyword_store = keyword_store self.embedder = embedder self.reranker = reranker self.fusion_method = fusion_method self.vector_weight = vector_weight async def search( self, query: str, top_k: int = 10, filter: Optional[Dict] = None, use_rerank: bool = True ) -> List[SearchResult]: """Execute hybrid search pipeline.""" # Step 1: Get query embedding query_embedding = self.embedder.embed(query) # Step 2: Execute parallel searches vector_results, keyword_results = await asyncio.gather( self._vector_search(query_embedding, top_k * 3, filter), self._keyword_search(query, top_k * 3, filter) ) # Step 3: Fuse results if self.fusion_method == "rrf": fused = self._rrf_fusion(vector_results, keyword_results) else: fused = self._linear_fusion(vector_results, keyword_results) # Step 4: Rerank if enabled if use_rerank and self.reranker: fused = await self._rerank(query, fused[:top_k * 2]) return fused[:top_k] async def _vector_search( self, embedding: List[float], limit: int, filter: Dict ) -> List[SearchResult]: results = await self.vector_store.search(embedding, limit, filter) return [ SearchResult( id=r["id"], content=r["content"], score=r["score"], source="vector", metadata=r.get("metadata") ) for r in results ] async def _keyword_search( self, query: str, limit: int, filter: Dict ) -> List[SearchResult]: results = await self.keyword_store.search(query, limit, filter) return [ SearchResult( id=r["id"], content=r["content"], score=r["score"], source="keyword", metadata=r.get("metadata") ) for r in results ] def _rrf_fusion( self, vector_results: List[SearchResult], keyword_results: List[SearchResult] ) -> List[SearchResult]: """Fuse with RRF.""" k = 60 scores = {} content_map = {} for rank, result in enumerate(vector_results): scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1) content_map[result.id] = result for rank, result in enumerate(keyword_results): scores[result.id] = scores.get(result.id, 0) + 1 / (k + rank + 1) if result.id not in content_map: content_map[result.id] = result sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True) return [ SearchResult( id=doc_id, content=content_map[doc_id].content, score=scores[doc_id], source="hybrid", metadata=content_map[doc_id].metadata ) for doc_id in sorted_ids ] async def _rerank( self, query: str, results: List[SearchResult] ) -> List[SearchResult]: """Rerank with cross-encoder.""" if not results: return results pairs = [(query, r.content) for r in results] scores = self.reranker.predict(pairs) for result, score in zip(results, scores): result.score = float(score) return sorted(results, key=lambda x: x.score, reverse=True) ``` ## Best Practices ### Do's - **Tune weights empirically** - Test on your data - **Use RRF for simplicity** - Works well without tuning - **Add reranking** - Significant quality improvement - **Log both scores** - Helps with debugging - **A/B test** - Measure real user impact ### Don'ts - **Don't assume one size fits all** - Different queries need different weights - **Don't skip keyword search** - Handles exact matches better - **Don't over-fetch** - Balance recall vs latency - **Don't ignore edge cases** - Empty results, single word queries ## Resources - [RRF Paper](https://plg.uwaterloo.ca/~gvcormac/cormacksigir09-rrf.pdf) - [Vespa Hybrid Search](https://blog.vespa.ai/improving-text-ranking-with-few-shot-prompting/) - [Cohere Rerank](https://docs.cohere.com/docs/reranking)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

langchain-architecture

Design LLM applications using LangChain 1.x and LangGraph for

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

llm-evaluation

Implement comprehensive evaluation strategies for LLM applications

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

prompt-engineering-patterns

Master advanced prompt engineering techniques to maximize LLM

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

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

similarity-search-patterns

Implement efficient similarity search with vector databases. Use

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

ml-pipeline-workflow

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

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

python-design-patterns

Python design patterns including KISS, Separation of Concerns,

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

protocol-reverse-engineering

Master network protocol reverse engineering including packet

security
⭐1
# Protocol Reverse Engineering Comprehensive techniques for capturing, analyzing, and documenting network protocols for security research, interoperability, and debugging. ## Traffic Capture ### Wireshark Capture ```bash # Capture on specific interface wireshark -i eth0 -k # Capture with filter wireshark -i eth0 -k -f "port 443" # Capture to file tshark -i eth0 -w capture.pcap # Ring buffer capture (rotate files) tshark -i eth0 -b filesize:100000 -b files:10 -w capture.pcap ``` ### tcpdump Capture ```bash # Basic capture tcpdump -i eth0 -w capture.pcap # With filter tcpdump -i eth0 port 8080 -w capture.pcap # Capture specific bytes tcpdump -i eth0 -s 0 -w capture.pcap # Full packet # Real-time display tcpdump -i eth0 -X port 80 ``` ### Man-in-the-Middle Capture ```bash # mitmproxy for HTTP/HTTPS mitmproxy --mode transparent -p 8080 # SSL/TLS interception mitmproxy --mode transparent --ssl-insecure # Dump to file mitmdump -w traffic.mitm # Burp Suite # Configure browser proxy to 127.0.0.1:8080 ``` ## Protocol Analysis ### Wireshark Analysis ``` # Display filters tcp.port == 8080 http.request.method == "POST" ip.addr == 192.168.1.1 tcp.flags.syn == 1 && tcp.flags.ack == 0 frame contains "password" # Following streams Right-click > Follow > TCP Stream Right-click > Follow > HTTP Stream # Export objects File > Export Objects > HTTP # Decryption Edit > Preferences > Protocols > TLS - (Pre)-Master-Secret log filename - RSA keys list ``` ### tshark Analysis ```bash # Extract specific fields tshark -r capture.pcap -T fields -e ip.src -e ip.dst -e tcp.port # Statistics tshark -r capture.pcap -q -z conv,tcp tshark -r capture.pcap -q -z endpoints,ip # Filter and extract tshark -r capture.pcap -Y "http" -T json > http_traffic.json # Protocol hierarchy tshark -r capture.pcap -q -z io,phs ``` ### Scapy for Custom Analysis ```python from scapy.all import * # Read pcap packets = rdpcap("capture.pcap") # Analyze packets for pkt in packets: if pkt.haslayer(TCP): print(f"Src: {pkt[IP].src}:{pkt[TCP].sport}") print(f"Dst: {pkt[IP].dst}:{pkt[TCP].dport}") if pkt.haslayer(Raw): print(f"Data: {pkt[Raw].load[:50]}") # Filter packets http_packets = [p for p in packets if p.haslayer(TCP) and (p[TCP].sport == 80 or p[TCP].dport == 80)] # Create custom packets pkt = IP(dst="target")/TCP(dport=80)/Raw(load="GET / HTTP/1.1\r\n") send(pkt) ``` ## Protocol Identification ### Common Protocol Signatures ``` HTTP - "HTTP/1." or "GET " or "POST " at start TLS/SSL - 0x16 0x03 (record layer) DNS - UDP port 53, specific header format SMB - 0xFF 0x53 0x4D 0x42 ("SMB" signature) SSH - "SSH-2.0" banner FTP - "220 " response, "USER " command SMTP - "220 " banner, "EHLO" command MySQL - 0x00 length prefix, protocol version PostgreSQL - 0x00 0x00 0x00 startup length Redis - "*" RESP array prefix MongoDB - BSON documents with specific header ``` ### Protocol Header Patterns ``` +--------+--------+--------+--------+ | Magic number / Signature | +--------+--------+--------+--------+ | Version | Flags | +--------+--------+--------+--------+ | Length | Message Type | +--------+--------+--------+--------+ | Sequence Number / Session ID | +--------+--------+--------+--------+ | Payload... | +--------+--------+--------+--------+ ``` ## Binary Protocol Analysis ### Structure Identification ```python # Common patterns in binary protocols # Length-prefixed message struct Message { uint32_t length; # Total message length uint16_t msg_type; # Message type identifier uint8_t flags; # Flags/options uint8_t reserved; # Padding/alignment uint8_t payload[]; # Variable-length payload }; # Type-Length-Value (TLV) struct TLV { uint8_t type; # Field type uint16_t length; # Field length uint8_t value[]; # Field data }; # Fixed header + variable payload struct Packet { uint8_t magic[4]; # "ABCD" signature uint32_t version; uint32_t payload_len; uint32_t checksum; # CRC32 or similar uint8_t payload[]; }; ``` ### Python Protocol Parser ```python import struct from dataclasses import dataclass @dataclass class MessageHeader: magic: bytes version: int msg_type: int length: int @classmethod def from_bytes(cls, data: bytes): magic, version, msg_type, length = struct.unpack( ">4sHHI", data[:12] ) return cls(magic, version, msg_type, length) def parse_messages(data: bytes): offset = 0 messages = [] while offset < len(data): header = MessageHeader.from_bytes(data[offset:]) payload = data[offset+12:offset+12+header.length] messages.append((header, payload)) offset += 12 + header.length return messages # Parse TLV structure def parse_tlv(data: bytes): fields = [] offset = 0 while offset < len(data): field_type = data[offset] length = struct.unpack(">H", data[offset+1:offset+3])[0] value = data[offset+3:offset+3+length] fields.append((field_type, value)) offset += 3 + length return fields ``` ### Hex Dump Analysis ```python def hexdump(data: bytes, width: int = 16): """Format binary data as hex dump.""" lines = [] for i in range(0, len(data), width): chunk = data[i:i+width] hex_part = ' '.join(f'{b:02x}' for b in chunk) ascii_part = ''.join( chr(b) if 32 <= b < 127 else '.' for b in chunk ) lines.append(f'{i:08x} {hex_part:<{width*3}} {ascii_part}') return '\n'.join(lines) # Example output: # 00000000 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d HTTP/1.1 200 OK. # 00000010 0a 43 6f 6e 74 65 6e 74 2d 54 79 70 65 3a 20 74 .Content-Type: t ``` ## Encryption Analysis ### Identifying Encryption ```python # Entropy analysis - high entropy suggests encryption/compression import math from collections import Counter def entropy(data: bytes) -> float: if not data: return 0.0 counter = Counter(data) probs = [count / len(data) for count in counter.values()] return -sum(p * math.log2(p) for p in probs) # Entropy thresholds: # < 6.0: Likely plaintext or structured data # 6.0-7.5: Possibly compressed # > 7.5: Likely encrypted or random # Common encryption indicators # - High, uniform entropy # - No obvious structure or patterns # - Length often multiple of block size (16 for AES) # - Possible IV at start (16 bytes for AES-CBC) ``` ### TLS Analysis ```bash # Extract TLS metadata tshark -r capture.pcap -Y "ssl.handshake" \ -T fields -e ip.src -e ssl.handshake.ciphersuite # JA3 fingerprinting (client) tshark -r capture.pcap -Y "ssl.handshake.type == 1" \ -T fields -e ssl.handshake.ja3 # JA3S fingerprinting (server) tshark -r capture.pcap -Y "ssl.handshake.type == 2" \ -T fields -e ssl.handshake.ja3s # Certificate extraction tshark -r capture.pcap -Y "ssl.handshake.certificate" \ -T fields -e x509sat.printableString ``` ### Decryption Approaches ```bash # Pre-master secret log (browser) export SSLKEYLOGFILE=/tmp/keys.log # Configure Wireshark # Edit > Preferences > Protocols > TLS # (Pre)-Master-Secret log filename: /tmp/keys.log # Decrypt with private key (if available) # Only works for RSA key exchange # Edit > Preferences > Protocols > TLS > RSA keys list ``` ## Custom Protocol Documentation ### Protocol Specification Template ```markdown # Protocol Name Specification ## Overview Brief description of protocol purpose and design. ## Transport - Layer: TCP/UDP - Port: XXXX - Encryption: TLS 1.2+ ## Message Format ### Header (12 bytes) | Offset | Size | Field | Description | | ------ | ---- | ------- | ----------------------- | | 0 | 4 | Magic | 0x50524F54 ("PROT") | | 4 | 2 | Version | Protocol version (1) | | 6 | 2 | Type | Message type identifier | | 8 | 4 | Length | Payload length in bytes | ### Message Types | Type | Name | Description | | ---- | --------- | ---------------------- | | 0x01 | HELLO | Connection initiation | | 0x02 | HELLO_ACK | Connection accepted | | 0x03 | DATA | Application data | | 0x04 | CLOSE | Connection termination | ### Type 0x01: HELLO | Offset | Size | Field | Description | | ------ | ---- | ---------- | ------------------------ | | 0 | 4 | ClientID | Unique client identifier | | 4 | 2 | Flags | Connection flags | | 6 | var | Extensions | TLV-encoded extensions | ## State Machine ``` [INIT] --HELLO--> [WAIT_ACK] --HELLO_ACK--> [CONNECTED] | DATA/DATA | [CLOSED] <--CLOSE--+ ``` ## Examples ### Connection Establishment ``` Client -> Server: HELLO (ClientID=0x12345678) Server -> Client: HELLO_ACK (Status=OK) Client -> Server: DATA (payload) ``` ``` ### Wireshark Dissector (Lua) ```lua -- custom_protocol.lua local proto = Proto("custom", "Custom Protocol") -- Define fields local f_magic = ProtoField.string("custom.magic", "Magic") local f_version = ProtoField.uint16("custom.version", "Version") local f_type = ProtoField.uint16("custom.type", "Type") local f_length = ProtoField.uint32("custom.length", "Length") local f_payload = ProtoField.bytes("custom.payload", "Payload") proto.fields = { f_magic, f_version, f_type, f_length, f_payload } -- Message type names local msg_types = { [0x01] = "HELLO", [0x02] = "HELLO_ACK", [0x03] = "DATA", [0x04] = "CLOSE" } function proto.dissector(buffer, pinfo, tree) pinfo.cols.protocol = "CUSTOM" local subtree = tree:add(proto, buffer()) -- Parse header subtree:add(f_magic, buffer(0, 4)) subtree:add(f_version, buffer(4, 2)) local msg_type = buffer(6, 2):uint() subtree:add(f_type, buffer(6, 2)):append_text( " (" .. (msg_types[msg_type] or "Unknown") .. ")" ) local length = buffer(8, 4):uint() subtree:add(f_length, buffer(8, 4)) if length > 0 then subtree:add(f_payload, buffer(12, length)) end end -- Register for TCP port local tcp_table = DissectorTable.get("tcp.port") tcp_table:add(8888, proto) ``` ## Active Testing ### Fuzzing with Boofuzz ```python from boofuzz import * def main(): session = Session( target=Target( connection=TCPSocketConnection("target", 8888) ) ) # Define protocol structure s_initialize("HELLO") s_static(b"\x50\x52\x4f\x54") # Magic s_word(1, name="version") # Version s_word(0x01, name="type") # Type (HELLO) s_size("payload", length=4) # Length field s_block_start("payload") s_dword(0x12345678, name="client_id") s_word(0, name="flags") s_block_end() session.connect(s_get("HELLO")) session.fuzz() if __name__ == "__main__": main() ``` ### Replay and Modification ```python from scapy.all import * # Replay captured traffic packets = rdpcap("capture.pcap") for pkt in packets: if pkt.haslayer(TCP) and pkt[TCP].dport == 8888: send(pkt) # Modify and replay for pkt in packets: if pkt.haslayer(Raw): # Modify payload original = pkt[Raw].load modified = original.replace(b"client", b"CLIENT") pkt[Raw].load = modified # Recalculate checksums del pkt[IP].chksum del pkt[TCP].chksum send(pkt) ``` ## Best Practices ### Analysis Workflow 1. **Capture traffic**: Multiple sessions, different scenarios 2. **Identify boundaries**: Message start/end markers 3. **Map structure**: Fixed header, variable payload 4. **Identify fields**: Compare multiple samples 5. **Document format**: Create specification 6. **Validate understanding**: Implement parser/generator 7. **Test edge cases**: Fuzzing, boundary conditions ### Common Patterns to Look For - Magic numbers/signatures at message start - Version fields for compatibility - Length fields (often before variable data) - Type/opcode fields for message identification - Sequence numbers for ordering - Checksums/CRCs for integrity - Timestamps for timing - Session/connection identifiers
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

market-sizing-analysis

This skill should be used when the user asks to "calculate TAM",

business
⭐1
# Market Sizing Analysis Comprehensive market sizing methodologies for calculating Total Addressable Market (TAM), Serviceable Available Market (SAM), and Serviceable Obtainable Market (SOM) for startup opportunities. ## Overview Market sizing provides the foundation for startup strategy, fundraising, and business planning. Calculate market opportunity using three complementary methodologies: top-down (industry reports), bottom-up (customer segment calculations), and value theory (willingness to pay). ## Core Concepts ### The Three-Tier Market Framework **TAM (Total Addressable Market)** - Total revenue opportunity if achieving 100% market share - Defines the universe of potential customers - Used for long-term vision and market validation - Example: All email marketing software revenue globally **SAM (Serviceable Available Market)** - Portion of TAM targetable with current product/service - Accounts for geographic, segment, or capability constraints - Represents realistic addressable opportunity - Example: AI-powered email marketing for e-commerce in North America **SOM (Serviceable Obtainable Market)** - Realistic market share achievable in 3-5 years - Accounts for competition, resources, and market dynamics - Used for financial projections and fundraising - Example: 2-5% of SAM based on competitive landscape ### When to Use Each Methodology **Top-Down Analysis** - Use when established market research exists - Best for mature, well-defined markets - Validates market existence and growth - Starts with industry reports and narrows down **Bottom-Up Analysis** - Use when targeting specific customer segments - Best for new or niche markets - Most credible for investors - Builds from customer data and pricing **Value Theory** - Use when creating new market categories - Best for disruptive innovations - Estimates based on value creation - Calculates willingness to pay for problem solution ## Three-Methodology Framework ### Methodology 1: Top-Down Analysis Start with total market size and narrow to addressable segments. **Process:** 1. Identify total market category from research reports 2. Apply geographic filters (target regions) 3. Apply segment filters (target industries/customers) 4. Calculate competitive positioning adjustments **Formula:** ``` TAM = Total Market Category Size SAM = TAM Γ— Geographic % Γ— Segment % SOM = SAM Γ— Realistic Capture Rate (2-5%) ``` **When to use:** Established markets with available research (e.g., SaaS, fintech, e-commerce) **Strengths:** Quick, uses credible data, validates market existence **Limitations:** May overestimate for new categories, less granular ### Methodology 2: Bottom-Up Analysis Build market size from customer segment calculations. **Process:** 1. Define target customer segments 2. Estimate number of potential customers per segment 3. Determine average revenue per customer 4. Calculate realistic penetration rates **Formula:** ``` TAM = Ξ£ (Segment Size Γ— Annual Revenue per Customer) SAM = TAM Γ— (Segments You Can Serve / Total Segments) SOM = SAM Γ— Realistic Penetration Rate (Year 3-5) ``` **When to use:** B2B, niche markets, specific customer segments **Strengths:** Most credible for investors, granular, defensible **Limitations:** Requires detailed customer research, time-intensive ### Methodology 3: Value Theory Calculate based on value created and willingness to pay. **Process:** 1. Identify problem being solved 2. Quantify current cost of problem (time, money, inefficiency) 3. Calculate value of solution (savings, gains, efficiency) 4. Estimate willingness to pay (typically 10-30% of value) 5. Multiply by addressable customer base **Formula:** ``` Value per Customer = Problem Cost Γ— % Solved by Solution Price per Customer = Value Γ— Willingness to Pay % (10-30%) TAM = Total Potential Customers Γ— Price per Customer SAM = TAM Γ— % Meeting Buy Criteria SOM = SAM Γ— Realistic Adoption Rate ``` **When to use:** New categories, disruptive innovations, unclear existing markets **Strengths:** Shows value creation, works for new markets **Limitations:** Requires assumptions, harder to validate ## Step-by-Step Process ### Step 1: Define the Market Clearly specify what market is being measured. **Questions to answer:** - What problem is being solved? - Who are the target customers? - What's the product/service category? - What's the geographic scope? - What's the time horizon? **Example:** - Problem: E-commerce companies struggle with email marketing automation - Customers: E-commerce stores with >$1M annual revenue - Category: AI-powered email marketing software - Geography: North America initially, global expansion - Horizon: 3-5 year opportunity ### Step 2: Gather Data Sources Identify credible data for calculations. **Top-Down Sources:** - Industry research reports (Gartner, Forrester, IDC) - Government statistics (Census, BLS, trade associations) - Public company filings and earnings - Market research firms (Statista, CB Insights, PitchBook) **Bottom-Up Sources:** - Customer interviews and surveys - Sales data and CRM records - Industry databases (LinkedIn, ZoomInfo, Crunchbase) - Competitive intelligence - Academic research **Value Theory Sources:** - Customer problem quantification - Time/cost studies - ROI case studies - Pricing research and willingness-to-pay surveys ### Step 3: Calculate TAM Apply chosen methodology to determine total market. **For Top-Down:** 1. Find total category size from research 2. Document data source and year 3. Apply growth rate if needed 4. Validate with multiple sources **For Bottom-Up:** 1. Count total potential customers 2. Calculate average annual revenue per customer 3. Multiply to get TAM 4. Break down by segment **For Value Theory:** 1. Quantify total addressable customer base 2. Calculate value per customer 3. Estimate pricing based on value 4. Multiply for TAM ### Step 4: Calculate SAM Narrow TAM to serviceable addressable market. **Apply Filters:** - Geographic constraints (regions you can serve) - Product limitations (features you currently have) - Customer requirements (size, industry, use case) - Distribution channel access - Regulatory or compliance restrictions **Formula:** ``` SAM = TAM Γ— (% matching all filters) ``` **Example:** - TAM: $10B global email marketing - Geographic filter: 40% (North America) - Product filter: 30% (e-commerce focus) - Feature filter: 60% (need AI capabilities) - SAM = $10B Γ— 0.40 Γ— 0.30 Γ— 0.60 = $720M ### Step 5: Calculate SOM Determine realistic obtainable market share. **Consider:** - Current market share of competitors - Typical market share for new entrants (2-5%) - Resources available (funding, team, time) - Go-to-market effectiveness - Competitive advantages - Time to achieve (3-5 years typically) **Conservative Approach:** ``` SOM (Year 3) = SAM Γ— 2% SOM (Year 5) = SAM Γ— 5% ``` **Example:** - SAM: $720M - Year 3 SOM: $720M Γ— 2% = $14.4M - Year 5 SOM: $720M Γ— 5% = $36M ### Step 6: Validate and Triangulate Cross-check using multiple methods. **Validation Techniques:** 1. Compare top-down and bottom-up results (should be within 30%) 2. Check against public company revenues in space 3. Validate customer count assumptions 4. Sense-check pricing assumptions 5. Review with industry experts 6. Compare to similar market categories **Red Flags:** - TAM that's too small (< $1B for VC-backed startups) - TAM that's too large (unsupported by data) - SOM that's too aggressive (> 10% in 5 years for new entrant) - Inconsistency between methodologies (> 50% difference) ## Industry-Specific Considerations ### SaaS Markets **Key Metrics:** - Number of potential businesses in target segment - Average contract value (ACV) - Typical market penetration rates - Expansion revenue potential **TAM Calculation:** ``` TAM = Total Target Companies Γ— Average ACV Γ— (1 + Expansion Rate) ``` ### Marketplace Markets **Key Metrics:** - Gross Merchandise Value (GMV) of category - Take rate (% of GMV you capture) - Total transactions or users **TAM Calculation:** ``` TAM = Total Category GMV Γ— Expected Take Rate ``` ### Consumer Markets **Key Metrics:** - Total addressable users/households - Average revenue per user (ARPU) - Engagement frequency **TAM Calculation:** ``` TAM = Total Users Γ— ARPU Γ— Purchase Frequency per Year ``` ### B2B Services **Key Metrics:** - Number of target companies by size/industry - Average project value or retainer - Typical buying frequency **TAM Calculation:** ``` TAM = Total Target Companies Γ— Average Deal Size Γ— Deals per Year ``` ## Presenting Market Sizing ### For Investors **Structure:** 1. Market definition and problem scope 2. TAM/SAM/SOM with methodology 3. Data sources and assumptions 4. Growth projections and drivers 5. Competitive landscape context **Key Points:** - Lead with bottom-up calculation (most credible) - Show triangulation with top-down - Explain conservative assumptions - Link to revenue projections - Highlight market growth rate ### For Strategy **Structure:** 1. Addressable customer segments 2. Prioritization by opportunity size 3. Entry strategy by segment 4. Expected penetration timeline 5. Resource requirements **Key Points:** - Focus on SAM and SOM - Show segment-level detail - Connect to go-to-market plan - Identify expansion opportunities - Discuss competitive positioning ## Common Mistakes to Avoid **Mistake 1: Confusing TAM with SAM** - Don't claim entire market as addressable - Apply realistic product/geographic constraints - Be honest about serviceable market **Mistake 2: Overly Aggressive SOM** - New entrants rarely capture > 5% in 5 years - Account for competition and resources - Show realistic ramp timeline **Mistake 3: Using Only Top-Down** - Investors prefer bottom-up validation - Top-down alone lacks credibility - Always triangulate with multiple methods **Mistake 4: Cherry-Picking Data** - Use consistent, recent data sources - Don't mix methodologies inappropriately - Document all assumptions clearly **Mistake 5: Ignoring Market Dynamics** - Account for market growth/decline - Consider competitive intensity - Factor in switching costs and barriers ## Additional Resources ### Reference Files For detailed methodologies and frameworks: - **`references/methodology-deep-dive.md`** - Comprehensive guide to each methodology with step-by-step worksheets - **`references/data-sources.md`** - Curated list of market research sources, databases, and tools - **`references/industry-templates.md`** - Specific templates for SaaS, marketplace, consumer, B2B, and fintech markets ### Example Files Working examples with complete calculations: - **`examples/saas-market-sizing.md`** - Complete TAM/SAM/SOM for a B2B SaaS product - **`examples/marketplace-sizing.md`** - Marketplace platform market opportunity calculation - **`examples/value-theory-example.md`** - Value-based market sizing for disruptive innovation Use these examples as templates for your own market sizing analysis. Each includes real numbers, data sources, and assumptions documented clearly. ## Quick Start To perform market sizing analysis: 1. **Define the market** - Problem, customers, category, geography 2. **Choose methodology** - Bottom-up (preferred) or top-down + triangulation 3. **Gather data** - Industry reports, customer data, competitive intelligence 4. **Calculate TAM** - Apply methodology formula 5. **Narrow to SAM** - Apply product, geographic, segment filters 6. **Estimate SOM** - 2-5% realistic capture rate 7. **Validate** - Cross-check with alternative methods 8. **Document** - Show methodology, sources, assumptions 9. **Present** - Structure for audience (investors, strategy, operations) For detailed step-by-step guidance on each methodology, reference the files in `references/` directory. For complete worked examples, see `examples/` directory.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

startup-financial-modeling

This skill should be used when the user asks to "create financial

business
⭐1
# Startup Financial Modeling Build comprehensive 3-5 year financial models with revenue projections, cost structures, cash flow analysis, and scenario planning for early-stage startups. ## Overview Financial modeling provides the quantitative foundation for startup strategy, fundraising, and operational planning. Create realistic projections using cohort-based revenue modeling, detailed cost structures, and scenario analysis to support decision-making and investor presentations. ## Core Components ### Revenue Model **Cohort-Based Projections:** Build revenue from customer acquisition and retention by cohort. **Formula:** ``` MRR = Ξ£ (Cohort Size Γ— Retention Rate Γ— ARPU) ARR = MRR Γ— 12 ``` **Key Inputs:** - Monthly new customer acquisitions - Customer retention rates by month - Average revenue per user (ARPU) - Pricing and packaging assumptions - Expansion revenue (upsells, cross-sells) ### Cost Structure **Operating Expenses Categories:** 1. **Cost of Goods Sold (COGS)** - Hosting and infrastructure - Payment processing fees - Customer support (variable portion) - Third-party services per customer 2. **Sales & Marketing (S&M)** - Customer acquisition cost (CAC) - Marketing programs and advertising - Sales team compensation - Marketing tools and software 3. **Research & Development (R&D)** - Engineering team compensation - Product management - Design and UX - Development tools and infrastructure 4. **General & Administrative (G&A)** - Executive team - Finance, legal, HR - Office and facilities - Insurance and compliance ### Cash Flow Analysis **Components:** - Beginning cash balance - Cash inflows (revenue, fundraising) - Cash outflows (operating expenses, CapEx) - Ending cash balance - Monthly burn rate - Runway (months of cash remaining) **Formula:** ``` Runway = Current Cash Balance / Monthly Burn Rate Monthly Burn = Monthly Revenue - Monthly Expenses ``` ### Headcount Planning **Role-Based Hiring Plan:** Track headcount by department and role. **Key Metrics:** - Fully-loaded cost per employee - Revenue per employee - Headcount by department (% of total) **Typical Ratios (Early-Stage SaaS):** - Engineering: 40-50% - Sales & Marketing: 25-35% - G&A: 10-15% - Customer Success: 5-10% ## Financial Model Structure ### Three-Scenario Framework **Conservative Scenario (P10):** - Slower customer acquisition - Lower pricing or conversion - Higher churn rates - Extended sales cycles - Used for cash management **Base Scenario (P50):** - Most likely outcomes - Realistic assumptions - Primary planning scenario - Used for board reporting **Optimistic Scenario (P90):** - Faster growth - Better unit economics - Lower churn - Used for upside planning ### Time Horizon **Detailed Projections: 3 Years** - Monthly detail for Year 1 - Monthly detail for Year 2 - Quarterly detail for Year 3 **High-Level Projections: Years 4-5** - Annual projections - Key metrics only - Support long-term planning ## Step-by-Step Process ### Step 1: Define Business Model Clarify revenue model and pricing. **SaaS Model:** - Subscription pricing tiers - Annual vs. monthly contracts - Free trial or freemium approach - Expansion revenue strategy **Marketplace Model:** - GMV projections - Take rate (% of transactions) - Buyer and seller economics - Transaction frequency **Transactional Model:** - Transaction volume - Revenue per transaction - Frequency and seasonality ### Step 2: Build Revenue Projections Use cohort-based methodology for accuracy. **Monthly Customer Acquisition:** Define new customers acquired each month. **Retention Curve:** Model customer retention over time. **Typical SaaS Retention:** - Month 1: 100% - Month 3: 90% - Month 6: 85% - Month 12: 75% - Month 24: 70% **Revenue Calculation:** For each cohort, calculate retained customers Γ— ARPU for each month. ### Step 3: Model Cost Structure Break down costs by category and behavior. **Fixed vs. Variable:** - Fixed: Salaries, software, rent - Variable: Hosting, payment processing, support **Scaling Assumptions:** - COGS as % of revenue - S&M as % of revenue (CAC payback) - R&D growth rate - G&A as % of total expenses ### Step 4: Create Hiring Plan Model headcount growth by role and department. **Inputs:** - Starting headcount - Hiring velocity by role - Fully-loaded compensation by role - Benefits and taxes (typically 1.3-1.4x salary) **Example:** ``` Engineer: $150K salary Γ— 1.35 = $202K fully-loaded Sales Rep: $100K OTE Γ— 1.30 = $130K fully-loaded ``` ### Step 5: Project Cash Flow Calculate monthly cash position and runway. **Monthly Cash Flow:** ``` Beginning Cash + Revenue Collected (consider payment terms) - Operating Expenses Paid - CapEx = Ending Cash ``` **Runway Calculation:** ``` If Ending Cash < 0: Funding Need = Negative Cash Balance Runway = 0 Else: Runway = Ending Cash / Average Monthly Burn ``` ### Step 6: Calculate Key Metrics Track metrics that matter for stage. **Revenue Metrics:** - MRR / ARR - Growth rate (MoM, YoY) - Revenue by segment or cohort **Unit Economics:** - CAC (Customer Acquisition Cost) - LTV (Lifetime Value) - CAC Payback Period - LTV / CAC Ratio **Efficiency Metrics:** - Burn multiple (Net Burn / Net New ARR) - Magic number (Net New ARR / S&M Spend) - Rule of 40 (Growth % + Profit Margin %) **Cash Metrics:** - Monthly burn rate - Runway (months) - Cash efficiency ### Step 7: Scenario Analysis Create three scenarios with different assumptions. **Variable Assumptions:** - Customer acquisition rate (Β±30%) - Churn rate (Β±20%) - Average contract value (Β±15%) - CAC (Β±25%) **Fixed Assumptions:** - Pricing structure - Core operating expenses - Hiring plan (adjust timing, not roles) ## Business Model Templates ### SaaS Financial Model **Revenue Drivers:** - New MRR (customers Γ— ARPU) - Expansion MRR (upsells) - Contraction MRR (downgrades) - Churned MRR (lost customers) **Key Ratios:** - Gross margin: 75-85% - S&M as % revenue: 40-60% (early stage) - CAC payback: < 12 months - Net retention: 100-120% **Example Projection:** ``` Year 1: $500K ARR, 50 customers, $100K MRR by Dec Year 2: $2.5M ARR, 200 customers, $208K MRR by Dec Year 3: $8M ARR, 600 customers, $667K MRR by Dec ``` ### Marketplace Financial Model **Revenue Drivers:** - GMV (Gross Merchandise Value) - Take rate (% of GMV) - Net revenue = GMV Γ— Take rate **Key Ratios:** - Take rate: 10-30% depending on category - CAC for buyers vs. sellers - Contribution margin: 60-70% **Example Projection:** ``` Year 1: $5M GMV, 15% take rate = $750K revenue Year 2: $20M GMV, 15% take rate = $3M revenue Year 3: $60M GMV, 15% take rate = $9M revenue ``` ### E-Commerce Financial Model **Revenue Drivers:** - Traffic (visitors) - Conversion rate - Average order value (AOV) - Purchase frequency **Key Ratios:** - Gross margin: 40-60% - Contribution margin: 20-35% - CAC payback: 3-6 months ### Services / Agency Financial Model **Revenue Drivers:** - Billable hours or projects - Hourly rate or project fee - Utilization rate - Team capacity **Key Ratios:** - Gross margin: 50-70% - Utilization: 70-85% - Revenue per employee ## Fundraising Integration ### Funding Scenario Modeling **Pre-Money Valuation:** Based on metrics and comparables. **Dilution:** ``` Post-Money = Pre-Money + Investment Dilution % = Investment / Post-Money ``` **Use of Funds:** Allocate funding to extend runway and achieve milestones. **Example:** ``` Raise: $5M at $20M pre-money Post-Money: $25M Dilution: 20% Use of Funds: - Product Development: $2M (40%) - Sales & Marketing: $2M (40%) - G&A and Operations: $0.5M (10%) - Working Capital: $0.5M (10%) ``` ### Milestone-Based Planning **Identify Key Milestones:** - Product launch - First $1M ARR - Break-even on CAC - Series A fundraise **Funding Amount:** Ensure runway to achieve next milestone + 6 months buffer. ## Common Pitfalls **Pitfall 1: Overly Optimistic Revenue** - New startups rarely hit aggressive projections - Use conservative customer acquisition assumptions - Model realistic churn rates **Pitfall 2: Underestimating Costs** - Add 20% buffer to expense estimates - Include fully-loaded compensation - Account for software and tools **Pitfall 3: Ignoring Cash Flow Timing** - Revenue β‰  cash (payment terms) - Expenses paid before revenue collected - Model cash conversion carefully **Pitfall 4: Static Headcount** - Hiring takes time (3-6 months to fill roles) - Ramp time for productivity (3-6 months) - Account for attrition (10-15% annually) **Pitfall 5: Not Scenario Planning** - Single scenario is never accurate - Always model conservative case - Plan for what you'll do if base case fails ## Model Validation **Sanity Checks:** - [ ] Revenue growth rate is achievable (3x in Year 2, 2x in Year 3) - [ ] Unit economics are realistic (LTV/CAC > 3, payback < 18 months) - [ ] Burn multiple is reasonable (< 2.0 in Year 2-3) - [ ] Headcount scales with revenue (revenue per employee growing) - [ ] Gross margin is appropriate for business model - [ ] S&M spending aligns with CAC and growth targets **Benchmark Against Peers:** Compare key metrics to similar companies at similar stage. **Investor Feedback:** Share model with advisors or investors for feedback on assumptions. ## Additional Resources ### Reference Files For detailed model structures and advanced techniques: - **`references/model-templates.md`** - Complete financial model templates by business model - **`references/unit-economics.md`** - Deep dive on CAC, LTV, payback, and efficiency metrics - **`references/fundraising-scenarios.md`** - Modeling funding rounds and dilution ### Example Files Working financial models with formulas: - **`examples/saas-financial-model.md`** - Complete 3-year SaaS model with cohort analysis - **`examples/marketplace-model.md`** - Marketplace GMV and take rate projections - **`examples/scenario-analysis.md`** - Three-scenario framework with sensitivities ## Quick Start To create a startup financial model: 1. **Define business model** - Revenue drivers and pricing 2. **Project revenue** - Cohort-based with retention 3. **Model costs** - COGS, S&M, R&D, G&A by month 4. **Plan headcount** - Hiring by role and department 5. **Calculate cash flow** - Revenue - expenses = burn/runway 6. **Compute metrics** - CAC, LTV, burn multiple, runway 7. **Create scenarios** - Conservative, base, optimistic 8. **Validate assumptions** - Sanity check and benchmark 9. **Integrate fundraising** - Model funding rounds and milestones For complete templates and formulas, reference the `references/` and `examples/` files.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

code-review-excellence

Master effective code review practices to provide constructive

coding
⭐1
# Code Review Excellence Transform code reviews from gatekeeping to knowledge sharing through constructive feedback, systematic analysis, and collaborative improvement. ## When to Use This Skill - Reviewing pull requests and code changes - Establishing code review standards for teams - Mentoring junior developers through reviews - Conducting architecture reviews - Creating review checklists and guidelines - Improving team collaboration - Reducing code review cycle time - Maintaining code quality standards ## Core Principles ### 1. The Review Mindset **Goals of Code Review:** - Catch bugs and edge cases - Ensure code maintainability - Share knowledge across team - Enforce coding standards - Improve design and architecture - Build team culture **Not the Goals:** - Show off knowledge - Nitpick formatting (use linters) - Block progress unnecessarily - Rewrite to your preference ### 2. Effective Feedback **Good Feedback is:** - Specific and actionable - Educational, not judgmental - Focused on the code, not the person - Balanced (praise good work too) - Prioritized (critical vs nice-to-have) ```markdown ❌ Bad: "This is wrong." βœ… Good: "This could cause a race condition when multiple users access simultaneously. Consider using a mutex here." ❌ Bad: "Why didn't you use X pattern?" βœ… Good: "Have you considered the Repository pattern? It would make this easier to test. Here's an example: [link]" ❌ Bad: "Rename this variable." βœ… Good: "[nit] Consider `userCount` instead of `uc` for clarity. Not blocking if you prefer to keep it." ``` ### 3. Review Scope **What to Review:** - Logic correctness and edge cases - Security vulnerabilities - Performance implications - Test coverage and quality - Error handling - Documentation and comments - API design and naming - Architectural fit **What Not to Review Manually:** - Code formatting (use Prettier, Black, etc.) - Import organization - Linting violations - Simple typos ## Review Process ### Phase 1: Context Gathering (2-3 minutes) ```markdown Before diving into code, understand: 1. Read PR description and linked issue 2. Check PR size (>400 lines? Ask to split) 3. Review CI/CD status (tests passing?) 4. Understand the business requirement 5. Note any relevant architectural decisions ``` ### Phase 2: High-Level Review (5-10 minutes) ```markdown 1. **Architecture & Design** - Does the solution fit the problem? - Are there simpler approaches? - Is it consistent with existing patterns? - Will it scale? 2. **File Organization** - Are new files in the right places? - Is code grouped logically? - Are there duplicate files? 3. **Testing Strategy** - Are there tests? - Do tests cover edge cases? - Are tests readable? ``` ### Phase 3: Line-by-Line Review (10-20 minutes) ```markdown For each file: 1. **Logic & Correctness** - Edge cases handled? - Off-by-one errors? - Null/undefined checks? - Race conditions? 2. **Security** - Input validation? - SQL injection risks? - XSS vulnerabilities? - Sensitive data exposure? 3. **Performance** - N+1 queries? - Unnecessary loops? - Memory leaks? - Blocking operations? 4. **Maintainability** - Clear variable names? - Functions doing one thing? - Complex code commented? - Magic numbers extracted? ``` ### Phase 4: Summary & Decision (2-3 minutes) ```markdown 1. Summarize key concerns 2. Highlight what you liked 3. Make clear decision: - βœ… Approve - πŸ’¬ Comment (minor suggestions) - πŸ”„ Request Changes (must address) 4. Offer to pair if complex ``` ## Review Techniques ### Technique 1: The Checklist Method ```markdown ## Security Checklist - [ ] User input validated and sanitized - [ ] SQL queries use parameterization - [ ] Authentication/authorization checked - [ ] Secrets not hardcoded - [ ] Error messages don't leak info ## Performance Checklist - [ ] No N+1 queries - [ ] Database queries indexed - [ ] Large lists paginated - [ ] Expensive operations cached - [ ] No blocking I/O in hot paths ## Testing Checklist - [ ] Happy path tested - [ ] Edge cases covered - [ ] Error cases tested - [ ] Test names are descriptive - [ ] Tests are deterministic ``` ### Technique 2: The Question Approach Instead of stating problems, ask questions to encourage thinking: ```markdown ❌ "This will fail if the list is empty." βœ… "What happens if `items` is an empty array?" ❌ "You need error handling here." βœ… "How should this behave if the API call fails?" ❌ "This is inefficient." βœ… "I see this loops through all users. Have we considered the performance impact with 100k users?" ``` ### Technique 3: Suggest, Don't Command ````markdown ## Use Collaborative Language ❌ "You must change this to use async/await" βœ… "Suggestion: async/await might make this more readable: `typescript async function fetchUser(id: string) { const user = await db.query('SELECT * FROM users WHERE id = ?', id); return user; } ` What do you think?" ❌ "Extract this into a function" βœ… "This logic appears in 3 places. Would it make sense to extract it into a shared utility function?" ```` ### Technique 4: Differentiate Severity ```markdown Use labels to indicate priority: πŸ”΄ [blocking] - Must fix before merge 🟑 [important] - Should fix, discuss if disagree 🟒 [nit] - Nice to have, not blocking πŸ’‘ [suggestion] - Alternative approach to consider πŸ“š [learning] - Educational comment, no action needed πŸŽ‰ [praise] - Good work, keep it up! Example: "πŸ”΄ [blocking] This SQL query is vulnerable to injection. Please use parameterized queries." "🟒 [nit] Consider renaming `data` to `userData` for clarity." "πŸŽ‰ [praise] Excellent test coverage! This will catch edge cases." ``` ## Language-Specific Patterns ### Python Code Review ```python # Check for Python-specific issues # ❌ Mutable default arguments def add_item(item, items=[]): # Bug! Shared across calls items.append(item) return items # βœ… Use None as default def add_item(item, items=None): if items is None: items = [] items.append(item) return items # ❌ Catching too broad try: result = risky_operation() except: # Catches everything, even KeyboardInterrupt! pass # βœ… Catch specific exceptions try: result = risky_operation() except ValueError as e: logger.error(f"Invalid value: {e}") raise # ❌ Using mutable class attributes class User: permissions = [] # Shared across all instances! # βœ… Initialize in __init__ class User: def __init__(self): self.permissions = [] ``` ### TypeScript/JavaScript Code Review ```typescript // Check for TypeScript-specific issues // ❌ Using any defeats type safety function processData(data: any) { // Avoid any return data.value; } // βœ… Use proper types interface DataPayload { value: string; } function processData(data: DataPayload) { return data.value; } // ❌ Not handling async errors async function fetchUser(id: string) { const response = await fetch(`/api/users/${id}`); return response.json(); // What if network fails? } // βœ… Handle errors properly async function fetchUser(id: string): Promise<User> { try { const response = await fetch(`/api/users/${id}`); if (!response.ok) { throw new Error(`HTTP ${response.status}`); } return await response.json(); } catch (error) { console.error('Failed to fetch user:', error); throw error; } } // ❌ Mutation of props function UserProfile({ user }: Props) { user.lastViewed = new Date(); // Mutating prop! return <div>{user.name}</div>; } // βœ… Don't mutate props function UserProfile({ user, onView }: Props) { useEffect(() => { onView(user.id); // Notify parent to update }, [user.id]); return <div>{user.name}</div>; } ``` ## Advanced Review Patterns ### Pattern 1: Architectural Review ```markdown When reviewing significant changes: 1. **Design Document First** - For large features, request design doc before code - Review design with team before implementation - Agree on approach to avoid rework 2. **Review in Stages** - First PR: Core abstractions and interfaces - Second PR: Implementation - Third PR: Integration and tests - Easier to review, faster to iterate 3. **Consider Alternatives** - "Have we considered using [pattern/library]?" - "What's the tradeoff vs. the simpler approach?" - "How will this evolve as requirements change?" ``` ### Pattern 2: Test Quality Review ```typescript // ❌ Poor test: Implementation detail testing test('increments counter variable', () => { const component = render(<Counter />); const button = component.getByRole('button'); fireEvent.click(button); expect(component.state.counter).toBe(1); // Testing internal state }); // βœ… Good test: Behavior testing test('displays incremented count when clicked', () => { render(<Counter />); const button = screen.getByRole('button', { name: /increment/i }); fireEvent.click(button); expect(screen.getByText('Count: 1')).toBeInTheDocument(); }); // Review questions for tests: // - Do tests describe behavior, not implementation? // - Are test names clear and descriptive? // - Do tests cover edge cases? // - Are tests independent (no shared state)? // - Can tests run in any order? ``` ### Pattern 3: Security Review ```markdown ## Security Review Checklist ### Authentication & Authorization - [ ] Is authentication required where needed? - [ ] Are authorization checks before every action? - [ ] Is JWT validation proper (signature, expiry)? - [ ] Are API keys/secrets properly secured? ### Input Validation - [ ] All user inputs validated? - [ ] File uploads restricted (size, type)? - [ ] SQL queries parameterized? - [ ] XSS protection (escape output)? ### Data Protection - [ ] Passwords hashed (bcrypt/argon2)? - [ ] Sensitive data encrypted at rest? - [ ] HTTPS enforced for sensitive data? - [ ] PII handled according to regulations? ### Common Vulnerabilities - [ ] No eval() or similar dynamic execution? - [ ] No hardcoded secrets? - [ ] CSRF protection for state-changing operations? - [ ] Rate limiting on public endpoints? ``` ## Giving Difficult Feedback ### Pattern: The Sandwich Method (Modified) ```markdown Traditional: Praise + Criticism + Praise (feels fake) Better: Context + Specific Issue + Helpful Solution Example: "I noticed the payment processing logic is inline in the controller. This makes it harder to test and reuse. [Specific Issue] The calculateTotal() function mixes tax calculation, discount logic, and database queries, making it difficult to unit test and reason about. [Helpful Solution] Could we extract this into a PaymentService class? That would make it testable and reusable. I can pair with you on this if helpful." ``` ### Handling Disagreements ```markdown When author disagrees with your feedback: 1. **Seek to Understand** "Help me understand your approach. What led you to choose this pattern?" 2. **Acknowledge Valid Points** "That's a good point about X. I hadn't considered that." 3. **Provide Data** "I'm concerned about performance. Can we add a benchmark to validate the approach?" 4. **Escalate if Needed** "Let's get [architect/senior dev] to weigh in on this." 5. **Know When to Let Go** If it's working and not a critical issue, approve it. Perfection is the enemy of progress. ``` ## Best Practices 1. **Review Promptly**: Within 24 hours, ideally same day 2. **Limit PR Size**: 200-400 lines max for effective review 3. **Review in Time Blocks**: 60 minutes max, take breaks 4. **Use Review Tools**: GitHub, GitLab, or dedicated tools 5. **Automate What You Can**: Linters, formatters, security scans 6. **Build Rapport**: Emoji, praise, and empathy matter 7. **Be Available**: Offer to pair on complex issues 8. **Learn from Others**: Review others' review comments ## Common Pitfalls - **Perfectionism**: Blocking PRs for minor style preferences - **Scope Creep**: "While you're at it, can you also..." - **Inconsistency**: Different standards for different people - **Delayed Reviews**: Letting PRs sit for days - **Ghosting**: Requesting changes then disappearing - **Rubber Stamping**: Approving without actually reviewing - **Bike Shedding**: Debating trivial details extensively ## Templates ### PR Review Comment Template ```markdown ## Summary [Brief overview of what was reviewed] ## Strengths - [What was done well] - [Good patterns or approaches] ## Required Changes πŸ”΄ [Blocking issue 1] πŸ”΄ [Blocking issue 2] ## Suggestions πŸ’‘ [Improvement 1] πŸ’‘ [Improvement 2] ## Questions ❓ [Clarification needed on X] ❓ [Alternative approach consideration] ## Verdict βœ… Approve after addressing required changes ``` ## Resources - **references/code-review-best-practices.md**: Comprehensive review guidelines - **references/common-bugs-checklist.md**: Language-specific bugs to watch for - **references/security-review-guide.md**: Security-focused review checklist - **assets/pr-review-template.md**: Standard review comment template - **assets/review-checklist.md**: Quick reference checklist - **scripts/pr-analyzer.py**: Analyze PR complexity and suggest reviewers
πŸ‘0
πŸ‘οΈ3
πŸ€– Auto-discovered