Spaces:
Sleeping
Sleeping
File size: 3,155 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 | """ReflectStep — analyses a trace to produce a ReflectorOutput."""
from __future__ import annotations
import logging
from ..core.context import ACEStepContext
from ..core.outputs import AgentOutput
from ..protocols import ReflectorLike
logger = logging.getLogger(__name__)
class ReflectStep:
"""Run the Reflector role against the trace and current skillbook.
Receives ``ctx.trace`` — a dict from EvaluateStep (standard ACE pipeline),
a raw object from TraceAnalyser, or any integration-produced trace. When
the trace is a dict with known keys, the step extracts them and calls the
Reflector's existing API. For raw/opaque traces, it passes them as
keyword arguments for the Reflector to handle.
Declares ``async_boundary = True`` — everything from this step onward
runs in a background thread pool when the pipeline has background
execution enabled.
Pure — produces a reflection object, no side effects.
"""
requires = frozenset({"trace", "skillbook"})
provides = frozenset({"reflections"})
async_boundary = True
max_workers = 3
def __init__(self, reflector: ReflectorLike) -> None:
self.reflector = reflector
@staticmethod
def _is_batch_container(trace: dict) -> bool:
for key in ("items", "tasks"):
if isinstance(trace.get(key), list):
return True
steps = trace.get("steps")
return (
isinstance(steps, list)
and bool(steps)
and all(
isinstance(step, dict)
and step.get("role") == "conversation"
and isinstance(step.get("content"), dict)
for step in steps
)
)
def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
trace = ctx.trace
if isinstance(trace, dict) and not self._is_batch_container(trace):
# Structured trace from EvaluateStep — extract known fields
agent_output = AgentOutput(
reasoning=trace.get("reasoning", ""),
final_answer=trace.get("answer", ""),
)
reflection = self.reflector.reflect(
question=trace.get("question", ""),
agent_output=agent_output,
skillbook=ctx.skillbook,
ground_truth=trace.get("ground_truth"),
feedback=trace.get("feedback"),
injected_skill_ids=ctx.injected_skill_ids,
mode=ctx.mode,
)
else:
# Raw trace from TraceAnalyser or integration — pass as-is
# The Reflector must handle the trace type via **kwargs
reflection = self.reflector.reflect(
question="",
agent_output=AgentOutput(reasoning="", final_answer=""),
skillbook=ctx.skillbook,
trace=trace,
injected_skill_ids=ctx.injected_skill_ids,
mode=ctx.mode,
)
return ctx.replace(reflections=(reflection,))
|