mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-08-28 05:15:00 +00:00
884 lines
32 KiB
Python
884 lines
32 KiB
Python
"""
|
|
Tool Quality Manager
|
|
|
|
Core API (called by main flow):
|
|
- record_execution(): Called by BaseTool after execution
|
|
- adjust_ranking(): Called by SearchCoordinator for quality-aware sorting
|
|
- evolve(): Called periodically by ToolLayer for self-evolution
|
|
|
|
Query API (for inspection/debugging):
|
|
- get_quality_report(), get_tool_insights()
|
|
"""
|
|
|
|
import hashlib
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
|
|
|
|
from .types import ToolQualityRecord, ExecutionRecord, DescriptionQuality
|
|
from .store import QualityStore
|
|
from openspace.utils.logging import Logger
|
|
|
|
if TYPE_CHECKING:
|
|
from openspace.grounding.core.tool import BaseTool
|
|
from openspace.grounding.core.types import ToolResult
|
|
from openspace.llm import LLMClient
|
|
|
|
logger = Logger.get_logger(__name__)
|
|
|
|
|
|
class ToolQualityManager:
|
|
"""
|
|
Manages tool quality tracking and quality-aware ranking.
|
|
|
|
Features:
|
|
- Track execution success rate and latency
|
|
- LLM-based description quality evaluation (optional, requires llm_client)
|
|
- Persistent memory across sessions
|
|
- Quality-integrated tool ranking
|
|
- Incremental update detection
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
db_path: Optional[Path] = None,
|
|
cache_dir: Optional[Path] = None, # deprecated, ignored
|
|
llm_client: Optional["LLMClient"] = None,
|
|
enable_persistence: bool = True,
|
|
auto_save: bool = True,
|
|
evolve_interval: int = 5,
|
|
):
|
|
self._llm_client = llm_client
|
|
self._enable_persistence = enable_persistence
|
|
self._auto_save = auto_save
|
|
self._evolve_interval = evolve_interval
|
|
|
|
# In-memory cache
|
|
self._records: Dict[str, ToolQualityRecord] = {}
|
|
self._global_execution_count: int = 0
|
|
self._last_evolve_count: int = 0
|
|
|
|
# Persistent store (SQLite, shares DB file with SkillStore)
|
|
self._store = QualityStore(db_path=db_path) if enable_persistence else None
|
|
|
|
# Load from DB
|
|
if self._store:
|
|
self._records, self._global_execution_count = self._store.load_all()
|
|
self._last_evolve_count = (
|
|
(self._global_execution_count // self._evolve_interval)
|
|
* self._evolve_interval
|
|
)
|
|
|
|
logger.info(
|
|
f"ToolQualityManager initialized "
|
|
f"(persistence={enable_persistence}, records={len(self._records)}, "
|
|
f"global_count={self._global_execution_count}, "
|
|
f"evolve_interval={self._evolve_interval})"
|
|
)
|
|
|
|
def get_tool_key(self, tool: "BaseTool") -> str:
|
|
"""Generate unique key for a tool."""
|
|
from openspace.grounding.core.types import BackendType
|
|
|
|
if tool.is_bound:
|
|
backend = tool.runtime_info.backend.value
|
|
server = tool.runtime_info.server_name or "default"
|
|
else:
|
|
backend = tool.backend_type.value if tool.backend_type != BackendType.NOT_SET else "unknown"
|
|
server = "default"
|
|
|
|
return f"{backend}:{server}:{tool.name}"
|
|
|
|
def _compute_description_hash(self, tool: "BaseTool") -> str:
|
|
"""Compute hash of tool description for change detection."""
|
|
content = f"{tool.name}|{tool.description or ''}|{tool.schema.parameters}"
|
|
return hashlib.md5(content.encode()).hexdigest()[:16]
|
|
|
|
def get_record(self, tool: "BaseTool") -> ToolQualityRecord:
|
|
"""Get or create quality record for a tool."""
|
|
key = self.get_tool_key(tool)
|
|
|
|
if key not in self._records:
|
|
backend, server, name = key.split(":", 2)
|
|
self._records[key] = ToolQualityRecord(
|
|
tool_key=key,
|
|
backend=backend,
|
|
server=server,
|
|
tool_name=name,
|
|
description_hash=self._compute_description_hash(tool),
|
|
)
|
|
|
|
return self._records[key]
|
|
|
|
def get_quality_score(self, tool: "BaseTool") -> float:
|
|
"""Get quality score for a tool (0-1)."""
|
|
return self.get_record(tool).quality_score
|
|
|
|
# Key-based record access (for cross-system integration)
|
|
def get_or_create_record_by_key(self, tool_key: str) -> ToolQualityRecord:
|
|
"""Get or create a ToolQualityRecord by its canonical key.
|
|
|
|
Used by ExecutionAnalyzer integration where no BaseTool instance
|
|
is available. Parses ``tool_key`` into backend/server/tool_name.
|
|
|
|
Key formats:
|
|
- ``backend:server:tool_name`` → three-part key (canonical for MCP)
|
|
- ``backend:tool_name`` → two-part; tries ``backend:default:tool_name``
|
|
first for matching existing records.
|
|
"""
|
|
# 1. Direct match
|
|
if tool_key in self._records:
|
|
return self._records[tool_key]
|
|
|
|
parts = tool_key.split(":", 2)
|
|
if len(parts) == 3:
|
|
backend, server, name = parts
|
|
elif len(parts) == 2:
|
|
backend, name = parts
|
|
server = "default"
|
|
# Try normalized 3-part key before creating a new record
|
|
canonical = f"{backend}:default:{name}"
|
|
if canonical in self._records:
|
|
return self._records[canonical]
|
|
else:
|
|
backend, server, name = "unknown", "default", tool_key
|
|
|
|
canonical_key = f"{backend}:{server}:{name}"
|
|
if canonical_key in self._records:
|
|
return self._records[canonical_key]
|
|
|
|
record = ToolQualityRecord(
|
|
tool_key=canonical_key,
|
|
backend=backend,
|
|
server=server,
|
|
tool_name=name,
|
|
)
|
|
self._records[canonical_key] = record
|
|
return record
|
|
|
|
def find_record_by_key(self, key: str) -> Optional[ToolQualityRecord]:
|
|
"""Find a record by exact or partial tool key.
|
|
|
|
Tries in order:
|
|
1. Exact match (3-part ``backend:server:tool`` or 2-part)
|
|
2. Normalized 2-part → ``backend:default:tool``
|
|
3. Linear scan matching backend + tool_name (ignoring server)
|
|
"""
|
|
# 1. Exact
|
|
if key in self._records:
|
|
return self._records[key]
|
|
|
|
parts = key.split(":", 2)
|
|
if len(parts) == 2:
|
|
backend, tool_name = parts
|
|
# 2. Normalize
|
|
canonical = f"{backend}:default:{tool_name}"
|
|
if canonical in self._records:
|
|
return self._records[canonical]
|
|
# 3. Scan
|
|
for record in self._records.values():
|
|
if record.backend == backend and record.tool_name == tool_name:
|
|
return record
|
|
return None
|
|
|
|
async def record_llm_tool_issues(
|
|
self, tool_issues: List[str], task_id: str = "",
|
|
) -> int:
|
|
"""Record LLM-identified tool issues into the quality tracking system.
|
|
|
|
Each issue is injected as a failed ``ExecutionRecord`` in the tool's
|
|
``recent_executions`` via ``add_llm_issue()``, so it feeds into the
|
|
same ``recent_success_rate`` → ``penalty`` pipeline as rule-based
|
|
tracking. This means one unified set of quality metrics drives
|
|
ranking adjustments and future batch skill updates.
|
|
|
|
The LLM may catch semantic failures (HTTP 200 but wrong data,
|
|
misleading output, etc.) that rule-based tracking misses.
|
|
|
|
Args:
|
|
tool_issues: List of ``"key — description"`` strings.
|
|
Key formats: ``mcp:server:tool`` or ``backend:tool``.
|
|
task_id: Task ID for traceability (optional).
|
|
"""
|
|
updated = 0
|
|
for issue in tool_issues:
|
|
# Parse "key — description" or "key - description"
|
|
if "—" in issue:
|
|
key_part, _, description = issue.partition("—")
|
|
elif " - " in issue:
|
|
key_part, _, description = issue.partition(" - ")
|
|
else:
|
|
key_part, description = issue, ""
|
|
key_part = key_part.strip()
|
|
description = description.strip()
|
|
if not key_part:
|
|
continue
|
|
|
|
record = self.find_record_by_key(key_part)
|
|
if record is None:
|
|
record = self.get_or_create_record_by_key(key_part)
|
|
|
|
# Inject into the unified quality pipeline
|
|
tag = f"(task={task_id}) " if task_id else ""
|
|
record.add_llm_issue(f"{tag}{description}" if description else f"{tag}flagged by analysis LLM")
|
|
updated += 1
|
|
|
|
# Persist
|
|
if updated and self._auto_save and self._store:
|
|
await self._store.save_all(self._records, self._global_execution_count)
|
|
|
|
if updated:
|
|
logger.info(
|
|
f"Recorded {updated} LLM tool issue(s) into quality pipeline"
|
|
f"{f' (task={task_id})' if task_id else ''}"
|
|
)
|
|
return updated
|
|
|
|
def get_llm_flagged_tools(
|
|
self, min_flags: int = 2,
|
|
) -> List[ToolQualityRecord]:
|
|
"""Get tools repeatedly flagged by the analysis LLM.
|
|
|
|
Useful for identifying tools whose descriptions, reliability, or
|
|
behavior may need attention — and for batch-triggering skill
|
|
updates on skills that depend on those tools.
|
|
|
|
Each returned record carries LLM-identified failures in its
|
|
``recent_executions`` (prefixed ``[LLM]``), providing actionable context.
|
|
|
|
Args:
|
|
min_flags: Minimum number of LLM flags to include.
|
|
"""
|
|
return [
|
|
r for r in self._records.values()
|
|
if r.llm_flagged_count >= min_flags
|
|
]
|
|
|
|
# Execution Tracking
|
|
async def record_execution(
|
|
self,
|
|
tool: "BaseTool",
|
|
result: "ToolResult",
|
|
execution_time_ms: float,
|
|
) -> None:
|
|
"""Record tool execution result and increment global counter."""
|
|
record = self.get_record(tool)
|
|
|
|
# Extract error message if failed
|
|
error_message = None
|
|
if result.is_error and result.error:
|
|
error_message = str(result.error)[:500]
|
|
|
|
# Add execution record
|
|
record.add_execution(ExecutionRecord(
|
|
timestamp=datetime.now(),
|
|
success=result.is_success,
|
|
execution_time_ms=execution_time_ms,
|
|
error_message=error_message,
|
|
))
|
|
|
|
# Increment global execution count
|
|
self._global_execution_count += 1
|
|
|
|
# Auto-save
|
|
if self._auto_save and self._store:
|
|
await self._store.save_record(record, self._records, self._global_execution_count)
|
|
|
|
logger.debug(
|
|
f"Recorded execution: {record.tool_key} "
|
|
f"success={result.is_success} time={execution_time_ms:.0f}ms "
|
|
f"(global_count={self._global_execution_count})"
|
|
)
|
|
|
|
async def evaluate_description(
|
|
self,
|
|
tool: "BaseTool",
|
|
force: bool = False,
|
|
) -> Optional[DescriptionQuality]:
|
|
"""
|
|
Evaluate tool description quality using LLM.
|
|
"""
|
|
try:
|
|
from gdpval_bench.token_tracker import set_call_source, reset_call_source
|
|
_src_tok = set_call_source("quality")
|
|
except ImportError:
|
|
_src_tok = None
|
|
|
|
if not self._llm_client:
|
|
logger.debug("LLM client not available for description evaluation")
|
|
if _src_tok is not None:
|
|
reset_call_source(_src_tok)
|
|
return None
|
|
|
|
record = self.get_record(tool)
|
|
|
|
# Skip if already evaluated and not forced
|
|
if record.description_quality and not force:
|
|
# Check if description changed
|
|
current_hash = self._compute_description_hash(tool)
|
|
if current_hash == record.description_hash:
|
|
return record.description_quality
|
|
|
|
# Build evaluation prompt
|
|
desc = tool.description or "No description provided"
|
|
if len(desc) > 4000:
|
|
desc = desc[:4000] + "\n... (truncated for length)"
|
|
|
|
params = tool.schema.parameters or {}
|
|
if params:
|
|
param_lines = []
|
|
# Extract parameter names and types from JSON schema
|
|
if "properties" in params:
|
|
for param_name, param_info in params.get("properties", {}).items():
|
|
param_type = param_info.get("type", "unknown")
|
|
param_desc = param_info.get("description", "")
|
|
param_lines.append(f"- {param_name} ({param_type}): {param_desc}" if param_desc else f"- {param_name} ({param_type})")
|
|
param_text = "\n".join(param_lines) if param_lines else "No parameter descriptions available"
|
|
else:
|
|
param_text = "No parameters"
|
|
|
|
prompt = f"""# Task: Evaluate this tool's documentation quality
|
|
|
|
## Tool Information
|
|
|
|
Name: {tool.name}
|
|
|
|
Description:
|
|
{desc}
|
|
|
|
Parameters:
|
|
{param_text}
|
|
|
|
## Evaluation Task
|
|
|
|
Rate the documentation on two dimensions (0.0 to 1.0 scale):
|
|
|
|
### 1. Clarity
|
|
How clear is the tool's purpose and usage?
|
|
|
|
- 0.0-0.3: No description or completely unclear
|
|
- 0.4-0.6: Basic purpose but vague
|
|
- 0.7-0.8: Clear purpose and functionality
|
|
- 0.9-1.0: Very clear with usage examples or context
|
|
|
|
### 2. Completeness
|
|
Are inputs/outputs properly documented?
|
|
|
|
- 0.0-0.3: Missing critical information
|
|
- 0.4-0.6: Basic info but lacks details
|
|
- 0.7-0.8: Well documented with types
|
|
- 0.9-1.0: Comprehensive with constraints and examples
|
|
|
|
## Scoring Guidelines
|
|
|
|
- Short descriptions can score high if clear and accurate
|
|
- If parameters exist but aren't explained in description, reduce completeness score
|
|
- Missing description means clarity = 0.0
|
|
|
|
## Output
|
|
|
|
Respond with JSON only:
|
|
|
|
```json
|
|
{{
|
|
"reasoning": "Brief 1-2 sentence analysis",
|
|
"clarity": 0.8,
|
|
"completeness": 0.7
|
|
}}
|
|
```"""
|
|
|
|
try:
|
|
response = await self._llm_client.complete(prompt)
|
|
content = response["message"]["content"]
|
|
|
|
# Parse JSON response
|
|
import json
|
|
|
|
# Extract complete JSON object
|
|
def extract_json_object(text: str) -> str | None:
|
|
"""Extract first complete JSON object from text by counting braces."""
|
|
start = text.find('{')
|
|
if start == -1:
|
|
return None
|
|
|
|
count = 0
|
|
in_string = False
|
|
escape_next = False
|
|
|
|
for i, char in enumerate(text[start:], start):
|
|
if escape_next:
|
|
escape_next = False
|
|
continue
|
|
|
|
if char == '\\':
|
|
escape_next = True
|
|
continue
|
|
|
|
if char == '"' and not escape_next:
|
|
in_string = not in_string
|
|
continue
|
|
|
|
if not in_string:
|
|
if char == '{':
|
|
count += 1
|
|
elif char == '}':
|
|
count -= 1
|
|
if count == 0:
|
|
return text[start:i+1]
|
|
return None
|
|
|
|
json_str = extract_json_object(content)
|
|
if not json_str:
|
|
logger.warning(f"Could not find JSON in LLM response for {tool.name}")
|
|
return None
|
|
|
|
data = json.loads(json_str)
|
|
|
|
# Extract and validate scores with robust error handling
|
|
def safe_float(value, default=0.5, min_val=0.0, max_val=1.0):
|
|
"""Safely convert to float and clamp to valid range."""
|
|
try:
|
|
if value is None:
|
|
return default
|
|
f = float(value)
|
|
return max(min_val, min(max_val, f))
|
|
except (ValueError, TypeError):
|
|
logger.warning(f"Invalid score value: {value}, using default {default}")
|
|
return default
|
|
|
|
clarity = safe_float(data.get("clarity"), default=0.5)
|
|
completeness = safe_float(data.get("completeness"), default=0.5)
|
|
reasoning = str(data.get("reasoning", ""))[:500] # Limit reasoning length
|
|
|
|
quality = DescriptionQuality(
|
|
clarity=clarity,
|
|
completeness=completeness,
|
|
evaluated_at=datetime.now(),
|
|
reasoning=reasoning,
|
|
)
|
|
|
|
# Update record
|
|
record.description_quality = quality
|
|
record.description_hash = self._compute_description_hash(tool)
|
|
record.last_updated = datetime.now()
|
|
|
|
# Save
|
|
if self._auto_save and self._store:
|
|
await self._store.save_record(record, self._records, self._global_execution_count)
|
|
|
|
logger.info(f"Evaluated description: {tool.name} score={quality.overall_score:.2f}")
|
|
return quality
|
|
|
|
except Exception as e:
|
|
logger.error(f"Description evaluation failed for {tool.name}: {e}")
|
|
return None
|
|
finally:
|
|
if _src_tok is not None:
|
|
reset_call_source(_src_tok)
|
|
|
|
# Quality-Aware Ranking
|
|
def adjust_ranking(
|
|
self,
|
|
tools_with_scores: List[Tuple["BaseTool", float]],
|
|
) -> List[Tuple["BaseTool", float]]:
|
|
"""
|
|
Adjust tool ranking using penalty-based approach.
|
|
|
|
Args:
|
|
tools_with_scores: List of (tool, semantic_score) tuples
|
|
"""
|
|
adjusted = []
|
|
for tool, semantic_score in tools_with_scores:
|
|
penalty = self.get_penalty(tool)
|
|
|
|
adjusted_score = semantic_score * penalty
|
|
|
|
adjusted.append((tool, adjusted_score))
|
|
|
|
# Sort by adjusted score (descending)
|
|
adjusted.sort(key=lambda x: x[1], reverse=True)
|
|
|
|
return adjusted
|
|
|
|
def get_penalty(self, tool: "BaseTool") -> float:
|
|
"""Get penalty factor for a tool (0.2-1.0)."""
|
|
return self.get_record(tool).penalty
|
|
|
|
# Change Detection
|
|
def check_changes(self, tools: List["BaseTool"]) -> Dict[str, str]:
|
|
"""
|
|
Check for tool changes (new/updated/unchanged).
|
|
|
|
Returns dict: {tool_key: "new"|"updated"|"unchanged"}
|
|
"""
|
|
changes = {}
|
|
|
|
for tool in tools:
|
|
key = self.get_tool_key(tool)
|
|
current_hash = self._compute_description_hash(tool)
|
|
|
|
if key not in self._records:
|
|
changes[key] = "new"
|
|
elif self._records[key].description_hash != current_hash:
|
|
changes[key] = "updated"
|
|
# Clear old evaluation on description change
|
|
self._records[key].description_quality = None
|
|
self._records[key].description_hash = current_hash
|
|
else:
|
|
changes[key] = "unchanged"
|
|
|
|
new_count = sum(1 for v in changes.values() if v == "new")
|
|
updated_count = sum(1 for v in changes.values() if v == "updated")
|
|
|
|
if new_count or updated_count:
|
|
logger.info(f"Tool changes: {new_count} new, {updated_count} updated")
|
|
|
|
return changes
|
|
|
|
async def save(self) -> None:
|
|
"""
|
|
Manually save all records to disk.
|
|
|
|
Note: Usually not needed - auto_save handles persistence in
|
|
record_execution(), evaluate_description(), and evolve().
|
|
Provided as public API for explicit save when needed.
|
|
"""
|
|
if self._store:
|
|
await self._store.save_all(self._records)
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Clear all cached data."""
|
|
self._records.clear()
|
|
if self._store:
|
|
self._store.clear()
|
|
|
|
def get_stats(self) -> Dict:
|
|
"""
|
|
Get quality tracking statistics.
|
|
|
|
Note: Query API for inspection, may not be called in main flow.
|
|
"""
|
|
if not self._records:
|
|
return {"total_tools": 0}
|
|
|
|
records = list(self._records.values())
|
|
|
|
return {
|
|
"total_tools": len(records),
|
|
"total_executions": sum(r.total_calls for r in records),
|
|
"avg_success_rate": (
|
|
sum(r.success_rate for r in records) / len(records)
|
|
if records else 0
|
|
),
|
|
"avg_quality_score": (
|
|
sum(r.quality_score for r in records) / len(records)
|
|
if records else 0
|
|
),
|
|
"tools_with_description_eval": sum(
|
|
1 for r in records if r.description_quality
|
|
),
|
|
"tools_llm_flagged": sum(
|
|
1 for r in records if r.llm_flagged_count > 0
|
|
),
|
|
}
|
|
|
|
def get_top_tools(
|
|
self,
|
|
n: int = 10,
|
|
backend: Optional[str] = None,
|
|
min_calls: int = 3,
|
|
) -> List[ToolQualityRecord]:
|
|
"""
|
|
Get top N tools by quality score.
|
|
|
|
Args:
|
|
n: Number of tools to return
|
|
backend: Filter by backend type (optional)
|
|
min_calls: Minimum calls required (to filter untested tools)
|
|
"""
|
|
records = [
|
|
r for r in self._records.values()
|
|
if r.total_calls >= min_calls
|
|
and (backend is None or r.backend == backend)
|
|
]
|
|
|
|
records.sort(key=lambda r: r.quality_score, reverse=True)
|
|
return records[:n]
|
|
|
|
def get_problematic_tools(
|
|
self,
|
|
success_rate_threshold: float = 0.5,
|
|
min_calls: int = 5,
|
|
) -> List[ToolQualityRecord]:
|
|
"""
|
|
Get tools with low success rate (candidates for review/removal).
|
|
|
|
Args:
|
|
success_rate_threshold: Tools below this rate are flagged
|
|
min_calls: Minimum calls required (avoid flagging new tools)
|
|
"""
|
|
return [
|
|
r for r in self._records.values()
|
|
if r.total_calls >= min_calls
|
|
and r.recent_success_rate < success_rate_threshold
|
|
]
|
|
|
|
def get_quality_report(self) -> Dict:
|
|
"""
|
|
Generate comprehensive quality report for upper layer.
|
|
|
|
Returns structured report with:
|
|
- Overall stats
|
|
- Per-backend breakdown
|
|
- Top/problematic tools
|
|
- Improvement suggestions
|
|
"""
|
|
if not self._records:
|
|
return {"status": "no_data", "message": "No quality data collected yet"}
|
|
|
|
records = list(self._records.values())
|
|
tested_records = [r for r in records if r.total_calls >= 3]
|
|
|
|
# Per-backend stats
|
|
backends = {}
|
|
for r in records:
|
|
if r.backend not in backends:
|
|
backends[r.backend] = {
|
|
"tools": 0,
|
|
"total_calls": 0,
|
|
"success_count": 0,
|
|
"servers": set()
|
|
}
|
|
backends[r.backend]["tools"] += 1
|
|
backends[r.backend]["total_calls"] += r.total_calls
|
|
backends[r.backend]["success_count"] += r.success_count
|
|
backends[r.backend]["servers"].add(r.server)
|
|
|
|
# Convert sets to counts
|
|
for b in backends:
|
|
backends[b]["servers"] = len(backends[b]["servers"])
|
|
backends[b]["success_rate"] = (
|
|
backends[b]["success_count"] / backends[b]["total_calls"]
|
|
if backends[b]["total_calls"] > 0 else 0
|
|
)
|
|
|
|
# Top and problematic tools
|
|
top_tools = self.get_top_tools(5)
|
|
problematic = self.get_problematic_tools()
|
|
|
|
return {
|
|
"summary": {
|
|
"total_tools": len(records),
|
|
"tested_tools": len(tested_records),
|
|
"total_executions": sum(r.total_calls for r in records),
|
|
"overall_success_rate": (
|
|
sum(r.success_count for r in records) /
|
|
max(1, sum(r.total_calls for r in records))
|
|
),
|
|
"avg_quality_score": (
|
|
sum(r.quality_score for r in tested_records) / len(tested_records)
|
|
if tested_records else 0
|
|
),
|
|
},
|
|
"by_backend": backends,
|
|
"top_tools": [
|
|
{"key": r.tool_key, "score": r.quality_score, "success_rate": r.success_rate}
|
|
for r in top_tools
|
|
],
|
|
"problematic_tools": [
|
|
{"key": r.tool_key, "success_rate": r.success_rate, "calls": r.total_calls}
|
|
for r in problematic
|
|
],
|
|
"recommendations": self._generate_recommendations(records, problematic),
|
|
}
|
|
|
|
def _generate_recommendations(
|
|
self,
|
|
records: List[ToolQualityRecord],
|
|
problematic: List[ToolQualityRecord],
|
|
) -> List[str]:
|
|
"""Generate actionable recommendations based on quality data."""
|
|
recommendations = []
|
|
|
|
# Check for problematic tools
|
|
if problematic:
|
|
tool_names = [r.tool_name for r in problematic[:3]]
|
|
recommendations.append(
|
|
f"Review low-success tools: {', '.join(tool_names)}"
|
|
)
|
|
|
|
# Check for tools needing description evaluation
|
|
unevaluated = [r for r in records if not r.description_quality and r.total_calls >= 3]
|
|
if unevaluated:
|
|
recommendations.append(
|
|
f"{len(unevaluated)} tools need description quality evaluation"
|
|
)
|
|
|
|
# Check for low description quality
|
|
poor_docs = [
|
|
r for r in records
|
|
if r.description_quality and r.description_quality.overall_score < 0.5
|
|
]
|
|
if poor_docs:
|
|
recommendations.append(
|
|
f"{len(poor_docs)} tools have poor documentation quality"
|
|
)
|
|
|
|
return recommendations
|
|
|
|
def compute_adaptive_quality_weight(self) -> float:
|
|
"""
|
|
Compute adaptive quality weight based on data confidence.
|
|
|
|
Returns higher weight when we have more reliable quality data,
|
|
lower weight when data is sparse.
|
|
"""
|
|
if not self._records:
|
|
return 0.1 # Low weight when no data
|
|
|
|
records = list(self._records.values())
|
|
tested_count = sum(1 for r in records if r.total_calls >= 3)
|
|
|
|
if tested_count == 0:
|
|
return 0.1
|
|
|
|
# More tested tools -> higher confidence -> higher weight
|
|
coverage = tested_count / len(records)
|
|
|
|
# Average calls per tested tool -> data richness
|
|
avg_calls = sum(r.total_calls for r in records) / len(records)
|
|
richness = min(1.0, avg_calls / 20) # Cap at 20 calls average
|
|
|
|
# Combine coverage and richness
|
|
confidence = (coverage * 0.5 + richness * 0.5)
|
|
|
|
# Map to weight range [0.1, 0.5]
|
|
weight = 0.1 + confidence * 0.4
|
|
|
|
return round(weight, 2)
|
|
|
|
def should_reevaluate_description(self, tool: "BaseTool") -> bool:
|
|
"""
|
|
Check if a tool's description should be re-evaluated.
|
|
|
|
Triggers re-evaluation when:
|
|
- Description hash changed
|
|
- Success rate dropped significantly
|
|
- No evaluation yet but enough calls
|
|
"""
|
|
record = self._records.get(self.get_tool_key(tool))
|
|
if not record:
|
|
return True
|
|
|
|
# Check hash change
|
|
current_hash = self._compute_description_hash(tool)
|
|
if current_hash != record.description_hash:
|
|
return True
|
|
|
|
# No evaluation yet but enough data
|
|
if not record.description_quality and record.total_calls >= 5:
|
|
return True
|
|
|
|
# Success rate dropped significantly (maybe description is misleading)
|
|
if record.description_quality and record.total_calls >= 10:
|
|
if record.recent_success_rate < 0.5 and record.description_quality.overall_score > 0.7:
|
|
# High doc quality but low success -> mismatch
|
|
return True
|
|
|
|
return False
|
|
|
|
async def evolve(self, tools: List["BaseTool"]) -> Dict:
|
|
"""
|
|
Run self-evolution cycle on given tools.
|
|
|
|
This method:
|
|
1. Detects tool changes
|
|
2. Re-evaluates descriptions where needed
|
|
3. Updates quality weights
|
|
4. Returns evolution report
|
|
"""
|
|
report = {
|
|
"changes_detected": {},
|
|
"descriptions_evaluated": 0,
|
|
"adaptive_weight": 0.0,
|
|
"recommendations": [],
|
|
}
|
|
|
|
# 1. Detect changes
|
|
report["changes_detected"] = self.check_changes(tools)
|
|
|
|
# 2. Find tools needing re-evaluation
|
|
needs_eval = [t for t in tools if self.should_reevaluate_description(t)]
|
|
|
|
# 3. Evaluate descriptions (limit to avoid too many LLM calls)
|
|
if needs_eval and self._llm_client:
|
|
for tool in needs_eval[:5]: # Max 5 per cycle
|
|
result = await self.evaluate_description(tool, force=True)
|
|
if result:
|
|
report["descriptions_evaluated"] += 1
|
|
|
|
# 4. Compute adaptive weight
|
|
report["adaptive_weight"] = self.compute_adaptive_quality_weight()
|
|
|
|
# 5. Generate recommendations
|
|
problematic = self.get_problematic_tools()
|
|
report["recommendations"] = self._generate_recommendations(
|
|
list(self._records.values()), problematic
|
|
)
|
|
|
|
# 6. Update last evolve count
|
|
self._last_evolve_count = self._global_execution_count
|
|
|
|
# Save
|
|
if self._store:
|
|
await self._store.save_all(self._records, self._global_execution_count)
|
|
|
|
logger.info(
|
|
f"Evolution cycle complete: "
|
|
f"changes={len([v for v in report['changes_detected'].values() if v != 'unchanged'])}, "
|
|
f"evaluated={report['descriptions_evaluated']}, "
|
|
f"weight={report['adaptive_weight']}, "
|
|
f"global_count={self._global_execution_count}"
|
|
)
|
|
|
|
return report
|
|
|
|
def should_evolve(self) -> bool:
|
|
"""Check if evolution should be triggered based on global execution count."""
|
|
return self._global_execution_count >= self._last_evolve_count + self._evolve_interval
|
|
|
|
def get_tool_insights(self, tool: "BaseTool") -> Dict:
|
|
"""
|
|
Get detailed insights for a specific tool (for debugging/analysis).
|
|
|
|
Returns comprehensive info about tool's quality history.
|
|
"""
|
|
record = self._records.get(self.get_tool_key(tool))
|
|
if not record:
|
|
return {"status": "not_tracked", "tool": tool.name}
|
|
|
|
# Count recent failures
|
|
recent_failures_count = sum(
|
|
1 for e in record.recent_executions[-20:]
|
|
if not e.success
|
|
)
|
|
|
|
return {
|
|
"tool_key": record.tool_key,
|
|
"total_calls": record.total_calls,
|
|
"success_rate": record.success_rate,
|
|
"recent_success_rate": record.recent_success_rate,
|
|
"avg_execution_time_ms": record.avg_execution_time_ms,
|
|
"quality_score": record.quality_score,
|
|
"description_quality": {
|
|
"overall_score": record.description_quality.overall_score,
|
|
"clarity": record.description_quality.clarity,
|
|
"completeness": record.description_quality.completeness,
|
|
"reasoning": record.description_quality.reasoning,
|
|
} if record.description_quality else None,
|
|
"llm_flagged_count": record.llm_flagged_count,
|
|
"recent_failures_count": recent_failures_count,
|
|
"first_seen": record.first_seen.isoformat(),
|
|
"last_updated": record.last_updated.isoformat(),
|
|
}
|