synalinks

Synalinks Framework

Synalinks is an open-source Keras-inspired framework for building neuro-symbolic LLM applications with in-context reinforcement learning.

Core Concepts

• DataModel: Pydantic-style schema defining structured I/O (replaces tensors)
• Module: Computational unit processing JSON data (replaces layers)
• Program: DAG of modules with conditional logic (replaces models)
• Rewards: Guide training (maximize reward, not minimize loss)
• Optimizers: Update prompts/examples via LLM reasoning (no gradients)

Quick Start

import synalinks
import asyncio

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

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

async def main():
    lm = synalinks.LanguageModel(model="ollama/mistral")

    inputs = synalinks.Input(data_model=Query)
    outputs = await synalinks.Generator(
        data_model=Answer,
        language_model=lm,
    )(inputs)

    program = synalinks.Program(
        inputs=inputs,
        outputs=outputs,
        name="simple_qa",
        description="A simple Q&A program",
    )

    result = await program(Query(query="What is the capital of France?"))
    print(result.prettify_json())

asyncio.run(main())

Four Ways to Build Programs

1. Functional API (Recommended for most cases)

inputs = synalinks.Input(data_model=Query)
outputs = await synalinks.Generator(data_model=Answer, language_model=lm)(inputs)
program = synalinks.Program(inputs=inputs, outputs=outputs)

2. Sequential API (Simple linear chains)

program = synalinks.Sequential([
    synalinks.Input(data_model=Query),
    synalinks.Generator(data_model=Answer, language_model=lm),
])

3. Subclassing (Advanced custom logic)

class MyProgram(synalinks.Program):
    def __init__(self, language_model):
        super().__init__()
        self.gen = synalinks.Generator(data_model=Answer, language_model=language_model)

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

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

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

4. Mixed (Functional + Subclassing)

class MyProgram(synalinks.Program):
    def __init__(self, language_model):
        super().__init__()
        self.language_model = language_model

    async def build(self, inputs):
        outputs = await synalinks.Generator(data_model=Answer, language_model=self.language_model)(inputs)
        super().__init__(inputs=inputs, outputs=outputs)

JSON Operators (Circuit-like Logic)

Operator	Symbol	Behavior
Concatenate	+	Merge fields; raises Exception if either is None
Logical And	&	Merge fields; returns None if either is None
Logical Or	|	Merge fields; returns non-None value if one is None
Logical Xor	^	Returns None if both present; otherwise returns the non-None value
Contains	in	Check if one DataModel's fields are a subset of another

# Concatenation (strict)
combined = x1 + x2  # Exception if either is None

# Logical And (safe concatenation)
combined = inputs & branch_output  # None if branch not taken

# Logical Or (merge branches)
result = branch1_output | branch2_output  # Returns whichever is not None

# Logical Xor (computation bypass for guards)
guarded = warning ^ inputs  # Returns inputs only if warning is None

# Contains operator
print(Query in (Query + Answer))  # True
print(Query in Answer)            # False

Module Alternatives to Operators

You can also use explicit module classes instead of operators:

# Using And module (equivalent to &)
merged = await synalinks.And()([b0, b1, b2])

# Using Or module (equivalent to |)
result = await synalinks.Or()([b0, b1, b2])

Control Flow

Parallel Branches (auto-detected)

x1 = await synalinks.Generator(data_model=Answer, language_model=lm)(inputs)
x2 = await synalinks.Generator(data_model=Answer, language_model=lm)(inputs)
# Both run in parallel via asyncio

Decision Making

decision = await synalinks.Decision(
    question="Evaluate query difficulty",
    labels=["easy", "difficult"],
    language_model=lm,
)(inputs)

Conditional Branching

(easy_answer, hard_answer) = await synalinks.Branch(
    question="Evaluate query difficulty",
    labels=["easy", "difficult"],
    branches=[
        synalinks.Generator(data_model=Answer, language_model=lm),
        synalinks.Generator(data_model=AnswerWithThinking, language_model=lm),
    ],
    language_model=lm,
    return_decision=False,   # If True, returns decision alongside branch outputs
    inject_decision=False,   # If True, injects decision into branch inputs
)(inputs)

# Merge with logical or - non-activated branches return None
final = easy_answer | hard_answer

Key Branch behaviors:

• Non-activated branches return None (not executed, not just empty)
• Each branch module gets optimized separately during training (specialized modules)
• Labels constrain LLM output - prevents hallucination by enforcing valid choices

Self-Consistency Pattern (Parallel Reasoning)

Use parallel branches with temperature > 0 to generate multiple answers, then merge:

async def build_self_consistency_program(lm):
    inputs = synalinks.Input(data_model=Query)

    # Generate multiple answers with randomness
    b0 = await synalinks.Generator(data_model=AnswerWithRationale, language_model=lm, temperature=1.0)(inputs)
    b1 = await synalinks.Generator(data_model=AnswerWithRationale, language_model=lm, temperature=1.0)(inputs)
    b2 = await synalinks.Generator(data_model=AnswerWithRationale, language_model=lm, temperature=1.0)(inputs)

    # Merge all answers
    merged = b0 & b1 & b2

    # Final answer that critically analyzes all attempts
    outputs = await synalinks.Generator(
        data_model=AnswerWithRationale,
        language_model=lm,
        instructions="Critically analyze the given answers to produce the final answer.",
    )(inputs & merged)

    return synalinks.Program(inputs=inputs, outputs=outputs)

Input/Output Guards (XOR Pattern)

Use XOR (^) to bypass computation based on conditions:

Input Guard - Block processing if input is invalid:

class InputGuard(synalinks.Module):
    """Block invalid inputs."""
    async def call(self, inputs, training=False):
        if self._is_blocked(inputs):
            return synalinks.ChatMessage(role="assistant", content="Cannot process this request")
        return None  # Allow through

async def build_guarded_program(lm):
    inputs = synalinks.Input(data_model=synalinks.ChatMessages)

    # Check input
    warning = await InputGuard()(inputs)

    # XOR: if warning exists, inputs becomes None (bypassing generator)
    guarded_inputs = warning ^ inputs

    # Generator only runs if guarded_inputs is not None
    answer = await synalinks.Generator(language_model=lm)(guarded_inputs)

    # OR: return warning if it exists, otherwise return answer
    outputs = warning | answer

    return synalinks.Program(inputs=inputs, outputs=outputs)

Output Guard - Replace invalid outputs:

async def build_output_guarded_program(lm):
    inputs = synalinks.Input(data_model=synalinks.ChatMessages)

    answer = await synalinks.Generator(language_model=lm)(inputs)

    # Check output
    warning = await OutputGuard()(answer)

    # XOR + OR: if warning exists, replace answer with warning
    outputs = (answer ^ warning) | warning

    return synalinks.Program(inputs=inputs, outputs=outputs)

Training Programs

# Compile with reward and optimizer
program.compile(
    reward=synalinks.rewards.ExactMatch(in_mask=["answer"]),
    optimizer=synalinks.optimizers.RandomFewShot(),
    metrics=[synalinks.metrics.F1Score(in_mask=["answer"])],
)

# Train
history = await program.fit(
    x=x_train,
    y=y_train,
    validation_split=0.2,
    epochs=10,
    batch_size=32,
    callbacks=[synalinks.callbacks.ProgramCheckpoint(filepath="best.json", monitor="val_reward", mode="max")],
)

# Evaluate
metrics = await program.evaluate(x=x_test, y=y_test, batch_size=32)

# Batch prediction
predictions = await program.predict(x_test, batch_size=32)

Built-in Rewards

• synalinks.rewards.ExactMatch(in_mask=["field"]) - Exact string match
• synalinks.rewards.CosineSimilarity(embedding_model=em, in_mask=["field"]) - Semantic similarity
• synalinks.rewards.LMAsJudge(language_model=lm) - LLM-based evaluation

Custom Rewards

@synalinks.saving.register_synalinks_serializable()
async def my_reward(y_true, y_pred):
    # Return float between 0.0 and 1.0
    return 1.0 if y_true.get("answer") == y_pred.get("answer") else 0.0

program.compile(reward=synalinks.rewards.MeanRewardWrapper(fn=my_reward))

Built-in Optimizers

RandomFewShot (Baseline)

synalinks.optimizers.RandomFewShot(
    few_shot_learning=False,    # Enable few-shot examples
    nb_min_examples=1,          # Min examples per prompt
    nb_max_examples=3,          # Max examples per prompt
)

Use as baseline. Fast, no extra LLM calls. Only manipulates examples, not prompts.

OMEGA (Advanced Evolutionary Optimizer)

OptiMizEr as Genetic Algorithm - Uses LLM-based mutation/crossover with Dominated Novelty Search for quality-diversity optimization.

synalinks.optimizers.OMEGA(
    # Required
    language_model=lm,              # For mutation/crossover reasoning
    embedding_model=em,             # For diversity metrics

    # D

---

*Content truncated.*