用 Exa 搜索 API 生成带引用简报：去重来源并压缩上下文

Exa 是给程序调用的搜索 API，返回网页的相关摘录（highlights）或正文。 这篇做 exa_brief.py：用官方 SDK 搜索一个主题，去重后生成带编号引用的简报（brief.md），以及一份限定长度、可以直接交给大模型的上下文（context.txt）。

先看清成品

• 真实接口 401，SDK 抛出 ValueError；脚本识别后给出“去控制台创建 key”的提示。
• 替身上跑通：4 条结果去重为 3 条（utm 参数、www.、末尾斜杠的同一篇文章合并）；没有发布日期的结果显示“日期未知”。
• 发现并修复：引用链接中带着 ?utm_source=newsletter。pytest 6 项通过。

合规提醒：Exa 返回的是第三方网页内容，转载或商用前要遵守原网站的版权和使用条款；不要用 Exa 采集个人信息。本文只使用 Exa 官方 API 和官方 Python SDK（exa-py）。API key 在 Exa 控制台（dashboard.exa.ai）创建，放在环境变量 EXA_API_KEY 中。调用费用以 Exa 官方价格页为准（本文不列价格）。

运行说明：box 上没有 Exa API key。我用无效 key 真实调用了 POST https://api.exa.ai/search，得到 HTTP 401 {"error":"Invalid API key. …","tag":"INVALID_API_KEY"}；通过官方 SDK（exa-py 2.26.0）调用时，SDK 抛出的是 ValueError（2026-10-21 20:06，UTC+8）。其余响应来自我写的本地替身：两条结果照抄官方快速入门页面的示例响应，另外的结果和网页正文是我编的（example.com）。SDK 通过官方的 base_url 参数指向替身。所有脚本和测试都在 box 上真实运行过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

74-exa-A-src/
├── exa_brief.py
├── fake_exa.py
├── demo_local.py
└── test_brief.py

第 1 步：主脚本

"""用 Exa 官方 Python SDK（exa-py）搜索一个主题，生成带编号引用的简报（brief.md），以及一份限定长度的上下文（context.txt），可交给大模型使用。
官方建议：内容默认用 highlights；只有需要完整上下文时才用 text，且两者只选一个（同时请求会返回并计费两份）。
key 放在环境变量 EXA_API_KEY。"""
import argparse, os, re, sys
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
from exa_py import Exa

def norm_url(u):
    """去掉 utm_* 参数、www. 前缀和末尾斜杠，用于去重。"""
    p = urlsplit(u); q = urlencode([(k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith("utm_")])
    return urlunsplit((p.scheme.lower(), p.netloc.lower().removeprefix("www."), p.path.rstrip("/") or "/", q, ""))

def strip_tracking(u):
    """引用时只去掉 utm_* 跟踪参数，其余保持原样（演示输出里第一次就把 ?utm_source=newsletter 带进了引用）。"""
    p = urlsplit(u); q = urlencode([(k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith("utm_")])
    return urlunsplit((p.scheme, p.netloc, p.path, q, p.fragment))

def client():
    return Exa(api_key=os.environ.get("EXA_API_KEY"), base_url=os.environ.get("EXA_BASE_URL", "https://api.exa.ai"), user_agent="binbho-tutorial/1.0")

def search(query, n=10, domains=None, since=None, exa=None):
    exa = exa or client()
    try:
        r = exa.search(query, type="auto", num_results=n, include_domains=domains or None, start_published_date=since, contents={"highlights": True})
    except ValueError as e:                     # SDK 把 HTTP 错误包装成 ValueError（实测 401 就是这样）
        if "401" in str(e): raise SystemExit("Exa 401：API key 无效。在 https://dashboard.exa.ai/api-keys 创建 key，放进 EXA_API_KEY") from None
        raise
    seen, out = set(), []
    for x in r.results:
        k = norm_url(x.url)
        if k not in seen: seen.add(k); out.append(x)
    return out

def brief(query, results, per_source=600, budget=4000):
    md = [f"# {query}", "", "> 由 Exa 搜索结果自动整理，摘录未经核实；引用前请打开原文。", ""]
    ctx, used = [], 0
    for i, x in enumerate(results, 1):
        date = (x.published_date or "")[:10] or "日期未知"
        hl = re.sub(r"\s+", " ", " … ".join(x.highlights or [])).strip()
        if len(hl) > per_source: hl = hl[:per_source].rstrip() + "…"
        url = strip_tracking(x.url)
        md += [f"## [{i}] {x.title}", f"- 来源：{url}", f"- 发布：{date}", "", f"> {hl}", ""]
        chunk = f"[{i}] {x.title} ({url}, {date})\n{hl}\n"
        if used + len(chunk) > budget: md.append(f"<!-- 上下文已满：[{i}] 及之后的来源没有放进 context.txt -->"); break
        ctx.append(chunk); used += len(chunk)
    return "\n".join(md) + "\n", "\n".join(ctx)

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("query"); ap.add_argument("-n", type=int, default=10); ap.add_argument("--domain", action="append")
    ap.add_argument("--since", help="只要这个日期之后发布的，例如 2026-01-01"); ap.add_argument("--budget", type=int, default=4000)
    a = ap.parse_args(argv)
    res = search(a.query, a.n, a.domain, a.since)
    md, ctx = brief(a.query, res, budget=a.budget)
    open("brief.md", "w", encoding="utf-8").write(md); open("context.txt", "w", encoding="utf-8").write(ctx)
    print(f"{len(res)} 个来源（已去重）→ brief.md；context.txt {len(ctx)} 字符（上限 {a.budget}）")
    return 0

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

第 2 步：本地替身

"""本地替身（我写的，不是 Exa 官方的）。/search 的结果照抄 Exa 官方 Search 快速入门页面中的示例响应（两条：arXiv 和 Thoughtworks），
另加两条我编的结果（example.com）用来测试去重和截断；401 的响应体是对真实接口的实测结果。"""
import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

OFFICIAL = [
    {"id": "https://arxiv.org/html/2608.21702", "title": "From Association to Causation: Improving Retrieval Precision ofRetrieval-Augmented Generation via Causal Relations and an Attention Mechanism",
     "url": "https://arxiv.org/html/2608.21702", "highlights": ["Retrieval-Augmented Generation (RAG) grounds LLM generation on retrieved documents, but the standard terminal retrieval stage—dense-vector similarity, optionally followed by reranking—often returns documents that merely share keywords with the query without containing the needed information...\n..."]},
    {"id": "https://www.thoughtworks.com/en-us/insights/blog/generative-ai/four-retrieval-techniques-improve-rag", "title": "Four retrieval techniques to improve RAG you need to know",
     "url": "https://www.thoughtworks.com/en-us/insights/blog/generative-ai/four-retrieval-techniques-improve-rag", "publishedDate": "2025-04-14T00:00:00.000Z",
     "highlights": ["It's not surprising, then, that we've seen a range of different approaches emerge that attempt to address RAG's limitations over the last year or so.\n..."]},
]
MINE = [
    {"id": "https://example.com/rag-guide?utm_source=newsletter", "title": "示例：RAG 检索调优指南", "url": "https://example.com/rag-guide?utm_source=newsletter",
     "publishedDate": "2026-10-19T08:00:00.000Z", "highlights": ["混合检索（BM25 + 向量）在长尾查询上更稳定。" * 30]},
    {"id": "https://www.example.com/rag-guide/", "title": "示例：RAG 检索调优指南", "url": "https://www.example.com/rag-guide/",
     "publishedDate": "2026-10-19T08:00:00.000Z", "highlights": ["同一篇文章的另一个网址。"]},
]
PAGES = {"https://example.com/benchmark": "In our benchmark, hybrid retrieval improved recall@10 from 0.61 to 0.74 on the long-tail query set. Latency rose by 12 ms.",
         "https://example.com/blocked": None}

class State:
    def __init__(self): self.bodies = []; self.results = OFFICIAL + MINE

def make(st):
    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_POST(self):
            body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))); st.bodies.append((self.path, body))
            if self.headers.get("x-api-key") != "good":
                return self.send(401, {"requestId": "f5c0d981250e865f0008dba10a9dee5f", "error": "Invalid API key. Provide a valid key using 'Authorization: Bearer <key>' or 'x-api-key: <key>'. Create a key at https://dashboard.exa.ai/api-keys", "tag": "INVALID_API_KEY"})
            if self.path == "/search":
                since = body.get("startPublishedDate")
                res = [r for r in st.results if not since or (r.get("publishedDate") or "") >= since]
                return self.send(200, {"requestId": "demo", "results": res[: body.get("numResults", 10)], "searchTime": 1.0})
            if self.path == "/contents":
                res, stat = [], []
                for u in body["urls"]:
                    t = PAGES.get(u)
                    if t is None: stat.append({"id": u, "status": "error", "error": {"tag": "CRAWL_NOT_FOUND", "httpStatusCode": 404}})
                    else: res.append({"id": u, "url": u, "title": u, "text": t}); stat.append({"id": u, "status": "success"})
                return self.send(200, {"requestId": "demo", "results": res, "statuses": stat})
            self.send(404, {"error": "not found"})
    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]}"

第 3 步：演示脚本

"""用本地替身演示（不连接 Exa，不花费额度）：python demo_local.py "查询" [其他参数]"""
import os, sys, fake_exa as F
srv, url = F.make(F.State()); os.environ["EXA_BASE_URL"] = url; os.environ.setdefault("EXA_API_KEY", "good")
import exa_brief as B
try: code = B.main(sys.argv[1:])
finally: srv.shutdown()
sys.exit(code)

第 4 步：测试

import pytest, fake_exa as F, exa_brief as B

@pytest.fixture
def exa(monkeypatch, tmp_path):
    st = F.State(); srv, url = F.make(st)
    monkeypatch.setenv("EXA_BASE_URL", url); monkeypatch.setenv("EXA_API_KEY", "good"); monkeypatch.chdir(tmp_path)
    yield st; srv.shutdown()

def test_request_shape(exa):
    B.search("recent techniques for improving retrieval in RAG systems", n=5, domains=["arxiv.org"], since="2026-01-01")
    path, body = exa.bodies[0]
    assert path == "/search" and body == {"query": "recent techniques for improving retrieval in RAG systems", "type": "auto", "numResults": 5,
                                          "includeDomains": ["arxiv.org"], "startPublishedDate": "2026-01-01", "contents": {"highlights": True}}

def test_dedupe_and_brief(exa):
    res = B.search("rag")
    assert len(res) == 3                                                       # utm 参数和 www./末尾斜杠版本合并
    md, ctx = B.brief("rag", res)
    assert md.count("## [") == 3 and "发布：日期未知" in md and "发布：2025-04-14" in md
    assert "[3] 示例：RAG 检索调优指南" in ctx

def test_budget(exa):
    md, ctx = B.brief("rag", B.search("rag"), budget=700)
    assert len(ctx) <= 700 and "[1]" in ctx and "[3]" not in ctx and "上下文已满" in md

def test_bad_key(exa, monkeypatch):
    monkeypatch.setenv("EXA_API_KEY", "bad")
    with pytest.raises(SystemExit, match="401"): B.search("x")

def test_norm_url():
    assert B.norm_url("https://www.Example.com/a/?utm_source=x&id=3") == "https://example.com/a?id=3"

def test_citation_without_utm(exa):
    md, ctx = B.brief("rag", B.search("rag"))
    assert "utm_source" not in md + ctx and "- 来源：https://example.com/rag-guide\n" in md

第 5 步：运行

$ python demo_local.py "recent techniques for improving retrieval in RAG systems" --budget 1500
3 个来源（已去重）→ brief.md；context.txt 1484 字符（上限 1500）
$ head brief.md
# recent techniques for improving retrieval in RAG systems

> 由 Exa 搜索结果自动整理，摘录未经核实；引用前请打开原文。

## [1] From Association to Causation: Improving Retrieval Precision ofRetrieval-Augmented Generation via Causal Relations and an Attention Mechanism
- 来源：https://arxiv.org/html/2608.21702
- 发布：日期未知

> Retrieval-Augmented Generation (RAG) grounds LLM generation on retrieved documents, but the standard terminal retrieval stage—dense-vector similarity, optionally followed by reranking—often returns documents that merely share keywords with the query without containing the needed information... ...

## [2] Four retrieval techniques to improve RAG you need to know
- 来源：https://www.thoughtworks.com/en-us/insights/blog/generative-ai/four-retrieval-techniques-improve-rag
- 发布：2025-04-14

> It's not surprising, then, that we've seen a range of different approaches emerge that attempt to address RAG's limitations over the last year or so. ...

## [3] 示例：RAG 检索调优指南
- 来源：https://example.com/rag-guide
- 发布：2026-10-19

> 混合检索（BM25 + 向量）在长尾查询上更稳定。混合检索（BM25 + 向量）在长尾查询上更稳定。混合检索（BM25 + 向量）在长尾查询上更稳定。混合检索（BM25 + 向量）在长尾查询上更稳定。混合检索（BM25 + 
…
$ EXA_API_KEY=bad python demo_local.py x
Exa 401：API key 无效。在 https://dashboard.exa.ai/api-keys 创建 key，放进 EXA_API_KEY
$ curl -s -X POST https://api.exa.ai/search -H 'Content-Type: application/json' -H 'Authorization: Bearer invalid' -d '{"query":"test","type":"lite"}'     # 真实接口
{"requestId":"f5c0d981250e865f0008dba10a9dee5f","error":"Invalid API key. Provide a valid key using 'Authorization: Bearer <key>' or 'x-api-key: <key>'. Create a key at https://dashboard.exa.ai/api-keys","tag":"INVALID_API_KEY"} (HTTP 401)
$ python -m pytest -q
6 passed

怎么检查结果？

• 测试会检查 SDK 实际发出的请求体（camelCase：numResults、includeDomains、startPublishedDate）。
• context.txt 不超过 --budget；放不下的来源在 brief.md 中有注释说明。
• 简报中的摘录是 Exa 选出的片段，引用前打开原文。

调试示例

1. SDK 的 401 是 ValueError。 用无效 key 通过 exa-py 调用，得到的是：

ValueError Request failed with status code 401: {"requestId":"5af9…","error":"Invalid API key. …","tag":"INVALID_API_KEY"}

不是专门的认证异常类型。脚本捕获 ValueError，看到 401 时给出明确提示，其他错误原样抛出。

2. 官方示例中第一条结果没有 publishedDate。 我照抄官方示例响应做替身时注意到，arXiv 那条结果没有 publishedDate 字段。简报中显示“日期未知”，而不是报错或留空。

3. 引用中带跟踪参数。 第一次运行演示，第 3 个来源的链接是 https://example.com/rag-guide?utm_source=newsletter。去重时已经去掉了 utm 参数，但输出时用的仍是原始网址。修复：输出前去掉 utm_*，其他参数保留。

常见问题

模板规则

• 默认用 highlights。
• 引用时去掉跟踪参数。
• 摘录要回到原文核对。
• key 只放在环境变量里。

项目资源

让 Exa 找来源，让引用编号把每句话和来源对上。