企业官网建设流程全解析

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/BaseRunner（base.py）：从 YAML 或 trial 目录加载流水线，提供单次查询的run()方法；
• ApiRunner（api.py）：在BaseRunner之上构建 Quart 异步 Web 服务，暴露 4 个 HTTP 端点；
• GradioRunner（gradio.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_dir=project_dir, **module_params, ) self.module_instances.append(module_instance) self.module_params.append(module_params)

2.2 两种加载方式

BaseRunner提供两个类方法作为标准入口：

• from_yaml(yaml_path, project_dir=None)：从 YAML 文件加载。该 YAML 必须是由extract_best_config提取出来的单模块配置，project_dir默认为当前目录；
• from_trial_folder(trial_path)：直接从已评估的 trial 目录加载。它会自动调用extract_best_config生成配置，并把project_dir设置为 trial 目录的父目录（即项目目录）。

底层链路是：extract_best_config（base.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_column="generated_texts")方法（base.py）。其执行方式与评估器一脉相承：先构造一行"伪 QA 数据"（qid用uuid4()生成、retrieval_gt为空列表、generation_gt为空串），然后让每个模块实例依次调用module_instance.pure(previous_result=previous_result, **module_param)，再对 DataFrame 做列合并：

duplicated_columns = previous_result.columns.intersection(new_result.columns) drop_previous_result = previous_result.drop(columns=duplicated_columns) previous_result = pd.concat([drop_previous_result, new_result], axis=1)

最终从result_column（默认generated_texts，即 generation 模块的输出列）取出答案。一个前提是：流水线的第一个模块必须是query_expansion或retrieval，以保证流水线以单个 query 为起点。

三、ApiRunner：把流水线发布为 REST API

ApiRunner在BaseRunner基础上用 Quart（异步 Flask 兼容框架）构建服务（api.py）。构造函数会加载项目目录下的data/corpus.parquet（用于把检索到的 doc_id 反查为正文内容），并注册全部路由。

3.1 启动 API 服务器

Python 方式（api_endpoint.md）：

from autorag.deploy import ApiRunner import nest_asyncio nest_asyncio.apply() # 方式一：从 YAML 配置加载 runner = ApiRunner.from_yaml('your/path/to/pipeline.yaml', project_dir='your/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 8000

run_api_server的签名与参数（api.py）：

注意：nest_asyncio.apply()是必需的——AutoRAG 的流水线内部已有事件循环，Quart 又是异步框架，需要在已有事件循环中再次运行，因此必须借助nest_asyncio打补丁（CLI 的run_api命令内部同样调用了nest_asyncio.apply()，见 cli.py）。

3.2 端点一：/v1/run（POST）——完整问答

接收 JSON 请求体：

请求示例：

curl -X POST "http://example.com:8000/v1/run" \ -H "Content-Type: application/json" \ -d '{"query": "example query", "result_column": "generated_texts"}'

响应（200 OK，application/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/retrieve（POST）——仅检索

该端点跳过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/stream（POST）——流式输出

这是面向真实对话体验的核心端点（api.py）。响应为text/event-stream，按顺序先推送retrieved_passage类型的消息（每条携带passage_index表示段落序号），随后进入生成阶段：取出previous_result["prompts"]中的 prompt，调用生成器模块的异步流式接口module_instance.astream(prompt=prompt, **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]

即：type=generated_text时仅generated_text有值；type=retrieved_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-buffer

3.5 端点四：/version（GET）——版本查询

读取包内的VERSION文件并返回：

curl -X GET "http://example.com:8000/version"

响应：

{ "version": "当前版本号" }

3.6 源码视角：检索段落如何被组装

/v1/run、/v1/retrieve、/v1/stream三个端点最终都会调用extract_retrieve_passage（api.py）把检索结果组装成响应结构。它按优先级从流水线中间结果中挑选检索列：

1. 存在retrieved_ids→ 使用retrieved_ids/retrieve_scores（综合检索）；
2. 否则若存在retrieved_ids_semantic→ 使用语义检索列；
3. 否则回退到retrieved_ids_lexical/retrieve_scores_lexical（词法检索）。

随后通过fetch_contents（util.py）按 doc_id 从corpus.parquet反查content、path、metadata、start_end_idx等列，最终映射为RetrievedPassage（file_page取自metadata["page"]，start_idx/end_idx取自start_end_idx二元组）。这意味着 API 返回的溯源信息（文件路径、页码、段落起止位置）完整来自语料库元数据，可放心用于证据引用。

3.7 通过 ngrok 暴露公网

run_api_server(remote=True)（默认开启）会通过pyngrok自动创建公网隧道，并在日志中打印Public API URL：

INFO [api.py:199] >> Public API URL: api.py:199 https://8a31-14-52-132-205.ngrok-free.app

拿到该 URL 后，把请求的 host 换成它即可从公网访问本地服务。

四、GradioRunner：对话式 Web 界面

GradioRunner（gradio.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/trial

Python 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_name="0.0.0.0", server_port=7680, share=True)

run_web参数（gradio.py）：

界面标题为 "📚 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_column="generated_texts"): url = f"{BASE_URL}/v1/run" payload = { "query": query, "result_column": result_column } response = requests.post(url, json=payload) if response.status_code == 200: return response.json() else: response.raise_for_status() def stream_query(query, result_column="generated_texts"): url = f"{BASE_URL}/v1/stream" payload = { "query": query, "result_column": result_column } with requests.Session() as session: response = session.post(url, json=payload, stream=True) 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_size=None)): 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(f"Request failed with status code: {response.status_code}") print(f"Response 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_bytes（util.py）：由于流式消息是逐条 JSON 字节串，且可能在一个 chunk 中粘连多条，需要该工具把字节块解码为多个 JSON 对象再按type分派——这正是StreamResponse模型双类型设计的落地用法。

七、部署流程总结与注意事项

完整部署链路可归纳为四步：

1. 评估：用Evaluator对流水线进行带数据评估，产出包含summary.csv与config.yaml的 trial 目录；
2. 提取：部署时自动（from_trial_folder）或手动（extract_best_config）把多模块配置收敛为单模块最优配置；
3. 选择形态：需要程序化访问选ApiRunner（REST/流式），需要人机交互选GradioRunner（Web 对话）；
4. 启动与调用：通过 Python API、CLI 或 curl 启动并调用。

需要留意的约束：

• 单模块限制：部署配置中每个节点只能有 1 个模块，多模块配置会直接抛出ValueError；
• 首节点约束：流水线必须以query_expansion或retrieval开头，保证单 query 输入成立；
• nest_asyncio：Python 方式启动 API 前必须nest_asyncio.apply()；
• 语料依赖：API 的溯源字段（filepath、file_page、start_idx/end_idx）依赖corpus.parquet中是否含有path、metadata、start_end_idx列，缺列时对应字段返回null；
• 目录约定：ApiRunner需要项目目录下的data/corpus.parquet，extract_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