企业官网建设流程全解析

• 人工智能
• 大模型
• 预训练
• 微调
• LoRA
• RLHF
• 强化学习
• 分布式训练

本篇技术指南围绕 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( title="pipelines REST API", debug=True, version=pipelines_version, root_path=ROOT_PATH, ) application.add_middleware(CORSMiddleware, ...) application.add_exception_handler(HTTPException, http_error_handler) application.include_router(api_router) return application

• title / version：title固定为pipelines REST API，version则优先从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"])

由此，最终对外暴露的端点被清晰划分为四组：

四、search 控制器：核心查询端点详解

search.py是 REST API 中最核心的控制器。模块加载时，会通过Pipeline.load_from_yaml()从 YAML 配置中实例化查询 Pipeline（QUERY_PIPELINE）和可选的 QA 对生成 Pipeline（QA_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.py：

class QueryRequest(BaseModel): query: str params: Optional[dict] = None debug: Optional[bool] = False class Config: extra = Extra.forbid

• query：问题文本（必填）。
• params：可选，透传给 Pipeline 的额外参数，可用于指定节点级过滤器等。
• debug：可选，是否返回调试信息。
• extra = Extra.forbid：请求体中出现任何未定义字段都会报错，避免静默失败。

响应模型QueryResponse包含query、answers、documents与可选别名_debug字段。处理请求的核心逻辑在_process_request()中完成，它在调用pipeline.run()之前会做两件关键的事：

1. 过滤器格式化：将全局顶层过滤器（params["filters"]）和节点级过滤器（如params["Retriever"]["filters"]）统一交给_format_filters()处理——把非列表的值包装成列表、剔除值为null的过滤条件，并对已废弃的过滤格式打印告警。
2. 结果兜底与序列化：确保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：文生图查询，返回QueryImageResponse（answers为字符串列表）。
• 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 中的RequestLimiter：

class RequestLimiter: def __init__(self, limit): self.semaphore = Semaphore(limit - 1) @contextmanager def run(self): acquired = self.semaphore.acquire(blocking=False) if not acquired: raise HTTPException(status_code=503, detail="The 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 控制器：文件上传与索引 Pipeline

file_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 模型）：

• FileConverterParams：remove_numeric_tables、valid_languages（文本/PDF 转换参数）。
• PreprocessorParams：clean_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_paths=file_paths, meta=file_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实现了一套完整的用户反馈闭环，用于沉淀标注数据并评估模型效果：

其中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, host="0.0.0.0", port=port)启动服务，监听来自任意主机的请求。

所有关键配置均集中在 rest_api/config.py，且全部支持环境变量覆盖：

仓库根目录下还提供了 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）；
• 查询请求的过滤器传递（全局过滤器、节点级过滤器、过滤器列表形式，以及无效过滤器返回空答案）；
• 无文档/无答案等空结果场景的兜底逻辑。

这些测试既是服务行为的规格说明书，也是二次开发时验证端点改动正确性的直接手段。

• 人工智能
• 大模型
• 预训练
• 微调
• LoRA
• RLHF
• 强化学习
• 分布式训练