Lab L3.3: Performance Optimization

🎯 Assignment: Accept this lab on GitHub Classroom
You’ll get your own repository with starter code, instructions, and automatic grading.

Duration	90 minutes
Prerequisites	Previous module completed

Objectives

• Profile agent performance
• Optimize response times
• Implement caching strategies

---

How to Complete This Lab

1. Accept the Assignment — Click the GitHub Classroom link above
2. Clone Your Repo — git clone <your-repo-url>
3. Read the README — Your repo has detailed requirements and grading criteria
4. Write Your Code — Implement the solution in solution/agent.py
5. Test Locally — Use swaig-test to verify your agent works
6. Push to Submit — git push triggers auto-grading

---

Key Concepts

The following exercises walk through the concepts you’ll need. Your GitHub Classroom repo README has the specific requirements for grading.

Scenario

You have a slow agent that needs optimization. Identify bottlenecks and fix them.

---

Part 1: Baseline Measurement (20 min)

The Slow Agent

Create slow_agent.py:

#!/usr/bin/env python3
"""Intentionally slow agent for optimization lab."""

import time
import requests
from signalwire_agents import AgentBase, SwaigFunctionResult


class SlowAgent(AgentBase):
    def __init__(self):
        super().__init__(name="slow-agent")

        # Verbose prompt (anti-pattern)
        self.prompt_add_section(
            "Role",
            "You are an incredibly helpful, extremely knowledgeable, and "
            "wonderfully friendly customer service representative who works "
            "for our amazing company. Your job is to assist customers with "
            "any and all inquiries they might have about our products, "
            "services, policies, procedures, and anything else they need. "
            "You should always be polite, professional, and patient. "
            "Remember to greet customers warmly and thank them for calling. "
            "Always ask if there's anything else you can help with before "
            "ending the conversation. Be sure to speak clearly and at a "
            "moderate pace so customers can understand you easily."
        )

        self.add_language("English", "en-US", "rime.spore")
        self._setup_functions()

    def _setup_functions(self):
        @self.tool(description="Look up product information")
        def get_product(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            product_id = args.get("product_id", "")
            # Simulated slow API call (no timeout!)
            time.sleep(2)  # Artificial delay
            return SwaigFunctionResult(f"Product {product_id}: Widget Pro, $99.99")

        @self.tool(description="Check inventory")
        def check_inventory(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            sku = args.get("sku", "")
            # Multiple slow calls
            time.sleep(1)
            warehouse_a = 10
            time.sleep(1)
            warehouse_b = 5
            time.sleep(1)
            warehouse_c = 15

            return SwaigFunctionResult(
                f"Inventory for {sku}: {warehouse_a + warehouse_b + warehouse_c} total"
            )

        @self.tool(description="Calculate shipping")
        def calculate_shipping(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            zip_code = args.get("zip_code", "")
            weight = args.get("weight", 0.0)
            # Slow external call without timeout
            try:
                response = requests.get(
                    f"https://httpbin.org/delay/3",  # Intentionally slow
                    timeout=30  # Way too long
                )
                return SwaigFunctionResult(f"Shipping to {zip_code}: $12.99")
            except Exception as e:
                return SwaigFunctionResult(f"Error: {str(e)}")


if __name__ == "__main__":
    agent = SlowAgent()
    agent.run()

Measure Baseline

# Time function execution
time swaig-test slow_agent.py --exec get_product --product_id "12345"

# Time inventory check
time swaig-test slow_agent.py --exec check_inventory --sku "WIDGET-001"

# Time shipping calculation
time swaig-test slow_agent.py --exec calculate_shipping \
  --zip_code "90210" --weight "2.5"

Record Results

Create performance_baseline.md:

# Performance Baseline

| Function | Time (seconds) | Notes |
|----------|----------------|-------|
| get_product | | |
| check_inventory | | |
| calculate_shipping | | |

## Prompt Analysis
- Word count: [count]
- Issues identified: [list]

---

Part 2: Implement Optimizations (40 min)

Task

Create optimized_agent.py with all performance improvements.

Optimizations to Implement

1. Concise Prompt

def _configure_prompts(self):
    self.prompt_add_section(
        "Role",
        "Customer service agent. Help with products, inventory, and shipping."
    )

    self.prompt_add_section(
        "Guidelines",
        bullets=[
            "Be helpful and concise",
            "Confirm before taking actions",
            "Offer alternatives when needed"
        ]
    )

2. Function with Timeout and Fillers

@self.tool(
    description="Look up product information",
    fillers=["Looking that up...", "Checking our catalog..."]
)
def get_product(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
    product_id = args.get("product_id", "")
    try:
        # Simulated API with timeout
        result = self._fetch_product_with_timeout(product_id, timeout=2)
        return SwaigFunctionResult(f"Product {product_id}: {result}")
    except TimeoutError:
        return SwaigFunctionResult(
            "I'm having trouble loading product details. "
            "Can I help with something else?"
        )

def _fetch_product_with_timeout(self, product_id: str, timeout: int):
    import concurrent.futures

    with concurrent.futures.ThreadPoolExecutor() as executor:
        future = executor.submit(self._fetch_product, product_id)
        try:
            return future.result(timeout=timeout)
        except concurrent.futures.TimeoutError:
            raise TimeoutError("Product fetch timed out")

def _fetch_product(self, product_id: str):
    # Actual API call
    time.sleep(0.5)  # Simulated reasonable delay
    return "Widget Pro, $99.99"

3. Parallel Inventory Check

import concurrent.futures

@self.tool(
    description="Check inventory",
    fillers=["Checking all warehouses..."]
)
def check_inventory(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
    sku = args.get("sku", "")
    # Parallel warehouse checks
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        futures = {
            executor.submit(self._check_warehouse, "A"): "A",
            executor.submit(self._check_warehouse, "B"): "B",
            executor.submit(self._check_warehouse, "C"): "C"
        }

        total = 0
        for future in concurrent.futures.as_completed(futures, timeout=3):
            try:
                total += future.result()
            except Exception:
                pass  # Skip failed warehouse

    return SwaigFunctionResult(f"Inventory for {sku}: {total} total")

def _check_warehouse(self, warehouse: str) -> int:
    time.sleep(0.3)  # Simulated delay
    return {"A": 10, "B": 5, "C": 15}.get(warehouse, 0)

4. Cached Shipping Calculation

from functools import lru_cache

@lru_cache(maxsize=100)
def _get_shipping_rate(self, zone: str) -> float:
    """Cached shipping rate lookup."""
    # This would normally call external API
    zones = {"west": 9.99, "central": 11.99, "east": 12.99}
    return zones.get(zone, 14.99)

@self.tool(
    description="Calculate shipping",
    fillers=["Calculating shipping cost..."]
)
def calculate_shipping(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
    zip_code = args.get("zip_code", "")
    weight = args.get("weight", 0.0)
    # Determine zone from zip (cached logic)
    zone = self._zip_to_zone(zip_code)

    # Get cached rate
    base_rate = self._get_shipping_rate(zone)
    total = base_rate + (weight * 0.50)

    return SwaigFunctionResult(f"Shipping to {zip_code}: ${total:.2f}")

def _zip_to_zone(self, zip_code: str) -> str:
    first_digit = zip_code[0] if zip_code else "5"
    if first_digit in "012":
        return "east"
    elif first_digit in "345":
        return "central"
    else:
        return "west"

5. Speech Timing Configuration

def _configure_timing(self):
    self.set_params({
        "end_of_speech_timeout": 400,  # Slightly faster
        "attention_timeout": 8000,
        "barge_min_words": 2  # Require 2 words to interrupt
    })

---

Part 3: Complete Optimized Agent (10 min)

Full Implementation

#!/usr/bin/env python3
"""Optimized agent with performance improvements."""

import time
import concurrent.futures
from functools import lru_cache
from signalwire_agents import AgentBase, SwaigFunctionResult


class OptimizedAgent(AgentBase):
    def __init__(self):
        super().__init__(name="optimized-agent")

        self._configure_prompts()
        self._configure_timing()
        self.add_language("English", "en-US", "rime.spore")
        self._setup_functions()

    def _configure_prompts(self):
        self.prompt_add_section(
            "Role",
            "Customer service agent. Help with products, inventory, shipping."
        )
        self.prompt_add_section(
            "Guidelines",
            bullets=["Be helpful and concise", "Confirm before acting"]
        )

    def _configure_timing(self):
        self.set_params({
            "end_of_speech_timeout": 400,
            "attention_timeout": 8000,
            "barge_min_words": 2
        })

    @lru_cache(maxsize=100)
    def _get_shipping_rate(self, zone: str) -> float:
        return {"west": 9.99, "central": 11.99, "east": 12.99}.get(zone, 14.99)

    def _zip_to_zone(self, zip_code: str) -> str:
        first = zip_code[0] if zip_code else "5"
        return {"0": "east", "1": "east", "2": "east",
                "3": "central", "4": "central", "5": "central"}.get(first, "west")

    def _check_warehouse(self, warehouse: str) -> int:
        time.sleep(0.3)
        return {"A": 10, "B": 5, "C": 15}.get(warehouse, 0)

    def _setup_functions(self):
        @self.tool(description="Look up product", fillers=["Looking that up..."])
        def get_product(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            product_id = args.get("product_id", "")
            time.sleep(0.5)  # Fast simulated lookup
            return SwaigFunctionResult(f"Product {product_id}: Widget Pro, $99.99")

        @self.tool(description="Check inventory", fillers=["Checking warehouses..."])
        def check_inventory(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            sku = args.get("sku", "")
            with concurrent.futures.ThreadPoolExecutor(max_workers=3) as ex:
                futures = [ex.submit(self._check_warehouse, w) for w in ["A", "B", "C"]]
                total = sum(f.result(timeout=2) for f in futures)
            return SwaigFunctionResult(f"Inventory for {sku}: {total} total")

        @self.tool(description="Calculate shipping", fillers=["Calculating..."])
        def calculate_shipping(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            zip_code = args.get("zip_code", "")
            weight = args.get("weight", 0.0)
            zone = self._zip_to_zone(zip_code)
            rate = self._get_shipping_rate(zone)
            total = rate + (weight * 0.50)
            return SwaigFunctionResult(f"Shipping to {zip_code}: ${total:.2f}")


if __name__ == "__main__":
    agent = OptimizedAgent()
    agent.run()

---

Part 4: Measure Improvement (20 min)

Benchmark Optimized Version

# Time function execution
time swaig-test optimized_agent.py --exec get_product --product_id "12345"

time swaig-test optimized_agent.py --exec check_inventory --sku "WIDGET-001"

time swaig-test optimized_agent.py --exec calculate_shipping \
  --zip_code "90210" --weight "2.5"

Record Results

Update performance_baseline.md:

# Performance Results

## Baseline (slow_agent.py)

| Function | Time (seconds) |
|----------|----------------|
| get_product | X.XX |
| check_inventory | X.XX |
| calculate_shipping | X.XX |

## Optimized (optimized_agent.py)

| Function | Time (seconds) | Improvement |
|----------|----------------|-------------|
| get_product | X.XX | XX% faster |
| check_inventory | X.XX | XX% faster |
| calculate_shipping | X.XX | XX% faster |

## Optimizations Applied

1. Concise prompts (reduced from ~150 words to ~30)
2. Parallel warehouse queries (3x speedup)
3. Cached shipping rates (instant after first call)
4. Timeout protection
5. Fillers for user experience
6. Optimized speech timing

---

Validation Checklist

• Baseline measurements recorded
• All three functions optimized
• Caching implemented for shipping
• Parallel execution for inventory
• Timeouts added to external calls
• Fillers configured for slow operations
• Speech timing configured
• Improvement measured and documented

---

Challenge Extension

Add request profiling decorator:

import functools
import time
import logging

logger = logging.getLogger(__name__)

def profile(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = (time.perf_counter() - start) * 1000
        logger.info(f"{func.__name__}: {elapsed:.2f}ms")
        return result
    return wrapper

---

Submission

Submit:

1. slow_agent.py (original)
2. optimized_agent.py (improved)
3. performance_baseline.md (measurements)

---

Complete Agent Code

Click to reveal complete solution

#!/usr/bin/env python3
"""Performance-optimized agent.

Lab 3.3 Deliverable (Part 2): Demonstrates performance best practices
including caching, parallel execution, timeouts, and fillers.
"""

import time
import concurrent.futures
from functools import lru_cache
from signalwire_agents import AgentBase, SwaigFunctionResult


class OptimizedAgent(AgentBase):
    """Agent with performance optimizations applied."""

    def __init__(self):
        super().__init__(name="optimized-agent")

        # OPTIMIZATION 1: Concise, focused prompts
        self._configure_prompts()
        self._configure_timing()

        self.add_language("English", "en-US", "rime.spore")
        self._setup_functions()

    def _configure_prompts(self):
        """Concise prompts for better performance."""
        self.prompt_add_section(
            "Role",
            "Customer service agent. Help with products, inventory, shipping."
        )

        self.prompt_add_section(
            "Guidelines",
            bullets=[
                "Be helpful and concise",
                "Confirm before taking actions",
                "Offer alternatives when needed"
            ]
        )

    def _configure_timing(self):
        """OPTIMIZATION 2: Optimized speech timing."""
        self.set_params({
            "end_of_speech_timeout": 400,  # Slightly faster
            "attention_timeout": 8000,
            "barge_min_words": 2  # Require 2 words to interrupt
        })

    # OPTIMIZATION 3: Cached shipping zone lookup
    @staticmethod
    @lru_cache(maxsize=100)
    def _get_shipping_zone(zip_prefix: str) -> str:
        """Cached zone lookup by zip prefix."""
        zones = {
            "9": "west", "8": "west", "7": "central",
            "6": "central", "5": "central", "4": "central",
            "3": "east", "2": "east", "1": "east", "0": "east"
        }
        return zones.get(zip_prefix, "standard")

    @staticmethod
    @lru_cache(maxsize=100)
    def _get_shipping_rate(zone: str) -> float:
        """Cached shipping rate by zone."""
        rates = {"west": 9.99, "central": 11.99, "east": 12.99, "standard": 14.99}
        return rates.get(zone, 14.99)

    def _check_warehouse(self, warehouse: str) -> int:
        """Simulated warehouse check."""
        time.sleep(0.3)  # Simulated API latency
        inventory = {"A": 10, "B": 5, "C": 15}
        return inventory.get(warehouse, 0)

    def _setup_functions(self):
        """Define optimized functions."""

        # OPTIMIZATION 4: Fillers for slow operations
        @self.tool(
            description="Look up product information",
            fillers=["Looking that up...", "Checking our catalog..."]
        )
        def get_product(product_id: str) -> SwaigFunctionResult:
            # Fast simulated lookup
            time.sleep(0.3)
            return SwaigFunctionResult(f"Product {product_id}: Widget Pro, $99.99")

        # OPTIMIZATION 5: Parallel warehouse checks
        @self.tool(
            description="Check inventory across warehouses",
            fillers=["Checking all warehouses...", "One moment..."]
        )
        def check_inventory(sku: str) -> SwaigFunctionResult:
            # Parallel execution - 3 calls in ~0.3s instead of ~1s
            with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
                futures = [
                    executor.submit(self._check_warehouse, "A"),
                    executor.submit(self._check_warehouse, "B"),
                    executor.submit(self._check_warehouse, "C")
                ]

                total = 0
                for future in concurrent.futures.as_completed(futures, timeout=2):
                    try:
                        total += future.result()
                    except Exception:
                        pass  # Skip failed warehouse

            return SwaigFunctionResult(f"Inventory for {sku}: {total} total")

        # OPTIMIZATION 6: Cached calculations
        @self.tool(
            description="Calculate shipping cost",
            fillers=["Calculating shipping..."]
        )
        def calculate_shipping(
            zip_code: str,
            weight: float
        ) -> SwaigFunctionResult:
            # Use cached lookups
            zip_prefix = zip_code[0] if zip_code else "5"
            zone = self._get_shipping_zone(zip_prefix)
            base_rate = self._get_shipping_rate(zone)

            # Simple weight calculation
            total = base_rate + (weight * 0.50)

            return SwaigFunctionResult(
                f"Shipping to {zip_code} ({zone} zone): ${total:.2f}"
            )

        @self.tool(description="Get shipping zone for zip code")
        def get_shipping_zone(zip_code: str) -> SwaigFunctionResult:
            # Instant from cache after first call
            zone = self._get_shipping_zone(zip_code[0] if zip_code else "5")
            return SwaigFunctionResult(f"Zip {zip_code} is in {zone} zone")

        # OPTIMIZATION 7: Timeout protection
        @self.tool(
            description="Check external service status",
            fillers=["Checking service status..."]
        )
        def check_external_service() -> SwaigFunctionResult:
            try:
                # Reasonable timeout
                import requests
                response = requests.get(
                    "https://httpbin.org/get",
                    timeout=2  # 2 second timeout
                )
                return SwaigFunctionResult("External service: Online")
            except requests.Timeout:
                return SwaigFunctionResult(
                    "External service check timed out. Functionality may be limited."
                )
            except Exception:
                return SwaigFunctionResult(
                    "Could not check external service. Please try again."
                )


if __name__ == "__main__":
    agent = OptimizedAgent()
    agent.run()