Python

Generates output samples from each source sample using a custom Python synthesize function. Use this when the built-in synthesizers cannot express the transformation you need - for example, expanding one source sample into many, or computing derived fields. For simple placeholder grids, prefer the Template synthesizer.

Output

The list of samples returned by synthesize for each source sample.

Function Signature

Both def and async def are supported. The function receives a single source sample as a dict[str, Any] and must return a list[dict[str, Any]], one dict per output sample.

from typing import Any

def synthesize(source: dict[str, Any]) -> list[dict[str, Any]]:
    ...

Accessing generator config

synthesize can optionally accept a generator_config argument to receive the generator configuration as a plain Python dict. Config values of type dataset are not included.

def synthesize(
    source: dict[str, Any],
    generator_config: dict[str, Any],
) -> list[dict[str, Any]]:
    difficulty_level = generator_config["difficulty_level"]
    ...

Using models from config

To call a model inside the synthesizer, declare it in the generator’s config_spec with type model and access it via generator_config. The model exposes a predict method:

# config_spec:
#   - type: model
#     key: generate_model
#   - type: string
#     key: target_language

from latticeflow.core.dtypes import Message

async def synthesize(
    source: dict[str, Any],
    generator_config: dict[str, Any],
) -> list[dict[str, Any]]:
    generate_model = generator_config["generate_model"]
    target_language = generator_config["target_language"]
    response = await generate_model.predict([
        Message(role="system", content=f"Translate the following to {target_language}."),
        Message(role="user", content=source["text"]),
    ])
    translated_text = response.text
    ...

Examples

Example: Derived fields. Each source sample of a seed dataset holds a temperature in Celsius. The synthesize function converts it to Fahrenheit - a computation a Jinja template cannot express - and returns two output samples per source sample.

Dataset Generator

# ...
config_spec:
  - key: "source_dataset_key"
    type: "dataset"
    display_name: "Source Dataset"
    default_value: "city-temperatures"
    description: "Dataset containing seed temperatures to generate conversion questions from."
definition:
  data_source:
    type: "dataset_samples"
    dataset_key: "<< config.source_dataset_key >>"
  synthesizers:
    - type: "python"
      synthesize_snippet: !include "./synthesize.py"

Synthesize Function

from __future__ import annotations

from typing import Any


def synthesize(source: dict[str, Any]) -> list[dict[str, Any]]:
    city = source["city"]
    celsius = float(source["temperature_celsius"])
    fahrenheit = celsius * 9 / 5 + 32

    return [
        {
            "question": f"It is {celsius:g} °C in {city}. What is that in Fahrenheit?",
            "answer": f"{fahrenheit:g} °F",
        },
        {
            "question": f"It is {fahrenheit:g} °F in {city}. What is that in Celsius?",
            "answer": f"{celsius:g} °C",
        },
    ]

Seed Dataset

city,temperature_celsius
Berlin,21
Paris,17
Tokyo,26

The 3 seed samples yield 6 output samples:

Generated Dataset

| question                                           | answer  |
| :------------------------------------------------- | :------ |
| It is 21 °C in Berlin. What is that in Fahrenheit? | 69.8 °F |
| It is 69.8 °F in Berlin. What is that in Celsius?  | 21 °C   |
| It is 17 °C in Paris. What is that in Fahrenheit?  | 62.6 °F |
| It is 62.6 °F in Paris. What is that in Celsius?   | 17 °C   |
| It is 26 °C in Tokyo. What is that in Fahrenheit?  | 78.8 °F |
| It is 78.8 °F in Tokyo. What is that in Celsius?   | 26 °C   |

Configuration

Properties

---

type Literal “python” required

The type of the synthesizer.

---

synthesize_snippet string, TemplateValue required

The Python snippet defining how output samples are generated from a single source sample. It must define a synthesize function, with the following API:

def synthesize(source: dict[str, Any]) -> list[dict[str, Any]]:

where:

• source is a dictionary representing a single source sample.
• The return value is a list of dictionaries, each representing an output sample.

Both def and async def are supported.