Skip to main content

synkro.generate()

Generate training traces from a policy document.
synkro.generate(
    policy: str | Policy,
    traces: int = 20,
    turns: int | str = "auto",
    dataset_type: DatasetType = DatasetType.CONVERSATION,
    generation_model: str = "gpt-5-mini",
    grading_model: str = "gpt-5.2",
    max_iterations: int = 3,
    skip_grading: bool = False,
    reporter: ProgressReporter | None = None,
    return_logic_map: bool = False,
    enable_hitl: bool = True,
    base_url: str | None = None,
    temperature: float = 0.7,
) -> Dataset | GenerationResult
ParameterTypeDefaultDescription
policystr | PolicyrequiredPolicy text or Policy object
tracesint20Number of traces to generate
turnsint | str"auto"Conversation turns. "auto" for policy complexity-driven
dataset_typeDatasetTypeCONVERSATIONType of dataset
generation_modelstr"gpt-5-mini"Model for generating responses
grading_modelstr"gpt-5.2"Model for grading (use stronger model)
max_iterationsint3Max refinement iterations per trace
skip_gradingboolFalseSkip grading phase for faster generation
reporterProgressReporterNoneProgress reporter (default: RichReporter)
return_logic_mapboolFalseReturn GenerationResult with Logic Map
enable_hitlboolTrueEnable Human-in-the-Loop editing
base_urlstrNoneAPI base URL for local LLM providers
temperaturefloat0.7Sampling temperature (0.0-2.0)

synkro.generate_scenarios()

Generate eval scenarios without synthetic responses.
synkro.generate_scenarios(
    policy: str | Policy,
    count: int = 100,
    generation_model: str = "gpt-4o-mini",
    temperature: float = 0.8,
    reporter: ProgressReporter | None = None,
    enable_hitl: bool = False,
    base_url: str | None = None,
) -> ScenariosResult
ParameterTypeDefaultDescription
policystr | PolicyrequiredPolicy text or Policy object
countint100Number of scenarios to generate
generation_modelstr"gpt-4o-mini"Model for generation
temperaturefloat0.8Sampling temperature
reporterProgressReporterNoneProgress reporter
enable_hitlboolFalseEnable Human-in-the-Loop editing
base_urlstrNoneAPI base URL

synkro.grade()

Grade a response against a scenario and policy.
synkro.grade(
    response: str,
    scenario: EvalScenario,
    policy: str | Policy,
    model: str = "gpt-4o",
    base_url: str | None = None,
) -> GradeResult
ParameterTypeDefaultDescription
responsestrrequiredModel response to grade
scenarioEvalScenariorequiredEval scenario with expected outcome
policystr | PolicyrequiredPolicy for grading context
modelstr"gpt-4o"LLM for grading
base_urlstrNoneAPI base URL

synkro.create_pipeline()

Create a customized generation pipeline.
synkro.create_pipeline(
    model: str = "gpt-4o-mini",
    grading_model: str = "gpt-4o",
    dataset_type: DatasetType = DatasetType.CONVERSATION,
    max_iterations: int = 3,
    skip_grading: bool = False,
    reporter: ProgressReporter | None = None,
    tools: list[ToolDefinition] | None = None,
    checkpoint_dir: str | None = None,
    enable_hitl: bool = True,
    base_url: str | None = None,
    temperature: float = 0.7,
    thinking: bool = False,
) -> Pipeline
ParameterTypeDefaultDescription
modelstr"gpt-4o-mini"Model for generating responses
grading_modelstr"gpt-4o"Model for grading
dataset_typeDatasetTypeCONVERSATIONType of dataset
max_iterationsint3Max refinement iterations
skip_gradingboolFalseSkip grading phase
reporterProgressReporterNoneProgress reporter
toolslist[ToolDefinition]NoneTool definitions for TOOL_CALL
checkpoint_dirstrNoneDirectory for checkpointing
enable_hitlboolTrueEnable Human-in-the-Loop
base_urlstrNoneAPI base URL
temperaturefloat0.7Sampling temperature
thinkingboolFalseEnable <think> tags

Dataset.save()

Save dataset to a JSONL file.
dataset.save(
    path: str | Path | None = None,
    format: str = "messages",
    pretty_print: bool = False,
) -> Dataset
ParameterTypeDefaultDescription
pathstr | PathNoneOutput path (auto-generated if None)
formatstr"messages"Output format
pretty_printboolFalseFormat with indentation
Format options: "messages", "qa", "langsmith", "langfuse", "tool_call", "chatml", "bert", "bert:<task>"

Dataset.filter()

Filter traces by criteria.
dataset.filter(
    passed: bool | None = None,
    category: str | None = None,
    min_length: int | None = None,
) -> Dataset
ParameterTypeDefaultDescription
passedboolNoneFilter by grade status
categorystrNoneFilter by category
min_lengthintNoneMinimum response length

Dataset.dedupe()

Remove duplicates.
dataset.dedupe(
    threshold: float = 0.85,
    method: str = "semantic",
    field: str = "user",
) -> Dataset
ParameterTypeDefaultDescription
thresholdfloat0.85Similarity threshold (0-1)
methodstr"semantic"Method: "exact" or "semantic"
fieldstr"user"Field: "user", "assistant", or "both"

Dataset.push_to_hub()

Push to HuggingFace Hub.
dataset.push_to_hub(
    repo_id: str,
    format: str = "messages",
    private: bool = False,
    split: str = "train",
    token: str | None = None,
) -> str
ParameterTypeDefaultDescription
repo_idstrrequiredHuggingFace repo ID
formatstr"messages"Output format
privateboolFalsePrivate repo
splitstr"train"Dataset split
tokenstrNoneHuggingFace token

ToolDefinition

Define a tool for TOOL_CALL datasets.
ToolDefinition(
    name: str,
    description: str,
    parameters: dict = {"type": "object", "properties": {}},
    examples: list[dict] = [],
    mock_responses: list[str] = [],
)
ParameterTypeDefaultDescription
namestrrequiredTool name
descriptionstrrequiredTool description
parametersdict{}JSON Schema for parameters
exampleslist[dict][]Example tool calls
mock_responseslist[str][]Mock responses for simulation

FileLoggingReporter

Log to file with optional console output.
FileLoggingReporter(
    delegate: ProgressReporter | None = None,
    log_dir: str = ".",
    log_filename: str | None = None,
)
ParameterTypeDefaultDescription
delegateProgressReporterRichReporter()Reporter for display
log_dirstr"."Log directory
log_filenamestrNoneCustom filename (auto if None)

CallbackReporter

Custom callbacks for progress events.
CallbackReporter(
    on_progress: Callable[[str, dict], None] | None = None,
    on_start: Callable[[int, str, str], None] | None = None,
    on_plan_complete: Callable[[Plan], None] | None = None,
    on_scenario_progress: Callable[[int, int], None] | None = None,
    on_scenarios_complete: Callable[[list[Scenario]], None] | None = None,
    on_response_progress: Callable[[int, int], None] | None = None,
    on_responses_complete: Callable[[list[Trace]], None] | None = None,
    on_grading_progress: Callable[[int, int], None] | None = None,
    on_grading_complete: Callable[[list[Trace], float], None] | None = None,
    on_complete: Callable[[int, float, float | None], None] | None = None,
)