PaddleNLP pipelines REST API 应用模块深度解析：FastAPI 服务启动、路由装配与 OpenAPI 文档生成

人工智能大模型预训练微调LoRARLHF强化学习分布式训练【免费下载链接】PaddleNLPEasy-to-use and powerful LLM and SLM library with awesome model zoo.项目地址https://gitcode.com/gh_mirrors/pa/PaddleNLP点击查看免费下载本篇技术指南围绕 PaddleNLP 仓库中slm/pipelines子项目的 REST API 服务层展开系统讲解其入口应用模块pipelines.rest_api.application如何基于 FastAPI 完成服务实例创建、CORS 中间件配置、异常处理器注册与路由装配并串联起 search、feedback、file-upload、document 四大控制器及其背后的 Pipeline 加载机制。读完本文你将掌握该服务的启动方式、端点清单、请求/响应数据模型、并发限流策略以及如何在当前仓库中定位并扩展这套 REST API 服务。一、模块定位REST API 在 pipelines 中的角色slm/pipelines是 PaddleNLP 下的一个独立子项目它提供了一套基于 Pipeline 编排的检索、问答、文件解析与语义搜索能力。为了让这些能力可以被 HTTP 客户端浏览器、curl、第三方应用方便地调用仓库在 slm/pipelines/rest_api 目录下实现了一套完整的 REST API 服务层。整个rest_api包的结构如下rest_api/ ├── application.py # 应用入口创建 FastAPI 实例、注册中间件与路由 ├── config.py # 全部环境变量配置的读取与默认值 ├── schema.py # 请求/响应 Pydantic 数据模型 ├── controller/ # 四个路由控制器 │ ├── router.py # 汇总所有子路由的 APIRouter │ ├── search.py # 查询类端点/query、/query_documents 等 │ ├── feedback.py # 用户反馈类端点/feedback、/eval-feedback 等 │ ├── file_upload.py # 文件上传与索引端点/file-upload 等 │ ├── document.py # 文档检索与删除端点/documents/get_by_filters 等 │ └── utils.py # RequestLimiter 限流器与 as_form 辅助装饰器 └── pipeline/ # 各场景的 Pipeline YAML 定义 ├── pipelines.yaml # 默认 Pipeline 配置含 query/indexing 等 └── ...chatfile、semantic_search、senta 等场景本文档聚焦的 application.md 正是对该模块的 API 文档入口其对应的实现位于 rest_api/application.py。控制器模块的文档见 controller.md。二、get_application()FastAPI 服务实例的创建application.py的核心函数是get_application()它负责创建并配置一个完整的 FastAPI 应用实例。其逻辑可以拆解为以下四个步骤def get_application() - FastAPI: application FastAPI( titlepipelines REST API, debugTrue, versionpipelines_version, root_pathROOT_PATH, ) application.add_middleware(CORSMiddleware, ...) application.add_exception_handler(HTTPException, http_error_handler) application.include_router(api_router) return applicationtitle / versiontitle固定为pipelines REST APIversion则优先从pipelines.__version__读取当作为开发模式运行无法导入 pipelines 包时降级为0.0.0。root_path取自定义环境变量ROOT_PATH默认/用于支持将服务挂载在反向代理的子路径下。CORS 中间件通过CORSMiddleware允许浏览器跨域访问默认对所有来源、方法、请求头开放allow_origins[*]等。源码注释也明确提示在生产部署中建议收紧这一配置。异常处理注册HTTPException的统一处理函数http_error_handler定义在 rest_api/controller/errors/http_error.py保证错误响应格式统一。路由装配调用application.include_router(api_router)将 controller/router.py 中的全部路由挂载到应用上。启动前还会执行use_route_names_as_operation_ids(app)遍历所有APIRoute把operation_id设置为路由对应的 Python 方法名。这样自动生成的 API 客户端会得到更简洁的函数名例如客户端方法直接以query、upload_file命名。三、路由装配四大控制器如何汇总router.py是整个 API 的路由中枢它创建一个空的APIRouter然后按标签依次挂载四个子控制器router APIRouter() router.include_router(search.router, tags[search]) router.include_router(feedback.router, tags[feedback]) router.include_router(file_upload.router, tags[file-upload]) router.include_router(document.router, tags[document])由此最终对外暴露的端点被清晰划分为四组标签端点功能searchGET /initialized、GET /hs_version、POST /query、POST /chatfile_query、POST /query_images、POST /query_text_to_images、POST /query_documents、POST /senta_file、POST /query_qa_pairs各类查询、健康检查与版本信息feedbackPOST /feedback、GET /feedback、DELETE /feedback、POST /eval-feedback、GET /export-feedback用户反馈的提交、查询、删除、评估与导出file-uploadPOST /file-upload-qa-generate、POST /file-upload、POST /file-upload-splitter、GET /files文件上传、解析、索引与结果文件下载documentPOST /documents/get_by_filters、POST /documents/delete_by_filters按元数据过滤查询/删除文档四、search 控制器核心查询端点详解search.py是 REST API 中最核心的控制器。模块加载时会通过Pipeline.load_from_yaml()从 YAML 配置中实例化查询 PipelineQUERY_PIPELINE和可选的 QA 对生成 PipelineQA_PAIR_PIPELINE并从中取出文档存储DOCUMENT_STORE。默认的查询 Pipeline 定义在 pipeline/pipelines.yaml其拓扑为pipelines: - name: query type: Query nodes: - name: Retriever inputs: [Query] - name: Reader inputs: [Retriever]即「DensePassageRetriever 召回 ErnieReader 抽取式阅读理解」的经典抽取式问答链路。4.1 健康检查与版本端点router.get(/initialized) def check_status(): return True router.get(/hs_version) def pipelines_version(): return {hs_version: pipelines.__version__}GET /initialized返回true表示服务已就绪客户端可设置约 500ms 的短超时轮询该端点来判断服务是否仍在加载中GET /hs_version返回当前 pipelines 运行时版本。4.2 POST /query核心问答端点POST /query接收一个QueryRequest其数据模型定义在 rest_api/schema.pyclass QueryRequest(BaseModel): query: str params: Optional[dict] None debug: Optional[bool] False class Config: extra Extra.forbidquery问题文本必填。params可选透传给 Pipeline 的额外参数可用于指定节点级过滤器等。debug可选是否返回调试信息。extra Extra.forbid请求体中出现任何未定义字段都会报错避免静默失败。响应模型QueryResponse包含query、answers、documents与可选别名_debug字段。处理请求的核心逻辑在_process_request()中完成它在调用pipeline.run()之前会做两件关键的事过滤器格式化将全局顶层过滤器params[filters]和节点级过滤器如params[Retriever][filters]统一交给_format_filters()处理——把非列表的值包装成列表、剔除值为null的过滤条件并对已废弃的过滤格式打印告警。结果兜底与序列化确保documents、answers、result字段始终存在若文档中的 embedding 是 numpyndarray则转换为list[float]以便 JSON 序列化。最后以 JSON 形式记录请求、响应与耗时日志。4.3 其余查询端点速览POST /chatfile_query面向 ChatFile 场景的查询响应模型Chatfile_QueryResponse额外包含result字段。POST /query_images接收图片文件List[UploadFile]与 JSON 序列化的meta表单字段将文件写入上传目录后交给 Pipeline 处理。POST /query_text_to_images文生图查询返回QueryImageResponseanswers为字符串列表。POST /query_documents按meta元数据查询文档响应为DocumentResponse。POST /senta_file情感分析场景端点响应包含img_dict。POST /query_qa_pairs调用QA_PAIR_PIPELINE返回过滤后的 CQA 三元组filtered_cqa_triples。4.4 并发限流RequestLimiter为避免高并发请求压垮 GPU/CPU 推理服务search 控制器引入了 utils.py 中的RequestLimiterclass RequestLimiter: def __init__(self, limit): self.semaphore Semaphore(limit - 1) contextmanager def run(self): acquired self.semaphore.acquire(blockingFalse) if not acquired: raise HTTPException(status_code503, detailThe server is busy processing requests.) ...当并发请求数超过配置上限CONCURRENT_REQUEST_PER_WORKER默认 4时请求会立即收到503 The server is busy processing requests.从而保护推理 Pipeline 不被击穿。五、document 控制器文档的过滤查询与删除document.py提供两个端点均通过DOCUMENT_STORE即查询 Pipeline 关联的文档存储操作文档POST /documents/get_by_filters根据过滤器返回文档列表响应模型为List[DocumentSerialized]。示例过滤器{filters: {name: [some, more], category: [only_one]}}传入空字典{filters: {}}即返回全部文档。出于响应体积考虑返回的文档中embedding字段会被置为None。POST /documents/delete_by_filters根据同样的过滤器删除文档并返回true传入空字典即可清空整个文档存储。在测试文件 test/test_rest_api.py 中test_get_documents与test_delete_documents验证了按meta_key、meta_index等元数据字段过滤查询和删除文档的完整行为先统计文档数量再按条件删除并断言剩余数量。六、file-upload 控制器文件上传与索引 Pipelinefile_upload.py负责把上传的文件喂给 Indexing Pipeline使文档能够被解析、切分并写入文档存储。模块加载时读取 YAML 中的indexing与indexing_qa_generating两个 Pipeline 定义若配置中不存在 Indexing Pipeline则相关端点会返回501错误。默认indexingPipeline 的拓扑见 pipelines.yaml为File → FileTypeClassifier → TextFileConverter / PDFFileConverter / DocxFileConverter / ImageFileConverter → Preprocessor → Retriever → DocumentStore即按文件类型分流转换 → 文本预处理 → 向量化召回 → 写入文档存储的完整索引链路。6.1 POST /file-upload该端点接收文件列表、JSON 序列化的meta字符串以及表单参数fileconverter_params和preprocessor_params分别对应FileConverterParams与PreprocessorParams两个as_form装饰的 Pydantic 模型FileConverterParamsremove_numeric_tables、valid_languages文本/PDF 转换参数。PreprocessorParamsclean_whitespace、clean_empty_lines、clean_header_footer、split_by、split_length、split_overlap、split_respect_sentence_boundary文本切分参数。上传的文件会以uuid4().hex 原文件名的形式保存到FILE_UPLOAD_PATH目录并附带name元数据随后调用INDEXING_PIPELINE.run( file_pathsfile_paths, metafile_metas, params{ TextFileConverter: fileconverter_params.dict(), PDFFileConverter: fileconverter_params.dict(), Preprocessor: preprocessor_params.dict(), }, )6.2 其余上传端点POST /file-upload-qa-generate驱动indexing_qa_generatingPipeline该链路在转换文件后依次经过AnswerExtractorPreprocessor → AnswerExtractor → QuestionGenerator → QAFilter → QAFilterPostprocessor实现「从文档抽取答案并自动生成问答对」后再写入文档存储。POST /file-upload-splitter仅接收文件与meta直接透传给 Indexing Pipeline适合已预置切分参数的场景。GET /files根据file_name返回解析后的结果文件FileResponse文件默认从FILE_PARSE_PATH目录读取。七、feedback 控制器反馈闭环与模型迭代feedback.py实现了一套完整的用户反馈闭环用于沉淀标注数据并评估模型效果端点方法说明/feedbackPOST提交用户反馈LabelSerialized/CreateLabelSerializedorigin缺省时记为user-feedback写入DOCUMENT_STORE/feedbackGET读取全部反馈标签/feedbackDELETE删除所有origin user-feedback的反馈标签/eval-feedbackPOST基于反馈计算answer_accuracy、document_accuracy与反馈条数n_feedback/export-feedbackGET将反馈导出为 SQuAD 格式 JSON供下游模型训练使用其中POST /eval-feedback的请求体FilterRequest允许通过document_id等字段过滤统计范围GET /export-feedback提供context_size默认 100000用于限制上下文长度、full_document_context是否使用完整文档作为 context与only_positive_labels三个查询参数导出时会自动做 offset 对齐校验并落盘为feedback_squad_direct.json。测试文件中的FEEDBACK样例即演示了提交一条 PDF 问答反馈的完整 JSON 结构。八、服务启动方式与核心配置项application.py的模块级代码在导入时即完成应用构建与路由 operation_id 简化并输出两条日志提示使用方式访问http://127.0.0.1:8000/docs查看 Swagger API 文档或直接调用POST /query例如curl --request POST --url http://127.0.0.1:8000/query \ -H Content-Type: application/json \ --data {query: Who is the father of Arya Stark?}当以脚本方式运行python application.py port时会通过uvicorn.run(app, host0.0.0.0, portport)启动服务监听来自任意主机的请求。所有关键配置均集中在 rest_api/config.py且全部支持环境变量覆盖环境变量默认值作用PIPELINE_YAML_PATHrest_api/pipeline/pipelines.yamlPipeline YAML 配置文件路径QUERY_PIPELINE_NAMEquery查询 Pipeline 名称QUERY_QA_PAIRS_NAMEquery_qa_pairsQA 对查询 Pipeline 名称INDEXING_PIPELINE_NAMEindexing索引 Pipeline 名称INDEXING_QA_GENERATING_PIPELINE_NAMEindexing_qa_generating问答对生成索引 Pipeline 名称FILE_UPLOAD_PATHrest_api/file-upload上传文件落盘目录FILE_PARSE_PATHparse_files解析结果文件目录LOG_LEVELINFO日志级别ROOT_PATH/FastAPI root_path用于反向代理子路径挂载CONCURRENT_REQUEST_PER_WORKER4每 worker 的最大并发请求数仓库根目录下还提供了 Docker 化的部署入口slm/pipelines/docker/docker-compose.yml 与 run_server.sh以及各业务场景的启动脚本例如问答场景 examples/question-answering/run_qa_server.sh、FAQ 场景 examples/FAQ/run_faq_server.sh、语义检索场景 examples/semantic-search/run_search_server.sh。这些脚本展示了如何通过环境变量为不同业务定制 Pipeline 配置并拉起服务。九、测试验证REST API 的行为保障仓库为 REST API 提供了完整的自动化测试 test/test_rest_api.py它基于 FastAPI 的TestClient直接对application.py中构建的app发起测试请求。测试通过环境变量将PIPELINE_YAML_PATH指向测试专用的test_pipeline.yaml并在每个用例前后清理文档与反馈数据覆盖了以下关键行为文档过滤查询与删除含删除后数量断言文件上传含无meta、非法meta两种边界非法 meta 预期返回500查询请求的过滤器传递全局过滤器、节点级过滤器、过滤器列表形式以及无效过滤器返回空答案无文档/无答案等空结果场景的兜底逻辑。这些测试既是服务行为的规格说明书也是二次开发时验证端点改动正确性的直接手段。赞分享人工智能大模型预训练微调LoRARLHF强化学习分布式训练【免费下载链接】PaddleNLPEasy-to-use and powerful LLM and SLM library with awesome model zoo.项目地址https://gitcode.com/gh_mirrors/pa/PaddleNLP点击查看免费下载相关推荐Lightdash API 自动生成体系深度解析TSOA 驱动的 Express 路由与 OpenAPI 文档实践Lightdash API 自动生成体系深度解析TSOA 驱动的 Express 路由与 OpenAPI 文档实践 本文以 Lightdash 后端 gene后端前端数据分析数据可视化人工智能AI AgentFlink 文档生成器flink-docs模块深度解析从源码自动生成 REST API 与配置文档的完整指南Flink 文档生成器flink docs模块深度解析从源码自动生成 REST API 与配置文档的完整指南 导读 本文围绕 Flink 仓库中的 fli后端大数据流处理批处理AWX REST API Reference 深度解析Swagger UI 驱动的交互式 API 文档与 OpenAPI Schema 生成机制AWX REST API Reference 深度解析Swagger UI 驱动的交互式 API 文档与 OpenAPI Schema 生成机制 AWX 官方后端运维任务调度上一篇symfony/var-dumper源码解析DOMCaster下一篇OctoPrint JS 客户端库 Socket 模块完全指南SockJS 实时通信、消息订阅与通信节流机制创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考