使用 MLflow 跟踪大型语言模型（LLM）：完整指南

pip install mlflow>=2.8.1
pip install openai
pip install chromadb==0.4.15
pip install langchain==0.0.348
pip install tiktoken
pip install 'mlflow[genai]'
pip install databricks-sdk --upgrade

import mlflow
import chromadb

<p>print(f"MLflow 版本：{mlflow.__version__}")
print(f"ChromaDB 版本：{chromadb.__version__}")

import mlflow
import openai

<p>def query_llm(prompt, max_tokens=100):
response = openai.Completion.create(
engine="text-davinci-002",
prompt=prompt,
max_tokens=max_tokens
)
return response.choices[0].text.strip()</p>

<p>with mlflow.start_run():
prompt = "用简单的术语解释机器学习的概念。"</p>

<p># 记录参数
mlflow.log_param("model", "text-davinci-002")
mlflow.log_param("max_tokens", 100)</p>

<p># 查询 LLM 并记录结果
result = query_llm(prompt)
mlflow.log_metric("response_length", len(result))</p>

<p># 记录提示和响应
mlflow.log_table("prompt_responses", {"prompt": [prompt], "response": [result]})</p>

<p>print(f"响应：{result}")

import mlflow
from mlflow.deployments import get_deploy_client

<p># 初始化部署客户端
client = get_deploy_client("databricks")</p>

<p># 定义端点配置
endpoint_name = "llm-endpoint"
endpoint_config = {
"served_entities": [{
"name": "gpt-model",
"external_model": {
"name": "gpt-3.5-turbo",
"provider": "openai",
"task": "llm/v1/completions",
"openai_config": {
"openai_api_type": "azure",
"openai_api_key": "{{secrets/scope/openai_api_key}}",
"openai_api_base": "{{secrets/scope/openai_api_base}}",
"openai_deployment_name": "gpt-35-turbo",
"openai_api_version": "2023-05-15",
},
},
}],
}</p>

<p># 创建端点
client.create_endpoint(name=endpoint_name, config=endpoint_config)

<div class="relative flex flex-col rounded-lg">

<p>response = client.predict(
endpoint=endpoint_name,
inputs={"prompt": "简要解释神经网络的概念。", "max_tokens": 100,},
)</p>

print(response)

这将向我们的部署模型发送一个提示并返回生成的响应。

使用 MLflow 评估 LLM

评估对于理解 LLM 的性能和行为至关重要。MLflow 提供了全面工具来评估 LLM，包括内置和自定义指标。

准备您的 LLM 进行评估

要使用 mlflow.evaluate() 评估您的 LLM，您的模型需要以以下形式之一存在：

1. 一个 mlflow.pyfunc.PyFuncModel 实例或指向已记录的 MLflow 模型的 URI。
2. 一个接受字符串输入并输出单个字符串的 Python 函数。
3. 一个 MLflow Deployments 端点 URI。
4. 设置 model=None 并在评估数据中包含模型输出。

让我们看一个使用已记录的 MLflow 模型的示例：

import mlflow
import openai

<p>with mlflow.start_run():
system_prompt = "简要回答以下问题。"
logged_model_info = mlflow.openai.log_model(
model="gpt-3.5-turbo",
task=openai.chat.completions,
artifact_path="model",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": "{question}"},
],
)</p>

<p># 准备评估数据
eval_data = pd.DataFrame({
"question": ["什么是机器学习？", "解释神经网络。"],
"ground_truth": [
"机器学习是人工智能的一个子集，能够让系统从经验中学习和改进，而无需显式编程。",
"神经网络是受生物神经网络启发的计算系统，由处理和传输信息的相互连接的节点组成。",
]
})</p>

<p># 评估模型
results = mlflow.evaluate(
logged_model_info.model_uri,
eval_data,
targets="ground_truth",
model_type="question-answering",
)</p>

<p>print(f"评估指标：{results.metrics}")

此示例记录一个 OpenAI 模型，准备评估数据，然后使用 MLflow 的内置指标评估模型，用于问答任务。

自定义评估指标

MLflow 允许您定义自定义指标来评估 LLM。以下是创建一个自定义指标的示例，用于评估响应的专业性：

from mlflow.metrics.genai import EvaluationExample, make_genai_metric

<p>professionalism = make_genai_metric(
name="professionalism",
definition="正式和适当的沟通风格的衡量标准。",
grading_prompt=(
"根据回答的正式和专业程度对其进行评分（0-4 分）。\n"
"0：非常随意或不恰当\n"
"1：随意但尊重\n"
"2：中度正式\n"
"3：专业和适当\n"
"4：非常正式和精心制作"
),
examples=[
EvaluationExample(
input="什么是 MLflow？",
output="MLflow 是一个用于管理机器学习项目的工具包。它非常酷！",
score=1,
justification="回答很随意，使用非正式语言。",
),
EvaluationExample(
input="什么是 MLflow？",
output="MLflow 是一个用于机器学习生命周期的开源平台，包括实验、可复现性和部署。",
score=4,
justification="回答正式、简洁、专业措辞。",
)
],
model="openai:/gpt-3.5-turbo-16k",
parameters={"temperature": 0.0},
aggregations=["mean", "variance"],
greater_is_better=True,
)</p>

<p># 在评估中使用自定义指标
results = mlflow.evaluate(
logged_model_info.model_uri,
eval_data,
targets="ground_truth",
model_type="question-answering",
extra_metrics=[professionalism]
)</p>

<p>print(f"专业性评分：{results.metrics['professionalism_mean']}")

此自定义指标使用 GPT-3.5-turbo 来评估响应的专业性，展示了如何利用 LLM 本身进行评估。

高级 LLM 评估技术

随着 LLM 的日益复杂，评估它们的技术也变得更加先进。让我们探索一些使用 MLflow 的高级评估方法。

检索增强生成（RAG）评估

RAG 系统结合了检索和生成模型的力量。评估 RAG 系统需要评估检索和生成组件。以下是如何设置 RAG 系统并使用 MLflow 进行评估：

from langchain.document_loaders import WebBaseLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

<p># 加载和预处理文档
loader = WebBaseLoader(["https://mlflow.org/docs/latest/index.html"])
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)</p>

<p># 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)</p>

<p># 创建 RAG 链
llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(),
return_source_documents=True
)</p>

<p># 评估函数
def evaluate_rag(question):
result = qa_chain({"query": question})
return result["result"], [doc.page_content for doc in result["source_documents"]]</p>

<p># 准备评估数据
eval_questions = [
"什么是 MLflow？",
"MLflow 如何处理实验跟踪？",
"MLflow 的主要组件是什么？"
]</p>

<p># 使用 MLflow 进行评估
with mlflow.start_run():
for question in eval_questions:
answer, sources = evaluate_rag(question)</p>

<p>mlflow.log_param(f"question", question)
mlflow.log_metric("num_sources", len(sources))
mlflow.log_text(answer, f"answer_{question}.txt")</p>

<p>for i, source in enumerate(sources):
mlflow.log_text(source, f"source_{question}_{i}.txt")</p>

<p># 记录自定义指标
mlflow.log_metric("avg_sources_per_question", sum(len(evaluate_rag(q)[1]) for q in eval_questions) / len(eval_questions))