版本跟踪快速入门
使用 MLflow 的版本管理功能构建和跟踪基于 LangChain 的聊天机器人。本快速入门演示了使用 MLflow 的 GenAI 功能进行提示版本控制、应用程序跟踪、跟踪生成和性能评估。
先决条件
安装所需软件包
需要 MLflow 3
本快速入门需要 MLflow 3.0 或更高版本才能获得完整的 GenAI 功能。
bash
pip install --upgrade mlflow
pip install langchain-openai
设置 OpenAI API 密钥
配置您的 OpenAI API 密钥以进行 OpenAI 服务身份验证
bash
export OPENAI_API_KEY=your_api_key_here
您将学到什么
本快速入门涵盖了使用 MLflow 的版本管理系统构建可跟踪的 GenAI 应用程序的基本概念。
版本控制提示
注册和跟踪提示模板，拥有完整的版本历史记录，以实现可重复的实验
构建 LangChain 代理
创建具有自动 MLflow 集成和可观测性的对话代理
跟踪一切
启用全面的跟踪日志记录，以监控和调试模型的行为
评估性能
使用内置指标和自定义评估框架评估模型质量
让我们构建一个简单的 IT 支持聊天机器人，并使用 MLflow 跟踪其开发生命周期。
步骤 1：注册提示模板
首先创建一个版本化的提示模板。这使您可以跟踪提示的演变并确保跨实验的可重复性。
python
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 中查看您的提示
导航到“提示”选项卡以查看已注册的提示
步骤 2：构建 LangChain 对话链
创建一个简单的链，将提示模板与 OpenAI 的聊天模型结合起来
python
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：启用跟踪可观测性
设置自动跟踪日志记录，以监控模型在开发过程中的行为。这会创建所有模型交互的链接历史记录。
配置活动模型和自动日志记录
python
# 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()
生成测试跟踪
运行多个查询以生成用于分析的跟踪
python
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 中探索跟踪
- 查看已记录的模型：在实验的“模型”选项卡中进行检查
- 访问模型详细信息：单击模型以查看其唯一的
model_id
- 分析生成的跟踪：导航到“跟踪”选项卡以检查单个交互
步骤 4：评估模型性能
评估对于理解聊天机器人的性能以及确保随时间的质量改进至关重要。MLflow 为系统评估跟踪提供了基础。
关键评估能力
- 质量评估：使用关键字覆盖率和内容分析系统地评估响应质量，以确保聊天机器人满足期望
- 性能指标：跟踪响应长度和关键字匹配等量化度量，以监控随时间的改进
- 持续监控：将评估结果记录到 MLflow 中，以进行持续的性能跟踪和模型迭代之间的比较
使用 MLflow 的跟踪功能来评估聊天机器人的准确性和相关性与预期响应的匹配程度。
手动分析输出
通过将模型输出与预期响应进行比较来评估它们
python
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 中以进行跟踪和比较。导航到“实验”选项卡以查看评估运行并比较不同迭代的结果。
您构建了什么
现在您拥有一个完整的版本跟踪 GenAI 应用程序，具有全面的可观测性和评估功能。
您已完成的成就
- 版本化提示模板：您的提示现已在 MLflow 中注册，具有完整的版本历史记录，可实现可重复的实验和系统性的改进
- 集成 LangChain 代理：构建了一个具有自动 MLflow 集成的对话代理，该代理捕获每次交互以进行分析和调试
- 完整的跟踪可观测性：启用了全面的跟踪日志记录，将所有模型交互链接到您的版本化应用程序，以实现完全可见性
- 系统化评估：实施了具有基于关键字的指标的性能跟踪，这些指标将结果记录到 MLflow 中以进行持续的质量监控