tools-only Synalinks API Reference
Core Classes
synalinks.DataModel
Base class for defining structured I/O schemas. Inherits from Pydantic BaseModel.
class Query(synalinks.DataModel):
query: str = synalinks.Field(description="The user query")
context: str = synalinks.Field(default="", description="Optional context")
# Methods
model.get_schema() # Returns JSON schema dict
model.prettify_json() # Returns formatted JSON string
model.prettify_schema() # Returns formatted schema string
model.get("field_name") # Get field value
synalinks.Field
Field descriptor with metadata for LLM prompts.
synalinks.Field(
description="Field description for LLM", # Required for LLM understanding
default=None, # Default value
default_factory=list, # Factory for mutable defaults
)
synalinks.JsonDataModel
Runtime data model with actual values. Created when DataModel is instantiated.
synalinks.SymbolicDataModel
Schema-only placeholder used during graph construction. No actual data.
# Metaclass operations create SymbolicDataModel
combined_schema = Query + Answer # SymbolicDataModel
isinstance(combined_schema, synalinks.SymbolicDataModel) # True
Module Classes
synalinks.Module
Base class for all modules. Encapsulates state and computation.
class MyModule(synalinks.Module):
def __init__(self, name=None, description=None, trainable=True):
super().__init__(name=name, description=description, trainable=trainable)
async def call(self, inputs, training=False):
# Core computation
return outputs
async def compute_output_spec(self, inputs, training=False):
# Define output schema (optional, auto-inferred if not implemented)
return symbolic_output
def get_config(self):
return {"name": self.name, ...}
@classmethod
def from_config(cls, config):
return cls(**config)
Properties:
module.trainable- Whether variables are updated during trainingmodule.trainable_variables- List of trainable variablesmodule.variables- All variablesmodule.name- Module namemodule.description- Module description
synalinks.Input
Entry point for programs. Defines expected input schema.
inputs = synalinks.Input(data_model=Query)
synalinks.Generator
Core module for LLM generation with structured output.
outputs = await synalinks.Generator(
data_model=Answer, # Or schema=Answer.get_schema()
language_model=lm,
prompt_template=None, # Custom prompt template
instructions=["Be concise"], # List of instructions
examples=None, # Few-shot examples
return_inputs=False, # Include inputs in output
use_inputs_schema=False, # Include input schema in prompt
use_outputs_schema=False, # Include output schema in prompt
streaming=False, # Enable streaming output
)(inputs)
synalinks.ChainOfThought
Generator with automatic thinking step.
outputs = await synalinks.ChainOfThought(
data_model=Answer, # Output schema
language_model=lm,
return_inputs=True,
)(inputs)
# Output includes "thinking" field automatically
synalinks.Decision
Single-label classification for routing.
decision = await synalinks.Decision(
question="What type of query is this?",
labels=["factual", "opinion", "creative"],
language_model=lm,
)(inputs)
# Returns data model with "label" field
synalinks.Branch
Conditional routing based on decision.
outputs = await synalinks.Branch(
question="Difficulty level?",
labels=["easy", "hard"],
branches=[module_for_easy, module_for_hard],
language_model=lm,
return_decision=True, # Include decision in output
)(inputs)
# Returns tuple of outputs; non-selected branches return None
synalinks.SelfCritique
Self-evaluation with reward score.
critique = await synalinks.SelfCritique(
language_model=lm,
return_reward=True, # Include reward float
return_inputs=True,
)(previous_output)
# Output includes "critique" and "reward" fields
synalinks.Identity
Pass-through module (no-op).
outputs = await synalinks.Identity()(inputs)
Program Classes
synalinks.Program
Main container for module DAGs.
program = synalinks.Program(
inputs=inputs,
outputs=outputs,
name="my_program",
description="Program description",
trainable=True,
)
# Execution
result = await program(input_data) # Single inference
results = await program.predict(batch, batch_size=32) # Batch inference
# Training
program.compile(reward=..., optimizer=..., metrics=[...])
history = await program.fit(x=..., y=..., epochs=10, validation_split=0.2)
metrics = await program.evaluate(x=..., y=..., batch_size=32)
# Saving
program.save("path.json")
program = synalinks.Program.load("path.json")
program.save_variables("vars.json")
program.load_variables("vars.json")
# Building
await program.build(InputDataModel) # Explicit build
# Inspection
program.modules # List of modules
program.summary() # Print summary
synalinks.Sequential
Linear chain of modules.
program = synalinks.Sequential(
[
synalinks.Input(data_model=Query),
synalinks.Generator(data_model=Thinking, language_model=lm),
synalinks.Generator(data_model=Answer, language_model=lm),
],
name="sequential_chain",
)
Model Classes
synalinks.LanguageModel
LLM wrapper supporting multiple providers via LiteLLM.
lm = synalinks.LanguageModel(
model="ollama/mistral", # Provider/model format
# model="openai/gpt-4",
# model="anthropic/claude-3-opus",
# model="groq/llama-3-70b",
temperature=0.7,
max_tokens=1000,
)
Supported providers: ollama, openai, anthropic, mistral, groq, cohere, etc.
synalinks.EmbeddingModel
Embedding model wrapper.
em = synalinks.EmbeddingModel(
model="ollama/mxbai-embed-large",
# model="openai/text-embedding-3-small",
)
Operations (synalinks.ops)
Concatenation
result = await synalinks.ops.concat(x1, x2, name="combined")
# Or use operator: result = x1 + x2
Logical Operations
result = await synalinks.ops.logical_and(x1, x2) # Or: x1 & x2
result = await synalinks.ops.logical_or(x1, x2) # Or: x1 | x2
result = await synalinks.ops.logical_xor(x1, x2) # Or: x1 ^ x2
result = await synalinks.ops.logical_not(x) # Or: ~x
Rewards (synalinks.rewards)
Built-in Rewards
synalinks.rewards.ExactMatch(in_mask=["answer"])
synalinks.rewards.CosineSimilarity(embedding_model=em, in_mask=["field"])
synalinks.rewards.LMAsJudge(language_model=lm)
synalinks.rewards.ProgramAsJudge(program=judge_program)
Custom Rewards
@synalinks.saving.register_synalinks_serializable()
async def custom_reward(y_true, y_pred):
return float(y_true.get("x") == y_pred.get("x"))
program.compile(reward=synalinks.rewards.MeanRewardWrapper(fn=custom_reward))
Optimizers (synalinks.optimizers)
synalinks.optimizers.RandomFewShot()
synalinks.optimizers.OMEGA(language_model=lm, embedding_model=em)
Callbacks (synalinks.callbacks)
synalinks.callbacks.ProgramCheckpoint(
filepath="best.json",
monitor="val_reward",
mode="max",
save_best_only=True,
)
Utilities (synalinks.utils)
synalinks.utils.plot_program(program, to_folder=".", show_schemas=True, show_trainable=True)
synalinks.utils.plot_history(history, to_folder=".")
synalinks.utils.plot_metrics_with_mean_and_std(metrics_list, to_folder=".")
synalinks.utils.plot_metrics_comparison_with_mean_and_std(metrics_dict, to_folder=".")
@synalinks.utils.register_synalinks_serializable()
def my_function(): ...
Configuration
synalinks.enable_logging() # Debug logging
synalinks.enable_observability() # Tracing (Arize Phoenix compatible)
synalinks.set_seed(42) # Reproducibility
synalinks.clear_session() # Clear global state
Datasets (synalinks.datasets)
(x_train, y_train), (x_test, y_test) = synalinks.datasets.gsm8k.load_data()
input_model = synalinks.datasets.gsm8k.get_input_data_model()
output_model = synalinks.datasets.gsm8k.get_output_data_model()