2568 Modules Catalog 4a0b2423

# Synalinks Modules Catalog

## Generation Modules

### Generator

Core LLM generation module with structured output.

```python
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.

```python
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.

```python
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.

```python
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.

```python
(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).

```python
outputs = await synalinks.Identity()(inputs)
```

---

## Agent Modules

### FunctionCallingAgent

Autonomous agent with parallel tool calling capabilities.

```python
@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:**
```python
synalinks.Tool(
    function,                   # Async function
    name=None,                  # Override function name
    description=None,           # Override docstring
)
```

### ToolCalling

Single tool selection and execution.

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

---

## Knowledge Modules

### EmbedKnowledge

Generate embeddings for data models to enable vector similarity search.

```python
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.

```python
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.

```python
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.

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

### SequentialPlanSynthesis

Generate step-by-step execution plans.

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

---

## Operators (synalinks.ops)

### Concatenate

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

# Operator form
result = x1 + x2
```

### And

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

### Or

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

### Xor

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

### Not

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

---

## Creating Custom Modules

### Subclass Template

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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)

```python
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)

```python
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
```