Skip to main content

Syncpulse Transformation Builder

⚠️ EXPERIMENTAL: This is an experimental implementation not intended for production use. Despite the async API, there may be blocking I/O calls that could impact performance in production async environments.

The SyncpulseTransformationBuilder is an AI Agent that automatically generates transformation code to convert standardized SDIF data into target output files. It uses examples and optional instructions to create Python code that transforms your data to match the desired output structure and format.

1. Basic Usage

from satif_ai.transformation_builders.syncpulse import SyncpulseTransformationBuilder
from pathlib import Path

# Initialize MCP components

# Initialize the transformation builder
builder = SyncpulseTransformationBuilder(
mcp_server=mcp_server,
mcp_session=mcp_session,
llm_model="o4-mini"
)

# Generate transformation code
transformation_code = await builder.build(
sdif="input_data.sdif",
output_target_files={"example_output.csv": "output.csv"},
instructions="Convert the customer data to CSV with columns: id, name, email"
)

# Save the generated code to a file
with open("transformation.py", "w") as f:
f.write(transformation_code)

# The generated code can now be used with a CodeTransformer

2. The Build Method

transformation_code = await builder.build(
sdif: Path, # Required: Input SDIF file path
output_target_files: Dict[FilePath, str] | List[FilePath] | FilePath, # Required: Output example files
output_sdif: Optional[Path] = None, # Optional: Target SDIF output path
instructions: str = "", # Optional: Natural language instructions
schema_only: bool = False, # Optional: Focus on output schema without data
representer_kwargs: Optional[Dict[str, Any]] = None # Optional: Additional parameters for representers
)
  • sdif: Path to the input SDIF file containing standardized data
  • output_target_files: Target output files in one of these formats:
    • A dictionary mapping file paths to their display names
    • A list of file paths
    • A single file path
  • output_sdif: Optional path to a target SDIF file
  • instructions: Natural language instructions for the transformation
  • schema_only: When True, focuses on structural transformation from the example file
  • representer_kwargs: Additional parameters passed to file representers

3. How It Works

The SyncpulseTransformationBuilder follows these steps to generate transformation code:

  1. Input Analysis: Examines the structure and content of the input SDIF file
  2. Target Analysis: Analyzes the example output files to understand the desired format
  3. Prompt Construction: Creates a detailed prompt combining the input analysis, output examples, and user instructions
  4. Code Generation: Uses an AI agent to generate Python code that performs the transformation
  5. Code Testing: Executes the generated code against the input SDIF and compares the result with the target examples
  6. Refinement: If the output doesn't match the target, the code is refined and tested again

4. Output Example Files

The builder uses example output files to understand the desired transformation. These files should:

  • Have the same structure and format as your desired output
  • Contain representative data (though not necessarily the exact data you expect)
  • Include all fields and relationships that should appear in the final output

Example output files can be provided in several ways:

# Single file
output_target_files = "example_output.xlsx"

# Multiple files
output_target_files = ["customers.csv", "orders.json"]

# Files with custom display names
output_target_files = {
"path/to/example_customers.csv": "customers.csv",
"path/to/example_orders.json": "orders.json"
}

5. Using the Generated Code

The code generated by the builder can be executed using a CodeTransformer:

from satif_sdk.transformers import CodeTransformer
from pathlib import Path

# Load the generated transformation code
with open("transformation.py", "r") as f:
transformation_code = f.read()

# Create a transformer
transformer = CodeTransformer(function=transformation_code)

# Execute the transformation on real data
result = transformer.export(
sdif="real_data.sdif",
output_path="output_directory"
)

6. Implementing Your Own Builder

We recommend implementing your own TransformationBuilder by subclassing the base class:

from satif_core import AsyncTransformationBuilder
from pathlib import Path
from typing import Dict, Optional, Any

class CustomTransformationBuilder(AsyncTransformationBuilder):
"""Custom implementation of a transformation builder."""

def __init__(self, **kwargs):
# Initialize your builder
pass

async def build(
self,
sdif: Path,
output_target_files: Dict[Path, str],
instructions: str = "",
**kwargs
) -> str:
# Implement your code generation logic
# Return the generated transformation code as a string
pass