版本跟踪快速入门
利用 MLflow 的版本管理功能，构建并跟踪一个基于 LangChain 的聊天机器人。本快速入门演示了如何使用 MLflow 的 GenAI 功能进行提示词版本控制、应用程序跟踪、追踪生成和性能评估。
先决条件
安装所需软件包
需要 MLflow 3
本快速入门需要 MLflow 3.0 或更高版本才能使用完整的 GenAI 功能。
pip install --upgrade mlflow
pip install langchain-openai
设置 OpenAI API 密钥
配置您的 OpenAI API 密钥以通过 OpenAI 服务进行身份验证
export OPENAI_API_KEY=your_api_key_here
您将学到什么
本快速入门涵盖了使用 MLflow 的版本管理系统构建可跟踪的 GenAI 应用程序的基本概念。
版本控制提示词
注册并跟踪具有完整版本历史的提示词模板，以实现可复现的实验
构建 LangChain 代理
创建具有自动 MLflow 集成和可观察性的对话代理
追踪一切
启用全面的追踪日志记录，以监控和调试模型的行为
评估性能
使用内置指标和自定义评估框架评估模型质量
让我们构建一个简单的 IT 支持聊天机器人，并使用 MLflow 跟踪其开发生命周期。
第 1 步：注册提示词模板
首先创建一个带版本的提示词模板。这使您能够跟踪提示词的演变，并确保实验的可复现性。
import mlflow
system_prompt = mlflow.genai.register_prompt(
name="chatbot_prompt",
template="You are a chatbot that can answer questions about IT. Answer this question: {{question}}",
commit_message="Initial version of chatbot",
)
在 MLflow UI 中查看您的提示词
导航到**提示词 (Prompts)** 选项卡以查看您注册的提示词
第 2 步：构建 LangChain 对话链
创建一个简单的链，将您的提示词模板与 OpenAI 的聊天模型结合起来
from langchain.schema.output_parser import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
# Convert MLflow prompt to LangChain format
prompt = ChatPromptTemplate.from_template(system_prompt.to_single_brace_format())
# Build the chain: prompt → LLM → output parser
chain = prompt | ChatOpenAI(temperature=0.7) | StrOutputParser()
# Test the chain
question = "What is MLflow?"
print(chain.invoke({"question": question}))
# MLflow is an open-source platform for managing the end-to-end machine learning lifecycle...
第 3 步：启用追踪可观察性
设置自动追踪日志记录，以在开发过程中监控模型的行为。这将创建所有模型交互的链接历史记录。
配置活动模型和自动日志记录
# Set the active model for linking traces
mlflow.set_active_model(name="langchain_model")
# Enable autologging - all traces will be automatically linked to the active model
mlflow.langchain.autolog()
生成测试追踪
运行多个查询以生成用于分析的追踪
questions = [
{"question": "What is MLflow Tracking and how does it work?"},
{"question": "What is Unity Catalog?"},
{"question": "What are user-defined functions (UDFs)?"},
]
outputs = []
for question in questions:
outputs.append(chain.invoke(question))
# Verify traces are linked to the active model
active_model_id = mlflow.get_active_model_id()
mlflow.search_traces(model_id=active_model_id)
在 UI 中探索追踪
- **查看已记录的模型**：在您的实验中检查**模型 (Models)** 选项卡
- **访问模型详情**：点击您的模型以查看其唯一的 `model_id`
- **分析生成的追踪**：导航到**追踪 (Traces)** 选项卡以检查单个交互
第 4 步：评估模型性能
评估对于了解聊天机器人的性能以及确保随时间推移的质量改进至关重要。MLflow 为系统性评估跟踪提供了基础。
关键评估功能
- **质量评估**：使用关键字覆盖率和内容分析系统地评估响应质量，以确保您的聊天机器人达到预期
- **性能指标**：跟踪响应长度和关键字匹配等定量指标，以监控随时间推移的改进
- **持续监控**：将评估结果记录到 MLflow，以便进行持续的性能跟踪和模型迭代间的比较
使用 MLflow 的跟踪功能，根据预期响应评估聊天机器人的准确性和相关性。
手动分析输出
通过将模型输出与预期响应进行比较来评估它们
import pandas as pd
# Create evaluation dataset
eval_data = [
{
"question": "What is MLflow Tracking and how does it work?",
"expected_keywords": [
"experiment tracking",
"parameters",
"metrics",
"artifacts",
"UI",
],
},
{
"question": "What is Unity Catalog?",
"expected_keywords": [
"data assets",
"centralized",
"collaboration",
"governance",
],
},
{
"question": "What are user-defined functions (UDFs)?",
"expected_keywords": [
"custom functions",
"data transformations",
"Spark",
"SQL",
],
},
]
# Simple evaluation metrics
def evaluate_response(response, expected_keywords):
"""Simple keyword-based evaluation."""
response_lower = response.lower()
keyword_matches = sum(
1 for keyword in expected_keywords if keyword.lower() in response_lower
)
coverage_score = keyword_matches / len(expected_keywords)
response_length = len(response.split())
return {
"keyword_coverage": coverage_score,
"response_length": response_length,
"keyword_matches": keyword_matches,
}
# Evaluate each response
evaluation_results = []
for i, (output, eval_item) in enumerate(zip(outputs, eval_data)):
metrics = evaluate_response(output, eval_item["expected_keywords"])
evaluation_results.append(
{
"question": eval_item["question"],
"response": output,
"keyword_coverage": metrics["keyword_coverage"],
"response_length": metrics["response_length"],
"keyword_matches": metrics["keyword_matches"],
}
)
print(
f"Question {i+1}: {metrics['keyword_matches']}/{len(eval_item['expected_keywords'])} keywords found"
)
print(f"Coverage: {metrics['keyword_coverage']:.1%}")
print(f"Response length: {metrics['response_length']} words\n")
# Log evaluation metrics
with mlflow.start_run():
avg_coverage = sum(r["keyword_coverage"] for r in evaluation_results) / len(
evaluation_results
)
avg_length = sum(r["response_length"] for r in evaluation_results) / len(
evaluation_results
)
mlflow.log_metric("avg_keyword_coverage", avg_coverage)
mlflow.log_metric("avg_response_length", avg_length)
print(f"📊 Average keyword coverage: {avg_coverage:.1%}")
print(f"📊 Average response length: {avg_length:.0f} words")
在 MLflow UI 中查看结果
评估指标被记录到 MLflow 中用于跟踪和比较。导航到**实验 (Experiments)** 选项卡以查看您的评估运行并比较不同迭代的结果。
您构建了什么
您现在拥有一个完整的、经过版本跟踪的 GenAI 应用程序，具备全面的可观察性和评估功能。
您完成了什么
- **版本化的提示词模板**：您的提示词现在已在 MLflow 中注册，并具有完整的版本历史，从而实现了可复现的实验和系统性的改进
- **集成的 LangChain 代理**：构建了一个具有自动 MLflow 集成的对话代理，该集成捕获每次交互以进行分析和调试
- **完整的追踪可观察性**：启用了全面的追踪日志记录，将所有模型交互链接到您的版本化应用程序，以实现完全的可见性
- **系统性评估**：使用基于关键字的指标实施了性能跟踪，并将结果记录到 MLflow 以进行持续的质量监控