Spaces:
Sleeping
Sleeping
File size: 7,056 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | """TraceAnalyser — batch learning from pre-recorded execution traces."""
from __future__ import annotations
from collections.abc import Sequence
from pathlib import Path
from types import MappingProxyType
from typing import Any
from pipeline import Pipeline
from pipeline.errors import CancellationToken
from pipeline.protocol import SampleResult, StepProtocol
from ..core.context import ACEStepContext, SkillbookView
from ..core.insight_source import TRACE_IDENTITY_METADATA_KEY, infer_trace_identity
from ..protocols import (
DeduplicationManagerLike,
ReflectorLike,
SkillManagerLike,
)
from ..core.skillbook import Skillbook
from ..steps import learning_tail
from .base import ACERunner
class TraceAnalyser(ACERunner):
"""Analyse pre-recorded traces to build a skillbook.
Runs the learning tail only — Reflect and Update — with optional
deduplication and checkpoint steps. No AgentStep, no EvaluateStep.
The agentic SkillManager mutates the skillbook directly through its
tools, so no ApplyStep is needed.
Accepts raw trace objects of any type. They are placed directly on
``ctx.trace`` for the Reflector to interpret.
Use when you have execution logs from an external system (browser-use
``AgentHistoryList``, LangChain intermediate steps, Claude Code
transcripts) and want to build or refine a skillbook from historical
data.
"""
@classmethod
def build_steps(
cls,
*,
reflector: ReflectorLike,
skill_manager: SkillManagerLike,
skillbook: Skillbook | None = None,
dedup_manager: DeduplicationManagerLike | None = None,
dedup_interval: int = 10,
checkpoint_dir: str | Path | None = None,
checkpoint_interval: int = 10,
extra_steps: list[StepProtocol] | None = None,
) -> list[StepProtocol]:
"""Return the steps that ``from_roles()`` would compose.
Use this to inspect, modify, or extend the pipeline before
constructing it yourself::
steps = TraceAnalyser.build_steps(reflector=r, skill_manager=sm, ...)
steps.append(MyCustomStep())
pipe = Pipeline(steps)
runner = ACERunner(pipeline=pipe, skillbook=skillbook)
Args:
reflector: Reflector role for analysing traces.
skill_manager: SkillManager role for producing update operations.
skillbook: Starting skillbook. Creates an empty one if ``None``.
dedup_manager: Optional deduplication manager.
dedup_interval: Samples between deduplication runs.
checkpoint_dir: Directory for checkpoint files.
checkpoint_interval: Samples between checkpoint saves.
extra_steps: Additional steps appended after the learning
tail (e.g. ``OpikStep``).
"""
skillbook = skillbook or Skillbook()
steps = learning_tail(
reflector,
skill_manager,
skillbook,
dedup_manager=dedup_manager,
dedup_interval=dedup_interval,
checkpoint_dir=checkpoint_dir,
checkpoint_interval=checkpoint_interval,
)
if extra_steps:
steps.extend(extra_steps)
return steps
@classmethod
def from_roles(
cls,
*,
reflector: ReflectorLike,
skill_manager: SkillManagerLike,
skillbook: Skillbook | None = None,
dedup_manager: DeduplicationManagerLike | None = None,
dedup_interval: int = 10,
checkpoint_dir: str | Path | None = None,
checkpoint_interval: int = 10,
extra_steps: list[StepProtocol] | None = None,
) -> TraceAnalyser:
"""Construct from pre-built role instances.
Args:
reflector: Reflector role for analysing traces.
skill_manager: SkillManager role for producing update operations.
skillbook: Starting skillbook. Creates an empty one if ``None``.
dedup_manager: Optional deduplication manager. Appends a
``DeduplicateStep`` when provided.
dedup_interval: Samples between deduplication runs.
checkpoint_dir: Directory for checkpoint files. Appends a
``CheckpointStep`` when provided.
checkpoint_interval: Samples between checkpoint saves.
extra_steps: Additional steps appended after the learning
tail (e.g. ``OpikStep``).
"""
skillbook = skillbook or Skillbook()
steps = cls.build_steps(
reflector=reflector,
skill_manager=skill_manager,
skillbook=skillbook,
dedup_manager=dedup_manager,
dedup_interval=dedup_interval,
checkpoint_dir=checkpoint_dir,
checkpoint_interval=checkpoint_interval,
extra_steps=extra_steps,
)
return cls(pipeline=Pipeline(steps), skillbook=skillbook)
def run(
self,
traces: Sequence[Any],
epochs: int = 1,
*,
wait: bool = True,
cancel_token: CancellationToken | None = None,
) -> list[SampleResult]:
"""Analyse traces and evolve the skillbook.
Args:
traces: Sequence of raw trace objects (any type).
epochs: Number of passes over all traces.
wait: If ``True``, block until background learning completes.
cancel_token: Optional cancellation signal. Forwarded to
``Pipeline.run()`` — checked between steps and inside
LLM calls (via contextvar).
Returns:
List of ``SampleResult``, one per trace per epoch.
"""
return self._run(traces, epochs=epochs, wait=wait, cancel_token=cancel_token)
def _build_context( # type: ignore[override]
self,
raw_trace: Any,
*,
epoch: int,
total_epochs: int,
index: int,
total: int | None,
global_sample_index: int,
**_: Any,
) -> ACEStepContext:
"""Place a raw trace directly on the context.
No extraction, no conversion — the Reflector receives the trace
as-is and has full freedom to analyse it.
"""
return ACEStepContext(
skillbook=SkillbookView(self.skillbook),
trace=raw_trace,
metadata=MappingProxyType(
{
TRACE_IDENTITY_METADATA_KEY: infer_trace_identity(
trace=raw_trace,
default_source_system="trace",
).to_dict()
}
),
epoch=epoch,
total_epochs=total_epochs,
step_index=index,
total_steps=total,
global_sample_index=global_sample_index,
)
|