标记会话

标记会话提供了一种结构化的方法，用于收集来自域专家的关于 GenAI 应用程序行为的反馈。 标记会话是一种特殊的 MLflow 运行类型，其中包含一组特定的跟踪，供领域专家使用 MLflow 评审应用进行审核。

标记会话的目标是在现有 Assessments上收集人工生成的（标签）。 可以捕获Feedback或Expectations，然后可以通过系统评估来使用它们改善 GenAI 应用。

注释

由于标记会话是 MLflow 运行，因此可以使用 MLflow SDK（例如Assessments）以编程方式访问收集的数据（跟踪及其关联mlflow.search_runs()），并在 MLflow UI 中可视化 - 每个标记会话将显示在“评估”选项卡中。

标记会话的工作原理

创建标记会话时，可以定义：

• 名称：会话的描述性标识符
• 分配的用户：将提供标签的域专家
• 代理：（可选） GenAI 应用，以根据需要生成响应
• 标签架构：反馈收集的问题和格式
• 多回合聊天：是否支持聊天样式标签

会话充当跟踪及其关联标签的容器，可实现可驱动评估和改进工作流的系统反馈收集。

创建标记会话

用于 mlflow.genai.labeling.create_labeling_session() 创建具有特定配置的新会话。

通过 UI 创建会话

导航到 MLflow UI 中的 “标记 ”选项卡以直观方式创建会话。 这提供了一个直观的界面，用于定义会话参数、分配用户和选择标签架构，而无需编写代码。

通过 UI 查看会话

导航到 MLflow UI 中的 “标记 ”选项卡，直观地查看会话。

以编程方式创建会话

使用 MLflow SDK 创建会话，并完全以编程方式控制所有配置选项。

基本会话创建

import mlflow.genai.labeling as labeling
import mlflow.genai.label_schemas as schemas

# Create a simple labeling session with built-in schemas
session = labeling.create_labeling_session(
    name="customer_service_review_jan_2024",
    assigned_users=["alice@company.com", "bob@company.com"],
    label_schemas=[schemas.EXPECTED_FACTS]  # Required: at least one schema needed
)

print(f"Created session: {session.name}")
print(f"Session ID: {session.labeling_session_id}")

警告

不保证标记会话名称是唯一的。 多个会话可以具有相同的名称。 对于可靠的程序化访问，请通过其 MLflow 运行 ID（session.mlflow_run_id）而非名称来存储和引用会话。

注释

创建标记会话时需要标签架构。 可以使用内置架构（EXPECTED_FACTS、、EXPECTED_RESPONSEGUIDELINES）或创建自定义架构。 有关创建和使用架构的详细信息，请参阅 标记架构指南 。

使用自定义标签架构的会话

import mlflow.genai.labeling as labeling
import mlflow.genai.label_schemas as schemas

# Create custom schemas first (see Labeling Schemas guide)
quality_schema = schemas.create_label_schema(
    name="response_quality",
    type="feedback",
    title="Rate the response quality",
    input=schemas.InputCategorical(options=["Poor", "Fair", "Good", "Excellent"]),
    overwrite=True,
)

# Create session using the schemas
session = labeling.create_labeling_session(
    name="quality_assessment_session",
    assigned_users=["expert@company.com"],
    label_schemas=["response_quality", schemas.EXPECTED_FACTS],
)

管理标记会话

警告

由于标记会话名称不唯一，按名称查找会话可能会返回多个匹配项或意外结果。 对于生产工作流，建议以 MLflow 运行 ID 存储和引用会话。

检索会话

import mlflow.genai.labeling as labeling

# Get all labeling sessions
all_sessions = labeling.get_labeling_sessions()
print(f"Found {len(all_sessions)} sessions")

for session in all_sessions:
    print(f"- {session.name} (ID: {session.labeling_session_id})")
    print(f"  Assigned users: {session.assigned_users}")

获取特定会话

import mlflow
import mlflow.genai.labeling as labeling
import pandas as pd

# Get all labeling sessions first
all_sessions = labeling.get_labeling_sessions()

# Find session by name (note: names may not be unique)
target_session = None
for session in all_sessions:
    if session.name == "customer_service_review_jan_2024":
        target_session = session
        break

if target_session:
    print(f"Session name: {target_session.name}")
    print(f"Experiment ID: {target_session.experiment_id}")
    print(f"MLflow Run ID: {target_session.mlflow_run_id}")
    print(f"Label schemas: {target_session.label_schemas}")
else:
    print("Session not found")

# Alternative: Get session by MLflow Run ID (if you know it)
run_id = "your_labeling_session_run_id"
run = mlflow.search_runs(
    experiment_ids=["your_experiment_id"],
    filter_string=f"tags.mlflow.runName LIKE '%labeling_session%' AND attribute.run_id = '{run_id}'"
).iloc[0]

print(f"Found labeling session run: {run['run_id']}")
print(f"Session name: {run['tags.mlflow.runName']}")

注释

由于标记会话名称不一定是唯一的，因此当需要以编程方式检索特定会话时，建议按其 MLflow 运行 ID 来存储和引用会话。

删除会话

import mlflow.genai.labeling as labeling

# Find the session to delete by name
all_sessions = labeling.get_labeling_sessions()
session_to_delete = None
for session in all_sessions:
    if session.name == "customer_service_review_jan_2024":
        session_to_delete = session
        break

if session_to_delete:
    # Delete the session (removes from Review App)
    review_app = labeling.delete_labeling_session(session_to_delete)
    print(f"Deleted session: {session_to_delete.name}")
else:
    print("Session not found")

将跟踪添加到会话

创建后，使用 add_traces() 方法将跟踪信息填入会话以供专家评审。

注释

有关在审核应用程序 UI 中跟踪如何被渲染和显示给标注者的详细信息，包括不同数据类型（字典、OpenAI 消息、工具调用）是如何呈现的，请参阅审核应用程序指南。

通过用户界面添加跟踪信息

还可以通过导航到 “跟踪 ”选项卡、选择要包含的跟踪以及使用导出功能将跟踪添加到标记会话，直接通过 MLflow UI 添加跟踪。

添加来自搜索结果的跟踪

import mlflow
import mlflow.genai.labeling as labeling
from openai import OpenAI

# First, let's create some sample traces with a simple app
# Connect to a Databricks LLM via OpenAI using the same credentials as MLflow
# Alternatively, you can use your own OpenAI credentials here
mlflow_creds = mlflow.utils.databricks_utils.get_databricks_host_creds()
client = OpenAI(
    api_key=mlflow_creds.token,
    base_url=f"{mlflow_creds.host}/serving-endpoints"
)

@mlflow.trace
def support_app(question: str):
    """Simple support app that generates responses"""
    mlflow.update_current_trace(tags={"test_tag": "C001"})
    response = client.chat.completions.create(
        model="databricks-claude-3-7-sonnet",  # This example uses Databricks hosted Claude 3.5 Sonnet. If you provide your own OpenAI credentials, replace with a valid OpenAI model e.g., gpt-4o, etc.
        messages=[
            {"role": "system", "content": "You are a helpful customer support agent."},
            {"role": "user", "content": question},
        ],
    )
    return {"response": response.choices[0].message.content}

# Generate some sample traces
with mlflow.start_run():
    # Create traces with negative feedback for demonstration
    support_app("My order is delayed")

    support_app("I can't log into my account")

# Now search for traces to label
traces_df = mlflow.search_traces(
    filter_string="tags.test_tag = 'C001'", max_results=50
)

# Create session and add traces
session = labeling.create_labeling_session(
    name="negative_feedback_review",
    assigned_users=["quality_expert@company.com"],
    label_schemas=["response_quality", "expected_facts"]
)

# Add traces from search results
session.add_traces(traces_df)
print(f"Added {len(traces_df)} traces to session")

添加单个跟踪对象

import mlflow
import mlflow.genai.labeling as labeling
from openai import OpenAI

# Set up the app to generate traces
# Connect to a Databricks LLM via OpenAI using the same credentials as MLflow
# Alternatively, you can use your own OpenAI credentials here
mlflow_creds = mlflow.utils.databricks_utils.get_databricks_host_creds()
client = OpenAI(
    api_key=mlflow_creds.token,
    base_url=f"{mlflow_creds.host}/serving-endpoints"
)

@mlflow.trace
def support_app(question: str):
    """Simple support app that generates responses"""
    mlflow.update_current_trace(tags={"test_tag": "C001"})
    response = client.chat.completions.create(
        model="databricks-claude-3-7-sonnet",  # This example uses Databricks hosted Claude 3.5 Sonnet. If you provide your own OpenAI credentials, replace with a valid OpenAI model e.g., gpt-4o, etc.
        messages=[
            {"role": "system", "content": "You are a helpful customer support agent."},
            {"role": "user", "content": question},
        ],
    )
    return {"response": response.choices[0].message.content}

# Generate specific traces for edge cases
with mlflow.start_run() as run:
    # Create traces for specific scenarios
    support_app("What's your refund policy?")
    trace_id_1 = mlflow.get_last_active_trace_id()

    support_app("How do I cancel my subscription?")
    trace_id_2 = mlflow.get_last_active_trace_id()

    support_app("The website is down")
    trace_id_3 = mlflow.get_last_active_trace_id()

# Get the trace objects
trace1 = mlflow.get_trace(trace_id_1)
trace2 = mlflow.get_trace(trace_id_2)
trace3 = mlflow.get_trace(trace_id_3)

# Create session and add traces
session = labeling.create_labeling_session(
    name="negative_feedback_review",
    assigned_users=["nikhil.thorat@databricks.com"],
    label_schemas=["response_quality", schemas.EXPECTED_FACTS],
)

# Add individual traces
session.add_traces([trace1, trace2, trace3])

管理分配的用户

用户访问要求

在 Databricks 帐户中，任何用户都可以被分配到标记会话中，无论他们是否有工作区访问权限。 但是，授予用户对标记会话的权限将允许他们访问标记会话的 MLflow 试验。

为用户设置权限

• 对于无权访问工作区的用户，帐户管理员使用帐户级 SCIM 预配将用户和组从标识提供者自动同步到 Databricks 帐户。 还可以在 Databricks 中设置标识时手动注册这些用户和组，以授予他们访问权限。 请参阅 用户和组管理。
• 对于已有权访问包含评审应用的工作区的用户，无需进行其他配置。

重要

将用户分配到标记会话时，系统会自动向包含标记会话的 MLflow 试验授予必要的 WRITE 权限。 这样，分配的用户可以访问查看和交互试验数据。

将用户添加到现有会话

import mlflow.genai.labeling as labeling

# Find existing session by name
all_sessions = labeling.get_labeling_sessions()
session = None
for s in all_sessions:
    if s.name == "customer_review_session":
        session = s
        break

if session:
    # Add more users to the session
    new_users = ["expert2@company.com", "expert3@company.com"]
    session.set_assigned_users(session.assigned_users + new_users)
    print(f"Session now has users: {session.assigned_users}")
else:
    print("Session not found")

替换已分配的用户

import mlflow.genai.labeling as labeling

# Find session by name
all_sessions = labeling.get_labeling_sessions()
session = None
for s in all_sessions:
    if s.name == "session_name":
        session = s
        break

if session:
    # Replace all assigned users
    session.set_assigned_users(["new_expert@company.com", "lead_reviewer@company.com"])
    print("Updated assigned users list")
else:
    print("Session not found")

同步到评估数据集

强大的功能是能够同步收集 Expectations 到 评估数据集。

数据集同步的工作原理

该sync()方法执行智能更新插入操作：

• 唯一键：每个跟踪的输入充当用于标识数据集中的记录的唯一键
• 预期更新：对于具有匹配输入的跟踪，标记会话的预期将覆盖数据集中的现有预期（如果预期名称相同）
• 新记录：标记会话中与输入匹配但数据集中不存在的跟踪，会被添加为新记录。
• 保留：具有不同输入的现有数据集记录保持不变

此方法允许你通过为现有示例添加新的示例和更新基础事实来迭代地改进评估数据集，而不会丢失以前的工作。

数据集同步

import mlflow.genai.labeling as labeling

# Find session with completed labels by name
all_sessions = labeling.get_labeling_sessions()
session = None
for s in all_sessions:
    if s.name == "completed_review_session":
        session = s
        break

if session:
    # Sync expectations to dataset
    session.sync(dataset_name="customer_service_eval_dataset")
    print("Synced expectations to evaluation dataset")
else:
    print("Session not found")

最佳做法

会话组织

• 描述性名称：使用明确的日期标记名称，例如 customer_service_review_march_2024
• 重点范围：使会话专注于特定的评估目标或时间段
• 适当大小：每个会话的跟踪数量应为 25 到 100 个，以避免审阅者疲劳

警告

会话标识：由于会话名称不唯一，在创建会话时始终存储 session.mlflow_run_id 。 使用运行 ID 进行编程访问，而不是依赖于会话名称。

import mlflow.genai.labeling as labeling

# Good: Store run ID for later reference
session = labeling.create_labeling_session(name="my_session", ...)
session_run_id = session.mlflow_run_id  # Store this!

# Later: Use run ID to find session via mlflow.search_runs()
# rather than searching by name through all sessions

用户管理

• 明确分配：根据域专业知识和可用性分配用户
• 均衡的工作负荷：跨多个专家均匀分配标记工作
• 权限感知：请记住，用户需要 Databricks 工作区访问权限

概要

MLflow 标记会话提供了一个结构化框架，用于收集有关 GenAI 应用程序的专家反馈。 通过将会话与 标签架构 和 评审应用相结合，您可以系统地收集高质量的人类评估，从而推动评估和改进工作流。

关键功能包括：

• 使用自定义配置创建灵活的会话
• 用户分配和权限管理
• 评估工作流的数据集同步

使用标记会话将即席反馈收集转换为系统、可重复的过程，从而不断改进 GenAI 应用程序。

后续步骤

• 标记现有轨迹 - 使用标记过程的分步指南
• 创建自定义标记架构 - 定义结构化反馈问题
• 生成评估数据集 - 将标记的会话转换为测试数据集