lambdagent/agentexample/agentbuilder67/tools/config_search.py

"""
config_search.py — 配置库检索工具
==================================

为 Retriever agent 提供配置搜索和读取能力。
搜索 experiments/raw_configs/ 中的 437 个真实世界配置。
"""
from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Dict, List, Optional

import yaml

PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent
RAW_CONFIGS_DIR = PROJECT_ROOT / "experiments" / "raw_configs"

# ── 索引缓存 ──────────────────────────────────────────────
_index: Optional[List[Dict]] = None


def _build_index() -> List[Dict]:
    """扫描 raw_configs 目录，构建配置索引。"""
    global _index
    if _index is not None:
        return _index

    _index = []
    if not RAW_CONFIGS_DIR.exists():
        return _index

    for yml_path in RAW_CONFIGS_DIR.rglob("*.yml"):
        try:
            text = yml_path.read_text(encoding="utf-8", errors="ignore")
            config = yaml.safe_load(text)
            if not isinstance(config, dict):
                continue

            entry = {
                "path": str(yml_path.relative_to(PROJECT_ROOT)),
                "source": yml_path.parent.name,
                "text": text[:2000],
                "fields": {
                    "type": config.get("type"),
                    "tools": _extract_tools(config),
                    "model_provider": _nested_get(config, "model", "provider"),
                    "has_rag": bool(config.get("rag", {}).get("enabled")),
                    "has_mcp": bool(config.get("mcp")),
                    "has_guard": bool(config.get("guard")),
                },
                "keywords": _extract_keywords(text),
            }
            _index.append(entry)
        except Exception:
            continue

    # 同样扫描 .yaml 文件
    for yml_path in RAW_CONFIGS_DIR.rglob("*.yaml"):
        try:
            text = yml_path.read_text(encoding="utf-8", errors="ignore")
            config = yaml.safe_load(text)
            if not isinstance(config, dict):
                continue
            entry = {
                "path": str(yml_path.relative_to(PROJECT_ROOT)),
                "source": yml_path.parent.name,
                "text": text[:2000],
                "fields": {
                    "type": config.get("type"),
                    "tools": _extract_tools(config),
                    "model_provider": _nested_get(config, "model", "provider"),
                    "has_rag": bool(config.get("rag", {}).get("enabled")),
                    "has_mcp": bool(config.get("mcp")),
                    "has_guard": bool(config.get("guard")),
                },
                "keywords": _extract_keywords(text),
            }
            _index.append(entry)
        except Exception:
            continue

    return _index


def _extract_tools(config: dict) -> List[str]:
    """从配置中提取工具列表。"""
    tools = []
    mcp = config.get("mcp", {})
    if isinstance(mcp, dict):
        tools.extend(mcp.get("localTools", []))
        for server, tool_list in mcp.get("onlineTool", {}).items():
            if isinstance(tool_list, list):
                tools.extend(tool_list)
    return tools


def _nested_get(d: dict, *keys):
    """安全的嵌套字典取值。"""
    for k in keys:
        if isinstance(d, dict):
            d = d.get(k)
        else:
            return None
    return d


def _extract_keywords(text: str) -> List[str]:
    """从配置文本中提取关键词（用于搜索匹配）。"""
    # 提取英文单词和中文词
    words = re.findall(r'[a-zA-Z_]{3,}', text.lower())
    # 去重，保留前 50 个
    seen = set()
    result = []
    for w in words:
        if w not in seen and w not in _STOP_WORDS:
            seen.add(w)
            result.append(w)
            if len(result) >= 50:
                break
    return result


_STOP_WORDS = {
    "the", "and", "for", "with", "that", "this", "from", "have", "has",
    "are", "was", "were", "been", "being", "will", "would", "could",
    "should", "may", "might", "must", "shall", "can", "not", "but",
    "true", "false", "null", "none", "string", "int", "float", "bool",
    "default", "type", "name", "value", "description",
}


# ── 公开接口 ──────────────────────────────────────────────

def search_configs(keywords: List[str], max_results: int = 5) -> List[Dict]:
    """
    按关键词搜索配置库。

    Args:
        keywords: 搜索关键词列表（英文）
        max_results: 最大返回数量

    Returns:
        匹配的配置列表，按相关性排序
    """
    index = _build_index()
    if not index:
        return []

    keywords_lower = [k.lower() for k in keywords]

    scored = []
    for entry in index:
        score = 0.0
        entry_keywords = set(entry["keywords"])
        entry_text = entry["text"].lower()

        for kw in keywords_lower:
            # 精确关键词匹配
            if kw in entry_keywords:
                score += 2.0
            # 文本子串匹配
            elif kw in entry_text:
                score += 1.0
            # 部分匹配（关键词包含搜索词）
            elif any(kw in ek for ek in entry_keywords):
                score += 0.5

        # 字段加分
        fields = entry["fields"]
        for kw in keywords_lower:
            if fields.get("type") and kw in str(fields["type"]).lower():
                score += 1.5
            if kw in [t.lower() for t in fields.get("tools", [])]:
                score += 1.5

        if score > 0:
            scored.append((score, entry))

    scored.sort(key=lambda x: x[0], reverse=True)

    results = []
    for score, entry in scored[:max_results]:
        results.append({
            "path": entry["path"],
            "source": entry["source"],
            "relevance": round(min(score / (len(keywords_lower) * 3), 1.0), 2),
            "fields": entry["fields"],
            "preview": entry["text"][:500],
        })

    return results


def read_config(config_path: str) -> Optional[Dict]:
    """
    读取指定路径的配置文件全文。

    Args:
        config_path: 相对于 PROJECT_ROOT 的路径

    Returns:
        解析后的配置字典，或 None
    """
    full_path = PROJECT_ROOT / config_path
    if not full_path.exists():
        return None

    try:
        text = full_path.read_text(encoding="utf-8")
        config = yaml.safe_load(text)
        return {
            "path": config_path,
            "raw_text": text,
            "parsed": config if isinstance(config, dict) else {},
        }
    except Exception as e:
        return {"path": config_path, "error": str(e)}