Synalinks Modules Catalog

Synalinks Modules Catalog

Generation Modules

Generator

Core LLM generation module with structured output.

outputs = await synalinks.Generator(
    data_model=Answer,          # Required: output schema
    language_model=lm,          # Required: LanguageModel instance
    prompt_template=None,       # Custom Jinja2 template
    instructions=None,          # MUST be a string or None (not a list!)
    examples=None,              # Few-shot examples (list of input/output dict tuple) optional. (generated by the training)
    return_inputs=False,        # Include inputs in output
    use_inputs_schema=False,    # Show input schema in prompt. False is better in most cases
    use_outputs_schema=False,   # Show output schema in prompt. False is better in most cases
    streaming=False,            # Enable streaming.
    temperature=None,           # LLM temperature (higher = more random)
    name=None,
    description=None,
    trainable=True,
)(inputs)

Trainable Variables:

• Instructions (optimized by optimizer)
• Examples (selected/generated by optimizer)

ChainOfThought

Generator that automatically adds a thinking step.

class Answer(synalinks.DataModel):
    answer: str = synalinks.Field(description="The answer")

outputs = await synalinks.ChainOfThought(
    data_model=Answer,
    language_model=lm,
    return_inputs=True,
)(inputs)
# Output includes: inputs + thinking + answer

SelfCritique

Self-evaluation module that generates critique and reward.

outputs = await synalinks.SelfCritique(
    language_model=lm,		# LanguageModel instance
    return_reward=True,         # Include reward score between 0.0 and 1.0
    return_inputs=True,		#  Control if the inputs is forwarded or not
)(previous_output)
# Output includes: inputs + critique + reward

Control Flow Modules

Decision

Single-label classification for routing decisions.

outputs = await synalinks.Decision(
    question="What type of query is this?",
    labels=["factual", "opinion", "creative"],
    language_model=lm,
    return_inputs=True,
)(inputs)
# Output includes: inputs + label

Branch

Conditional routing to different modules based on decision.

(output1, output2) = await synalinks.Branch(
    question="Query difficulty?",
    labels=["easy", "hard"],
    branches=[
        synalinks.Generator(data_model=SimpleAnswer, language_model=lm),
        synalinks.Generator(data_model=DetailedAnswer, language_model=lm),
    ],
    language_model=lm,
    return_decision=False,      # Include decision in outputs
    inject_decision=False,      # Inject decision into branch inputs
)(inputs)
# Non-selected branches return None
# Use | operator to merge: result = output1 | output2

Key behaviors:

• Non-activated branches return None (not executed)
• Each branch module is optimized separately during training (specialized)
• Labels constrain LLM output - prevents hallucination

Identity

Pass-through module (no-op).

outputs = await synalinks.Identity()(inputs)

Agent Modules

FunctionCallingAgent

Autonomous agent with parallel tool calling capabilities.

@synalinks.utils.register_synalinks_serializable()
async def search(query: str):
    """Search the web for information.

    Args:
        query: Search query string
    """
    return {"results": [...], "log": "Success"}

tools = [synalinks.Tool(search)]

outputs = await synalinks.FunctionCallingAgent(
    data_model=FinalAnswer,     # Final output schema
    tools=tools,                # List of Tool instances
    language_model=lm,
    max_iterations=5,           # Max tool calls
    autonomous=True,            # Run autonomously
    return_inputs_with_trajectory=True,  # Include full trajectory
)(inputs)

Tool Definition:

synalinks.Tool(
    function,                   # Async function
    name=None,                  # Override function name
    description=None,           # Override docstring
)

ToolCalling

Single tool selection and execution.

outputs = await synalinks.ToolCalling(
    tools=tools,
    language_model=lm,
)(inputs)

Knowledge Modules

EmbedKnowledge

Generate embeddings for data models to enable vector similarity search.

outputs = await synalinks.EmbedKnowledge(
    embedding_model=embedding_model,
    in_mask=["content"],        # Fields to embed (keep only these)
    # out_mask=["id"],          # Or exclude specific fields
)(inputs)

Note: Each data model should have exactly one field for embedding after masking.

UpdateKnowledge

Store data models in the knowledge base.

outputs = await synalinks.UpdateKnowledge(
    knowledge_base=knowledge_base,
)(data_model_input)

Uses the first field as primary key for upsert operations.

RetrieveKnowledge

Retrieve relevant records using LM-generated search queries.

outputs = await synalinks.RetrieveKnowledge(
    knowledge_base=knowledge_base,
    language_model=lm,
    search_type="hybrid",       # "similarity", "fulltext", or "hybrid"
    k=10,                       # Number of results
    return_inputs=True,
    return_query=True,          # Include generated search query
)(inputs)

Search Types:

• "similarity": Vector-based semantic search (requires embedding_model)
• "fulltext": BM25-based full-text search
• "hybrid": Combines both using Reciprocal Rank Fusion (default)

Synthesis Modules (Test-time adaptation)

PythonSynthesis

Generate and execute Python code.

outputs = await synalinks.PythonSynthesis(
    data_model=Result,
    language_model=lm,
    allowed_modules=["math", "datetime"],
)(inputs)

SequentialPlanSynthesis

Generate step-by-step execution plans.

outputs = await synalinks.SequentialPlanSynthesis(
    data_model=Plan,
    language_model=lm,
)(inputs)

Operators (synalinks.ops)

Concatenate

# Function form
result = await synalinks.ops.concat(x1, x2, name="combined")

# Operator form
result = x1 + x2

And

result = await synalinks.ops.logical_and(x1, x2)
result = x1 & x2

Or

result = await synalinks.ops.logical_or(x1, x2)
result = x1 | x2

Xor

result = await synalinks.ops.logical_xor(x1, x2)
result = x1 ^ x2

Not

result = await synalinks.ops.logical_not(x)
result = ~x

Creating Custom Modules

Subclass Template

class MyModule(synalinks.Module):
    def __init__(
        self,
        language_model=None,
        return_inputs=False,
        name=None,
        description=None,
        trainable=True,
    ):
        super().__init__(name=name, description=description, trainable=trainable)
        self.language_model = language_model
        self.return_inputs = return_inputs

        # Initialize sub-modules
        # Note: instructions must be a string, not a list
        self.generator = synalinks.Generator(
            data_model=OutputSchema,
            language_model=language_model,
            instructions="Your instructions here as a single string.",
            return_inputs=return_inputs,
        )

    async def call(self, inputs, training=False):
        """Core computation logic."""
        if not inputs:
            return None  # Support logical flows (XOR guards)

        result = await self.generator(inputs, training=training)
        return result

    async def compute_output_spec(self, inputs, training=False):
        """Define output schema (required for custom modules)."""
        return await self.generator(inputs)

    def to_config(self):
        """Serialization config for Module subclasses."""
        return {
            "return_inputs": self.return_inputs,
            "name": self.name,
            "description": self.description,
            "trainable": self.trainable,
            "language_model": synalinks.saving.serialize_synalinks_object(
                self.language_model
            ),
        }

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

Note: Module subclasses use to_config() while Program subclasses use get_config().

Adding Variables

class MyModule(synalinks.Module):
    def __init__(self, ...):
        super().__init__(...)

        # Add trainable variable
        self.my_var = self.add_variable(
            name="my_variable",
            initializer=synalinks.initializers.Constant(value="initial"),
            trainable=True,
        )

Module Composition Patterns

Sequential Chain

inputs = synalinks.Input(data_model=Query)
x1 = await ModuleA(language_model=lm)(inputs)
x2 = await ModuleB(language_model=lm)(x1)
outputs = await ModuleC(language_model=lm)(x2)

Parallel Branches

inputs = synalinks.Input(data_model=Query)
x1 = await ModuleA(language_model=lm)(inputs)
x2 = await ModuleB(language_model=lm)(inputs)
outputs = x1 + x2  # Concatenate results

Conditional Flow

inputs = synalinks.Input(data_model=Query)
(branch1, branch2) = await synalinks.Branch(
    question="...",
    labels=["a", "b"],
    branches=[ModuleA(...), ModuleB(...)],
)(inputs)
outputs = branch1 | branch2  # Merge branches

Self-Consistency

inputs = synalinks.Input(data_model=Query)
# Multiple generators with temperature for diversity
x1 = await synalinks.Generator(data_model=Answer, language_model=lm, temperature=1.0)(inputs)
x2 = await synalinks.Generator(data_model=Answer, language_model=lm, temperature=1.0)(inputs)
combined = inputs & x1 & x2
critique = await synalinks.SelfCritique(language_model=lm)(combined)
outputs = await synalinks.Generator(
    data_model=FinalAnswer,
    language_model=lm,
    instructions="Critically analyze the given answers to produce the final answer.",
)(critique)

Input Guard (XOR Pattern)

inputs = synalinks.Input(data_model=synalinks.ChatMessages)
warning = await InputGuard()(inputs)
guarded_inputs = warning ^ inputs       # Bypass if warning exists
answer = await synalinks.Generator(language_model=lm)(guarded_inputs)
outputs = warning | answer              # Return warning or answer

Output Guard (XOR Pattern)

inputs = synalinks.Input(data_model=synalinks.ChatMessages)
answer = await synalinks.Generator(language_model=lm)(inputs)
warning = await OutputGuard()(answer)
outputs = (answer ^ warning) | warning  # Replace if warning exists