用 Consensus API 整理论文检索结果：按证据类型生成阅读清单

Consensus 的官方 API 可以按研究问题检索论文，并返回研究类型、样本量、是否预印本等字段。 这篇做 evidence_table.py：把检索结果整理成按证据类型排序的阅读清单（Markdown + CSV），自动去重，并标出预印本、小样本、动物研究和缺少 DOI 的条目。

先看清成品

• 用无效 key 真实调用 Consensus API，得到 401，脚本给出可读提示。
• 本地替身上跑通：自动翻页、在 is_end 处停止；7 篇去重为 6 篇；荟萃分析排第一，RCT 按年份排序。
• 发现并修正：布尔参数被 requests 发成 "True"，改为 "true"。pytest 5 项通过。

合规提醒：Consensus 的回答和要点摘录是 AI 生成的，可能遗漏或误读论文；引用时必须引用原论文，并阅读原文核对。医疗、健康、法律问题请咨询专业人士，不要把检索结果当成诊疗建议。本文只使用官方功能：Consensus 网页版、官方 API（x-api-key），以及 Crossref 公开 REST API。不抓取网页，不共享账号。调用次数和套餐以官方页面为准（本文不列价格）。

运行说明：box 上没有 Consensus 账号和 API key。我用无效 key 和不带 key 各真实调用了一次 GET https://api.consensus.app/v1/search，分别得到 HTTP 401 {"detail":"Invalid authentication credentials."} 和 {"detail":"Not authenticated"}（2026-10-21 20:06，UTC+8）。A 篇其余响应来自我写的本地替身（字段名照官方 OpenAPI，论文是虚构的，DOI 用 10.5555 测试前缀）。B 篇真实调用了 Crossref 公开 API。C 篇是纯本地规则。所有脚本和测试都在 box 上真实运行过（Python 3.13）。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

本文按“系统综述/荟萃分析 → RCT → 观察性研究 → 综述 → 个案 → 动物/体外”排序，只是阅读顺序，不代表单篇论文质量；一项设计差的荟萃分析不如一项好的 RCT。

文件结构

71-consensus-A-src/
├── evidence_table.py
├── fake_consensus.py
├── demo_local.py
└── test_evidence.py

第 1 步：创建 API key（手动）

登录 Consensus，在账户设置中创建 API key，放进环境变量 CONSENSUS_API_KEY。（本文没有账号，没有实际操作这一步。）

第 2 步：主脚本

"""用 Consensus 官方 API（GET /v1/search）把一个研究问题的检索结果整理成证据表（Markdown + CSV）。
- 认证：请求头 x-api-key（官方文档）；key 放在环境变量 CONSENSUS_API_KEY
- 分页：page 从 0 开始，看 is_end / next_page；(page+1)×page_size 不能超过 1000
- 每 100 篇结果算 1 次调用（官方说明），所以默认只取 1 页、20 篇
证据表只是阅读清单，不是结论：每篇都要打开原文核对。"""
import argparse, csv, os, sys, requests

API = os.environ.get("CONSENSUS_API", "https://api.consensus.app")
# 证据等级排序（数值越小越靠前）；只用于排序阅读顺序，不代表单篇论文质量
RANK = {"systematic review": 0, "meta-analysis": 0, "rct": 1, "non-rct experimental": 2, "non-randomized experimental study": 2,
        "cohort study": 3, "case-control study": 3, "longitudinal / panel data study": 3, "cross-sectional study": 4,
        "non-rct observational study": 4, "literature review": 5, "case report": 6, "case study": 6, "animal": 7,
        "non-rct in vitro": 7, "bench experiment": 7, "commentary or perspective": 8}

class ConsensusError(Exception): pass

def search(query, key=None, pages=1, **filters):
    key = key or os.environ.get("CONSENSUS_API_KEY")
    if not key: raise ConsensusError("没有设置 CONSENSUS_API_KEY")
    h = {"x-api-key": key, "User-Agent": "binbho-tutorial/1.0"}
    out, page = [], 0
    # 布尔值按 true/false 发送（requests 默认会把 True 变成字符串 "True"）
    params = {k: (str(v).lower() if isinstance(v, bool) else v) for k, v in filters.items() if v is not None}
    for _ in range(pages):
        r = requests.get(f"{API}/v1/search", params={"query": query, "page": page, **params}, headers=h, timeout=60)
        if r.status_code == 401: raise ConsensusError(f"401：{r.json().get('detail')}（检查 API key）")
        if r.status_code == 422: raise ConsensusError(f"422 参数错误：{r.json().get('detail')}")
        r.raise_for_status(); j = r.json(); out += j["results"]
        if j.get("is_end") or j.get("next_page") is None: break
        page = j["next_page"]
    return out

def flags(p, min_n=100):
    f = []
    if p.get("is_preprint"): f.append("预印本")
    if p.get("sample_size") is not None and p["sample_size"] < min_n and (p.get("study_type") or "") not in ("meta-analysis", "systematic review"): f.append(f"样本 {p['sample_size']}")
    if p.get("population_type") and p["population_type"] != "human": f.append(p["population_type"])
    if not p.get("doi"): f.append("无 DOI")
    return "；".join(f)

def table(papers):
    rows = []
    for p in papers:
        st = (p.get("study_type") or "").lower()
        rows.append({"rank": RANK.get(st, 9), "year": p.get("publish_year"), "study_type": st or "未知", "n": p.get("sample_size") or p.get("study_count") or "",
                     "title": p["title"], "journal": p.get("journal_name", ""), "doi": p.get("doi", ""), "citations": p.get("citation_count", 0),
                     "takeaway": p.get("takeaway") or "", "flags": flags(p), "url": p.get("url", "")})
    rows.sort(key=lambda r: (r["rank"], -(r["year"] or 0), -r["citations"]))
    seen, uniq = set(), []
    for r in rows:                                   # 同一篇论文可能以不同版本出现，按 DOI（没有时按标题）去重
        k = (r["doi"] or r["title"]).lower().strip()
        if k not in seen: seen.add(k); uniq.append(r)
    return uniq

def to_md(q, rows):
    L = [f"# 证据表：{q}", "", "| 类型 | 年份 | 样本/研究数 | 标题 | 期刊 | 被引 | 要点（Consensus 摘录，需核对） | 提醒 |", "|---|---|---|---|---|---|---|---|"]
    for r in rows:
        t = f"[{r['title']}](https://doi.org/{r['doi']})" if r["doi"] else r["title"]
        L.append(f"| {r['study_type']} | {r['year']} | {r['n']} | {t} | {r['journal']} | {r['citations']} | {r['takeaway']} | {r['flags']} |")
    return "\n".join(L) + "\n"

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("query"); ap.add_argument("--year-min", type=int); ap.add_argument("--human", action="store_true")
    ap.add_argument("--exclude-preprints", action="store_true"); ap.add_argument("--pages", type=int, default=1); ap.add_argument("--out", default="evidence")
    a = ap.parse_args(argv)
    try:
        papers = search(a.query, pages=a.pages, year_min=a.year_min, human=True if a.human else None, exclude_preprints=True if a.exclude_preprints else None)
    except ConsensusError as e:
        print("Consensus API 错误：", e); return 1
    rows = table(papers)
    open(a.out + ".md", "w", encoding="utf-8").write(to_md(a.query, rows))
    with open(a.out + ".csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0]) if rows else ["title"]); w.writeheader(); w.writerows(rows)
    print(f"{len(papers)} 篇 → 去重后 {len(rows)} 篇，写入 {a.out}.md / {a.out}.csv")
    return 0

if __name__ == "__main__":
    sys.exit(main())

第 3 步：本地替身

"""本地替身（我写的，不是 Consensus 官方的）。字段名照官方 OpenAPI 中的 QueryResult；论文标题、DOI、数据都是虚构的（DOI 用 10.5555 测试前缀）。
401 的响应体是对真实接口的实测结果。"""
import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

def P(i, st, year, n, cites, pre=False, pop="human", doi=True, title=None):
    return {"title": title or f"虚构研究 {i}", "abstract": "…", "authors": ["A. Author"], "doi": f"10.5555/demo.{i}" if doi else "",
            "journal_name": "Demo Journal", "pages": "1-10", "volume": "1", "publish_year": year, "url": f"https://consensus.app/papers/demo-{i}",
            "citation_count": cites, "study_type": st, "takeaway": f"研究 {i} 的要点", "sample_size": n, "study_count": None,
            "population_type": pop, "is_preprint": pre}

PAPERS = [P(1, "rct", 2021, 48, 30), P(2, "meta-analysis", 2023, None, 120), P(3, "cohort study", 2019, 5200, 210),
          P(4, "animal", 2022, 24, 15, pop="animal"), P(5, "rct", 2024, 300, 5, pre=True), P(6, None, 2018, None, 2, doi=False),
          P(2, "meta-analysis", 2023, None, 120)]          # 第 7 条与第 2 条重复（同一 DOI）

def make(page_size=4):
    calls = []
    class H(BaseHTTPRequestHandler):
        def log_message(self, *a): pass
        def send(self, code, obj):
            b = json.dumps(obj).encode(); self.send_response(code); self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
        def do_GET(self):
            u = urlparse(self.path); q = {k: v[0] for k, v in parse_qs(u.query).items()}; calls.append(q)
            if "x-api-key" not in {k.lower() for k in self.headers.keys()}: return self.send(401, {"detail": "Not authenticated"})
            if self.headers.get("x-api-key") != "good": return self.send(401, {"detail": "Invalid authentication credentials."})
            if u.path != "/v1/search": return self.send(404, {"detail": "Not Found"})
            page = int(q.get("page", 0)); res = PAPERS[page * page_size:(page + 1) * page_size]
            end = (page + 1) * page_size >= len(PAPERS)
            self.send(200, {"results": res, "page": page, "page_size": page_size, "is_end": end, "next_page": None if end else page + 1})
    srv = ThreadingHTTPServer(("127.0.0.1", 0), H); threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv, f"http://127.0.0.1:{srv.server_address[1]}", calls

第 4 步：演示脚本

"""用本地替身演示（不连接 Consensus，不消耗调用次数）：python demo_local.py "问题" [参数]"""
import os, sys, fake_consensus as F
srv, url, calls = F.make(); os.environ["CONSENSUS_API"] = url; os.environ.setdefault("CONSENSUS_API_KEY", "good")
import evidence_table as E
try: code = E.main(sys.argv[1:])
finally: srv.shutdown()
print("请求参数：", calls); sys.exit(code)

第 5 步：测试

import pytest, fake_consensus as F

@pytest.fixture
def api(monkeypatch):
    srv, url, calls = F.make(); monkeypatch.setattr("evidence_table.API", url); yield calls; srv.shutdown()

def test_pagination_and_dedupe(api):
    import evidence_table as E
    papers = E.search("q", key="good", pages=5)
    assert len(papers) == 7 and [c["page"] for c in api] == ["0", "1"]       # is_end 后停止
    rows = E.table(papers)
    assert len(rows) == 6 and rows[0]["study_type"] == "meta-analysis"

def test_order_and_flags(api):
    import evidence_table as E
    rows = E.table(E.search("q", key="good", pages=2))
    assert [r["study_type"] for r in rows] == ["meta-analysis", "rct", "rct", "cohort study", "animal", "未知"]
    assert rows[1]["flags"] == "预印本"                    # 2024 年的 RCT 排在 2021 年之前，且是预印本
    assert rows[2]["flags"] == "样本 48"
    assert "animal" in rows[4]["flags"] and rows[5]["flags"] == "无 DOI"

def test_filters_passed(api):
    import evidence_table as E
    E.search("q", key="good", year_min=2020, human=True, exclude_preprints=None)
    assert api[0]["year_min"] == "2020" and api[0]["human"] == "true" and "exclude_preprints" not in api[0]

def test_bad_key(api):
    import evidence_table as E
    with pytest.raises(E.ConsensusError, match="Invalid authentication credentials"):
        E.search("q", key="bad")

def test_cli(api, tmp_path, monkeypatch):
    import evidence_table as E
    monkeypatch.setenv("CONSENSUS_API_KEY", "good")
    assert E.main(["间歇性禁食与胰岛素敏感性", "--pages", "2", "--out", str(tmp_path / "ev")]) == 0
    md = (tmp_path / "ev.md").read_text(encoding="utf-8")
    assert "[虚构研究 2](https://doi.org/10.5555/demo.2)" in md and md.count("虚构研究 2") == 1

第 6 步：运行

$ python demo_local.py "间歇性禁食是否改善成年人的胰岛素敏感性" --pages 3 --human --year-min 2018
7 篇 → 去重后 6 篇，写入 evidence.md / evidence.csv
请求参数： [{'query': '间歇性禁食是否改善成年人的胰岛素敏感性', 'page': '0', 'year_min': '2018', 'human': 'true'}, {'query': '间歇性禁食是否改善成年人的胰岛素敏感性', 'page': '1', 'year_min': '2018', 'human': 'true'}]
$ cat evidence.md
# 证据表：间歇性禁食是否改善成年人的胰岛素敏感性

| 类型 | 年份 | 样本/研究数 | 标题 | 期刊 | 被引 | 要点（Consensus 摘录，需核对） | 提醒 |
|---|---|---|---|---|---|---|---|
| meta-analysis | 2023 |  | [虚构研究 2](https://doi.org/10.5555/demo.2) | Demo Journal | 120 | 研究 2 的要点 |  |
| rct | 2024 | 300 | [虚构研究 5](https://doi.org/10.5555/demo.5) | Demo Journal | 5 | 研究 5 的要点 | 预印本 |
| rct | 2021 | 48 | [虚构研究 1](https://doi.org/10.5555/demo.1) | Demo Journal | 30 | 研究 1 的要点 | 样本 48 |
| cohort study | 2019 | 5200 | [虚构研究 3](https://doi.org/10.5555/demo.3) | Demo Journal | 210 | 研究 3 的要点 |  |
| animal | 2022 | 24 | [虚构研究 4](https://doi.org/10.5555/demo.4) | Demo Journal | 15 | 研究 4 的要点 | 样本 24；animal |
| 未知 | 2018 |  | 虚构研究 6 | Demo Journal | 2 | 研究 6 的要点 | 无 DOI |
$ CONSENSUS_API_KEY=bad python demo_local.py x
Consensus API 错误： 401：Invalid authentication credentials.（检查 API key）
$ curl -s -H 'x-api-key: invalid' 'https://api.consensus.app/v1/search?query=intermittent%20fasting'      # 真实接口
{"detail":"Invalid authentication credentials."}
$ python -m pytest -q
5 passed

怎么检查结果？

• 第二页的 is_end 为 true 后不再请求（替身中只有 2 页）。
• 同一 DOI 只出现一次。
• 每一行的“提醒”列都要看：预印本没有经过同行评审，动物研究不能直接推到人。
• 本地替身不执行筛选，year_min 等参数只检查是否正确发送。

调试示例

1. 布尔参数的格式。 第一版直接把 human=True 交给 requests，测试中替身收到的是字符串 "True"。Consensus API 是否接受首字母大写的写法，我没有 key，无法验证；为了不依赖服务端的宽松解析，改为发送小写 true/false。

2. 两种 401。 真实接口对“没带 key”和“key 无效”给出不同提示（Not authenticated 与 Invalid authentication credentials.）。脚本直接显示服务端的 detail，看到第一种说明环境变量没生效，看到第二种说明 key 本身不对。

常见问题

模板规则

• 证据表是阅读清单，不是结论。
• 每篇论文都要打开原文核对要点。
• 预印本、小样本、动物研究要单独看待。
• key 只放在环境变量里。

项目资源

让 API 帮你找论文、排顺序，结论要自己读原文得出。