Dify 中级实验（10）：知识库深度调优——如何科学评估检索质量？

def main() -> dict:

    test_cases = [

        {"query": "如何重置密码", "expected_keywords": ["密码重置", "忘记密码", "重置步骤"], "expected_doc": "用户管理_密码重置"},

        {"query": "API 调用频率限制", "expected_keywords": ["限流", "Rate Limit", "API配额"], "expected_doc": "API文档_限流策略"},

        {"query": "错误码 E1001 是什么", "expected_keywords": ["E1001", "错误码", "排查"], "expected_doc": "排障手册_错误码"},

        # ... 共 10 组

    ]

    return {"test_cases": test_cases}

- data:

    dataset_ids:

    - 459c4981-6b76-43e7-b44f-c430f33ef035   # 你的知识库 ID

    isInIteration: true

    iteration_id: iter_test

    output_retrieval_result: true            # 必须 true，否则 result 为空

    parentId: iter_test

    query_attachment_selector: []

    query_variable_selector: [cd_split, query]

    retrieval_mode: single

    score_threshold: 0.0

    title: 技术文档库检索

    top_k: 3

    type: knowledge-retrieval

  id: kb_retrieve

def main(query: str, expected_keywords: str, expected_doc: str, retrieved: list) -> dict:

    import json

    retrieved = retrieved if isinstance(retrieved, list) else []

    texts = [str(r.get("content", "")) for r in retrieved if isinstance(r, dict)]

    all_text = " ".join(texts).lower()

    try:

        kws = json.loads(expected_keywords or "[]")

    except Exception:

        kws = []

    kws = [str(k).lower() for k in kws if k]

    hits = sum(1 for k in kws if k in all_text)

    keyword_rate = round(hits / len(kws), 2) if kws else 0.0

    top1_title = str(retrieved[0].get("title", "")) if retrieved and isinstance(retrieved[0], dict) else ""

    top1_match = bool(top1_title) and str(expected_doc or "").lower() in top1_title.lower()

    score = round(keyword_rate * 0.5 + (1.0 if top1_match else 0.0) * 0.5, 2)

    return {"eval_text": json.dumps({...}, ensure_ascii=False)}   # 序列化成 string 供迭代收集

def main(results: list) -> dict:

    import json

    total = 0; score_sum = 0.0; kw_sum = 0.0; top1_ok = 0

    for r in results:

        try:

            d = json.loads(r) if isinstance(r, str) else {}

        except Exception:

            d = {}

        total += 1

        score_sum += float(d.get("score", 0) or 0)

        kw_sum += float(d.get("keyword_hit_rate", 0) or 0)

        if d.get("top1_match"):

            top1_ok += 1

    return {

        "total_cases": total,

        "avg_score": round(score_sum / total, 2) if total else 0.0,

        "avg_keyword_rate": round(kw_sum / total, 2) if total else 0.0,

        "top1_matched": top1_ok,

        "summary_text": "检索质量评估：共 {} 组用例，综合评分 {}，关键词命中率均值 {}，Top1 文档匹配 {} 组".format(

            total, round(score_sum / total, 2) if total else 0.0,

            round(kw_sum / total, 2) if total else 0.0, top1_ok),

    }