🤖 Gen AI & Copilot Cheat Sheet

# Bad:
"Write a function"

# Good:
"Write a Python function called `top_performers`
that takes a list[dict] with 'name' (str) and
'score' (float) keys, returns the top N items
sorted by score descending. Include type hints
and a docstring."

ROLE: You are a senior data engineer specializing
      in Snowflake and dbt.

CONTEXT: Our pipeline runs hourly via Airflow.
  Source: S3 Parquet files (100GB/day)
  Target: Snowflake analytics schema
  Stack: Python 3.11, Snowpark, dbt

TASK: Review this Snowflake SQL for performance.
  Must handle 500M+ rows efficiently.

OUTPUT: Bullet list with severity (P0/P1/P2)
  and specific fix for each issue.

Convert natural language to SQL.

Example 1:
  Input:  "total sales by region last quarter"
  Output: SELECT region, SUM(amount) AS total
    FROM sales
    WHERE sale_date >= DATE_TRUNC('quarter',
      CURRENT_DATE - INTERVAL '3 month')
    GROUP BY region;

Example 2:
  Input:  "customers who ordered more than 5 times"
  Output: SELECT customer_id, COUNT(*) AS cnt
    FROM orders GROUP BY customer_id
    HAVING COUNT(*) > 5;

Now convert:
  Input: "average order value by month for 2024"

Write a Python data pipeline function.

DO:
- Use type hints on all parameters and return
- Add structured logging with timestamps
- Include retry logic with exponential backoff
- Handle edge cases (empty data, network errors)

DO NOT:
- Use print statements (use logging module)
- Hardcode credentials (use env vars)
- Catch bare Exception without re-raising
- Use mutable default arguments

Solve this step by step:

1. Identify the tables and their relationships
2. Determine which JOIN type is appropriate
3. Apply the WHERE filters
4. Add GROUP BY and aggregation
5. Apply HAVING to filter groups
6. Add ORDER BY and LIMIT
7. Write the complete SQL query

Question: Find the top 3 departments where
the average salary of employees hired in the
last 2 years exceeds the company-wide average.

After generating the solution, verify it by:

1. Walk through with this sample data:
   employees: [(1,'Alice','Eng',90000),
               (2,'Bob','Sales',70000)]

2. Check edge cases:
   - Empty table
   - All same department
   - NULL salary values

3. Verify the output matches expected results

4. If any check fails, fix the solution
   before presenting it.

I need to design a data pipeline for real-time
user event processing.

Explore 3 different approaches:

Approach A: Kafka → Spark Streaming → Snowflake
Approach B: Kafka → Flink → Delta Lake
Approach C: Kinesis → Lambda → DynamoDB

For each approach, evaluate:
- Latency (ms)
- Cost at 1M events/day
- Operational complexity
- Team skill requirements

Recommend the best approach with justification.

Round 1: "Write a Python ETL pipeline that
  reads from S3 and loads into Snowflake"

Round 2: "Add error handling with retry logic
  and dead letter queue for failed records"

Round 3: "Add structured JSON logging with
  correlation IDs and execution metrics"

Round 4: "Add unit tests with pytest using
  mocked S3 and Snowflake connections"

Round 5: "Add a Dockerfile and docker-compose
  for local development"

Tab            → Accept suggestion
Esc            → Dismiss suggestion
Alt + ]        → Next suggestion
Alt + [        → Previous suggestion
Ctrl + Enter   → Open completions panel
Alt + \        → Trigger inline suggestion

# Accept partial suggestion:
Ctrl + →       → Accept next word
Cmd + →        → Accept next word (Mac)

/explain       → Explain selected code
/fix           → Fix bugs in selection
/tests         → Generate unit tests
/doc           → Generate documentation
/optimize      → Suggest performance improvements
/new           → Scaffold new project/file
/clear         → Clear chat history

@workspace     → Ask about entire codebase
@terminal      → Ask about terminal output
@vscode        → Ask about VS Code settings
#file:name.py  → Reference specific file
#selection     → Reference selected code

# Enable Copilot in terminal
Ctrl + I           → Open Copilot in terminal

# Natural language to commands:
"find all python files larger than 1MB"
→ find . -name '*.py' -size +1M

"show git commits from last week"
→ git log --since='1 week ago' --oneline

"count lines of code in src directory"
→ find src -name '*.py' | xargs wc -l

# Function to calculate the exponential moving
# average of a time series with a configurable
# span parameter. Returns a list of same length
# as input with None for insufficient data points.
def ema(data: list[float], span: int) -> list[float | None]:
    # Copilot generates implementation from comments
    pass

# Tip: Write detailed function docstrings FIRST,
# then let Copilot implement the body.
# The more specific your docstring, the better
# the generated code.

# Write tests FIRST, then ask AI to implement

def test_merge_intervals():
    assert merge([[1,3],[2,6],[8,10]]) == [[1,6],[8,10]]
    assert merge([[1,4],[4,5]]) == [[1,5]]
    assert merge([]) == []
    assert merge([[1,1]]) == [[1,1]]

# Now prompt: "Implement the merge() function
# that passes all the above test cases."
# AI has clear spec from the tests.

Review this code for:

1. BUGS: Logic errors, off-by-one, null handling
2. SECURITY: SQL injection, XSS, credential leaks
3. PERFORMANCE: N+1 queries, unnecessary loops,
   missing indexes
4. READABILITY: Naming, complexity, dead code
5. BEST PRACTICES: Error handling, logging,
   type safety

For each issue found, provide:
- Severity: Critical / Warning / Info
- Line number
- Current code
- Suggested fix
- Explanation

from openai import OpenAI
client = OpenAI()  # Uses OPENAI_API_KEY env var

response = client.chat.completions.create(
    model='gpt-4o',
    messages=[
        {'role': 'system', 'content': 'You are a helpful assistant.'},
        {'role': 'user', 'content': 'Explain Snowflake streams.'}
    ],
    temperature=0.3,    # Lower = more deterministic
    max_tokens=1000,
    top_p=0.9,
)

answer = response.choices[0].message.content
print(answer)

# Streaming response
stream = client.chat.completions.create(
    model='gpt-4o',
    messages=[{'role': 'user', 'content': prompt}],
    stream=True,
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end='')

# Structured output (JSON mode)
response = client.chat.completions.create(
    model='gpt-4o',
    response_format={'type': 'json_object'},
    messages=[{
        'role': 'user',
        'content': 'Return a JSON with keys: summary, tags, score'
    }]
)

temperature    0.0 = deterministic, 1.0 = creative
top_p          0.1 = narrow focus, 1.0 = all tokens
max_tokens     Maximum response length
stop           Stop sequences ['\n', '###']
frequency_penalty  0-2, reduce repetition
presence_penalty   0-2, encourage new topics

# Recommendations:
# Code generation:     temp=0.2, top_p=0.9
# Creative writing:    temp=0.8, top_p=0.95
# Data extraction:     temp=0.0, top_p=1.0
# Brainstorming:       temp=1.0, top_p=0.95

# Retrieval-Augmented Generation

1. INDEXING (offline):
   Documents → Chunk (500 tokens) → Embed → Vector DB

2. RETRIEVAL (runtime):
   User Query → Embed → Vector Search → Top-K chunks

3. GENERATION:
   System: "Answer using ONLY the provided context."
   Context: [retrieved chunks]
   User: [original question]
   → LLM generates grounded answer

# Key decisions:
# - Chunk size: 256-1024 tokens
# - Overlap: 10-20% between chunks
# - Top-K: 3-10 most relevant
# - Embedding model: text-embedding-3-small/large

from openai import OpenAI
client = OpenAI()

# Generate embedding
response = client.embeddings.create(
    model='text-embedding-3-small',
    input='How do Snowflake streams work?'
)
vector = response.data[0].embedding  # 1536-dim

# Cosine similarity
import numpy as np
def cosine_sim(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Batch embeddings
texts = ['doc1 content', 'doc2 content', ...]
response = client.embeddings.create(
    model='text-embedding-3-small',
    input=texts
)

# Guard against prompt injection:

1. Separate system instructions from user input
2. Never put user input directly in system prompt
3. Validate and sanitize user input
4. Use output validation / content filters
5. Set clear boundaries in system prompt:

   System: "You are a SQL assistant. ONLY generate
   SELECT queries. Never generate DROP, DELETE,
   UPDATE, or any DDL statements. If asked to do
   anything outside SQL queries, politely decline."

# Reduce API costs:

1. Use smaller models for simple tasks
   gpt-4o-mini for classification/extraction
   gpt-4o for complex reasoning

2. Cache responses for identical queries
   from functools import lru_cache

3. Batch requests when possible

4. Set appropriate max_tokens
   Don't use 4096 when 200 suffices

5. Use streaming for long responses
   (faster perceived latency)

6. Compress prompts: remove fluff,
   use abbreviations in system prompts

7. Fine-tune for repetitive tasks
   (cheaper per-token after training)

# Test AI outputs systematically:

1. DETERMINISTIC TESTS (temp=0)
   - Exact match for structured outputs
   - JSON schema validation
   - Code syntax verification

2. SEMANTIC TESTS
   - Key facts are present
   - No hallucinated information
   - Correct format/structure

3. SAFETY TESTS
   - No harmful content
   - No leaked training data
   - Handles adversarial inputs

4. REGRESSION TESTS
   - Golden dataset of Q&A pairs
   - Run after model/prompt changes
   - Track quality scores over time