让搜索 Agent 自己改进：从结果评估到迭代实验的工程闭环

用户问题 -> LLM 生成查询 -> 搜索服务 -> LLM 汇总答案

任务集
  -> 当前搜索策略
  -> 执行搜索
  -> 评估结果
  -> 定位失败类型
  -> 生成候选策略
  -> 对照实验
  -> 保留更优版本

总分 = 0.5 × 证据召回 + 0.3 × 答案可信度 + 0.2 × 来源覆盖
       - 延迟惩罚 - Token 成本惩罚 - 无效搜索惩罚

from __future__ import annotations

import re
from dataclasses import dataclass, replace

DOCS = [
    ('k8s', 'Diagnose a pod that keeps restarting with crash loop logs'),
    ('auth', 'Rotate an API credential safely without service downtime'),
    ('db', 'Speed up slow SQL queries by adding and validating indexes'),
    ('cache', 'Use a cache to reduce repeated network requests'),
]

TRAIN_TASKS = [
    ('pod keeps restarting', 'k8s'),
    ('renew service token', 'auth'),
]

HOLDOUT_TASKS = [
    ('reduce database query latency', 'db'),
]

STOP_WORDS = {'a', 'an', 'the', 'how', 'to', 'can', 'i'}
SYNONYMS = {
    'renew': ['rotate'],
    'service': ['api'],
    'token': ['credential'],
    'reduce': ['speed'],
    'database': ['sql'],
    'latency': ['slow'],
}


def tokens(text: str) -> list[str]:
    return re.findall(r'[a-z0-9]+', text.lower())


@dataclass(frozen=True)
class Strategy:
    drop_stop_words: bool = False
    expand_synonyms: bool = False
    top_k: int = 1


def rewrite(query: str, strategy: Strategy) -> list[str]:
    result = tokens(query)

    if strategy.drop_stop_words:
        result = [word for word in result if word not in STOP_WORDS]

    if strategy.expand_synonyms:
        expanded = list(result)
        for word in result:
            expanded.extend(SYNONYMS.get(word, []))
        result = expanded

    return result


def search(query: str, strategy: Strategy) -> list[str]:
    query_words = set(rewrite(query, strategy))
    ranked = []

    for doc_id, text in DOCS:
        doc_words = set(tokens(text))
        overlap = len(query_words & doc_words)
        ranked.append((overlap, doc_id))

    ranked.sort(key=lambda item: (-item[0], item[1]))
    return [doc_id for _, doc_id in ranked[:strategy.top_k]]


def evaluate(tasks, strategy: Strategy):
    failures = []
    hits = 0

    for query, expected_doc in tasks:
        results = search(query, strategy)
        if expected_doc in results:
            hits += 1
        else:
            failures.append({
                'query': query,
                'expected': expected_doc,
                'actual': results,
            })

    return hits / len(tasks), failures


def mutate(strategy: Strategy) -> set[Strategy]:
    candidates = {
        strategy,
        replace(strategy, drop_stop_words=not strategy.drop_stop_words),
        replace(strategy, expand_synonyms=not strategy.expand_synonyms),
        replace(strategy, top_k=min(3, strategy.top_k + 1)),
    }
    return candidates


def complexity_penalty(strategy: Strategy) -> tuple[int, int, int]:
    # 分数相同时，优先选择更少结果、更少改写的简单策略。
    return (
        strategy.top_k,
        int(strategy.expand_synonyms),
        int(strategy.drop_stop_words),
    )


def main() -> None:
    current = Strategy()

    for round_no in range(1, 4):
        candidates = mutate(current)
        scored = []

        for candidate in candidates:
            score, failures = evaluate(TRAIN_TASKS, candidate)
            scored.append((score, candidate, failures))

        scored.sort(
            key=lambda item: (-item[0], complexity_penalty(item[1]))
        )
        best_score, current, failures = scored[0]

        print(f'round={round_no} train_score={best_score:.2f}')
        print(f'strategy={current}')
        print(f'failures={failures}\n')

    holdout_score, holdout_failures = evaluate(HOLDOUT_TASKS, current)
    print(f'holdout_score={holdout_score:.2f}')
    print(f'holdout_failures={holdout_failures}')


if __name__ == '__main__':
    main()

python self_iter_search.py

{
  "failure_type": "vocabulary_mismatch",
  "changes": {
    "expand_synonyms": true,
    "top_k": 2
  },
  "reason": "The query uses token while the target document uses credential."
}