第 20 章 · 工具调用进阶：契约、校验与边界

# ❌ 只说功能 —— 模型不知道该不该用
{
    "name": "run_qc_checks",
    "description": "运行数据质量检查",
}

# ✅ 说清用法与前置条件 —— 模型知道何时用、用之前要做什么
{
    "name": "run_qc_checks",
    "description": (
        "对某个数据集运行标准化质量检查，返回按严重度分级的问题清单。"
        "可检查项：required_vars（必填变量是否齐全）、missing_rate（缺失率）、"
        "key_unique（主键唯一性）、date_pairs（日期先后关系）、codelist（受控术语）。"
        "在调用本工具前，应先调用 describe_dataset 确认变量名。"
        "本工具只读，不修改任何数据。"
    ),
}

# ❌ 自由文本：模型会写出 "requiredVars"、"必填变量"、"required_vars " 各种变体
{"checks": {"type": "array", "items": {"type": "string"}}}

# ✅ 枚举：模型只能在有限集合里选
{
    "checks": {
        "type": "array",
        "items": {
            "type": "string",
            "enum": ["required_vars", "missing_rate", "key_unique",
                     "date_pairs", "codelist"],
        },
        "description": "要执行的检查项；不传则执行全部",
    }
}

{
    "dataset": {
        "type": "string",
        "enum": sorted(ALLOWED_DATASETS),      # dm / adsl / adae / ...
        "description": "数据集名（小写，不含扩展名）",
    }
}

"parameters": {
    "type": "object",
    "properties": {...},
    "required": ["dataset"],
    "additionalProperties": False,     # ← 拦住模型自创的参数
}

# ❌ 三个名字长得太像，模型经常选错
get_data()   /  get_info()   /  get_stats()

# ✅ 名字自带语义，无需看描述也能猜到区别
describe_dataset()      # 结构：行数、列数、缺失、示例值
summarize_by_group()    # 分组描述统计：n / Mean (SD) / Median
frequency()             # 频数分布：n (%)

name="sas_round_rate"    # 明确告诉模型：这是 SAS 口径的舍入，不是 Python 的

# 你以为会收到：
{"dataset": "adsl", "subjects": ["01-701-1015", "01-701-1023"]}

# 实际可能收到：
{"dataset": "ADSL"}                                  # 大小写
{"dataset": "adsl.csv"}                              # 带扩展名
{"dataset": "adsl "}                                 # 带空格
{"subjects": "01-701-1015"}                          # 单值当列表
{"subjects": "01-701-1015,01-701-1023"}              # 逗号串
{"subjects": ["01-701-1015", "01-701-1015"]}         # 重复
{"subjects": "全部"}                                  # 自然语言
{"dataset": "../../../../etc/passwd"}                # ★ 恶意的路径穿越

def _norm_dataset(raw: Any) -> str:
    """把 LLM 给的各种形态归一化成一个干净的数据集名。"""
    s = str(raw).strip().lower()
    if s.endswith(".csv"):
        s = s[:-4]
    if s.endswith(".xpt"):
        s = s[:-4]
    return s

def _norm_subjects(raw: Any) -> list[str]:
    """单值 / 逗号串 / 列表 都接受，去重保序。"""
    if raw is None:
        return []
    if isinstance(raw, str):
        parts = [p.strip() for p in raw.replace(";", ",").split(",")]
    elif isinstance(raw, (list, tuple, set)):
        parts = [str(p).strip() for p in raw]
    else:
        parts = [str(raw).strip()]
    out, seen = [], set()
    for p in parts:
        if p and p not in seen:
            seen.add(p)
            out.append(p)
    return out

# ❌ 危险：看似校验了目录，实际可以被绕过
def load_dataset(name: str):
    if "../" in name:
        raise ValueError("非法数据集名")
    path = DATA_DIR / f"{name}.csv"
    return pd.read_csv(path)

# ✅ 白名单 + 只映射不拼接
ALLOWED = {
    "dm":    "dm.csv",
    "adsl":  "adsl.csv",
    "adae":  "adae.csv",
    "ae":    "ae.csv",
    "vs":    "vs_bp.csv",
    "adlbc": "adlbc_shift.csv",
}

def _load(name: str) -> pd.DataFrame:
    key = _norm_dataset(name)
    if key not in ALLOWED:                       # 不在白名单 → 直接拒绝
        raise PermissionError(
            f"数据集 {name!r} 不在允许列表中。可用：{sorted(ALLOWED)}"
        )
    path = (DATA_DIR / ALLOWED[key]).resolve()   # 路径完全由白名单决定

    # 双保险：确认解析后的路径确实还在数据目录内
    if not str(path).startswith(str(DATA_DIR.resolve())):
        raise PermissionError("路径越界")

    return pd.read_csv(path, low_memory=False)

# ❌ 第 16 章的做法（教学版可以，工程版不行）
except Exception as e:
    return f"错误：{e}"

# 实际会返回这种东西给模型：
# "错误：'NoneType' object has no attribute 'upper'"
# 模型看到这条，唯一能做的就是瞎猜着改参数 —— 或者干脆放弃

# ✅ 结构化错误：类型 + 说明 + 怎么改 + 候选值
@dataclass
class ToolResult:
    ok: bool
    data: Any = None
    error: str | None = None
    error_type: str | None = None       # validation/not_found/permission/transient/internal
    hint: str | None = None             # ★ 给模型的修复建议
    options: list[str] | None = None    # ★ 合法取值

    def to_llm_text(self) -> str:
        if self.ok:
            return _render(self.data)
        parts = [f"[{self.error_type}] {self.error}"]
        if self.options:
            parts.append(f"可选值：{self.options}")
        if self.hint:
            parts.append(f"提示：{self.hint}")
        return "\n".join(parts)

# ❌ 工具直接返回 DataFrame
return df.head(100)

# 后面某处 state.to_json() 时：
# TypeError: Object of type DataFrame is not JSON serializable

# ❌ 这段看起来成功了 —— 但它把 DataFrame 转成了一坨
json.dumps(df, default=str)     # → "   USUBJID  TRT01P  AGE\n0  01-701-1015 ..."

# ✅ 工具自己负责把结果转成"小而结构化"的形式
def summarize_by_group(dataset: str, group: str, var: str) -> ToolResult:
    ...
    table = []
    for g, sub in df.groupby(group, observed=True):
        s = pd.to_numeric(sub[var], errors="coerce")
        table.append({
            "组": str(g),
            "n": int(s.notna().sum()),
            "Mean (SD)": f"{s.mean():.2f} ({s.std(ddof=1):.2f})",
            "Median": float(s.median()),
            "Min, Max": f"{s.min():.1f}, {s.max():.1f}",
        })
    return ToolResult.ok({
        "数据集": dataset, "分组": group, "变量": var,
        "结果": table,                      # ← 小而结构化
        "缺失": int(s.isna().sum()),
    })

def _cap(rows: list, limit: int = 50) -> dict:
    """统一的结果长度控制：截断 + 摘要 + 明确告知被截断。"""
    if len(rows) <= limit:
        return {"行": rows, "总数": len(rows)}
    return {
        "行": rows[:limit],
        "总数": len(rows),
        "已截断": True,
        "说明": f"仅返回前 {limit} 行，共 {len(rows)} 行；"
                f"如需完整结果，请按条件缩小范围或使用汇总工具。",
    }

@dataclass
class ToolSpec:
    ...
    side_effect: str = "read"      # read / write / irreversible
    idempotent: bool = True        # 重复执行是否安全

def append_audit(event: dict, idem_key: str | None = None) -> ToolResult:
    """追加审计事件。带 idem_key 时，同一 key 只写入一次。"""
    if idem_key and _seen(idem_key):
        return ToolResult.ok({"重复": True, "说明": "该事件已记录，跳过"})
    ...

# tests/test_agent_tools.py
import pytest
from clinic.agent_tools import build_registry, ALLOWED

@pytest.fixture(scope="module")
def reg():
    return build_registry()

# ---- 1) Schema 自身要合法 ----
def test_every_tool_has_valid_schema(reg):
    for spec in reg.specs():
        assert spec.name and spec.description, f"{spec.name} 缺名字或描述"
        assert spec.parameters["type"] == "object"
        assert spec.parameters.get("additionalProperties") is False
        for p in spec.parameters.get("properties", {}).values():
            assert "description" in p, "每个参数都要有 description"

# ---- 2) 参数归一化的边界 ----
@pytest.mark.parametrize("raw,expect", [
    ("ADSL", "adsl"), ("adsl.csv", "adsl"), (" adsl ", "adsl"),
])
def test_dataset_normalization(reg, raw, expect):
    assert reg.execute("describe_dataset", {"dataset": raw}).ok

@pytest.mark.parametrize("raw", ["../etc/passwd", "adls", "", "全部", None])
def test_rejects_bad_dataset(reg, raw):
    res = reg.execute("describe_dataset", {"dataset": raw})
    assert not res.ok
    assert res.error_type in {"not_found", "validation", "permission"}

# ---- 3) 结果必须可 JSON 序列化 ----
def test_results_are_serializable(reg):
    import json
    for name in ("describe_dataset", "run_qc_checks", "frequency"):
        res = reg.execute(name, {"dataset": "adsl"})
        json.dumps(res.data, ensure_ascii=False)      # 不抛异常即通过

# ---- 4) 错误必须结构化 ----
def test_errors_are_actionable(reg):
    res = reg.execute("describe_dataset", {"dataset": "nope"})
    assert res.error_type and res.options            # 有类型、有候选值