跳到主内容
快讯直播
AI智模界
教程

上手 Exa 3.0:给智能体接一层语义搜索

智能体在回答「有没有人踩过把搜索结果整段塞进上下文、结果模型跑偏的坑」这类问题时,靠关键词搜索很难命中,因为用户的说法和文章里的说法往往差着好几个词。语义搜索解决的正是这件事:把一句话的意思拿去匹配网页,而不是拿字面去匹配。这篇教程带你从零把 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. 并发化。 中英双查询用异步请求并行发出,端到端延迟能明显下降。

到这里,一个带缓存、带预算熔断、结果已经去过重的语义检索工具就接进智能体了。剩下的工作不是加功能,而是拿真实问题反复跑,把改写规则和条数上限调到适合自己业务的样子。

AI 生成本文由 AI 基于公开信息自动生成,仅供参考。