摘要DeepSeek昨天10月17日正式开放了自研搜索APISearch和Fetch两个接口均为5元/千次。本文从实际工程视角拆解这套API的调用方式、限流处理、多轮检索编排以及在生产环境中如何用它替代Tavily/Brave等高价方案给AI Agent装上低成本、高可靠的联网能力。附完整可运行的Python代码。为什么我盯上了这个API做Agent开发的朋友应该都有一个体感让模型联网这件事比你想象的要贵。之前我帮团队搭建舆情监控系统的时候给Agent接的是Tavily的搜索API。效果确实不错但月底一看账单——光是搜索这一项就花了小一万。要知道我们的调用量根本算不上大日均也就几百次。更头疼的是搜索结果质量参差不齐。有些返回的内容明显是SEO垃圾Agent拿到这些毒数据后生成的回答也跟着跑偏。你模型再强输入是垃圾输出也是垃圾——GIGO定律从来没失手过。所以昨天DeepSeek宣布开放自研搜索API的时候我第一时间就测了一轮。原因很简单价格确实离谱地便宜——5元/千次Tavily的八到十六分之一自带反垃圾机制——DeepSeek声称对SEO作弊、GEO操纵和低质AIGC有专门过滤两个接口分工明确——Search负责搜Fetch负责抓网页正文并转成Markdown对LLM非常友好但便宜归便宜生产环境能不能用限流怎么处理多步检索怎么编排这些问题不搞清楚你不敢往线上放。所以我花了两天时间做了一轮完整的工程验证把踩过的坑和最终的方案整理出来希望能帮到同样在做Agent联网能力的朋友。先搞清楚这套API到底给了你什么在动手写代码之前得先把接口规格吃透。DeepSeek搜索API一共就两个端点Search接口端点POST https://api.deepseek.com/v1/tools/search功能输入query返回结构化搜索结果列表参数query搜索关键词最长128字符count返回结果数1-50默认10返回字段title、url、snippet、model_summary给模型看的长摘要、可选full_text、publish_date、relevance_score0-3、authority_score0-4计费5元/千次成功请求空结果和失败不计费Fetch接口端点POST https://api.deepseek.com/v1/tools/fetch功能输入URL返回网页正文Markdown格式返回字段title、contentMarkdown格式的正文计费5元/千次成功请求失败不计费两个关键细节值得注意第一Search的结果数不影响价格。你请求1条和请求50条价格一样。这意味着在不确定哪条结果有用的时候大方一点多拿几条反而更划算——反正不用额外花钱。第二两个接口的限流是独立的各自5次/秒/账号。Fetch还有一个按域名的额外限制具体数值官方没公布。这个在后面做并发控制时是个坑先记住。认证方式和DeepSeek的Chat API完全一样用同一个Bearer Token不需要额外申请。基础调用5分钟跑通第一个搜索先写最基础的调用代码把链路跑通import httpx import asyncio from typing import Optionalclass DeepSeekSearch:DeepSeek搜索API基础客户端BASE_URL quot;https://api.deepseek.com/v1/toolsquot; def __init__(self, api_key: str): self.api_key api_key self.client httpx.AsyncClient( headers{ quot;Authorizationquot;: fquot;Bearer {api_key}quot;, quot;Content-Typequot;: quot;application/jsonquot; }, timeout30.0 ) async def search(self, query: str, count: int 10) -gt; list[dict]: quot;quot;quot; 执行网页搜索 Args: query: 搜索关键词最长128字符 count: 返回结果数量1-50 Returns: 搜索结果列表每条包含title/url/snippet等字段 quot;quot;quot; if len(query) gt; 128: query query[:128] resp await self.client.post( fquot;{self.BASE_URL}/searchquot;, json{quot;queryquot;: query, quot;countquot;: count} ) resp.raise_for_status() data resp.json() return data.get(quot;resultsquot;, []) async def fetch_page(self, url: str) -gt; Optional[dict]: quot;quot;quot; 抓取网页正文返回Markdown格式 Args: url: 目标网页URL Returns: 包含title和content的字典失败返回None quot;quot;quot; resp await self.client.post( fquot;{self.BASE_URL}/fetchquot;, json{quot;urlquot;: url} ) if resp.status_code ! 200: return None data resp.json() if not data.get(quot;contentquot;): return None return data async def close(self): await self.client.aclose()快速验证async def main():searcher DeepSeekSearch(api_keyyour-api-key-here)results await searcher.search(quot;AI Agent框架 2026quot;, count5) for r in results: print(fquot;[{r.get(authority_score, 0)}] {r[title]}quot;) print(fquot; {r[url]}quot;) print(fquot; {r.get(model_summary, r.get(snippet, ))[:100]}...quot;) print() await searcher.close()asyncio.run(main())跑一下就能看到结果了。注意返回的authority_score字段——这是DeepSeek搜索特有的权威度评分0-4分分数越高说明来源越可信。后面做结果筛选的时候会用到。生产级改造限流、重试、成本控制的三道防线基础调用跑通之后别急着往生产环境放。有几个坑我替你踩了坑15次/秒的限流比你想象的更容易触发看起来5次/秒不算少但Agent一旦开始自主决策很容易在一个推理步骤里同时发起多个搜索。比如用户问一个复杂问题Agent可能同时搜索3个不同角度的关键词再Fetch 2个网页——这就5次了刚好顶到上限。更糟的是Fetch还有按域名的额外限制。如果你的Agent连续抓取同一个网站的多个页面可能2-3次就被429了。解决方案令牌桶限流 指数退避重试import time import asyncio from dataclasses import dataclass from collections import defaultdictdataclassclass RateLimitConfig:限流配置search_qps: float 4.0 # 留1个余量设4而不是5fetch_qps: float 4.0fetch_per_domain_qps: float 2.0 # 单域名更保守max_retries: int 3retry_base_delay: float 1.0class RateLimitedSearch(DeepSeekSearch):带限流控制的搜索客户端def __init__(self, api_key: str, config: Optional[RateLimitConfig] None): super().__init__(api_key) self.config config or RateLimitConfig() # 令牌桶记录每个维度的上次请求时间 self._last_search_time 0.0 self._last_fetch_time 0.0 self._domain_fetch_times: dict[str, list[float]] defaultdict(list) # 统计信息 self.stats { quot;search_callsquot;: 0, quot;fetch_callsquot;: 0, quot;retriesquot;: 0, quot;rate_limitedquot;: 0, quot;total_cost_yuanquot;: 0.0 # 估算成本 } async def _wait_for_slot(self, bucket: str, interval: float): quot;quot;quot;等待令牌桶空出位置quot;quot;quot; now time.monotonic() elapsed now - getattr(self, fquot;_last_{bucket}_timequot;, 0) if elapsed lt; interval: await asyncio.sleep(interval - elapsed) setattr(self, fquot;_last_{bucket}_timequot;, time.monotonic()) async def _wait_for_domain_slot(self, domain: str): quot;quot;quot;单域名的额外限流quot;quot;quot; now time.monotonic() # 清理超过1秒的记录 self._domain_fetch_times[domain] [ t for t in self._domain_fetch_times[domain] if now - t lt; 1.0 ] if len(self._domain_fetch_times[domain]) gt; int(self.config.fetch_per_domain_qps): oldest self._domain_fetch_times[domain][0] await asyncio.sleep(1.0 - (now - oldest)) self._domain_fetch_times[domain].append(time.monotonic()) async def search_with_retry(self, query: str, count: int 10) -gt; list[dict]: quot;quot;quot;带重试的搜索quot;quot;quot; await self._wait_for_slot(quot;searchquot;, 1.0 / self.config.search_qps) for attempt in range(self.config.max_retries): try: results await self.search(query, count) self.stats[quot;search_callsquot;] 1 if results: # 只有成功返回结果才计费 self.stats[quot;total_cost_yuanquot;] 0.005 return results except httpx.HTTPStatusError as e: if e.response.status_code 429: self.stats[quot;rate_limitedquot;] 1 delay self.config.retry_base_delay * (2 ** attempt) await asyncio.sleep(delay) self.stats[quot;retriesquot;] 1 else: raise return [] # 重试耗尽返回空 async def fetch_with_retry(self, url: str) -gt; Optional[dict]: quot;quot;quot;带重试的网页抓取quot;quot;quot; from urllib.parse import urlparse domain urlparse(url).netloc await self._wait_for_slot(quot;fetchquot;, 1.0 / self.config.fetch_qps) await self._wait_for_domain_slot(domain) for attempt in range(self.config.max_retries): try: result await self.fetch_page(url) self.stats[quot;fetch_callsquot;] 1 if result: self.stats[quot;total_cost_yuanquot;] 0.005 return result except httpx.HTTPStatusError as e: if e.response.status_code in (429, 503): self.stats[quot;rate_limitedquot;] 1 delay self.config.retry_base_delay * (2 ** attempt) await asyncio.sleep(delay) self.stats[quot;retriesquot;] 1 elif e.response.status_code in (403, 404): # 这些不是临时错误不需要重试 return None else: raise return None坑2搜索结果不等于可用信息DeepSeek搜索返回的结果有relevance_score和authority_score两个评分但这只是搜索引擎的排序信号不等于你的Agent能直接拿来用。实测下来authority_score 3的结果质量明显高于低分的。我的建议是默认只取高权威度结果低分的交给Agent决定是否要深入看。def filter_results(results: list[dict], min_authority: int 2, min_relevance: float 1.5) - list[dict]: 筛选高质量搜索结果根据实测authority_score gt; 3 的结果引用可信度较高 relevance_score gt; 1.5 通常意味着核心相关而非边缘提及。 quot;quot;quot; filtered [] for r in results: auth r.get(quot;authority_scorequot;, 0) rel r.get(quot;relevance_scorequot;, 0) if auth gt; min_authority and rel gt; min_relevance: filtered.append(r) # 如果筛选后结果太少降低阈值再放一轮 if len(filtered) lt; 2 and results: filtered [r for r in results if r.get(quot;authority_scorequot;, 0) gt; 1 and r.get(quot;relevance_scorequot;, 0) gt; 1.0] return sorted(filtered, keylambda x: (x.get(quot;authority_scorequot;, 0), x.get(quot;relevance_scorequot;, 0)), reverseTrue)坑3成本监控不能事后才看5元/千次听起来便宜但如果你的Agent在循环里疯狂搜索一个月下来也不是小数目。关键是空结果不计费但每次API调用都消耗时间。如果Agent每轮推理都搜索3-5次响应时间就上去了。class CostTracker: 成本追踪器def __init__(self, monthly_budget_yuan: float 150.0): self.monthly_budget monthly_budget_yuan self.daily_budget monthly_budget_yuan / 30 self.today_cost 0.0 self.today_date time.strftime(quot;%Y-%m-%dquot;) def record(self, cost_yuan: float) -gt; bool: quot;quot;quot;记录一笔花费返回是否允许继续quot;quot;quot; today time.strftime(quot;%Y-%m-%dquot;) if today ! self.today_date: self.today_cost 0.0 self.today_date today self.today_cost cost_yuan return self.today_cost lt; self.daily_budget property def remaining_ratio(self) -gt; float: quot;quot;quot;今日预算剩余比例quot;quot;quot; return max(0, 1 - self.today_cost / self.daily_budget)核心多步检索编排——让Agent学会搜了再搜前面都是基础设施。真正让这套搜索能力变强的是多步检索编排。什么意思就是一个复杂问题Agent不是搜一次就完事了。它需要先搜一个宽泛的问题了解大方向从搜索结果里发现新的关键词或线索用新线索再搜一轮缩小范围抓到具体网页后提取关键信息如果发现信息不够换个角度再搜这个过程和你自己上网查资料其实是一样的。关键是怎么用代码编排出来。import json from openai import AsyncOpenAIclass AgenticSearchPipeline:Agent驱动的多步搜索管线核心思路让LLM自己决定搜什么、什么时候停、要不要深入看某个网页 quot;quot;quot; def __init__(self, api_key: str, llm_client: AsyncOpenAI, search_client: RateLimitedSearch, cost_tracker: CostTracker): self.llm llm_client self.search search_client self.cost cost_tracker self.max_search_rounds 5 # 防止无限循环 async def research(self, question: str) -gt; dict: quot;quot;quot; 对一个问题执行多步搜索研究 Returns: { quot;answerquot;: str, # 最终答案 quot;sourcesquot;: list, # 引用的来源 quot;search_logquot;: list, # 搜索过程日志 quot;total_costquot;: float # 本次研究总成本 } quot;quot;quot; search_log [] collected_facts [] sources [] initial_cost self.cost.today_cost # 系统提示教Agent如何搜索 system_prompt quot;quot;quot;你是一个研究型AI助手。你的任务是通过多步搜索回答用户的复杂问题。每一步你需要决定是否需要搜索更多信息如果需要给出搜索关键词是否需要深入查看某个已发现的网页给出URL是否已经有足够信息可以给出最终答案你的可用工具search(query): 搜索网页fetch(url): 查看某个网页的正文answer(final_answer): 给出最终答案搜索策略建议第一轮搜索用宽泛的关键词了解全貌从搜索结果中发现具体线索后再用精确关键词深入注意检查信息的时效性和权威性如果3轮搜索后仍信息不足基于已有信息给出最佳回答并标注不确定性请以JSON格式回复包含{thinking: 你的分析思考过程,action: search | fetch | answer,params: {query/url/answer: 具体参数},confidence: 0.0-1.0 // 对当前答案的信心}messages [ {quot;rolequot;: quot;systemquot;, quot;contentquot;: system_prompt}, {quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;请研究以下问题{question}quot;} ] for round_num in range(self.max_search_rounds): # 让LLM决定下一步 response await self.llm.chat.completions.create( modelquot;deepseek-flashquot;, # 用Flash模型做调度成本低 messagesmessages, temperature0.3, # 低温度决策更稳定 response_format{quot;typequot;: quot;json_objectquot;} ) decision json.loads(response.choices[0].message.content) thinking decision.get(quot;thinkingquot;, quot;quot;) action decision.get(quot;actionquot;, quot;answerquot;) params decision.get(quot;paramsquot;, {}) search_log.append({ quot;roundquot;: round_num 1, quot;thinkingquot;: thinking, quot;actionquot;: action, quot;paramsquot;: params }) # 执行动作 if action quot;searchquot;: query params.get(quot;queryquot;, quot;quot;) if not query: break if not self.cost.record(0.005): search_log.append({quot;roundquot;: round_num 1, quot;errorquot;: quot;预算耗尽quot;}) break results await self.search.search_with_retry(query, count10) filtered filter_results(results) # 把搜索结果喂回给LLM search_summary quot;\nquot;.join([ fquot;[{i1}] {r[title]} (权威度:{r.get(authority_score, ?)}/4)\nquot; fquot; URL: {r[url]}\nquot; fquot; 摘要: {r.get(model_summary, r.get(snippet, ))[:200]}quot; for i, r in enumerate(filtered[:8]) ]) messages.append({ quot;rolequot;: quot;assistantquot;, quot;contentquot;: json.dumps(decision, ensure_asciiFalse) }) messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;搜索「{query}」返回了{len(filtered)}条相关结果\n{search_summary}\n\n请决定下一步继续搜索、查看某个网页、还是给出答案quot; }) elif action quot;fetchquot;: url params.get(quot;urlquot;, quot;quot;) if not url: break if not self.cost.record(0.005): break page await self.search.fetch_with_retry(url) if page: content page.get(quot;contentquot;, quot;quot;)[:4000] # 截断防止超token sources.append({quot;urlquot;: url, quot;titlequot;: page.get(quot;titlequot;, quot;quot;)}) messages.append({ quot;rolequot;: quot;assistantquot;, quot;contentquot;: json.dumps(decision, ensure_asciiFalse) }) messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;网页《{page.get(title, )}》正文截断\n{content}\n\n请基于这篇内容更新你的认知然后决定下一步。quot; }) else: messages.append({ quot;rolequot;: quot;assistantquot;, quot;contentquot;: json.dumps(decision, ensure_asciiFalse) }) messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;无法抓取该网页可能超时或被拒绝请换其他方式获取信息。quot; }) elif action quot;answerquot;: # Agent认为信息足够给出最终答案 final_answer params.get(quot;answerquot;, quot;无法回答quot;) confidence decision.get(quot;confidencequot;, 0) return { quot;answerquot;: final_answer, quot;sourcesquot;: sources, quot;search_logquot;: search_log, quot;total_costquot;: self.cost.today_cost - initial_cost, quot;confidencequot;: confidence, quot;rounds_usedquot;: round_num 1 } # 达到最大轮次强制总结 messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: quot;你已达到最大搜索轮次。请基于已收集的所有信息给出你目前能给出的最佳答案。如果信息不足请标注不确定的部分。quot; }) final_response await self.llm.chat.completions.create( modelquot;deepseek-flashquot;, messagesmessages, temperature0.3 ) return { quot;answerquot;: final_response.choices[0].message.content, quot;sourcesquot;: sources, quot;search_logquot;: search_log, quot;total_costquot;: self.cost.today_cost - initial_cost, quot;confidencequot;: 0.5, quot;rounds_usedquot;: self.max_search_rounds }实际测试3个场景的成本和质量对比光说代码没意义来看看实际效果。我用3个不同类型的问题做了测试对比方案是之前用的Tavily。测试1技术事实查询问题DeepSeek V4.1 Flash 1017相比上一版本有哪些具体改进指标TavilyDeepSeek Search搜索轮次22网页抓取33总成本¥0.25¥0.025回答准确度高高耗时8.2s6.1s测试2多维度对比分析问题2026年主流的AI Agent记忆方案有哪些各自的优缺点是什么指标TavilyDeepSeek Search搜索轮次43网页抓取54总成本¥0.45¥0.035回答完整度覆盖4种方案覆盖3种方案耗时15.3s11.7s测试3实时信息观点整合问题最近一周AI行业有什么重大事件对开发者有什么影响指标TavilyDeepSeek Search搜索轮次33网页抓取44总成本¥0.35¥0.035时效性好好来源多样性5个来源4个来源结论很明显在大多数场景下DeepSeek Search能达到Tavily 80-90%的信息质量但成本只有十分之一甚至更低。对于日均搜索量大的Agent应用来说这个差距会被放大。唯一的劣势是来源多样性略差——毕竟DeepSeek搜索引擎的索引覆盖和Tavily不完全一样。但对于中文内容的搜索质量DeepSeek反而更好这可能跟它的中文训练数据更充足有关。在舆情监控场景中的落地实践前面说了我最初的动力是帮团队降低舆情系统的搜索成本。这里简单分享一下落地方案。我们的舆情系统需要监控多个关键词组合的实时信息然后交给Agent判断是否构成需要关注的舆情事件。之前用Tavily每小时跑一轮一天24次×10个关键词组合光搜索就240次/天。成本可控但如果要增加监控频率或关键词费用就线性增长。切换到DeepSeek Search后同样的监控频率成本降了92%。更重要的是DeepSeek搜索返回的authority_score帮我们做了第一轮筛选——权威度低于2的结果直接过滤掉省掉了后面Agent的分析开销。这套方案的核心架构很简单async def monitor_keywords(keywords: list[str], pipeline: AgenticSearchPipeline): 关键词监控主循环每小时执行一轮每个关键词搜索最近24小时的相关信息 quot;quot;quot; while True: for keyword in keywords: # 构造带时间范围的搜索query today time.strftime(quot;%Y-%m-%dquot;) query fquot;{keyword} site:weibo.com OR site:zhihu.com OR site:36kr.comquot; results await pipeline.search.search_with_retry(query, count20) # 用authority_score做初筛 high_auth [r for r in results if r.get(quot;authority_scorequot;, 0) gt; 3] for r in high_auth[:5]: # 每条高分结果深入看 page await pipeline.search.fetch_with_retry(r[quot;urlquot;]) if page: # 交给舆情分析Agent判断 # ... 这里接入你的舆情分类逻辑 pass await asyncio.sleep(3600) # 1小时后下一轮类似这样的架构在像黑箭科技的舆安系统这类需要高频、大量搜索的场景中尤其能发挥成本优势。毕竟舆情监控是7×24不间断运行的搜索成本直接决定了你能监控多少关键词、跑多高的频率。一些实战建议最后总结几点我这几天的实操心得1. 用Flash模型做搜索调度别用ProAgent做搜索编排时决定搜什么这个任务本身不需要太强的推理能力。用DeepSeek Flash来做调度成本只有Pro的几十分之一效果差距不大。把Pro的算力留给最终的答案生成。2. Search的count参数设大一些既然返回1条和返回50条价格一样那就设20-30。多出来的结果你可以用authority_score和relevance_score在本地筛选反正不要钱。3. Fetch要有降级策略不是所有网页都能成功抓取。403、404、超时都很常见。建议的做法是如果Fetch失败退而使用Search返回的model_summary字段——那个摘要信息量其实比snippet丰富得多很多时候够用了。4. 注意数据合规DeepSeek的隐私政策提到数据存储在中华人民共和国境内。如果你的Agent处理的是敏感信息比如用户隐私数据、企业内部文档的关键词搜索之前要先评估合规风险。公开信息搜索没问题但别把内部数据当query搜出去。5. 缓存缓存缓存相同或相似的query结果短期内不会大变。加一层TTL缓存比如30分钟内同一query直接返回缓存结果能显著降低API调用量和成本。from functools import lru_cache import hashlibclass CachedSearch(RateLimitedSearch):带缓存的搜索客户端def __init__(self, *args, cache_ttl: int 1800, **kwargs): super().__init__(*args, **kwargs) self._cache: dict[str, tuple[float, list]] {} self.cache_ttl cache_ttl async def search_with_retry(self, query: str, count: int 10) -gt; list[dict]: cache_key hashlib.md5(fquot;{query}:{count}quot;.encode()).hexdigest() if cache_key in self._cache: ts, results self._cache[cache_key] if time.time() - ts lt; self.cache_ttl: return results results await super().search_with_retry(query, count) self._cache[cache_key] (time.time(), results) return results总结DeepSeek这次开放的搜索API核心优势就三个字便宜、干净、好用。便宜是指5元/千次的定价在同类产品中几乎是最低的干净是指它自带的反SEO和反AIGC垃圾机制给Agent返回的信息质量相对可靠好用是指Search Fetch的接口设计很克制两个端点覆盖了你需要的全部能力不用学一堆复杂参数。当然它也有局限5次/秒的限流对高并发场景是个瓶颈索引覆盖可能不如Google/Bing全面数据合规问题需要具体评估。但对于大部分Agent应用来说这套API已经足够撑起一个生产级的联网搜索方案了。如果你的Agent也有联网搜索的需求建议趁现在免费期到11月1日前通过Claude Code/Codex调用免计费先跑几轮测试感受一下效果和限制再决定要不要正式接入。作者黑箭科技技术团队声明本文为黑箭科技原创技术分享转载请注明出处。文中代码基于DeepSeek Search API公开文档编写接口规格以官方最新文档为准。如果你在Agent开发、AI应用架构方面有更多问题欢迎在评论区讨论交流。觉得有用的话点个赞收个藏后续我们还会分享更多AI Agent工程实践的干货。 
长河君 · 2026/10/30 9:49:39 · 阅读 1.2k 我用DeepSeek刚开放的搜索API给Agent装了个联网大脑：成本降了92%，附完整生产级代码
摘要DeepSeek昨天10月17日正式开放了自研搜索APISearch和Fetch两个接口均为5元/千次。本文从实际工程视角拆解这套API的调用方式、限流处理、多轮检索编排以及在生产环境中如何用它替代Tavily/Brave等高价方案给AI Agent装上低成本、高可靠的联网能力。附完整可运行的Python代码。为什么我盯上了这个API做Agent开发的朋友应该都有一个体感让模型联网这件事比你想象的要贵。之前我帮团队搭建舆情监控系统的时候给Agent接的是Tavily的搜索API。效果确实不错但月底一看账单——光是搜索这一项就花了小一万。要知道我们的调用量根本算不上大日均也就几百次。更头疼的是搜索结果质量参差不齐。有些返回的内容明显是SEO垃圾Agent拿到这些毒数据后生成的回答也跟着跑偏。你模型再强输入是垃圾输出也是垃圾——GIGO定律从来没失手过。所以昨天DeepSeek宣布开放自研搜索API的时候我第一时间就测了一轮。原因很简单价格确实离谱地便宜——5元/千次Tavily的八到十六分之一自带反垃圾机制——DeepSeek声称对SEO作弊、GEO操纵和低质AIGC有专门过滤两个接口分工明确——Search负责搜Fetch负责抓网页正文并转成Markdown对LLM非常友好但便宜归便宜生产环境能不能用限流怎么处理多步检索怎么编排这些问题不搞清楚你不敢往线上放。所以我花了两天时间做了一轮完整的工程验证把踩过的坑和最终的方案整理出来希望能帮到同样在做Agent联网能力的朋友。先搞清楚这套API到底给了你什么在动手写代码之前得先把接口规格吃透。DeepSeek搜索API一共就两个端点Search接口端点POST https://api.deepseek.com/v1/tools/search功能输入query返回结构化搜索结果列表参数query搜索关键词最长128字符count返回结果数1-50默认10返回字段title、url、snippet、model_summary给模型看的长摘要、可选full_text、publish_date、relevance_score0-3、authority_score0-4计费5元/千次成功请求空结果和失败不计费Fetch接口端点POST https://api.deepseek.com/v1/tools/fetch功能输入URL返回网页正文Markdown格式返回字段title、contentMarkdown格式的正文计费5元/千次成功请求失败不计费两个关键细节值得注意第一Search的结果数不影响价格。你请求1条和请求50条价格一样。这意味着在不确定哪条结果有用的时候大方一点多拿几条反而更划算——反正不用额外花钱。第二两个接口的限流是独立的各自5次/秒/账号。Fetch还有一个按域名的额外限制具体数值官方没公布。这个在后面做并发控制时是个坑先记住。认证方式和DeepSeek的Chat API完全一样用同一个Bearer Token不需要额外申请。基础调用5分钟跑通第一个搜索先写最基础的调用代码把链路跑通import httpx import asyncio from typing import Optionalclass DeepSeekSearch:DeepSeek搜索API基础客户端BASE_URL quot;https://api.deepseek.com/v1/toolsquot; def __init__(self, api_key: str): self.api_key api_key self.client httpx.AsyncClient( headers{ quot;Authorizationquot;: fquot;Bearer {api_key}quot;, quot;Content-Typequot;: quot;application/jsonquot; }, timeout30.0 ) async def search(self, query: str, count: int 10) -gt; list[dict]: quot;quot;quot; 执行网页搜索 Args: query: 搜索关键词最长128字符 count: 返回结果数量1-50 Returns: 搜索结果列表每条包含title/url/snippet等字段 quot;quot;quot; if len(query) gt; 128: query query[:128] resp await self.client.post( fquot;{self.BASE_URL}/searchquot;, json{quot;queryquot;: query, quot;countquot;: count} ) resp.raise_for_status() data resp.json() return data.get(quot;resultsquot;, []) async def fetch_page(self, url: str) -gt; Optional[dict]: quot;quot;quot; 抓取网页正文返回Markdown格式 Args: url: 目标网页URL Returns: 包含title和content的字典失败返回None quot;quot;quot; resp await self.client.post( fquot;{self.BASE_URL}/fetchquot;, json{quot;urlquot;: url} ) if resp.status_code ! 200: return None data resp.json() if not data.get(quot;contentquot;): return None return data async def close(self): await self.client.aclose()快速验证async def main():searcher DeepSeekSearch(api_keyyour-api-key-here)results await searcher.search(quot;AI Agent框架 2026quot;, count5) for r in results: print(fquot;[{r.get(authority_score, 0)}] {r[title]}quot;) print(fquot; {r[url]}quot;) print(fquot; {r.get(model_summary, r.get(snippet, ))[:100]}...quot;) print() await searcher.close()asyncio.run(main())跑一下就能看到结果了。注意返回的authority_score字段——这是DeepSeek搜索特有的权威度评分0-4分分数越高说明来源越可信。后面做结果筛选的时候会用到。生产级改造限流、重试、成本控制的三道防线基础调用跑通之后别急着往生产环境放。有几个坑我替你踩了坑15次/秒的限流比你想象的更容易触发看起来5次/秒不算少但Agent一旦开始自主决策很容易在一个推理步骤里同时发起多个搜索。比如用户问一个复杂问题Agent可能同时搜索3个不同角度的关键词再Fetch 2个网页——这就5次了刚好顶到上限。更糟的是Fetch还有按域名的额外限制。如果你的Agent连续抓取同一个网站的多个页面可能2-3次就被429了。解决方案令牌桶限流 指数退避重试import time import asyncio from dataclasses import dataclass from collections import defaultdictdataclassclass RateLimitConfig:限流配置search_qps: float 4.0 # 留1个余量设4而不是5fetch_qps: float 4.0fetch_per_domain_qps: float 2.0 # 单域名更保守max_retries: int 3retry_base_delay: float 1.0class RateLimitedSearch(DeepSeekSearch):带限流控制的搜索客户端def __init__(self, api_key: str, config: Optional[RateLimitConfig] None): super().__init__(api_key) self.config config or RateLimitConfig() # 令牌桶记录每个维度的上次请求时间 self._last_search_time 0.0 self._last_fetch_time 0.0 self._domain_fetch_times: dict[str, list[float]] defaultdict(list) # 统计信息 self.stats { quot;search_callsquot;: 0, quot;fetch_callsquot;: 0, quot;retriesquot;: 0, quot;rate_limitedquot;: 0, quot;total_cost_yuanquot;: 0.0 # 估算成本 } async def _wait_for_slot(self, bucket: str, interval: float): quot;quot;quot;等待令牌桶空出位置quot;quot;quot; now time.monotonic() elapsed now - getattr(self, fquot;_last_{bucket}_timequot;, 0) if elapsed lt; interval: await asyncio.sleep(interval - elapsed) setattr(self, fquot;_last_{bucket}_timequot;, time.monotonic()) async def _wait_for_domain_slot(self, domain: str): quot;quot;quot;单域名的额外限流quot;quot;quot; now time.monotonic() # 清理超过1秒的记录 self._domain_fetch_times[domain] [ t for t in self._domain_fetch_times[domain] if now - t lt; 1.0 ] if len(self._domain_fetch_times[domain]) gt; int(self.config.fetch_per_domain_qps): oldest self._domain_fetch_times[domain][0] await asyncio.sleep(1.0 - (now - oldest)) self._domain_fetch_times[domain].append(time.monotonic()) async def search_with_retry(self, query: str, count: int 10) -gt; list[dict]: quot;quot;quot;带重试的搜索quot;quot;quot; await self._wait_for_slot(quot;searchquot;, 1.0 / self.config.search_qps) for attempt in range(self.config.max_retries): try: results await self.search(query, count) self.stats[quot;search_callsquot;] 1 if results: # 只有成功返回结果才计费 self.stats[quot;total_cost_yuanquot;] 0.005 return results except httpx.HTTPStatusError as e: if e.response.status_code 429: self.stats[quot;rate_limitedquot;] 1 delay self.config.retry_base_delay * (2 ** attempt) await asyncio.sleep(delay) self.stats[quot;retriesquot;] 1 else: raise return [] # 重试耗尽返回空 async def fetch_with_retry(self, url: str) -gt; Optional[dict]: quot;quot;quot;带重试的网页抓取quot;quot;quot; from urllib.parse import urlparse domain urlparse(url).netloc await self._wait_for_slot(quot;fetchquot;, 1.0 / self.config.fetch_qps) await self._wait_for_domain_slot(domain) for attempt in range(self.config.max_retries): try: result await self.fetch_page(url) self.stats[quot;fetch_callsquot;] 1 if result: self.stats[quot;total_cost_yuanquot;] 0.005 return result except httpx.HTTPStatusError as e: if e.response.status_code in (429, 503): self.stats[quot;rate_limitedquot;] 1 delay self.config.retry_base_delay * (2 ** attempt) await asyncio.sleep(delay) self.stats[quot;retriesquot;] 1 elif e.response.status_code in (403, 404): # 这些不是临时错误不需要重试 return None else: raise return None坑2搜索结果不等于可用信息DeepSeek搜索返回的结果有relevance_score和authority_score两个评分但这只是搜索引擎的排序信号不等于你的Agent能直接拿来用。实测下来authority_score 3的结果质量明显高于低分的。我的建议是默认只取高权威度结果低分的交给Agent决定是否要深入看。def filter_results(results: list[dict], min_authority: int 2, min_relevance: float 1.5) - list[dict]: 筛选高质量搜索结果根据实测authority_score gt; 3 的结果引用可信度较高 relevance_score gt; 1.5 通常意味着核心相关而非边缘提及。 quot;quot;quot; filtered [] for r in results: auth r.get(quot;authority_scorequot;, 0) rel r.get(quot;relevance_scorequot;, 0) if auth gt; min_authority and rel gt; min_relevance: filtered.append(r) # 如果筛选后结果太少降低阈值再放一轮 if len(filtered) lt; 2 and results: filtered [r for r in results if r.get(quot;authority_scorequot;, 0) gt; 1 and r.get(quot;relevance_scorequot;, 0) gt; 1.0] return sorted(filtered, keylambda x: (x.get(quot;authority_scorequot;, 0), x.get(quot;relevance_scorequot;, 0)), reverseTrue)坑3成本监控不能事后才看5元/千次听起来便宜但如果你的Agent在循环里疯狂搜索一个月下来也不是小数目。关键是空结果不计费但每次API调用都消耗时间。如果Agent每轮推理都搜索3-5次响应时间就上去了。class CostTracker: 成本追踪器def __init__(self, monthly_budget_yuan: float 150.0): self.monthly_budget monthly_budget_yuan self.daily_budget monthly_budget_yuan / 30 self.today_cost 0.0 self.today_date time.strftime(quot;%Y-%m-%dquot;) def record(self, cost_yuan: float) -gt; bool: quot;quot;quot;记录一笔花费返回是否允许继续quot;quot;quot; today time.strftime(quot;%Y-%m-%dquot;) if today ! self.today_date: self.today_cost 0.0 self.today_date today self.today_cost cost_yuan return self.today_cost lt; self.daily_budget property def remaining_ratio(self) -gt; float: quot;quot;quot;今日预算剩余比例quot;quot;quot; return max(0, 1 - self.today_cost / self.daily_budget)核心多步检索编排——让Agent学会搜了再搜前面都是基础设施。真正让这套搜索能力变强的是多步检索编排。什么意思就是一个复杂问题Agent不是搜一次就完事了。它需要先搜一个宽泛的问题了解大方向从搜索结果里发现新的关键词或线索用新线索再搜一轮缩小范围抓到具体网页后提取关键信息如果发现信息不够换个角度再搜这个过程和你自己上网查资料其实是一样的。关键是怎么用代码编排出来。import json from openai import AsyncOpenAIclass AgenticSearchPipeline:Agent驱动的多步搜索管线核心思路让LLM自己决定搜什么、什么时候停、要不要深入看某个网页 quot;quot;quot; def __init__(self, api_key: str, llm_client: AsyncOpenAI, search_client: RateLimitedSearch, cost_tracker: CostTracker): self.llm llm_client self.search search_client self.cost cost_tracker self.max_search_rounds 5 # 防止无限循环 async def research(self, question: str) -gt; dict: quot;quot;quot; 对一个问题执行多步搜索研究 Returns: { quot;answerquot;: str, # 最终答案 quot;sourcesquot;: list, # 引用的来源 quot;search_logquot;: list, # 搜索过程日志 quot;total_costquot;: float # 本次研究总成本 } quot;quot;quot; search_log [] collected_facts [] sources [] initial_cost self.cost.today_cost # 系统提示教Agent如何搜索 system_prompt quot;quot;quot;你是一个研究型AI助手。你的任务是通过多步搜索回答用户的复杂问题。每一步你需要决定是否需要搜索更多信息如果需要给出搜索关键词是否需要深入查看某个已发现的网页给出URL是否已经有足够信息可以给出最终答案你的可用工具search(query): 搜索网页fetch(url): 查看某个网页的正文answer(final_answer): 给出最终答案搜索策略建议第一轮搜索用宽泛的关键词了解全貌从搜索结果中发现具体线索后再用精确关键词深入注意检查信息的时效性和权威性如果3轮搜索后仍信息不足基于已有信息给出最佳回答并标注不确定性请以JSON格式回复包含{thinking: 你的分析思考过程,action: search | fetch | answer,params: {query/url/answer: 具体参数},confidence: 0.0-1.0 // 对当前答案的信心}messages [ {quot;rolequot;: quot;systemquot;, quot;contentquot;: system_prompt}, {quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;请研究以下问题{question}quot;} ] for round_num in range(self.max_search_rounds): # 让LLM决定下一步 response await self.llm.chat.completions.create( modelquot;deepseek-flashquot;, # 用Flash模型做调度成本低 messagesmessages, temperature0.3, # 低温度决策更稳定 response_format{quot;typequot;: quot;json_objectquot;} ) decision json.loads(response.choices[0].message.content) thinking decision.get(quot;thinkingquot;, quot;quot;) action decision.get(quot;actionquot;, quot;answerquot;) params decision.get(quot;paramsquot;, {}) search_log.append({ quot;roundquot;: round_num 1, quot;thinkingquot;: thinking, quot;actionquot;: action, quot;paramsquot;: params }) # 执行动作 if action quot;searchquot;: query params.get(quot;queryquot;, quot;quot;) if not query: break if not self.cost.record(0.005): search_log.append({quot;roundquot;: round_num 1, quot;errorquot;: quot;预算耗尽quot;}) break results await self.search.search_with_retry(query, count10) filtered filter_results(results) # 把搜索结果喂回给LLM search_summary quot;\nquot;.join([ fquot;[{i1}] {r[title]} (权威度:{r.get(authority_score, ?)}/4)\nquot; fquot; URL: {r[url]}\nquot; fquot; 摘要: {r.get(model_summary, r.get(snippet, ))[:200]}quot; for i, r in enumerate(filtered[:8]) ]) messages.append({ quot;rolequot;: quot;assistantquot;, quot;contentquot;: json.dumps(decision, ensure_asciiFalse) }) messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;搜索「{query}」返回了{len(filtered)}条相关结果\n{search_summary}\n\n请决定下一步继续搜索、查看某个网页、还是给出答案quot; }) elif action quot;fetchquot;: url params.get(quot;urlquot;, quot;quot;) if not url: break if not self.cost.record(0.005): break page await self.search.fetch_with_retry(url) if page: content page.get(quot;contentquot;, quot;quot;)[:4000] # 截断防止超token sources.append({quot;urlquot;: url, quot;titlequot;: page.get(quot;titlequot;, quot;quot;)}) messages.append({ quot;rolequot;: quot;assistantquot;, quot;contentquot;: json.dumps(decision, ensure_asciiFalse) }) messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;网页《{page.get(title, )}》正文截断\n{content}\n\n请基于这篇内容更新你的认知然后决定下一步。quot; }) else: messages.append({ quot;rolequot;: quot;assistantquot;, quot;contentquot;: json.dumps(decision, ensure_asciiFalse) }) messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: fquot;无法抓取该网页可能超时或被拒绝请换其他方式获取信息。quot; }) elif action quot;answerquot;: # Agent认为信息足够给出最终答案 final_answer params.get(quot;answerquot;, quot;无法回答quot;) confidence decision.get(quot;confidencequot;, 0) return { quot;answerquot;: final_answer, quot;sourcesquot;: sources, quot;search_logquot;: search_log, quot;total_costquot;: self.cost.today_cost - initial_cost, quot;confidencequot;: confidence, quot;rounds_usedquot;: round_num 1 } # 达到最大轮次强制总结 messages.append({ quot;rolequot;: quot;userquot;, quot;contentquot;: quot;你已达到最大搜索轮次。请基于已收集的所有信息给出你目前能给出的最佳答案。如果信息不足请标注不确定的部分。quot; }) final_response await self.llm.chat.completions.create( modelquot;deepseek-flashquot;, messagesmessages, temperature0.3 ) return { quot;answerquot;: final_response.choices[0].message.content, quot;sourcesquot;: sources, quot;search_logquot;: search_log, quot;total_costquot;: self.cost.today_cost - initial_cost, quot;confidencequot;: 0.5, quot;rounds_usedquot;: self.max_search_rounds }实际测试3个场景的成本和质量对比光说代码没意义来看看实际效果。我用3个不同类型的问题做了测试对比方案是之前用的Tavily。测试1技术事实查询问题DeepSeek V4.1 Flash 1017相比上一版本有哪些具体改进指标TavilyDeepSeek Search搜索轮次22网页抓取33总成本¥0.25¥0.025回答准确度高高耗时8.2s6.1s测试2多维度对比分析问题2026年主流的AI Agent记忆方案有哪些各自的优缺点是什么指标TavilyDeepSeek Search搜索轮次43网页抓取54总成本¥0.45¥0.035回答完整度覆盖4种方案覆盖3种方案耗时15.3s11.7s测试3实时信息观点整合问题最近一周AI行业有什么重大事件对开发者有什么影响指标TavilyDeepSeek Search搜索轮次33网页抓取44总成本¥0.35¥0.035时效性好好来源多样性5个来源4个来源结论很明显在大多数场景下DeepSeek Search能达到Tavily 80-90%的信息质量但成本只有十分之一甚至更低。对于日均搜索量大的Agent应用来说这个差距会被放大。唯一的劣势是来源多样性略差——毕竟DeepSeek搜索引擎的索引覆盖和Tavily不完全一样。但对于中文内容的搜索质量DeepSeek反而更好这可能跟它的中文训练数据更充足有关。在舆情监控场景中的落地实践前面说了我最初的动力是帮团队降低舆情系统的搜索成本。这里简单分享一下落地方案。我们的舆情系统需要监控多个关键词组合的实时信息然后交给Agent判断是否构成需要关注的舆情事件。之前用Tavily每小时跑一轮一天24次×10个关键词组合光搜索就240次/天。成本可控但如果要增加监控频率或关键词费用就线性增长。切换到DeepSeek Search后同样的监控频率成本降了92%。更重要的是DeepSeek搜索返回的authority_score帮我们做了第一轮筛选——权威度低于2的结果直接过滤掉省掉了后面Agent的分析开销。这套方案的核心架构很简单async def monitor_keywords(keywords: list[str], pipeline: AgenticSearchPipeline): 关键词监控主循环每小时执行一轮每个关键词搜索最近24小时的相关信息 quot;quot;quot; while True: for keyword in keywords: # 构造带时间范围的搜索query today time.strftime(quot;%Y-%m-%dquot;) query fquot;{keyword} site:weibo.com OR site:zhihu.com OR site:36kr.comquot; results await pipeline.search.search_with_retry(query, count20) # 用authority_score做初筛 high_auth [r for r in results if r.get(quot;authority_scorequot;, 0) gt; 3] for r in high_auth[:5]: # 每条高分结果深入看 page await pipeline.search.fetch_with_retry(r[quot;urlquot;]) if page: # 交给舆情分析Agent判断 # ... 这里接入你的舆情分类逻辑 pass await asyncio.sleep(3600) # 1小时后下一轮类似这样的架构在像黑箭科技的舆安系统这类需要高频、大量搜索的场景中尤其能发挥成本优势。毕竟舆情监控是7×24不间断运行的搜索成本直接决定了你能监控多少关键词、跑多高的频率。一些实战建议最后总结几点我这几天的实操心得1. 用Flash模型做搜索调度别用ProAgent做搜索编排时决定搜什么这个任务本身不需要太强的推理能力。用DeepSeek Flash来做调度成本只有Pro的几十分之一效果差距不大。把Pro的算力留给最终的答案生成。2. Search的count参数设大一些既然返回1条和返回50条价格一样那就设20-30。多出来的结果你可以用authority_score和relevance_score在本地筛选反正不要钱。3. Fetch要有降级策略不是所有网页都能成功抓取。403、404、超时都很常见。建议的做法是如果Fetch失败退而使用Search返回的model_summary字段——那个摘要信息量其实比snippet丰富得多很多时候够用了。4. 注意数据合规DeepSeek的隐私政策提到数据存储在中华人民共和国境内。如果你的Agent处理的是敏感信息比如用户隐私数据、企业内部文档的关键词搜索之前要先评估合规风险。公开信息搜索没问题但别把内部数据当query搜出去。5. 缓存缓存缓存相同或相似的query结果短期内不会大变。加一层TTL缓存比如30分钟内同一query直接返回缓存结果能显著降低API调用量和成本。from functools import lru_cache import hashlibclass CachedSearch(RateLimitedSearch):带缓存的搜索客户端def __init__(self, *args, cache_ttl: int 1800, **kwargs): super().__init__(*args, **kwargs) self._cache: dict[str, tuple[float, list]] {} self.cache_ttl cache_ttl async def search_with_retry(self, query: str, count: int 10) -gt; list[dict]: cache_key hashlib.md5(fquot;{query}:{count}quot;.encode()).hexdigest() if cache_key in self._cache: ts, results self._cache[cache_key] if time.time() - ts lt; self.cache_ttl: return results results await super().search_with_retry(query, count) self._cache[cache_key] (time.time(), results) return results总结DeepSeek这次开放的搜索API核心优势就三个字便宜、干净、好用。便宜是指5元/千次的定价在同类产品中几乎是最低的干净是指它自带的反SEO和反AIGC垃圾机制给Agent返回的信息质量相对可靠好用是指Search Fetch的接口设计很克制两个端点覆盖了你需要的全部能力不用学一堆复杂参数。当然它也有局限5次/秒的限流对高并发场景是个瓶颈索引覆盖可能不如Google/Bing全面数据合规问题需要具体评估。但对于大部分Agent应用来说这套API已经足够撑起一个生产级的联网搜索方案了。如果你的Agent也有联网搜索的需求建议趁现在免费期到11月1日前通过Claude Code/Codex调用免计费先跑几轮测试感受一下效果和限制再决定要不要正式接入。作者黑箭科技技术团队声明本文为黑箭科技原创技术分享转载请注明出处。文中代码基于DeepSeek Search API公开文档编写接口规格以官方最新文档为准。如果你在Agent开发、AI应用架构方面有更多问题欢迎在评论区讨论交流。觉得有用的话点个赞收个藏后续我们还会分享更多AI Agent工程实践的干货。