企业官网建设流程全解析

• 人工智能
• NLP
• 深度学习

导读

本文深入解析 DeepPavlov 开源库中deeppavlov.models.api_requester模块——一个将模型推理管线与外部 HTTP API 服务衔接起来的通用组件层。模块包含ApiRequester（转发请求到 API 端点）与ApiRouter（并行调度多个 API 请求器）两个核心类，是 KBQA 等知识问答场景将 Wikidata 解析、实体链接等重计算任务外包给独立服务的关键桥梁。读完本文，你将掌握这两个组件的参数语义、批处理与异步逐条请求两种工作模式，以及如何在 DeepPavlov 的 JSON 配置中把它们注册进 Chainer 管线。

模块概览：API 参考文档背后的真实实现

docs/apiref/models/api_requester.rst是 DeepPavlov 文档体系中针对api_requester模块的 API 参考页。该页面通过 Sphinx 的automodule与autoclass指令，自动从源码抽取两个类的公开接口：

• ApiRequester：核心请求转发组件，文档重点标注了__call__与get_async_response两个方法；
• ApiRouter：多个 API 请求器的并行调度器，文档标注了其__call__方法。

与之对应的源码位于 deeppavlov/models/api_requester/api_requester.py 与 deeppavlov/models/api_requester/api_router.py，模块入口 deeppavlov/models/api_requester/init.py 负责导出。两个类均继承自 deeppavlov/core/models/component.py 中定义的Component抽象基类——这意味着它们天然适配 DeepPavlov 的 Chainer 管线约定：实现__call__方法、可被reset()与destroy()生命周期管理（后者会递归销毁其持有的子组件）。

ApiRequester：把管线参数转发给外部 API

ApiRequester的定位（源码 docstring）是 "Component for forwarding parameters to APIs"，即一个纯粹的转发器：它不参与模型计算，而是把管线中某个环节产生的中间结果打包成 JSON，POST 到指定的 API 端点，再把响应 JSON 解析后返回管线。

构造参数与语义

实现细节见 api_requester.py 的__init__。

__call__：两种请求模式

__call__的完整逻辑（api_requester.py）可以拆解为三步：

第一步：组装请求数据。data = kwargs or dict(zip(self.param_names, args))——如果调用时传入命名参数（**kwargs），则直接以命名参数作为请求体；否则把位置参数*args按param_names的顺序压缩成字典。因此param_names决定了位置参数到 JSON 字段名的映射。

第二步：按debatchify分支处理。

• debatchify=False（默认）：直接requests.post(self.url, json=data).json()，整批数据一次性作为 JSON body 发送，同步等待响应；
• debatchify=True：先从data的任一字段值推断批大小batch_size（取第一个字段的列表长度，并断言其大于 0），随后调用get_async_response(data, batch_size)异步地逐条发送；若out_count > 1，还会把结果做转置list(zip(*response))，把"每条请求的多个返回值"重排为"每个返回值字段的所有结果"，以便下游按字段取用。

第三步：返回响应。无论哪种模式，返回值都是 API 响应 JSON 解析后的结果（列表或字段化结构）。

get_async_response：异步逐条请求的实现

get_async_response（api_requester.py）是文档重点标注的辅助方法，专门解决"API 端点不支持批处理"的瓶颈：

async def get_async_response(self, data: dict, batch_size: int) -> AsyncIterable: loop = asyncio.get_event_loop() futures = [ loop.run_in_executor( None, requests.post, self.url, None, {k: v[i] for k, v in data.items()} ) for i in range(batch_size) ] for r in await asyncio.gather(*futures): yield r.json()

其核心思想是：通过loop.run_in_executor把阻塞式的requests.post投递到线程池执行，用asyncio.gather并发等待所有单条请求，最后逐个yield解析后的 JSON。注意这里逐条请求的 payload 是{k: v[i] for k, v in data.items()}——即取出每个字段的第i个元素组成单条实例。外层__call__中通过asyncio.get_event_loop()与loop.run_until_complete(collect())驱动整个协程完成。

ApiRouter：并行调度多个 ApiRequester

当一次推理需要同时询问多个独立的外部服务（例如同时查询实体链接与 Wiki 解析）时，ApiRouter提供进程级并行能力。其构造参数为：

• api_requesters：ApiRequester对象的列表；
• n_workers：int，默认1，即ProcessPoolExecutor的最大子进程数。

__call__（api_router.py）把所有api_requester以相同输入*args提交到进程池：

with ProcessPoolExecutor(self.n_workers) as executor: futures = [executor.submit(api_requester, *args) for api_requester in self.api_requesters] concurrent.futures.wait(futures) results = [] for future, api_requester in zip(futures, self.api_requesters): result = future.result() if api_requester.out_count > 1: results += result # 多返回值请求器：结果展开拼接到列表 else: results.append(result) # 单返回值请求器：整体追加

两个值得注意的设计点：其一，结果组装逻辑依据每个请求器的out_count分支——多返回值的结果被展平拼入总列表，单返回值则作为整体元素追加，这要求使用者清楚各请求器的输出形态；其二，由于使用ProcessPoolExecutor，ApiRequester对象需可被 pickle 传递，输入参数也需能在进程间序列化。

在 DeepPavlov 配置中注册与使用

两个组件都已写入组件注册表 deeppavlov/core/common/registry.json：

"api_requester": "deeppavlov.models.api_requester.api_requester:ApiRequester", "api_router": "deeppavlov.models.api_requester.api_router:ApiRouter",

注册名分别为api_requester与api_router（对应源码中的@register('api_requester')与@register("api_router")装饰器），因此可以在任意 DeepPavlov 管线配置的chainer.units中以字符串引用。一个典型的配置片段形如：

{ "chainer": { "in": ["x"], "out": ["api_result"], "pipe": [ { "class_name": "api_requester", "in": ["x"], "out": ["api_result"], "url": "http://localhost:8000/api", "param_names": ["text"], "out": 1, "debatchify": false } ] } }

参数映射关系如下：in指定组件输入名（ApiRequester在未显式传param_names时会将其作为请求字段名回退来源）；out列表中的名字即该组件输出在 Chainer 内存中的变量名，同时被out_count用于判定返回结构；url、param_names、debatchify直接透传给构造器。从 chainer.py 的_compute可以看出，Chainer 会把上一个组件的输出按in取出，以位置参数方式调用组件，再依据out_params长度决定单值存储或多值解包——这与ApiRequester中out_count > 1时的zip(*response)转置逻辑正好配套。

与 KBQA 组件的联动：源码级证据

api_requester模块并非孤立存在，它在 KBQA 知识问答链中承担"把重计算外包给独立服务"的角色。从源码可以确认三处明确的联动开关：

1. query_generator_base.py 中QueryGeneratorBase构造器接受use_wp_api_requester与use_el_api_requester两个布尔参数，docstring 明确说明其含义为"该组件是否使用deeppavlov.models.api_requester.api_requester组件来调用 Wiki Parser / Entity Linking"；
2. 在 query_generator_base.py 的get_entity_ids中，开启use_el_api_requester时会对实体链接输出做一层解包（el_output = el_output[0]）以适应 API 请求器的返回形态，未开启时则进一步取entity_ids[0]——这印证了两者数据形态的差异；
3. 在 query_generator_base.py 的find_top_rels中，开启use_wp_api_requester时从 Wiki Parser 输出中抽取rel[0]作为关系候选；
4. rel_ranking_infer.py 中RelRankerInfer也提供use_api_requester参数，docstring 说明其含义是"wiki parser 是否作为外部 API 使用"。

这些开关说明：ApiRequester的设计目标之一，就是让 KBQA 中成本较高的 Wikidata 查询（WikiParser的find_rels、find_label等）与实体链接推理可以部署为独立 HTTP 服务，管线只负责转发与接收。

使用注意事项与边界

结合源码与仓库现状，以下几点对实际使用尤为关键：

• 阻塞式请求：默认模式下requests.post是同步阻塞的，高并发场景建议开启debatchify以利用run_in_executor的线程池并发，或借助ApiRouter做进程级并行；
• 异常处理边界：源码未对网络错误、非 JSON 响应做显式兜底，调用方需自行保证 API 端点可用性与响应格式，否则异常会直接冒泡到管线；
• out_count的一致性：out同时承担"返回值个数"与"Chainer 输出名列表"两种职责，多返回值 API 必须确保响应 JSON 的结构顺序与out列表顺序一致；
• 进程池约束：ApiRouter依赖ProcessPoolExecutor，其中的ApiRequester对象与输入参数必须可 pickle；
• 组件生命周期：作为Component子类，ApiRequester不涉及模型参数保存/加载（无save/load逻辑），但它持有的子组件会被 component.py 的destroy递归清理。

结语

api_requester模块以极小的接口面（两个类、两个公开方法）解决了 DeepPavlov 管线与外部服务集成的通用问题：ApiRequester负责把管线的中间结果按字段映射成 JSON 请求并解析响应，debatchify与get_async_response补齐了"端点不支持批处理"的并发短板，ApiRouter则提供了多请求器并行调度能力。从注册表到 KBQA 组件的参数开关，仓库内的每一处引用都指向同一个用途——把计算密集的知识库查询和实体链接等环节，从本地推理进程中解耦为可独立部署、可水平扩展的 HTTP 服务。

• 人工智能
• NLP
• 深度学习