最近写 Python 脚本的时候,我越来越频繁被问到一类需求:让程序自己去搜索引擎里搜实时信息,再把结果整理成一段能直接用的答案。这类能力现在大多通过“搜索增强型问答 API”提供,接口形式跟普通大模型聊天很接近,但实时检索、引用来源、流量控制这些细节都有一些不太一样的玩法。
这篇文章不是从装 Python 开始手把手教学,而是给已经会点 Python、想把问答 API 接进自己项目的朋友一份可以直接照着跑的实战方案。我会把请求头、请求体、参数选择、返回解析、异常处理、配额管理这些环节全部拆开讲,最后附上我实际调试时踩过的几个坑。项目里用到的调用方式不单独绑定某家厂商,接口设计大同小异,你可以把它当成一套通用落地模板。
1. 用 Python 调问答 API:先想清楚它解决什么问题
1.1 检索增强问答 API 与普通大模型接口的差别
这类问答 API 的核心特点是:收到你的问题之后,不是直接凭模型“记忆”里的知识硬答,而是先去执行一轮检索,把相关的网页、文档片段捞回来,再让大模型基于这些检索结果做归纳和生成。所以它的回答往往带有较强的时效性,能回答“昨天刚发生的事”,同时返回里通常还会附带引用来源列表。
普通大模型接口通常走POST /chat/completions,传入模型名、消息列表、温度参数,返回choices[0].message.content;而检索增强问答 API 有的沿用这套结构,有的则独立一个/answers或/search端点,返回字段也从单一的content变成了answer + citations + related_queries这种组合。我在最开始接触这类接口时,笼统地按大模型聊天那一套去写,结果各种 KeyError,后来才明白字段结构不一样是正常的。
这个差别直接决定了你写代码时的策略:如果服务商提供的是 OpenAI 兼容格式,可以省很多事;如果返回结构不同,就要自己写字段兼容层。后面我会给一个适配两种结构的函数,省得来回改。
1.2 为什么偏要用 Python,而不是直接开网页去搜
很多人第一反应是“我直接在网页里问不就行了”,但当你有一批问题要批量处理时,脚本的价值就出来了。比如抓取几十个股票代码的最新公告摘要、整理竞品新闻、给内部知识库补充词条、每天定时跑一个资讯汇总,这些重复劳动用人工去问,效率低还容易漏。用 Python 把问答 API 包装成一个函数之后,想怎么调度都行,配合定时任务就能实现全自动。
另外,程序调用可以结构化地拿到“答案”和“引用来源”,这点比浏览器里点开一个个结果页要方便得多。引用来源可以直接入库,给调用方做来源追溯,也方便后续人工复核。Python 生态在这里也很顺手,requests做请求、json做解析、pandas做批量处理,一条链路非常顺。
1.3 适合谁看,以及动手前需要什么
如果你属于下面任一情况,这篇文章比较对路:
- 想用 Python 给自己的小工具加上“实时问答”能力;
- 需要批量把问题转换成带来源的文本答案;
- 看到文档头大,希望有一份可以直接改改就跑的代码;
- 已经跑通接口,但被 401、429、超时这些报错搞得焦头烂额。
动手前的准备也很简单:一台装了 Python 3.8+ 的电脑,一个能访问目标服务商 API 的合规网络环境,一个申请好的 API Key,还需要装requests和python-dotenv两个库。如果你还没有 Key,先进服务商后台创建账号、开通 API 权限,拿一个测试 Key 再回来。注意申请时看清楚免费额度和计费规则,别把主 Key 散到代码里。
2. 动手前先定好调用方式:requests 还是 SDK
2.1 三种调用方案的对比
我在不同项目里用过三种方式,各有优势,先列个表说明白:
| 方案 | 优点 | 缺点 | 适合场景 |
|---|---|---|---|
| 直接用 requests | 依赖最少,结构完全可控,方便调试 | 要自己处理鉴权、重试、流式解析 | 大多数一次性或中小批量任务 |
| 用服务商提供的 Python SDK | 封装好连接池和流式响应,代码更少 | 有的 SDK 更新不及时,参数名和文档不一致 | 官方维护积极、需要流式输出时 |
| 用 aiohttp 异步请求 | 并发效率高,适合大量问题 | 代码复杂,事件循环容易踩坑 | 高并发、需要秒级吞吐的场景 |
我的建议是:第一版先用requests。它的代码非常直白,出了问题你打开返回的 JSON 一眼就能定位。等你把业务逻辑跑通、确认字段之后,再考虑要不要换 SDK 或改异步。
2.2 密钥管理与安全配置
很多刚接触 API 的人会直接把 Key 写在脚本里,比如API_KEY = "sk-xxx",这在小 demo 里没问题,但一旦代码提交到 Git 仓库或者发给别人,就相当于把账号上限和账单权限交出去了。我自己中过一次招,某个小工具项目把测试 Key 提交到了公开仓库,几分钟内就被扫描机器人抓取,白白损失了一天的免费额度。
所以请你务必这样配置:
- 在项目根目录新建
.env文件,写入敏感配置:QA_API_KEY=sk-你的密钥 QA_API_BASE_URL=https://api.example-qa-service.com/v1 QA_MODEL_NAME=qa-model-v1 - 安装
python-dotenv,在代码里通过load_dotenv()加载环境变量。 - 把
.env加进.gitignore,确保它永远不会被提交。
加载代码就两行:
pip install python-dotenv requestsimport os from dotenv import load_dotenv load_dotenv() API_KEY = os.getenv("QA_API_KEY") BASE_URL = os.getenv("QA_API_BASE_URL", "https://api.example-qa-service.com/v1") MODEL = os.getenv("QA_MODEL_NAME", "qa-model-v1")如果API_KEY为空,代码里应该立刻抛出错误,而不是拿着空字符串去请求,这和“配置缺失”优先级高于“请求失败”是同一个道理。
2.3 请求结构与关键参数解析
多数检索增强问答 API 的请求体长这样:
{ "model": "qa-model-v1", "messages": [ {"role": "system", "content": "你是一个严谨的信息整理助手。"}, {"role": "user", "content": "Python 3.12 有哪些新特性?"} ], "temperature": 0.2, "max_tokens": 1024, "stream": false }几个关键参数我展开说说:
model:指定模型名。不同模型支持的检索类型、上下文长度、计费价格都不一样。建议先看文档或跑一次测试,确认你申请的 Key 能调用哪个模型,不要想当然填一个名称。temperature:控制回答随机性。带检索的问答 API 我建议先调到 0.2 左右,因为这种场景一般更看重事实准确性,太高的随机性会让回答跑偏。注意有些服务商会固定温度,传了也会被忽略,这倒不是 bug。max_tokens:控制最大生成长度。检索增强型回答普遍比普通聊天长,因为要带上理由和引用,不要太抠。但也别一次给到 4096,费用会明显上涨。stream:是否流式输出。如果只是拿最终答案,保持false最简单。如果要做打字机效果或长回答实时展示,再打开true。流式返回通常是 SSE 格式,需要按data:前缀逐行解析。citations或return_sources:部分服务商有这个开关,用于控制是否返回引用来源。如果你的下游要展示来源,记得打开并且重点关注。
鉴权头一般是这样:
headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" }有的服务商要求的是自定义 Header,比如X-Api-Key,或者Authorization: Token xxx。这个只差一个字就可能导致 401,我建议第一次联调时先用服务商的在线调试工具跑通,再照抄到代码里。
3. 关键环节实现:一个能直接用的 Python 封装
3.1 最小可运行代码:构造请求与解析返回
下面是一段可以直接复制到本地跑通的代码,我把健壮性也做进去了。第一次运行前,先确认你已经把.env配置好,并且安装了依赖。
# qa_api.py import os import requests from dotenv import load_dotenv load_dotenv() API_KEY = os.getenv("QA_API_KEY") BASE_URL = os.getenv("QA_API_BASE_URL", "https://api.example-qa-service.com/v1") MODEL = os.getenv("QA_MODEL_NAME", "qa-model-v1") def ask_qa(question: str, temperature: float = 0.2, max_tokens: int = 1024, timeout: int = 60) -> str: if not API_KEY: raise RuntimeError("缺少 QA_API_KEY,请检查 .env 文件") url = f"{BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL, "messages": [ {"role": "system", "content": "你是一个严谨的中文信息整理助手。回答时先给结论,再给理由,引用来源用编号标注。"}, {"role": "user", "content": question}, ], "temperature": temperature, "max_tokens": max_tokens, "stream": False, } response = requests.post(url, headers=headers, json=payload, timeout=timeout) response.raise_for_status() data = response.json() # 兼容不同返回结构 content = None if "choices" in data: content = data["choices"][0]["message"]["content"] elif "answer" in data: content = data["answer"] elif "data" in data and isinstance(data["data"], dict): content = data["data"].get("answer") or data["data"].get("content") if not content: raise RuntimeError(f"无法从返回中解析答案,原始返回:{data}") return content这里有一个细节:timeout我设成了 60 秒,而不是常见的 10 秒。原因是检索增强型 API 内部要先做一次网页检索,再加上大模型生成,链路明显比普通大模型更长。我之前用默认 5 秒超时去调,十次里四次报超时,改成 30 秒以上才稳定。
3.2 把“引用来源”一起处理
很多场景下我们不光要答案,还要保留来源。比如你做财经简报,如果只给结论不给出处,读者根本不敢信。所以我在实际项目里一般会写一个同时返回答案和引用的版本。
def ask_qa_with_citations(question: str, timeout: int = 60): if not API_KEY: raise RuntimeError("缺少 QA_API_KEY,请检查 .env 文件") url = f"{BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL, "messages": [ {"role": "system", "content": "你是一个严谨的信息整理助手,重要结论后面用 [来源序号] 标注。"}, {"role": "user", "content": question}, ], "temperature": 0.2, "max_tokens": 1024, } response = requests.post(url, headers=headers, json=payload, timeout=timeout) response.raise_for_status() data = response.json() answer = data.get("answer") or data["choices"][0]["message"]["content"] citations = [] raw_citations = data.get("citations") or data.get("sources") or [] for item in raw_citations: citations.append({ "title": item.get("title", ""), "url": item.get("url", ""), "snapshot": item.get("snippet") or item.get("summary") or "", }) return answer, citations这里要注意一点:有的服务商会把引用写在citations数组里,每条有title/url/snippet;有的则直接把引用链接塞在回答文本尾部,像[1]: https://...。前者好办,后者就要你写一段正则去提取。我习惯返回结构化citations,便于直接入库,而不是去解析文本。
3.3 支持多轮对话和历史记录
如果要用在聊天机器人里,历史记录是绕不开的。把上面函数改造一下,接收一个完整的messages列表就行:
def chat_once(messages: list, timeout: int = 60) -> str: if not API_KEY: raise RuntimeError("缺少 QA_API_KEY") url = f"{BASE_URL}/chat/completions" headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"} payload = { "model": MODEL, "messages": messages, "temperature": 0.3, "max_tokens": 1024, } response = requests.post(url, headers=headers, json=payload, timeout=timeout) response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"]调用端维护历史:
history = [ {"role": "system", "content": "你是一个信息整理助手,回答尽量简洁。"}, {"role": "user", "content": "帮我查一下某公司最近一周的公开动态。"}, ] answer = chat_once(history) history.append({"role": "assistant", "content": answer})如果你做了多轮,记得控制历史长度。检索增强 API 比普通聊天更容易撑爆上下文,因为系统提示词、历史对话、检索到的网页片段都会占用 token。我常用的策略是:保留最近 6 条对话,超过就丢弃最老的,避免单次请求 token 超限。
3.4 批量提问与并发限速
批量场景下,一个个顺序请求虽然稳,但效率太低。如果你有几十个问题,顺序跑可能要几分钟,用线程池能快好几倍。但要小心:并发太大会触发 429 限流,所以“有限并发 + 请求间隔”是比较稳妥的组合。
from concurrent.futures import ThreadPoolExecutor, as_completed import time def ask_qa_batch(questions: list, workers: int = 3, sleep_between: float = 0.5): results = [] with ThreadPoolExecutor(max_workers=workers) as pool: future_map = {pool.submit(ask_qa, q): q for q in questions} for future in as_completed(future_map): question = future_map[future] try: answer = future.result() results.append({"question": question, "answer": answer, "ok": True}) except Exception as exc: results.append({"question": question, "answer": str(exc), "ok": False}) time.sleep(sleep_between) return results上面这段代码给每个任务执行完都加了一个sleep(0.5),相当于人为限速,避免瞬间打满你的免费额度。如果你对速度要求高,可以去掉sleep并适当提高workers,但请务必先确认你的账号配额能承受。
4. 常见问题与排查技巧实录
4.1 401 身份验证失败
401 是新手最常见的报错,而且原因往往很蠢。我把遇到过的情况整理成一张速查表:
| 现象 | 常见原因 | 排查方法 |
|---|---|---|
| 401 Unauthorized | Header 名字写错 | 检查是不是Authorization: Bearer xxx,有的服务商要求X-API-Key |
| 401 + “invalid api key” | Key 前后有空格或无痕换行 | 检查.env,不要用sk-xxxx\n |
| 401 但文档里明明能通 | 用了错误的 Key 类型 | 确认是该服务 API Key,而不是控制台账号密码 |
| 偶发 401 | Key 已过期或被后台禁用 | 登录控制台查看 Key 状态和到期时间 |
排查这类问题我一律建议先手动用 curl 试一次:
curl -X POST https://api.example-qa-service.com/v1/chat/completions \ -H "Authorization: Bearer $QA_API_KEY" \ -H "Content-Type: application/json" \ -d '{"model":"qa-model-v1","messages":[{"role":"user","content":"你好"}]}'curl 能通、Python 报错,说明代码里的 header 或 URL 有问题;curl 都不通,就是 Key 或者网络环境的问题,别急着调代码。
4.2 429 请求太频繁
429 的意思是触发了限流。不同服务商给的额度维度不一样,有的是每分钟请求数,有的是每分钟 token 数,还有的是搜索次数。出现 429 时,响应体里一般会有Retry-After头,告诉你多少秒之后再试。
我的通用处理方案是:请求函数里加指数退避重试,不超过 3 次,避免死循环。
import time def post_with_retry(url, headers, payload, timeout=60, max_retries=3): for attempt in range(max_retries): response = requests.post(url, headers=headers, json=payload, timeout=timeout) if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 2 ** attempt)) time.sleep(retry_after) continue response.raise_for_status() return response.json() raise RuntimeError("重试次数已用完,仍返回 429")这里我最想强调的一点:不要一遇到 429 就把重试间隔设得很小,这会让服务商把你当成恶意调用,反而加重限流。间隔从 2 秒、4 秒、8 秒递增,是比较温和的信号。
4.3 返回结构对不上:choices 不存在
检索增强问答 API 的一个大坑就是返回字段“似曾相识但不完全一致”。比如/answers接口返回的是:
{ "answer": "xxxx", "citations": [{"title": "a", "url": "https://..."}], "related_queries": ["..."] }而不是choices[0].message.content。如果你死死按 OpenAI 兼容格式解析,就直接 KeyError。我在一个写资讯助手的项目里就吃过这个亏,当时为了适配多个接口,最后写了很笨的字段兜底逻辑:
def extract_answer(data): choices = data.get("choices") or data.get("data") or {} if isinstance(choices, list) and choices: return choices[0].get("message", {}).get("content") if isinstance(data.get("answer"), str): return data["answer"] if isinstance(data.get("data"), dict): inner = data["data"] if inner.get("answer"): return inner["answer"] return None这样至少不会一崩到底。建议你拿到任何返回,先打印一次完整 JSON 再写解析,别凭记忆猜字段。
4.4 超时和莫名其妙的连接错误
前面提过,搜索增强型 API 比普通大模型接口慢。有的同学一上来设timeout=3,然后狂问“为什么 Timeout”。我这里直接建议:普通请求给 30 秒,长问题给 60 秒。如果你做的是批量任务,可以考虑把超时分文件、超时重试、队列结合起来,而不是单纯增加 timeout。
连接错误里还有一种情况是 SSL 证书报错或 DNS 解析失败,这通常跟本机网络环境或代理设置有关。遇到这种问题,先看看是不是全局代理把请求劫持了,或者公司出口网络对 API 域名做了限制。调试时可以用:
response = requests.post(url, headers=headers, json=payload, timeout=timeout, verify=True)真的需要先快速定位网络问题时,可以暂时verify=False,但生产代码千万别长期关掉证书校验,那是拿安全换一时的便利。
4.5 成本控制与免费额度玩法
检索增强 API 的计费通常分两块:一是按 token 计费,二是按检索次数计费。其中检索次数往往比 token 更费,而且容易被忽略。我见过有人批量跑 500 条问题,结果每个问题都附带多次自动检索,费用直接翻了几倍。
控制成本我有几个习惯:
- 给同样的输入做本地缓存,用 SQLite 或内存字典存
(question, model)到answer的映射,重复问题直接读缓存,不重复调 API。 - 批量任务先拿 2 条数据测费用。在请求里把
max_tokens调成 512,跑一次看返回消耗,再决定要不要放开,别上来就跑全量。 - 给请求函数加一个总预算保护,统计累计使用量,超过设定值就主动熔断。
- 利用免费额度时,注意额度的统计口径。有的按自然日刷新,有的按账号周期刷新,别在月底最后一天临时跑大任务。
5. 最后分享几个我的实操习惯
文章快写完了,最后说几个我踩过几次坑之后沉淀下来的习惯,希望对你有参考价值。
第一,第一次调用任何问答 API,我一定会先打印原始返回 JSON,绝不直接写解析代码。这一步能帮你省掉至少半小时的“字段名猜来猜去”时间。你可以把返回内容保存成一个 json 文件,之后调试时反复翻看,比在命令行里看截断输出舒服得多。
第二,所有调用函数统一入口,不要每个文件各写一份请求逻辑。我会把ask_qa、ask_qa_with_citations、post_with_retry这些都收进一个qa_client.py模块,其他业务代码只管调用。这样后续要加日志、加缓存、换模型时,只改一个文件就行。
第三,注意控制并发“匀着来”。即使服务商给了很高的并发上限,我也会故意留一点余量。批量任务宁可多跑 10% 的时间,也不要因为触发限流导致任务中断重来。我习惯先跑 10 条数据观察一下响应耗时,再决定最终并发数。
第四,日志里不要记录完整 Key,但要把最后 4 位记录下来。排查 401 时,“用户传进来的 Key 尾号是不是我预期的那个”,一眼就能看出来,又不会泄露完整密钥。这个技巧在很多协作项目里救过我很多次。
对我来说,调用这类 API 本质上就是三件事:把请求体构造对、把返回结构解析兼容、把异常处理补全。做到这三件事,不管后端怎么换服务商,你的 Python 代码都能很快接手。希望这篇实战记录能让你少走几步弯路。