将多源文献API接入Agent：统一检索层的设计与实现-平芜编程栈

做 Agent 最常被问到的问题，不是怎么调提示词，而是：“你的 Agent 查文献是怎么连的？” 我自己在折腾科研助手类 Agent 时也卡了很久。文献数据库很多，每个库都有自己的接口风格、限流策略、字段命名，如果只接一个库，召回率不够；全接过来，工作量又大得吓人。后来我收敛出一个免费好用的方案：用一个轻量的统一接入层，把多个文献数据库的服务包装成一套标准接口，再给 Agent 一个工具入口。这个工具我把它叫做 LiteScholar，名字不重要，重要的是这套接线方法，它能把“让模型自己查文献”这件事从失控变可控。

适合谁看？适合正在做大模型 Agent 应用，需要让模型自己查文献、整理综述的人；也适合刚接触开放文献接口、不知道从哪个库开始的人。我的目标很简单：免费、不依赖付费订阅、不碰网页抓取，只走各家开放 API，把 Agent 接到“能查、能筛、能拿全文链接”的程度。这篇文章里你会看到我是怎么定接口模型、怎么区分不同数据源的适配逻辑、怎么用缓存和限流让免费额度长期够用，以及几个让我调试到深夜的坑。

1. 开始前，先解决三个设计问题

1.1 免费接入的边界在哪里

“免费”两个字最容易让人误解。不是所有文献数据库都会把完整能力开放出来。有的库只给检索结果，不保证能拿全文；有的库允许个人账号拉数据，但限制并发；有的库完全不提供开放接口，只能在网页上人工点。

所以我的理解是，免费接入的正确做法，是用官方提供的开放接口，把能够合法获取的元数据和开放获取全文接出来。不要试图绕过付费墙，那既不合法，也没有可持续性。在这个前提下，绝大多数面向科研的公开数据源都支持关键词检索、返回结构化字段。这就已经够 Agent 做很多事了：回答“某领域有哪些最新论文”、生成综述草稿、按引用关系追踪文献。

另外要认清，免费额度不等于无限额度。几乎所有开放接口都会设限流，有的按分钟，有的按天。对原型和低并发场景来说完全够用，但你不能拿它当生产级搜索后端来压。我在 LiteScholar 里专门做了限流控制和缓存，目的就是让有限额度用得更久。

1.2 多库接入为什么比单库麻烦

单库接入非常简单，申请一个 key，调一个 URL，解析 JSON 就结束了。多库接入的麻烦在于接口风格极度不统一。说几个让我加班的差异点：

分页方式不同。有的用 page/page_size，有的用 offset/limit，还有的用游标。
返回字段不同。有的库把作者拼成字符串，有的返回对象数组。
排序方式不同。有的只支持相关度排序，有的允许按引用数或时间排序。
认证方式不同。有的要请求头带 token，有的要 query 参数。
限流策略不同。有的每分钟 30 次，有的每天几千次。

如果直接在 Agent 代码里分别写死每个库的调用逻辑，那等于把数据库的复杂性全灌进了 Agent 流程。后续想加新库、改字段映射、换 key，都得动主流程。我用 LiteScholar 时许过一个愿望：以后再加新库，只新增一个适配器文件，不改 Agent 的任何代码。这个愿望后来真的实现了。

1.3 统一接入层的设计原则

我画的边界很简单：Agent 只和统一接口说话，统一接口负责分发。数据源之间互不可见，每个源对应一个适配器。设计时我给自己定了三条原则。

第一，对外只暴露一个 search 方法。不管底层是医学库、预印本库还是综合专利库，Agent 都只传关键词、过滤条件、分页参数。第二，返回必须标准化。title、authors、abstract、doi、year、source、open_access_url，这几个字段每条记录都必须有；缺失的字段填空字符串或 null，不能报错。第三，错误不能拖垮 Agent。某个库挂了，统一接口要能自动跳过并继续查询其他库，最后在响应里标记哪些库失败。

这套设计做完之后，再往 Agent 里接工具就非常快。后面我从配置到代码逐步展开，你看完完全可以照着重现一套。

2. 核心细节：统一检索层要处理什么

2.1 统一字段模型：先定标准再写适配器

我最开始犯的错，是先写适配器、后定字段。结果每个适配器返回的 JSON 都不一样，Agent 解析时经常拿错值。后来我学乖了，先把标准响应结构固定下来。LiteScholar 里一个标准记录，我定义了这些字段：

字段	类型	说明
id	string	来源系统的唯一 ID
title	string	标题
authors	list[str]	作者名列表
abstract	string	摘要，可能为空
year	int	发表年份
doi	string	DOI 标识符，可能为空
source	string	标识来源库的短名称
open_access_url	string	可访问全文或摘要页的链接
score	float	相关度得分，统一归一化到 0-100

为什么要统一 score？因为不同库的相关度分数量纲完全不一样，有的越高越好，有的越低越好。如果不归一化，Agent 拿到结果后根本无法比较“哪个结果更相关”。作者字段一开始我用字符串，后来改成列表，因为大模型处理列表比拆字符串稳定得多。字符串作者名经常带分号或逗号，模型切分时容易出错，列表可以直接用。

2.2 适配器模式：每个库只做三件事

有了标准字段，每个数据源适配器只需要完成三件固定的活：把 Agent 的查询参数翻译成目标库的请求参数；发起 HTTP 请求，处理认证和分页；把目标库的响应映射到标准字段。

这里贴一个适配器的伪代码骨架。不用太纠结具体库，重点是理解每个适配器都长这样。

import requests class BaseAdapter: source_name = "base" def __init__(self, api_key, timeout=10): self.api_key = api_key self.timeout = timeout def search(self, query, page=1, size=10, **kwargs): params = self.build_params(query, page, size, kwargs) headers = self.build_headers() resp = requests.get(self.endpoint, params=params, headers=headers, timeout=self.timeout) resp.raise_for_status() return self.normalize(resp.json()) def build_params(self, query, page, size, kwargs): raise NotImplementedError def build_headers(self): return {} def normalize(self, raw): raise NotImplementedError

你可能觉得这个骨架太简单。实际上真正的适配器也就是这么简单。复杂的是每个库的 normalize 函数：因为目标库返回的字段名千奇百怪，你要在 normalize 里做大量的 get 和容错。经验是，写 normalize 的第一版时，先手动打几条真实响应存成样例，然后照着样例写映射，不要靠记忆去猜字段名。靠猜的后果就是某个字段在真实数据里叫 abstractText，而你以为叫 abstract，上线之后一路报错。

2.3 限流、缓存和降级：免费接口能不能用的关键

免费接口最大的问题不是功能弱，是限流。我实测过，大部分免费接口给个人使用的限额大致在每分钟几次到几百次之间。如果 Agent 同时接多个库，一次检索请求可能触发好几个库的调用，稍不注意就会打到限流，返回 429。处理不好，Agent 整个链路都会超时。

我在 LiteScholar 里用了四个策略：

每个库独立配置每秒请求数上限，用本地令牌桶控制。
结果缓存 30 分钟。相同的关键词组合不再请求后端。
连续失败两次后自动熔断，后续请求在 1 分钟内直接跳过该库。
优先级降级：关键库失败后，把语义相近的备选库放进本轮查询。

令牌桶可以用一个简单的队列实现，也可以直接用现成的限流库。另一个容易忽略的点是超时。免费接口偶尔会慢，甚至卡住。我统一设置了 10 秒超时，并且所有请求都使用同一个连接池，避免每次新建连接浪费握手时间。你如果发现请求总是时快时慢，大概率不是接口问题，是你没有复用连接。

3. 实操记录：把一个统一检索服务跑起来

3.1 准备阶段：申请 key 和定配置

虽然不能点名具体库，但市面上常见的开放文献接口，申请流程都差不多：注册账号、创建应用、拿到一个 API key。有的库还要求填写用途说明，一般选“个人学术研究”就能通过。我建议在配置里至少准备两个数据源 key，一是为了保证召回率，二是为了在某个库限流时依然能工作。

配置我用 YAML 存着，放在服务启动时读取。大致长这样：

db_providers: - name: bio_med type: medicine api_key: ${BIO_MED_KEY} rate_limit: 3 # 每秒最多 3 次 cache_ttl: 1800 - name: preprint type: preprint api_key: ${PREPRINT_KEY} rate_limit: 5 cache_ttl: 600

注意 key 不要硬编码进代码，用环境变量引用。这样即使仓库代码泄漏，key 也不会跟着泄漏。我还习惯在本地建一个 .env.example 文件，把需要的环境变量全部列出来，方便换机器时快速配环境。

3.2 核心服务：多库并行检索的写法

统一检索服务里，我推荐对上游做并行请求，而不是串行。串行的问题很直接：三个库平均耗时 0.5 秒，串行就要 1.5 秒，Agent 等不起；并行的话，整体耗时只取决于最慢的那个库。下面是 LiteScholar 里服务端的示例实现。

from concurrent.futures import ThreadPoolExecutor class UnifiedSearchService: def __init__(self, adapters, cache=None): self.adapters = adapters self.cache = cache def search_all(self, query, page=1, size=10): cache_key = (query, page, size) if self.cache: result = self.cache.get(cache_key) if result: return result def run_one(adapter): try: records = adapter.search(query, page=page, size=size) return records, adapter.source_name, None except Exception as e: return [], adapter.source_name, str(e) merged = [] errors = {} with ThreadPoolExecutor(max_workers=len(self.adapters)) as pool: futures = [pool.submit(run_one, adp) for adp in self.adapters] for fut in futures: records, name, err = fut.result() if err: errors[name] = err else: merged.extend(records) merged.sort(key=lambda x: x["score"], reverse=True) payload = {"query": query, "records": merged[:size], "errors": errors} if self.cache and not errors: self.cache.set(cache_key, payload) return payload

这段代码里有两个细节值得说。第一，如果某个库完全失败，我仍然会返回其他库的结果，只在 errors 字段里带上失败信息，这样 Agent 不会因为这个库挂了而中止整个任务。第二，排序放在合并之后。因为不同库相关度分数已经被适配器归一化到同一量纲，所以才可以直接比较。实际项目里，我还会在 merge 后做一遍去重，按 DOI 和 title 判断是否是同一篇文献，如果重复，优先保留 open_access_url 更完整的那条。

3.3 把统一服务包装成 Agent 工具

服务跑起来之后，接入 Agent 就只剩最后一步：把 search_all 这个函数描述给大模型，让它在需要查文献时主动调用。大多数 Agent 框架都支持“工具定义加函数执行”的模式。你只需要声明工具名、描述、参数结构，再写一个执行函数。

{ "type": "function", "function": { "name": "literature_search", "description": "搜索公开学术文献。支持关键词、发表年份、排序方式。返回标准化字段。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "检索关键词，要求语义完整" }, "year_from": { "type": "integer", "description": "起始年份，可选" }, "page": { "type": "integer", "description": "页码，从1开始" }, "size": { "type": "integer", "description": "每次返回条数，最大20" } }, "required": ["query"] } } }

Agent 侧的执行函数，本质就是调用刚才的 UnifiedSearchService。返回的字符串不要直接丢给大模型，最好精简成一行一条的摘要文本。大模型对结构化文本的跟随能力，远好于处理一个嵌套很深的 JSON。我的做法是，把每条记录渲染成类似下面的文本：

[来源:bio_med] 标题：视网膜图像分割方法综述 作者：A, B, C 年份：2024 DOI: 10.xxxx 摘要前80字：本文回顾了近年来深度学习在视网膜血管分割中的进展...

这样 Agent 读取时消耗的上下文更少，定位信息也更准。

3.4 一个可运行的调用示例

接好之后，用户在对话框里问“帮我找一下近三年关于视网膜图像分割的开放获取文献”，后台发生的完整链路是：

Agent 判断需要查文献，生成 literature_search 工具调用。
LiteScholar 把 query 翻译成各库的参数，并行请求多个数据源。
各库返回结果，适配器 normalize 成统一格式，合并去重排序。
Agent 拿到精简文本，从中挑出合适的文献，整理成回答。

这一步我测试时最深的感受是：工具调用链路不复杂，复杂的是把上游数据打磨得干净。如果上游返回的摘要里带换行符、带 HTML 标签，Agent 回答时就会很痛苦。所以适配器里一定要加一层文本清洗，把多余的空格、HTML 实体、乱码全部清理掉再进统一模型。

4. 问题排查与避坑经验

4.1 请求超时、连接重置怎么办

免费接口没有商业 SLA，不稳定是常态。遇到超时，第一件事不是调大 timeout，而是看日志里哪个库消耗了最多时间。我的排查思路：

先看是不是所有库都慢。如果是，可能是本机网络 DNS 解析有问题。
如果只有某个库慢，去看这个库的 API 文档，确认是不是近期改了入口域名。
如果慢在连接池耗尽，就调大连接池上限，开启 Keep-Alive。

我踩过最大的坑，是“用单例 session 但没开连接池复用”，结果高并发下每个线程还在各自建连接，慢到怀疑人生。后来改成所有线程共享一个 session，整体耗时立刻降了一半。你可以在适配器初始化时创建全局 session，然后每个请求都从同一个 session 发出去。

4.2 返回结果总是缺字段

有的库对于某些老文献确实不返回摘要或 DOI，这不是 bug。但如果你发现某个类型的记录总是缺字段，就要回去检查 normalize 映射。这里有个小技巧：写适配器的时候，把原始响应和标准响应同时打到日志里，等积累了几十条样本，再一次性把漏映射的字段补上。靠猜是补不全的。

还要注意字段类型。有的库把年份返回成“2024.0”，有的是字符串“2024”。normalize 里一定要做 int() 并容错。年份解析失败时，我一般直接设成 0，而不是抛异常。因为单独一篇文献年份解析失败，不应该影响整轮检索结果。

4.3 Agent 每次搜索都带上一堆乱七八糟参数

我发现大模型在调用工具时，有时候会把“最近三年”理解成 year_from=2021，但有时又会传一些非预期参数，比如 language 或 sort。解决办法是在参数 schema 里把可选项尽量收敛，并把描述写清楚。另外，提示词里可以加一句约束：“除非用户明确要求，否则不要传递过滤参数。” 这句话能明显减少误传。

还要在服务端做好参数白名单。超出范围的 year_from 就默认用 None 处理，不要抛异常给 Agent。多余的未知参数直接忽略，不要返回错误。因为 Agent 对你的接口容错能力判断通常很粗暴，一旦报错，它可能连续重试，反而浪费更多请求额度。