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 dataoutput_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 fileinstructions
: Natural language instructions for the transformationschema_only
: When True, focuses on structural transformation from the example filerepresenter_kwargs
: Additional parameters passed to file representers
3. How It Works
The SyncpulseTransformationBuilder
follows these steps to generate transformation code:
- Input Analysis: Examines the structure and content of the input SDIF file
- Target Analysis: Analyzes the example output files to understand the desired format
- Prompt Construction: Creates a detailed prompt combining the input analysis, output examples, and user instructions
- Code Generation: Uses an AI agent to generate Python code that performs the transformation
- Code Testing: Executes the generated code against the input SDIF and compares the result with the target examples
- 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