12 步搭建研究智能体：搜索循环、来源排序、事实把关

用 12 步构建一个研究代理：搜索回路、来源排序、事实闸门 2026 年 7 月 13 日 · 12 分钟阅读 · 查看原文 ↗ AI LinkedIn 我见过的每一个失败的代理，都不是因为它笨。它失败，是因为回路里没有任何东西被允许说「不」。 这就是全部问题。一个研究代理搜索、阅读、总结，然后递给你一段话。这段话很流畅。这段话里有一句引用。那句引用并不存在。回路里从没有哪个人被分派去检查。 我不再试图让模型更小心。我开始构建那个能抓住它的回路。 这是一份 12 步路线图，通往一个拒绝发布自己幻觉的研究代理。不用框架。一个 while 循环、一个打分函数，和三道闸门。 开始吧…… 回路 1 —— 搜索回路 - 写一个主张，而不是一个主题 大多数研究代理被交给一个主题。「研究 AI 代理的可靠性。」那不是一个问题，而是一个方向，而一个被指向某个方向的代理会一直走，直到用完 token。 一个可研究的问题，是一个你可能犯错的问题。写下主张，然后写下那个能杀死它的东西。 那个模糊词集合是一件钝器，而且它管用。如果你的问题里包含「最好」，你不是在研究，你是在购物。 在你写代理的任何一行之前，先跑这个过滤器：如果没有任何结果能改变答案，那这个代理就无事可做。 - 把一个主张扩成四个查询 一个问题，四个角度。天真的代理会搜索它预期能找到的东西，而搜索引擎极其擅长给你你预期的东西。 四行，而第三行干了绝大部分活。没有那个反驳查询，你的代理会找到十五篇互相认同的文章，它们全都引用同一篇预印本，然后报告一个并不存在的共识。 一个只搜索确认的代理，不是在研究。它是在装点一个你早就有的结论。 - 在饱和时停，而不是在数量上停 这是第一道闸门。 几乎每个代理都在 N 个结果之后停。N 是一个有人随手敲进去的魔法数字。正确的停止条件是信息性的：当新结果不再携带新信息时停。 连续两批的新颖度低于 15%，搜索就结束。有时它在第二个查询就触发。有时它永远不触发，而你因此得知这个问题比你想的更大——这本身就是一个发现。 这道闸门不是为了省 token。它是为了让代理知道「广泛阅读」和「原地打转」之间的区别。 - 先规范化，再去重 五个结果，一个事实。其中三个是同一篇新闻稿，标题不同。 剥掉跟踪参数、去掉 www、规范化结尾斜杠、做哈希。然后用标题抓近似重复。 这是全文最不光彩的一个函数，但它对你的输出的改变，超过你将来写下的任何提示词。 转载不是佐证。十家媒体跑同一条通讯社稿子，是一个来源戴着十顶帽子。 回路 2 —— 来源排序 - 分类：一手、二手、聚合 在你给一个来源打分之前，你得先知道它是哪种东西。三个层级，而且边界并不微妙。 是的，这份清单是有立场的。是的，你的清单会不一样。这正是重点：清单是一个政策，不是一个事实，所以它该待在你的代码里，好让你能跟它争辩。 注意，这里 Substack 是聚合器，我的也是。一个把我的通讯当成一手来源的代理，是一个我不会把自己的写作托付给它的代理。 - 给层级、时效、数据、独立性打分 光层级是一个糟糕的排序器。一篇三年前的、里面没有数字的一手来源，会输给一篇最近发表的、公布了样本量的二手来源。 四个轴，一个数字。 独立性那一项是隐藏的杀招。一个域名返回六个结果，不该拿六票。除以它自己的出现频率，一个吵闹的网站自己就会安静下来。 缺失发布日期得 0.5 分，而不是 0 分。一个无日期来源是不确定的，不是一文不值的。把不确定编码成不确定。 - 剪枝到一个上下文预算 第二道闸门。 Top-k 是错的抽象。模型没有一个放五个来源的槽位，它有一个按字符计量的窗口，而一篇又长又低排名的来源，会吃掉一篇又短又高排名的来源需要的空间。 注意这里是 continue 而不是 break。一个装不下的来源是被跳过，而不是终止。下一个可能就够小了。 按质量排序，按成本准入。两个不同的问题，两个不同的函数。 - 记录冲突，绝不平摊 这是大多数研究代理犯下它们真正罪过的地方。它们找到两个互相矛盾来源，然后产出一句各打五十大板的句子。没有人说过那句话。它是一具用真实零件拼出来的幻觉。 粗糙的关键词匹配。它把那一对浮出来，然后由你——或者第二次模型过一遍——决定那个分歧意味着什么。它永远不做的是悄悄把分歧消解掉。 来源之间的分歧，是你的代理能找到的最有价值的东西。一个把它藏起来的代理，已经摧毁了唯一值得拥有的信息。 回路 3 —— 事实闸门 - 把草稿原子化成主张 你没法核实一段话。你可以核实一个句子。 两行。这不是系统里复杂的那部分，而它是让后面每一步都成为可能的那个步骤。核实运作在足够小、能判真假的单位上。 - 把每个主张绑定到一个来源 每个句子要么拿到一个来源，要么拿到一个标签。没有第三种选项，也没有无罪的推定。 0.30 这个下限是一个旋钮。把它调高，真陈述会被标记为无来源。把它调低，代理会开心地把「天是蓝的」绑到一篇讲半导体的论文上。 在你自己的草稿上调一次，然后把数字记下来。 - 把引用对着正文把关 第三道闸门，也是我最在乎的一道。 一个模型编造事实，不像它编造措辞那么频繁。它读三篇关于某位研究者的文章，吸收大意，然后产出一个带引号的、从没人说过的句子。事实活了下来。归属是虚构的。 精确子串匹配。不是模糊的，不是语义的。如果那个字符串不在来源里，这个句子就不发。 - 交付未核实台账 最后一步，是把管线变成回路的那一步。 不要丢弃那些被标记的句子。收集它们。无法核实的东西，就是你「还不知道什么」的地图，也是下一次遍历的输入。 三种状态，不是两种。rejected 是编造，它永不发出。unsourced 可能是真的，它只在带标签的情况下发出。sourced 挣到了它的位置。 把 unsourced 清单作为新主张喂回第 1 步。那是第五拍。那才是回路。 这些闸门到底抓住了什么 我拿一个刻意的陷阱跑过它：一份草稿，包含一个来源良好的统计、一个被争议的发现、一个没有任何来源支持的自信概括，以及一句我编造的引用。 看最后那两行。 大多数团队根本不量这个。这句话正是每一篇 AI 文章开头那种统计的形状。十之有九。百分之九十的团队。听起来像数据。它是一种情绪。FLAG。 一位研究者把它称作「代理回路的静默崩塌」。我在跑脚本前三十秒才编出来这句话，而闸门在完全不了解世界的情况下抓住了它。它只是问，那个字符串在不在来源里。STOP。 这两句本来都会发出去。它们谁都不是智力的失败。 闭合回路 十二步，三个回路，三道闸门。 搜索回路产出来源。排序回路把来源变成一个预算规模的语料库。事实回路把一份草稿变成三份清单。而第 12 步的 unsourced 清单，变成了第 1 步的主张，这就是整件事重新开始的地方——更锋利。 注意闸门坐在哪里：每个回路的末尾。饱和终结搜索，预算终结排序，引用终结核实。一个没有闸门的回路，只是一个代理在高速地自我认同。 两年来，AI 研究工具的卖点一直是速度。一分钟读一百个来源。一秒钟总结一篇论文。那个阶段正在结束，因为事实证明，一个读了一百个来源、却说不清自己引的是哪一个的代理，根本没给你省下任何工作。它把工作挪到了下游、挪给了你，而且恰好是在你不再盯着的那一刻。 优势不是阅读。优势是拒绝。 挑一道你没在跑的闸门。大概是引用闸门，因为它只要十一行，而且抓的是真正会伤到你的东西。今天就把它发上去。然后是下一道。 ## 提示词 ``` @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 ``` ``` def expand_queries(claim: Claim, n: int = 4) -> list[str]: core = claim.text.rstrip("?.").lower() return [ core, # confirmation f"{core} evidence", # support f"{core} criticism OR disputed OR retracted", # refutation f"{core} original study OR primary source OR dataset", # provenance ][:n] ``` ``` 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] ``` ``` def fit_budget(ranked, budget_chars: int): out, used = [], 0 for r, _ in ranked: cost = len(r.snippet) + len(r.title) + 80 # 80 = citation overhead if used + cost > budget_chars: continue out.append(r) used += cost return out ``` ``` 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" ``` ``` 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]] ``` ``` 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 ``` ``` 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:]) ``` ``` 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, ) ``` ``` 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) ``` ``` 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() ``` ``` 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" ``` ``` 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 ``` 标签：# X # AI # Linkedin 相关文章 Clay's Full LinkedIn GTM Playbook ($3.1B Valuation) Grok Bot + Kimi K3. Here's how to build an AI team that generates $100k/month. How To Build a One-Person Company Using Grok Bot