用 DeepSeek Search 与 Fetch API 给 Agent 装上实时联网检索

两个接口都是只读请求，都挂在 https://api.deepseek.com 下，用和 DeepSeek 模型 API 完全相同的 Bearer 鉴权。Search 的端点是 POST /v1/tools/search，你给它一个查询词，它返回一组结构化结果（标题、链接、摘要、正文、站点名、发布时间、质量分）；Fetch 的端点是 POST /v1/tools/fetch，你给它一个 URL，它把那个页面抓成 Markdown 正文返回。典型用法是两段式：先用 Search 在千亿级索引里定位候选，挑出质量分高的几条，再用 Fetch 把选中的页面读成 Markdown 喂给模型——这样既省 token，又避免模型被整页噪声干扰。

# 两个端点，同一种鉴权
POST https://api.deepseek.com/v1/tools/search   # 找：query → 结果列表
POST https://api.deepseek.com/v1/tools/fetch    # 读：url   → Markdown 正文

# 请求头两个都必填
Authorization: Bearer <你的 DEEPSEEK_API_KEY>
Content-Type: application/json

Search 的核心请求参数有五个：query（查询词，1 到 128 个 Unicode 字符，不能全是空白）、count（返回条数，默认 10，可取值 1 到 50）、need_content（是否附带网页正文）、ori_question（把用户的原始问题传进去，帮它理解搜索意图）、localization（传一个 IP 地址，用于优先返回本地化结果）。返回数组里每一项都带 rerank_score（0 到 3 的排序分，官方标注 1.5 分及以上为优质结果）、authority（0 到 4 的权威性分级，3 级及以上为极权威，未知时为 -1）、snippet（约 100 到 200 字符，给界面展示用）、summary（约 600 到 800 字符，给模型读用）、可选的 content（网页正文）、site_name、site_icon 与 RFC 3339 格式的 date。

curl https://api.deepseek.com/v1/tools/search \
  -H "Authorization: Bearer $DEEPSEEK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "MCP 无状态规范 变更",
    "count": 5,
    "need_content": false,
    "ori_question": "MCP 协议 2026 年有哪些破坏性变更？"
  }'

Search 会把你的查询词连同搜索语法一起转发给引擎，支持 site: 限定站点、-site: 排除站点、减号排除词、英文双引号精确短语（含 -"..." 排除短语）、lang: 限定语言、after: 与 before: 限定发布时间。这些操作符可以在同一条查询里组合使用，这是控制检索质量最省力的手段——与其让模型事后从 50 条结果里筛，不如在查询阶段就把范围钉死。

# 只搜 GitHub 上的内容
"query": "DeepSeek Search API site:github.com"

# 排除某个站点与广告页
"query": "智能体框架 -site:example.com -advertisement"

# 精确短语 + 排除短语
"query": "MCP 规范 \"streamable http\" -\"release notes\""

# 限定中英文 + 时间范围
"query": "智能体记忆 lang:zh,en after:2026-01-01 before:2026-10-01"

Fetch 的请求体只有一个 url 字段，长度不超过 8192 个字符，返回 title 与 content（Markdown 正文）。它适合放在 Search 之后：Agent 先看检索结果的标题与长摘要，挑出两三条最相关的，再逐条抓取正文交给模型分析。需要注意的是，抓取结果受目标站点自身的可用性与访问策略影响——站点不可用、被站点策略屏蔽、内容类型不受支持，或页面根本提取不出 Markdown，都会返回错误而不是内容。

curl https://api.deepseek.com/v1/tools/fetch \
  -H "Authorization: Bearer $DEEPSEEK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://api-docs.deepseek.com/updates"}'

这是这套接口最实用的设计：每条结果都自带 rerank_score 与 authority，你可以先按分数过滤，只把通过闸门的页面抓下来喂给模型。一次研究任务跑 200 次搜索加 500 次抓取的场景下，提前砍掉一半低质结果，省下的是实打实的模型上下文和钱。下面这段 Python 把「检索 → 过滤 → 抓取 → 拼装上下文」串成一条可直接复用的函数。

import os, json, requests

BASE = "https://api.deepseek.com"
HEADERS = {
    "Authorization": f"Bearer {os.environ['DEEPSEEK_API_KEY']}",
    "Content-Type": "application/json",
}

def search(query, count=10, ori_question=None):
    body = {"query": query[:128], "count": count}
    if ori_question:
        body["ori_question"] = ori_question
    r = requests.post(f"{BASE}/v1/tools/search", headers=HEADERS, json=body, timeout=30)
    r.raise_for_status()
    return r.json().get("results", [])

def fetch(url):
    r = requests.post(f"{BASE}/v1/tools/fetch", headers=HEADERS, json={"url": url}, timeout=30)
    r.raise_for_status()
    return r.json().get("content", "")

def research(topic, min_score=1.5, min_authority=3, top_n=3):
    hits = search(topic, count=20, ori_question=topic)
    # 闸门：排序分与权威性双达标才进入下一轮
    good = [h for h in hits
            if h.get("rerank_score", 0) >= min_score
            and h.get("authority", -1) >= min_authority]
    ctx = []
    for h in good[:top_n]:
        ctx.append(f"# {h['title']}\n来源: {h['url']}\n时间: {h.get('date')}\n"
                   f"{fetch(h['url'])[:6000]}")
    return hits, good, "\n\n---\n\n".join(ctx)

Search 与 Fetch 各自独立限速，均为每账号每秒 5 次（按账号计，不按 Key 计），互不占用对方额度，也不占用模型并发额度；Fetch 还额外按目标域名限速，超限返回 429。错误响应采用与 OpenAI 兼容的格式，官方给出的重试策略很明确：408、429 与 5xx 等待后重试；402（余额不足）先充值再重试；其余 4xx 在未修改请求的情况下不要重试。把它封成一个带退避的包装器，Agent 循环就不用关心这些细节。

import time

def call_with_retry(fn, *args, max_retry=3, **kwargs):
    for i in range(max_retry):
        try:
            return fn(*args, **kwargs)
        except requests.HTTPError as e:
            code = e.response.status_code
            if code in (408, 429, 500, 504) and i < max_retry - 1:
                time.sleep(2 ** i)      # 指数退避：1s、2s、4s
                continue
            if code == 402:
                raise RuntimeError("账户余额不足，充值后重试")
            raise                        # 其余 4xx 不要盲目重试
    return None

# 串行节流：把 QPS 压在 5 以内
_last = [0.0]
def throttle(qps=4):
    gap = 1.0 / qps - (time.time() - _last[0])
    if gap > 0:
        time.sleep(gap)
    _last[0] = time.time()