Subclassing

Building Programs by Subclassing

In Lesson 1a, you learned to build programs using the Functional API. Now, let's explore a more advanced approach: subclassing the Program class.

When to Use Subclassing

Subclassing is useful when you need:

• Custom logic in your program's execution flow
• Stateful behavior that persists across calls
• Reusable components that can be shared across projects
• Full control over serialization and deserialization

Think of it like the difference between using a pre-built function vs writing your own class in object-oriented programming.

The Subclassing Pattern

When you subclass synalinks.Program, you must implement:

1. __init__(): Define your modules and initialize state
2. call(): Define how data flows through your modules
3. get_config(): Define how to save your program (serialization)
4. from_config(): Define how to load your program (deserialization)

classDiagram
    class Program {
        +__init__()
        +call(inputs)
        +get_config()
        +from_config(config)
    }
    class MyProgram {
        +generator
        +__init__(language_model)
        +call(inputs)
        +get_config()
        +from_config(config)
    }
    Program <|-- MyProgram

class MyProgram(synalinks.Program):

    def __init__(self, language_model=None):
        super().__init__()  # Always call super().__init__()!
        self.generator = synalinks.Generator(
            data_model=OutputModel,
            language_model=language_model,
        )

    async def call(self, inputs, training=False):
        # Define the forward pass
        return await self.generator(inputs)

    def get_config(self):
        # Return a dict with everything needed to recreate this program
        return {"language_model": serialize(self.language_model)}

    @classmethod
    def from_config(cls, config):
        # Recreate the program from the config dict
        return cls(language_model=deserialize(config["language_model"]))

Important: The build() Method

Unlike the Functional API, subclassed programs need to be built before first use. This tells Synalinks what input type to expect:

program = MyProgram(language_model=lm)
await program.build(InputDataModel)  # <-- Required before first call!

If you used a subclassed module inside a functional API program, your module is built automatically!

Complete Example

import asyncio
from dotenv import load_dotenv
import synalinks

class Query(synalinks.DataModel):
    query: str = synalinks.Field(description="The user query")

class AnswerWithThinking(synalinks.DataModel):
    thinking: str = synalinks.Field(description="Your step by step thinking")
    answer: str = synalinks.Field(description="The correct answer")

class ChainOfThought(synalinks.Program):
    def __init__(self, language_model=None):
        super().__init__()
        self.answer_generator = synalinks.Generator(
            data_model=AnswerWithThinking,
            language_model=language_model,
        )

    async def call(self, inputs, training=False):
        return await self.answer_generator(inputs)

    def get_config(self):
        return {
            "language_model": synalinks.saving.serialize_synalinks_object(
                self.language_model
            )
        }

    @classmethod
    def from_config(cls, config):
        language_model = synalinks.saving.deserialize_synalinks_object(
            config.pop("language_model")
        )
        return cls(language_model=language_model)

async def main():
    load_dotenv()
    language_model = synalinks.LanguageModel(model="openai/gpt-4.1")

    program = ChainOfThought(language_model=language_model)
    await program.build(Query)  # Required before first call!

    result = await program(Query(query="What is 15% of 80?"))
    print(f"Answer: {result['answer']}")

asyncio.run(main())

Key Takeaways

• Subclassing: Inherit from synalinks.Program for full control over program behavior and custom logic.
• Four Methods: Implement __init__(), call(), get_config(), and from_config() for a complete subclassed program.
• Build Required: Call await program.build(InputDataModel) before first use when using standalone subclassed programs.
• Serialization: get_config() and from_config() enable saving and loading your custom programs.

Program Visualization

API References

• DataModel
• LanguageModel
• Generator
• Program

AnswerWithThinking

Bases: DataModel

The output from our program - reasoning + final answer.

Source code in examples/1b_subclassing.py

class AnswerWithThinking(synalinks.DataModel):
    """The output from our program - reasoning + final answer."""

    thinking: str = synalinks.Field(
        description="Your step by step thinking",
    )
    answer: str = synalinks.Field(
        description="The correct answer",
    )

ChainOfThought

Bases: Program

A program that answers questions with step-by-step reasoning.

Note: The first line of the docstring becomes the program's description if not explicitly provided in super().init().

Source code in examples/1b_subclassing.py

class ChainOfThought(synalinks.Program):
    """A program that answers questions with step-by-step reasoning.

    Note: The first line of the docstring becomes the program's description
    if not explicitly provided in super().__init__().
    """

    def __init__(self, language_model=None):
        # Always call super().__init__() first!
        super().__init__()

        # Define the modules your program will use
        # These are like instance variables in regular Python classes
        self.answer_generator = synalinks.Generator(
            data_model=AnswerWithThinking,
            language_model=language_model,
        )

    async def call(
        self, inputs: synalinks.JsonDataModel, training: bool = False
    ) -> synalinks.JsonDataModel:
        """Define how data flows through your program.

        This method is called when you do `await program(input_data)`.

        Args:
            inputs (JsonDataModel): The input data (will be a Query instance)
            training (bool): Whether we're in training mode (for optimization)

        Returns:
            JsonDataModel: The output data (will be an AnswerWithThinking instance)
        """
        # In this simple case, we just pass inputs through one module
        # More complex programs might have multiple steps, conditionals, etc.
        result = await self.answer_generator(inputs)
        return result

    def get_config(self):
        """Return configuration needed to recreate this program.

        This is called when saving the program to disk.
        """
        config = {
            "name": self.name,
            "description": self.description,
            "trainable": self.trainable,
        }
        # Serialize the language model so it can be saved
        language_model_config = {
            "language_model": synalinks.saving.serialize_synalinks_object(
                self.language_model
            )
        }
        return {**config, **language_model_config}

    @classmethod
    def from_config(cls, config):
        """Recreate the program from a configuration dict.

        This is called when loading the program from disk.
        """
        # Deserialize the language model first
        language_model = synalinks.saving.deserialize_synalinks_object(
            config.pop("language_model")
        )
        return cls(language_model=language_model, **config)

call(inputs, training=False) async

Define how data flows through your program.

This method is called when you do await program(input_data).

Parameters:

Name	Type	Description	Default
inputs	JsonDataModel	The input data (will be a Query instance)	required
training	bool	Whether we're in training mode (for optimization)	False

Returns:

Name	Type	Description
JsonDataModel	JsonDataModel	The output data (will be an AnswerWithThinking instance)

Source code in examples/1b_subclassing.py

async def call(
    self, inputs: synalinks.JsonDataModel, training: bool = False
) -> synalinks.JsonDataModel:
    """Define how data flows through your program.

    This method is called when you do `await program(input_data)`.

    Args:
        inputs (JsonDataModel): The input data (will be a Query instance)
        training (bool): Whether we're in training mode (for optimization)

    Returns:
        JsonDataModel: The output data (will be an AnswerWithThinking instance)
    """
    # In this simple case, we just pass inputs through one module
    # More complex programs might have multiple steps, conditionals, etc.
    result = await self.answer_generator(inputs)
    return result

from_config(config) classmethod

Recreate the program from a configuration dict.

This is called when loading the program from disk.

Source code in examples/1b_subclassing.py

@classmethod
def from_config(cls, config):
    """Recreate the program from a configuration dict.

    This is called when loading the program from disk.
    """
    # Deserialize the language model first
    language_model = synalinks.saving.deserialize_synalinks_object(
        config.pop("language_model")
    )
    return cls(language_model=language_model, **config)

get_config()

Return configuration needed to recreate this program.

This is called when saving the program to disk.

Source code in examples/1b_subclassing.py

def get_config(self):
    """Return configuration needed to recreate this program.

    This is called when saving the program to disk.
    """
    config = {
        "name": self.name,
        "description": self.description,
        "trainable": self.trainable,
    }
    # Serialize the language model so it can be saved
    language_model_config = {
        "language_model": synalinks.saving.serialize_synalinks_object(
            self.language_model
        )
    }
    return {**config, **language_model_config}

Query

Bases: DataModel

The input to our program - a user's question.

Source code in examples/1b_subclassing.py

class Query(synalinks.DataModel):
    """The input to our program - a user's question."""

    query: str = synalinks.Field(
        description="The user query",
    )