AutoRAG 部署实战：将优化后的 RAG 流水线发布为 API 服务与 Web 界面

AutoRAG 部署实战将优化后的 RAG 流水线发布为 API 服务与 Web 界面【免费下载链接】AutoRAGAutoRAG: Now your agent can find anything in your computer. It gets smarter if you are using it frequently.项目地址: https://gitcode.com/GitHub_Trending/au/AutoRAG导读AutoRAG 在完成 RAG 流水线的构建与评估后还需要把选出的最优流水线投入实际使用。autorag.deploy包正是解决这一环节的统一部署入口它提供了基于 Quart 异步框架的 REST API 服务ApiRunner、基于 Gradio 的对话式 Web 界面GradioRunner以及承载二者共同逻辑的BaseRunner/Runner基类。读完本文你将掌握如何把评估得到的 trial 目录或提取出的 YAML 配置一键加载为可运行的 RAG 服务理解/v1/run、/v1/retrieve、/v1/stream、/version四个 API 端点的数据流与请求/响应格式并能通过 CLI、Python 客户端与curl完成端到端调用。本文以 autorag.deploy.rst 这一 API 规范文档为骨架结合 api.py、base.py、gradio.py 源码及配套的 API 端点指南、Web 界面指南 展开。一、deploy 包结构总览autorag.deploy是 AutoRAG 的部署子包顶层__init__.py向外暴露了全部核心类与函数Runner/BaseRunnerbase.py从 YAML 或 trial 目录加载流水线提供单次查询的run()方法ApiRunnerapi.py在BaseRunner之上构建 Quart 异步 Web 服务暴露 4 个 HTTP 端点GradioRunnergradio.py在BaseRunner之上构建 Gradio 聊天界面extract_best_config、summary_df_to_yaml、extract_node_line_names、extract_node_strategy等工具函数负责从评估过的 trial 目录中提取最优流水线配置。整个模块只依赖流水线执行层的统一抽象因此无论是哪种节点组合检索、重排、生成……的流水线部署方式完全一致这也是评估完即可一键部署的设计基础。二、BaseRunner 与 Runner一切部署的统一底座BaseRunner是 API 与 Web 两种部署形态共享的核心抽象负责把优化好的流水线配置翻译成一串可顺序执行的模块实例。2.1 流水线实例化原理在 base.py 的构造函数中BaseRunner读取配置里的node_lines逐条逐节点检查for node_line in node_lines: for node in node_line[nodes]: if len(node[modules]) ! 1: raise ValueError( The number of modules in a node must be 1 for using runner. Please use extract_best_config method for extracting yaml file from evaluated trial. )这里体现了一个重要约束用于部署的流水线每个节点只能保留 1 个模块。因为评估阶段每个节点会并行尝试多种模块并择优而部署阶段只需要最优的那个。所以部署前必须先通过extract_best_config()把多模块配置收敛为单模块配置。随后通过get_support_modules(module_type)见 support.py实例化每个模块并收集其参数module_instance get_support_modules(module_type)( project_dirproject_dir, **module_params, ) self.module_instances.append(module_instance) self.module_params.append(module_params)2.2 两种加载方式BaseRunner提供两个类方法作为标准入口from_yaml(yaml_path, project_dirNone)从 YAML 文件加载。该 YAML 必须是由extract_best_config提取出来的单模块配置project_dir默认为当前目录from_trial_folder(trial_path)直接从已评估的 trial 目录加载。它会自动调用extract_best_config生成配置并把project_dir设置为 trial 目录的父目录即项目目录。底层链路是extract_best_configbase.py读取{trial_path}/summary.csv与{trial_path}/config.yaml经summary_df_to_yaml把每个节点的best_module_name、best_module_params与strategy重组为部署配置再补充vectordb配置读取resources/vectordb.yaml若为空则回退到默认的 chroma 持久化配置见extract_vectordb_config。因此评估 → 提取 → 部署整条链路是闭合的。2.3 Runner.run单次查询的执行引擎Runner继承BaseRunner补上了最常用的run(query, result_columngenerated_texts)方法base.py。其执行方式与评估器一脉相承先构造一行伪 QA 数据qid用uuid4()生成、retrieval_gt为空列表、generation_gt为空串然后让每个模块实例依次调用module_instance.pure(previous_resultprevious_result, **module_param)再对 DataFrame 做列合并duplicated_columns previous_result.columns.intersection(new_result.columns) drop_previous_result previous_result.drop(columnsduplicated_columns) previous_result pd.concat([drop_previous_result, new_result], axis1)最终从result_column默认generated_texts即 generation 模块的输出列取出答案。一个前提是流水线的第一个模块必须是query_expansion或retrieval以保证流水线以单个 query 为起点。三、ApiRunner把流水线发布为 REST APIApiRunner在BaseRunner基础上用 Quart异步 Flask 兼容框架构建服务api.py。构造函数会加载项目目录下的data/corpus.parquet用于把检索到的 doc_id 反查为正文内容并注册全部路由。3.1 启动 API 服务器Python 方式api_endpoint.mdfrom autorag.deploy import ApiRunner import nest_asyncio nest_asyncio.apply() # 方式一从 YAML 配置加载 runner ApiRunner.from_yaml(your/path/to/pipeline.yaml, project_diryour/project/directory) runner.run_api_server() # 方式二从 trial 目录加载 runner ApiRunner.from_trial_folder(/your/path/to/trial_dir) runner.run_api_server()CLI 方式autorag run_api --trial_dir /trial/dir/0 --host 0.0.0.0 --port 8000run_api_server的签名与参数api.py参数默认值说明host0.0.0.0服务监听地址port8000服务监听端口remoteTrue是否通过 ngrok 暴露到公网**kwargs—透传给 Quart/Flaskapp.run的其他参数注意nest_asyncio.apply()是必需的——AutoRAG 的流水线内部已有事件循环Quart 又是异步框架需要在已有事件循环中再次运行因此必须借助nest_asyncio打补丁CLI 的run_api命令内部同样调用了nest_asyncio.apply()见 cli.py。3.2 端点一/v1/runPOST——完整问答接收 JSON 请求体字段类型必填说明querystring是用户查询result_columnstring否结果列名默认generated_texts请求示例curl -X POST http://example.com:8000/v1/run \ -H Content-Type: application/json \ -d {query: example query, result_column: generated_texts}响应200 OKapplication/json{ result: 生成的答案文本, retrieved_passage: [ { content: 段落内容, doc_id: 文档ID, score: 0.98, filepath: 文件路径, file_page: 2, start_idx: 100, end_idx: 150 } ] }其中filepath、file_page、start_idx、end_idx均为可空字段取决于语料库corpus.parquet中是否存在path、metadata、start_end_idx列。3.3 端点二/v1/retrievePOST——仅检索该端点跳过promptmaker与generator两类节点api.py 中通过isinstance判断并continue只返回检索到的段落适合构建检索即服务的下游应用curl -X POST http://example.com:8000/v1/retrieve \ -H Content-Type: application/json \ -d {query: latest trends in AI}响应体结构{ passages: [ { doc_id: doc123, content: Artificial Intelligence is transforming industries., score: 0.98, filepath: path/to/file, file_page: 2, start_idx: 100, end_idx: 150 } ] }若请求体缺少query字段会返回 400 与错误提示Invalid request. You need to include query in the request body.。3.4 端点三/v1/streamPOST——流式输出这是面向真实对话体验的核心端点api.py。响应为text/event-stream按顺序先推送retrieved_passage类型的消息每条携带passage_index表示段落序号随后进入生成阶段取出previous_result[prompts]中的 prompt调用生成器模块的异步流式接口module_instance.astream(promptprompt, **module_param)逐 delta 推送generated_text类型的消息。消息体由StreamResponse模型约束class StreamResponse(BaseModel): type: Literal[generated_text, retrieved_passage] generated_text: Optional[str] retrieved_passage: Optional[RetrievedPassage] passage_index: Optional[int]即typegenerated_text时仅generated_text有值typeretrieved_passage时仅retrieved_passage与passage_index有值其余字段为null。因此客户端只需按type字段分派即可。curl流式调用需--no-buffer关闭缓冲curl -X POST http://example.com:8000/v1/stream \ -H Content-Type: application/json \ -d {query: example query, result_column: generated_texts} \ --no-buffer3.5 端点四/versionGET——版本查询读取包内的VERSION文件并返回curl -X GET http://example.com:8000/version响应{ version: 当前版本号 }3.6 源码视角检索段落如何被组装/v1/run、/v1/retrieve、/v1/stream三个端点最终都会调用extract_retrieve_passageapi.py把检索结果组装成响应结构。它按优先级从流水线中间结果中挑选检索列存在retrieved_ids→ 使用retrieved_ids/retrieve_scores综合检索否则若存在retrieved_ids_semantic→ 使用语义检索列否则回退到retrieved_ids_lexical/retrieve_scores_lexical词法检索。随后通过fetch_contentsutil.py按 doc_id 从corpus.parquet反查content、path、metadata、start_end_idx等列最终映射为RetrievedPassagefile_page取自metadata[page]start_idx/end_idx取自start_end_idx二元组。这意味着 API 返回的溯源信息文件路径、页码、段落起止位置完整来自语料库元数据可放心用于证据引用。3.7 通过 ngrok 暴露公网run_api_server(remoteTrue)默认开启会通过pyngrok自动创建公网隧道并在日志中打印Public API URLINFO [api.py:199] Public API URL: api.py:199 https://8a31-14-52-132-205.ngrok-free.app拿到该 URL 后把请求的 host 换成它即可从公网访问本地服务。四、GradioRunner对话式 Web 界面GradioRunnergradio.py用 Gradio 的ChatInterface包装run()方法提供浏览器对话界面。之所以用 Gradio 而非 Streamlit官方文档给出的理由是Streamlit 必须在新进程中启动导致自定义模型无法在进程内复用而 Gradio 可在同一进程中运行。4.1 启动 Web 界面CLI 方式web.md# 方式一YAML 路径 autorag run_web --yaml_path your/path/to/pipeline.yaml # 指定项目目录 autorag run_web --yaml_path your/path/to/pipeline.yaml --project_dir your/project/directory # 方式二trial 路径 autorag run_web --trial_path your/path/to/trialPython Runner 方式from autorag.deploy import Runner runner Runner.from_yaml(your/path/to/pipeline.yaml) runner.run_web() runner Runner.from_trial_folder(your/path/to/trial_folder) runner.run_web(server_name0.0.0.0, server_port7680, shareTrue)run_web参数gradio.py参数默认值说明server_name0.0.0.0监听地址server_port7680监听端口shareFalse是否生成公网分享链接有效期 72 小时**kwargs—透传给gr.ChatInterface.launch界面标题为 AutoRAG并关闭了retry_btn与undo_btn。CLI 的run_web命令cli.py实际上是启动 Streamlit 的web.py一个独立进程并强制要求yaml_path与trial_path二选一、不可同时给出。4.2 界面效果CLI 启动的 Streamlit 版界面通过Runner.run_web()启动的 Gradio 版对话界面两种形态都是输入查询 → 实时获得流水线响应的交互环境适合在部署前快速验证流水线效果。五、OpenAPI 规范swagger.yml仓库内置了 OpenAPI 3.0 规范文件 swagger.yml完整描述了/v1/run、/v1/retrieve、/v1/stream、/version的请求/响应 schema包括StreamResponse中type字段的generated_text/retrieved_passage枚举语义type: type: string enum: - generated_text - retrieved_passage description: | When the type is generated_text, only generated_text is returned. The other fields are None. When the type is retrieved_passage, only retrieved_passage and passage_index are returned. The other fields are None.该文件可直接用于生成客户端 SDK、导入 Postman/Swagger UI 或编写契约测试是前后端联调的标准依据。六、Python 客户端调用示例以下示例来自 api_endpoint.md覆盖了三个 POST/GET 端点的完整调用逻辑import requests from autorag.utils.util import decode_multiple_json_from_bytes # Base URL of the API BASE_URL http://example.com:8000 # Replace with the actual base URL of the API def run_query(query, result_columngenerated_texts): url f{BASE_URL}/v1/run payload { query: query, result_column: result_column } response requests.post(url, jsonpayload) if response.status_code 200: return response.json() else: response.raise_for_status() def stream_query(query, result_columngenerated_texts): url f{BASE_URL}/v1/stream payload { query: query, result_column: result_column } with requests.Session() as session: response session.post(url, jsonpayload, streamTrue) retrieved_passages [] # This will store retrieved passages # Check if the request was successful if response.status_code 200: # Process the streaming response for i, chunk in enumerate(response.iter_content(chunk_sizeNone)): if chunk: data_list decode_multiple_json_from_bytes(chunk) for data in data_list: if data[type] retrieved_passage: retrieved_passages.append(data[retrieved_passage]) else: print(data[generated_text], end) # Stream the generated texts else: print(fRequest failed with status code: {response.status_code}) print(fResponse content: {response.text}) def get_version(): url f{BASE_URL}/version response requests.get(url) if response.status_code 200: return response.json() else: response.raise_for_status()流式响应的解析要点在于decode_multiple_json_from_bytesutil.py由于流式消息是逐条 JSON 字节串且可能在一个 chunk 中粘连多条需要该工具把字节块解码为多个 JSON 对象再按type分派——这正是StreamResponse模型双类型设计的落地用法。七、部署流程总结与注意事项完整部署链路可归纳为四步评估用Evaluator对流水线进行带数据评估产出包含summary.csv与config.yaml的 trial 目录提取部署时自动from_trial_folder或手动extract_best_config把多模块配置收敛为单模块最优配置选择形态需要程序化访问选ApiRunnerREST/流式需要人机交互选GradioRunnerWeb 对话启动与调用通过 Python API、CLI 或 curl 启动并调用。需要留意的约束单模块限制部署配置中每个节点只能有 1 个模块多模块配置会直接抛出ValueError首节点约束流水线必须以query_expansion或retrieval开头保证单 query 输入成立nest_asyncioPython 方式启动 API 前必须nest_asyncio.apply()语料依赖API 的溯源字段filepath、file_page、start_idx/end_idx依赖corpus.parquet中是否含有path、metadata、start_end_idx列缺列时对应字段返回null目录约定ApiRunner需要项目目录下的data/corpus.parquetextract_best_config需要项目目录下的resources/vectordb.yaml部署时务必保持项目目录结构完整。从 tests/autorag 的部署相关测试与 deploy 文档 可以看出这套部署体系把评估得到的最优流水线无缝衔接为可对外服务的 API/Web 应用是 AutoRAG 从实验到生产的关键一环。若需要更细粒度的端点 schema可对照 swagger.yml 与 api.py 阅读源码。【免费下载链接】AutoRAGAutoRAG: Now your agent can find anything in your computer. It gets smarter if you are using it frequently.项目地址: https://gitcode.com/GitHub_Trending/au/AutoRAG创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考