让大模型思考，让小模型执行：在 Elastic Workflows 中拆分 LLM 成本

作者来自 Elastic Jeffrey Rengifo构建一个 Elastic workflow将数据样本发送给大模型由大模型提出分类标签。人工审核确认后再由小模型将这些标签应用于整个数据集。将大语言模型LLM分类中昂贵的部分与廉价的部分拆分开来。本文构建了一个 Elastic workflow让 Claude Sonnet 阅读 NASA 飞行员事故报告的分层样本并根据发现的内容提出分类标签。人工审核 schema 并确认后再由 Mistral Small 3.1 将这些标签应用于整个数据集。路由使用 YAML结果以结构化数据的形式写入 Elasticsearch而这种模式适用于任何需要对自由文本进行标注的场景。NASA 航空安全报告系统ASRS 报告描述了飞行过程中发生的异常事件例如高度偏差、令人困惑的许可指令、跑道问题或机械故障。每份报告已经有一个官方类别例如高度偏差、航向偏差或地面事件。在本文中我们提出了一个不同的问题这份报告揭示了撰写报告的飞行员的哪些信息我们希望让模型根据数据本身推断出一个 schema根据有助于了解报告撰写者的信息对报告进行分类。然后再让第二个模型应用这些标签。你可以在这里找到完整的 workflow 定义和辅助脚本。运行此 LLM 流水线所需的内容Elastic Stack 9.4 或 Elastic Cloud Serverless。Elastic Workflows 自 9.4 起正式发布GA。在你的部署中启用 Elastic Agent Builder。一个指向 Claude Sonnet或等效推理模型的 Kibana 生成式 AIGenAI连接器。它充当规划器。一个 Mistral API 密钥。我们将使用它来注册一个 Elasticsearch inference endpoint。Python 3.10以及elasticsearch9.0和pandas。数据集加载器会使用这些组件。双层 LLM 编排如何工作这个 workflow 有两个任务确定应该有哪些标签然后将这些标签应用到每份报告。第一个任务是开放式的。大模型读取一组具有多样性的报告样本并提出一个由分类字段组成的小型 schema。一个字段是一种描述报告撰写者的方式例如attribution_style或procedure_orientation。每个字段都有几个允许的值例如self_critical、system_attributing或balanced。第二个任务是可重复的。人工批准 schema 后小模型读取每份报告并为每个字段选择一个值。我们使用 Elastic Workflows是因为这些步骤都是预先确定的抽取报告样本、提出标签、等待批准、对每份文档进行分类以及存储结果。将这些步骤写入 YAML可以让整个流程具备可复现性、可观测性并且能够以更低的成本重新运行。为什么要在两个模型层级之间拆分 LLM 工作小模型可以处理分类但 schema 发现是另一种类型的问题。它需要读取多样化的样本、发现潜在模式并提出复杂的结构。在实际使用中小模型往往会过度依赖表面关键词并生成重复或互斥性不足的字段。分类更加简单因为 schema 已经存在值也已经枚举出来了任务只是为每个字段选择一个值。小模型可以可靠地处理这项任务而且成本只是大模型的一小部分因为它需要对整个数据集中的每份文档运行一次。这里的大和小指的是推理能力。在本文中Claude Sonnet 充当规划器而 Mistral Small 3.1 充当执行器。使用双层 LLM 流水线对 NASA 飞行员报告进行分类我们将使用 NASA ASRS 数据库该数据库收集来自飞行员、空中交通管制员和机械师的自愿、匿名事故报告。该数据集是公开的报告以自由文本叙述的形式撰写。我们想要提出的问题是这份报告揭示了撰写报告的飞行员的哪些信息规划器读取一组具有多样性的报告样本并根据报告实际的撰写方式来判断哪些差异具有意义。步骤角色模型层级sample从数据集中抽取一组具有多样性的报告子集。无 LLMdiscover阅读问题和样本提出由字段及枚举值组成的 schema。大模型approve人工审核提出的 schema并进行批准或编辑。通过 waitForInput 由人工完成apply遍历整个数据集为每份报告的每个字段分配一个值。小模型store将 schema 和每份文档的字段值写入 Elasticsearch。注册 Mistral 和 Claude 为 Elasticsearch inference endpoint小模型将使用原生的mistral服务集成注册为 Elasticsearch inference endpoint。INFERENCE_ID mistral-small-extractor es.inference.put( task_typechat_completion, inference_idINFERENCE_ID, inference_config{ service: mistral, service_settings: { api_key: MISTRAL_API_KEY, model: mistral-small-latest, # 6 RPM is conservative for the Mistral free tier to avoid 429s. rate_limit: {requests_per_minute: 6}, }, }, )别名mistral-small-latest解析为 Mistral Small 3.1。它具有 128k 的上下文窗口并支持 JSON 模式输出。大模型将使用一个指向 Claude Sonnet 的 AI connector。Agent Builder UI 会引导你完成 connector 的创建。记下 connector ID因为我们将在 workflow 中引用它。将 NASA ASRS 事故报告索引到 ElasticsearchASRS 数据集使用keyword映射来处理用于聚合的字段并使用text映射来处理模型将要读取的叙述文本。下载 ASRS CSV 文件该数据库会在 ASRS Database Online 页面发布季度数据提取文件然后将其建立索引。映射如下{ properties: { acn: { type: keyword }, flight_phase: { type: keyword }, anomaly: { type: keyword }, synopsis: { type: text }, narrative: { type: text } } }映射类型根据每个字段的用途来确定。flight_phase和anomaly被映射为keyword因为我们将对它们执行terms聚合来构建样本而聚合需要精确的、未经分析的值。narrative和synopsis被映射为text因为它们包含模型将要读取的自由文本。配套的 notebook 包含完整的加载脚本用于读取 CSV 并批量建立文档索引。为规划 LLM 构建分层样本本节以及后续章节中的 YAML 代码片段都是 workflow 定义中的步骤notebook 会通过 Workflows API 注册这些步骤。前两个步骤用于生成具有代表性的样本它们分别按照飞行阶段和异常情况进行聚合并使用top_hits从每个分组中抽取几份文档。- name: by_phase type: elasticsearch.request with: method: POST path: /incident_reports/_search body: size: 0 aggs: per_phase: terms: field: flight_phase size: 8 aggs: sampled_docs: top_hits: size: 5 _source: [acn, synopsis, narrative] - name: by_anomaly type: elasticsearch.request with: method: POST path: /incident_reports/_search body: size: 0 aggs: per_anomaly: terms: field: anomaly size: 8 aggs: sampled_docs: top_hits: size: 3 _source: [acn, synopsis, narrative]大型 LLM 如何从数据中发现分类 schema提示词需要同时包含问题和样本。只有问题可能会产生与数据集无关的通用标签而只有样本则会产生描述性分组却忽略问题的角度。当两者同时存在并且输出采用结构化形式时模型会生成基于数据且面向任务的标签由分类字段组成的 schema每个字段包含 2 到 4 个互斥的值选项并由样本中的证据提供支持。下面是 workflow 中的规划器步骤- name: discover type: ai.prompt connector-id: claude-sonnet with: systemPrompt: | You design categorical schemas for use by downstream classifiers. A schema is a small set of fields, each with a few mutually exclusive values. Every value you propose must be grounded in evidence from the provided sample and must serve the stated question. You do not invent values that are not supported by at least two documents in the sample. You do not propose fields that a reasonable analyst could have written without reading the documents. prompt: | Question: ${{ inputs.goal }} Sample documents stratified by flight phase: ${{ steps.by_phase.output.aggregations.per_phase.buckets | json }} Sample documents stratified by anomaly type: ${{ steps.by_anomaly.output.aggregations.per_anomaly.buckets | json }} Propose between 2 and 4 categorical fields that: - serve the question (you can explain how) - depend on patterns visible in the sample (you can cite document IDs) - would not be obvious to someone who has not read the sample For each field, return: name (snake_case), definition, why_useful, and values (2 to 4 mutually exclusive options). For each value, return: value (snake_case) and definition. schema: type: object properties: fields: type: array minItems: 2 maxItems: 4 items: type: object required: [name, definition, why_useful, values] properties: name: { type: string } definition: { type: string } why_useful: { type: string } values: type: array minItems: 2 maxItems: 4 items: type: object required: [value, definition] properties: value: { type: string } definition: { type: string } temperature: 0.3结构化输出 schema 强制规定了响应的结构name分类字段的标识符。definition该字段衡量的内容用一句话描述。why_useful该字段如何服务于问题这也有助于下游分类器理解其意图。values2 到 4 个互斥的选项。每个选项包含一个value和一个definition。下面是一个生成的 schema 示例。我们可以看到报告撰写者是如何被分类的以及模型决定创建该类别的原因。第二个模型会使用definition和why_useful字段来对文档进行分类。{ fields: [ { name: attribution_style, definition: How the reporter frames responsibility for what happened., why_useful: Surfaces reporting culture independent of the technical event. Useful for training and safety-management programmes that want to distinguish reporter style from incident type., values: [ { value: self_critical, definition: Assigns the cause primarily to their own action, even when external factors clearly contributed. }, { value: system_attributing, definition: Frames the cause as external: ATC, equipment, weather, or organisational factors. }, { value: balanced, definition: Distributes responsibility across self and system without emphasising either. } ] }, { name: procedure_orientation, definition: How the reporter relates to written procedure., why_useful: Distinguishes pilots who frame events through SOPs from those who frame them through personal judgment., values: [ // procedure_first, experience_first (same structure as above) ] } ] }使用 waitForInput 实现人工参与的 schema 审批现在提出的 schema 会交给人工进行审批。Elastic Workflows 提供了一个 waitForInput 步骤可以使用 schema 暂停 workflow、展示表单并在提交输入后恢复执行。waitForInput自身没有超时设置因此如果没有人响应执行会无限期等待。如果希望设置时间限制可以设置 workflow 级别的settings.timeout如果审核者在超时时间到达之前没有提交表单执行就会被取消。- name: human_gate type: waitForInput with: message: Review and edit the proposed schema. The approved fields will be applied across the full corpus. schema: type: object required: [approved_fields] properties: approved_fields: type: array items: type: object properties: name: { type: string } definition: { type: string } values: type: array items: type: object properties: value: { type: string } notes: type: string当 workflow 执行到这一步时执行会暂停Kibana UI 会显示一个“需要执行操作”的标记。点击提供操作会打开一个表单审核者可以在其中粘贴或编辑 schema JSON。由于waitForInput无法使用前一个步骤的输出进行预填充因此代码会轮询discover步骤的输出并打印一个可以直接复制到该表单中的 JSON 代码块。discover step_output(execution_id, discover) # polls until the step completes # Strip why_useful (not part of the human_gate form) and wrap in the shape # expected by the waitForInput form so this is paste-ready. approved_fields [ { name: field[name], definition: field[definition], values: [ {value: v[value], definition: v[definition]} for v in field[values] ], } for field in discover[content][fields] ] print(json.dumps({approved_fields: approved_fields, notes: }, indent2))notebook 中的step_output辅助函数会通过GET /api/workflows/executions/{id}轮询执行状态直到discover步骤完成然后返回其输出。粘贴到 Kibana 中的代码 JSON 输出审核者可以保留有用的字段、重写不清晰的字段、合并重叠的值并添加备注。批准后workflow 会恢复执行并将最终 schema 发送给执行器步骤。对于新的数据集请保留这个人工审核环节。等 schema 稳定后你可以自动批准并且只在确实需要时再进行人工审核只将低置信度的提取结果交给人工或者将新的发现运行结果与存储在schemas索引中的 schema 进行比较仅当字段或值的变化超过阈值时才触发审核。使用较小的 LLM 对整个数据集进行分类当 workflow 执行到这一步时任务中开放式的部分已经完成。从这里开始小模型接管工作并根据已批准的 schema 对每份报告进行分类。- name: fetch_corpus type: elasticsearch.request with: method: POST path: /incident_reports/_search body: size: 100 _source: [acn, narrative] query: match_all: {} - name: classify_all type: foreach foreach: ${{ steps.fetch_corpus.output.hits.hits }} iteration-on-failure: retry: max-attempts: 5 delay: 3s fallback: - name: notify_failure type: slack_api.postMessage connector-id: team-alerts with: channelNames: - #pipeline-alerts text: Classification failed for ACN ${{ foreach.item._source.acn }} after all retries. continue: true steps: - name: classify type: ai.agent inference-id: mistral-small-extractor timeout: 120s with: message: | You will classify the following report against a fixed schema. For each field in the schema, assign exactly one of its value options, or null if none of the values clearly applies. Include the short quote that supports the assignment and a confidence score between 0 and 1. Set review_required to true if any field returned null or any confidence is below 0.5. Schema: ${{ steps.human_gate.output.approved_fields | json }} Report: ${{ foreach.item._source.narrative }} schema: type: object properties: field_values: type: object additionalProperties: true review_required: { type: boolean } - name: write_extraction type: elasticsearch.index with: index: extractions document: acn: ${{ foreach.item._source.acn }} field_values: ${{ steps.classify.output.structured_output.field_values }} review_required: ${{ steps.classify.output.structured_output.review_required }}注意分类步骤使用ai.agent而不是ai.prompt因为ai.agent 接受一个 inference-id这使它能够直接调用 Elasticsearch 的_inferenceendpoint而ai.prompt只接受connector-id。fetch_corpus步骤是 workflow 中第三个elasticsearch.request因此有必要说明为什么我们要再次从索引中读取数据。前两个步骤by_phase和by_anomaly只抽取了一小部分分层样本供规划器进行推理而不是用于实际标注的数据。现在 schema 已经获得批准fetch_corpus会抽取我们真正希望进行分类的文档。为了让演示运行得更快我们使用match_all将数量限制为 100在实际场景中你可以在这里对整个数据集进行分页处理。对于每个字段它都会返回一个值或 null、一个置信度以及一段简短的引用。将 JSON schema 中的additionalProperties设置为true可以让该步骤为每个字段返回一个条目而不需要 workflow 预先知道字段名称。存储的提取结果如下所示{ acn: 2238341, field_values: { attribution_style: { value: self_critical, confidence: 0.82, quote: I should have caught the altitude bust earlier }, procedure_orientation: { value: procedure_first, confidence: 0.44, quote: we ran the QRH before doing anything else } }, review_required: true }这里review_required为true因为procedure_orientation返回的置信度为0.44低于我们设定的0.5阈值这正是基于置信度的质量门控会采取行动的信号。fetch_corpus步骤抽取需要分类的文档。foreach 步骤依次遍历这些文档而iteration-on-failure负责处理错误retry用于处理 inference endpoint 返回的临时 API 错误如果所有尝试都失败fallback步骤会向 Slack 发送消息从而避免失败被悄无声息地忽略。email connector 也可以采用相同的方式。continue: true随后允许循环继续处理下一份文档而不是让整个运行失败。对于生产规模的数据集可以考虑使用executeAsync它是execute的扇出版本。将 schema 和提取结果写回 Elasticsearchworkflow 会生成两类内容经过批准的 schema以及每份文档对应的字段值。store_schema步骤紧接人工审核环节之后运行在分类步骤进行扇出之前执行- name: store_schema type: elasticsearch.index with: index: schemas document: question: ${{ inputs.goal }} approved_fields: ${{ steps.human_gate.output.approved_fields }} reviewer_notes: ${{ steps.human_gate.output.notes }}每次提取都会在foreach循环中写入因此结果会在生成后立即持久化而不是等到最后再进行批量写入。schemas索引每次发现运行对应一个文档其中包含问题、经过批准的字段以及审核者备注。extractions索引则每个报告的每个 schema 版本对应一个文档。这种双层 LLM 编排模式能为你带来什么我们构建了一个 Elastic workflow它从事故报告索引中抽取分层样本将样本和一个问题一起发送给大型推理模型由模型根据数据和用户定义的角度生成分类 schema暂停等待人工审核然后使用小型 Mistral 模型遍历整个数据集为每个字段分配一个值。经过批准的 schema 和每份文档的字段值会以结构化数据的形式写回 Elasticsearch。这个实践的意义在于两种不同类型的工作——schema 发现和 schema 应用——可以使用两个不同层级的模型而 workflow 则让你能够明确记录这种路由决策。下一步构建你自己的 LLM 流水线在你自己的数据集上试试。无论输入是事故报告、客户反馈、每周状态更新还是房产信息这种模式都不受影响。当你准备好在大规模场景下进行并行扇出后将foreach步骤提升为workflow.executeAsync。通过 cron 触发器定期运行重新发现 workflow这样你就可以根据不断进入的数据发现不同的 schema 变体。阅读 Elastic Workflows 文档了解完整的步骤目录。相关阅读使用 Elastic Inference Service 构建 AI agentsEIS介绍了通过 EIS 实现更广泛的多模型连接模式与本文所展示的由 Workflows 编排的拆分模式相互补充。如何使用 Elasticsearch 构建 AI agentic workflows 从更高层面介绍了 Agent Builder 和 Workflows 如何协同工作。LangExtract 和 Elasticsearch 教程 探索了另一种使用手工编写 schema 的提取模式可以与上文介绍的先发现、再应用的方法进行对比。原文Elastic Workflows for human-in-the-loop LLM orchestration | Elasticsearch Labs