企业官网建设流程全解析

导读

本文围绕 RocketRide 开源仓库中的landing_ai节点包展开，系统讲解如何通过 Landing.ai 的 Agentic Document Extraction（ADE）服务，在 RocketRide 管道中完成「非结构化文档 → 干净 Markdown 与表格 → 符合 JSON Schema 的结构化字段」的完整处理链路。读完本文，你将掌握两个子节点（Landing.ai Parse、Landing.ai Extract）的 lane 接线方式、全部配置参数与默认值、API Key 认证机制，以及底层 Python 实现的运行细节与边界行为，可以直接在画布上搭建可运行的文档抽取管道。

Landing.ai ADE 与节点包的定位

Landing.ai 提供 Agentic Document Extraction（ADE），一种面向文档处理的云端服务，可以把非结构化文件转换成可用文本和结构化数据。在本仓库中，ADE 被封装为两个可组合的子节点，位于 nodes/src/nodes/landing_ai：

• Landing.ai Parse：接收文档字节，调用 ADE 的parse操作，输出干净的 Markdown 文本以及表格块；
• Landing.ai Extract：接收 Parse 产出的 Markdown，结合上传的 JSON Schema 调用 ADE 的extract操作，输出结构化答案（answers）或文档（documents）。

选择本包的前提是你需要同一个 ADE 服务同时提供解析与抽取两种操作；如果只需要可读的文档文本和表格，单独使用 Parse 即可。典型的管道接线为Parse → Extract，因为 Extract 消费的是 Parse 产生的 Markdown 而不是原始文档。

数据流与 Lanes

两个子节点的 lane 定义集中在各自的 service 描述文件里：services.parse.json 声明"lanes": {"tags": ["text", "table"]}，services.extract.json 声明"lanes": {"text": ["answers", "documents"]}。汇总如下：

Parse 的输入是二进制字节流（由 tag 流状态机接收），支持 PDF、图片、电子表格（XLSX/CSV）等输入类型；Extract 的输入是textlane 上的解析后 Markdown，因此必须放置在 Parse 下游。

配置模型与参数详解

两个服务都只有一个defaultprofile。节点读取配置时，get_node_config会先获取节点配置，再解包嵌套的default字典（landing_ai_base.py）：

config = Config.getNodeConfig(logical_type, conn_config) if not isinstance(config, dict): return {} inner = config.get('default') return inner if isinstance(inner, dict) else config

因此画布上的服务配置会落到landing_ai_parse.default/landing_ai_extract.default这样的对象字段上（见两个 service JSON 中的shape与fields.default定义）。

Landing.ai Parse 配置

Model控制发送给 ADEparse调用的模型值，默认dpt-2-latest，这也是当前节点唯一提供的取值（service JSON 的enum只有["dpt-2-latest", "DPT-2 (latest)"]）。除非节点配置被扩展支持其他 ADE 模型，否则保持默认即可。

Region默认production，需要欧盟部署时选择eu（对应 eu-west-1 数据驻留端点）。解析器会把任何其他配置值静默重置为production——对应源码 parse.py：

if self._region not in ('production', 'eu'): self._region = 'production'

所以请只使用production与eu两个取值。

Landing.ai Extract 配置

Extraction Schema（JSON Schema 文件）是服务元数据中的必填项（"required": ["landing_ai_extract.schema_file"]）。上传的必须是JSON 对象形式的>_MAX_SCHEMA_BYTES = 2 * 1024 * 1024 if len(raw) > _MAX_SCHEMA_BYTES: raise ValueError(f'... uploaded schema is too large ...') schema = json.loads(raw.decode('utf-8')) ... if not isinstance(schema, dict): raise ValueError('Landing.ai Extract: extraction schema must be a JSON object')

注意json.loads同时捕获RecursionError，用来拦截在体积上限之下、但嵌套深度病态的 JSON，避免解析时递归崩溃。Schema 内容被当作不透明数据，只重新序列化发送给 ADE，不会被读取字段值或写盘。

Schema 用于定义响应必须包含的字段，并支持 ADE 的x-alternativeNames与format关键字。上传错误会在校验阶段以warning形式报告，但不会阻塞画布编辑；真正运行时抽取调用会失败。service JSON 的preconfig.default自带一个最小可用示例：

{ "type": "object", "properties": { "summary": { "type": "string" } } }

Strict默认关闭。关闭时，文档不完全满足 Schema 也会返回部分抽取结果；开启后，不满足 Schema 的文档会失败而非产出部分结果。

Region与 Parse 行为一致（production/eu二选一），其他值同样被归一化为production（extract.py）。

完整参数表（由 nodes:docs-generate 生成）

Landing.ai Extract（services.extract.json）

Landing.ai Parse（services.parse.json）

认证：API Key 与环境变量回退

在每个服务的API Key字段填入 Landing.ai ADE Key，或在该字段为空时设置环境变量ROCKETRIDE_LANDING_AI_KEY。解析顺序由 resolve_api_key 实现：优先取配置中的api_key（去除首尾空白），为空则回退到环境变量，再为空返回None：

return (config.get('api_key') or '').strip() or (os.environ.get('ROCKETRIDE_LANDING_AI_KEY') or '').strip() or None

保存配置时，validateConfig会做一次便宜的只读校验：调用parse_jobs.list（landing_ai_base.py），该接口是只读的、不消耗额度。校验失败只输出 warning，不会阻止编辑管道：

client = build_client(api_key, config.get('region') or 'production') client.parse_jobs.list(page=0, page_size=1)

两个子节点的IGlobal.validateConfig（parse/IGlobal.py、extract/IGlobal.py）都用try/except包裹校验逻辑并统一降级为 warning；Extract 还会额外检查 schema_file 是否为空以及能否成功解码。

运行机制：从源码看底层调用链

Parse：tag 流状态机与 ADEparse调用

文档字节通过tagslane 流入，由 parse/IInstance.py 的writeTag状态机处理：识别OMET元数据标签、SBGN流开始、SDAT二进制数据块（通过 header 长度剥离 tag 头、累积document_data）、SEND流结束并触发处理。

处理前会先检查输出是否有人监听（IInstance.py）：如果text和table都没有 listener，则完全跳过 ADE 远程调用，这是节省额度的关键优化：

has_text_listener = self.instance.hasListener('text') has_table_listener = self.instance.hasListener('table') if not has_text_listener and not has_table_listener: debug('Landing.ai Parse: no text/table listeners connected; skipping parse') return

真正调用发生在Parser.parse（parse.py）：

• 未配置 API Key 时记录error并返回空文本、无表格（fail-closed）；
• 文档字节为空时直接返回空结果；
• 缺文件名时用guess_filename(file_data, 'pdf')嗅探出带类型的文件名，因为 ADE 依赖文件名推断文档类型；
• 通过build_client创建一次性的LandingAIADE客户端（每次调用新建，保证并发安全，见 landing_ai_base.py），以(filename, bytes)元组直传 SDK、不落临时文件：

response = client.parse(document=(file_name, file_data), model=self._model)

响应映射_map_response（parse.py）把response.markdown作为文本，遍历response.chunks，只把type（不区分大小写）为table且 Markdown 非空的 chunk 放进表格列表；同时从metadata读取credit_usage（debug 日志）和failed_pages（若有则 warning 提示部分页解析失败）。

Extract：Markdown 累积与 ADEextract调用

extract/IInstance.py 通过writeText把textlane 上收到的所有 Markdown 片段累积进markdown_parts，对象关闭时用空行连接成一个整体再触发抽取：

markdown = '\n\n'.join(self.markdown_parts).strip() if markdown: extraction = self.IGlobal.extractor.extract(markdown)

Extractor.extract（extract.py）的行为：

• 构造阶段捕获到的 schema 错误在运行时抛出ValueError（延迟到运行时而非在beginGlobal崩溃整个管道）；
• 无 API Key 或 Markdown 为空时返回{}；
• 调用 ADE 时schema 必须传 JSON 字符串——直接传 dict 会被 multipart 展平导致 422：

response = client.extract(markdown=markdown, schema=json.dumps(self._schema), strict=self._strict)

• _map_response取response.extraction，若metadata.warnings存在则 warning 提示部分抽取，但依然返回响应中的 extraction。

Extract 的 documents 输出规则

当documentslane 有连接时，IInstance._to_documents 会把抽取结果包装成文档：对象结果转成一个 JSON 文档；若抽取值是列表，则每个元素各转成一个 JSON 文档，并且每个输入对象的 chunk ID 从 0 开始递增编号：

items = extraction if isinstance(extraction, list) else [extraction] for item in items: ... documents.append(Doc(page_content=json.dumps(item), metadata=metadata)) self.chunkId += 1

失败与边界行为速查

两条失败语义值得注意：缺 key 与空输入是"fail-closed 返回空"，而 ADE 远端失败是"log-and-re-raise"（见 parse.py 与 extract.py），避免把远端故障静默伪装成"没有内容"。

测试验证与源码佐证

节点包附带了两组独立单测，用项目 mock 遮蔽landingai_adeSDK，无需真实服务即可运行：

• nodes/test/landing_ai/parse/test_parse.py：验证parse()把 ADE 响应映射为(markdown, [table blocks])；SDK 收到的是(filename, bytes)元组与配置的模型；缺文件名时嗅探出.pdf后缀；无 Key 时返回空结果；SDK 异常被重新抛出；_map_response仅按不区分大小写的table类型过滤表格 chunk。
• nodes/test/landing_ai/extract/test_extract.py：验证 base64>赞分享