Lab L3.2: Knowledge and RAG

🎯 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

Integrate knowledge bases
Implement RAG patterns
Optimize retrieval

How to Complete This Lab

Accept the Assignment — Click the GitHub Classroom link above
Clone Your Repo — git clone <your-repo-url>
Read the README — Your repo has detailed requirements and grading criteria
Write Your Code — Implement the solution in solution/agent.py
Test Locally — Use swaig-test to verify your agent works
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

Build a technical support agent for a software product with:

Product documentation (FAQ, user guide)
Live API for license/subscription status
Accurate, sourced answers

Part 1: Prepare Knowledge Base (25 min)

Task

Create sample documentation and build a search index.

Create Documentation

Create docs/ directory with sample files:

docs/faq.md

# Frequently Asked Questions

## Installation

### How do I install the software?
Download the installer from our website and run it. Windows users
should right-click and select "Run as Administrator."

### What are the system requirements?

- Windows 10 or later, macOS 10.15+, or Ubuntu 20.04+
- 8GB RAM minimum, 16GB recommended
- 500MB disk space
- Internet connection for activation

### How do I activate my license?
Go to Settings > License > Activate. Enter your license key
(found in your purchase confirmation email) and click Activate.

## Troubleshooting

### The software won't start
1. Restart your computer
2. Run as administrator (Windows)
3. Check for conflicting software
4. Reinstall if issue persists

### I'm getting a license error
Your license may have expired or be used on too many devices.
Check your account at account.example.com or contact support.

### The software is running slowly
1. Close other applications
2. Check you meet system requirements
3. Clear the cache (Settings > Advanced > Clear Cache)
4. Update to the latest version

docs/features.md

# Feature Guide

## Dashboard
The main dashboard shows your recent projects and activity.
Click any project to open it.

## Projects
Create new projects with File > New Project.
Projects auto-save every 5 minutes.

## Export
Export your work in multiple formats:
- PDF (File > Export > PDF)
- PNG (File > Export > Image)
- SVG (File > Export > Vector)

## Collaboration
Share projects with team members:
1. Click Share in the top right
2. Enter email addresses
3. Choose permission level (View/Edit)
4. Click Send Invitation

## Keyboard Shortcuts
- Ctrl+S: Save
- Ctrl+Z: Undo
- Ctrl+Shift+Z: Redo
- Ctrl+N: New project
- Ctrl+O: Open project

Build Index

# Using sw-search to build index
sw-search index \
  --source ./docs/ \
  --output ./support_kb.index \
  --chunk-size 300 \
  --chunk-overlap 50

# Verify index
sw-search query \
  --index ./support_kb.index \
  --query "How do I install?"

Part 2: Create RAG Agent (35 min)

Task

Build an agent that uses the knowledge index.

Deliverable: `support_agent.py`

#!/usr/bin/env python3
"""Technical support agent with RAG."""

import os
from signalwire_agents import AgentBase, SwaigFunctionResult


class SupportAgent(AgentBase):
    def __init__(self, kb_path: str):
        super().__init__(name="support-agent")

        self.prompt_add_section(
            "Role",
            "Technical support agent for ExampleSoft. "
            "Answer questions using the knowledge base. "
            "If information isn't in the knowledge base, say so."
        )

        self.prompt_add_section(
            "Instructions",
            bullets=[
                "Search the knowledge base before answering",
                "Cite sources when providing information",
                "Admit when you don't have information",
                "Offer to escalate complex issues"
            ]
        )

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

        # Add search skill with knowledge base
        self.add_skill(
            "search",
            {
                "index_path": kb_path,
                "count": 3,
                "distance": 0.7
            }
        )

        self._setup_functions()

    def _setup_functions(self):
        @self.tool(description="Get installation help")
        def installation_help(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            # This uses the search skill implicitly
            return SwaigFunctionResult(
                "For installation: Download from our website and run the installer. "
                "Windows users should run as Administrator. "
                "Need system requirements or activation help?"
            )

        @self.tool(
            description="Troubleshoot an issue",
            parameters={
                "type": "object",
                "properties": {
                    "issue": {
                        "type": "string",
                        "description": "The issue to troubleshoot"
                    }
                },
                "required": ["issue"]
            }
        )
        def troubleshoot(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            issue = args.get("issue", "")
            # Search would provide context here
            common_issues = {
                "start": "Try: 1) Restart computer 2) Run as admin 3) Reinstall",
                "license": "Check license at account.example.com or contact support",
                "slow": "Try: 1) Close other apps 2) Clear cache 3) Update software"
            }

            for keyword, solution in common_issues.items():
                if keyword in issue.lower():
                    return SwaigFunctionResult(solution)

            return SwaigFunctionResult(
                "I'll search our knowledge base for that issue. "
                "If I can't find a solution, I can create a support ticket."
            )

        @self.tool(
            description="Create support ticket",
            parameters={
                "type": "object",
                "properties": {
                    "description": {"type": "string"},
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"]
                    }
                },
                "required": ["description"]
            }
        )
        def create_ticket(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            description = args.get("description", "")
            priority = args.get("priority", "medium")
            import datetime
            ticket_id = f"TKT-{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"

            return (
                SwaigFunctionResult(
                    f"Created ticket {ticket_id}. "
                    "Our team will respond within 24 hours."
                )
                .update_global_data( {
                    "ticket_id": ticket_id,
                    "ticket_description": description,
                    "ticket_priority": priority
                })
            )


if __name__ == "__main__":
    kb_path = os.getenv("KB_PATH", "./support_kb.index")
    agent = SupportAgent(kb_path=kb_path)
    agent.run()

Part 3: Add Live Data Integration (20 min)

Task

Add DataMap for real-time license status.

Add to Agent

from signalwire_agents import AgentBase, DataMap, SwaigFunctionResult


def _setup_live_data(self):
    """Add DataMap for live API queries."""

    # License status lookup
    check_license = (
        DataMap("check_license")
        .description("Check customer license status")
        .parameter("license_key", "string", "Customer's license key", required=True)
        .webhook(
            "GET",
            "https://api.example.com/licenses/${enc:args.license_key}",
            headers={"Authorization": "Bearer ${env.API_KEY}"}
        )
        .output(
            SwaigFunctionResult(
                "License ${args.license_key}: ${response.status}. "
                "Expires: ${response.expires}. "
                "Seats: ${response.seats_used}/${response.seats_total}."
            )
            .update_global_data( {
                "license_status": "${response.status}",
                "license_expires": "${response.expires}"
            })
        )
        .fallback_output(SwaigFunctionResult(
            "I couldn't find that license. Please verify the key and try again."
        ))
    )
    self.register_swaig_function(check_license.to_swaig_function())

    # Version check
    check_version = (
        DataMap("check_version")
        .description("Check if customer has latest version")
        .parameter("current_version", "string", "Customer's current version", required=True)
        .webhook("GET", "https://api.example.com/versions/latest")
        .output(SwaigFunctionResult(
            "Latest version is ${response.latest}. You have ${args.current_version}."
        ))
    )
    self.register_swaig_function(check_version.to_swaig_function())

Update Constructor

def __init__(self, kb_path: str):
    super().__init__(name="support-agent")

    # ... existing code ...

    self._setup_live_data()  # Add this line

Part 4: Testing (10 min)

Test Commands

# Test SWML output
swaig-test support_agent.py --dump-swml

# Test knowledge search
swaig-test support_agent.py --exec troubleshoot \
  --issue "software won't start"

# Test ticket creation
swaig-test support_agent.py --exec create_ticket \
  --description "Cannot export to PDF" \
  --priority "medium"

# Verify DataMaps in SWML
swaig-test support_agent.py --dump-swml | jq '.sections.main[0].ai.SWAIG.functions[] | select(.function | contains("check_"))'

Test Scenarios

Query	Expected Source	Expected Response
“How do I install?”	Knowledge base	Installation steps
“What’s my license status?”	Live API	Status from API
“Software is slow”	Knowledge base	Troubleshooting steps
“Unknown error XYZ”	Ticket creation	Create ticket offer

Validation Checklist

Knowledge index built from docs
Search skill configured correctly
Troubleshooting uses knowledge base
DataMap for license lookup working
Version check DataMap working
Ticket creation function implemented
Appropriate fallback for unknown issues

Challenge Extension

Add a feedback loop for knowledge base improvement:

@self.tool(
    description="Report that an answer was incorrect or incomplete",
    parameters={
        "type": "object",
        "properties": {
            "question": {"type": "string"},
            "feedback": {"type": "string"}
        },
        "required": ["question", "feedback"]
    }
)
def report_feedback(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
    question = args.get("question", "")
    feedback = args.get("feedback", "")
    # Log feedback for KB improvement
    log_feedback(question, feedback)

    return SwaigFunctionResult(
        "Thank you for the feedback. I've logged this for our team to review."
    )

Submission

Submit:

docs/ directory with documentation files
support_agent.py with RAG implementation
Screenshot of successful knowledge query

Complete Agent Code

Click to reveal complete solution

#!/usr/bin/env python3
"""Technical support agent with RAG capabilities.

Lab 3.2 Deliverable: Demonstrates knowledge management with search skill,
live data integration via DataMap, and ticket creation.
"""

import os
from datetime import datetime
from signalwire_agents import AgentBase, SwaigFunctionResult
from signalwire_agents.core.data_map import DataMap
from signalwire_agents.core.function_result import SwaigFunctionResult as DmResult


class SupportAgent(AgentBase):
    """Technical support agent with RAG and live data integration."""

    def __init__(self, kb_path: str = None):
        super().__init__(name="support-agent")

        self.prompt_add_section(
            "Role",
            "Technical support agent for ExampleSoft. "
            "Answer questions using the knowledge base. "
            "If information isn't in the knowledge base, say so."
        )

        self.prompt_add_section(
            "Instructions",
            bullets=[
                "Search the knowledge base before answering",
                "Cite sources when providing information",
                "Admit when you don't have information",
                "Offer to escalate complex issues",
                "Create tickets for unresolved problems"
            ]
        )

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

        # Add search skill with knowledge base (if available)
        if kb_path and os.path.exists(kb_path):
            self.add_skill(
                "search",
                {
                    "index_path": kb_path,
                    "count": 3,
                    "distance": 0.7
                }
            )

        self._setup_datamaps()
        self._setup_functions()

    def _setup_datamaps(self):
        """Configure DataMaps for live data queries."""

        # License status lookup
        check_license_dm = (
            DataMap("check_license")
            .description("Check customer license status by license key")
            .parameter("license_key", "string", "Customer's license key", required=True)
            .webhook(
                "GET",
                "https://api.example.com/licenses/${args.license_key}"
            )
            .output(DmResult(
                "License status: ${status}. Expires: ${expires}."
            ))
            .fallback_output(DmResult(
                "I couldn't find that license key. Please verify and try again."
            ))
        )
        self.register_swaig_function(check_license_dm.to_swaig_function())

        # Version check
        check_version_dm = (
            DataMap("check_version")
            .description("Check if a software version is current")
            .parameter("current_version", "string", "Customer's current version", required=True)
            .webhook(
                "GET",
                "https://api.example.com/versions/latest"
            )
            .output(DmResult(
                "Latest version is ${latest}. You have ${args.current_version}."
            ))
            .fallback_output(DmResult(
                "Unable to check version information."
            ))
        )
        self.register_swaig_function(check_version_dm.to_swaig_function())

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

        @self.tool(description="Get installation help")
        def installation_help(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            return SwaigFunctionResult(
                "For installation: Download from our website and run the installer. "
                "Windows users should run as Administrator. "
                "System requirements: Windows 10+, macOS 10.15+, or Ubuntu 20.04+. "
                "8GB RAM minimum, 500MB disk space. "
                "Need help with activation or system requirements?"
            )

        @self.tool(
            description="Troubleshoot a specific issue",
            parameters={
                "type": "object",
                "properties": {
                    "issue": {
                        "type": "string",
                        "description": "Description of the issue"
                    }
                },
                "required": ["issue"]
            },
            fillers=["Let me search our knowledge base..."]
        )
        def troubleshoot(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Troubleshoot common issues."""
            issue = args.get("issue", "")
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            issue_lower = issue.lower()

            # Common issue patterns
            if any(word in issue_lower for word in ["start", "launch", "open", "won't run"]):
                return SwaigFunctionResult(
                    "For startup issues, try: "
                    "1) Restart your computer, "
                    "2) Run as administrator (Windows), "
                    "3) Check for conflicting software, "
                    "4) Reinstall if needed. "
                    "Did any of these help?"
                )

            if any(word in issue_lower for word in ["license", "activate", "key"]):
                return SwaigFunctionResult(
                    "For license issues: "
                    "Check your license at account.example.com. "
                    "Make sure it hasn't expired and isn't used on too many devices. "
                    "Would you like me to check your license status?"
                )

            if any(word in issue_lower for word in ["slow", "performance", "lag"]):
                return SwaigFunctionResult(
                    "For performance issues: "
                    "1) Close other applications, "
                    "2) Verify system requirements (8GB RAM minimum), "
                    "3) Clear cache (Settings > Advanced > Clear Cache), "
                    "4) Update to latest version. "
                    "Which of these would you like help with?"
                )

            if any(word in issue_lower for word in ["password", "login", "forgot"]):
                return SwaigFunctionResult(
                    "For password reset: "
                    "Click 'Forgot Password' on the login screen, "
                    "enter your email, and check your inbox for a reset link. "
                    "The link arrives within 5 minutes. "
                    "Still having trouble?"
                )

            # Generic response for unknown issues
            return SwaigFunctionResult(
                "I'll search our knowledge base for that issue. "
                "If I can't find a solution, I can create a support ticket. "
                "Could you provide more details about what's happening?"
            )

        @self.tool(
            description="Create a support ticket",
            parameters={
                "type": "object",
                "properties": {
                    "description": {
                        "type": "string",
                        "description": "Detailed description of the issue"
                    },
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"],
                        "description": "Ticket priority"
                    }
                },
                "required": ["description"]
            }
        )
        def create_ticket(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            description = args.get("description", "")
            priority = args.get("priority", "medium")
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            ticket_id = f"TKT-{datetime.now().strftime('%Y%m%d%H%M%S')}"

            return (
                SwaigFunctionResult(
                    f"Created ticket {ticket_id} with {priority} priority. "
                    "Our team will respond within 24 hours for standard issues, "
                    "or 4 hours for high priority. "
                    "Is there anything else I can help with?"
                )
                .update_global_data({
                    "ticket_id": ticket_id,
                    "ticket_description": description,
                    "ticket_priority": priority,
                    "ticket_created": datetime.now().isoformat()
                })
            )

        @self.tool(
            description="Report incorrect or incomplete answer",
            parameters={
                "type": "object",
                "properties": {
                    "question": {
                        "type": "string",
                        "description": "The original question"
                    },
                    "feedback": {
                        "type": "string",
                        "description": "What was wrong or missing"
                    }
                },
                "required": ["question", "feedback"]
            }
        )
        def report_feedback(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Log feedback for knowledge base improvement."""
            question = args.get("question", "")
            feedback = args.get("feedback", "")
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            return (
                SwaigFunctionResult(
                    "Thank you for the feedback. I've logged this for our team to review. "
                    "Your input helps us improve. "
                    "Would you like me to create a ticket for further assistance?"
                )
                .update_global_data({
                    "feedback_question": question,
                    "feedback_content": feedback,
                    "feedback_time": datetime.now().isoformat()
                })
            )

        @self.tool(description="Escalate to human support")
        def escalate_to_human(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Transfer to human support agent."""
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            ticket_id = global_data.get("ticket_id")

            context = f"Ticket {ticket_id}" if ticket_id else "New inquiry"

            return (
                SwaigFunctionResult(
                    "I'm connecting you with a human support agent. "
                    "Please hold while I transfer your call.",
                    post_process=True
                )
                .update_global_data({
                    "escalated": True,
                    "escalation_time": datetime.now().isoformat()
                })
                .swml_transfer("/human-support", "Goodbye!", final=True)
            )


if __name__ == "__main__":
    kb_path = os.getenv("KB_PATH", "./support_kb.index")
    agent = SupportAgent(kb_path=kb_path)
    agent.run()

Lab L3.2: Knowledge and RAG

Objectives

How to Complete This Lab

Key Concepts

Scenario

Part 1: Prepare Knowledge Base (25 min)

Task

Create Documentation

Build Index

Part 2: Create RAG Agent (35 min)

Task

Deliverable: support_agent.py

Part 3: Add Live Data Integration (20 min)

Task

Add to Agent

Update Constructor

Part 4: Testing (10 min)

Test Commands

Test Scenarios

Validation Checklist

Challenge Extension

Submission

Complete Agent Code

Deliverable: `support_agent.py`