butler/tools/orthogonality_lint.py

Rama main · réplica de lectura del repositorio original.

"""Builtin tool orthogonality lint — warn on overlapping descriptions within toolsets."""

from __future__ import annotations

from typing import Any, cast

# Builtin search tools intentionally overlap MCP scrape/fetch (layered fallback).
_BENIGN_SEARCH_TOOLS = frozenset({"web_search", "web_fetch"})


def _tool_name(defn: dict[str, Any]) -> str:
    fn_raw = defn.get("function")
    fn = fn_raw if isinstance(fn_raw, dict) else {}
    return str(fn.get("name") or defn.get("name") or "").strip()


def _tool_description(defn: dict[str, Any]) -> str:
    fn_raw = defn.get("function")
    fn = fn_raw if isinstance(fn_raw, dict) else {}
    return str(fn.get("description") or "").strip().lower()


def _tool_text(defn: dict[str, Any]) -> str:
    """Embedding text for orthogonality: description only (names are fixed API)."""
    return _tool_description(defn)


def _toolset_for(name: str) -> str:
    from butler.tools.orthogonality_lint_ops import toolset_for_safe

    ts = toolset_for_safe(name)
    if ts:
        return cast(str, ts)
    if name.startswith("mcp_"):
        parts = name.split("_", 2)
        if len(parts) >= 2:
            return f"mcp:{parts[1]}"
    return "default"


def _is_benign_overlap(name_a: str, name_b: str) -> bool:
    """Suppress expected builtin↔MCP search/scrape redundancy in /诊断."""
    a = str(name_a or "").lower()
    b = str(name_b or "").lower()
    names = {a, b}
    if names & _BENIGN_SEARCH_TOOLS:
        for n in names - _BENIGN_SEARCH_TOOLS:
            if "firecrawl" in n or n.startswith("mcp_fetch"):
                return True
    return False


def _canonical_pair(name_a: str, name_b: str) -> tuple[str, str]:
    return (min(name_a, name_b), max(name_a, name_b))


def _collect_tool_vectors(
    defs: list[dict[str, Any]],
    embedder: Any,
) -> list[tuple[str, str, list[float]]]:
    out: list[tuple[str, str, list[float]]] = []
    for defn in defs:
        name = _tool_name(defn)
        if not name:
            continue
        ts = _toolset_for(name)
        from butler.tools.orthogonality_lint_ops import embed_tool_text_safe

        vec = embed_tool_text_safe(embedder, _tool_text(defn))
        if vec is None:
            continue
        out.append((name, ts, vec))
    return out


def _lint_tool_vectors(
    items: list[tuple[str, str, list[float]]],
    *,
    threshold: float,
    max_pairs: int,
    cross_toolset: bool,
) -> list[str]:
    from butler.memory.embedding import cosine_similarity

    issues: list[tuple[float, str]] = []
    seen: set[tuple[str, str]] = set()
    for i in range(len(items)):
        n1, ts1, v1 = items[i]
        for j in range(i + 1, len(items)):
            n2, ts2, v2 = items[j]
            if not cross_toolset and ts1 != ts2:
                continue
            pair = _canonical_pair(n1, n2)
            if pair in seen:
                continue
            if _is_benign_overlap(n1, n2):
                continue
            sim = cosine_similarity(v1, v2)
            if sim >= threshold:
                seen.add(pair)
                scope = "cross" if ts1 != ts2 else ts1
                issues.append(
                    (
                        sim,
                        f"{scope}: {n1} ↔ {n2} cosine={sim:.3f} (≥{threshold})",
                    )
                )
    issues.sort(key=lambda x: x[0], reverse=True)
    return [line for _, line in issues[:max_pairs]]


def lint_builtin_tool_orthogonality(
    *,
    threshold: float = 0.82,
    max_pairs: int = 12,
) -> list[str]:
    """Return warning lines for semantically overlapping builtin tools."""
    from butler.tools.orthogonality_lint_ops import load_embedder_for_lint_safe
    from butler.tools.registry import get_tool_definitions

    embedder, skip = load_embedder_for_lint_safe()
    if embedder is None:
        return [skip or "orthogonality lint skipped: embedder unavailable"]
    if embedder.model_id.startswith("hashing"):
        return ["orthogonality lint skipped: hashing embedder (set real embedding model)"]

    defs = get_tool_definitions()
    items = _collect_tool_vectors(defs, embedder)
    return _lint_tool_vectors(items, threshold=threshold, max_pairs=max_pairs, cross_toolset=False)


def lint_tool_orthogonality_for_diagnostics(
    *,
    session_key: str = "",
    threshold: float = 0.82,
    max_pairs: int = 2,
) -> list[str]:
    """Builtin + MCP overlap warnings for /诊断 (deduped, noise-reduced)."""
    from butler.tools.orthogonality_lint_ops import (
        get_builtin_tool_definitions_safe,
        get_mcp_tool_definitions_safe,
        load_embedder_for_diagnostics_safe,
    )

    embedder = load_embedder_for_diagnostics_safe()
    if embedder is None:
        return []
    if embedder.model_id.startswith("hashing"):
        return []

    defs: list[dict[str, Any]] = list(get_builtin_tool_definitions_safe())

    mcp_defs = get_mcp_tool_definitions_safe(session_key)
    known = {_tool_name(d) for d in defs}
    for defn in mcp_defs:
        name = _tool_name(defn)
        if name and name not in known:
            defs.append(defn)

    if len(defs) < 2:
        return []

    items = _collect_tool_vectors(defs, embedder)
    return _lint_tool_vectors(
        items,
        threshold=threshold,
        max_pairs=max_pairs,
        cross_toolset=True,
    )


def format_orthogonality_report(issues: list[str]) -> str:
    if not issues:
        return "Builtin tool orthogonality: OK (no high-overlap pairs)"
    lines = ["Builtin tool orthogonality warnings:"]
    lines.extend(f"  - {i}" for i in issues)
    return "\n".join(lines)


__all__ = [
    "format_orthogonality_report",
    "lint_builtin_tool_orthogonality",
    "lint_tool_orthogonality_for_diagnostics",
]