Skip to content

synthesizer

synthesizer

Synthesize personal benchmarks from interaction traces.

Classes

PersonalBenchmarkSample dataclass 

PersonalBenchmarkSample(
    trace_id: str,
    query: str,
    reference_answer: str,
    agent: str = "",
    category: str = "chat",
    feedback_score: float = 0.0,
    metadata: Dict[str, Any] = dict(),
)

A single sample in a personal benchmark.

PersonalBenchmark dataclass 

PersonalBenchmark(
    workflow_id: str,
    samples: List[PersonalBenchmarkSample] = list(),
    created_at: float = 0.0,
)

A synthesized benchmark from user interaction traces.

PersonalBenchmarkSynthesizer 

PersonalBenchmarkSynthesizer(trace_store: TraceStore)

Mines interaction traces into a reusable personal benchmark.

Source code in src/diapason/learning/optimize/personal/synthesizer.py 
56
57
def __init__(self, trace_store: TraceStore) -> None:
    self._store = trace_store
Methods:
synthesize 
synthesize(
    workflow_id: str = "default",
    min_feedback: float = 0.7,
    max_samples: int = 100,
) -> PersonalBenchmark

Build a personal benchmark from high-quality traces.

  1. Query traces that have feedback >= min_feedback.
  2. Group by query class (agent + first 50 chars of query).
  3. For each class, pick the trace with the highest feedback as reference.
  4. Return a :class:PersonalBenchmark capped at max_samples.
Source code in src/diapason/learning/optimize/personal/synthesizer.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def synthesize(
    self,
    workflow_id: str = "default",
    min_feedback: float = 0.7,
    max_samples: int = 100,
) -> PersonalBenchmark:
    """Build a personal benchmark from high-quality traces.

    1. Query traces that have feedback >= *min_feedback*.
    2. Group by query class (agent + first 50 chars of query).
    3. For each class, pick the trace with the highest feedback as reference.
    4. Return a :class:`PersonalBenchmark` capped at *max_samples*.
    """
    # Fetch a large pool of traces (limit high enough to cover most stores)
    all_traces = self._store.list_traces(limit=10_000)

    # Filter to traces with sufficient feedback
    qualified = [
        t
        for t in all_traces
        if t.feedback is not None and t.feedback >= min_feedback
    ]

    # Group by query class
    groups: Dict[str, list] = defaultdict(list)
    for trace in qualified:
        key = _query_class_key(trace.agent, trace.query)
        groups[key].append(trace)

    # Pick best trace per class
    samples: List[PersonalBenchmarkSample] = []
    for _key, traces in groups.items():
        best = max(traces, key=lambda t: t.feedback or 0.0)
        samples.append(
            PersonalBenchmarkSample(
                trace_id=best.trace_id,
                query=best.query,
                reference_answer=best.result,
                agent=best.agent,
                category=_infer_category(best.agent),
                feedback_score=best.feedback or 0.0,
                metadata=best.metadata,
            ),
        )

    # Sort deterministically (highest feedback first) and cap
    samples.sort(key=lambda s: (-s.feedback_score, s.trace_id))
    samples = samples[:max_samples]

    return PersonalBenchmark(
        workflow_id=workflow_id,
        samples=samples,
        created_at=time.time(),
    )