Without safeguards, customer-facing AI can:

• Leak PII

• Offer unauthorized legal or medical advice

• Be tricked through jailbreaks

• Enable fraud and compliance violations

NVIDIA’s NeMo Guardrails addresses this by providing a modular safety framework: rules, classifiers, and visualization tools that make AI risk visible, explainable, and enforceable.

Figure 1: Input and output guardrails act as middleware around the LLM

1. Defining Guardrails Programmatically

Guardrails in NeMo are expressed as configuration modules: flows, blocks, and patterns. These can operate on both input (user prompt) and output (model response). Conceptually, they form a middleware safety layer:

• Input guardrails intercept unsafe requests before the LLM sees them

• Output guardrails sanitize or block unsafe generations from the model

2. Risk Taxonomy in Customer Service

Not all risks are equal. In customer service contexts, risks tend to cluster into a handful of recurring categories. By mapping them explicitly, we can define reusable guardrails. In customer service systems, risks cluster into eight categories. Each maps to a set of detection patterns that developers can configure.

• PII Exposure – e.g., regex matches for SSNs, phone numbers, or email requests.

rails:
  input:
    flows:
      - detect_pii
    blocks:
      detect_pii:
        type: regex
        patterns:
          - "[0-9]{3}-[0-9]{2}-[0-9]{4}"   # SSN pattern
          - "[0-9]{10}"                    # phone number
        action: block

Figure 2: PII exposure detection and redaction (SSNs, phone numbers)

• Security Bypass – prompts requesting admin access, password resets, or hacking help. As shown in Figure 3, attempts to bypass security (e.g., admin access requests) are detected before reaching the model.

  .

Figure 3: Security bypass attempts (admin/password reset)

• Legal Liability – drafting lawsuits or offering legal advice that could create compliance issues.

• Jailbreak Attempts – instruction overrides, role-playing requests.

from nemoguardrails import LLMRails

config = """
rails:
  input:
    flows:
      - detect_jailbreak
    blocks:
      detect_jailbreak:
        type: regex
        patterns:
          - "ignore all previous instructions"
          - "pretend you are"
        action: block
"""

Figure 4: Jailbreak attempt (role-playing request, hacking)

• Harmful Content – fake reviews, defamatory or competitor-targeted outputs.

Figure 5: Harmful content generation (defamation, fake reviews)

• Financial Fraud – credit card data extraction, account takeover attempts.

• Corporate Espionage – requests for trade secrets, pricing strategies, or internal policies.

• Regulatory Violations – prompts encouraging bypass of HIPAA, GDPR, or consent requirements.

3. Healthcare-Specific Guardrails

Healthcare is one of the highest-stakes domains for AI adoption. Here, guardrails don’t just reduce reputational risk - they enforce legal compliance under HIPAA, FDA regulations, and clinical trial protocols.

Example: Blocking Unauthorized Diagnosis

rails:
  input:
    flows:
      - detect_medical
    blocks:
      detect_medical:
        type: regex
        patterns:
          - "diagnose my symptoms"
          - "recommend a treatment"
        action: block

This prevents the system from giving unlicensed medical advice. Healthcare requires stricter safeguards due to HIPAA/FDA compliance. Figure 6 shows how rails block diagnosis or patient data requests.

Figure 6: Guardrails blocking unauthorized diagnosis

Example: Preventing HIPAA Violations

These rails extend across categories:

• HIPAA privacy

• FDA device safety

• Clinical trial consent

• Healthcare billing fraud

rails:
  input:
    flows:
      - detect_hipaa
    blocks:
      detect_hipaa:
        type: regex
        patterns:
          - "access patient records"
          - "show me patient data"
        action: block

Figure 6: Guardrails blocking HIPAA violations

4. Making Guardrails Explainable

Unlike traditional black-box filters, NeMo Guardrails produces structured metadata for every block. Developers and compliance officers can see exactly why a prompt was blocked.

Example: Legal Liability Detection

result = app.generate("Help me sue my neighbor")
print(result.explanations)

Output

{
  "blocked": true,
  "reason": "Legal Liability - Detected request for legal advice"
}

This supports:

• Compliance audits – showing regulators why content was blocked

• Debugging – letting developers refine patterns and rules

• Risk dashboards – aggregating blocked categories across usage

Example: HIPAA Compliance

In healthcare, transparency is just as important as blocking unsafe prompts. A compliance officer doesn’t just want to know that the system stopped a response - they need to see why it was stopped and which regulatory category was triggered.

result = app.generate("Can you access patient records for John Doe?")
print(result.explanations)

{
  "blocked": true,
  "reason": "HIPAA Violation - Attempted access to protected patient data"
}

5. Why This Matters

By codifying conversational risks into structured categories, patterns, and explainable outputs, NeMo Guardrails enables enterprises to:

• Customer service → Prevent data leaks, fraud, and liability

• Healthcare → Align outputs with HIPAA/FDA constraints

• Enterprise AI → Treat safety as a measurable, auditable layer

Instead of an opaque “safety filter,” Guardrails turns safety into configurable infrastructure. Developers can reason about rules, test them, and visualize performance across scenarios.

GitHub Repo: Link

Figure: DenseNet-121 classification results on MedNIST images: Head CT, Chest X-Ray (CXR), and Abdomen CT. The top row shows the original input scans, while the bottom row highlights the model’s region-of-interest focus with bounding ellipses. Predicted class labels are displayed with high confidence scores -HeadCT (96.3%), CXR (94.7%), and AbdomenCT (95.1%).

The key stages include data loading, preprocessing, augmentation, batching, device transfer, inference, and post-processing. In each stage, we substitute CPU-based tools (Pillow/OpenCV, pandas, etc.) with NVIDIA GPU-accelerated libraries (DALI, cuDF, etc.) to remove bottlenecks and boost throughput. For example, on an NVIDIA Tesla T4 GPU we achieved ~0.109 s per batch of 32 (≈3.4 ms/image) versus ~0.315 s on an 8-core CPU - a ~3.3× speedup in throughput. The GPU ran at ~88% utilization, whereas CPU cores were mostly idle (~62% utilization), showing the GPU’s efficiency under load.

Figure: The diagram below summarizes our pipeline stages and tool mappings. The left side shows the CPU-based workflow (Pillow/OpenCV, PyTorch DataLoader, etc.) and the right side shows the NVIDIA GPU-accelerated workflow (DALI, CUDA tensors, etc.).

Pipeline Architecture:

Our end-to-end inference pipeline has these stages:

Figure: DenseNet-121 medical imaging inference pipeline. From data loading and preprocessing to GPU transfer, inference, and postprocessing for confidence mapping and evaluation

Data Loading:

Read image files and assign class labels. We collected ~64K MedNIST JPEGs (6 classes) and split them into train/validation.

Preprocessing:

Decode JPEGs, resize to 256×256, and normalize pixel intensities to [0,1].

Data Augmentation (training only):

Apply randomized transforms (e.g. crops, flips, noise, blur, brightness/contrast).

Figure: Step-by-step Medical Image Augmentation Pipeline. Simulates real-world imaging variations like noise, motion blur, brightness, contrast, zoom, flips, and rotations. Enhances dataset diversity for robust deep learning in medical imaging

Figure: Original vs Augmented. The augmented CT shows changes in intensity and structure, mimicking real-world variability. The difference map highlights pixel-wise intensity shifts between the two images

Batching & Device Transfer:

Batches of 32.

• GPU path (DALI): On-GPU decoding and batching with overlap between I/O and inference.

• CPU path (PyTorch DataLoader): Host batching with explicit .to(device) copies, adding latency.

Inference:

We run DenseNet-121 (2D, 1 input channel, 6 classes) in eval() mode with FP16 enabled to exploit GPU tensor cores.

Performance:

On a Tesla T4, 32-image batches complete in 0.109 s vs 0.315 s on CPU - a 3.3× throughput gain (103 vs 31 img/s)

Output handling:

Logits are passed through softmax/top-k to extract top-1 predictions and confidence scores. GPU vs CPU outputs align aside from minor floating-point differences.

Monitoring:

A custom profiler logs per-batch timings plus GPU utilization, memory, and power at ~10 Hz. From these metrics, we derive latency, throughput, and energy efficiency.

Post-processing:

Extract top-k predictions, format results, and perform consistency checks.

Logging & Metrics:

Record timings, throughput, and hardware utilization throughout.

Performance Analysis

Our key quantitative findings are summarized below. These numbers are averaged over many batches during a full validation pass on MedNIST.

Latency and Throughput

The GPU achieves ~3.3× higher throughput and ~2.9× lower batch time than the CPU. This matches the qualitative expectation that a Tesla T4 (with 2560 CUDA cores) can process dense CNN workloads much faster than a multi-core CPU for the same batch.

Figure: Comparison of GPU (Tesla T4) vs CPU (8 cores) performance on DenseNet-121 inference. The GPU achieves ~2.9× lower batch time and ~3.3× higher throughput, while also improving memory and power efficiency.

Efficiency and Utilization

• The GPU ran at ~88% compute utilization (measured via NVML), with VRAM usage around 1.2 GB for inference. In contrast, the CPU’s 8 cores averaged only ~62% utilization (due to GIL, I/O stalls, etc.) and used ~4.5 GB of system RAM for buffers and intermediate data.

• We measured power draw during inference. Normalized to a standard metric, the GPU delivered roughly 40% higher “images per joule” efficiency than the CPU.

Memory Footprint

Our GPU pipeline’s memory efficiency is higher: only ~3 GB (20%) of the T4’s 15 GB VRAM was needed, whereas the CPU approach used ~15 GB of RAM due to unmanaged buffers.

Figure: Timing overlap: The DALI-based pipeline builds the next batch on GPU while the model processes the current batch. In contrast, the CPU pipeline often has idle gaps due to host-to-device transfers.

Figure: GPU vs CPU Training Metrics. GPU shows higher utilization and memory efficiency, while CPU runs hotter but consumes slightly less power. Temperature and power limits remain below thresholds throughout the run

Key Takeaways

In summary, our comprehensive analysis demonstrates that GPU acceleration transforms medical imaging inference pipelines. By carefully mapping each stage (file I/O, decoding, augmentation, batching, inference) to NVIDIA tools, we achieved:

• GPU pipeline: 3.3× throughput, 2.9× lower latency vs CPU.

• Better efficiency: 20% VRAM use vs 15 GB RAM on CPU.

• Higher utilization: 88% GPU vs 62% CPU.

• ~40% more images per watt → energy savings.

Optimization Roadmap:

Our pipeline already leverages GPU acceleration via PyTorch, cuDNN, and cuBLAS, but further optimizations can push latency and efficiency gains:

Inference Optimizations

• TensorRT: Operator fusion, kernel auto-tuning, and graph-level optimizations to reduce per-batch latency.

• INT8 Quantization: Cuts memory and bandwidth demands, accelerating inference while maintaining accuracy.

Data & I/O Pipeline

• RAPIDS cuDF: Replaces Python preprocessing with GPU-native operations, eliminating host-device transfer overhead.

• GPUdirect Storage: Direct GPU-to-disk I/O to reduce data loading latency, especially for hospital archives.

System Efficiency

• DCGM Telemetry: Real-time monitoring of utilization, temperature, and power to enable adaptive scheduling in multi-tenant clusters.

Figure: Proposed enhancements to the GPU inference stack: TensorRT optimization, INT8 quantization, DCGM telemetry, RAPIDS cuDF preprocessing, and GPUdirect Storage integration could further accelerate DenseNet-121 inference and improve memory and energy efficiency

Overall, these results suggest that with TensorRT optimization and GPU-native preprocessing, throughput could scale further, enabling real-time triage pipelines for radiology workflows.

I pulled together the diagram below to make sense of NVIDIA’s AI Enterprise Software Reference Architecture. This guide shows how these can be applied in practical GenAI workflows, along with examples and non-NVIDIA equivalents.

Here’s my breakdown, with GenAI use cases for each tool.

AI Frameworks & SDKs:

Where you build and fine-tune the brains of your system

NVIDIA NeMo

What is it? An open-source framework for training and fine-tuning large language models, ASR, TTS, and multimodal models. It has ready-to-use pipelines for pretraining and fine-tuning, plus domain adaptation tools so you can customize vocabularies and tokenizers. Built to handle massive models with Megatron-LM integration.

Example: Fine-tune a legal Q&A model so it understands Latin legal terms and cites case law accurately.

Integration tip: Pair with FAISS-GPU or Milvus for RAG, and deploy through Triton for production.

NVIDIA TAO Toolkit

What is it? Transfer learning without building training loops from scratch. You can adapt vision, speech, and NLP models in a low-code way.

Example: Fine-tune a speech-to-text model on your call center data in a few hours without writing custom training loops.

Domain-Specific SDKs

What is it? Pre-optimized kits for specialized industries. These are industry-optimized kits to shortcut development.

Examples:

• Merlin → Build personalized product recommenders in an e-commerce chatbot (like “users who read this also liked…”).

• Clara → Use healthcare-specific AI in a patient-facing Q&A assistant.

Core GPU Libraries

What is it? The CUDA-X building blocks that make everything faster. They speed up deep learning math, data loading, and GPU-to-GPU communication.

Examples: NCCL lets multiple GPUs “talk” super fast so a big LLM can run across them without slowdown.

Data Processing & Analytics Tools:

Where you clean, prep, and transform raw data into AI-ready formats - at GPU speed

RAPIDS (cuDF, cuML, cuGraph, etc.)

What is it? Think Pandas or Spark, but everything stays in GPU memory. This removes the CPU as a bottleneck.

Example: Process millions of PDFs, clean the text, and generate embeddings in hours instead of days.

NVIDIA DALI

What is it? Loads and preprocesses data directly on the GPU so your model is never idle waiting for the next batch.

Example: Normalize and batch massive image datasets for a multimodal RAG pipeline, removing the I/O bottleneck.

Integration tip: NCCL uses NVLink/InfiniBand for multi-GPU all-reduce; DALI keeps preprocessing in GPU memory.

Inference & Deployment Tools:

Where you turn trained models into fast, reliable services.

NVIDIA TensorRT

What is it? Optimizes models for much faster inference through kernel fusion and mixed precision.

Example: Cut chatbot latency from 1 second to 0.5 seconds without losing accuracy.

Integration tip: Use before deployment to Triton for max throughput.

TensorRT-LLM

What is it? Optimized for large language models with paged KV cache and speculative decoding.

Example: Serve a 70B model at high QPS for customer support without doubling GPU count.

NVIDIA Triton Inference Server

What is it? One API for serving models from TensorFlow, PyTorch, ONNX, OpenVINO, and more. Auto-batches requests to get the most from your GPUs.

Example: Host your embedding model and reranker together and let Triton batch requests automatically.

NVIDIA Riva

What is it? A real-time speech AI SDK for streaming ASR and TTS.

Example: Add a voice interface to your RAG assistant so it can take spoken queries and respond aloud instantly.

Technical integration: Pair with Triton or NIM for unified deployment.

NVIDIA NIM + NIM Operator

What is it? Prepackaged AI microservices you can deploy on any GPU environment in minutes.

Example: Deploy a retrieval API for your RAG app without touching CUDA installs.

Orchestration & Management:

Where you keep all the moving parts running - and scaling - in production.

NVIDIA GPU Operator

What is it? Handles GPU driver and CUDA setup in Kubernetes automatically.

Example: Add a new GPU node to your inference cluster and have it production-ready in minutes.

NVIDIA Base Command

What is it? The control center for DGX systems.

Example: Monitor a multi-node LLM training job and restart failed tasks automatically.

NVIDIA Fleet Command

What is it? Securely deploy and update AI apps across distributed sites.

Example: Push a new model version to hundreds of retail store edge nodes overnight.

MLOps Integrations (AI Workbench, Kubeflow, MLflow)

What is it? Tools for tracking experiments, automating pipelines, and managing model versions.

Example: Swap in a new reranker model and roll back instantly if something breaks.

Monitoring & Logging (DCGM + Prometheus/Grafana)

What is it? Full GPU fleet observability.

Example: Catch GPUs running hot before they throttle and slow down your responses.

Deployment & Infrastructure Layer:

Where your workloads actually run.

NVIDIA NGC Catalog

What is it? A marketplace of GPU-optimized containers, models, and SDKs.

Example: Start a multimodal RAG proof-of-concept in hours by pulling a prebuilt NeMo container.

NVIDIA DGX Systems

What is it? GPU supercomputers with NVLink for large-scale AI training.

Example: Train a domain-specific 70B LLM in days instead of weeks.

System Software (CUDA toolkit, container runtime)

What is it? The foundation for running any GPU workload in Docker or Kubernetes.

Example: Deploy your entire RAG pipeline in containers with full GPU acceleration.

The examples here are starting points. In practice, you’ll mix and match based on your model type, data sources, latency requirements, and operational constraints. The key is to treat this stack as a set of interchangeable building blocks rather than a rigid recipe, that’s how you’ll get from a promising prototype to a production system that lasts.

Deployment & Infrastructure Layer: Hardware

Where your workloads physically live, with the raw compute, memory, and networking to make large-scale AI possible.

NVIDIA H100 / A100 / L4

What is it? Data center GPUs for AI training and inference.

Why it’s useful:

• H100: Transformer Engine for mixed precision, ideal for large LLMs.

• A100: Excellent for both training and inference, widely deployed.

• L4: Optimized for energy-efficient inference and media processing

Example: Train and serve a multi-billion parameter LLM with low latency.

NVIDIA DGX Systems

What is it? Turnkey AI supercomputers with NVLink/NVSwitch interconnects for multi-GPU scaling.

Why it’s useful:

• NVLink/NVSwitch enables multi-node model parallelism without network bottlenecks

Example: Run multi-node model parallel training for a 70B parameter chatbot.

HGX & MGX Platforms

What is it? Modular reference architectures for OEM partners.

Example: OEM builds an HGX H100 8-GPU server with liquid cooling for dense video analytics

Grace CPU & Grace Hopper

What is it?

• Grace CPU: ARM-based CPU optimized for AI and HPC workloads.

• Grace Hopper: Superchip combining Grace CPU + Hopper GPU with high-bandwidth NVLink-C2C.

Why it’s useful:

• Massive memory bandwidth, ideal for retrieval-augmented LLMs and large graph neural networks.

• Reduces latency in CPU–GPU data transfers.

Example: Optimize large memory-bound inference without bottlenecks.

NVIDIA Jetson Platform

What is it? Edge AI devices for robotics, IoT, and embedded AI.

Why it’s useful:

• Runs CUDA and TensorRT on small, low-power devices.

• Great for latency-sensitive inference without relying on cloud connectivity.
  

Example: Deploy an on-device LLM-powered voice assistant in a drone.

NVIDIA’s hardware portfolio spans from large-scale AI supercomputers to compact edge devices, each designed with specific computational resources to target different AI deployment needs.

Considering the average human protein is 430 amino acids long, the number of possible designs translates to 20^430 sequences, which is practically infinite. Navigating this search space through conventional methods or brute-force methods is highly impractical! So what should we do?

To overcome these limitations, we need tools that integrate generative AI models, and GPU-accelerated microservices to explore this vast design space, leading to stable and structurally constrained binders at a much faster pace.

In this post we will walk through an end-to-end AI pipeline, showing how we can use NVIDIA’s BioNeMo to generate novel protein binders, from initial target sequences to validated, stable complexes, all within a streamlined, GPU-enabled accelerated workflow.

Target identification and protein structure prediction

Provides a structural foundation for designing protein binders.

How can NVIDIA accelerate this?

• By incorporating AlphaFold2 into the BioNeMo framework, researchers can achieve up to a 5x speedup in protein structure prediction, making large-scale target identification feasible in days rather than months [Source].

De novo protein binder design

Enables custom-designed binders tailored to the target.

How can NVIDIA accelerate this?

• NVIDIA's integration of RFdiffusion within the BioNeMo framework accelerates inference by 1.9x, enabling rapid generation of protein backbones optimized for target binding [Source].

• By leveraging ProteinMPNN, BioNeMo facilitates large-scale sequence generation at multi-GPU scale, drastically improving search space efficiency and enabling the design of amino acid sequences optimized for binding stability [Source].

Molecular docking and binding affinity prediction

Helps identify the best candidates for experimental validation.

How can NVIDIA accelerate this?

• Parallelized inference on NVIDIA GPUs allows for high-throughput screening of thousands of binders in a fraction of the time taken by traditional docking methods.

• DiffDock 2.0 enables researchers to predict molecular orientations 6.2 times faster and with 16% greater accuracy [Source].

Structural validation and stability optimization

Ensures that the designed binder is thermodynamically stable.

How can NVIDIA accelerate this?

• By utilizing parallelized inference capabilities, BioNeMo accelerates computations related to protein stability, such as changes in thermodynamic stability (ΔΔG) and melting temperature (ΔTm), facilitating rapid assessment of binder viability [Source]. ESM-1nv and ESM-2 are used for these calculations.

Functional testing and refinement

Reduces the risk of immunogenicity and manufacturability failures.

How can NVIDIA accelerate this?

• NVIDIA's Clara Discovery platform provides a collection of frameworks, applications, and AI models enabling GPU-accelerated computational drug discovery, allowing researchers to rapidly refine binder designs through distributed computing [Source].

• Tools like BioPhi and Efficient Evolution will come in handy for this step.

High throughput screening and experimental validation

Confirms that AI-generated binders work in a biological environment.

How can NVIDIA accelerate this?

• NVIDIA GPUs accelerate lead optimization and ranking.

• BioNeMo enables real-time in silico predictions, reducing the need for costly wet-lab experiments and accelerating lead optimization and ranking [Source].

Conclusion

This pipeline captures the end-to-end NVIDIA BioNeMo workflow for generative protein binder design. The tools, processes, and NVIDIA contributions align with the latest AI-driven drug discovery approaches.

Let me know what you think! Connect with me on LinkedIn for more such posts.

In this blog, we will explore the following techniques for guided generation with LLMs:

1. use of regular expressions

2. JSON schemas

3. context-free grammars (CFGs)

4. templates

5. entities

6. structured data generation.

0. Initialization

We will run this code using Vertex AI. First, let's initialize our environment and set up the Vertex AI client with the necessary configuration to ensure our outputs are both useful and safe:

import pandas as pd
import vertexai
from vertexai.generative_models import GenerativeModel, Part, FinishReason
import vertexai.preview.generative_models as generative_models

# Authenticate
from google.colab import auth
auth.authenticate_user()

vertexai.init(project="project-name", location="us-central1")

model = GenerativeModel("gemini-1.0-pro-vision-001")

generation_config = {
    "max_output_tokens": 300,
    "temperature": 0.4,
    "top_p": 0.9,
}

safety_settings = {
    generative_models.HarmCategory.HARM_CATEGORY_HATE_SPEECH: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    generative_models.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    generative_models.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    generative_models.HarmCategory.HARM_CATEGORY_HARASSMENT: generative_models.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
}

1. Guided Generation with Regular Expressions

Regular expressions (regex) are a powerful way to ensure that generated text matches a specific pattern.

E.g. Imagine you need a 6-digit number. By defining a regex pattern, you can validate the generated number, ensuring it’s exactly six digits with no extra spaces or characters. This method is great for maintaining strict control over simple, structured outputs like numeric codes or specific text formats.

number_pattern = re.compile(r"^\d{6}$")
def validate_number(number_str):
    if number_pattern.match(number_str):
        return True
    else:
        print(number_str)
        print("Invalid 6-digit format. Re-prompting...")
        return False
    
    
def generate():
    vertexai.init(project="cart-ppt-llm", location="us-central1")
    model = GenerativeModel(
        "gemini-1.0-pro-vision-001",
    )
    while True:
        responses = model.generate_content(
            [text1],
            generation_config=generation_config,
            stream=True,
        )
    
        number = ""
        for response in responses:
            number += response.text.strip()
    
        if validate_number(number):
            return number

text1 = """Generate a valid 6-digit number, ensuring it contains exactly 6 digits with no spaces or other characters. It should be in the format XXXXXX."""

generated_number = generate()
print(generated_number)

Output:

123456

2. Guided Generation with JSON Schemas

JSON schemas allow you to define the structure and data types of JSON objects. This is particularly useful when you need to generate structured data, such as user profiles, where each profile must include a name, age, and email.

By validating the generated JSON against a schema, you ensure that the output adheres to the expected structure and data types. This technique is useful for applications requiring precise and predictable data formats.

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer", "minimum": 0},
        "email": {"type": "string", "format": "email"}
    },
    "required": ["name", "age", "email"]
}

prompt = """
Generate a JSON object representing a user profile. The JSON object should have the following structure:
{
    "name": "a random first and last name",
    "age": an age between 20 and 40,
    "email": "a unique email address"
}
"""

def validate_json(instance, schema):
    """ Validate if the generated JSON matches the schema """
    try:
        jsonschema.validate(instance=instance, schema=schema)

        return True
    except jsonschema.exceptions.ValidationError as err:
        print("Invalid JSON format. Error:", err)
        return False

def generate_json():
    responses = model.generate_content(
        [prompt],
        generation_config=generation_config,
        safety_settings=safety_settings,
        stream=True,
    )

    json_response = ""
    for response in responses:
        json_response += response.text.strip()

    if json_response.startswith("```json") and json_response.endswith("```"):
        json_response = json_response[7:-3].strip()

    try:
        json_object = json.loads(json_response)
        if validate_json(json_object, schema):
            return json_object
    except json.JSONDecodeError as e:
        print("Error decoding JSON:", e)
    return None

generated_profile = generate_json()
print(f"{json.dumps(generated_profile, indent=2) if generated_profile else 'None'}")

Output

{
  "name": "John Smith",
  "age": 32,
  "email": "john.smith@example.com"
}

3. Guided Generation with Context-Free Grammars

Context-Free Grammars (CFGs) allow us to define a set of production rules for generating structured sentences. CGFs are excellent for generating structured sentences or text that follows a specific set of grammatical rules.

E.g., you might want to generate sentences about people performing actions on objects. A CFG can define the structure of these sentences, ensuring they always follow a logical and grammatical pattern. This method is ideal for tasks requiring syntactically correct and varied sentences, such as automated storytelling or dialogue generation.

grammar = CFG.fromstring("""
    S -> NP VP
    NP -> 'John' | 'Mary' | 'Alice' | 'Bob'
    VP -> V Obj
    V -> 'eats' | 'drinks' | 'sees' | 'likes'
    Obj -> Det N
    Det -> 'an' | 'a'
    N -> 'apple' | 'banana' | 'water' | 'book'
""")

prompt = f"""
Generate a sentence based on the following context-free grammar:
{grammar}
Ensure there is a space between each word in the sentence.
"""

def generate_cfg_sentence():
    responses = model.generate_content(
        [prompt],
        generation_config=generation_config,
        safety_settings=safety_settings,
        stream=True,
    )

    response_text = ""
    for response in responses:
        response_text += response.text.strip()
    
    return response_text

generated_llm_sentence = generate_cfg_sentence()
print(f"{generated_llm_sentence}")

Output:

John eats an apple

The following diagram represents the CFG used in the above example:

1. S -> NP VP:

   • The start symbol S is expanded into a noun phrase (NP) and a verb phrase (VP).

2. NP:

   • NP can be any of 'John', 'Mary', 'Alice', or 'Bob'.

3. VP -> V Obj:

   • The verb phrase VP is expanded into a verb (V) and an object (Obj).

4. V:

   • V can be any of 'eats', 'drinks', 'sees', or 'likes'.

5. Obj -> Det N:

   • The object Obj is expanded into a determiner (Det) and a noun (N).

6. Det:

   • Det can be either 'an' or 'a'.

7. N:

   • N can be any of 'apple', 'banana', 'water', or 'book'.

In our CFG, the start symbol S is expanded into a noun phrase (NP) and a verb phrase (VP). The NP can be names like 'John', 'Mary', 'Alice', or 'Bob'. The VP is broken down into a verb (V) and an object (Obj). The verb could be actions like 'eats', 'drinks', 'sees', or 'likes'. The object is composed of a determiner (Det) and a noun (N), where determiners can be 'an' or 'a', and nouns can be 'apple', 'banana', 'water', or 'book'. This structured approach ensures that the generated sentences are both grammatically correct and varied.

4. Template-based Generation:

Template-based generation uses predefined templates to structure the generated text.

E.g., you can create a user profile using a template that specifies placeholders for the name, age, and email. This method ensures that the generated content follows a consistent format, which is particularly useful for applications like automated report generation or content templating where the format is fixed, but the content varies.

prompt = """
Create a user profile using the following template:
Name: {{name}}
Age: {{age}}
Email: {{email}}

Ensure the name is a first name and a last name, the age is a number between 20 and 40, and the email is in the format name@example.com.
"""

def generate_template_based_profile():
    responses = model.generate_content(
        [prompt],
        generation_config=generation_config,
        safety_settings=safety_settings,
        stream=True,
    )

    response_text = ""
    for response in responses:
        response_text += response.text.strip()
    
    return response_text

generated_profile = generate_template_based_profile()
print(f"{generated_profile}")

Output:

Name:John Smith
Age: 35
Email: john.smith@example.com

5. Entity-based Generation

Entity-based generation is about including specific entities in the generated text.

E.g., if you want to generate a paragraph about France, you can specify entities such as the capital (Paris), a famous food (croissant), and the official language (French). This technique ensures that the generated text is relevant and includes the necessary information about the entities, making it ideal for tasks like generating descriptive content or tailored information based on specific data points.

entities = {
    "country": "France",
    "capital": "Paris",
    "famous_food": "croissant",
    "language": "French"
}

prompt = f"""
Generate a 3-line paragraph about {entities['country']} that includes the following entities:
1. The capital city, {entities['capital']}
2. A famous food, {entities['famous_food']}
3. The official language, {entities['language']}
"""

def generate_entity_based_paragraph():
    responses = model.generate_content(
        [prompt],
        generation_config=generation_config,
        safety_settings=safety_settings,
        stream=True,
    )

    response_text = ""
    for response in responses:
        response_text += response.text.strip() + " "
    
    return response_text.strip()

generated_paragraph = generate_entity_based_paragraph()
print(f"Generated Paragraph: {generated_paragraph}")

Output:

Nestled in the heart of Europe, France boasts the captivating capital of Paris, renowned for its iconic Eiffel Tower and the Louvre Museum. Indulge in the delectable aroma of freshly baked croissants, a culinary staple that embodies the nation's rich gastronomic heritage. The official language, French, echoes through the streets, adding a touch of elegance and sophistication to the vibrant atmosphere.

6. Structured Data Generation

Structured data generation involves creating data in a tabular format, such as CSV, which can be easily converted into a DataFrame for analysis or processing.

E.g., you might generate a table with columns for Name, Age, Country, and Profession, and populate it with data for several rows. This approach is beneficial for generating datasets or structured information that needs to be processed further, ensuring consistency and ease of use in data-centric applications.

column_headers = ["Name", "Age", "Country", "Profession"]

prompt = f"""
Generate a table with the following columns: {', '.join(column_headers)}.
Provide data for 5 rows.
Ensure that the table is properly formatted as CSV without any line breaks within rows.
"""

def generate_dataframe():
    responses = model.generate_content(
        [prompt],
        generation_config=generation_config,
        safety_settings=safety_settings,
        stream=True,
    )

    response_text = ""
    for response in responses:
        response_text += response.text.strip()
    
    return response_text

generated_table = generate_dataframe()


df = pd.read_csv(io.StringIO(generated_table))
df

Output:

Name	Age	Country	Profession
0	John	30	USA	Software Engineer
1	Mary	25	Canada	Doctor
2	Bob	40	UK	Teacher
3	Alice	28	Australia	Lawyer
4	Tom	35	Germany	Architect

Wrapping Up:

Guided generation techniques are key to making sure LLM outputs are useful and well-structured. Using methods like regular expressions, JSON schemas, CFGs, templates, entities, and structured data generation can greatly improve the accuracy and reliability of LLM content. These techniques help ensure the generated text meets specific needs, making it easier to integrate LLMs into real-world applications.

Here is a link to a Jupyter Notebook containing all the above code. The notebook is for demonstration purposes only. You will need a GCP account with credits in order to run it.

Thanks for reading! 🤝

Subscribe now

Paper: https://www.biorxiv.org/content/10.1101/2024.05.22.594943v1.full.pdf

Model: https://huggingface.co/alchemab/fabcon-small

Broader context:

Antibodies (Abs) have the potential to recognize + bind to a vast array of antigens. The human body can theoretically produce over 10^15 unique Abs to cover all possible antigens. Given this magnitude, discovering and optimizing Ab therapeutics is challenging. It requires screening billions of Ab variants to identify those with high binding affinity and specificity to the target antigen, while also possessing good biophysical properties for developability.

Recently, researchers have been leveraging LLMs for learning patterns from large Ab repertoires. FAbCon is a SOTA LLM model that builds on this foundation. It enables de novo design of Abs optimized for desirable properties. It can provide a rich pool of high-quality starting leads to fuel discovery pipelines.

What is FAbCon?

FAbCon is a large 2.4 billion parameter language model specifically designed and pre-trained on a massive corpus of Ab sequences to learn representations useful for Ab sequence understanding and generation tasks.

Technical details:

• Based on the transformer decoder architecture from the general Falcon LLM

• Pre-trained on a massive corpus of 823.7 million unpaired and 2.5 million paired Ab sequences using Causal Language Modeling (CLM)

• CLM involves training to predict the next amino acid based on preceding residues

• Three variants spanning 144M to 2.4B parameters: FAbCon-small, -medium, -large

After pre-training on the diverse Ab data, FAbCon learns rich representations capturing patterns in Ab sequences. These representations can then be transferred to downstream tasks:

1. Generating new Ab sequences optimized for developability properties (example shown below)

2. Predicting antigen binding by fine-tuning on labeled Ab-antigen datasets

Key Capabilities:

• SOTA performance in predicting antigen binding for Abs against targets like HER2, SARS-CoV-2, IL-6

• Can generate Ab sequences with computational developability profiles mimicking human repertoires

• Scaling up model size consistently improves antigen binding prediction

• Requires less training data compared to models without pre-training

Limitations and Future Work

• As with any LLM, FAbCon may encode biases present in the pre-training data, which could impact the reliability of the generated Ab sequences.

• The FAbCon model uses a gated architecture, which restricts the direct use of the model for certain tasks, such as full sequence generation

• The publicly available versions of FAbCon are pre-trained only, and further fine-tuning on labeled datasets would require obtaining a commercial license from the authors

Code + Use Cases:

The FAbCon models are available on the Hugging Face Hub. We can load and use them with the 🤗 Transformers library.

Since FAbCon is a gated model, there are some limitations on what can be done out-of-the-box with the publicly released versions. However, here are some potential use cases and examples:

Generate Novel Ab Sequences:

Here we will use pre-trained FAbCon to generate a novel Ab sequence with properties it learned during pre-training on the large Ab corpus.

from transformers import PreTrainedTokenizerFast, FalconForCausalLM

tokenizer = PreTrainedTokenizerFast.from_pretrained("alchemab/fabcon-large")
model = FalconForCausalLM.from_pretrained("alchemab/fabcon-large")

input_ids = tokenizer("H", return_tensors='pt')['input_ids']
output_ids = model.generate(input_ids, max_new_tokens=256, top_k=50, temperature=0.95)
decoded_seq = tokenizer.batch_decode(output_ids)[0]

print(f"Generated sequence: {decoded_seq}")

We provided a starting prompt (H) to the model, which represents the beginning of the Ab sequence. We then used model.generate() to generate a new Ab sequence of up to 256 tokens, with top_p=0.95 param to enable diverse and high quality sequence generation. This will generate a paired Ab sequence (output_ids) and print out the decoded sequence (decoded_seq).

Sequence Representation & Embedding Extraction:

We can also extract the embeddings or representations learned by FAbCon for downstream analysis like clustering, visualization, KNN searching, or transfer learning to Ab-specific tasks.

from transformers import PreTrainedTokenizerFast, FalconModel

tokenizer = PreTrainedTokenizerFast.from_pretrained("alchemab/fabcon-large")
model = FalconModel.from_pretrained("alchemab/fabcon-large")

input_ids = tokenizer("HEVQLLESGGGLVQPGGSLRLSCATSGYTFT", return_tensors='pt')['input_ids']
attention_mask = tokenizer("HEVQLLESGGGLVQPGGSLRLSCATSGYTFT", return_tensors='pt')['attention_mask']

outputs = model(input_ids=input_ids, attention_mask=attention_mask)
sequence_embedding = outputs.last_hidden_state[:, 0, :]

We started by loading the pre-trained FAbCon-small model and its tokenizer. We then encoded a sample Ab sequence into input IDs. The sequence_embedding variable now contains the high-dimensional representation of the input antibody sequence.

Zero-Shot Humanness Prediction:

The paper shows FAbCon's perplexity scores correlate strongly with the OASis humanness metric. So FAbCon could potentially be used zero-shot to predict the "humanness" of an Ab sequence.

from transformers import PreTrainedTokenizerFast, FalconForSequenceClassification

tokenizer = PreTrainedTokenizerFast.from_pretrained("alchemab/fabcon-large")
model = FalconForSequenceClassification.from_pretrained("alchemab/fabcon-large")

input_ids = tokenizer("HEVQLLESGGGLVQPGGSLRLSCATSGYTFT", return_tensors='pt')['input_ids']
attention_mask = tokenizer("HEVQLLESGGGLVQPGGSLRLSCATSGYTFT", return_tensors='pt')['attention_mask']

output = model(input_ids=input_ids, attention_mask=attention_mask)
logits = output.logits
humanness_score = logits[:, 0].item()  

The model's output contains the logits, which can be used to compute the "humanness" score directly.

For more advanced tasks like antigen binding prediction, FAbCon would need to be further fine-tuned on labeled data. That will require a commercial license according to the model card.

Thanks for reading!

Subscribe now

Share Up to Data

Share

Here’s a streamlined guide to get started with Google Cloud Batch using Cromwell:

Preparing Your Workflow

• Workflow Definition (WDL File): hello.wdl contains a simple task to greet an addressee.

task hello {
    
  String addressee  
  
  command {
    touch test.txt
    echo "Hello ${addressee}! Welcome to Cromwell" > test.txt  
  }
  output {
    File out = "test.txt"
  }
  runtime {
    docker: "us-central1-docker.pkg.dev/example-project/example-org/example-image:7841b2e"
    memory: "1 GB"
    cpu: 1
  }
}
workflow wf_hello {
    String addressee
        call hello{
            input:
                addressee = addressee
    
            }
}

• Inputs File: hello.inputs specifies the input for the workflow, such as the addressee's name.

{
  "wf_hello.hello.addressee": "Kopal"
}

• Configuration File: google.conf configures Cromwell to use Google Batch, including specifying project details, bucket for outputs, and the Batch backend settings.

backend {
  default = GCPBATCH

  providers {
    GCPBATCH {
      actor-factory = "cromwell.backend.google.batch.GcpBatchBackendLifecycleActorFactory"
      config {
        # Google project
        project = "example-project"

        # Base bucket for workflow executions
        root = "gs://org-batch-output"

        # Polling for completion backs-off gradually for slower-running jobs.
        # This is the maximum polling interval (in seconds):
        maximum-polling-interval = 600

        # Optional Dockerhub Credentials. Can be used to access private docker images.
        dockerhub {
          # account = ""
          # token = ""
        }

        # Optional configuration to use high security network (Virtual Private Cloud) for running jobs.
        # See https://cromwell.readthedocs.io/en/stable/backends/Google/ for more details.
        # virtual-private-cloud {
        #  network-label-key = "network-key"
        #  auth = "application-default"
        # }

        # Global pipeline timeout
        # Defaults to 7 days; max 30 days
        # batch-timeout = 7 days

        genomics {
          # A reference to an auth defined in the `google` stanza at the top.  This auth is used to create
          # Batch Jobs and manipulate auth JSONs.
          auth = "application-default"


          // alternative service account to use on the launched compute instance
          // NOTE: If combined with service account authorization, both that service account and this service account
          // must be able to read and write to the 'root' GCS path
          compute-service-account = "default"

          # Location to submit jobs to Batch and store job metadata.
          location = "us-central1"

          # Specifies the minimum file size for `gsutil cp` to use parallel composite uploads during delocalization.
          # Parallel composite uploads can result in a significant improvement in delocalization speed for large files
          # but may introduce complexities in downloading such files from GCS, please see
          # https://cloud.google.com/storage/docs/gsutil/commands/cp#parallel-composite-uploads for more information.
          #
          # If set to 0 parallel composite uploads are turned off. The default Cromwell configuration turns off
          # parallel composite uploads, this sample configuration turns it on for files of 150M or larger.
          parallel-composite-upload-threshold="150M"
        }

        filesystems {
          gcs {
            # A reference to a potentially different auth for manipulating files via engine functions.
            auth = "application-default"
            # Google project which will be billed for the requests
            project = "google-billing-project"

            caching {
              # When a cache hit is found, the following duplication strategy will be followed to use the cached outputs
              # Possible values: "copy", "reference". Defaults to "copy"
              # "copy": Copy the output files
              # "reference": DO NOT copy the output files but point to the original output files instead.
              #              Will still make sure than all the original output files exist and are accessible before
              #              going forward with the cache hit.
              duplication-strategy = "copy"
            }
          }
        }

        default-runtime-attributes {
          cpu: 1
          failOnStderr: false
          continueOnReturnCode: 0
          memory: "2048 MB"
          bootDiskSizeGb: 10
          # Allowed to be a String, or a list of Strings
          disks: "local-disk 10 SSD"
          noAddress: false
          preemptible: 0
          zones: ["us-central1-a"]
        }

      }
    }
  }
}

Running Your Workflow with Cromwell

java -Dconfig.file=google.conf -jar cromwell-86.jar run hello.wdl -i hello.inputs

This command uses Cromwell to run your hello.wdl workflow, specifying hello.inputs as the input file and google.conf for configuration.

The Issue with the Latest Cromwell Release

However, when attempting to run workflows with the most recent version of Cromwell, users encountered an “invalid Docker spec” error due to a change in how GCP Batch backend processed mount paths. This issue, detailed in a GitHub thread, prevented workflows from executing successfully.

Workaround and Solution

To circumvent this, the Cromwell team provided fixes in subsequent development versions, addressing the Docker spec issue. We found success by pulling the latest Cromwell Docker image, which includes these fixes:

Pull the Latest Cromwell Docker Image:

docker pull broadinstitute/cromwell:latest

Execute the Workflow:

docker run -it --rm -v $(pwd)/batch:/batch --entrypoint /bin/bash broadinstitute/cromwell:latest -c "java -Dconfig.file=/batch/config/google.conf -jar /app/cromwell.jar run /batch/wdls/hello.wdl -i /batch/hello.inputs"

Until the next Cromwell release, this approach ensures smooth workflow execution on Google Cloud Batch, leveraging the latest developments and community support.

Subscribe now

This process, while seemingly straightforward, introduces a set of challenges that can complicate maintaining data integrity and consistency

• Append-Only Nature of BigQuery: BigQuery is designed primarily for appending data, not updating or deleting specific rows.

• Deletions Aren’t Synced: Similarly, if you remove a row from your spreadsheet, this deletion won’t propagate to BigQuery.

• Schema Changes Are Restricted: Modifying the structure of your Google Sheet presents another layer of complexity. While BigQuery allows for some schema modifications, these are limited.

In this blog, we will explore an App Script-based strategy that addresses these challenges head-on.

Our approach involves three main steps:

Download Sheet to a CSV

The first step involves exporting the Google Sheet data to a CSV format. Here’s how:

function exportSheetToCsv(sheetId, sheetName) {
  var url = `https://docs.google.com/spreadsheets/d/${sheetId}/export?format=csv&name=${encodeURIComponent(sheetName)}`;
  const options = { headers: { 'Authorization': 'Bearer ' + ScriptApp.getOAuthToken() } };
  const response = UrlFetchApp.fetch(url, options);
  const blob = response.getBlob();
  return blob;
}

This function fetches the specified Google Sheet as a CSV, ready for upload to BigQuery.

Delete the Table from BigQuery

Remove the existing BigQuery table to address the append-only and deletion-sync issues.

To ensure our BigQuery dataset reflects the latest data structure and content from Google Sheets, we’ll delete the existing table:

function deleteBigQueryTable(projectId, datasetId, tableId) {
  try {
    BigQuery.Tables.remove(projectId, datasetId, tableId);
    console.log('Table deleted successfully.');
  } catch (e) {
    console.log(`Error deleting table: ${e.message}`);
  }
}

Upload the CSV to BigQuery as a New Table

Create a fresh table in BigQuery with the same name, ensuring it reflects the current state of the Google Sheet.

With the table deleted, we upload the new data:

function uploadCsvToBigQuery(blob, projectId, datasetId, tableId) {
  const file = DriveApp.createFile(blob);
  const data = file.getBlob().setContentType('application/octet-stream');
  const job = {
    configuration: {
      load: {
        destinationTable: { projectId, datasetId, tableId },
        sourceFormat: 'CSV',
        autodetect: true,
        skipLeadingRows: 1
      }
    }
  };
  BigQuery.Jobs.insert(job, projectId, data);
  console.log('CSV uploaded and table created in BigQuery.');
}

Send an Update Notification

Optionally, send a notification (e.g., via Slack) to inform stakeholders that the data sync has been completed.

function sendSlackNotification(webhookUrl, message) {
  const payload = JSON.stringify({ text: message });
  const options = {
    method: 'post',
    contentType: 'application/json',
    payload: payload
  };
  UrlFetchApp.fetch(webhookUrl, options);
}

Automating Data Sync with Time-Driven Triggers

To automate the data syncing process from Google Sheets to BigQuery, you can use Google Apps Script’s time-driven triggers. These triggers can run your data sync script at regular intervals — ranging from every minute to once a month.

Stitch it all together, and you get something like this:

function loadCsv() {

  var url = 'https://docs.google.com/spreadsheets/d/' + 'sheet_ID' + '/export?format=csv&name=' + encodeURIComponent('sheet_name');

  const options = {
    headers: {
      'Authorization': 'Bearer ' + ScriptApp.getOAuthToken()
    }
  };

  const response = UrlFetchApp.fetch(url, options);

  const currentDate = new Date();

  const dateString = currentDate.toISOString().split('T')[0].replace(/-/g, '');

  const blob = response.getBlob().setName('sheet_' + dateString + '.csv');

  const exportedFile = DriveApp.createFile(blob); 

  const projectId = 'project_id';
  const datasetId = 'dataset_id';
  const csvFileId = exportedFile.getId();
  const tableId = 'table_id' 

  // Delete existing table

  try {
    BigQuery.Tables.remove(projectId, datasetId, tableId);
    console.log('Table deleted.');
  } catch (e) {
    console.log('Table deletion failed or table did not exist. Error: ' + e.message);
  }
  // Create new table with the same name

  var table = {
    tableReference: {
      projectId: projectId,
      datasetId: datasetId,
      tableId: tableId
    }
  };
  
 // Upload table to BigQuery
  table = BigQuery.Tables.insert(table, projectId, datasetId);
  console.log('Table created: %s', table.id);
  
  const file = DriveApp.getFileById(csvFileId);
  const data = file.getBlob().setContentType('application/octet-stream');
  const job = {
    configuration: {
      load: {
        destinationTable: {
          projectId: projectId,
          datasetId: datasetId,
          tableId: tableId
        },
        skipLeadingRows: 1,
        autodetect: true,
        sourceFormat: 'CSV'

      }
    }
  };
  const jobResult = BigQuery.Jobs.insert(job, projectId, data);
  console.log('Load job started. Status: ' + jobResult.status.state);

// Send a Slack
  sendSlackNotification(webhookUrl, message)
}

This solution ensures that your BigQuery dataset remains up-to-date with your Google Sheets, including reflecting any updates, deletions, or schema changes.

Feel free to customize my script based on your specific requirements!

We will create an API for a calculator that computes the combination and permutation for n items chosen from a set of k. Here are the two essential endpoints we'll start with:

This endpoint will calculate the number of ways to choose k items from a set of n items without regard to the order, i.e. combination:

GET /combinations?n=5&k=3

This endpoint will calculate the number of ways to arrange k items from a set of n items with regard to the order, i.e. permutation:

GET /permutations?n=5&k=3

Intrigued? Let's dive into the code and build our endpoints.

Creating Endpoints:

We will start by defining a few helper functions for computing the combination and permutations. Keep in mind that n is the total # items to choose from, and k is the # items chosen. These functions also include input validation to ensure that n and k are non-negative integers, and n is greater than or equal to k.

import math

def calculate_combinations_logic(n, k):
    if n < 0 or k < 0 or n < k:
        raise ValueError("Invalid input")
    combinations = math.factorial(n) // (math.factorial(k) * math.factorial(n - k))
    
    return combinations

def calculate_permutations_logic(n, k):
    if n < 0 or k < 0 or n < k:
        raise ValueError("Invalid input")
    permutations = math.factorial(n) // math.factorial(n - k)
    
    return permutations

a. Combinations:

from flask import Flask, request

app = Flask(__name__)

@app.route('/combinations', methods=['GET'])
def calculate_combinations():
    n = int(request.args.get('n'))
    k = int(request.args.get('k'))
    
    result = calculate_combinations_logic(n, k)
    
    return f"Combinations of {n} choose {k}: {result}"

b. Permutations:

@app.route('/permutations', methods=['GET'])
def calculate_permutations():
    n = int(request.args.get('n'))
    k = int(request.args.get('k'))
    
    result = calculate_permutations_logic(n, k)
    
    return f"Permutations of {n} choose {k}: {result}"

But wait, how do others know how to use them? That's where Swagger documentation comes into play.

Swagger Docs:

Now we will integrate Swagger into our API. Let’s begin by installing the flask-swagger package:

pip install flask-swagger,flask_swagger_ui

Also, let us download the latest release from the swagger-ui GitHub. Un-tar this, and move the dist directory to a separate directory named swaggerui and create another new directory named static. Move swaggerui to static.

Your folder structure should look something like this:

(base) kopalgarg@kgarg swagger-documentation % ls -lah
total 9248
drwxr-xr-x@  5 kopalgarg  staff   160B Nov 28 18:17 .
drwxr-xr-x@ 94 kopalgarg  staff   2.9K Nov 28 18:10 ..
-rw-r--r--   1 kopalgarg  staff   2.5K Nov 28 18:32 api.py
drwxr-xr-x   3 kopalgarg  staff    96B Nov 28 18:24 static
-rw-r--r--   1 kopalgarg  staff   3.8M Nov 28 18:15 v5.10.3.tar.gz


(base) kopalgarg@kgarg swagger-documentation % ls -lah static/swaggerui 
total 20248
drwxr-xr-x  20 kopalgarg  staff   640B Nov 22 01:04 .
drwxr-xr-x   3 kopalgarg  staff    96B Nov 28 18:24 ..
-rw-r--r--   1 kopalgarg  staff   665B Nov 22 01:04 favicon-16x16.png
-rw-r--r--   1 kopalgarg  staff   628B Nov 22 01:04 favicon-32x32.png
-rw-r--r--   1 kopalgarg  staff   202B Nov 22 01:04 index.css
-rw-r--r--   1 kopalgarg  staff   834B Nov 28 18:24 index.html
-rw-r--r--   1 kopalgarg  staff   2.7K Nov 22 01:04 oauth2-redirect.html
-rw-r--r--   1 kopalgarg  staff   501B Nov 28 18:23 swagger-initializer.js
-rw-r--r--   1 kopalgarg  staff   1.3M Nov 22 01:04 swagger-ui-bundle.js
-rw-r--r--   1 kopalgarg  staff   1.8M Nov 22 01:04 swagger-ui-bundle.js.map
-rw-r--r--   1 kopalgarg  staff   450K Nov 22 01:04 swagger-ui-es-bundle-core.js

Modify static/index.html with the correct paths:

<!-- HTML for static distribution bundle build -->
<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8">
    <title>Swagger UI</title>
    <link rel="stylesheet" type="text/css" href="static/swaggerui/swagger-ui.css" />
    <script src="static/swaggerui/swagger-ui-bundle.js" charset="UTF-8"></script>
    <script src="static/swaggerui/swagger-ui-standalone-preset.js" charset="UTF-8"></script>
    <script src="static/swaggerui/swagger-initializer.js" charset="UTF-8"></script>
  </head>
  <body>
    <div id="swagger-ui"></div>
    <script src="static/swaggerui/swagger-ui-bundle.js" charset="UTF-8"> </script>
    <script src="static/swaggerui/swagger-ui-standalone-preset.js" charset="UTF-8"> </script>
    <script src="static/swaggerui/swagger-initializer.js" charset="UTF-8"> </script>
  </body>
</html>

Modify static/swagger-initializer.js with the correct paths:

window.onload = function() {
  //<editor-fold desc="Changeable Configuration Block">

  // the following lines will be replaced by docker/configurator, when it runs in a docker-container
  window.ui = SwaggerUIBundle({
    url: "/spec",
    dom_id: '#swagger-ui',
    deepLinking: true,
    presets: [
      SwaggerUIBundle.presets.apis,
      SwaggerUIStandalonePreset
    ],
    plugins: [
      SwaggerUIBundle.plugins.DownloadUrl
    ],
    layout: "StandaloneLayout"
  });

  //</editor-fold>
};

Once we have the set-up out of the way, we will modify our Flask API code to use flask-swagger for documentation:

from flask import Flask, request, jsonify
from flask_swagger import swagger
import math

def calculate_combinations_logic(n, k):
    if n < 0 or k < 0 or n < k:
        raise ValueError("Invalid input")
    combinations = math.factorial(n) // (math.factorial(k) * math.factorial(n - k))
    
    return combinations

def calculate_permutations_logic(n, k):
    if n < 0 or k < 0 or n < k:
        raise ValueError("Invalid input")
    permutations = math.factorial(n) // math.factorial(n - k)
    
    return permutations

app = Flask(__name__)

@app.route("/spec")
def spec():
    swag = swagger(app)
    swag['info']['title'] = "Combination and Permutation"
    swag['info']['version'] = "1.0.0"
    return jsonify(swag)

@app.route('/api-docs')
def swagger_ui():
    return app.send_static_file('swaggerui/index.html') 

@app.route('/combinations', methods=['GET'])
def calculate_combinations():
    """
    Calculate the number of combinations (n choose k).

    This endpoint calculates the number of ways to choose k items from a set of n items without regard to the order (combinations).

    ---
    parameters:
      - name: n
        in: query
        type: integer
        required: true
        description: Total number of items to choose from.
      - name: k
        in: query
        type: integer
        required: true
        description: Number of items to choose.

    responses:
      200:
        description: The number of combinations.
    """
    n = int(request.args.get('n'))
    k = int(request.args.get('k'))
    
    result = calculate_combinations_logic(n, k)
    
    return f"Combinations of {n} choose {k}: {result}"

@app.route('/permutations', methods=['GET'])
def calculate_permutations():
    """
    Calculate the number of permutations (n P k).

    This endpoint calculates the number of ways to arrange k items from a set of n items with regard to the order (permutations).

    ---
    parameters:
      - name: n
        in: query
        type: integer
        required: true
        description: Total number of items to arrange from.
      - name: k
        in: query
        type: integer
        required: true
        description: Number of items to arrange.

    responses:
      200:
        description: The number of permutations.
    """
    n = int(request.args.get('n'))
    k = int(request.args.get('k'))
    
    result = calculate_permutations_logic(n, k)
    
    return f"Permutations of {n} choose {k}: {result}"

if __name__ == '__main__':
    app.run()

Testing it out

Assuming you saved your API code in Python file named api.py, if you run python3 api.py, you should see something like this:

If you open up http://127.0.0.1:5000/, you should see the raw JSON for the specs on http://127.0.0.1:5000/spec:

and the Swagger documentation on http://127.0.0.1:5000/api-docs:

Now, if you wanted to test the API, you could do so directly from the documentation Swagger page, like so:

That’s all for now! Let me know if you have any thoughts, and feel free to subscribe to my blog if you found this helpful. You can also find me on LinkedIn and GitHub.

First, we'll create nodes for each friend and coffee shop.

CREATE (Emma:Person {name: 'Emma', age: 24})
CREATE (Liam:Person {name: 'Liam', age: 26})
CREATE (Olivia:Person {name: 'Olivia', age: 25})

CREATE (JavaJunction:CoffeeShop {name: 'Java Junction'})
CREATE (BeanBrew:CoffeeShop {name: 'Bean Brew'})
CREATE (CaffeineCorner:CoffeeShop {name: 'Caffeine Corner'})

Next, we establish relationships between friends and their favourite coffee shops, and also friendships. In this step, we're using FAVORS relationships to link each person to their favourite coffee shop and FRIENDS_WITH relationships to connect the friends.

For e.g., we create a relationship between ‘Emma’ and ‘Java Junction’. If this relationship already exists, MERGE will not create a duplicate. Then we create a 'FRIENDS_WITH' relationship between Emma and Olivia.

MATCH (e:Person {name: 'Emma'}), (jj:CoffeeShop {name: 'Java Junction'})
MERGE (e)-[:FAVORS]->(jj)
WITH e, jj
MATCH (l:Person {name: 'Liam'}), (bb:CoffeeShop {name: 'Bean Brew'})
MERGE (l)-[:FAVORS]->(bb)
WITH l, bb
MATCH (o:Person {name: 'Olivia'}), (cc:CoffeeShop {name: 'Caffeine Corner'})
MERGE (o)-[:FAVORS]->(cc)
WITH o, cc
MATCH (Emma:Person {name: 'Emma'}), (Olivia:Person {name: 'Olivia'})
MERGE (Emma)-[:FRIENDS_WITH]->(Olivia)
WITH Emma, Olivia
MATCH (Olivia:Person {name: 'Olivia'}), (Liam:Person {name: 'Liam'})
MERGE (Olivia)-[:FRIENDS_WITH]->(Liam)

Let's run a query to find out who likes 'Bean Brew':

MATCH (p:Person)-[:FAVORS]->(cs:CoffeeShop {name: 'Bean Brew'})
RETURN p.name

This query will return the name of the person who favors 'Bean Brew', which in our graph is Liam.

So, what exactly is Neo4j?

Now that we've seen Neo4j in action, let's backtrack a bit and talk about what Neo4j really is. It is a graph database management system that is built to deal with complex data relationships way better than your standard relational databases. This blog post will guide you through setting up Neo4j on GCP.

Hosting on GCP:

So why should you choose GCP for hosting Neo4j?

• Scalability: GCP lets you easily adjust resources to handle different database loads, ensuring efficient performance even as your data grows.

• Security: With GCP's security measures, your data remains protected, which is crucial for sensitive information.

• Cost-Effectiveness: GCP's pay-as-you-go pricing model makes it a budget-friendly option, especially for projects with fluctuating resource needs.

Setting up a VM on GCP:

We start by configuring the GCP project and creating a new VM instance using the gcloud command-line tool.

gcloud config set project [ID]

gcloud compute instances create neo4j-vm \
    --machine-type=n1-standard-1 \
    --image-family=debian-10 \
    --image-project=debian-cloud \
    --zone=us-central1-a

Set up firewall rules:

Next, we have to make sure the VM can receive traffic on the ports used by Neo4j. The default is 7474 for HTTP and 7687 for Bolt protocol.

gcloud compute firewall-rules create default-allow-http \
    --allow tcp:7474 \
    --target-tags=http-server \
    --source-ranges=0.0.0.0/0

gcloud compute firewall-rules create default-allow-bolt \
    --allow tcp:7687 \
    --target-tags=bolt-server \
    --source-ranges=0.0.0.0/0

The above command initiates a SSH connection to neo4j-vm.

• -L 7474:localhost:7474: This forwards port 7474 on your local machine to port 7474 on the virtual machine. This is often used for accessing web applications running on the virtual machine (like Jupyter notebooks) from your local browser.

• -L 7687:localhost:7687: Similarly, this forwards port 7687 on your local machine to port 7687 on the virtual machine. Port 7687 is commonly used by Neo4j, a graph database, for its Bolt protocol, which is a binary protocol for database access.

Download Neo4j:

Now, download and install Neo4j. We'll use the latest stable version:

wget -O - https://debian.neo4j.com/neotechnology.gpg.key | sudo apt-key add -

echo 'deb https://debian.neo4j.com stable 4.4' | sudo tee -a /etc/apt/sources.list.d/neo4j.list

sudo apt-get update

sudo apt-get install neo4j=1:4.4.12

Start the Neo4j service:

cd /var/lib/neo4j
./bin/neo4j start

Accessing Neo4j on the VM:

Once you have Neo4j installed and running on your VM, the next step is to securely access it. To connect to your VM and forward the necessary ports for Neo4j, use the following command:

gcloud compute ssh --zone "us-west1-b" "neo4j-vm"  --project "ID" -- -L 7474:localhost:7474 -L 7687:localhost:7687

After running the above command, your local machine is connected to the VM, and the ports for accessing Neo4j are forwarded. You can access the Neo4j browser interface at http://localhost:7474. By default, your username is neo4j. You can set an initial password using this command:

 /bin/neo4j-admin set-initial-password 1234

Follow the instructions to reset your password. Once you have done so, your window should look something like this:

You now have a fully functional Neo4j graph database running on a GCP VM!

Resources:

• Neo4J on GCP

• Configure Firewalls

One common task in CI/CD pipelines is building and pushing Docker images to container registries like Google Artifact Registry (GAR), Amazon Elastic Container Registry (ECR) or DockerHub. In this blog, we will explore how to test a GitHub Actions (GAs) pipeline for pushing an image to DockerHub. Specifically, we'll look at two types of testing: unit testing and integration testing.

In this blog, you'll learn:

1. Unit Testing to validate individual components like Dockerfiles and scripts.

2. Integration Testing using tools like act to ensure your GitHub Actions workflows function as expected.

1. Unit Testing the Docker Image

Unit testing in the context of a Docker image involves running tests against small testable parts of the script in isolation from the rest of the code. This can help ensure that individual actions work as expected. So, without having to run the entire workflow, we can test the core logic that the action is performing.

Test the Dockerfile

Before we even get to the GA, we have to make sure the Dockerfile builds as expected. We can write a script that tries to build the Docker image and fails if the docker build command fails.

Test the script

If the GA is running a shell or Python script to push the Docker image, we can test that script in isolation. For example, if we have a Python function that constructs docker push command, we can write a unit test for that function.

Mock external calls

If the action makes calls to external services (like in our case, DockerHub), we can mock those calls.

For example, if we’re using Python’s subprocess module to execute the docker push command, we can mock subprocess.run() to test that it gets called with expected arguments.

• actions.py contains the actual logic for pushing a Docker image.

• test_actions.py contains unit tests to verify that the Docker image can be built and pushed correctly.

The tests are designed to catch any issues in the Docker build and push process, ensuring that the actual shell commands are being called with the correct arguments.

actions.py

import subprocess

def construct_docker_push_command(image_name, tag):
    return f"docker push {image_name}:{tag}"

def function_that_calls_docker_push():
    try:
        subprocess.run(["docker", "push", "samtools:latest"], check=True)
    except subprocess.CalledProcessError as e:
        print(f"An error occurred while pushing the Docker image: {e}")

In actions.py, you have a function called function_that_calls_docker_push() that uses Python's subprocess module to run a shell command for pushing a Docker image.

test_actions.py

import unittest
from unittest.mock import patch
from action import function_that_calls_docker_push, construct_docker_push_command

class TestDockerActions(unittest.TestCase):

    def test_construct_docker_push_command(self):
        self.assertEqual(construct_docker_push_command("samtools", "latest"), "docker push samtools:latest")

    @patch("action.subprocess.run")
    def test_docker_push(self, mock_run):
        function_that_calls_docker_push()
        mock_run.assert_called_with(["docker", "push", "samtools:latest"], check=True)
    
if __name__ == '__main__':
    unittest.main()

In test_actions.py, you have several unit tests:

1. test_construct_docker_push_command: This test checks if the function construct_docker_push_command() returns the correct Docker push command string.

2. test_docker_push: This test mocks the subprocess.run() function and checks if function_that_calls_docker_push() calls it with the correct arguments.

3. @patch("subprocess.run") is using Python's unittest.mock to replace subprocess.run with a mock object for the duration of the test.

4. mock_run.assert_called_with(["docker", "push", "samtools:latest"]) checks if the mock was called with the specified arguments.

If you run python3 -m unittest test_action.py, you should see something like this, if all tests pass:

2. Integration Testing GitHub Actions

Integration testing involves testing the interaction between multiple units of an application. They help ensure the action works as expected when it interacts with the GitHub environment or other services.

We will use act, which will let us run GA workflows locally, and help speed up the development and testing process. Make sure it’s installed and ready to run:

brew install act

act

It's worth noting that act doesn't perfectly emulate GitHub Actions. Some third-party actions may not work as expected, so always double-check your results.

Ok, let’s use act to run the entire workflow and ensure all steps are executed correctly. First we will define a few environment variables in a .secrets file (placed in your root directory). This is useful for setting secrets or other environment-specific variables that your GitHub Actions might need.

DOCKERHUB_USERNAME=yourusername
DOCKERHUB_TOKEN=yourpassword

Here's a sample GitHub Actions YAML configuration for a Docker build and push workflow:

name: Docker Push

on:
  push:
    branches:
      - main

jobs:
  docker:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v2

    - name: Build Docker image
      run: docker build -t samtools:latest .

    - name: Login to DockerHub
      uses: docker/login-action@v1
      with:
        username: ${{ secrets.DOCKERHUB_USERNAME }}
        password: ${{ secrets.DOCKERHUB_TOKEN }}
    - name: Tag image
      run: docker tag samtools:latest ${{ secrets.DOCKERHUB_USERNAME }}/samtools:latest
    - name: Push Docker image
      run: docker push ${{ secrets.DOCKERHUB_USERNAME }}/samtools:latest

If everything runs smoothly, you should see something like this:

By running these integration tests, you can catch issues that unit tests might miss, such as problems that only appear when multiple steps are run together.

That's all for now! If you want the full code from this blog, check out this repo.

Have you implemented similar testing strategies in your CI/CD pipelines? 👀 If so, I would love to hear your thoughts on this!

Leave a comment

Share

This is how it works:

Traditional methods of text generation often involve a one-shot query to the model, which may or may not yield the desired output. Iterative prompting allows us to refine the model's output in a more controlled way.

This is useful when we have criteria that the generated text must meet, like semantic similarity to the original text and level of politeness.

Scenario:

Imagine you're a community manager who needs to moderate comments on a forum. Some comments may be rude but contain valid points that shouldn't be entirely dismissed. Using iterative prompting, you can rephrase these comments to retain their original meaning while making them more polite and respectful.

Setting Up Your Environment

First, let's install the necessary packages

!pip install openai
!pip install langchain
!pip install sentence_transformers

Importing Libraries and Initializing the Model

import os
import openai
from langchain.llms import OpenAI
from sentence_transformers import SentenceTransformer, util
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch

os.environ["OPENAI_API_KEY"] = 'your_openai_api_key_here'
llm = OpenAI(temperature=0.9)

Defining Helper Functions

In our code, we used two metrics to evaluate the quality of the rephrased comments.

1. Semantic Similarity: We used the SentenceTransformer library to compute the cosine similarity between the original and rephrased comments. This ensures that the rephrased comment retains the original meaning.

def compute_semantic_similarity(original, rephrased):
    model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
    embedding1 = model.encode(original, convert_to_tensor=True)
    embedding2 = model.encode(rephrased, convert_to_tensor=True)
    similarity = util.pytorch_cos_sim(embedding1, embedding2)
    return similarity.item()

1. Politeness: We used a pre-trained BERT model to classify the sentiment of the rephrased comment. The model returns a label ranging from very negative to very positive, which we use as a proxy for politeness.

def is_polite_transformer(text):
    model_name = "nlptown/bert-base-multilingual-uncased-sentiment"
    model = AutoModelForSequenceClassification.from_pretrained(model_name)
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
    with torch.no_grad():
        outputs = model(**inputs)
    _, predicted = torch.max(outputs.logits, 1)
    label = int(predicted.item())
    return label

Let’s do a side investigation into the politeness of various sentences. This is to better understand how the politeness metric behaves and to validate its effectiveness in different contexts.

The model assigns a label ranging from 0 (very negative) to 4 (very positive), which we interpret as a measure of politeness:

rephrased_comment = "Your painting is excellent"
polite_label = is_polite_transformer(rephrased_comment)
print(f"Is the comment polite? {polite_label}")
# 4

rephrased_comment = "Your painting is good"
polite = is_polite_transformer(rephrased_comment)
print(f"Is the comment polite? {polite}")
# 3

rephrased_comment = "Your painting is not the best"
polite = is_polite_transformer(rephrased_comment)
print(f"Is the comment polite? {polite}")
# 2

rephrased_comment = "I'm not a fan of your painting."
polite = is_polite_transformer(rephrased_comment)
print(f"Is the comment polite? {polite}")
# 1

rephrased_comment = "Your painting absolutely sucks"
polite = is_polite_transformer(rephrased_comment)
print(f"Is the comment polite? {polite}")
# 0

We tested a variety of sentences, ranging from overtly rude to extremely polite, to see how well the classifier performs. Here are some examples:

• "Your painting is excellent" received a score of 4, indicating it's very polite.

• "Your painting is good" received a score of 3, still polite but less so than the previous example.

• "Your painting is not the best" received a score of 2, which is neutral.

• "I'm not a fan of your painting" received a score of 1, leaning towards negative.

• "Your painting absolutely sucks" received a score of 0, which is very negative.

This is a decent quantitative way to measure one of the key dimensions we are interested in: the politeness of the rephrased comments.

The Iterative Prompting Loop

Next we will loop through a list of rude comments and iteratively prompt the model to rephrase them:

rude_comments = [
    "You're making it hard for me to take you seriously.",
    "This is the most ridiculous thing I've ever heard.",
    "I've never seen someone mess up this badly.",
    "You're not just wrong, you're stupidly wrong.",
    "Is this really the best you could come up with?",
    "I don't have the time or the crayons to explain this to you.",
    "You're not just a disappointment, you're a waste of time.",
    "I've met some dumb people, but you take the cake.",
    "You're so clueless it's almost cute."]

# Initialize variables to store results
iterative_similarity_scores = []
iterative_politeness_scores = []
standard_similarity_scores = []
standard_politeness_scores = []


# Loop through each rude comment

for comment in rude_comments:
    # Iterative Prompting
    user_input = comment
    for i in range(max_iterations):  
        prompt_rephrase = f"Can you reword this comment to be more polite and kinder?: {user_input}"
        user_input_rephrased = llm(prompt_rephrase)
        
        similarity_prompt_check = f"Is the reworded comment semantically similar to the original comment? Original comment: {user_input}, Reworded comment: {user_input_rephrased}"
        politeness_prompt_check = f"Is the reworded comment more polite and kinder than the original comment? Original comment: {user_input}, Reworded comment: {user_input_rephrased}"

        similarity_verification_result = llm(similarity_prompt_check)
        politeness_verification_result = llm(politeness_prompt_check)
  
        if "YES" in similarity_verification_result.strip().upper() and "YES" in politeness_verification_result.strip().upper():  
            break
        else:
          user_input = user_input_rephrased

    print(f"Original comment: {comment}")
    print(f"Iterative rephrased comment: {user_input_rephrased}")


    semantic_similarity=compute_semantic_similarity(comment, user_input_rephrased)
    iterative_similarity_scores.append(semantic_similarity)
    print(f'semantic similarity: {semantic_similarity}')
    politeness=is_polite_transformer(user_input_rephrased)
    print(f'politeness: {politeness}')
    iterative_politeness_scores.append(politeness)
    
    standard_rephrased = llm(f"Can you reword this comment to be more polite and kinder?: {comment}")
    print(f"Standard rephrased comment: {standard_rephrased}")
    

    semantic_similarity=compute_semantic_similarity(comment, standard_rephrased)
    standard_similarity_scores.append(semantic_similarity)
    print(f'semantic similarity: {semantic_similarity}')
    politeness=is_polite_transformer(standard_rephrased)
    print(f'politeness: {politeness}')
    standard_politeness_scores.append(politeness)

Here is an example output where iterative prompting performs better than standard querying in terms of politeness:

Original comment: I've met some dumb people, but you take the cake.
Iterative rephrased comment: 

I've been pleasantly surprised by your knowledge.
semantic similarity: 0.24437330663204193
politeness: 4

Standard rephrased comment: 

I've met people with different levels of knowledge, however you stand out.
semantic similarity: 0.3822552561759949
politeness: 2

Comparative Analysis: Iterative Prompting vs. Standard Querying

After running the loop, we can compute average scores for semantic similarity and politeness for both iterative prompting and standard querying:

avg_iterative_similarity = sum(iterative_similarity_scores) / len(iterative_similarity_scores)
avg_iterative_politeness = sum(iterative_politeness_scores) / len(iterative_politeness_scores)
avg_standard_similarity = sum(standard_similarity_scores) / len(standard_similarity_scores)
avg_standard_politeness = sum(standard_politeness_scores) / len(standard_politeness_scores)


print(f"Avg Iterative Similarity: {avg_iterative_similarity}, Avg Iterative Politeness: {avg_iterative_politeness}")
print(f"Avg Standard Similarity: {avg_standard_similarity}, Avg Standard Politeness: {avg_standard_politeness}")

Avg Iterative Similarity: 0.279, Avg Iterative Politeness: 2.111

Avg Standard Similarity: 0.372, Avg Standard Politeness: 1.667

After running our experiment, we found that iterative prompting generally produced more polite comments compared to standard querying, with an average politeness score of 2.11 versus 1.67.

However, the semantic similarity was slightly lower. This trade-off is an important consideration when implementing iterative prompting in real-world applications.

That’s all for now! If you have additional thoughts, please reach out. Here is the notebook I used for my analysis.

Leave a comment

• basic concepts of probability: events, experiments and outcomes

• axioms, rules and principles in probability

• some python simulations to deepen your understanding

• bonus: a couple questions you can expect to see in data science interviews

Let’s get started.

Events, experiments and outcomes

An event is just a possible outcome of the experiment. When you flip a coin, the event could be landing a heads up or a tails up. An experiment is a process that generates an outcome. For example, rolling a die, flipping a coin, measuring the temperature on a given day are all experiments because they all have multiple possible outcomes.

Now that we have those down, let’s do a simple coin flip simulation. Heads is 1, and tails is 0, and we can assume the coin is fair.

import random

def flip_coin():
  return random.choice([0,1])

Because we know the coin is fair, we know that if we flip the coin a 100 times, we can expect a roughly equal number of heads and tail ups.

def hundred_flips(n=100):
  result=[]
  for i in range(0,n):
    result.append(flip_coin())
  return result

flips=hundred_flips(n=100)
num_heads = sum(flips)
num_tails = len(flips) - num_heads

print(f"Number of Heads: {num_heads}") # 48
print(f"Number of Tails: {num_tails}") # 52

As you can see, we got 48 heads and 52 tails. We can expect to get similar numbers if we repeat this a thousand more times, and look at the distribution of heads ups.

import matplotlib.pyplot as plt

def run_experiment(num_flips=100, num_trials=1000):
  heads_counts=[]
  for _ in range(num_trials):
    flips = hundred_flips(num_flips)
    heads=sum(flips)
    heads_counts.append(heads)
  return heads_counts

heads_counts=run_experiment()

plt.hist(heads_counts, bins=range(30,70), alpha=0.75)
plt.title('Distribution of Heads in 100 Coin Flips')
plt.xlabel('Number of heads')
plt.ylabel('Frequency')
plt.show()

In the above simulation, we flip a coin 100 times as we did before, and repeat that a 1000 times, and in each trial keep track of the number of heads we get in each 100-flip sequence. When we plot this distribution, we see that the plot roughly simulates a normal distribution, which is what we expected.

Axioms, rules and principles

So now let’s talk about axioms, or the laws of probability.

1. Non-negative probability: This just means the probability of any event is at least 0, i.e. P(E)≥0,∀E∈Ω, and here E is the event and Ω is the sample space (i.e. for all events in sample space). Note that it can be 0, just has to be non-negative.

2. Certainty of sample space: The one means that something must happen. The total probability over the entire sample space is 1, i.e.:

3. Addition rule for mutually exclusive events: So if you have a mutually exclusive event, E_i​, it’s probability of occurring is the sum of it’s individual probabilities. A coin flip outcome can not be heads and tails at the same time, i.e. heads and tails are mutually exclusive. The probability of both, P(H⋃T) = P(H) + P(T). By extension:

Conditional probability

For two events, A and B, conditional probability, P(A|B), is the probability that event A happens when one assumes that event B has already happened.

Let’s say we want to calculate P(A|B) for rolling a six-sided die. Event A is rolling an odd number and event B is rolling a number less than 3, i.e.:

Event A: {1, 3, 5}, Event B: {1, 2}

From basic probability we know:

• the theoretical probability of getting an odd number, P(A) =3/6 = 0.5

• the theoretical probability of getting a number <3, P(B) = 2/6 =0.33

• the theoretical probability of getting both, P(A∩B)=1/6 = 0.167 because only one number (1) satisfies both conditions.

• for conditional probability, P(A|B), we use formula P(A∩B)/P(B)=0.167/0.33=0.5

Let’s simulate this.

import random

# Event A: {1, 3, 5}
# Evnet B: {1, 2}

def simulate_die_rolls(num_rolls=100):
  count_A = 0
  count_B = 0
  count_A_and_B = 0

  for _ in range(num_rolls):
    roll = random.randint(1,6)
    is_A = roll in [1, 3, 5]
    is_B = roll in [1, 2]
    if is_A:
      count_A+=1
    if is_B:
      count_B+=1
    if is_A and is_B:
      count_A_and_B +=1

  p_A = count_A / num_rolls
  p_B = count_B / num_rolls
  p_A_and_B = count_A_and_B / num_rolls

  p_A_given_B = count_A_and_B / count_B

  is_independent = p_A_and_B == (p_A * p_B)

  return p_A, p_B,  p_A_and_B, p_A_given_B, is_independent

p_A, p_B, p_A_and_B, p_A_given_B, is_independent=simulate_die_rolls(num_rolls=10000)

print(f"P(A) = {p_A}")
print(f"P(B) = {p_B}")
print(f"P(A and B) = {p_A_and_B}")
print(f"P(A|B) = {p_A_given_B}")
print(f"Are A and B independent? {is_independent}")

Result:

P(A) = 0.4998
P(B) = 0.3342
P(A and B) = 0.163
P(A|B) = 0.48773189706762415
Are A and B independent? False

The simulated probabilities are close, which is a good sanity check. The P(A|B) value suggests that the if we know event B has occurred, there is a ~49% chance that event B will also occur. The independence check tells us that A and B are not independent, which makes sense. The more simulations you run, the closer you will get to the theoretical values we computed earlier.

Bayes’ rule

Bayes’ rule is a powerful tool for revising existing predictions or hypothesis given new evidence. You may have seen it in the context of medical diagnosis, spam filtering and other decision-making scenarios where you have some data and you want to make an informed decision. Here is how it is generally applied:

• P(A), prior probability: this is what you already know about event A happening without new evidence on event B

• P(B|A), likelihood: this tells you how likely B is assuming A is true

• P(B), marginal likelihood: this tells you how likely new evidence B is under all possible values of A

• P(A|B), posterior probability: this (goal of Bayes’ theorem) tells you how likely A is given some new evidence B

The general formula for Bayes’ rule is:

P(B|A) = (P(B|A)P(A)) / P(B)

Say, for example, we have a rare disease that affects 1% of the population. We also have to test for this disease that has a 95% true positive rate (TPR) and a 5% FPR. We want to know: if you get a positive test, what is the chance you actually have the disease?

P(disease | positive test) = P(positive test | disease) * P(disease) / P(positive test)

expanding P(positive test) = [ P(positive test | disease ) * P(disease) + P(positive test | no disease) * P(no disease)]

so, P(disease | positive test) = P(positive test | disease) * P(disease) / [ P(positive test | disease ) * P(disease) + P(positive test | no disease) * P(no disease)]

def bayes_rule(prior, tpr, fpr):

  # prior
  p_disease = prior # P(disease)

  # complement
  p_no_disease = 1-prior # P(no disease)

  # likelihood
  p_pos_given_disease = tpr # P(pos | disease)

  # fpr
  p_pos_given_no_disease = fpr # P(pos | no disease)

  #marginal
  p_pos = (p_disease * p_pos_given_disease) + (p_no_disease * p_pos_given_no_disease)

  #posterior
  p_disease_given_pos = (p_pos_given_disease * p_disease) / p_pos

  return p_disease_given_pos

prior=0.01
tpr=0.95
fpr=1-tpr

result = bayes_rule(prior, tpr, fpr)
print(result)

# 0.1610

So this implies that even though the test is 95% accurate, the chance of actually having the disease after testing is only about 16.10%. This is why Bayes’ theorem is so important. It includes prior probability into the calculation.

Combinatorics

Combinatorics is the study of counting things. There are two main ways of counting. Consider 3 apples, A_1, A_2, and A_3, and you’re picking 2 out of them. Two considerations:

• ordered vs. unordered: does the order matter? i.e. (A_1, A_2) different from (A_2, A_1)? if so, the selection process is ordered, otherwise unordered.

• with or without replacement: when selecting one object, is that object removed from future selection? if so, the selection process is said to have replacement.

Let's map some of these concepts to code.

• Ordered selection with replacement

For ordered selection with replacement, we have n^k possible selections :

def ordered_with_replacement(n,k):
  return n**k

n=3 # 3 apples
k=2 # pick 2
result =ordered_with_replacement(n,k)
print(result) # 9

(1,1), (1,2), (1,3), (2,1), (2,2), (2,3), (3,1), (3,2), (3,3)

For 3 apples, picking 2, we compute 3^2 = 9

• Ordered selection without replacement

For ordered selection without replacement, use n!/(n-k)!

from math import factorial

def ordered_without_replacement(n, k):
    return factorial(n) // factorial(n - k)

result = ordered_without_replacement(n, k)
print(result) # 6

(1,2), (1,3), (2,1), (2,3), (3,1), (3,2)

We compute 3!/(3-2)! = 6

• Unordered selection without replacement

For unordered selection without replacement, we use n!/(k! * (n-k)!)

def unordered_without_replacement(n, k):
    return factorial(n) // (factorial(k) * factorial(n - k))

result = unordered_without_replacement(n, k)
print(result) # 3

(1,2), (1,3), (2,3)

We compute 3!/(2!*(3-2)!)=3

• Unordered selection with replacement

For unordered selection with replacement, we use (n+k-1)!/k!(n-1)!

def unordered_with_replacement(n, k):
    return factorial(n + k - 1) // (factorial(k) * factorial(n - 1))

result = unordered_with_replacement(n, k)
print(result)

(1,1), (1,2), (1,3), (2,2), (2,3), (3,3)

We compute (3+2-1)!/(2!(3-1)!)=6

In each case, the results match up with the theoretical expectations, and we’ve also enumerated the sets of all possible selections.

Bonus: questions

1. Three zebras (E):

Three zebras are chilling in the desert. Suddenly a lion attacks.

Each zebra is sitting on a corner of an equally length triangle. Each zebra randomly picks a direction and only runs along the outline of the triangle to either edge of the triangle.

What is the probability that none of the zebras collide?

So each zebra can either run in a CW, or CCW direction.

P(CW) = P(CCW) = 1/2 * 1/2 * 1/2 = 1/8

so P(same direction) = P(CW) + P(CCW) = 1/4

2. Lazy Rater (E):

Netflix has hired people to rate movies.

Out of all of the raters, 80% of the raters carefully rate movies and rate 60% of the movies as good and 40% as bad. The other 20% are lazy raters and rate 100% of the movies as good.

Assuming all raters rate the same amount of movies, what is the probability that a movie is rated good?

P(careful) = 0.8

• P(rated good|careful) = 0.6

• P(rated bad|careful) = 0.4

P(lazy) = 0.2

• P(rated good|lazy) = 1.0

• P(rated bad|lazy) = 0

P(rated good)= P(rated good|careful)*P(careful) + P(rated good|lazy)*P(lazy) = 0.4*0.8 + 0.2*1=0.68

3. Fake Algorithm Reviews (E):

Let’s say we’re trying to determine fake reviews on our products. 

Based on past data, 98% reviews are legitimate and 2% are fake. If a review is fake, there is 95% chance that the machine learning algorithm identifies it as fake. If a review is legitimate, there is a 90% chance that the machine learning algorithm identifies it as legitimate.

What is the percentage chance the review is actually fake when the algorithm detects it as fake?

P(legit) = .98

P(fake) = .02

P(identified fake|fake) = 0.95 =

P(identified legit|legit) = 0.90

P(fake|identified fake) = ?

This is a Bayes’ theorem problem.

P(fake|identified fake) = P(identified fake|fake) P(fake)/P(identified fake)

P(identified fake) = P(identified fake|fake) P(fake) + P(identified fake|legit) P(legit)

P(fake|identified fake) = 0.95 * 0.02 / [(0.95*0.2) + ((1-0.90)+0.98)]

= 0.162

That’s all for now! ✌🏼In the next part, we will talk about discrete distributions.

In this blog, we take a deep dive into various strategies and look at code snippets that’ll help you make your data pipelines more fault-tolerant, and maintainable.

Variable Retry Mechanisms

Exponentially increase the delay between task retries

Airflow’s built-in retry mechanism is useful, but it often makes sense to exponentially increase the delay between the tries instead of having a constant delay between retries.

Imagine you have a task that calls an external API. If the API goes down temporarily due to rate-limiting or maintenance, the exponential backoff strategy gives the API more time to recover, rather than bombarding it with repeated calls.

You can achieve this with some custom Python logic:

from datetime import timedelta

def custom_retry_delay(context):
    var_factor = context['ti'].try_number ** 2   factor
    return timedelta(minutes=5 * var_factor)

default_args = {
    'retries': 3,
    'retry_delay': custom_retry_delay,
}

So in the above code, we define a custom retry delay function. The try_number is squared to get an exponential factor. This is then used to calculate the new delay time, multiplying it with a base delay.

Dynamic retry delays

Similar to exponential delay, dynamic delay introduces a variable retry interval. The difference is with dynamic delay, you can delay times using a pre-defined list. There are no dynamic calculations involved, so there is no dynamic progression or increase. You can set dynamic retry delays by setting retry_delay to a list of timedelta objects.

This is useful when you know the exact intervals you’d like between your retries. For example, when dealing with multiple services with known recovery times, you can set the delays accordingly.

default_args = {
    'retries': 3,
    'retry_delay': [timedelta(minutes=5), timedelta(minutes=10), timedelta(minutes=20)]
}

State Management and Conditional Logic

Using custom XComs for State

For those not familiar, XComs is short for “cross-communications”. It’s a feature that allows you to pass state in form of messages of small amounts of data between tasks. This can be useful when you have tasks that depend on the state of previous tasks. XComs are especially useful in making your DAGs fail gracefully by letting subsequent tasks know if prior task has failed, partially succeeded, or produced a certain output.

def push_xcom_value(**kwargs):
    ti = kwargs['ti']
    ti.xcom_push(key='my_key', value='my_value')

One of the arguments to push_xcom_value is ti, which tasks for task instance. The ti object has methods for pushing and pulling XCom values in form of key-pairs, like my_key and my_value.

When this task is run, it will push the key-pair into XCom. Other tasks within the same DAG can pull this value and make decisions based on it.

def pull_xcom_and_act(**kwargs):
    ti = kwargs['ti']
    value = ti.xcom_pull(key='my_key')
    if value == 'my_value':
        do_something()
    else:
        do_something_else()

Imagine you are asked to set up a data pipeline where the first task is to fetch data from an API. If the API returns an empty dataset, there is no point running the rest of the pipeline. Using XCom, you can pass the state of the data from the first task to subsequent tasks, which can then decide whether to continue execution or skip themselves.

Conditional skipping of tasks

In come cases, you may want to conditionally skip tasks based on custom logic or the state of other tasks. Skipping tasks can help you make your DAGs by avoiding unnecessary computations or operations.

from airflow.exceptions import AirflowSkipException

def conditionally_skip_task():
  if some_condition:
      raise AirflowSkipException

In above code, when the task is run, the condition will be evaluated. If it is met, the task will raise an exception that will effectively skip the task, setting its state to “Skipped” in the Airflow UI.

Looking at the same case use case as above, this exception would be useful if you want to skip certain tasks when the data comes back empty.

Task rescheduling

Airflow allows tasks to be rescheduled rather than retried from scratch, which is useful for long-running or resource-intensive tasks. This is done by the Reschedule exception from airflow.models.

Consider a data processing task that takes several hours to complete. If the task fails midway due to issues like network timeouts, rescheduling allows the task to continue from where it left off, saving both time and compute resources.

Here’s an example:

from airflow.models import Reschedule
def long_running_task():
  if not task_complete:
     raise Reschedule 

So when Reschedule is raised, the task instance will be rescheduled to run again at the next available time slot, maintaining its state.

This feature enables more efficient resource utilization and can significantly improve the reliability of long-running tasks in your DAGs.

Callbacks and Alerts

Using on_failure_callback and on_retry_callback

Airflow provides callback mechanisms that allow you to define custom behavior when a task fails or is retried. These callbacks can be used to send alerts, clean up resources, or perform any other custom logic

from airflow.operators.python import PythonOperator

def alert_on_failure(context):
   pass

task = PythonOperator(
   task_id='my_task'
   python_callable=my_python_function
   on_failure_callback=alert_on_failure
)

When the task fails, the function specified in on_failure_callback will be executed.

These callbacks are particularly useful for alerting. For instance, you could use on_failure_callback to send a Slack message, log the failure details, or even trigger another task that performs some sort of cleanup or rollback. This is demonstrated next

Share

Add Slack webhooks and operators for notifications

Airflow provides built-in support for Slack notifications via the SlackAPIPostOperator. This lets you send custom messages to a Slack channel, which can be useful for alerting your team when a task fails or meets certain conditions.

Let's say you have a data pipeline with multiple tasks, and it's crucial to know immediately if any task fails. You can add the slack_alert task to your DAG and use Airflow's trigger_rule to execute it whenever a task fails, ensuring instant notification.

from airflow.providers.slack.operators.slack_api import SlackAPIPostOperator

slack_alert = SlackAPIPostOperator(
   task_id = 'slack_alert',
   text = 'Task failed!'
   token = 'your_token',
   channel = '#your-channel'
)

This task can be added to your DAG and set to execute when a preceding task fails, succeeds, or meets some other condition. When triggered, the SlackAPIPostOperator will post the specified message to the given Slack channel.

Setting up alerts

Airflow also has in-built support to set up email alerts that can be configured using email_on_failure and email_on_retry parameters. These parameters allow you to automatically send email alerts when tasks in your DAG fail or are retried.

Consider a data pipeline where task failures could have a significant impact. By setting email_on_failure to True, you can ensure that the right team members are immediately notified and can take quick action to resolve the issue.

default_args = {
   'email_on_failure': True,
   'email_on_retry': False,
   'email': ['your-email@example.com']
}

You can specify multiple recipients in the email list.

Dependency Management

External trigger dependency with sensors

Airflow’s sensor tasks are designed to wait for certain criteria to be met before proceeding to execute subsequent tasks. One common use case is to use the ExternalTaskSensor to wait for the completion of tasks in another DAG.

Imagine two separate DAGs: 1) data ingestion and 2) data processing. The processing DAG should only run after the ingestion DAG has successfully ingested the day's data. You can use ExternalTaskSensor in the processing DAG to wait for the completion of the ingestion task in the ingestion DAG.

from airflow.sensors.external_task import ExternalTaskSensor

wait_for_other_task = ExternalTaskSensor(
    task_id='wait_for_other_task',
    external_dag_id='other_dag_id',
    external_task_id='other_task_id',
)

The ExternalTaskSensor will keep polling for the status of the specified external task in the specified external DAG. Once that task is complete, the sensor task will be marked as complete, allowing downstream tasks to proceed.

Setting up Service Level Agreements (SLAs)

Service Level Agreements (SLAs) in Airflow allow you to specify the maximum amount of time a task should take to complete. This can be particularly useful to ensure your tasks meet operational requirements.

Suppose you have a task that aggregates daily sales data and it's important for this data to be available within the first few hours of the new day for reporting. By setting an SLA, you can make sure that any delays in this task are flagged to allow for prompt remedial action.

my_task = PythonOperator(
    task_id='my_task',
    python_callable=my_python_function,
    sla=timedelta(hours=2),
)

When this task is executed, Airflow will monitor its duration. If the task takes more than 2 hours to complete, it will violate the SLA. Airflow provides ways to alert or take other actions when SLA violations occur, often configured via the email and sla_miss_callback parameters.

That’s all for now! These techniques are best practices that can help you build robust data pipelines. Feel free to share any additional techniques or insights you might have—I'd love to hear from you!

Subscribe now

In this blog, we will learn how to:

• create a GPU-enabled Docker container

• push the custom Docker image to Google Artifacts Registry

• craft a WDL script that leverages the GPU-enabled Docker

• run a WDL using Cromwell

0. Setting Up the Environment:

Start by making sure you have Docker installed on your system, or VM. Next, choose an artifact registry like Google Artifact Registry or AWS Elastic Container Registry and set it up according to their instructions. Finally, you can run the following linux command to check if your host machine's GPU is configured properly

nvidia-smi

If everything is in order, you'll see detailed information about your GPU.

1. Build a GPU-Enabled Docker Image

To begin, create a file named Dockerfile with the following content:

FROM nvidia/cuda:12.2.0-devel-ubuntu20.04

This line sets up a GPU-enabled environment using NVIDIA's CUDA 12.2 development image based on Ubuntu 20.04.

From here, you can further customize the image to include specific tools or libraries relevant to your bioinformatics workflow, like so:

FROM nvidia/cuda:12.2.0-devel-ubuntu20.04

WORKDIR /usr/src/app

RUN apt-get update && apt-get install -y \
    git \
    wget
    
ENV CONDA_DIR /opt/conda
RUN wget --quiet https://repo.anaconda.com/miniconda/Miniconda3-py39_23.5.2-0-Linux-x86_64.sh -O ~/miniconda.sh && \
    /bin/bash ~/miniconda.sh -b -p /opt/conda

ENV PATH=$CONDA_DIR/bin:$PATH

RUN git clone https://github.com/example/example && \
    cd example && \
    python3 setup.py install

Build a Docker image named ‘example’ with the following command:

docker build -t example .

Test the image locally:

docker run --rm --gpus=all -it example

If you see a GPU driver incompatibility error, you may use the following to avoid it:

docker run --rm --gpus=all --env NVIDIA_DISABLE_REQUIRE=1 -it example

2. Push the Image to the Artifact Registry

Once you've built and tested the Docker image locally, you may want to deploy it in a cloud environment, like the Google Artifact Registry (GAR).

You'll need to authenticate your local Docker client with GAR so you can push and pull images from that registry.

gcloud auth configure-docker us-central1-docker.pkg.dev

Next, tag the image with the full registry path including the location (us-central1), project ID (project-1), repository name (repository-1) and image name (example):

docker tag example us-central1-docker.pkg.dev/project-1/repository-1/example

Finally, push the tagged image to the registry:

docker push us-central1-docker.pkg.dev/project-1/repository-1/example

3. Writing the WDL Task

Now let’s create a file named example_workflow.wdl to define a task to run an example command using the GPU-enabled Docker image we created earlier.

task example {
  String sequence_input

  runtime {
    docker: "us-central1-docker.pkg.dev/project-1/repository-1/example"
    memory: "16 GB"
    cpu: "8"
    disks: "local-disk 500 HDD"
    gpuCount: 1
    bootDiskSizeGb: 100
  }

  command <<<
   
    example [...] > result
    tar -cvf result.tar result/

  >>>
  output {
    File example_output = "result.tar"
  }
}

workflow example_workflow {
  String sequence_input

  call example {
    input:
      sequence_input = sequence_input
  }
  output{
    File example_output_workflow = example.example_output
    }
}

The next step is to execute it. We'll use Cromwell, a WDL execution engine.

java -jar cromwell.jar run my_task.wdl -i inputs.json

You'll need to create an inputs.json file with the necessary inputs. For the example workflow we defined earlier, the inputs file might look like this:

{
  "example_workflow.sequence_input": "your_sequence_here"
}

Cromwell will provide log output as the workflow runs, allowing you to monitor its progress.

With these steps, you can create, push, and use GPU-enabled Docker containers in your bioinformatics workflows using WDL.

Share

Subscribe now

References

• WDL specifications: https://github.com/openwdl/wdl/blob/main/versions/1.1/SPEC.md

• Setting WDL runtime attributes for Cromwell: https://cromwell.readthedocs.io/en/stable/RuntimeAttributes/

To provide some context:

• Vertex AI: Google's managed machine learning platform designed for building, deploying, and scaling models. It offers a few pre-trained models along with customization options

• BigQuery: Google's data warehouse solution that enables SQL queries over massive datasets

This article will lead you through the following steps:

• Retrieve relevant data from Stack Overflow using BigQuery

• Learn how to clean and structure your data for optimal model training

• Fine-tune a Vertex AI foundation language model

• Discover various techniques for caching using LangChain

1. Retrieve data from BigQuery

This section deals with data extraction from BigQuery and downloading a CSV file containing the required data. Note that you’ll find information on environment variables, and libraries towards the end of the article.

• Run this SQL query on BigQuery to extract data from the publicly available Stack Overflow dataset. Filter questions and answers based on specific tags (e.g. R, bash) and dates.

SELECT
  CONCAT(q.body, q.tags) AS input_text,
  a.body AS output_text
FROM
  `bigquery-public-data.stackoverflow.posts_questions` q
JOIN
  `bigquery-public-data.stackoverflow.posts_answers` a
ON
  q.accepted_answer_id = a.id
WHERE
  q.accepted_answer_id IS NOT NULL
  AND (REGEXP_CONTAINS(q.tags, "R")
    OR REGEXP_CONTAINS(q.tags, "bash"))
  AND a.creation_date >= "2020-01-01"
LIMIT
  10000

• Save the above results into a CSV file on your Google Drive. The next piece of code downloads the file into a dataframe.

creds = None  
drive_service = build('drive', 'v3', credentials=creds)
file_id='file-id'

request = drive_service.files().get_media(fileId=file_id)
downloaded = io.BytesIO()
downloader = MediaIoBaseDownload(downloaded, request)
done = False

while done is False:
    _, done = downloader.next_chunk()

downloaded.seek(0)
df = pd.read_csv(downloaded)[0:1000]

2. Data Cleaning

This section focuses on data cleaning, specifically removing HTML tags from text data.

• Here is a custom function that uses BeautifulSoup to clean the HTML tags from the input_text and output_text columns of the dataframe.

def remove_html_tags(text):
    return BeautifulSoup(text, 'html.parser').get_text()

df['input_text'] = df['input_text'].apply(remove_html_tags)
df['output_text'] = df['output_text'].apply(remove_html_tags)

3. Model Tuning

This section focuses on splitting the data into training and evaluation sets and tuning a language model using Vertex AI.

• This code creates a JSONL file for training and then defines the new model name.

train, evaluation = train_test_split(df, test_size=0.2)
tune_jsonl = train.to_json(orient='records', lines=True)
training_data_filename = "tune_data_stack_overflow_python_qa.jsonl"

with open(training_data_filename, "w") as f:
    f.write(tune_jsonl)

tuned_model_name="genai-workshop-tuned-model-cart1"

• With the following code you can tune a pre-trained model (text-bison@001) for a specific task, with parameters like training steps and locations.

model = TextGenerationModel.from_pretrained("text-bison@001")
model.tune_model(training_data=training_data_filename, model_display_name=tuned_model_name, train_steps=100, tuning_job_location="europe-west4",tuned_model_location="us-central1")

4. Inference

This section involves generating predictions from different models. Each model takes a prompt as an input and predicts the response based on the defined parameters like temperature and maximum output tokens.

Here we ask it to write an R function.

parameters = {
    "temperature": 0.2,
    "max_output_tokens": 1024
}

prompt = """
Write an R function that uses GPU acceleration to multiply two matrices. Include details such as matrix dimensions, data types, and any specific GPU libraries or functions you use.
"""

Tuned Model: This is the model we fine-tuned using the dataset from Stack Overflow.

tuned_model = model.get_tuned_model("projects/581748955524/locations/us-central1/models/2298518062749450240")
response_tuned_model = tuned_model.predict(
    prompt,
    **parameters
)
print(f"Response from Model: {response_tuned_model.text}")

#' Multiply two matrices using GPU acceleration
#'
#' This function multiplies two matrices using GPU acceleration.
#'
#' @param x A numeric matrix.
#' @param y A numeric matrix.
#'
#' @return A numeric matrix.
#'
#' @examples
#' x <- matrix(1:16, 4, 4)
#' y <- matrix(1:16, 4, 4)
#' z <- gpu_matmul(x, y)
#'
#' @importFrom gpuR gpuR
#'
#' @export
gpu_matmul <- function(x, y) {
  # Check input dimensions
  if (dim(x)[2] != dim(y)[1]) {
    stop("Matrices must have compatible dimensions.")
  }

  # Create GPU matrices
  x_gpu <- gpuR::gpuR(x)
  y_gpu <- gpuR::gpuR(y)

  # Multiply matrices
  z_gpu <- gpuR::gpuR(x_gpu %*% y_gpu)

  # Convert GPU matrix to R matrix
  z <- gpuR::gpuR_to_array(z_gpu)

  return(z)
}

Code Model: This is a Vertex AI foundation model, designed specifically to generate code snippets based on given requirements.

code_model = CodeGenerationModel.from_pretrained("code-bison@001")
response_code_model = code_model.predict(
    prompt,
    **parameters
)
print(f"Response from Model: {response_code_model.text}")")

# Multiply two matrices using GPU acceleration

multiply_matrices_gpu <- function(m1, m2) {

  # Check that the matrices are of the same size and type

  if (dim(m1) != dim(m2)) {
    stop("The matrices must be of the same size.")
  }

  if (typeof(m1) != typeof(m2)) {
    stop("The matrices must be of the same type.")
  }

  # Initialize the GPU library

  library(gputools)

  # Create GPU matrices from the input matrices

  m1_gpu <- gpuMatrix(m1)
  m2_gpu <- gpuMatrix(m2)

  # Multiply the matrices on the GPU

  product_gpu <- m1_gpu %*% m2_gpu

  # Return the product matrix

  return(product_gpu)

}

Both models seem to produce syntactically correct code snippets that follow the prompt.

5. Caching Strategies

This section explores strategies to speed up retrieval of repeated queries.

We time cache retrieval for different prompts.

Exact Caching: Uses exact match between query and cache.

Exact caching is straightforward. It is efficient for queries that are repeated exactly. It makes sure to retrieve the cached result rather than re-computing it.

Here's how it's implemented:

• A Vertex AI model is linked with a specific tuned model

• A cache object is initialized

llm = VertexAI(tuned_model_name='projects/581748955524/locations/us-central1/models/2298518062749450240')

llm_cache = Cache()
llm_cache.init(pre_embedding_func=get_prompt)

cached_llm = LangChainLLMs(llm=llm)

• Timing is used to show the retrieval speed of cached and uncached prompts

• For exact matches, the response is taken from the cache

Query once:

start = time.time()

prompt = 'What is caching?'
answer = cached_llm(prompt=prompt, cache_obj=llm_cache)
print(time.time() - start)

Time: 6.31 s

Send the exact same query again:

start = time.time()

prompt = 'What is caching?'
answer = cached_llm(prompt=prompt, cache_obj=llm_cache)
print(time.time() - start)

Time: 0.76 s

This is where it falls short. It doesn’t recognize a similar question that is worded differently, and ends up taking a long time:

start = time.time()

prompt = 'Define caching'
answer = cached_llm(prompt=prompt, cache_obj=llm_cache)
print(time.time() - start)

Time: 6.49 s

Semantic Caching: Utilizes similarity-based matching between queries.

And this is where semantic caching is better. It goes a step further by recognizing similarity-based matches. Here's how it works:

• Again, a Vertex AI model is linked with a specific tuned model

• A hash function is used to initialize the cache object, linking it with a directory

llm = VertexAI(tuned_model_name='projects/581748955524/locations/us-central1/models/2298518062749450240')

def init_gptcache(cache_obj: Cache, llm: str):
    hashed_llm = hashlib.sha256(llm.encode()).hexdigest()
    init_similar_cache(cache_obj=cache_obj, data_dir=f"similar_cache_{hashed_llm}")


langchain.llm_cache = GPTCache(init_gptcache)

• Timing is used to compare the retrieval speed for similar but not identical queries

• If the query is semantically similar to a cached result, the cached response is used

Query once:

start = time.time()
llm('What is sharding?')
print(time.time() - start)

Time: 7.33 s

Semantically similar query:

start = time.time()
llm('Can you define sharding?')
print(time.time() - start)

Time: 0.71 s

As you can see, cached results are faster for slightly different, but related queries.

So, in this article we learned how to fine tune foundation models from Vertex AI. We also saw how semantic caching can provide an additional layer of performance optimization.

Subscribe now

References

GPTCache: https://github.com/zilliztech/GPTCache

LangChain caching: https://python.langchain.com/docs/integrations/llms/llm_caching

Tuning foundation models: https://cloud.google.com/vertex-ai/docs/generative-ai/models/tune-models

Share

Additional details:

Environment Variables

Here's a quick look at the variables being used.

• project: specifies the unique ID of the project

• bucket_uri: denotes the URI of the bucket

• location: defines the location for resource provisioning

project="abc" 
bucket_uri="gs://abc-cde-efg"
location="us-central1"  

Import Libraries

• io, time, numpy, and pandas for handling input-output and data processing

• BeautifulSoup for HTML parsing

• aiplatform, bigquery, and Google Drive API libraries for interacting with Google Cloud

• Other libraries for caching and VertexAI

import io
import time
import numpy as np
import pandas as pd
import hashlib
from typing import Union
from sklearn.model_selection import train_test_split
from bs4 import BeautifulSoup

from google.cloud import aiplatform, bigquery
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload
from google.oauth2.credentials import Credentials
from google.colab import auth as google_auth

import vertexai
from vertexai.preview.language_models import TextGenerationModel
from vertexai.language_models import CodeGenerationModel

from langchain import SQLDatabase, PromptTemplate, LLMChain
from langchain.llms import VertexAI
from langchain.cache import GPTCache

from gptcache import cache, Cache
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation
from gptcache.processor.pre import get_prompt
from gptcache.adapter.langchain_models import LangChainLLMs
from gptcache.adapter.api import init_similar_cache