Merge pull request #85 from kayba-ai/test/David/Opik

feat(opik): Redesign Opik integration with pipeline-level tracing

Author: davidfarah2003

AGENTS.md

@@ -4,11 +4,13 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ## Repository Guidelines
 
-### Design Document Maintenance
+### Documentation Maintenance
 Before working on code in `ace/` or `ace_next/`, read `docs/ACE_DESIGN.md` to understand the current architecture.
 Before working on code in `pipeline/` or `ace_next/core/`, read `docs/PIPELINE_DESIGN.md` to understand the pipeline engine.
 
-When changes to code alter architecture, add new modules, change public APIs, rename concepts, or modify execution flow described in these documents, update the respective design doc to reflect the changes:
+**Docs MUST be kept in sync with code.** Any change that alters a public API, renames a concept, adds/removes a module, or changes execution flow **requires** a corresponding update to the relevant docs. Do not merge code changes that make the documentation inaccurate.
+
+Key design docs:
 - `docs/ACE_DESIGN.md` — core ACE architecture: roles, skillbook, adaptation loops, insight levels, integration patterns
 - `docs/PIPELINE_DESIGN.md` — pipeline engine: steps, StepProtocol, Pipeline, SubRunner, RR pipeline
 
@@ -20,6 +22,14 @@ When changes to code alter architecture, add new modules, change public APIs, re
 - `examples/` — runnable demos grouped by integration
 - `benchmarks/`, `scripts/` — research/evaluation tooling (not shipped to PyPI)
 - `docs/` — guides and reference material
+  - `docs/getting-started/` — installation, setup, quick-start
+  - `docs/concepts/` — core concepts: roles, skillbook, insight levels
+  - `docs/guides/` — in-depth guides: full pipeline, integration, testing, prompts
+  - `docs/integrations/` — per-integration docs (LiteLLM, LangChain, browser-use, Claude Code, Opik)
+  - `docs/pipeline/` — pipeline engine docs: core concepts, custom steps, branching, error handling
+  - `docs/api/` — API reference
+  - `docs/ACE_DESIGN.md` — architecture design doc (keep in sync with code)
+  - `docs/PIPELINE_DESIGN.md` — pipeline engine design doc (keep in sync with code)
 
 ### Commands
 - `uv sync` — install all dependencies

ace_next/__init__.py

@@ -28,6 +28,7 @@
     LangChain,
     TraceAnalyser,
 )
+from .rr import RRConfig, RRStep
 from .steps.opik import OPIK_AVAILABLE, OpikStep, register_opik_litellm_callback
 
 __all__ = [
@@ -56,6 +57,9 @@
     "ClaudeCode",
     "LangChain",
     "TraceAnalyser",
+    # Recursive Reflector
+    "RRStep",
+    "RRConfig",
     # Deduplication
     "DeduplicationConfig",
     "DeduplicationManager",

ace_next/runners/ace.py

@@ -7,7 +7,7 @@
 from typing import Any
 
 from pipeline import Pipeline
-from pipeline.protocol import SampleResult
+from pipeline.protocol import SampleResult, StepProtocol
 
 from ..core.context import ACEStepContext, SkillbookView
 from ..core.environments import Sample, TaskEnvironment
@@ -48,6 +48,7 @@ def from_roles(
         dedup_interval: int = 10,
         checkpoint_dir: str | Path | None = None,
         checkpoint_interval: int = 10,
+        extra_steps: list[StepProtocol] | None = None,
     ) -> ACE:
         """Construct from pre-built role instances.
 
@@ -64,6 +65,8 @@ def from_roles(
             dedup_interval: Samples between deduplication runs.
             checkpoint_dir: Directory for checkpoint files.
             checkpoint_interval: Samples between checkpoint saves.
+            extra_steps: Additional steps appended after the learning
+                tail (e.g. ``OpikStep``).
         """
         skillbook = skillbook or Skillbook()
         steps = [
@@ -79,6 +82,8 @@ def from_roles(
                 checkpoint_interval=checkpoint_interval,
             ),
         ]
+        if extra_steps:
+            steps.extend(extra_steps)
         return cls(pipeline=Pipeline(steps), skillbook=skillbook)
 
     def run(

ace_next/runners/litellm.py

@@ -67,6 +67,9 @@ def __init__(
         checkpoint_dir: str | Path | None = None,
         checkpoint_interval: int = 10,
         is_learning: bool = True,
+        opik: bool = False,
+        opik_project: str = "ace-framework",
+        opik_tags: list[str] | None = None,
     ) -> None:
         # Resolve skillbook
         if skillbook_path:
@@ -98,6 +101,29 @@ def __init__(
         self._checkpoint_dir = checkpoint_dir
         self._checkpoint_interval = checkpoint_interval
 
+        # Opik observability (explicit opt-in — fail loudly)
+        self._opik_step: Any = None
+        if opik:
+            from ..steps.opik import OPIK_AVAILABLE, OpikStep, register_opik_litellm_callback
+
+            if not OPIK_AVAILABLE:
+                raise ImportError(
+                    "opik=True requires the 'opik' package. "
+                    "Install it with: pip install ace-framework[observability]"
+                )
+
+            self._opik_step = OpikStep(
+                project_name=opik_project,
+                tags=opik_tags,
+            )
+            if not self._opik_step.enabled:
+                raise RuntimeError(
+                    "OpikStep failed to initialize. Check your Opik configuration "
+                    "(~/.opik.config, OPIK_API_KEY, OPIK_WORKSPACE env vars)."
+                )
+            # Register LiteLLM-level callback for per-call token/cost tracking
+            register_opik_litellm_callback(project_name=opik_project)
+
         # Lazy-init caches
         self._ace: ACE | None = None
         self._analyser: TraceAnalyser | None = None
@@ -128,6 +154,9 @@ def from_model(
         checkpoint_dir: Optional[Union[str, Path]] = None,
         checkpoint_interval: int = 10,
         is_learning: bool = True,
+        opik: bool = False,
+        opik_project: str = "ace-framework",
+        opik_tags: Optional[list[str]] = None,
         **llm_kwargs: Any,
     ) -> ACELiteLLM:
         """Build from a model string (creates LiteLLMClient + roles).
@@ -148,6 +177,10 @@ def from_model(
             checkpoint_dir: Directory for checkpoint files.
             checkpoint_interval: Samples between checkpoint saves.
             is_learning: Whether learning is enabled.
+            opik: Enable Opik observability (pipeline traces +
+                LiteLLM per-call token/cost tracking).
+            opik_project: Opik project name.
+            opik_tags: Tags applied to every Opik trace.
             **llm_kwargs: Extra kwargs forwarded to ``LiteLLMClient``.
         """
         from ..providers import LiteLLMClient
@@ -170,12 +203,21 @@ def from_model(
             checkpoint_dir=checkpoint_dir,
             checkpoint_interval=checkpoint_interval,
             is_learning=is_learning,
+            opik=opik,
+            opik_project=opik_project,
+            opik_tags=opik_tags,
         )
 
     # ------------------------------------------------------------------
     # Lazy-init runners
     # ------------------------------------------------------------------
 
+    def _get_extra_steps(self) -> list[Any] | None:
+        """Return extra pipeline steps (e.g. OpikStep) or None."""
+        if self._opik_step is not None:
+            return [self._opik_step]
+        return None
+
     def _get_ace(self, environment: TaskEnvironment | None = None) -> ACE:
         """Return (or build) the cached ACE runner."""
         env = environment or self.environment
@@ -194,6 +236,7 @@ def _get_ace(self, environment: TaskEnvironment | None = None) -> ACE:
                 dedup_interval=self._dedup_interval,
                 checkpoint_dir=self._checkpoint_dir,
                 checkpoint_interval=self._checkpoint_interval,
+                extra_steps=self._get_extra_steps(),
             )
         return self._ace
 
@@ -208,6 +251,7 @@ def _get_analyser(self) -> TraceAnalyser:
                 dedup_interval=self._dedup_interval,
                 checkpoint_dir=self._checkpoint_dir,
                 checkpoint_interval=self._checkpoint_interval,
+                extra_steps=self._get_extra_steps(),
             )
         return self._analyser
 

ace_next/runners/trace_analyser.py

@@ -7,7 +7,7 @@
 from typing import Any
 
 from pipeline import Pipeline
-from pipeline.protocol import SampleResult
+from pipeline.protocol import SampleResult, StepProtocol
 
 from ..core.context import ACEStepContext, SkillbookView
 from ..protocols import (
@@ -47,6 +47,7 @@ def from_roles(
         dedup_interval: int = 10,
         checkpoint_dir: str | Path | None = None,
         checkpoint_interval: int = 10,
+        extra_steps: list[StepProtocol] | None = None,
     ) -> TraceAnalyser:
         """Construct from pre-built role instances.
 
@@ -60,6 +61,8 @@ def from_roles(
             checkpoint_dir: Directory for checkpoint files.  Appends a
                 ``CheckpointStep`` when provided.
             checkpoint_interval: Samples between checkpoint saves.
+            extra_steps: Additional steps appended after the learning
+                tail (e.g. ``OpikStep``).
         """
         skillbook = skillbook or Skillbook()
         steps = learning_tail(
@@ -71,6 +74,8 @@ def from_roles(
             checkpoint_dir=checkpoint_dir,
             checkpoint_interval=checkpoint_interval,
         )
+        if extra_steps:
+            steps.extend(extra_steps)
         return cls(pipeline=Pipeline(steps), skillbook=skillbook)
 
     def run(

ace_next/steps/opik.py

@@ -2,17 +2,30 @@
 
 Terminal side-effect step that creates an Opik trace per sample with
 pipeline metadata, agent output, reflection insights, and skill
-manager operations.  Optionally registers the LiteLLM OpikLogger
-callback for per-LLM-call token/cost tracking.
+manager operations.
 
 Place at the end of the pipeline (after ApplyStep).  Gracefully
 degrades to a no-op when Opik is not installed or is disabled.
+
+**Explicit opt-in only** — constructing an ``OpikStep`` is the
+opt-in signal.  Opik is never auto-enabled just because the package
+is installed.
+
+Two independent tracing modes:
+
+1. **Pipeline step** (this class) — client-agnostic, reads
+   ``ACEStepContext`` fields and creates one Opik trace per sample.
+2. **LiteLLM callback** (``register_opik_litellm_callback``) —
+   LiteLLM-specific, registers ``OpikLogger`` on
+   ``litellm.callbacks`` for per-LLM-call token/cost tracking.
+   Call separately when needed; ``OpikStep`` does NOT register it.
 """
 
 from __future__ import annotations
 
 import logging
 import os
+import warnings
 from typing import Any
 
 from ..core.context import ACEStepContext
@@ -44,7 +57,8 @@ def register_opik_litellm_callback(
     """Register ``OpikLogger`` on ``litellm.callbacks`` for token/cost tracking.
 
     Standalone helper — call this when you want LiteLLM-level Opik
-    tracing without adding ``OpikStep`` to the pipeline.
+    tracing.  This is **separate** from ``OpikStep`` (pipeline-level
+    tracing) and must be called explicitly.
 
     Returns ``True`` if the callback was successfully registered.
     """
@@ -54,10 +68,28 @@ def register_opik_litellm_callback(
         import litellm
         from litellm.integrations.opik.opik import OpikLogger
 
-        opik_logger = OpikLogger(project_name=project_name)
+        # OpikLogger.__init__ calls asyncio.create_task() which spews
+        # ERROR logs + RuntimeWarnings when no event loop is running.
+        # Suppress only those specific messages during init.
+        class _AsyncInitFilter(logging.Filter):
+            def filter(self, record: logging.LogRecord) -> bool:
+                return "Asynchronous processing not initialized" not in record.getMessage()
+
+        _ll = logging.getLogger("LiteLLM")
+        _f = _AsyncInitFilter()
+        _ll.addFilter(_f)
+        try:
+            with warnings.catch_warnings():
+                warnings.filterwarnings(
+                    "ignore",
+                    message="coroutine.*periodic_flush.*never awaited",
+                    category=RuntimeWarning,
+                )
+                opik_logger = OpikLogger(project_name=project_name)
+        finally:
+            _ll.removeFilter(_f)
         already = any(
-            getattr(cb, "__class__", None) and cb.__class__.__name__.lower() == "opik"
-            for cb in getattr(litellm, "callbacks", [])
+            isinstance(cb, OpikLogger) for cb in getattr(litellm, "callbacks", [])
         )
         if not already:
             litellm.callbacks.append(opik_logger)
@@ -77,11 +109,13 @@ class OpikStep:
     Pure side-effect step — reads context fields and creates an Opik
     trace per sample.  Never mutates the context.
 
+    **Does NOT register the LiteLLM callback.**  Call
+    ``register_opik_litellm_callback()`` separately if you also want
+    per-LLM-call token/cost tracking.
+
     Args:
         project_name: Opik project name.
         tags: Tags applied to every trace.
-        register_litellm_callback: Also register ``OpikLogger`` on
-            ``litellm.callbacks`` for per-LLM-call token/cost tracking.
     """
 
     requires = frozenset({"skillbook"})
@@ -91,7 +125,6 @@ def __init__(
         self,
         project_name: str = "ace-framework",
         tags: list[str] | None = None,
-        register_litellm_callback: bool = True,
     ) -> None:
         self.project_name = project_name
         self.tags = tags or ["ace"]
@@ -100,14 +133,21 @@ def __init__(
 
         if self.enabled:
             try:
-                self._client = _opik.Opik(project_name=project_name)
+                # Pass config explicitly from env vars so we never depend
+                # on the global ~/.opik.config file.
+                api_key = os.environ.get("OPIK_API_KEY")
+                workspace = os.environ.get("OPIK_WORKSPACE")
+                host = os.environ.get("OPIK_URL_OVERRIDE")
+                self._client = _opik.Opik(
+                    project_name=project_name,
+                    api_key=api_key or None,
+                    workspace=workspace or None,
+                    host=host or None,
+                )
             except Exception as exc:
                 logger.debug("OpikStep: failed to create Opik client: %s", exc)
                 self.enabled = False
 
-        if self.enabled and register_litellm_callback:
-            register_opik_litellm_callback(project_name=project_name)
-
     def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
         if not self.enabled:
             return ctx