智能体在回答「有没有人踩过把搜索结果整段塞进上下文、结果模型跑偏的坑」这类问题时,靠关键词搜索很难命中,因为用户的说法和文章里的说法往往差着好几个词。语义搜索解决的正是这件事:把一句话的意思拿去匹配网页,而不是拿字面去匹配。这篇教程带你从零把 Exa 接成一个智能体可调用的工具,输入一句人话问题,输出一组去重后的网页,每条带标题、链接、发布时间,以及和问题相关的那几段正文。
先看最终产物长什么样。调用一次工具,拿回的结构大致如下:
```json
{
"query": "把搜索结果整段塞进上下文导致模型跑偏",
"results": [
{
"title": "Context engineering for retrieval agents",
"url": "https://example.com/posts/context-engineering",
"published": "2024-11-02",
"highlights": ["……把 top-20 全文拼进去之后,模型开始引用不相关段落……"]
}
],
"billing": { "search_calls": 1, "content_urls": 3, "cached": false }
}
```
整条链路有六个环节:鉴权、查询改写、搜索、按需抓正文、去重、计费控制,最后再包一层工具描述交给模型。Exa 的接口是标准的 REST 风格,一个 key 打通所有端点;本文不绑定某个具体小版本,端点参数名和计费口径以官方文档当前版本为准。
前置条件清单
动手前先确认这几项:
- 一个 Exa 账号,并在控制台创建一个 API key。免费额度、单价与各端点的计费方式以官方定价页为准,不要凭记忆估算成本。
- Python 3.9 及以上(以本地实际环境为准),装好
requests。用httpx也可以,差别只在写法。 - 机器能正常访问外网,公司网络需要配置代理的提前配好。
- 会用环境变量或
.env管理密钥,不要把 key 写进代码再提交到仓库。 - 可选:一个 LLM 接口,用于查询改写。没有也能跑,用规则改写做降级。
步骤一:鉴权并打通第一个请求
Exa 的鉴权头是 x-api-key,不是常见的 Authorization: Bearer。这一点写错会直接拿到 401,后面所有环节都无从谈起。先用 curl 验证 key 有效:
```bash
export EXA_API_KEY="你的 key"
curl -s https://api.exa.ai/search \
-H "Content-Type: application/json" \
-H "x-api-key: $EXA_API_KEY" \
-d '{"query":"semantic search for ai agents","numResults":3}'
```
返回是 JSON,results 是数组,每项至少有 title、url、publishedDate 之类的字段。字段名以实际返回为准,建议第一次先把原始 JSON 完整打印出来看一遍,别凭印象写解析代码。
接着封装一个统一的请求函数,后面所有端点都复用它:
```python
import os
import requests
BASE = "https://api.exa.ai"
HEADERS = {
"Content-Type": "application/json",
"x-api-key": os.environ["EXA_API_KEY"],
}
def exa_post(path: str, payload: dict, timeout: int = 20) -> dict:
resp = requests.post(BASE + path, json=payload, headers=HEADERS, timeout=timeout)
resp.raise_for_status()
return resp.json()
```
把超时显式写出来很重要。语义检索偶发慢请求,不设超时的话智能体主线程会一直挂着。
步骤二:查询改写,别把用户原话直接丢给检索
用户的原话通常带指代、口语和模糊时间:「帮我看看最近大家都在聊啥关于上下文太长的问题」。直接拿这句去检索,召回质量会明显下降。检索前先做一次改写,目标是产出 15 到 40 字的检索式,并补上时间范围。
先写一个不依赖 LLM 的规则版,作为降级方案:
```python
import re
def rewrite_by_rule(q: str) -> list[str]:
q = q.strip()
去掉礼貌性前缀
q = re.sub(r"^(帮我|麻烦|请问|能不能|想了解)\s*", "", q)
口语词替换
q = q.replace("大家都在聊", "").replace("有没有人", "").replace("咋", "怎么")
压缩连续空白
q = re.sub(r"\s+", " ", q).strip(",。?? ")
return [q] if q else []
```
有 LLM 时用提示词改写,一次生成中英两条查询并行检索,中文语义和英文技术文章之间的用词差异靠这个补齐:
```text
你是检索查询改写器。把用户问题改写成 1 条中文检索式和 1 条英文检索式,
每条 40 字以内,只保留实词,不要疑问句。
如果问题里出现"最近""今年""上个月"等相对时间,换算成 YYYY-MM-DD 格式的起始日期。
只输出 JSON,不要解释:
{"queries": ["中文检索式", "english query"], "start": "YYYY-MM-DD 或 null"}
```
相对时间必须换算成绝对日期,因为接口的时间过滤参数要的是标准日期格式,写「最近三个月」会被当成普通字符串处理。
步骤三:搜索并用 contents 抓正文
/search 负责召回,contents 子对象负责顺手把正文带回来,省掉二次请求:
```python
def search(query: str, num_results: int = 8, start: str | None = None) -> dict:
payload = {
"query": query,
"type": "auto",
"numResults": num_results,
"contents": {
"highlights": {"numSentences": 3},
"text": {"maxCharacters": 1200},
},
}
if start:
payload["startPublishedDate"] = start
return exa_post("/search", payload)
```
几处要理解清楚:
type决定检索模式。auto会自行判断,keyword走字面匹配,neural走语义。具体可选值以官方文档为准。含明确专有名词、版本号的查询用keyword往往更准,纯语义问题用neural。highlights返回的是围绕查询词截取的几句话,token 消耗小,适合直接喂给模型。text返回正文,maxCharacters控制截断长度。正文长,成本和上下文占用都高,只在需要精读时开。- 子参数的确切键名可能随版本调整,以官方文档当前版本为准。
如果先搜索、后由模型决定精读哪几篇,就用独立的 /contents 端点二次抓取:
```python
def fetch_contents(urls: list[str], max_chars: int = 2000) -> dict:
return exa_post("/contents", {
"urls": urls[:5],
"text": {"maxCharacters": max_chars},
})
```
注意这里的 urls[:5]。这个端点按 URL 计费,传入多少条就算多少条,代码里必须硬性截断,不能把检索返回的 20 条一股脑传进去。
步骤四:结果去重
同一条内容经常以不同 URL 出现:带不带 www、带不带 utm_source、有没有结尾斜杠、有没有锚点。不去重的话,模型会看到五条「不同来源」其实是同一篇文章,然后把它当成共识反复引用。
先做 URL 归一化:
```python
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
TRACKING = {"utm_source", "utm_medium", "utm_campaign", "utm_term",
"utm_content", "ref", "spm", "from", "share_token"}
def norm_url(u: str) -> str:
parts = urlsplit(u.strip())
host = parts.netloc.lower()
if host.startswith("www."):
host = host[4:]
query = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
if k.lower() not in TRACKING]
path = parts.path.rstrip("/") or "/"
return urlunsplit(("https", host, path, urlencode(query), ""))
```
再做内容指纹,防止同一篇文章挂在两个域名下:
```python
import hashlib
import re
def fingerprint(item: dict) -> str:
raw = (item.get("title", "") + " " + item.get("text", "")[:300]).lower()
return hashlib.sha1(re.sub(r"\W+", "", raw).encode()).hexdigest()
def dedup(items: list[dict], per_domain: int = 2) -> list[dict]:
seen_url, seen_fp, domain_count, out = set(), set(), {}, []
for it in items:
u = norm_url(it.get("url", ""))
if not u or u in seen_url:
continue
fp = fingerprint(it)
if fp in seen_fp:
continue
host = u.split("/")[2]
if domain_count.get(host, 0) >= per_domain:
continue
seen_url.add(u)
seen_fp.add(fp)
domain_count[host] = domain_count.get(host, 0) + 1
out.append({**it, "url": u})
return out
```
per_domain 限制同一站点保留 2 条以内,避免一个内容农场把结果列表刷满。
步骤五:按次计费的控制手段
Exa 按请求计费,搜索调用和正文抓取分别计价,具体单价以官方定价页为准。工程上要做的是四件事:限制单次调用的数量、复用结果、预算熔断、记录消耗。
预算计数器:
```python
class BudgetExceeded(RuntimeError):
pass
class Budget:
def __init__(self, max_units: int):
self.left = max_units
def spend(self, units: int = 1, what: str = "") -> None:
if units > self.left:
raise BudgetExceeded(f"预算不足:需要 {units},剩余 {self.left}({what})")
self.left -= units
```
本地缓存,同一会话里重复的问题直接命中:
```python
import json
import os
import time
CACHE_DIR = ".exa_cache"
os.makedirs(CACHE_DIR, exist_ok=True)
def _cache_path(key: str) -> str:
return os.path.join(CACHE_DIR, hashlib.sha1(key.encode()).hexdigest() + ".json")
def cache_get(key: str, ttl: int = 86400):
p = _cache_path(key)
if os.path.exists(p) and time.time() - os.path.getmtime(p) < ttl:
with open(p, encoding="utf-8") as f:
return json.load(f)
return None
def cache_set(key: str, value) -> None:
with open(_cache_path(key), "w", encoding="utf-8") as f:
json.dump(value, f, ensure_ascii=False)
```
四条额度纪律,写进代码而不是靠自觉:
1. numResults 默认 8,别默认 20。检索是为了给模型找依据,不是做数据集。
2. 先搜索、后精读。搜索阶段只开 highlights,需要正文时再用 /contents 抓前 5 条。
3. 一次会话设总预算,超了抛异常而不是静默继续。
4. 重试最多两次,用指数退避。429 时盲目重试只会把额度烧得更快。
步骤六:包装成智能体可调用的工具
最后把上面几段串成单一入口,并给它写清楚工具描述。描述里要写明「什么时候该调用」,否则模型不会主动用:
```json
{
"name": "semantic_web_search",
"description": "当用户的问题需要外部资料、涉及技术方案对比、或提到最近/有没有人/怎么解决时调用。返回去重后的网页标题、链接和与问题相关的正文摘录。",
"parameters": {
"type": "object",
"properties": {
"question": { "type": "string", "description": "用户原话问题" },
"limit": { "type": "integer", "description": "返回条数,默认 5,不超过 8" }
},
"required": ["question"]
}
}
```
调度层大致这样组织:
```python
def semantic_web_search(question: str, limit: int = 5, budget: Budget | None = None) -> dict:
budget = budget or Budget(max_units=10)
queries = rewrite_by_rule(question) or [question]
merged: list[dict] = []
for q in queries:
cached = cache_get(q)
if cached:
merged.extend(cached)
continue
budget.spend(1, f"search:{q}")
data = search(q, num_results=8)
items = data.get("results", [])
cache_set(q, items)
merged.extend(items)
kept = dedup(merged)[:limit]
return {
"query": question,
"results": [
{
"title": it.get("title"),
"url": it.get("url"),
"published": it.get("publishedDate"),
"highlights": it.get("highlights") or [],
}
for it in kept
],
"billing": {"content_urls": 0},
}
```
把 highlights 交给模型而不是整段正文,能在多数问答场景下把上下文占用压下来一个量级。
常见坑与排错
401 Unauthorized。 八成是鉴权头写成了 Authorization: Bearer xxx。改成 x-api-key,并确认环境变量确实被进程读到,echo $EXA_API_KEY 先验一下。
429 Too Many Requests。 触发限流。加指数退避:第一次等 1 秒,第二次 2 秒,三次后放弃并向上层返回「检索暂不可用」,让智能体走无检索的降级路径。
搜索结果为空。 常见于三种情况:查询是一整句长问句、查询是纯 URL 或代码片段、查询里塞了太多不必要的修饰。缩短到 40 字以内,或把 type 换成 keyword 再试一次。
中文查询召回差。 中文技术内容的用词和英文差异大,同时发一条英文查询再合并去重,通常能补上一批结果。
contents 返回的正文是空字符串。 目标站点可能有反爬策略、或者页面是 PDF、或者是纯前端渲染。换用 highlights 或 summary 字段,或者显式启用实时抓取参数;具体参数名与行为以官方文档当前版本为准。
账单比预期高。 排查两处:一是 /contents 有没有对二三十个 URL 一次性发起请求,二是智能体的循环里有没有对同一问题反复检索。前者靠截断,后者靠缓存加预算熔断。
日期过滤不生效。 日期参数要 ISO 格式(YYYY-MM-DD)。传了「最近三个月」这种自然语言会被忽略,而且不会报错,属于静默失败,写完记得打印请求体确认。
字段名对不上。 检索结果里发布时间字段的确切名称、contents 子对象的结构都可能随版本变化。养成第一次调用就打印原始 JSON 的习惯,别照抄二手示例。
下一步建议
跑通之后,按收益排序的几个增强方向:
1. 建评测集。 拿 20 个真实问题,人工标注哪些链接算命中,每次改改写规则或检索参数就回归一遍。没有评测集的调参全靠感觉。
2. 加语义重排。 检索返回的排序是按相关性给的,但未必符合你的场景。可以再跑一轮轻量打分,或者直接用 highlights 的匹配度做二次排序。
3. 拆分快慢路径。 简单事实性问题直接返回 highlights,需要综合多篇时才抓正文,把成本压在真正需要的请求上。
4. 补可观测性。 每次调用记录:原始问题、改写后的查询、返回条数、去重掉几条、消耗多少单位。出问题时这些日志比断点有用。
5. 并发化。 中英双查询用异步请求并行发出,端到端延迟能明显下降。
到这里,一个带缓存、带预算熔断、结果已经去过重的语义检索工具就接进智能体了。剩下的工作不是加功能,而是拿真实问题反复跑,把改写规则和条数上限调到适合自己业务的样子。
