API Reference

This page contains the auto-generated API reference for dspyer classes, decorators, and utilities.

Decorators

dspyer.self_correcting(schema=None, max_retries=2, refine_instructions=None, validator=None, dataset_log_path=None, redact_hook=None, validation_log_path=None)

Decorator to wrap a dspy.Module class or a dspy.Predict instance with schema-enforced automatic correction retry loops.

Source code in dspyer/decorator.py

def self_correcting(
    schema: Optional[Type[BaseModel]] = None,
    max_retries: int = 2,
    refine_instructions: Optional[str] = None,
    validator: Optional[Callable[[Any], bool]] = None,
    dataset_log_path: Optional[str] = None,
    redact_hook: Optional[Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = None,
    validation_log_path: Optional[str] = None,
) -> Callable[[Any], Any]:
    """
    Decorator to wrap a dspy.Module class or a dspy.Predict instance with schema-enforced
    automatic correction retry loops.
    """

    def decorator(target: Any) -> Any:
        if isinstance(target, type) and issubclass(target, dspy.Module):
            # 1. Class-level decoration
            orig_init = target.__init__  # type: ignore[misc]

            @functools.wraps(orig_init)
            def new_init(self, *args, **kwargs):
                orig_init(self, *args, **kwargs)  # type: ignore[misc]
                # Walk the module attributes and auto-wrap all Predict instances
                for name, attr in list(self.__dict__.items()):
                    if hasattr(attr, "signature") and hasattr(attr, "forward"):
                        setattr(
                            self,
                            name,
                            wrap_predictor(
                                attr,
                                schema,
                                max_retries,
                                refine_instructions,
                                validator,
                                dataset_log_path,
                                redact_hook,
                                validation_log_path,
                            ),
                        )

            target.__init__ = new_init  # type: ignore[misc]

            # Wrap forward as a final safety gate/corrector if schema is explicitly passed
            if hasattr(target, "forward") and schema is not None:
                orig_forward = target.forward

                @functools.wraps(orig_forward)
                def new_forward(self, *args, **kwargs):
                    sig_inspect = inspect.signature(orig_forward)
                    bound = sig_inspect.bind(self, *args, **kwargs)
                    bound.apply_defaults()
                    current_inputs = {k: v for k, v in bound.arguments.items() if k != "self"}

                    prediction = orig_forward(self, *args, **kwargs)

                    attempt = 0
                    all_failed_fields: list[str] = []
                    while attempt < max_retries:
                        try:
                            parsed_model, raw_data = parse_and_validate(
                                prediction, schema, validator
                            )
                            if attempt > 0 and dataset_log_path is not None:
                                from dspyer.utils import log_self_correction_example

                                log_self_correction_example(
                                    dataset_log_path,
                                    current_inputs,
                                    parsed_model.model_dump(),
                                    redact_hook,
                                )
                            if validation_log_path is not None:
                                from dspyer.utils import log_validation_event

                                log_validation_event(
                                    validation_log_path,
                                    node_name=schema.__name__,
                                    success=True,
                                    retries_taken=attempt,
                                    failed_fields=all_failed_fields,
                                )
                            return prediction
                        except ValidationError as val_err:
                            attempt += 1
                            for err in val_err.errors():
                                loc_str = (
                                    ".".join(str(x) for x in err["loc"])
                                    if err.get("loc")
                                    else "unknown"
                                )
                                all_failed_fields.append(loc_str)

                            # Format error message
                            feedback_lines = []
                            for err in val_err.errors():
                                loc = " -> ".join(str(x) for x in err["loc"])
                                msg = err["msg"]
                                inp = err.get("input", "")
                                feedback_lines.append(f"Field '{loc}': {msg} (Value got: {inp})")
                            feedback_str = "\n".join(feedback_lines)

                            try:
                                from dspyer.telemetry import get_current_span

                                span = get_current_span()
                                if span is not None and hasattr(span, "record_validation_error"):
                                    span.record_validation_error(val_err)
                            except Exception:
                                pass

                            if attempt >= max_retries:
                                if validation_log_path is not None:
                                    from dspyer.utils import log_validation_event

                                    log_validation_event(
                                        validation_log_path,
                                        node_name=schema.__name__,
                                        success=False,
                                        retries_taken=attempt,
                                        failed_fields=all_failed_fields,
                                    )
                                raise val_err

                            # Resolve or instantiate module refiner dynamically
                            refiner_attr = f"_refiner_{schema.__name__}"
                            if not hasattr(self, refiner_attr):
                                orig_sig = make_signature_from_args_and_schema(
                                    list(current_inputs.keys()),
                                    schema,
                                    orig_forward.__doc__
                                    or self.__class__.__doc__
                                    or f"Correct output for {schema.__name__}",
                                )
                                fields = {}
                                for name, field in orig_sig.input_fields.items():
                                    fields[name] = (field.annotation, field)
                                fields["failed_output"] = (
                                    str,
                                    dspy.InputField(
                                        desc="The previous invalid output that failed schema validation."
                                    ),
                                )
                                fields["error_feedback"] = (
                                    str,
                                    dspy.InputField(
                                        desc="Natural language feedback outlining the schema validation errors to fix."
                                    ),
                                )
                                for name, field in orig_sig.output_fields.items():
                                    fields[name] = (field.annotation, field)

                                refine_sig = dspy.make_signature(
                                    signature=fields,
                                    instructions=refine_instructions
                                    or "Review inputs and correct the failed_output utilizing feedback.",
                                    signature_name=f"{schema.__name__}RefineSignature",
                                )
                                setattr(self, refiner_attr, dspy.Predict(refine_sig))

                            refiner = getattr(self, refiner_attr)
                            refiner_inputs = {
                                **current_inputs,
                                "failed_output": str(prediction),
                                "error_feedback": feedback_str,
                            }
                            prediction = refiner(**refiner_inputs)

                    return prediction

                target.forward = new_forward
            return target

        elif hasattr(target, "signature") and hasattr(target, "forward"):
            # 2. Predictor instance wrapping
            return wrap_predictor(
                target,
                schema,
                max_retries,
                refine_instructions,
                validator,
                dataset_log_path,
                redact_hook,
                validation_log_path,
            )

        elif inspect.isfunction(target):
            # 3. Typed function wrapping
            sig = inspect.signature(target)

            # Check return type schema
            return_anno = sig.return_annotation
            if (
                return_anno is inspect.Signature.empty
                or not isinstance(return_anno, type)
                or not issubclass(return_anno, BaseModel)
            ):
                raise TypeError(
                    f"Decorated function '{target.__name__}' must have a return type annotation that is a subclass of pydantic.BaseModel"
                )

            target_schema = return_anno

            # Determine instructions
            instructions = (
                target.__doc__ or f"Execute logic to produce {target_schema.__name__} outputs."
            )
            instructions = instructions.strip()

            # Build input & output fields for the signature
            fields = {}
            for param_name, param in sig.parameters.items():
                anno = param.annotation if param.annotation is not inspect.Parameter.empty else str
                desc = param_name.replace("_", " ").title()
                fields[param_name] = (anno, dspy.InputField(desc=desc))

            for name, field_info in target_schema.model_fields.items():
                desc = field_info.description or name.replace("_", " ").title()
                fields[name] = (field_info.annotation, dspy.OutputField(desc=desc))

            # Compile dynamic signature
            dyn_sig = dspy.make_signature(
                signature=fields,
                instructions=instructions,
                signature_name=f"{target_schema.__name__}Signature",
            )

            # Build and wrap predictor
            predictor = dspy.Predict(dyn_sig)
            wrapped_predictor = wrap_predictor(
                predictor,
                target_schema,
                max_retries,
                refine_instructions,
                validator,
                dataset_log_path,
                redact_hook,
                validation_log_path,
            )

            if inspect.iscoroutinefunction(target):

                @functools.wraps(target)
                async def async_wrapper(*args, **kwargs):
                    bound = sig.bind(*args, **kwargs)
                    bound.apply_defaults()
                    import asyncio

                    # Run the self-correcting predictor in a thread to avoid blocking the event loop
                    prediction = await asyncio.to_thread(wrapped_predictor, **bound.arguments)
                    # Parse and validate returned outputs back into target BaseModel
                    parsed, _ = parse_and_validate(prediction, target_schema, validator)
                    return parsed

                return async_wrapper

            @functools.wraps(target)
            def wrapper(*args, **kwargs):
                bound = sig.bind(*args, **kwargs)
                bound.apply_defaults()
                # Run the self-correcting predictor
                prediction = wrapped_predictor(**bound.arguments)
                # Parse and validate returned outputs back into target BaseModel
                parsed, _ = parse_and_validate(prediction, target_schema, validator)
                return parsed

            return wrapper

        else:
            raise TypeError(
                "@self_correcting decorator must wrap a dspy.Module class, a dspy.Predict instance, or a typed Python function"
            )

    return decorator

dspyer.dspyer_node(*args, input_model=None, output_model=None, is_llm=True, instructions=None)

Decorator to explicitly declare input/output schemas and prompts for a node, bypassing static AST parsing during LangGraph conversion.

Source code in dspyer/decorator.py

def dspyer_node(
    *args,
    input_model: Optional[Type[BaseModel]] = None,
    output_model: Optional[Type[BaseModel]] = None,
    is_llm: bool = True,
    instructions: Optional[str] = None,
) -> Callable[[Any], Any]:
    """
    Decorator to explicitly declare input/output schemas and prompts for a node,
    bypassing static AST parsing during LangGraph conversion.
    """
    # Check if called as @dspyer_node directly
    if len(args) == 1 and callable(args[0]):
        func = args[0]
        func._dspyer_input_model = None
        func._dspyer_output_model = None
        func._dspyer_is_llm = is_llm
        func._dspyer_instructions = None
        return func

    # Called with parameters: @dspyer_node(input_model=...)
    pos_input_model = args[0] if len(args) > 0 else None

    def decorator(func: Any) -> Any:
        func._dspyer_input_model = input_model or pos_input_model
        func._dspyer_output_model = output_model
        func._dspyer_is_llm = is_llm
        func._dspyer_instructions = instructions
        return func

    return decorator

---

Compiler & Program Execution

dspyer.AgentTranspiler

Public interface to transpile state machines into DSPy programs.

Source code in dspyer/compiler.py

class AgentTranspiler:
    """Public interface to transpile state machines into DSPy programs."""

    @overload
    @staticmethod
    def compile(
        graph: Graph,
        dataset_log_path: Optional[str] = None,
        redact_hook: Optional[Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = None,
        validation_log_path: Optional[str] = None,
        output_model: None = None,
        error_formatter: Optional[Callable[[Exception], str]] = None,
    ) -> TranspiledAgentProgram[BaseModel]: ...

    @overload
    @staticmethod
    def compile(
        graph: Graph,
        dataset_log_path: Optional[str] = None,
        redact_hook: Optional[Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = None,
        validation_log_path: Optional[str] = None,
        output_model: Type[T] = ...,
        error_formatter: Optional[Callable[[Exception], str]] = None,
    ) -> TranspiledAgentProgram[T]: ...

    @staticmethod
    def compile(
        graph: Graph,
        dataset_log_path: Optional[str] = None,
        redact_hook: Optional[Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = None,
        validation_log_path: Optional[str] = None,
        output_model: Optional[Type[Any]] = None,
        error_formatter: Optional[Callable[[Exception], str]] = None,
    ) -> TranspiledAgentProgram[Any]:
        return TranspiledAgentProgram(
            graph=graph,
            dataset_log_path=dataset_log_path,
            redact_hook=redact_hook,
            validation_log_path=validation_log_path,
            output_model=output_model,
            error_formatter=error_formatter,
        )

dspyer.compiler.TranspiledAgentProgram

Bases: Module, Generic[T]

A dynamically compiled, optimizable DSPy Module generated from an execution Graph. Supports branching, loops, and self-correction.

Source code in dspyer/compiler.py

class TranspiledAgentProgram(dspy.Module, Generic[T]):
    """
    A dynamically compiled, optimizable DSPy Module generated from
    an execution Graph. Supports branching, loops, and self-correction.
    """

    def __init__(
        self,
        graph: Graph,
        dataset_log_path: Optional[str] = None,
        redact_hook: Optional[Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = None,
        validation_log_path: Optional[str] = None,
        output_model: Optional[Type[T]] = None,
        error_formatter: Optional[Callable[[Exception], str]] = None,
    ):
        super().__init__()
        if graph.entry_point is None:
            raise ValueError("Graph entry point must be set before compilation.")

        self.entry_point = graph.entry_point
        self.edges = graph.edges
        self.conditional_edges = graph.conditional_edges
        self._nodes_map = graph.nodes
        self._last_refinement_steps_taken = 0
        self.dataset_log_path = dataset_log_path
        self.redact_hook = redact_hook
        self.validation_log_path = validation_log_path
        self.output_model = output_model
        self.error_formatter = error_formatter or format_validation_error

        # Validate that no node's input model has fields colliding with reserved execution control keywords
        reserved = {"max_retries", "max_steps", "on_loop_limit"}
        for node in graph.nodes.values():
            for field in node.input_model.model_fields.keys():
                if field in reserved:
                    raise ValueError(
                        f"Field name '{field}' in input model of node '{node.name}' is reserved. "
                        f"Please rename it to avoid collision with graph execution control parameters."
                    )

        # Detect unreachable nodes
        reachable = set()
        to_visit = [self.entry_point]
        while to_visit:
            curr = to_visit.pop(0)
            if curr in reachable:
                continue
            if curr in ("__end__", "END"):
                continue
            reachable.add(curr)
            # Normal static edges
            if curr in self.edges:
                to_visit.append(self.edges[curr])
            # Conditional router edges
            if curr in self.conditional_edges:
                router, path_map = self.conditional_edges[curr]
                for dest in path_map.values():
                    to_visit.append(dest)

        all_nodes = set(self._nodes_map.keys())
        unreachable = all_nodes - reachable
        if unreachable:
            logger.warning(
                f"Unreachable nodes detected in compilation graph: {sorted(list(unreachable))}. "
                f"These nodes are registered but cannot be reached from the entry point '{self.entry_point}'."
            )

        # Statically bind predictors and refiners for ALL registered nodes
        for node in graph.nodes.values():
            if getattr(node, "is_passthrough", False):
                continue
            sig = DynamicSignatureBuilder.build(node)
            predictor = dspy.Predict(sig)
            setattr(self, f"predictor_{node.name}", predictor)

            refine_sig = DynamicSignatureBuilder.build_refine(node)
            refiner = dspy.Predict(refine_sig)
            setattr(self, f"refiner_{node.name}", refiner)

    @property
    def refinement_steps_taken(self) -> int:
        val = _refinement_steps.get()
        if val == -1:
            return _last_refinement_steps.get()
        return val

    def _execute_node(
        self, node: StatefulNode, state: ImmutableState, max_retries: int
    ) -> ImmutableState:
        """
        Runs execution pipeline for a single node including pre-flight checks,
        predictions, output Pydantic validations, retries, and telemetry hooks.
        """
        if getattr(node, "is_passthrough", False) and node.callable is not None:
            with trace_span(f"node.{node.name}", state.to_dict()) as span:
                try:
                    patch = node.callable(state.to_dict())
                    if patch is None:
                        patch = {}
                    if not isinstance(patch, dict):
                        raise ValueError(
                            f"Deterministic node '{node.name}' must return a dictionary patch, got {type(patch)}."
                        )
                    state = state.apply_patch(patch)
                    for k, v in patch.items():
                        span.set_attribute(f"output.{k}", str(v))
                    return state
                except Exception as err:
                    span.set_status("ERROR", str(err))
                    raise GraphExecutionError(
                        node_name=node.name,
                        inputs=state.to_dict(),
                        raw_output=None,
                        error_feedback=str(err),
                        retries=0,
                        original_exception=err,
                    ) from err

        effective_max_retries = node.max_retries if node.max_retries is not None else max_retries
        predictor = getattr(self, f"predictor_{node.name}")
        refiner = getattr(self, f"refiner_{node.name}")

        # Extract and validate input payload from current state
        current_state_data = state.to_dict()
        node_inputs = {}
        for field_name, field_info in node.input_model.model_fields.items():
            if field_name in current_state_data:
                node_inputs[field_name] = current_state_data[field_name]
            else:
                # Check for default value
                if field_info.default is not PydanticUndefined:
                    node_inputs[field_name] = field_info.default
                elif field_info.default_factory is not None:
                    node_inputs[field_name] = field_info.default_factory()  # type: ignore[call-arg]
                else:
                    # Check if the field is nullable (Optional or None type)
                    annotation = field_info.annotation
                    is_nullable = False
                    if annotation is not None:
                        if get_origin(annotation) is Union:
                            is_nullable = type(None) in get_args(annotation)
                        elif annotation is type(None):
                            is_nullable = True

                    if is_nullable:
                        node_inputs[field_name] = None
                    else:
                        raise ValueError(
                            f"Node '{node.name}' expects input '{field_name}', "
                            f"but it was not found in the workflow state."
                        )

        node_validation_log_path = (
            getattr(node, "validation_log_path", None) or self.validation_log_path
        )

        # Trace execution of this node
        with trace_span(f"node.{node.name}", node_inputs) as span:
            # Execute initial attempt
            try:
                output_prediction = predictor(**node_inputs)
            except Exception as pred_err:
                span.set_status("ERROR", str(pred_err))
                raise GraphExecutionError(
                    node_name=node.name,
                    inputs=node_inputs,
                    raw_output=None,
                    error_feedback=str(pred_err),
                    retries=0,
                    original_exception=pred_err,
                ) from pred_err

            attempt = 0
            all_failed_fields: list[str] = []
            while True:
                # Extract outputs and attempt parsing for string fields that expect structured types
                raw_outputs = {}
                for output_field, field_info in node.output_model.model_fields.items():
                    # Check if the field is present in the output prediction
                    has_field = False
                    val = None

                    if isinstance(output_prediction, dict):
                        if output_field in output_prediction:
                            val = output_prediction[output_field]
                            has_field = True
                    elif hasattr(output_prediction, "__getitem__"):
                        try:
                            val = output_prediction[output_field]
                            has_field = True
                        except KeyError:
                            if hasattr(output_prediction, output_field):
                                val = getattr(output_prediction, output_field)
                                has_field = True
                    else:
                        if hasattr(output_prediction, output_field):
                            val = getattr(output_prediction, output_field)
                            has_field = True

                    if has_field:
                        # If field expects a collection or sub-model but received a string, attempt repair
                        if isinstance(val, str):
                            ann = field_info.annotation
                            origin = get_origin(ann) or ann
                            is_collection = isinstance(origin, type) and issubclass(
                                origin, (dict, list)
                            )
                            is_model = isinstance(ann, type) and issubclass(ann, BaseModel)
                            if is_collection or is_model:
                                try:
                                    val = repair_and_parse_json(val)
                                except Exception:
                                    pass

                        raw_outputs[output_field] = val

                try:
                    # Enforce schema validation at the transaction boundary
                    validated_patch = node.output_model.model_validate(raw_outputs)
                    # Commit state patch after dumping to JSON-compatible format
                    if getattr(node, "_is_autogenerated", False):
                        patch_data = validated_patch.model_dump(mode="json", exclude_unset=True)
                    else:
                        patch_data = validated_patch.model_dump(mode="json")
                    state = state.apply_patch(patch_data)

                    # Populate span output metadata
                    for k, v in validated_patch.model_dump().items():
                        span.set_attribute(f"output.{k}", str(v))

                    # If use_cot is True, collect and store rationale in telemetry/span metadata
                    if node.use_cot:
                        rationale_val = None
                        if isinstance(output_prediction, dict) and "rationale" in output_prediction:
                            rationale_val = output_prediction["rationale"]
                        elif hasattr(output_prediction, "__getitem__"):
                            try:
                                rationale_val = output_prediction["rationale"]  # type: ignore[index]
                            except KeyError:
                                rationale_val = getattr(output_prediction, "rationale", None)
                        else:
                            rationale_val = getattr(output_prediction, "rationale", None)

                        if rationale_val is not None:
                            span.set_attribute("output.rationale", str(rationale_val))
                            current_rationales = _rationales.get()
                            updated_rationales = dict(current_rationales)
                            updated_rationales[node.name] = str(rationale_val)
                            _rationales.set(updated_rationales)

                    if attempt > 0:
                        node_log_path = (
                            getattr(node, "dataset_log_path", None) or self.dataset_log_path
                        )
                        node_redact_hook = getattr(node, "redact_hook", None) or self.redact_hook
                        if node_log_path is not None:
                            from dspyer.utils import log_self_correction_example

                            log_self_correction_example(
                                node_log_path,
                                node_inputs,
                                validated_patch.model_dump(),
                                node_redact_hook,
                            )

                    if node_validation_log_path is not None:
                        from dspyer.utils import log_validation_event

                        log_validation_event(
                            node_validation_log_path,
                            node_name=node.name,
                            success=True,
                            retries_taken=attempt,
                            failed_fields=all_failed_fields,
                        )

                    break  # Validation succeeded, break retry loop
                except ValidationError as validation_err:
                    span.record_validation_error(validation_err)
                    for pydantic_error in validation_err.errors():
                        loc_str = (
                            ".".join(str(x) for x in pydantic_error["loc"])
                            if pydantic_error.get("loc")
                            else "unknown"
                        )
                        all_failed_fields.append(loc_str)
                    attempt += 1
                    _refinement_steps.set(_refinement_steps.get() + 1)

                    if attempt > effective_max_retries:
                        if node_validation_log_path is not None:
                            from dspyer.utils import log_validation_event

                            log_validation_event(
                                node_validation_log_path,
                                node_name=node.name,
                                success=False,
                                retries_taken=effective_max_retries,
                                failed_fields=all_failed_fields,
                            )
                        span.set_status("ERROR", str(validation_err))
                        feedback = self.error_formatter(validation_err)
                        raise GraphExecutionError(
                            node_name=node.name,
                            inputs=node_inputs,
                            raw_output=raw_outputs,
                            error_feedback=feedback,
                            retries=effective_max_retries,
                            original_exception=validation_err,
                        ) from validation_err

                    # Format feedback and prepare refine inputs
                    feedback = self.error_formatter(validation_err)
                    failed_output_str = json.dumps(raw_outputs)

                    # Annotate span with retry context
                    span.set_attribute(f"retry.{attempt}.error", feedback)
                    span.set_attribute(f"retry.{attempt}.failed_output", failed_output_str)

                    refine_inputs = {**node_inputs}
                    refine_inputs["failed_output"] = failed_output_str
                    refine_inputs["error_feedback"] = feedback

                    # Execute refiner
                    try:
                        output_prediction = refiner(**refine_inputs)
                    except Exception as refine_err:
                        span.set_status("ERROR", str(refine_err))
                        raise GraphExecutionError(
                            node_name=node.name,
                            inputs=node_inputs,
                            raw_output=raw_outputs,
                            error_feedback=str(refine_err),
                            retries=attempt,
                            original_exception=refine_err,
                        ) from refine_err
        return state

    def forward(
        self,
        *,
        _max_retries: int = 2,
        _max_steps: int = 15,
        _on_loop_limit: str = "raise",
        **initial_state_kwargs: Any,
    ) -> dspy.Prediction:
        token = _refinement_steps.set(0)
        rat_token = _rationales.set({})
        try:
            max_retries = initial_state_kwargs.pop("max_retries", _max_retries)
            max_steps = initial_state_kwargs.pop("max_steps", _max_steps)
            on_loop_limit = initial_state_kwargs.pop("on_loop_limit", _on_loop_limit)

            state = ImmutableState(initial_state_kwargs)

            current_node_name: Optional[str] = self.entry_point
            step_count = 0

            while current_node_name is not None and current_node_name not in ("__end__", "END"):
                step_count += 1
                if step_count > max_steps:
                    msg = f"Graph execution exceeded max_steps limit of {max_steps}."
                    logger.warning(msg)
                    if on_loop_limit == "raise":
                        raise RuntimeError(msg)
                    else:
                        break

                node = self._nodes_map[current_node_name]
                state = self._execute_node(node, state, max_retries)

                # Determine next step
                if current_node_name in self.edges:
                    current_node_name = self.edges[current_node_name]
                elif current_node_name in self.conditional_edges:
                    router, path_map = self.conditional_edges[current_node_name]
                    decision: Any = None

                    if callable(router) and not isinstance(router, StatefulNode):
                        # Python routing function
                        try:
                            decision = router(state.to_dict())
                        except Exception as router_err:
                            raise RuntimeError(
                                f"Python router function failed at node '{current_node_name}': {str(router_err)}"
                            ) from router_err
                    else:
                        # Semantic routing node (StatefulNode)
                        # We execute the router node as a child node step
                        state = self._execute_node(router, state, max_retries)

                        # Discover routing decision from router outputs
                        output_fields = list(router.output_model.model_fields.keys())
                        if len(output_fields) == 1:
                            decision_field = output_fields[0]
                        elif "next_step" in output_fields:
                            decision_field = "next_step"
                        elif "route" in output_fields:
                            decision_field = "route"
                        else:
                            raise ValueError(
                                f"Could not determine routing decision from router node '{router.name}' output. "
                                f"Please design the output model with a single field or a field named 'next_step' / 'route'."
                            )
                        decision = state.to_dict().get(decision_field)

                    # Retrieve destination
                    decision_str = str(decision) if decision is not None else ""
                    if decision_str in path_map:
                        current_node_name = path_map[decision_str]
                    else:
                        raise ValueError(
                            f"Router outcome '{decision_str}' at node '{current_node_name}' "
                            f"is not mapped to any destination in path_map: {list(path_map.keys())}"
                        )
                else:
                    current_node_name = None

            final_state = state.to_dict()
            metadata: Dict[str, Any] = {
                "refinement_steps_taken": self.refinement_steps_taken,
                "step_count": step_count,
            }
            rats = _rationales.get()
            if rats:
                metadata["rationales"] = rats
            final_state["_metadata"] = metadata
            self._last_refinement_steps_taken = self.refinement_steps_taken
            _last_refinement_steps.set(self.refinement_steps_taken)
            if self.output_model is not None:
                return self.output_model.model_validate(final_state)
            return dspy.Prediction(**final_state)
        finally:
            _refinement_steps.reset(token)
            _rationales.reset(rat_token)

    async def _aexecute_node(
        self,
        node: StatefulNode,
        state: ImmutableState,
        max_retries: int,
        event_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
    ) -> ImmutableState:
        """
        Runs async execution pipeline for a single node.
        """
        if getattr(node, "is_passthrough", False) and node.callable is not None:
            with trace_span(f"node.{node.name}", state.to_dict()) as span:
                try:
                    if inspect.iscoroutinefunction(node.callable):
                        patch = await node.callable(state.to_dict())
                    else:
                        patch = await asyncio.to_thread(node.callable, state.to_dict())
                    if patch is None:
                        patch = {}
                    if not isinstance(patch, dict):
                        raise ValueError(
                            f"Deterministic node '{node.name}' must return a dictionary patch, got {type(patch)}."
                        )
                    state = state.apply_patch(patch)
                    for k, v in patch.items():
                        span.set_attribute(f"output.{k}", str(v))
                    return state
                except Exception as err:
                    span.set_status("ERROR", str(err))
                    raise GraphExecutionError(
                        node_name=node.name,
                        inputs=state.to_dict(),
                        raw_output=None,
                        error_feedback=str(err),
                        retries=0,
                        original_exception=err,
                    ) from err

        effective_max_retries = node.max_retries if node.max_retries is not None else max_retries
        predictor = getattr(self, f"predictor_{node.name}")
        refiner = getattr(self, f"refiner_{node.name}")

        current_state_data = state.to_dict()
        node_inputs = {}
        for field_name, field_info in node.input_model.model_fields.items():
            if field_name in current_state_data:
                node_inputs[field_name] = current_state_data[field_name]
            else:
                if field_info.default is not PydanticUndefined:
                    node_inputs[field_name] = field_info.default
                elif field_info.default_factory is not None:
                    node_inputs[field_name] = field_info.default_factory()  # type: ignore[call-arg]
                else:
                    annotation = field_info.annotation
                    is_nullable = False
                    if annotation is not None:
                        if get_origin(annotation) is Union:
                            is_nullable = type(None) in get_args(annotation)
                        elif annotation is type(None):
                            is_nullable = True

                    if is_nullable:
                        node_inputs[field_name] = None
                    else:
                        raise ValueError(
                            f"Node '{node.name}' expects input '{field_name}', "
                            f"but it was not found in the workflow state."
                        )

        node_validation_log_path = (
            getattr(node, "validation_log_path", None) or self.validation_log_path
        )

        with trace_span(f"node.{node.name}", node_inputs) as span:
            try:
                output_prediction = await asyncio.to_thread(predictor, **node_inputs)
            except Exception as pred_err:
                span.set_status("ERROR", str(pred_err))
                raise GraphExecutionError(
                    node_name=node.name,
                    inputs=node_inputs,
                    raw_output=None,
                    error_feedback=str(pred_err),
                    retries=0,
                    original_exception=pred_err,
                ) from pred_err

            attempt = 0
            all_failed_fields: list[str] = []
            while True:
                raw_outputs = {}
                for output_field, field_info in node.output_model.model_fields.items():
                    has_field = False
                    val = None

                    if isinstance(output_prediction, dict):
                        if output_field in output_prediction:
                            val = output_prediction[output_field]
                            has_field = True
                    elif hasattr(output_prediction, "__getitem__"):
                        try:
                            val = output_prediction[output_field]
                            has_field = True
                        except KeyError:
                            if hasattr(output_prediction, output_field):
                                val = getattr(output_prediction, output_field)
                                has_field = True
                    else:
                        if hasattr(output_prediction, output_field):
                            val = getattr(output_prediction, output_field)
                            has_field = True

                    if has_field:
                        if isinstance(val, str):
                            ann = field_info.annotation
                            origin = get_origin(ann) or ann
                            is_collection = isinstance(origin, type) and issubclass(
                                origin, (dict, list)
                            )
                            is_model = isinstance(ann, type) and issubclass(ann, BaseModel)
                            if is_collection or is_model:
                                try:
                                    val = repair_and_parse_json(val)
                                except Exception:
                                    pass

                        raw_outputs[output_field] = val

                try:
                    validated_patch = node.output_model.model_validate(raw_outputs)
                    if getattr(node, "_is_autogenerated", False):
                        patch_data = validated_patch.model_dump(mode="json", exclude_unset=True)
                    else:
                        patch_data = validated_patch.model_dump(mode="json")
                    state = state.apply_patch(patch_data)

                    for k, v in validated_patch.model_dump().items():
                        span.set_attribute(f"output.{k}", str(v))

                    if node.use_cot:
                        rationale_val = None
                        if isinstance(output_prediction, dict) and "rationale" in output_prediction:
                            rationale_val = output_prediction["rationale"]
                        elif hasattr(output_prediction, "__getitem__"):
                            try:
                                rationale_val = output_prediction["rationale"]
                            except KeyError:
                                rationale_val = getattr(output_prediction, "rationale", None)
                        else:
                            rationale_val = getattr(output_prediction, "rationale", None)

                        if rationale_val is not None:
                            span.set_attribute("output.rationale", str(rationale_val))
                            current_rationales = _rationales.get()
                            updated_rationales = dict(current_rationales)
                            updated_rationales[node.name] = str(rationale_val)
                            _rationales.set(updated_rationales)

                    if attempt > 0:
                        node_log_path = (
                            getattr(node, "dataset_log_path", None) or self.dataset_log_path
                        )
                        node_redact_hook = getattr(node, "redact_hook", None) or self.redact_hook
                        if node_log_path is not None:
                            from dspyer.utils import log_self_correction_example_async

                            await log_self_correction_example_async(
                                node_log_path,
                                node_inputs,
                                validated_patch.model_dump(),
                                node_redact_hook,
                            )

                    if node_validation_log_path is not None:
                        from dspyer.utils import log_validation_event_async

                        await log_validation_event_async(
                            node_validation_log_path,
                            node_name=node.name,
                            success=True,
                            retries_taken=attempt,
                            failed_fields=all_failed_fields,
                        )

                    break
                except ValidationError as validation_err:
                    span.record_validation_error(validation_err)
                    for pydantic_error in validation_err.errors():
                        loc_str = (
                            ".".join(str(x) for x in pydantic_error["loc"])
                            if pydantic_error.get("loc")
                            else "unknown"
                        )
                        all_failed_fields.append(loc_str)
                    attempt += 1
                    _refinement_steps.set(_refinement_steps.get() + 1)

                    feedback = self.error_formatter(validation_err)
                    if event_callback:
                        event_callback(
                            {
                                "event": "validation_error",
                                "node": node.name,
                                "attempt": attempt,
                                "error": feedback,
                                "failed_fields": all_failed_fields,
                            }
                        )

                    if attempt > effective_max_retries:
                        if node_validation_log_path is not None:
                            from dspyer.utils import log_validation_event_async

                            await log_validation_event_async(
                                node_validation_log_path,
                                node_name=node.name,
                                success=False,
                                retries_taken=effective_max_retries,
                                failed_fields=all_failed_fields,
                            )
                        span.set_status("ERROR", str(validation_err))
                        raise GraphExecutionError(
                            node_name=node.name,
                            inputs=node_inputs,
                            raw_output=raw_outputs,
                            error_feedback=feedback,
                            retries=effective_max_retries,
                            original_exception=validation_err,
                        ) from validation_err

                    failed_output_str = json.dumps(raw_outputs)

                    span.set_attribute(f"retry.{attempt}.error", feedback)
                    span.set_attribute(f"retry.{attempt}.failed_output", failed_output_str)

                    refine_inputs = {**node_inputs}
                    refine_inputs["failed_output"] = failed_output_str
                    refine_inputs["error_feedback"] = feedback

                    try:
                        output_prediction = await asyncio.to_thread(refiner, **refine_inputs)
                    except Exception as refine_err:
                        span.set_status("ERROR", str(refine_err))
                        raise GraphExecutionError(
                            node_name=node.name,
                            inputs=node_inputs,
                            raw_output=raw_outputs,
                            error_feedback=str(refine_err),
                            retries=attempt,
                            original_exception=refine_err,
                        ) from refine_err
        return state

    async def aforward(
        self,
        *,
        _max_retries: int = 2,
        _max_steps: int = 15,
        _on_loop_limit: str = "raise",
        **initial_state_kwargs: Any,
    ) -> Union[T, dspy.Prediction]:
        """
        Asynchronously executes the state machine workflow.

        Runs predictions in a thread-safe thread pool to avoid blocking the event loop.
        Validates output fields against schemas and triggers self-correction retry loops
        on validation failures.

        Args:
            _max_retries (int): Default maximum correction attempts per node.
            _max_steps (int): Maximum total nodes traversed before breaking loops.
            _on_loop_limit (str): Behavior when step limit is hit ('raise' or 'return').
            **initial_state_kwargs: Initial state dictionary values.

        Returns:
            Union[T, dspy.Prediction]: Validated Pydantic model or dspy.Prediction result.
        """
        token = _refinement_steps.set(0)
        rat_token = _rationales.set({})
        try:
            max_retries = initial_state_kwargs.pop("max_retries", _max_retries)
            max_steps = initial_state_kwargs.pop("max_steps", _max_steps)
            on_loop_limit = initial_state_kwargs.pop("on_loop_limit", _on_loop_limit)

            state = ImmutableState(initial_state_kwargs)

            current_node_name: Optional[str] = self.entry_point
            step_count = 0

            while current_node_name is not None and current_node_name not in ("__end__", "END"):
                step_count += 1
                if step_count > max_steps:
                    msg = f"Graph execution exceeded max_steps limit of {max_steps}."
                    logger.warning(msg)
                    if on_loop_limit == "raise":
                        raise RuntimeError(msg)
                    else:
                        break

                node = self._nodes_map[current_node_name]
                state = await self._aexecute_node(node, state, max_retries)

                # Determine next step
                if current_node_name in self.edges:
                    current_node_name = self.edges[current_node_name]
                elif current_node_name in self.conditional_edges:
                    router, path_map = self.conditional_edges[current_node_name]
                    decision: Any = None

                    if callable(router) and not isinstance(router, StatefulNode):
                        try:
                            if inspect.iscoroutinefunction(router):
                                decision = await router(state.to_dict())
                            else:
                                decision = await asyncio.to_thread(router, state.to_dict())
                        except Exception as router_err:
                            raise RuntimeError(
                                f"Python router function failed at node '{current_node_name}': {str(router_err)}"
                            ) from router_err
                    else:
                        state = await self._aexecute_node(router, state, max_retries)
                        output_fields = list(router.output_model.model_fields.keys())
                        if len(output_fields) == 1:
                            decision_field = output_fields[0]
                        elif "next_step" in output_fields:
                            decision_field = "next_step"
                        elif "route" in output_fields:
                            decision_field = "route"
                        else:
                            raise ValueError(
                                f"Could not determine routing decision from router node '{router.name}' output. "
                                f"Please design the output model with a single field or a field named 'next_step' / 'route'."
                            )
                        decision = state.to_dict().get(decision_field)

                    decision_str = str(decision) if decision is not None else ""
                    if decision_str in path_map:
                        current_node_name = path_map[decision_str]
                    else:
                        raise ValueError(
                            f"Router outcome '{decision_str}' at node '{current_node_name}' "
                            f"is not mapped to any destination in path_map: {list(path_map.keys())}"
                        )
                else:
                    current_node_name = None

            final_state = state.to_dict()
            metadata: Dict[str, Any] = {
                "refinement_steps_taken": self.refinement_steps_taken,
                "step_count": step_count,
            }
            rats = _rationales.get()
            if rats:
                metadata["rationales"] = rats
            final_state["_metadata"] = metadata
            self._last_refinement_steps_taken = self.refinement_steps_taken
            _last_refinement_steps.set(self.refinement_steps_taken)

            if self.output_model is not None:
                return self.output_model.model_validate(final_state)
            return dspy.Prediction(**final_state)
        finally:
            _refinement_steps.reset(token)
            _rationales.reset(rat_token)

    async def astream(
        self,
        *,
        _max_retries: int = 2,
        _max_steps: int = 15,
        _on_loop_limit: str = "raise",
        **initial_state_kwargs: Any,
    ) -> AsyncGenerator[Dict[str, Any], None]:
        """
        Asynchronously executes the state machine workflow and streams step events.

        Yields events as they happen (e.g. node_start, node_end, validation_error)
        enabling frontends or interactive runtimes to track execution progress.

        Args:
            _max_retries (int): Default maximum correction attempts per node.
            _max_steps (int): Maximum total nodes traversed before breaking loops.
            _on_loop_limit (str): Behavior when step limit is hit ('raise' or 'return').
            **initial_state_kwargs: Initial state dictionary values.

        Yields:
            AsyncGenerator[Dict[str, Any], None]: Event payloads tracking steps.
        """
        token = _refinement_steps.set(0)
        rat_token = _rationales.set({})
        queue: asyncio.Queue[Optional[Dict[str, Any]]] = asyncio.Queue()

        def event_callback(ev: Optional[Dict[str, Any]]) -> None:
            queue.put_nowait(ev)

        async def run_workflow():
            try:
                max_retries = initial_state_kwargs.pop("max_retries", _max_retries)
                max_steps = initial_state_kwargs.pop("max_steps", _max_steps)
                on_loop_limit = initial_state_kwargs.pop("on_loop_limit", _on_loop_limit)

                state = ImmutableState(initial_state_kwargs)

                current_node_name: Optional[str] = self.entry_point
                step_count = 0

                while current_node_name is not None and current_node_name not in ("__end__", "END"):
                    step_count += 1
                    if step_count > max_steps:
                        msg = f"Graph execution exceeded max_steps limit of {max_steps}."
                        logger.warning(msg)
                        event_callback(
                            {
                                "event": "step_limit",
                                "max_steps": max_steps,
                                "message": msg,
                            }
                        )
                        if on_loop_limit == "raise":
                            raise RuntimeError(msg)
                        else:
                            break

                    node = self._nodes_map[current_node_name]

                    event_callback(
                        {
                            "event": "node_start",
                            "node": node.name,
                            "inputs": state.to_dict(),
                        }
                    )

                    state = await self._aexecute_node(node, state, max_retries, event_callback)

                    event_callback(
                        {
                            "event": "node_end",
                            "node": node.name,
                            "outputs": state.to_dict(),
                        }
                    )

                    # Determine next step
                    if current_node_name in self.edges:
                        current_node_name = self.edges[current_node_name]
                    elif current_node_name in self.conditional_edges:
                        router, path_map = self.conditional_edges[current_node_name]
                        decision: Any = None

                        if callable(router) and not isinstance(router, StatefulNode):
                            try:
                                if inspect.iscoroutinefunction(router):
                                    decision = await router(state.to_dict())
                                else:
                                    decision = await asyncio.to_thread(router, state.to_dict())
                            except Exception as router_err:
                                raise RuntimeError(
                                    f"Python router function failed at node '{current_node_name}': {str(router_err)}"
                                ) from router_err
                        else:
                            event_callback(
                                {
                                    "event": "node_start",
                                    "node": router.name,
                                    "inputs": state.to_dict(),
                                }
                            )
                            state = await self._aexecute_node(
                                router, state, max_retries, event_callback
                            )
                            event_callback(
                                {
                                    "event": "node_end",
                                    "node": router.name,
                                    "outputs": state.to_dict(),
                                }
                            )
                            output_fields = list(router.output_model.model_fields.keys())
                            if len(output_fields) == 1:
                                decision_field = output_fields[0]
                            elif "next_step" in output_fields:
                                decision_field = "next_step"
                            elif "route" in output_fields:
                                decision_field = "route"
                            else:
                                raise ValueError(
                                    f"Could not determine routing decision from router node '{router.name}' output. "
                                    f"Please design the output model with a single field or a field named 'next_step' / 'route'."
                                )
                            decision = state.to_dict().get(decision_field)

                        decision_str = str(decision) if decision is not None else ""
                        if decision_str in path_map:
                            current_node_name = path_map[decision_str]
                        else:
                            raise ValueError(
                                f"Router outcome '{decision_str}' at node '{current_node_name}' "
                                f"is not mapped to any destination in path_map: {list(path_map.keys())}"
                            )
                    else:
                        current_node_name = None

                final_state = state.to_dict()
                metadata: Dict[str, Any] = {
                    "refinement_steps_taken": self.refinement_steps_taken,
                    "step_count": step_count,
                }
                rats = _rationales.get()
                if rats:
                    metadata["rationales"] = rats
                final_state["_metadata"] = metadata
                self._last_refinement_steps_taken = self.refinement_steps_taken
                _last_refinement_steps.set(self.refinement_steps_taken)

                if self.output_model is not None:
                    pred = self.output_model.model_validate(final_state)
                else:
                    pred = dspy.Prediction(**final_state)

                event_callback(
                    {
                        "event": "finished",
                        "prediction": pred,
                    }
                )
            except Exception as exc:
                event_callback(
                    {
                        "event": "error",
                        "error": str(exc),
                    }
                )
                raise
            finally:
                event_callback(None)

        task = asyncio.create_task(run_workflow())

        try:
            while True:
                val = await queue.get()
                if val is None:
                    break
                yield val
        finally:
            _refinement_steps.reset(token)
            _rationales.reset(rat_token)
            if not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

    def save_config(self, path: str) -> None:
        """
        Serializes the compiled module's prompt instructions and refined instructions
        to a clean, human-readable JSON configuration file.
        """
        config = {}
        for node_name in self._nodes_map.keys():
            predictor = getattr(self, f"predictor_{node_name}", None)
            refiner = getattr(self, f"refiner_{node_name}", None)

            node_cfg = {}
            if predictor is not None and hasattr(predictor, "signature"):
                node_cfg["instructions"] = predictor.signature.instructions
            if refiner is not None and hasattr(refiner, "signature"):
                node_cfg["refine_instructions"] = refiner.signature.instructions

            config[node_name] = node_cfg

        with open(path, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=4)

    def load_config(self, path: str) -> None:
        """
        Loads optimized prompt instructions and refined instructions from a JSON
        configuration file and applies them to the module's predictors and refiners.
        """
        with open(path, "r", encoding="utf-8") as f:
            config = json.load(f)

        for node_name, node_cfg in config.items():
            predictor = getattr(self, f"predictor_{node_name}", None)
            refiner = getattr(self, f"refiner_{node_name}", None)

            if (
                predictor is not None
                and hasattr(predictor, "signature")
                and "instructions" in node_cfg
            ):
                sig = predictor.signature
                if hasattr(sig, "with_instructions"):
                    predictor.signature = sig.with_instructions(node_cfg["instructions"])
                else:
                    sig.instructions = node_cfg["instructions"]

            if (
                refiner is not None
                and hasattr(refiner, "signature")
                and "refine_instructions" in node_cfg
            ):
                sig = refiner.signature
                if hasattr(sig, "with_instructions"):
                    refiner.signature = sig.with_instructions(node_cfg["refine_instructions"])
                else:
                    sig.instructions = node_cfg["refine_instructions"]

    def save_prompts(self, path: str) -> None:
        """Alias for save_config to serialize optimized node instructions."""
        self.save_config(path)

    def load_prompts(self, path: str) -> None:
        """Alias for load_config to rehydrate optimized node instructions."""
        self.load_config(path)

Methods:

aforward(*, _max_retries=2, _max_steps=15, _on_loop_limit='raise', **initial_state_kwargs) async

Asynchronously executes the state machine workflow.

Runs predictions in a thread-safe thread pool to avoid blocking the event loop. Validates output fields against schemas and triggers self-correction retry loops on validation failures.

Parameters:

Name	Type	Description	Default
_max_retries	int	Default maximum correction attempts per node.	2
_max_steps	int	Maximum total nodes traversed before breaking loops.	15
_on_loop_limit	str	Behavior when step limit is hit ('raise' or 'return').	'raise'
**initial_state_kwargs	Any	Initial state dictionary values.	{}

Returns:

Type	Description
Union[T, Prediction]	Union[T, dspy.Prediction]: Validated Pydantic model or dspy.Prediction result.

Source code in dspyer/compiler.py

async def aforward(
    self,
    *,
    _max_retries: int = 2,
    _max_steps: int = 15,
    _on_loop_limit: str = "raise",
    **initial_state_kwargs: Any,
) -> Union[T, dspy.Prediction]:
    """
    Asynchronously executes the state machine workflow.

    Runs predictions in a thread-safe thread pool to avoid blocking the event loop.
    Validates output fields against schemas and triggers self-correction retry loops
    on validation failures.

    Args:
        _max_retries (int): Default maximum correction attempts per node.
        _max_steps (int): Maximum total nodes traversed before breaking loops.
        _on_loop_limit (str): Behavior when step limit is hit ('raise' or 'return').
        **initial_state_kwargs: Initial state dictionary values.

    Returns:
        Union[T, dspy.Prediction]: Validated Pydantic model or dspy.Prediction result.
    """
    token = _refinement_steps.set(0)
    rat_token = _rationales.set({})
    try:
        max_retries = initial_state_kwargs.pop("max_retries", _max_retries)
        max_steps = initial_state_kwargs.pop("max_steps", _max_steps)
        on_loop_limit = initial_state_kwargs.pop("on_loop_limit", _on_loop_limit)

        state = ImmutableState(initial_state_kwargs)

        current_node_name: Optional[str] = self.entry_point
        step_count = 0

        while current_node_name is not None and current_node_name not in ("__end__", "END"):
            step_count += 1
            if step_count > max_steps:
                msg = f"Graph execution exceeded max_steps limit of {max_steps}."
                logger.warning(msg)
                if on_loop_limit == "raise":
                    raise RuntimeError(msg)
                else:
                    break

            node = self._nodes_map[current_node_name]
            state = await self._aexecute_node(node, state, max_retries)

            # Determine next step
            if current_node_name in self.edges:
                current_node_name = self.edges[current_node_name]
            elif current_node_name in self.conditional_edges:
                router, path_map = self.conditional_edges[current_node_name]
                decision: Any = None

                if callable(router) and not isinstance(router, StatefulNode):
                    try:
                        if inspect.iscoroutinefunction(router):
                            decision = await router(state.to_dict())
                        else:
                            decision = await asyncio.to_thread(router, state.to_dict())
                    except Exception as router_err:
                        raise RuntimeError(
                            f"Python router function failed at node '{current_node_name}': {str(router_err)}"
                        ) from router_err
                else:
                    state = await self._aexecute_node(router, state, max_retries)
                    output_fields = list(router.output_model.model_fields.keys())
                    if len(output_fields) == 1:
                        decision_field = output_fields[0]
                    elif "next_step" in output_fields:
                        decision_field = "next_step"
                    elif "route" in output_fields:
                        decision_field = "route"
                    else:
                        raise ValueError(
                            f"Could not determine routing decision from router node '{router.name}' output. "
                            f"Please design the output model with a single field or a field named 'next_step' / 'route'."
                        )
                    decision = state.to_dict().get(decision_field)

                decision_str = str(decision) if decision is not None else ""
                if decision_str in path_map:
                    current_node_name = path_map[decision_str]
                else:
                    raise ValueError(
                        f"Router outcome '{decision_str}' at node '{current_node_name}' "
                        f"is not mapped to any destination in path_map: {list(path_map.keys())}"
                    )
            else:
                current_node_name = None

        final_state = state.to_dict()
        metadata: Dict[str, Any] = {
            "refinement_steps_taken": self.refinement_steps_taken,
            "step_count": step_count,
        }
        rats = _rationales.get()
        if rats:
            metadata["rationales"] = rats
        final_state["_metadata"] = metadata
        self._last_refinement_steps_taken = self.refinement_steps_taken
        _last_refinement_steps.set(self.refinement_steps_taken)

        if self.output_model is not None:
            return self.output_model.model_validate(final_state)
        return dspy.Prediction(**final_state)
    finally:
        _refinement_steps.reset(token)
        _rationales.reset(rat_token)

astream(*, _max_retries=2, _max_steps=15, _on_loop_limit='raise', **initial_state_kwargs) async

Asynchronously executes the state machine workflow and streams step events.

Yields events as they happen (e.g. node_start, node_end, validation_error) enabling frontends or interactive runtimes to track execution progress.

Parameters:

Name	Type	Description	Default
_max_retries	int	Default maximum correction attempts per node.	2
_max_steps	int	Maximum total nodes traversed before breaking loops.	15
_on_loop_limit	str	Behavior when step limit is hit ('raise' or 'return').	'raise'
**initial_state_kwargs	Any	Initial state dictionary values.	{}

Yields:

Type	Description
AsyncGenerator[Dict[str, Any], None]	AsyncGenerator[Dict[str, Any], None]: Event payloads tracking steps.

Source code in dspyer/compiler.py

async def astream(
    self,
    *,
    _max_retries: int = 2,
    _max_steps: int = 15,
    _on_loop_limit: str = "raise",
    **initial_state_kwargs: Any,
) -> AsyncGenerator[Dict[str, Any], None]:
    """
    Asynchronously executes the state machine workflow and streams step events.

    Yields events as they happen (e.g. node_start, node_end, validation_error)
    enabling frontends or interactive runtimes to track execution progress.

    Args:
        _max_retries (int): Default maximum correction attempts per node.
        _max_steps (int): Maximum total nodes traversed before breaking loops.
        _on_loop_limit (str): Behavior when step limit is hit ('raise' or 'return').
        **initial_state_kwargs: Initial state dictionary values.

    Yields:
        AsyncGenerator[Dict[str, Any], None]: Event payloads tracking steps.
    """
    token = _refinement_steps.set(0)
    rat_token = _rationales.set({})
    queue: asyncio.Queue[Optional[Dict[str, Any]]] = asyncio.Queue()

    def event_callback(ev: Optional[Dict[str, Any]]) -> None:
        queue.put_nowait(ev)

    async def run_workflow():
        try:
            max_retries = initial_state_kwargs.pop("max_retries", _max_retries)
            max_steps = initial_state_kwargs.pop("max_steps", _max_steps)
            on_loop_limit = initial_state_kwargs.pop("on_loop_limit", _on_loop_limit)

            state = ImmutableState(initial_state_kwargs)

            current_node_name: Optional[str] = self.entry_point
            step_count = 0

            while current_node_name is not None and current_node_name not in ("__end__", "END"):
                step_count += 1
                if step_count > max_steps:
                    msg = f"Graph execution exceeded max_steps limit of {max_steps}."
                    logger.warning(msg)
                    event_callback(
                        {
                            "event": "step_limit",
                            "max_steps": max_steps,
                            "message": msg,
                        }
                    )
                    if on_loop_limit == "raise":
                        raise RuntimeError(msg)
                    else:
                        break

                node = self._nodes_map[current_node_name]

                event_callback(
                    {
                        "event": "node_start",
                        "node": node.name,
                        "inputs": state.to_dict(),
                    }
                )

                state = await self._aexecute_node(node, state, max_retries, event_callback)

                event_callback(
                    {
                        "event": "node_end",
                        "node": node.name,
                        "outputs": state.to_dict(),
                    }
                )

                # Determine next step
                if current_node_name in self.edges:
                    current_node_name = self.edges[current_node_name]
                elif current_node_name in self.conditional_edges:
                    router, path_map = self.conditional_edges[current_node_name]
                    decision: Any = None

                    if callable(router) and not isinstance(router, StatefulNode):
                        try:
                            if inspect.iscoroutinefunction(router):
                                decision = await router(state.to_dict())
                            else:
                                decision = await asyncio.to_thread(router, state.to_dict())
                        except Exception as router_err:
                            raise RuntimeError(
                                f"Python router function failed at node '{current_node_name}': {str(router_err)}"
                            ) from router_err
                    else:
                        event_callback(
                            {
                                "event": "node_start",
                                "node": router.name,
                                "inputs": state.to_dict(),
                            }
                        )
                        state = await self._aexecute_node(
                            router, state, max_retries, event_callback
                        )
                        event_callback(
                            {
                                "event": "node_end",
                                "node": router.name,
                                "outputs": state.to_dict(),
                            }
                        )
                        output_fields = list(router.output_model.model_fields.keys())
                        if len(output_fields) == 1:
                            decision_field = output_fields[0]
                        elif "next_step" in output_fields:
                            decision_field = "next_step"
                        elif "route" in output_fields:
                            decision_field = "route"
                        else:
                            raise ValueError(
                                f"Could not determine routing decision from router node '{router.name}' output. "
                                f"Please design the output model with a single field or a field named 'next_step' / 'route'."
                            )
                        decision = state.to_dict().get(decision_field)

                    decision_str = str(decision) if decision is not None else ""
                    if decision_str in path_map:
                        current_node_name = path_map[decision_str]
                    else:
                        raise ValueError(
                            f"Router outcome '{decision_str}' at node '{current_node_name}' "
                            f"is not mapped to any destination in path_map: {list(path_map.keys())}"
                        )
                else:
                    current_node_name = None

            final_state = state.to_dict()
            metadata: Dict[str, Any] = {
                "refinement_steps_taken": self.refinement_steps_taken,
                "step_count": step_count,
            }
            rats = _rationales.get()
            if rats:
                metadata["rationales"] = rats
            final_state["_metadata"] = metadata
            self._last_refinement_steps_taken = self.refinement_steps_taken
            _last_refinement_steps.set(self.refinement_steps_taken)

            if self.output_model is not None:
                pred = self.output_model.model_validate(final_state)
            else:
                pred = dspy.Prediction(**final_state)

            event_callback(
                {
                    "event": "finished",
                    "prediction": pred,
                }
            )
        except Exception as exc:
            event_callback(
                {
                    "event": "error",
                    "error": str(exc),
                }
            )
            raise
        finally:
            event_callback(None)

    task = asyncio.create_task(run_workflow())

    try:
        while True:
            val = await queue.get()
            if val is None:
                break
            yield val
    finally:
        _refinement_steps.reset(token)
        _rationales.reset(rat_token)
        if not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

load_config(path)

Loads optimized prompt instructions and refined instructions from a JSON configuration file and applies them to the module's predictors and refiners.

Source code in dspyer/compiler.py

def load_config(self, path: str) -> None:
    """
    Loads optimized prompt instructions and refined instructions from a JSON
    configuration file and applies them to the module's predictors and refiners.
    """
    with open(path, "r", encoding="utf-8") as f:
        config = json.load(f)

    for node_name, node_cfg in config.items():
        predictor = getattr(self, f"predictor_{node_name}", None)
        refiner = getattr(self, f"refiner_{node_name}", None)

        if (
            predictor is not None
            and hasattr(predictor, "signature")
            and "instructions" in node_cfg
        ):
            sig = predictor.signature
            if hasattr(sig, "with_instructions"):
                predictor.signature = sig.with_instructions(node_cfg["instructions"])
            else:
                sig.instructions = node_cfg["instructions"]

        if (
            refiner is not None
            and hasattr(refiner, "signature")
            and "refine_instructions" in node_cfg
        ):
            sig = refiner.signature
            if hasattr(sig, "with_instructions"):
                refiner.signature = sig.with_instructions(node_cfg["refine_instructions"])
            else:
                sig.instructions = node_cfg["refine_instructions"]

load_prompts(path)

Alias for load_config to rehydrate optimized node instructions.

Source code in dspyer/compiler.py

def load_prompts(self, path: str) -> None:
    """Alias for load_config to rehydrate optimized node instructions."""
    self.load_config(path)

save_config(path)

Serializes the compiled module's prompt instructions and refined instructions to a clean, human-readable JSON configuration file.

Source code in dspyer/compiler.py

def save_config(self, path: str) -> None:
    """
    Serializes the compiled module's prompt instructions and refined instructions
    to a clean, human-readable JSON configuration file.
    """
    config = {}
    for node_name in self._nodes_map.keys():
        predictor = getattr(self, f"predictor_{node_name}", None)
        refiner = getattr(self, f"refiner_{node_name}", None)

        node_cfg = {}
        if predictor is not None and hasattr(predictor, "signature"):
            node_cfg["instructions"] = predictor.signature.instructions
        if refiner is not None and hasattr(refiner, "signature"):
            node_cfg["refine_instructions"] = refiner.signature.instructions

        config[node_name] = node_cfg

    with open(path, "w", encoding="utf-8") as f:
        json.dump(config, f, indent=4)

save_prompts(path)

Alias for save_config to serialize optimized node instructions.

Source code in dspyer/compiler.py

def save_prompts(self, path: str) -> None:
    """Alias for save_config to serialize optimized node instructions."""
    self.save_config(path)

---

Graph Specification

dspyer.graph.Graph

Represents the agent execution topology. Tracks nodes, linear edges, and routing connections.

Source code in dspyer/graph.py

class Graph:
    """
    Represents the agent execution topology.
    Tracks nodes, linear edges, and routing connections.
    """

    def __init__(self):
        self.nodes: Dict[str, StatefulNode] = {}
        self.edges: Dict[str, str] = {}  # Source node -> Destination node mapping
        self.conditional_edges: Dict[
            str, Tuple[Union[Callable[[Dict[str, Any]], str], StatefulNode], Dict[str, str]]
        ] = {}
        self.entry_point: Optional[str] = None

    def add_node(self, node: StatefulNode) -> None:
        """Registers a stateful node in the graph."""
        if node.name in self.nodes:
            raise ValueError(f"Node with name '{node.name}' already exists in the graph.")
        self.nodes[node.name] = node

    def set_entry_point(self, name: str) -> None:
        """Sets the starting node of the execution graph."""
        if name not in self.nodes:
            raise ValueError(f"Cannot set entry point to '{name}'; node is not registered.")
        self.entry_point = name

    def add_edge(self, source: str, destination: str) -> None:
        """Connects two nodes statically."""
        if source not in self.nodes:
            raise ValueError(f"Source node '{source}' must be registered before creating an edge.")
        if destination not in self.nodes and destination not in ("__end__", "END"):
            raise ValueError(
                f"Destination node '{destination}' must be registered before creating an edge."
            )
        self.edges[source] = destination

    def add_conditional_edges(
        self,
        source: str,
        router: Union[Callable[[Dict[str, Any]], str], StatefulNode],
        path_map: Dict[str, str],
    ) -> None:
        """
        Registers a conditional routing point from a source node.
        'router' can be a python callable or a StatefulNode.
        'path_map' maps router outcomes to destination nodes.
        """
        if source not in self.nodes:
            raise ValueError(
                f"Source node '{source}' must be registered before adding conditional edges."
            )

        # Verify destinations in path_map are registered
        for path_name, destination in path_map.items():
            if destination not in self.nodes and destination not in ("__end__", "END"):
                raise ValueError(
                    f"Destination node '{destination}' in path '{path_name}' must be registered."
                )

        # If the router is a StatefulNode, automatically register it if needed
        if isinstance(router, StatefulNode):
            if router.name not in self.nodes:
                self.add_node(router)

        self.conditional_edges[source] = (router, path_map)

Methods:

add_conditional_edges(source, router, path_map)

Registers a conditional routing point from a source node. 'router' can be a python callable or a StatefulNode. 'path_map' maps router outcomes to destination nodes.

Source code in dspyer/graph.py

def add_conditional_edges(
    self,
    source: str,
    router: Union[Callable[[Dict[str, Any]], str], StatefulNode],
    path_map: Dict[str, str],
) -> None:
    """
    Registers a conditional routing point from a source node.
    'router' can be a python callable or a StatefulNode.
    'path_map' maps router outcomes to destination nodes.
    """
    if source not in self.nodes:
        raise ValueError(
            f"Source node '{source}' must be registered before adding conditional edges."
        )

    # Verify destinations in path_map are registered
    for path_name, destination in path_map.items():
        if destination not in self.nodes and destination not in ("__end__", "END"):
            raise ValueError(
                f"Destination node '{destination}' in path '{path_name}' must be registered."
            )

    # If the router is a StatefulNode, automatically register it if needed
    if isinstance(router, StatefulNode):
        if router.name not in self.nodes:
            self.add_node(router)

    self.conditional_edges[source] = (router, path_map)

add_edge(source, destination)

Connects two nodes statically.

Source code in dspyer/graph.py

def add_edge(self, source: str, destination: str) -> None:
    """Connects two nodes statically."""
    if source not in self.nodes:
        raise ValueError(f"Source node '{source}' must be registered before creating an edge.")
    if destination not in self.nodes and destination not in ("__end__", "END"):
        raise ValueError(
            f"Destination node '{destination}' must be registered before creating an edge."
        )
    self.edges[source] = destination

add_node(node)

Registers a stateful node in the graph.

Source code in dspyer/graph.py

def add_node(self, node: StatefulNode) -> None:
    """Registers a stateful node in the graph."""
    if node.name in self.nodes:
        raise ValueError(f"Node with name '{node.name}' already exists in the graph.")
    self.nodes[node.name] = node

set_entry_point(name)

Sets the starting node of the execution graph.

Source code in dspyer/graph.py

def set_entry_point(self, name: str) -> None:
    """Sets the starting node of the execution graph."""
    if name not in self.nodes:
        raise ValueError(f"Cannot set entry point to '{name}'; node is not registered.")
    self.entry_point = name

dspyer.graph.StatefulNode

Defines a stateful execution node in the agent topology. Contains instructions, expected input Pydantic schemas, and expected output Pydantic schemas.

Source code in dspyer/graph.py

class StatefulNode:
    """
    Defines a stateful execution node in the agent topology.
    Contains instructions, expected input Pydantic schemas, and expected output Pydantic schemas.
    """

    def __init__(
        self,
        name: str,
        input_model: Type[BaseModel],
        output_model: Type[BaseModel],
        instructions: Optional[str] = None,
        max_retries: Optional[int] = None,
        refine_instructions: Optional[str] = None,
        use_cot: bool = False,
        dataset_log_path: Optional[str] = None,
        redact_hook: Optional[Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]] = None,
        validation_log_path: Optional[str] = None,
    ):
        self.name = name
        self.input_model = input_model
        self.output_model = output_model
        self.instructions = instructions
        self.max_retries = max_retries
        self.refine_instructions = refine_instructions
        self.use_cot = use_cot
        self.dataset_log_path = dataset_log_path
        self.redact_hook = redact_hook
        self.validation_log_path = validation_log_path
        self._is_autogenerated = False
        self.is_passthrough = False
        self.callable: Optional[Callable] = None

---

Conversions & Scaffolding

dspyer.from_langgraph(state_graph, node_mappings=None, node_configs=None)

Converts a LangGraph StateGraph builder or CompiledStateGraph into a dspyer Graph.

Parameters:

Name	Type	Description	Default
state_graph	Any	The LangGraph StateGraph or CompiledStateGraph instance.	required
node_mappings	Optional[Dict[str, StatefulNode]]	Optional dictionary mapping LangGraph node names to dspyer StatefulNodes. If not provided for a node, a default StatefulNode will be dynamically generated using the node's function docstring/signature.	None
node_configs	Optional[Dict[str, Dict[str, Any]]]	Optional dictionary mapping LangGraph node names to configuration override dictionaries. Can configure max_retries and refine_instructions.	None

Returns:

Name	Type	Description
Graph	Graph	A populated dspyer Graph topology ready for compilation.

Source code in dspyer/converter.py

def from_langgraph(
    state_graph: Any,
    node_mappings: Optional[Dict[str, StatefulNode]] = None,
    node_configs: Optional[Dict[str, Dict[str, Any]]] = None,
) -> Graph:
    """
    Converts a LangGraph StateGraph builder or CompiledStateGraph into a dspyer Graph.

    Args:
        state_graph: The LangGraph StateGraph or CompiledStateGraph instance.
        node_mappings: Optional dictionary mapping LangGraph node names to dspyer StatefulNodes.
                       If not provided for a node, a default StatefulNode will be dynamically
                       generated using the node's function docstring/signature.
        node_configs: Optional dictionary mapping LangGraph node names to configuration override dictionaries.
                      Can configure max_retries and refine_instructions.

    Returns:
        Graph: A populated dspyer Graph topology ready for compilation.
    """
    # 1. Extract LangGraph structures safely
    state_schema, nodes, edges, branches = _extract_langgraph_structures(state_graph)

    # 2. Extract and resolve State model
    pydantic_state: type[BaseModel]
    if state_schema is None:
        pydantic_state = DefaultState
    elif isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
        pydantic_state = state_schema
    else:
        try:
            pydantic_state = typeddict_to_pydantic(state_schema)
        except Exception:
            pydantic_state = DefaultState

    node_mappings = node_mappings or {}
    node_configs = node_configs or {}
    dspyer_graph = Graph()

    # 3. Process and register all nodes
    for node_name, node_spec in nodes.items():
        if node_name in node_mappings:
            node = node_mappings[node_name]
            if node_name in node_configs:
                cfg = node_configs[node_name]
                if "max_retries" in cfg:
                    node.max_retries = cfg["max_retries"]
                if "refine_instructions" in cfg:
                    node.refine_instructions = cfg["refine_instructions"]
                if "use_cot" in cfg:
                    node.use_cot = cfg["use_cot"]
                if "dataset_log_path" in cfg:
                    node.dataset_log_path = cfg["dataset_log_path"]
                if "redact_hook" in cfg:
                    node.redact_hook = cfg["redact_hook"]
                if "validation_log_path" in cfg:
                    node.validation_log_path = cfg["validation_log_path"]
            dspyer_graph.add_node(node)
        else:
            runnable = getattr(node_spec, "runnable", node_spec)
            func = getattr(runnable, "func", runnable)

            # Check if function is explicitly decorated
            if hasattr(func, "_dspyer_is_llm"):
                is_llm = getattr(func, "_dspyer_is_llm", True)
                input_model = getattr(func, "_dspyer_input_model", None) or make_permissive_model(
                    pydantic_state, f"{node_name}Input"
                )
                output_model = getattr(func, "_dspyer_output_model", None) or make_permissive_model(
                    pydantic_state, f"{node_name}Output"
                )
                instructions = (
                    getattr(func, "_dspyer_instructions", None)
                    or getattr(func, "__doc__", None)
                    or f"Execute agent step for node '{node_name}'."
                )
                if instructions:
                    instructions = instructions.strip()

                cfg = node_configs.get(node_name, {})
                node = StatefulNode(
                    name=node_name,
                    input_model=input_model,
                    output_model=output_model,
                    instructions=instructions,
                    max_retries=cfg.get("max_retries"),
                    refine_instructions=cfg.get("refine_instructions"),
                    use_cot=cfg.get("use_cot", False),
                    dataset_log_path=cfg.get("dataset_log_path"),
                    redact_hook=cfg.get("redact_hook"),
                    validation_log_path=cfg.get("validation_log_path"),
                )
                if not is_llm:
                    node.is_passthrough = True
                    node.callable = func
                else:
                    node._is_autogenerated = True
                dspyer_graph.add_node(node)
                continue

            is_llm, input_keys, output_keys, is_dynamic = analyze_node_function(func)

            if not is_llm:
                # Deterministic node: Execute as native Python passthrough
                permissive_input = make_permissive_model(pydantic_state, f"{node_name}Input")
                permissive_output = make_permissive_model(pydantic_state, f"{node_name}Output")
                cfg = node_configs.get(node_name, {})
                passthrough_node = StatefulNode(
                    name=node_name,
                    input_model=permissive_input,
                    output_model=permissive_output,
                    instructions=None,
                    max_retries=cfg.get("max_retries"),
                    refine_instructions=cfg.get("refine_instructions"),
                    use_cot=cfg.get("use_cot", False),
                    dataset_log_path=cfg.get("dataset_log_path"),
                    redact_hook=cfg.get("redact_hook"),
                    validation_log_path=cfg.get("validation_log_path"),
                )
                passthrough_node.is_passthrough = True
                passthrough_node.callable = func
                dspyer_graph.add_node(passthrough_node)
            else:
                if is_dynamic:
                    raise ValueError(
                        f"LLM-containing node '{node_name}' cannot be dynamically converted because it utilizes dynamic "
                        f"state accesses (e.g. dynamic keys, unpackings, or dynamic returns). "
                        f"Please map this node explicitly by providing a mapped dspyer.StatefulNode in 'node_mappings'."
                    )

                docstring = getattr(func, "__doc__", None)
                instructions = (
                    docstring.strip()
                    if docstring
                    else f"Execute agent step for node '{node_name}'."
                )

                import warnings

                warnings.warn(
                    f"Auto-generating LLM StatefulNode '{node_name}' with a dynamically derived narrow state schema. "
                    "This serves as a compiled DSPy scaffold stub and does not execute the original Python function body. "
                    "For a behavior-preserving execution or custom prompts, map this node explicitly using 'node_mappings'.",
                    UserWarning,
                    stacklevel=2,
                )

                input_fields: Dict[str, Any] = {}
                for key in input_keys:
                    if key in pydantic_state.model_fields:
                        field_info = pydantic_state.model_fields[key]
                        annotation = field_info.annotation or Any
                        input_fields[key] = (Optional[annotation], None)
                    else:
                        input_fields[key] = (Any, None)
                if not input_fields:
                    input_fields["placeholder"] = (Optional[str], "default")

                output_fields: Dict[str, Any] = {}
                for key in output_keys:
                    if key in pydantic_state.model_fields:
                        field_info = pydantic_state.model_fields[key]
                        annotation = field_info.annotation or Any
                        output_fields[key] = (Optional[annotation], None)
                    else:
                        output_fields[key] = (Any, None)
                if not output_fields:
                    output_fields["placeholder"] = (Optional[str], "default")

                narrow_input = create_model(f"{node_name}Input", **input_fields)
                narrow_output = create_model(f"{node_name}Output", **output_fields)

                cfg = node_configs.get(node_name, {})
                generated_node = StatefulNode(
                    name=node_name,
                    input_model=narrow_input,
                    output_model=narrow_output,
                    instructions=instructions,
                    max_retries=cfg.get("max_retries"),
                    refine_instructions=cfg.get("refine_instructions"),
                    use_cot=cfg.get("use_cot", False),
                    dataset_log_path=cfg.get("dataset_log_path"),
                    redact_hook=cfg.get("redact_hook"),
                    validation_log_path=cfg.get("validation_log_path"),
                )
                generated_node._is_autogenerated = True
                dspyer_graph.add_node(generated_node)

    # 4. Map static edges
    for source, target in edges:
        if source in ("__start__", "START"):
            # Set graph entry point
            dspyer_graph.set_entry_point(target)
        else:
            dspyer_graph.add_edge(source, target)

    # 5. Map conditional edges (branches)
    # branches maps source_node -> {router_name: BranchSpec}
    for source, branch_dict in branches.items():
        for router_key, branch_spec in branch_dict.items():
            router_callable = branch_spec.path
            router_func = getattr(router_callable, "func", router_callable)
            path_map = branch_spec.ends
            dspyer_graph.add_conditional_edges(source, router_func, path_map)

    return dspyer_graph

---

State Management

dspyer.state.ImmutableState

Represents an immutable snapshot of the agent's workflow state. Updates are applied via JSON Merge Patch (RFC 7396), returning a new ImmutableState instance without mutating the original.

Source code in dspyer/state.py

class ImmutableState:
    """
    Represents an immutable snapshot of the agent's workflow state.
    Updates are applied via JSON Merge Patch (RFC 7396), returning
    a new ImmutableState instance without mutating the original.
    """

    def __init__(self, data: Dict[str, Any], _skip_copy: bool = False):
        if _skip_copy:
            self._data = data
        else:
            self._data = deepcopy(data)

    def apply_patch(self, patch: Dict[str, Any]) -> "ImmutableState":
        """
        Applies a recursive JSON Merge Patch (RFC 7396) and returns
        a new ImmutableState instance containing the merged state using COW optimization.
        """
        new_data = self._merge_cow(self._data, patch)
        return ImmutableState(new_data, _skip_copy=True)

    def _merge_cow(self, target: Dict[str, Any], patch: Dict[str, Any]) -> Dict[str, Any]:
        """
        Copy-on-write merge routine. Reuses reference pointers for unchanged keys.
        """
        new_data = target.copy()
        for key, value in patch.items():
            if value is None:
                new_data.pop(key, None)
            elif isinstance(value, dict) and isinstance(new_data.get(key), dict):
                new_data[key] = self._merge_cow(new_data[key], value)
            else:
                if isinstance(value, (dict, list)):
                    new_data[key] = deepcopy(value)
                else:
                    new_data[key] = value
        return new_data

    def merge(self, other: "ImmutableState", policy: str = "last_write_wins") -> "ImmutableState":
        """
        Merges another state snapshot into this one.
        Supports conflict resolution policies: 'last_write_wins', 'combine_lists', 'raise'.
        """
        new_data = self._reconcile_cow(self._data, other.to_dict(), policy)
        return ImmutableState(new_data, _skip_copy=True)

    def _reconcile_cow(
        self, target: Dict[str, Any], source: Dict[str, Any], policy: str
    ) -> Dict[str, Any]:
        new_data = target.copy()
        for key, val in source.items():
            if key not in new_data:
                if isinstance(val, (dict, list)):
                    new_data[key] = deepcopy(val)
                else:
                    new_data[key] = val
            else:
                target_val = new_data[key]
                if isinstance(target_val, dict) and isinstance(val, dict):
                    new_data[key] = self._reconcile_cow(target_val, val, policy)
                elif (
                    isinstance(target_val, list)
                    and isinstance(val, list)
                    and policy == "combine_lists"
                ):
                    new_data[key] = target_val + deepcopy(val)
                elif target_val == val:
                    pass
                else:
                    if policy == "raise":
                        raise ValueError(
                            f"Conflict detected at key '{key}': target value '{target_val}' "
                            f"does not match source value '{val}'."
                        )
                    elif policy == "combine_lists":
                        if isinstance(val, (dict, list)):
                            new_data[key] = deepcopy(val)
                        else:
                            new_data[key] = val
                    else:
                        if isinstance(val, (dict, list)):
                            new_data[key] = deepcopy(val)
                        else:
                            new_data[key] = val
        return new_data

    def to_dict(self) -> Dict[str, Any]:
        """
        Returns a deep copy of the state data to protect mutability boundaries.
        """
        return deepcopy(self._data)

Methods:

apply_patch(patch)

Applies a recursive JSON Merge Patch (RFC 7396) and returns a new ImmutableState instance containing the merged state using COW optimization.

Source code in dspyer/state.py

def apply_patch(self, patch: Dict[str, Any]) -> "ImmutableState":
    """
    Applies a recursive JSON Merge Patch (RFC 7396) and returns
    a new ImmutableState instance containing the merged state using COW optimization.
    """
    new_data = self._merge_cow(self._data, patch)
    return ImmutableState(new_data, _skip_copy=True)

merge(other, policy='last_write_wins')

Merges another state snapshot into this one. Supports conflict resolution policies: 'last_write_wins', 'combine_lists', 'raise'.

Source code in dspyer/state.py

def merge(self, other: "ImmutableState", policy: str = "last_write_wins") -> "ImmutableState":
    """
    Merges another state snapshot into this one.
    Supports conflict resolution policies: 'last_write_wins', 'combine_lists', 'raise'.
    """
    new_data = self._reconcile_cow(self._data, other.to_dict(), policy)
    return ImmutableState(new_data, _skip_copy=True)

to_dict()

Returns a deep copy of the state data to protect mutability boundaries.

Source code in dspyer/state.py

def to_dict(self) -> Dict[str, Any]:
    """
    Returns a deep copy of the state data to protect mutability boundaries.
    """
    return deepcopy(self._data)

---

Utilities & Observability

dspyer.utils.generate_validation_report(validation_log_path)

Parses a validation log JSONL file and returns a formatted text report containing retry rates, failure rates, and top failing fields per node.

Source code in dspyer/utils.py

def generate_validation_report(validation_log_path: str) -> str:
    """
    Parses a validation log JSONL file and returns a formatted text report
    containing retry rates, failure rates, and top failing fields per node.
    """
    from collections import Counter

    node_events: Dict[str, List[Dict[str, Any]]] = {}
    try:
        with open(validation_log_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    event = json.loads(line)
                    node_name = event.get("node_name", "unknown")
                    if node_name not in node_events:
                        node_events[node_name] = []
                    node_events[node_name].append(event)
                except Exception:
                    continue
    except Exception:
        return "Error: Could not read validation log file."

    if not node_events:
        return "No validation events found in log file."

    lines = []
    lines.append("=" * 50)
    lines.append("           dspyer Batch Validation Report")
    lines.append("=" * 50)

    for node_name, events in sorted(node_events.items()):
        total_runs = len(events)
        successful_runs = sum(1 for e in events if e.get("success", False))
        failed_runs = total_runs - successful_runs
        runs_with_retries = sum(1 for e in events if e.get("retries_taken", 0) > 0)
        total_retries = sum(e.get("retries_taken", 0) for e in events)

        success_pct = (successful_runs / total_runs) * 100 if total_runs > 0 else 0.0
        failed_pct = (failed_runs / total_runs) * 100 if total_runs > 0 else 0.0
        retry_rate_pct = (runs_with_retries / total_runs) * 100 if total_runs > 0 else 0.0
        avg_retries = total_retries / total_runs if total_runs > 0 else 0.0

        field_counter: Counter = Counter()
        for e in events:
            for field in e.get("failed_fields", []):
                field_counter[field] += 1

        lines.append(f"\nNode: {node_name}")
        lines.append("-" * 50)
        lines.append(f"  Total Runs: {total_runs}")
        lines.append(f"  Successful Runs: {successful_runs} ({success_pct:.1f}%)")
        lines.append(f"  Failed Runs: {failed_runs} ({failed_pct:.1f}%)")
        lines.append(
            f"  Retry Rate: {retry_rate_pct:.1f}% ({runs_with_retries}/{total_runs} runs required retries)"
        )
        lines.append(f"  Average Retries: {avg_retries:.2f} per run")

        if field_counter:
            lines.append("  Top Failing Pydantic Fields:")
            total_errors = sum(field_counter.values())
            for field, count in field_counter.most_common():
                pct = (count / total_errors) * 100 if total_errors > 0 else 0.0
                lines.append(f"    - {field}: {count} errors ({pct:.1f}% of total errors)")
        else:
            lines.append("  Top Failing Pydantic Fields: None")

    lines.append("\n" + "=" * 50)
    return "\n".join(lines)

dspyer.utils.load_logged_dataset(dataset_log_path, input_keys)

Loads a JSONL file generated by self-correction logging and returns it as a list of dspy.Example objects ready for DSPy teleprompters.

Source code in dspyer/utils.py

def load_logged_dataset(
    dataset_log_path: str,
    input_keys: List[str],
) -> List[dspy.Example]:
    """
    Loads a JSONL file generated by self-correction logging and returns it
    as a list of dspy.Example objects ready for DSPy teleprompters.
    """
    examples = []
    try:
        with open(dataset_log_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    data = json.loads(line)
                    example = dspy.Example(**data).with_inputs(*input_keys)
                    examples.append(example)
                except Exception:
                    continue
    except Exception:
        pass
    return examples

dspyer.utils.BaseStorageAdapter

Abstract base class representing a storage sink for logging datasets and validation reports.

Source code in dspyer/utils.py

class BaseStorageAdapter:
    """Abstract base class representing a storage sink for logging datasets and validation reports."""

    def append_line(self, target: str, line: str) -> None:
        raise NotImplementedError

    async def append_line_async(self, target: str, line: str) -> None:
        raise NotImplementedError

dspyer.utils.FileStorageAdapter

Bases: BaseStorageAdapter

File storage adapter with thread-safe sync appending and thread-pooled async appending.

Source code in dspyer/utils.py

class FileStorageAdapter(BaseStorageAdapter):
    """File storage adapter with thread-safe sync appending and thread-pooled async appending."""

    def __init__(self):
        self._lock = threading.Lock()

    def append_line(self, target: str, line: str) -> None:
        with self._lock:
            with open(target, "a", encoding="utf-8") as f:
                f.write(line + "\n")

    async def append_line_async(self, target: str, line: str) -> None:
        import asyncio

        await asyncio.to_thread(self.append_line, target, line)

dspyer.utils.get_storage_adapter()

Retrieve the current globally configured storage adapter.

Source code in dspyer/utils.py

def get_storage_adapter() -> BaseStorageAdapter:
    """Retrieve the current globally configured storage adapter."""
    global _default_storage_adapter
    return _default_storage_adapter

dspyer.utils.set_storage_adapter(adapter)

Set the globally configured storage adapter.

Source code in dspyer/utils.py

def set_storage_adapter(adapter: BaseStorageAdapter) -> None:
    """Set the globally configured storage adapter."""
    global _default_storage_adapter
    _default_storage_adapter = adapter

---

Direct Connection Clients (DirectLM)

dspyer.compiler.DirectClient

Direct model client with optional httpx-based async connection pooling (bypassing LiteLLM at runtime). Supports Ollama, Gemini, Claude, and OpenAI API protocols with jittered exponential backoff.

Source code in dspyer/compiler.py

class DirectClient:
    """
    Direct model client with optional httpx-based async connection pooling (bypassing LiteLLM at runtime).
    Supports Ollama, Gemini, Claude, and OpenAI API protocols with jittered exponential backoff.
    """

    def __init__(
        self,
        provider: str,
        model: str,
        api_key: Optional[str] = None,
        api_base: Optional[str] = None,
        max_network_retries: int = 3,
        base_delay: float = 1.0,
    ):
        self.provider = provider.lower()
        self.model = model
        self.api_key = api_key or os.environ.get(f"{self.provider.upper()}_API_KEY")
        if not self.api_key and self.provider in ("google", "gemini"):
            self.api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")
        self.api_base = api_base
        self.max_network_retries = max_network_retries
        self.base_delay = base_delay
        self._sync_client: Optional[Any] = None
        self._async_client: Optional[Any] = None
        self.last_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}

        if not HAS_HTTPX:
            warnings.warn(
                "Optional dependency 'httpx' is missing. Falling back to urllib for network requests. "
                "Install 'httpx' (pip install httpx) to enable async execution and connection pooling.",
                UserWarning,
                stacklevel=2,
            )

    def _get_sync_client(self) -> Any:
        if not HAS_HTTPX:
            raise RuntimeError("httpx is required for connection pooling but not installed.")
        if self._sync_client is None:
            import httpx

            self._sync_client = httpx.Client(
                timeout=60.0,
                limits=httpx.Limits(max_keepalive_connections=10, max_connections=20),
            )
        return self._sync_client

    def _get_async_client(self) -> Any:
        if not HAS_HTTPX:
            raise RuntimeError("httpx is required for connection pooling but not installed.")
        if self._async_client is None:
            import httpx

            self._async_client = httpx.AsyncClient(
                timeout=60.0,
                limits=httpx.Limits(max_keepalive_connections=10, max_connections=20),
            )
        return self._async_client

    def close(self):
        """Close the sync connection pool."""
        if self._sync_client is not None:
            self._sync_client.close()
            self._sync_client = None

    async def aclose(self):
        """Close the async connection pool."""
        if self._async_client is not None:
            await self._async_client.aclose()
            self._async_client = None

    async def __aenter__(self):
        """Support for async context manager."""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Support for async context manager."""
        await self.aclose()

    def __del__(self):
        try:
            self.close()
        except Exception:
            pass
        if self._async_client is not None:
            logger.warning(
                "DirectClient connection pool leak detected: "
                "The async client was not closed. Please call await client.aclose() "
                "or use 'async with DirectClient(...) as client' context manager."
            )

    def _get_request_details(
        self, prompt: str, system_prompt: Optional[str] = None
    ) -> tuple[str, dict, dict]:
        """
        Formats URL, Headers, and JSON payload for the chosen provider.
        """
        url = ""
        headers = {"Content-Type": "application/json"}
        payload = {}

        if self.provider == "ollama":
            url = self.api_base or "http://localhost:11434/api/chat"
            payload = {
                "model": self.model,
                "messages": [
                    *([{"role": "system", "content": system_prompt}] if system_prompt else []),
                    {"role": "user", "content": prompt},
                ],
                "stream": False,
            }
        elif self.provider == "openai":
            url = self.api_base or "https://api.openai.com/v1/chat/completions"
            if self.api_key:
                headers["Authorization"] = f"Bearer {self.api_key}"
            payload = {
                "model": self.model,
                "messages": [
                    *([{"role": "system", "content": system_prompt}] if system_prompt else []),
                    {"role": "user", "content": prompt},
                ],
            }
        elif self.provider == "anthropic":
            url = self.api_base or "https://api.anthropic.com/v1/messages"
            if self.api_key:
                headers["x-api-key"] = self.api_key
            headers["anthropic-version"] = "2023-06-01"
            payload = {
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                **({"system": system_prompt} if system_prompt else {}),
                "max_tokens": 4096,
            }
        elif self.provider == "google":
            url = (
                self.api_base
                or f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent"
            )
            if self.api_key:
                headers["x-goog-api-key"] = self.api_key

            contents = []
            if system_prompt:
                payload["systemInstruction"] = {"parts": [{"text": system_prompt}]}
            contents.append({"parts": [{"text": prompt}]})
            payload["contents"] = contents
        else:
            raise ValueError(f"Unsupported provider: {self.provider}")

        return url, headers, payload

    def _extract_response(self, response_data: dict) -> str:
        """
        Extracts completion text from the provider's response JSON.
        """
        try:
            if self.provider == "ollama":
                return response_data["message"]["content"]
            elif self.provider == "openai":
                return response_data["choices"][0]["message"]["content"]
            elif self.provider == "anthropic":
                return response_data["content"][0]["text"]
            elif self.provider == "google":
                return response_data["candidates"][0]["content"]["parts"][0]["text"]
            else:
                raise ValueError(f"Unsupported provider: {self.provider}")
        except (KeyError, IndexError) as err:
            raise RuntimeError(
                f"Failed to parse model response: {err}. Raw response: {response_data}"
            )

    def _extract_usage(self, response_data: dict) -> dict[str, int]:
        """
        Extracts token usage details from response JSON.
        Returns a dict mapping keys to token counts.
        """
        usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
        try:
            if self.provider == "openai" and "usage" in response_data:
                u = response_data["usage"]
                usage["prompt_tokens"] = u.get("prompt_tokens", 0)
                usage["completion_tokens"] = u.get("completion_tokens", 0)
                usage["total_tokens"] = u.get("total_tokens", 0)
            elif self.provider == "anthropic" and "usage" in response_data:
                u = response_data["usage"]
                usage["prompt_tokens"] = u.get("input_tokens", 0)
                usage["completion_tokens"] = u.get("output_tokens", 0)
                usage["total_tokens"] = usage["prompt_tokens"] + usage["completion_tokens"]
            elif self.provider == "google" and "usageMetadata" in response_data:
                u = response_data["usageMetadata"]
                usage["prompt_tokens"] = u.get("promptTokenCount", 0)
                usage["completion_tokens"] = u.get("candidatesTokenCount", 0)
                usage["total_tokens"] = u.get("totalTokenCount", 0)
            elif self.provider == "ollama":
                usage["prompt_tokens"] = response_data.get("prompt_eval_count", 0)
                usage["completion_tokens"] = response_data.get("eval_count", 0)
                usage["total_tokens"] = usage["prompt_tokens"] + usage["completion_tokens"]
        except Exception:
            pass
        return usage

    def generate_sync(self, prompt: str, system_prompt: Optional[str] = None) -> str:
        url, headers, payload = self._get_request_details(prompt, system_prompt)
        data = json.dumps(payload).encode("utf-8")

        attempt = 0
        while True:
            try:
                if HAS_HTTPX:
                    client = self._get_sync_client()
                    res = client.post(url, headers=headers, content=data)
                    res.raise_for_status()
                    res_json = res.json()
                    self.last_usage = self._extract_usage(res_json)
                    return self._extract_response(res_json)
                else:
                    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
                    with urllib.request.urlopen(req, timeout=60.0) as response:
                        res_body = response.read().decode("utf-8")
                        res_json = json.loads(res_body)
                        self.last_usage = self._extract_usage(res_json)
                        return self._extract_response(res_json)
            except Exception as e:
                status_code = None
                if HAS_HTTPX:
                    import httpx

                    if isinstance(e, httpx.HTTPStatusError):
                        status_code = e.response.status_code
                if isinstance(e, urllib.error.HTTPError):
                    status_code = e.code

                is_transient = status_code in (429, 500, 502, 503, 504)
                if is_transient and attempt < self.max_network_retries:
                    attempt += 1
                    sleep_time = (self.base_delay * (2**attempt)) + random.uniform(0, 1.0)
                    logger.warning(
                        f"DirectClient sync request failed with status {status_code}. "
                        f"Retrying in {sleep_time:.2f}s (Attempt {attempt}/{self.max_network_retries})..."
                    )
                    time.sleep(sleep_time)
                else:
                    raise

    async def generate_async(self, prompt: str, system_prompt: Optional[str] = None) -> str:
        url, headers, payload = self._get_request_details(prompt, system_prompt)
        data = json.dumps(payload).encode("utf-8")

        attempt = 0
        while True:
            try:
                if HAS_HTTPX:
                    client = self._get_async_client()
                    res = await client.post(url, headers=headers, content=data)
                    res.raise_for_status()
                    res_json = res.json()
                    self.last_usage = self._extract_usage(res_json)
                    return self._extract_response(res_json)
                else:
                    loop = asyncio.get_event_loop()
                    return await loop.run_in_executor(
                        None, self.generate_sync, prompt, system_prompt
                    )
            except Exception as e:
                status_code = None
                if HAS_HTTPX:
                    import httpx

                    if isinstance(e, httpx.HTTPStatusError):
                        status_code = e.response.status_code
                if isinstance(e, urllib.error.HTTPError):
                    status_code = e.code

                is_transient = status_code in (429, 500, 502, 503, 504)
                if is_transient and attempt < self.max_network_retries:
                    attempt += 1
                    sleep_time = (self.base_delay * (2**attempt)) + random.uniform(0, 1.0)
                    logger.warning(
                        f"DirectClient async request failed with status {status_code}. "
                        f"Retrying in {sleep_time:.2f}s (Attempt {attempt}/{self.max_network_retries})...."
                    )
                    await asyncio.sleep(sleep_time)
                else:
                    raise

Methods:

__aenter__() async

Support for async context manager.

Source code in dspyer/compiler.py

async def __aenter__(self):
    """Support for async context manager."""
    return self

__aexit__(exc_type, exc_val, exc_tb) async

Support for async context manager.

Source code in dspyer/compiler.py

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Support for async context manager."""
    await self.aclose()

aclose() async

Close the async connection pool.

Source code in dspyer/compiler.py

async def aclose(self):
    """Close the async connection pool."""
    if self._async_client is not None:
        await self._async_client.aclose()
        self._async_client = None

close()

Close the sync connection pool.

Source code in dspyer/compiler.py

def close(self):
    """Close the sync connection pool."""
    if self._sync_client is not None:
        self._sync_client.close()
        self._sync_client = None

dspyer.compiler.DirectLM

Bases: BaseLM

Custom dspy.BaseLM subclass that wraps DirectClient. Integrates directly with DSPy's global runtime, history tracking, and teleprompters, bypassing LiteLLM entirely at execution time.

Source code in dspyer/compiler.py

class DirectLM(dspy.BaseLM):
    """
    Custom dspy.BaseLM subclass that wraps DirectClient.
    Integrates directly with DSPy's global runtime, history tracking, and teleprompters,
    bypassing LiteLLM entirely at execution time.
    """

    def __init__(
        self,
        model: str,
        api_key: Optional[str] = None,
        api_base: Optional[str] = None,
        max_network_retries: int = 3,
        base_delay: float = 1.0,
        **kwargs,
    ):
        provider = "openai"
        model_name = model
        if "/" in model:
            provider, model_name = model.split("/", 1)

        super().__init__(model=model, **kwargs)
        self.client = DirectClient(
            provider=provider,
            model=model_name,
            api_key=api_key,
            api_base=api_base,
            max_network_retries=max_network_retries,
            base_delay=base_delay,
        )

    async def __aenter__(self):
        """Support for async context manager."""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Support for async context manager."""
        await self.client.aclose()

    def forward(
        self, prompt: str | None = None, messages: list[dict[str, Any]] | None = None, **kwargs
    ):
        system_prompt = None
        user_prompt = ""
        if messages:
            system_msgs = [m["content"] for m in messages if m.get("role") == "system"]
            if system_msgs:
                system_prompt = "\n".join(system_msgs)

            user_msgs = [m["content"] for m in messages if m.get("role") in ("user", "developer")]
            if user_msgs:
                user_prompt = "\n".join(user_msgs)
            else:
                user_prompt = messages[-1]["content"] if messages else ""
        else:
            user_prompt = prompt or ""

        content = self.client.generate_sync(user_prompt, system_prompt=system_prompt)
        res = MockCompletionResult(content, self.model)
        res.usage = self.client.last_usage
        return res

    async def aforward(
        self, prompt: str | None = None, messages: list[dict[str, Any]] | None = None, **kwargs
    ):
        system_prompt = None
        user_prompt = ""
        if messages:
            system_msgs = [m["content"] for m in messages if m.get("role") == "system"]
            if system_msgs:
                system_prompt = "\n".join(system_msgs)

            user_msgs = [m["content"] for m in messages if m.get("role") in ("user", "developer")]
            if user_msgs:
                user_prompt = "\n".join(user_msgs)
            else:
                user_prompt = messages[-1]["content"] if messages else ""
        else:
            user_prompt = prompt or ""

        content = await self.client.generate_async(user_prompt, system_prompt=system_prompt)
        res = MockCompletionResult(content, self.model)
        res.usage = self.client.last_usage
        return res

Methods:

__aenter__() async

Support for async context manager.

Source code in dspyer/compiler.py

async def __aenter__(self):
    """Support for async context manager."""
    return self

__aexit__(exc_type, exc_val, exc_tb) async

Support for async context manager.

Source code in dspyer/compiler.py

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Support for async context manager."""
    await self.client.aclose()