Chapter 7 · Section 2

Adapters & Tools

Bridge DSPy with databases, APIs, file systems, and specialized tools.

~20 min read

Introduction

Adapters make DSPy modular and extensive. They handle communication with external resources like databases, APIs, and file systems, ensuring your core logic remains clean.

Built-in vs. Custom Adapters

Database Adapter (Example)

A typical PostgreSQL adapter pattern:

Python
class PostgreSQLAdapter(dspy.Adapter):
    def __init__(self, connection_string):
        super().__init__()
        self.connection_string = connection_string

    def query(self, sql, params=None):
        # Implementation of query execution...
        pass

Creating Custom Adapters

File System Adapter

Manage file operations seamlessly within your DSPy pipeline:

Python
import dspy
from pathlib import Path

class FileSystemAdapter(dspy.Adapter):
    def __init__(self, base_path="."):
        super().__init__()
        self.base_path = Path(base_path)

    def read_file(self, filename):
        return (self.base_path / filename).read_text()

    def write_file(self, filename, content):
        (self.base_path / filename).write_text(content)

API Adapter

Interact with external REST APIs:

Python
import requests

class APIAdapter(dspy.Adapter):
    def __init__(self, base_url):
        super().__init__()
        self.base_url = base_url

    def get(self, endpoint):
        return requests.get(f"{self.base_url}/{endpoint}").json()

Specialized Tools

Tools encapsulate specific functionality like calculations or text processing.

Calculator Tool

Python
class CalculatorTool(dspy.Tool):
    def calculate(self, expression):
        # Safe evaluation logic...
        pass

Integration Example: Tool-Enabled Agent

Combining adapters and tools in a powerful agent:

Python
class ToolEnabledAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        self.tools = {
            'calculator': CalculatorTool(),
            'file_system': FileSystemAdapter()
        }
        self.decide = dspy.Predict("task -> tool, params")

    def forward(self, task):
        decision = self.decide(task=task)
        # Execute selected tool...
        pass