icl_updater

icl_updater

ICL example updater + skill discovery from traces.

Classes

ICLUpdaterPolicy

ICLUpdaterPolicy(*, min_score: float = 0.7, max_examples: int = 20, min_skill_occurrences: int = 3, auto_apply: bool = False)

Bases: AgentLearningPolicy

Updates in-context examples and discovers skills from traces.

Analyzes traces for successful tool call patterns, extracts in-context learning examples, and discovers reusable multi-tool sequences ("skills"). This updates agent logic (ICL examples and tool-use strategies), not tool implementations themselves.

Source code in src/openjarvis/learning/icl_updater.py

def __init__(
    self,
    *,
    min_score: float = 0.7,
    max_examples: int = 20,
    min_skill_occurrences: int = 3,
    auto_apply: bool = False,
) -> None:
    self._min_score = min_score
    self._max_examples = max_examples
    self._min_skill_occurrences = min_skill_occurrences
    self._auto_apply = auto_apply
    self._examples: List[Dict[str, Any]] = []
    self._skills: List[Dict[str, Any]] = []
    # Versioned example database for add_example / rollback
    self._example_db: List[Dict[str, Any]] = []
    self._version: int = 0

Attributes

version property

version: int

Current version counter.

example_db property

example_db: List[Dict[str, Any]]

Return a copy of the versioned example database.

Functions

update

update(trace_store: Any, **kwargs: object) -> Dict[str, Any]

Analyze traces and extract ICL examples + skills.

Source code in src/openjarvis/learning/icl_updater.py

def update(self, trace_store: Any, **kwargs: object) -> Dict[str, Any]:
    """Analyze traces and extract ICL examples + skills."""
    try:
        traces = trace_store.list_traces()
    except Exception as exc:
        logger.warning("ICL updater failed: %s", exc)
        return {"examples": [], "skills": []}

    # Extract high-scoring traces with tool calls
    new_examples: List[Dict[str, Any]] = []
    tool_sequences: List[List[str]] = []

    for trace in traces:
        # Only consider successful traces
        if trace.outcome != "success":
            continue

        feedback = trace.feedback if trace.feedback is not None else 0.5
        if feedback < self._min_score:
            continue

        # Extract tool call steps
        tool_steps = [
            s
            for s in (trace.steps or [])
            if s.step_type.value == "tool_call"
        ]

        if tool_steps:
            # Build ICL example
            tool_names = [
                s.metadata.get("tool_name", "")
                for s in tool_steps
            ]
            example = {
                "query": trace.query,
                "tools_used": tool_names,
                "outcome": trace.outcome,
                "score": feedback,
            }
            new_examples.append(example)
            tool_sequences.append(tool_names)

    # Keep top examples by score
    new_examples.sort(key=lambda x: x["score"], reverse=True)
    self._examples = new_examples[: self._max_examples]

    # Discover skills: recurring multi-tool sequences
    self._skills = self._discover_skills(tool_sequences)

    return {
        "examples": list(self._examples),
        "skills": list(self._skills),
        "traces_analyzed": len(traces),
    }

add_example

add_example(query: str, response: str, outcome: float, metadata: Optional[Dict[str, Any]] = None) -> bool

Add an ICL example if it meets the quality threshold.

PARAMETER	DESCRIPTION
query	The user query that produced this example. TYPE: str
response	The agent/model response. TYPE: str
outcome	Quality score in [0, 1]. TYPE: float
metadata	Optional metadata dict attached to the example. TYPE: Optional[Dict[str, Any]] DEFAULT: None

RETURNS	DESCRIPTION
True if the example was accepted, False if rejected (below threshold).

Source code in src/openjarvis/learning/icl_updater.py

def add_example(
    self,
    query: str,
    response: str,
    outcome: float,
    metadata: Optional[Dict[str, Any]] = None,
) -> bool:
    """Add an ICL example if it meets the quality threshold.

    Parameters
    ----------
    query:
        The user query that produced this example.
    response:
        The agent/model response.
    outcome:
        Quality score in [0, 1].
    metadata:
        Optional metadata dict attached to the example.

    Returns
    -------
    True if the example was accepted, False if rejected (below threshold).
    """
    if outcome < self._min_score:
        return False

    self._version += 1
    entry: Dict[str, Any] = {
        "query": query,
        "response": response,
        "outcome": outcome,
        "metadata": metadata or {},
        "version": self._version,
    }
    self._example_db.append(entry)

    # Trim to max_examples (remove oldest first)
    if len(self._example_db) > self._max_examples:
        self._example_db = self._example_db[-self._max_examples:]

    return True

rollback

rollback(version: int) -> None

Remove all examples added after the given version.

PARAMETER	DESCRIPTION
version	The version checkpoint to rollback to. All examples with version > checkpoint are removed. TYPE: int

Source code in src/openjarvis/learning/icl_updater.py

def rollback(self, version: int) -> None:
    """Remove all examples added after the given version.

    Parameters
    ----------
    version:
        The version checkpoint to rollback to.  All examples with
        ``version > checkpoint`` are removed.
    """
    self._example_db = [
        ex for ex in self._example_db if ex["version"] <= version
    ]
    self._version = version

get_examples

get_examples(query_class: str = '', top_k: int = 5) -> List[Dict[str, Any]]

Retrieve the best examples, optionally filtered by query class.

PARAMETER	DESCRIPTION
query_class	If non-empty, only return examples whose query contains this substring (case-insensitive). TYPE: str DEFAULT: ''
top_k	Maximum number of examples to return. TYPE: int DEFAULT: 5

RETURNS	DESCRIPTION
Up to *top_k* examples sorted by outcome (descending).

Source code in src/openjarvis/learning/icl_updater.py

def get_examples(
    self,
    query_class: str = "",
    top_k: int = 5,
) -> List[Dict[str, Any]]:
    """Retrieve the best examples, optionally filtered by query class.

    Parameters
    ----------
    query_class:
        If non-empty, only return examples whose query contains this
        substring (case-insensitive).
    top_k:
        Maximum number of examples to return.

    Returns
    -------
    Up to *top_k* examples sorted by outcome (descending).
    """
    pool = self._example_db
    if query_class:
        lc = query_class.lower()
        pool = [ex for ex in pool if lc in ex["query"].lower()]

    # Sort by outcome descending, take top_k
    ranked = sorted(pool, key=lambda ex: ex["outcome"], reverse=True)
    return ranked[:top_k]