原文：towardsdatascience.com/building-an-llmops-pipeline-08d367b36d64

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/69f8fb9af675710a6fbeae2dfa926cae.png

Image from Unsplash by Sigmund

2023 年是见证生成式人工智能（Generative AI）领域各种大型语言模型（LLMs）崛起的一年。LLMs 拥有惊人的力量和潜力，但将它们投入生产一直是对用户的一个持续挑战。一个特别普遍的问题是应该使用哪个 LLM？更具体地说，如何评估一个 LLM 的准确性？当有大量模型可供选择、不同的数据集用于微调/RAG 以及各种提示工程/调整技术需要考虑时，这尤其具有挑战性。

为了解决这个问题，我们需要为 LLMs 建立DevOps最佳实践。拥有一个可以帮助评估不同模型、数据集和提示的工作流程或管道。这个领域开始被称为LLMOPs/FMOPs。在 LLMOPs 中可以考虑的一些参数如下，这是一个（极其）简化的流程：

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/f73468a3255aa268e22c511b9ecf8155.png

LLM 评估考虑因素（作者）

在这篇文章中，我们将尝试通过构建一个微调、部署和评估Llama 7B 模型的管道来解决此问题。您也可以通过将其用作模板来比较多个 LLMs、数据集和提示来扩展此示例。对于这个例子，我们将使用以下工具来构建这个管道：

  • SageMaker JumpStart: SageMaker JumpStart 提供了各种 FM/LLMs，用于即时的微调和部署。这两个过程可能相当复杂，因此 JumpStart 抽象出了具体细节，并允许您指定数据集和模型元数据以进行微调和部署。在这种情况下，我们选择了 Llama 7B 并进行了指令微调，该功能是即插即用的。要深入了解 JumpStart 微调，请参阅这篇博客和这个 Llama 代码示例，我们将以此作为参考。

  • SageMaker Clarify/FMEval: SageMaker Clarify 通过 SageMaker Studio UI 和开源的 Python FMEVal 库 提供了一个基础模型评估工具。该功能内置了多种不同的算法，覆盖了不同的自然语言处理领域，如文本生成和摘要。在这个例子中，我们使用这个库来评估 Llama 模型在摘要用例中的应用。要深入了解该库，请参阅我的入门文章这里。

  • SageMaker Pipelines Step Decorator: SageMaker Pipelines 是 SageMaker 中的一个 MLOps 功能，它帮助您实现 ML 工作流的运营化。使用 Pipelines，您可以定义不同的步骤和参数来构建您的 ML 工作流。在 Pipelines 中特别有一个名为步骤装饰器的功能，您可以将 Python 代码提升并转换为可以作为 Pipeline 链接在一起的功能。在这个例子中，我们将使用这个功能来构建使用上述工具进行训练和评估的函数。要了解步骤装饰器的介绍，请参阅我的入门文章这里。

现在我们已经了解了我们的堆栈，让我们动手实践吧！

注意：本文假设读者对 Python、LLMs 和 Amazon SageMaker 有基本了解。

免责声明：我是 AWS 的机器学习架构师，我的观点仅代表个人。

目录

  1. 设置与数据集准备

  2. 管道步骤构建

  3. 管道执行

  4. 其他资源与结论

1. 设置与数据集准备

对于开发，我们将在新的SageMaker Studio环境中工作（已启用本地内核支持）。我们将使用具有 Python3 内核的 ml.c5.18xlarge 实例。

对于我们的用例，我们将使用 Llama 进行微调和评估摘要用例。对于我们的数据集，我们将使用公共Dolly 数据集（许可证：cc-by-sa-3.0）。我们可以使用内置的 HuggingFace 数据集库来获取此数据集，并过滤摘要数据点。我们还创建了用于微调和评估的训练和测试数据集。

import datasets

# dolly dataset
dolly_dataset = load_dataset("databricks/databricks-dolly-15k", split="train")

# summarization use-case
summarization_dataset = dolly_dataset.filter(lambda example: example["category"] == "summarization")
summarization_dataset = summarization_dataset.remove_columns("category")

# train test split
train_and_test_dataset = summarization_dataset.train_test_split(test_size=0.1)

# local train dataset
train_and_test_dataset["train"].to_json("train.jsonl")

# test dataset
train_and_test_dataset["test"].to_json("test.jsonl")

我们还指定了模型元数据以供 JumpStart 检索适当的 Llama 7B 模型。

import sagemaker

model_id, model_version = "meta-textgeneration-llama-2-7b", "2.*"

对于训练，我们将实现指令微调，因此我们准备了一个提示和响应模板，或者在这种情况下，文本和摘要。我们将使用以下示例作为训练部分的参考。

import json

template = {
    "prompt": "Below is an instruction that describes a task, paired with an input that provides further context. "
    "Write a response that appropriately completes the request.nn"
    "### Instruction:n{instruction}nn### Input:n{context}nn",
    "completion": " {response}",
}
with open("template.json", "w") as f:
    json.dump(template, f)

然后，我们将这些文件上传到公共 S3 路径以进行训练，以及后续的推理/评估。

from sagemaker.s3 import S3Uploader
import sagemaker
import random

output_bucket = sagemaker.Session().default_bucket()
local_data_file = "train.jsonl"
test_data_file = "test.jsonl"
train_data_location = f"s3://{output_bucket}/dolly_dataset"
test_data_location = f"s3://{output_bucket}/test_dataset"
S3Uploader.upload(local_data_file, train_data_location)
S3Uploader.upload("template.json", train_data_location)
S3Uploader.upload(test_data_file, test_data_location)
print(f"Training data: {train_data_location}")
print(f"Test data: {test_data_location}")
print(f"Output bucket: {output_bucket}")

2. 流水线步骤构建

流水线设置

为了设置我们的流水线，我们需要几个单独的文件来定义我们的执行环境。其中一个是一个 config.yaml 文件，它将定义流水线执行的硬件以及您定义的任何其他配置。该配置文件还指向您的 requirements.txt 文件，该文件将在流水线环境中安装。

SchemaVersion: '1.0'
SageMaker:
  PythonSDK:
    Modules:
      RemoteFunction:
        InstanceType: ml.m5.xlarge
        Dependencies: ./requirements.txt
        IncludeLocalWorkDir: true
        CustomFileFilter:
          IgnoreNamePatterns: # files or directories to ignore
          - "*.ipynb" # all notebook files

jsonlines
sagemaker
fmeval

然后，我们可以将此配置文件作为环境变量指向：

import os

# Set path to config file
os.environ["SAGEMAKER_USER_CONFIG_OVERRIDE"] = os.getcwd()

使用流水线，您还可以定义将被注入到您的流水线中的参数。在这种情况下，我们定义了执行每个步骤的工作硬件实例类型：

import sagemaker
from sagemaker.workflow.function_step import step
from sagemaker.workflow.parameters import ParameterString

sagemaker_session = sagemaker.session.Session()
role = sagemaker.get_execution_role()
bucket = sagemaker_session.default_bucket()
region = sagemaker_session.boto_region_name

instance_type = ParameterString(name="TrainInstanceType", 
default_value="ml.c5.18xlarge")

训练与部署步骤

对于训练步骤，我们将使用 SageMaker JumpStart 来微调我们的 Llama 7B 模型。我们将向函数传递几个不同的参数：

  • 训练数据路径：这是指向 train.jsonl 和 template.json 文件的 S3 位置。

  • 模型元数据：JumpStart 根据您传递的模型 ID 和版本识别要下拉的模型。在这种情况下，我们指定了 Llama 7B 模型。

一旦我们定义了这些，我们就使用这些参数设置 JumpStart 估算器。可选地，您还可以根据您想要测试的性能旋钮定义模型特定的微调参数。注意带有函数的步骤装饰器，表示我们正在处理 SageMaker 流水线步骤而不是普通的 Python 函数。

# step one
@step(
    name = "train-deploy",
    instance_type = instance_type,
    keep_alive_period_in_seconds=300
)
def train_deploy(train_data_path: str, 
model_id: str = "meta-textgeneration-llama-2-7b", 
model_version: str = "2.*") -> str:
    import sagemaker
    from sagemaker.jumpstart.estimator import JumpStartEstimator

    # configure JumpStart Estimator
    estimator = JumpStartEstimator(
        model_id=model_id,
        model_version=model_version,
        environment={"accept_eula": "true"},
        disable_output_compression=True, 
    )
    estimator.set_hyperparameters(instruction_tuned="True", epoch="1", max_input_length="1024")
    estimator.fit({"training": train_data_path})

    ## deploy fine-tuned model
    finetuned_predictor = estimator.deploy()
    endpoint_name = finetuned_predictor.endpoint_name
    return endpoint_name

训练完成后，我们将微调后的 Llama 模型部署到SageMaker 实时端点，我们可以调用该端点进行推理。默认情况下，此实例类型用于训练和推理，具体取决于您选择的 LLM，但您可以进行调整，如果您想自己选择硬件的话。

我们将端点名称作为下一步的输入，以便我们可以使用该端点进行推理和评估。当您最终执行 Pipeline 时，在 Studio UI 中，您将看到成功的 JumpStart 训练作业和端点创建。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/393c4e9806f43c19df29dede63ba3460.png

JumpStart 训练作业（作者截图）

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/c8b874263dc36c899298c3f2772c9b14.png

端点创建截图（作者截图）

现在我们有了端点，我们可以继续运行样本推理并对这些结果进行评估。

推理与评估步骤

在我们的第二步中，我们再次为我们的函数定义某些参数：

  • 端点名称：我们在评估之前对此端点进行推理。

  • S3_Test_Path：我们之前也将一个与训练数据集分开的测试数据集推送到 S3。我们将从这个路径下载数据集，并在评估之前进行推理。

我们首先使用Boto3 Python SDK下载 S3 的"test.jsonl"文件：

def evaluate(endpoint_name: str, output_bucket: str = output_bucket, test_data_file: str = "test.jsonl",
            key_path: str = "test_dataset/test.jsonl") -> str:

    # download S3 file
    s3 = boto3.client("s3")
    s3.download_file(output_bucket, key_path, test_data_file)

然后，我们使用此文件进行推理，并创建一个新的 JSONLines 文件，我们可以在此文件上运行我们的评估算法。出于时间考虑，我们限制此样本仅为 20 个数据点，但您可以根据需要增加测试数据集。

with jsonlines.open(input_file) as input_fh, jsonlines.open(output_file, "w") as output_fh:
  for i, datapoint in enumerate(input_fh, start=1):
      instruction = datapoint["instruction"]
      context = datapoint["context"]
      summary = datapoint["response"]
      payload = prepare_payload(datapoint)
      response = runtime.invoke_endpoint(EndpointName=endpoint_name, Body=json.dumps(payload), 
                             ContentType=content_type, CustomAttributes='accept_eula=true')
      result = json.loads(response['Body'].read().decode())[0]['generation']
      line = {"instruction": instruction, "context": context, "summary": summary, "model_output": result}
      output_fh.write(line)

      # evaluate just 20 datapoints for example
      if i == 20:
          break

函数的这一部分运行完成后，应该生成一个"results.jsonl"文件，该文件将包含几个不同的参数：

  • 文档/输入：这是需要总结的原始文本。

  • 真实值/实际输出：这是测试数据集中存在的摘要，这是我们评估的真实值。

  • 模型输出：这是我们使用微调后的 Llama 7B 模型进行的推理。我们将运行我们的评估算法，比较真实值与模型推理结果。

现在我们有了用于评估的数据集，我们可以导入 FMEval 库：

import fmeval
from fmeval.data_loaders.data_config import DataConfig
from fmeval.constants import MIME_TYPE_JSONLINES
from fmeval.eval_algorithms.summarization_accuracy import SummarizationAccuracy

注意，我们拉取了 SummarizationAccuracy 算法，该算法将返回如 Meteor、Rouge 和 Bert 等指标。要查看这些算法的完整实现，您可以参考此链接。一般来说，每个指标都有其优缺点，您可以选择您想要用于评估的指标。

  • Rouge: Rouge-N 分数，本质上是在真实值和模型推理摘要之间搜索 N-gram 词重叠。

  • Meteor: 基于 Rouge，结合词干提取和同义词，这将捕获更多文本中的相似性，即使单词不完全匹配。

  • BertScore: 使用余弦相似度匹配单词，同时利用 BERT 的预训练嵌入，你会在安装包时看到这一点。

对于这些指标的深入探讨，我会参考以下文章。

对于我们使用 FMEval 包的实现，我们首先在 FMEval 特定的DataConfig 对象中配置我们的数据集，并指定真实值和模型推理列。

config = DataConfig(
        dataset_name="dolly_summary_model_outputs",
        dataset_uri="results.jsonl",
        dataset_mime_type=MIME_TYPE_JSONLINES,
        model_input_location="instruction",
        target_output_location="summary",
        model_output_location="model_output"
    )

然后，我们实例化 SummarizationAccuracy 算法，并使用指定的 DataConfig 对象运行评估。

eval_algo = SummarizationAccuracy()
eval_output = eval_algo.evaluate(dataset_config=config, save=True)
res = json.dumps(eval_output, default=vars, indent=4)
serialized_data = json.loads(res)
# print metrics to CW logs, realistically push to somewhere to visualize
for item in serialized_data:
    for key, value in item.items():
        print(f"Key: {key}, Value: {value}")

在这个例子中，我们将指标直接写入 CloudWatch 日志，在现实场景中，你可以将其导出到 S3 或 QuickSight 等可视化工具，以获得更美观的评估视图。如果你检查了管道执行后的第二步的 CloudWatch 日志，你会注意到指标已经被输出。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/6e8ddc5f289b5c7ae1698bfef73c1d5a.png

评估指标（作者截图）

3. 管道执行

一旦定义了两个管道步骤，你就可以像使用纯 Python 函数一样简单地将它们链接在一起。

# stitch together pipeline
from sagemaker.workflow.pipeline import Pipeline

endpoint_name = train_deploy(train_data_location)
eval_metrics = evaluate(endpoint_name)

然后，我们可以定义一个管道对象并启动执行：

pipeline = Pipeline(
    name="llm-train-eval-pipeline",
    parameters=[
        instance_type
    ],
    steps=[
        eval_metrics,
    ],
)

# execute Pipeline
pipeline.upsert(role_arn=role)
execution = pipeline.start()
execution.describe()
execution.wait()

你可以在 Studio UI 中查看和监控管道执行，这个管道大约需要 45 分钟才能成功完成。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/cc6f49ebeb9c14b53cb8e30e87114536.png

执行状态（作者截图）

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/1eb5883158b3f9ce13e008ba128ea4d5.png

DAG（作者截图）

4. 其他资源与结论

GenAI-Samples/LLMOps-Pipeline at master · RamVegiraju/GenAI-Samples

我希望这篇文章是 LLMOPs 和利用不同 SageMaker 功能构建管道的有用介绍。随着 LLM 用例的扩展，对正确实验的需求也在增加，以确定你 LLM 的理想配置。这个例子可以扩展以包含多个 LLM、数据集、提示模板等。请关注 GenAI/LLM 空间中的更多内容。

如常，感谢您阅读，欢迎随时留下任何反馈。

如果您喜欢这篇文章，请随意在LinkedIn上与我联系并订阅我的 Medium通讯。

Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

AI5 - 智慧城市交通调度：用大模型缓解早高峰拥堵

avatar 2048 AI社区

Spring AI 集成 Amazon Bedrock Converse API 技术博客

随着生成式 AI 应用的爆发，企业亟需安全、可扩展、多模态、可调用工具的 LLM 服务。Amazon Bedrock 提供多家主流模型统一接入，Converse API 更支持工具调用和多模态输入。Spring AI 作为开源 Java AI 框架，已原生支持 Bedrock Converse，帮助开发者快速构建智能应用。Bedrock Converse API 让 Java 应用具备主流 LLM

avatar 2048 AI社区

AI应用架构核心：AI辅助决策支持系统配置中心架构设计（动态参数调整）

一个参数ppkvτdsmpkvτdsmk：键，参数的唯一标识符，是一个字符串。例如，。v：值，参数的具体取值，其类型由τ决定。例如，0.85。τ：数据类型，取自一个类型集合。d：描述，人类可读的文本，说明参数的用途和影响。s：作用域，定义了参数的生效范围。这是一个复杂但关键的概念，我们将其进一步分解。假设我们正在构建一个名为“ShieldWave”的实时金融交易风控系统。

avatar 2048 AI社区