用 12 步造一个会拒绝自己幻觉的研究 agent:搜索循环、来源打分、事实闸

Yaqin Hei··14分钟阅读
中文EN
用 12 步造一个会拒绝自己幻觉的研究 agent:搜索循环、来源打分、事实闸

本文原题《Build a research agent in 12 steps: search loops, source ranking, fact gates》,作者 @de1lymoon——一位长期在自己 newsletter 上写 agent 工程的独立作者(这一点在原文里他自己点破:「Substack 是聚合器,我自己的也是」)。全文没有框架、没有黑话,只有一个 while 循环、一个打分函数、三道闸,却把「agent 为什么会一本正经地编造引文」这件事拆得干干净净。中文翻译、重排、5 张配图本地化:Yaqin Hei(非原创,已保留原文出处)。

译者的话:为什么翻这篇

国内现在用 AI 做调研、做 agent,很多团队还停在一个朴素的信仰上:模型不够强,读得不够多——于是拼命喂更多来源、追更长的上下文,指望「读得快、读得多」能自动换来「读得对」。

这篇最扎心的一句是:我见过的每一个失败的 agent,都不是因为它笨,而是因为整个循环里没有任何一环被允许说「不」。 它检索、它阅读、它总结,交给你一段通顺的话,话里带一句引文,引文不存在——因为从头到尾,没有一个环节的职责是「核对」。你以为瓶颈是智力,其实瓶颈是没人负责拒绝

对我最大的改变,是它换掉了我评判一个 agent 好不好的第一个问题。以前我盯着「输出漂不漂亮」,现在我先找那道闸:这个循环里,哪一步被允许说「不」? 搜索循环的出口是「信息饱和了就停」,排序循环的出口是「装不进预算就丢」,验证循环的出口是「引文不在原文里就毙掉」。三道闸,每道都在一个循环的末尾——没有闸的循环,只是一个 agent 在用高速反复给自己点赞。

如果你手里有任何一个会产出引文 / 数据 / 结论的 agent,这篇末尾那道引用闸(citation gate)你今天就能抄进去:11 行代码,不做语义匹配,只问「这个带引号的字符串,到底在不在原文里」。它拦下的,恰好是那种「读起来最像真的、最能骗过你」的编造。


以下是正文

我见过的每一个失败的 agent,都不是因为它笨。它失败,是因为整个循环里没有任何一环被允许说「」。

这就是全部的问题。一个研究 agent 检索、阅读、总结,交给你一段话。这段话很通顺。这段话里有一句引文。这句引文不存在。而循环里,从来没有人被分配到「核对」这份工作。

我不再试图把模型调得更谨慎。我开始搭那个能当场抓住它的循环。

下面是一份 12 步路线图,造一个拒绝把自己的幻觉当结论发出去的研究 agent。不用框架。一个 while 循环、一个打分函数、三道闸。

开始吧……

三个循环,三道闸:搜索循环产出来源,排序循环把来源压成预算大小的语料,事实循环把草稿拆成三份清单——未溯源的那份回流成新问题,循环重启。

三个循环,三道闸。每道闸都蹲在一个循环的末尾;末端那份「未溯源」清单回流成起点的新问题,整套东西更锋利地重启一遍。

循环一 · 搜索循环

01 · 写一个论断,不是一个话题

大多数研究 agent 接到的是一个话题。「研究一下 AI agent 的可靠性。」这不是问题,是一个方向;而一个被指了个方向的 agent,会一直走到把 token 烧光为止。

一个可研究的问题,是你有可能被证明是错的那种。先写下论断(claim),再写下那个能杀死它的东西——它的可证伪条件(falsifier)。

from dataclasses import dataclass
import re

@dataclass
class Claim:
    text: str
    falsifier: str

    def is_researchable(self) -> bool:
        vague = {"best", "good", "bad", "better", "important", "significant"}
        words = set(re.findall(r"[a-z]+", self.text.lower()))
        return bool(self.falsifier) and not (words & vague)

那个 vague(含糊词)集合是把钝刀子,但它管用。如果你的问题里带着「best(最好)」,你不是在做研究,你是在逛街比价。

c = Claim(
    text="Do LLM agents degrade on long horizon tasks?",
    falsifier="A benchmark showing flat success rate past step 7",
)
assert c.is_researchable()

在写下 agent 的第一行之前,先跑这个过滤器:如果没有任何结果能改变答案,那这个 agent 根本无事可做。

02 · 把一个论断炸成四个查询

一个问题,四个角度。天真的 agent 只搜它预期会找到的东西——而搜索引擎极其擅长把你预期的东西喂给你。

一个论断炸成四路查询:确认、佐证、反驳、溯源;真正干活的是「反驳」那一路。

一个论断炸成四路查询。真正干活的是第三路——反驳。少了它,你的 agent 会找来十五篇彼此附和的文章。
def expand_queries(claim: Claim, n: int = 4) -> list[str]:
    core = claim.text.rstrip("?.").lower()
    return [
        core,                                                    # 确认
        f"{core} evidence",                                      # 佐证
        f"{core} criticism OR disputed OR retracted",            # 反驳
        f"{core} original study OR primary source OR dataset",   # 溯源
    ][:n]

四行,第三行干了大部分活。没有那条反驳查询,你的 agent 会找来十五篇彼此附和的文章,它们全都引用同一篇预印本,然后向你报告一个根本不存在的共识

一个只搜确认信息的 agent 不是在做研究。它是在给你早就有的结论描边。

03 · 按饱和停,不按数量停

这是第一道闸。

几乎每个 agent 都在拿到 N 个结果后停下。N 是某人某次随手敲进去的一个魔法数字。正确的停止条件是信息层面的:当新结果不再带来新信息时,停。

from dataclasses import field

@dataclass
class Saturation:
    seen: set[str] = field(default_factory=set)
    novelty_window: list[float] = field(default_factory=list)
    threshold: float = 0.15
    window: int = 2

    def observe(self, batch) -> float:
        if not batch:
            return 0.0
        fresh = sum(1 for r in batch if r.fingerprint() not in self.seen)
        for r in batch:
            self.seen.add(r.fingerprint())
        novelty = fresh / len(batch)
        self.novelty_window.append(novelty)
        return novelty

    def exhausted(self) -> bool:
        if len(self.novelty_window) < self.window:
            return False
        return all(n < self.threshold for n in self.novelty_window[-self.window:])

连续两批新鲜度低于 15%,搜索就结束。有时它在第二个查询就触发。有时它永远不触发,于是你意识到这个问题比你想的更大——而这件事本身就是一个发现

这道闸不是为了省 token。它在那儿,是为了让 agent 分得清「广泛地读」和「原地打转地读」的区别。

04 · 先归一化,再去重

五个结果,一个事实。其中三个是同一份新闻稿,换了三个标题。

剥掉追踪参数、去掉 www、把结尾的斜杠归一化、做哈希。然后靠标题抓住那些近似重复

from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
from difflib import SequenceMatcher
import hashlib, re

TRACKING = re.compile(r"^(utm_|fbclid|gclid|ref|mc_)")

def canonical(url: str) -> str:
    parts = urlsplit(url.strip())
    host = parts.netloc.lower().removeprefix("www.")
    query = [(k, v) for k, v in parse_qsl(parts.query) if not TRACKING.match(k)]
    path = parts.path.rstrip("/") or "/"
    return urlunsplit(("https", host, path, urlencode(sorted(query)), ""))

def dedupe(results, near: float = 0.88):
    kept, seen = [], set()
    for r in results:
        fp = r.fingerprint()
        if fp in seen:
            continue
        if any(SequenceMatcher(None, r.title.lower(), k.title.lower()).ratio() > near
               for k in kept):
            continue
        seen.add(fp)
        kept.append(r)
    return kept

这是全文最不起眼的一个函数,而它对你输出的改变,超过你这辈子写过的任何一句 prompt。

转载不等于佐证。 十家媒体跑同一条通稿,那是一个来源戴了十顶帽子。

循环二 · 来源排序

05 · 分类:一手源、二手源、聚合器

在给一个来源打分之前,你得先知道它是个什么东西。三档,而且档与档之间的界线并不微妙。

三档来源:一手源(arxiv / 政府 / 期刊)、二手源、聚合器(Substack / Reddit / 社交媒体)。这张清单是一条策略,不是一条事实。

三档来源。这张清单是有立场的——因为它是一条策略,不是一条事实,所以它该待在你能跟它吵架的代码里。
PRIMARY = ("arxiv.org", "nature.com", "sec.gov", "who.int",
           "nih.gov", "europa.eu", "bls.gov")
AGGREGATOR = ("medium.com", "substack.com", "reddit.com",
              "news.ycombinator.com", "x.com", "linkedin.com")

def classify(r) -> str:
    d = r.domain()
    if any(d.endswith(p) for p in PRIMARY) or d.endswith(".gov"):
        return "primary"
    if any(d.endswith(a) for a in AGGREGATOR):
        return "aggregator"
    return "secondary"

是的,这张清单是有立场的。是的,你的清单会长得不一样。这正是重点:清单是一条策略,不是一条事实,所以它该待在你的代码里,一个你能跟它争论的地方。

注意这里 Substack 被归为聚合器——包括我自己的。一个把我的 newsletter 当一手源的 agent,是一个我连自己写的东西都不敢托付给它的 agent。

06 · 打分:档位、时效、数据、独立性

只看档位是个糟糕的排序器。一篇三年前、里面一个数字都没有的一手源,会输给一篇最近的、公布了样本量的二手源。

四个维度,一个数字。

四个打分维度合成一个数字:档位 0.45、时效 0.25、有无数据 0.20、独立性 0.10。独立性是那个不动声色的杀手。

四个维度,一个数字。独立性那一项是不动声色的杀手——一个域名返回六个结果,不该拿到六票。
from datetime import datetime, timezone

TIER_WEIGHT = {"primary": 1.0, "secondary": 0.6, "aggregator": 0.25}

def recency(published: str | None, half_life_days: float = 540.0) -> float:
    if not published:
        return 0.5
    try:
        then = datetime.fromisoformat(published).replace(tzinfo=timezone.utc)
    except ValueError:
        return 0.5
    age = (datetime.now(timezone.utc) - then).days
    return 0.5 ** (age / half_life_days)

def has_data(r) -> float:
    signal = re.search(r"\d{2,}|\bdataset\b|\bsample\b|\bn\s*=\s*\d+", r.snippet, re.I)
    return 1.0 if signal else 0.0

def score(r, domain_counts: dict[str, int]) -> float:
    tier = TIER_WEIGHT[classify(r)]
    independence = 1.0 / domain_counts[r.domain()]
    return round(
        0.45 * tier
        + 0.25 * recency(r.published)
        + 0.20 * has_data(r)
        + 0.10 * independence,
        4,
    )

独立性这一项是那个不动声色的杀手。一个域名返回六个结果,不该拿到六票。除以它自己出现的频次,一个大嗓门的站点就会自己安静下来。

缺失发布日期算 0.5 分,不是 0 分。一个没标日期的来源是不确定,不是没价值。把不确定,编码成不确定。

07 · 剪裁到上下文预算

第二道闸。

Top-k 是个错误的抽象。模型并没有一个「装五个来源」的槽位,它有的是一个以字符数计量的窗口(context window);一个又长又低分的来源,会吃掉一个又短又高分的来源本该用的空间。

def fit_budget(ranked, budget_chars: int):
    out, used = [], 0
    for r, _ in ranked:
        cost = len(r.snippet) + len(r.title) + 80   # 80 = 引用开销
        if used + cost > budget_chars:
            continue
        out.append(r)
        used += cost
    return out

注意这里用的是 continue 而不是 break。一个装不下的来源是被跳过,不是到此为止——下一个也许就小得刚好塞得进。

按质量排序,按成本准入。 两个不同的问题,两个不同的函数。

08 · 记录冲突,永远不要把它们平均掉

这是大多数研究 agent 真正犯罪的地方。它们找到两个彼此矛盾的来源,然后产出一句折中的话。这句话没有任何人说过。它是一个用真材料拼出来的幻觉。

NEGATION = re.compile(
    r"\b(no|not|never|fails?|disput|refut|contradict|retract)\w*", re.I
)

def find_conflicts(results):
    pos = [r for r in results if not NEGATION.search(r.snippet)]
    neg = [r for r in results if NEGATION.search(r.snippet)]
    return [(p.domain(), n.domain()) for p in pos[:2] for n in neg[:2]]

粗糙的关键词匹配。它把矛盾的那一对摆到台面上,然后由你、或第二次模型调用,来判断这个分歧意味着什么。它永远不做的,是悄悄把分歧抹平。

来源之间的分歧,是你的 agent 能找到的最有价值的东西。一个把它藏起来的 agent,毁掉的恰恰是唯一值得拥有的那点信息。

循环三 · 事实闸

09 · 把草稿原子化成一句句论断

你没法验证一个段落。你能验证一个句子。

SENT = re.compile(r"(?<=[.!?])\s+")

def atomize(draft: str) -> list[str]:
    return [s.strip() for s in SENT.split(draft.strip()) if len(s.strip()) > 15]

两行。这不是系统里精巧的那一部分,却是让后面每一步成为可能的那一步。验证,只能在小到足以判真伪的单元上进行。

10 · 把每句论断绑定到一个来源

每个句子都拿到一个来源,或者拿到一个标签。没有第三种选择,也没有任何「姑且信它」的余地。

STOP = set("the a an of in on and or to is are was were for with that this it as by".split())

def keywords(text: str) -> set[str]:
    return {w for w in re.findall(r"[a-z0-9]{3,}", text.lower()) if w not in STOP}

def bind(statement: str, corpus, floor: float = 0.30):
    kw = keywords(statement)
    if not kw:
        return None
    best, best_overlap = None, 0.0
    for r in corpus:
        overlap = len(kw & keywords(r.snippet)) / len(kw)
        if overlap > best_overlap:
            best, best_overlap = r, overlap
    return best if best_overlap >= floor else None

0.30 那个下限是个旋钮。调高,真话会被标成「无来源」;调低,agent 会兴高采烈地把「天是蓝的」绑定到一篇讲半导体的论文上。

在你自己的草稿上,调一次,然后把那个数字记下来。

11 · 拿引文去比对原文

第三道闸,也是我最在意的一道。

一个模型编造事实的频率,其实没有它编造措辞的频率高。它读三篇讲某位研究者的文章,吸收了大意,然后产出一句带引号的话——没有任何人说过这句话。事实都对,归属是虚构的。

引用闸:把带引号的字符串拿去和原文做精确子串匹配。不模糊、不语义。字符串不在原文里,这句话就不发。

引用闸——精确子串匹配,不模糊、不语义。字符串不在原文里,这句话就不发。这道闸拦下的,是那种「事实都对、只有归属是编的」的句子。
QUOTED = re.compile(r'"([^"]{10,})"')

def citation_gate(statement: str, source) -> tuple[bool, str]:
    quotes = QUOTED.findall(statement)
    if not quotes:
        return True, "ok"
    if source is None:
        return False, "quote with no source"
    for q in quotes:
        if q.lower() not in source.snippet.lower():
            return False, f"quote not found in {source.domain()}"
    return True, "ok"

精确子串匹配。不模糊,不语义。如果那串字符不在来源里,这个句子就不发货。

12 · 把「未核实」清单也发出去

最后一步,才是把这条流水线变成一个循环的那一步。

不要丢掉被标记的句子。把它们收集起来。没能被核实的东西,是一张「你还不知道什么」的地图——它是下一轮的输入。

@dataclass
class Verdict:
    statement: str
    status: str          # sourced | unsourced | rejected
    source: str | None
    note: str

def verify(draft: str, corpus) -> list[Verdict]:
    out = []
    for s in atomize(draft):
        src = bind(s, corpus)
        ok, note = citation_gate(s, src)
        if not ok:
            out.append(Verdict(s, "rejected", src.domain() if src else None, note))
        elif src is None:
            out.append(Verdict(s, "unsourced", None, "no source above floor"))
        else:
            out.append(Verdict(s, "sourced", src.domain(), "ok"))
    return out

三种状态,不是两种。rejected(已否决) 是编造,永不发货。unsourced(未溯源) 也许是真的,只有带着标签才发。sourced(已溯源) 挣到了它的位置。

unsourced 那份清单喂回第 1 步,当作新的论断。这就是那第五拍。这就是那个循环。

三种判决:rejected 永不发货、unsourced 带标签才发、sourced 挣到位置;未溯源的那份回流成第 1 步的新论断。

三种判决,不是两种。「未溯源」那份不是垃圾,是回流到第 1 步的新问题——这就是那个闭环。

这些闸到底拦下了什么

我拿一个故意设的陷阱跑了一遍:一份草稿里,一个有扎实来源的统计数字、一个有争议的结论、一个没有任何来源支持的自信泛化、以及一句我自己编的引文。

看最后两行。

大多数团队从来不测这个。这句话,正是每篇 AI 文章开头那个统计数字的标准长相。「十有八九」。「90% 的团队」。它听起来像数据。它其实是一种情绪。FLAG(标记)。

「一位研究者称之为『agent 循环的静默崩塌』。」我在跑脚本前三十秒现编的这句,而这道闸抓住了它——它对这个世界一无所知,它只是问了一句:这串字符,在原文里吗?STOP(拦截)。

这两句本来都会被发出去。而它们没有一个是智力的失败。

闭合这个循环

十二步,三个循环,三道闸。

搜索循环产出来源。排序循环把来源变成预算大小的语料。事实循环把草稿变成三份清单。而第 12 步产出的那份「未溯源」清单,变成第 1 步里的论断——整套东西从那里更锋利地重新开始。

注意这些闸蹲在哪儿:每一个循环的末尾。饱和终结搜索,预算终结排序,引用终结验证。一个没有闸的循环,只是一个 agent 在用高速反复认同自己。

过去两年,AI 研究工具的卖点一直是速度。一分钟读一百个来源。一秒钟总结一篇论文。这个阶段正在结束——因为事实证明,一个读了一百个来源、却说不清自己在引用哪一个的 agent,根本没帮你省下任何工作。它只是把工作挪到了下游,挪给了你,而且恰好在你放松警惕的那一刻。

优势不在阅读。优势在拒绝

挑一道你还没在跑的闸。大概率是引用闸,因为它只要十一行,却拦得住那件真会伤到你的事。今天就把它发上线。然后是下一道。

译自 @de1lymoon《Build a research agent in 12 steps: search loops, source ranking, fact gates》。中文翻译、重排、配图本地化:Yaqin Hei(非原创,已保留原文出处)。

Subscribe for updates

Get the latest AI engineering posts delivered to your inbox.

评论