import httpx import asyncio from typing import Optional

class DeepSeekSearch:
"""DeepSeek搜索API基础客户端"""

BASE_URL = "https://api.deepseek.com/v1/tools" def __init__(self, api_key: str): self.api_key = api_key self.client = httpx.AsyncClient( headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, timeout=30.0 ) async def search(self, query: str, count: int = 10) -> list[dict]: """ 执行网页搜索 Args: query: 搜索关键词，最长128字符 count: 返回结果数量，1-50 Returns: 搜索结果列表，每条包含title/url/snippet等字段 """ if len(query) > 128: query = query[:128] resp = await self.client.post( f"{self.BASE_URL}/search", json={"query": query, "count": count} ) resp.raise_for_status() data = resp.json() return data.get("results", []) async def fetch_page(self, url: str) -> Optional[dict]: """ 抓取网页正文，返回Markdown格式 Args: url: 目标网页URL Returns: 包含title和content的字典，失败返回None """ resp = await self.client.post( f"{self.BASE_URL}/fetch", json={"url": url} ) if resp.status_code != 200: return None data = resp.json() if not data.get("content"): return None return data async def close(self): await self.client.aclose()

快速验证

async def main():
searcher = DeepSeekSearch(api_key="your-api-key-here")

results = await searcher.search("AI Agent框架 2026", count=5) for r in results: print(f"[{r.get('authority_score', 0)}] {r['title']}") print(f" {r['url']}") print(f" {r.get('model_summary', r.get('snippet', ''))[:100]}...") print() await searcher.close()

asyncio.run(main())

跑一下就能看到结果了。注意返回的authority_score字段——这是DeepSeek搜索特有的权威度评分，0-4分，分数越高说明来源越可信。后面做结果筛选的时候会用到。

生产级改造：限流、重试、成本控制的三道防线

基础调用跑通之后，别急着往生产环境放。有几个坑我替你踩了：

坑1：5次/秒的限流比你想象的更容易触发

看起来5次/秒不算少，但Agent一旦开始"自主决策"，很容易在一个推理步骤里同时发起多个搜索。比如用户问一个复杂问题，Agent可能同时搜索3个不同角度的关键词，再Fetch 2个网页——这就5次了，刚好顶到上限。

更糟的是，Fetch还有按域名的额外限制。如果你的Agent连续抓取同一个网站的多个页面，可能2-3次就被429了。

解决方案：令牌桶限流 + 指数退避重试

import time import asyncio from dataclasses import dataclass from collections import defaultdict

@dataclass
class RateLimitConfig:
"""限流配置"""
search_qps: float = 4.0 # 留1个余量，设4而不是5
fetch_qps: float = 4.0
fetch_per_domain_qps: float = 2.0 # 单域名更保守
max_retries: int = 3
retry_base_delay: float = 1.0

class 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 = { "search_calls": 0, "fetch_calls": 0, "retries": 0, "rate_limited": 0, "total_cost_yuan": 0.0 # 估算成本 } async def _wait_for_slot(self, bucket: str, interval: float): """等待令牌桶空出位置""" now = time.monotonic() elapsed = now - getattr(self, f"_last_{bucket}_time", 0) if elapsed < interval: await asyncio.sleep(interval - elapsed) setattr(self, f"_last_{bucket}_time", time.monotonic()) async def _wait_for_domain_slot(self, domain: str): """单域名的额外限流""" now = time.monotonic() # 清理超过1秒的记录 self._domain_fetch_times[domain] = [ t for t in self._domain_fetch_times[domain] if now - t < 1.0 ] if len(self._domain_fetch_times[domain]) >= 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) -> list[dict]: """带重试的搜索""" await self._wait_for_slot("search", 1.0 / self.config.search_qps) for attempt in range(self.config.max_retries): try: results = await self.search(query, count) self.stats["search_calls"] += 1 if results: # 只有成功返回结果才计费 self.stats["total_cost_yuan"] += 0.005 return results except httpx.HTTPStatusError as e: if e.response.status_code == 429: self.stats["rate_limited"] += 1 delay = self.config.retry_base_delay * (2 ** attempt) await asyncio.sleep(delay) self.stats["retries"] += 1 else: raise return [] # 重试耗尽，返回空 async def fetch_with_retry(self, url: str) -> Optional[dict]: """带重试的网页抓取""" from urllib.parse import urlparse domain = urlparse(url).netloc await self._wait_for_slot("fetch", 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["fetch_calls"] += 1 if result: self.stats["total_cost_yuan"] += 0.005 return result except httpx.HTTPStatusError as e: if e.response.status_code in (429, 503): self.stats["rate_limited"] += 1 delay = self.config.retry_base_delay * (2 ** attempt) await asyncio.sleep(delay) self.stats["retries"] += 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 >= 3 的结果引用可信度较高， relevance_score >= 1.5 通常意味着核心相关而非边缘提及。 """ filtered = [] for r in results: auth = r.get("authority_score", 0) rel = r.get("relevance_score", 0) if auth >= min_authority and rel >= min_relevance: filtered.append(r) # 如果筛选后结果太少，降低阈值再放一轮 if len(filtered) < 2 and results: filtered = [r for r in results if r.get("authority_score", 0) >= 1 and r.get("relevance_score", 0) >= 1.0] return sorted(filtered, key=lambda x: (x.get("authority_score", 0), x.get("relevance_score", 0)), reverse=True)

坑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("%Y-%m-%d") def record(self, cost_yuan: float) -> bool: """记录一笔花费，返回是否允许继续""" today = time.strftime("%Y-%m-%d") if today != self.today_date: self.today_cost = 0.0 self.today_date = today self.today_cost += cost_yuan return self.today_cost <= self.daily_budget @property def remaining_ratio(self) -> float: """今日预算剩余比例""" return max(0, 1 - self.today_cost / self.daily_budget)

核心：多步检索编排——让Agent学会"搜了再搜"

前面都是基础设施。真正让这套搜索能力变强的，是多步检索编排。

什么意思？就是一个复杂问题，Agent不是搜一次就完事了。它需要：

1. 先搜一个宽泛的问题，了解大方向
2. 从搜索结果里发现新的关键词或线索
3. 用新线索再搜一轮，缩小范围
4. 抓到具体网页后，提取关键信息
5. 如果发现信息不够，换个角度再搜

这个过程，和你自己上网查资料其实是一样的。关键是怎么用代码编排出来。

import json from openai import AsyncOpenAI

class AgenticSearchPipeline:
"""
Agent驱动的多步搜索管线

核心思路：让LLM自己决定搜什么、什么时候停、要不要深入看某个网页 """ 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) -> dict: """ 对一个问题执行多步搜索研究 Returns: { "answer": str, # 最终答案 "sources": list, # 引用的来源 "search_log": list, # 搜索过程日志 "total_cost": float # 本次研究总成本 } """ search_log = [] collected_facts = [] sources = [] initial_cost = self.cost.today_cost # 系统提示：教Agent如何搜索 system_prompt = """你是一个研究型AI助手。你的任务是通过多步搜索回答用户的复杂问题。

每一步你需要决定：

1. 是否需要搜索更多信息（如果需要，给出搜索关键词）
2. 是否需要深入查看某个已发现的网页（给出URL）
3. 是否已经有足够信息可以给出最终答案

你的可用工具：

• search(query): 搜索网页
• fetch(url): 查看某个网页的正文
• answer(final_answer): 给出最终答案

搜索策略建议：

• 第一轮搜索用宽泛的关键词了解全貌
• 从搜索结果中发现具体线索后，再用精确关键词深入
• 注意检查信息的时效性和权威性
• 如果3轮搜索后仍信息不足，基于已有信息给出最佳回答并标注不确定性

请以JSON格式回复，包含：
{
"thinking": "你的分析思考过程",
"action": "search" | "fetch" | "answer",
"params": {"query/url/answer": "具体参数"},
"confidence": 0.0-1.0 // 对当前答案的信心
}"""

messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"请研究以下问题：{question}"} ] for round_num in range(self.max_search_rounds): # 让LLM决定下一步 response = await self.llm.chat.completions.create( model="deepseek-flash", # 用Flash模型做调度，成本低 messages=messages, temperature=0.3, # 低温度，决策更稳定 response_format={"type": "json_object"} ) decision = json.loads(response.choices[0].message.content) thinking = decision.get("thinking", "") action = decision.get("action", "answer") params = decision.get("params", {}) search_log.append({ "round": round_num + 1, "thinking": thinking, "action": action, "params": params }) # 执行动作 if action == "search": query = params.get("query", "") if not query: break if not self.cost.record(0.005): search_log.append({"round": round_num + 1, "error": "预算耗尽"}) break results = await self.search.search_with_retry(query, count=10) filtered = filter_results(results) # 把搜索结果喂回给LLM search_summary = "\n".join([ f"[{i+1}] {r['title']} (权威度:{r.get('authority_score', '?')}/4)\n" f" URL: {r['url']}\n" f" 摘要: {r.get('model_summary', r.get('snippet', ''))[:200]}" for i, r in enumerate(filtered[:8]) ]) messages.append({ "role": "assistant", "content": json.dumps(decision, ensure_ascii=False) }) messages.append({ "role": "user", "content": f"搜索「{query}」返回了{len(filtered)}条相关结果：\n{search_summary}\n\n请决定下一步：继续搜索、查看某个网页、还是给出答案？" }) elif action == "fetch": url = params.get("url", "") 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("content", "")[:4000] # 截断防止超token sources.append({"url": url, "title": page.get("title", "")}) messages.append({ "role": "assistant", "content": json.dumps(decision, ensure_ascii=False) }) messages.append({ "role": "user", "content": f"网页《{page.get('title', '')}》正文（截断）：\n{content}\n\n请基于这篇内容更新你的认知，然后决定下一步。" }) else: messages.append({ "role": "assistant", "content": json.dumps(decision, ensure_ascii=False) }) messages.append({ "role": "user", "content": f"无法抓取该网页（可能超时或被拒绝），请换其他方式获取信息。" }) elif action == "answer": # Agent认为信息足够，给出最终答案 final_answer = params.get("answer", "无法回答") confidence = decision.get("confidence", 0) return { "answer": final_answer, "sources": sources, "search_log": search_log, "total_cost": self.cost.today_cost - initial_cost, "confidence": confidence, "rounds_used": round_num + 1 } # 达到最大轮次，强制总结 messages.append({ "role": "user", "content": "你已达到最大搜索轮次。请基于已收集的所有信息，给出你目前能给出的最佳答案。如果信息不足，请标注不确定的部分。" }) final_response = await self.llm.chat.completions.create( model="deepseek-flash", messages=messages, temperature=0.3 ) return { "answer": final_response.choices[0].message.content, "sources": sources, "search_log": search_log, "total_cost": self.cost.today_cost - initial_cost, "confidence": 0.5, "rounds_used": self.max_search_rounds }

实际测试：3个场景的成本和质量对比

光说代码没意义，来看看实际效果。我用3个不同类型的问题做了测试，对比方案是之前用的Tavily。

测试1：技术事实查询

问题：DeepSeek V4.1 Flash 1017相比上一版本有哪些具体改进？

测试2：多维度对比分析

问题：2026年主流的AI Agent记忆方案有哪些？各自的优缺点是什么？

测试3：实时信息+观点整合

问题：最近一周AI行业有什么重大事件？对开发者有什么影响？

结论很明显：在大多数场景下，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小时的相关信息 """ while True: for keyword in keywords: # 构造带时间范围的搜索query today = time.strftime("%Y-%m-%d") query = f"{keyword} site:weibo.com OR site:zhihu.com OR site:36kr.com" results = await pipeline.search.search_with_retry(query, count=20) # 用authority_score做初筛 high_auth = [r for r in results if r.get("authority_score", 0) >= 3] for r in high_auth[:5]: # 每条高分结果深入看 page = await pipeline.search.fetch_with_retry(r["url"]) if page: # 交给舆情分析Agent判断 # ... 这里接入你的舆情分类逻辑 pass await asyncio.sleep(3600) # 1小时后下一轮

类似这样的架构，在像黑箭科技的舆安系统这类需要高频、大量搜索的场景中尤其能发挥成本优势。毕竟舆情监控是7×24不间断运行的，搜索成本直接决定了你能监控多少关键词、跑多高的频率。

一些实战建议

最后总结几点我这几天的实操心得：

1. 用Flash模型做搜索调度，别用Pro

Agent做搜索编排时，"决定搜什么"这个任务本身不需要太强的推理能力。用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 hashlib

class 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) -> list[dict]: cache_key = hashlib.md5(f"{query}:{count}".encode()).hexdigest() if cache_key in self._cache: ts, results = self._cache[cache_key] if time.time() - ts < 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工程实践的干货。