agent_evolver

agent_evolver

AgentConfigEvolver — analyze traces to evolve agent TOML configs.

Reads interaction traces to determine which agent/tool/parameter combinations perform best for different query classes, then writes updated TOML config files with automatic versioning and rollback.

Classes

AgentConfigEvolver

AgentConfigEvolver(
    trace_store: TraceStore,
    *,
    config_dir: Union[str, Path],
    min_quality: float = 0.5,
)

Analyze traces to evolve agent TOML configs with versioning.

PARAMETER	DESCRIPTION
trace_store	A :class:TraceStore used to fetch historical traces. TYPE: TraceStore
config_dir	Directory where agent TOML configs are written. TYPE: Union[str, Path]
min_quality	Minimum average feedback score for a recommendation to be emitted. TYPE: float DEFAULT: 0.5

Source code in src/diapason/learning/agents/agent_evolver.py

def __init__(
    self,
    trace_store: TraceStore,
    *,
    config_dir: Union[str, Path],
    min_quality: float = 0.5,
) -> None:
    self._store = trace_store
    self._config_dir = Path(config_dir)
    self._history_dir = self._config_dir / ".history"
    self._min_quality = min_quality

    self._config_dir.mkdir(parents=True, exist_ok=True)
    self._history_dir.mkdir(parents=True, exist_ok=True)

Methods:

analyze

analyze() -> List[Dict[str, Any]]

Analyze traces, return recommendations per query class.

Returns a list of dicts, each containing: - query_class: the classified query category - recommended_tools: list of tool names sorted by frequency - recommended_agent: the best-performing agent for this class - recommended_max_turns: suggested max_turns value - sample_count: number of traces analyzed for this class

Source code in src/diapason/learning/agents/agent_evolver.py

def analyze(self) -> List[Dict[str, Any]]:
    """Analyze traces, return recommendations per query class.

    Returns a list of dicts, each containing:
    - ``query_class``: the classified query category
    - ``recommended_tools``: list of tool names sorted by frequency
    - ``recommended_agent``: the best-performing agent for this class
    - ``recommended_max_turns``: suggested max_turns value
    - ``sample_count``: number of traces analyzed for this class
    """
    traces = self._store.list_traces(limit=10_000)
    if not traces:
        return []

    # Group traces by query class
    groups: Dict[str, List[Trace]] = defaultdict(list)
    for trace in traces:
        qclass = classify_query(trace.query)
        groups[qclass].append(trace)

    recommendations: List[Dict[str, Any]] = []
    for qclass, class_traces in sorted(groups.items()):
        rec = self._analyze_class(qclass, class_traces)
        if rec is not None:
            recommendations.append(rec)

    return recommendations

write_config

write_config(
    agent_name: str,
    *,
    tools: List[str],
    max_turns: int = 10,
    temperature: float = 0.3,
    system_prompt: str = "",
) -> Path

Write agent TOML config, archiving previous version first.

Returns the :class:Path to the written config file.

Source code in src/diapason/learning/agents/agent_evolver.py

def write_config(
    self,
    agent_name: str,
    *,
    tools: List[str],
    max_turns: int = 10,
    temperature: float = 0.3,
    system_prompt: str = "",
) -> Path:
    """Write agent TOML config, archiving previous version first.

    Returns the :class:`Path` to the written config file.
    """
    config_path = self._config_dir / f"{agent_name}.toml"

    # Archive the existing config before overwriting
    if config_path.exists():
        self._archive(agent_name, config_path)

    # Build the TOML data
    data = {
        "agent": {
            "name": agent_name,
            "tools": tools,
            "max_turns": max_turns,
            "temperature": temperature,
            "system_prompt": system_prompt,
        }
    }

    _write_toml(config_path, data)
    return config_path

list_versions

list_versions(agent_name: str) -> List[Dict[str, Any]]

List all versions (including current) for agent_name.

Returns a list of dicts with version, path, and modified. Versions are numbered starting from 1 (oldest archived) through to the current (highest version number).

Source code in src/diapason/learning/agents/agent_evolver.py

def list_versions(self, agent_name: str) -> List[Dict[str, Any]]:
    """List all versions (including current) for *agent_name*.

    Returns a list of dicts with ``version``, ``path``, and ``modified``.
    Versions are numbered starting from 1 (oldest archived) through to
    the current (highest version number).
    """
    versions: List[Dict[str, Any]] = []

    # Collect archived versions from .history/
    pattern = f"{agent_name}.v*.toml"
    archived = sorted(self._history_dir.glob(pattern))
    for idx, archived_path in enumerate(archived, start=1):
        versions.append(
            {
                "version": idx,
                "path": str(archived_path),
                "modified": archived_path.stat().st_mtime,
            }
        )

    # Current version
    current = self._config_dir / f"{agent_name}.toml"
    if current.exists():
        versions.append(
            {
                "version": len(versions) + 1,
                "path": str(current),
                "modified": current.stat().st_mtime,
            }
        )

    return versions

rollback

rollback(agent_name: str, version: int) -> None

Rollback to a specific version.

Raises :class:ValueError if the requested version does not exist.

Source code in src/diapason/learning/agents/agent_evolver.py

def rollback(self, agent_name: str, version: int) -> None:
    """Rollback to a specific version.

    Raises :class:`ValueError` if the requested version does not exist.
    """
    versions = self.list_versions(agent_name)
    target = None
    for v in versions:
        if v["version"] == version:
            target = v
            break

    if target is None:
        raise ValueError(
            f"Version {version} not found for agent '{agent_name}'. "
            f"Available versions: {[v['version'] for v in versions]}"
        )

    target_path = Path(target["path"])
    config_path = self._config_dir / f"{agent_name}.toml"

    # If the target is already the current file, nothing to do
    if target_path == config_path:
        return

    # Archive current before rollback
    if config_path.exists():
        self._archive(agent_name, config_path)

    # Copy the target version to become the current config
    shutil.copy2(str(target_path), str(config_path))

Functions: