提示注册表参考

概述

MLflow 提示注册表是一个集中存储库，用于在其生命周期内管理提示模板。 它使团队能够：

• 使用 Git 风格的版本控制、提交消息和回滚功能，对版本和提示进行跟踪
• 使用可变别名（例如，“生产”、“预发布”）安全地进行部署，以便进行 A/B 测试和逐步推出
• 通过允许非工程师通过 UI 修改提示，在没有代码更改的情况下进行协作
• 与任何框架（ 包括 LangChain、LlamaIndex 和其他 GenAI 框架）集成
• 通过 Unity Catalog 集成，进行访问控制和审计追踪以维护治理。
• 通过将提示链接到试验和评估结果来跟踪世系

先决条件

1. 通过 Unity 目录支持安装 MLflow：

   pip install --upgrade "mlflow[databricks]>=3.1.0"
   

2. 请按照 设置环境快速指南 创建 MLflow 试验。
3. 访问 Unity 目录中的 CREATE FUNCTION
   • 为什么？ 提示以函数的形式存储在 UC 中

快速入门

以下是使用提示注册表的基本工作流程：

import mlflow

# Register a prompt template
prompt = mlflow.genai.register_prompt(
    name="mycatalog.myschema.customer_support",
    template="You are a helpful assistant. Answer this question: {{question}}",
    commit_message="Initial customer support prompt"
)
print(f"Created version {prompt.version}")  # "Created version 1"

# Set a production alias
mlflow.genai.set_prompt_alias(
    name="mycatalog.myschema.customer_support",
    alias="production",
    version=1
)

# Load and use the prompt in your application
prompt = mlflow.genai.load_prompt(name_or_uri="prompts:/mycatalog.myschema.customer_support@production")
response = llm.invoke(prompt.render(question="How do I reset my password?"))

核心概念

SDK 概述

提示注册表提供六个主要操作：

register_prompt（）

创建新的提示或向现有提示添加新版本。

mlflow.genai.register_prompt(
    name: str,
    template: str,
    commit_message: str,
    tags: dict = None
) -> PromptVersion

参数

退货

PromptVersion 对象包含：

• name：提示名称
• version：版本号（自动递增）
• template：提示模板
• commit_message：提交消息
• creation_timestamp：创建版本时

示例：

prompt = mlflow.genai.register_prompt(
    name="mycatalog.myschema.summarization",
    template="""Summarize the following text in {{num_sentences}} sentences:

Text: {{content}}

Focus on: {{focus_areas}}""",
    commit_message="Added focus areas parameter",
    tags={
        "tested_with": "gpt-4",
        "avg_latency_ms": "1200",
        "team": "content",
        "project": "summarization-v2"
    }
)

load_prompt（）

按版本号或别名检索提示。

mlflow.genai.load_prompt(
    name_or_uri: str,
    version: Optional[Union[str, int]] = None,
    allow_missing: bool = False,
) -> Optional[PromptVersion]

参数

URI 格式

# Load specific version
prompt = mlflow.genai.load_prompt(name_or_uri="prompts:/mycatalog.myschema.chat_prompt/3")

# Load by alias
prompt = mlflow.genai.load_prompt(name_or_uri="prompts:/mycatalog.myschema.chat_prompt@production")

示例：

# Load specific version using URI format
v2 = mlflow.genai.load_prompt(name_or_uri="prompts:/mycatalog.myschema.qa_prompt/2")

# Load specific version using name + version parameter
v3 = mlflow.genai.load_prompt(name_or_uri="mycatalog.myschema.qa_prompt", version=3)

# Load by alias using URI
prod = mlflow.genai.load_prompt(name_or_uri="prompts:/mycatalog.myschema.qa_prompt@production")

# Load with allow_missing flag (returns None if not found)
optional_prompt = mlflow.genai.load_prompt(
    name_or_uri="mycatalog.myschema.qa_prompt",
    version=3,
    allow_missing=True
)

# Use the prompt
if optional_prompt is not None:
    response = llm.invoke(optional_prompt.render(
        question="What is MLflow?",
        context="MLflow is an open source platform..."
    ))

search_prompts（）

按目录和架构列出 Unity 目录中的提示。

mlflow.genai.search_prompts(
    filter_string: str,
    max_results: int = None
) -> PagedList[Prompt]

参数

Unity 目录要求

对于 Unity 目录提示注册表，必须同时指定目录和架构：

# REQUIRED format - list all prompts in a catalog.schema
results = mlflow.genai.search_prompts("catalog = 'mycatalog' AND schema = 'myschema'")

# This is the ONLY supported filter format
results = mlflow.genai.search_prompts("catalog = 'rohit' AND schema = 'default'")

局限性

Unity 目录中不支持以下筛选器：

• 名称模式： name LIKE '%pattern%'❌
• 标记筛选： tags.field = 'value'❌
• 确切的名称匹配： name = 'specific.name'❌
• 超越目录和架构的组合筛选器 ❌

若要查找特定提示，请使用返回的列表并以编程方式进行筛选：

# Get all prompts in schema
all_prompts = mlflow.genai.search_prompts("catalog = 'mycatalog' AND schema = 'myschema'")

# Filter programmatically
customer_prompts = [p for p in all_prompts if 'customer' in p.name.lower()]
tagged_prompts = [p for p in all_prompts if p.tags.get('team') == 'support']

set_prompt_alias（）

创建或更新指向特定版本的别名。

mlflow.genai.set_prompt_alias(
    name: str,
    alias: str,
    version: int
) -> None

参数

示例：

# Promote version 3 to production
mlflow.genai.set_prompt_alias(
    name="mycatalog.myschema.chat_assistant",
    alias="production",
    version=3
)

# Set up staging for testing
mlflow.genai.set_prompt_alias(
    name="mycatalog.myschema.chat_assistant",
    alias="staging",
    version=4
)

delete_prompt_alias（）

删除别名而不影响基础版本。

mlflow.genai.delete_prompt_alias(
    name: str,
    alias: str
) -> None

参数

delete_prompt（） 和delete_prompt_version（）

删除提示的全部或特定版本。

delete_prompt_version（）

删除提示的特定版本：

from mlflow import MlflowClient

client = MlflowClient()
client.delete_prompt_version(name: str, version: str) -> None

参数

delete_prompt（）

从注册表中删除整个提示：

from mlflow import MlflowClient

client = MlflowClient()
client.delete_prompt(name: str) -> None

参数

重要说明：

• 对于 Unity 目录注册表，如果任何版本仍然存在， delete_prompt() 将失败。 必须先使用 delete_prompt_version() 删除所有版本。
• 对于其他注册表类型， delete_prompt() 通常无需版本检查。

示例：

from mlflow import MlflowClient

client = MlflowClient()

# Delete specific versions first (required for Unity Catalog)
client.delete_prompt_version("mycatalog.myschema.chat_assistant", "1")
client.delete_prompt_version("mycatalog.myschema.chat_assistant", "2")
client.delete_prompt_version("mycatalog.myschema.chat_assistant", "3")

# Then delete the entire prompt
client.delete_prompt("mycatalog.myschema.chat_assistant")

# For convenience with Unity Catalog, you can also search and delete all versions:
search_response = client.search_prompt_versions("mycatalog.myschema.chat_assistant")
for version in search_response.prompt_versions:
    client.delete_prompt_version("mycatalog.myschema.chat_assistant", str(version.version))

client.delete_prompt("mycatalog.myschema.chat_assistant")

数据模型

提示注册表遵循类似于 Git 的模型：

• 提示：Unity Catalog 中的命名实体
• 版本：不可变快照，并具有自动递增编号
• 别名：指向特定版本的可变指针
• 标记：特定于版本的键值对

常见模式

模板变量

对变量使用双大括号语法：

# Single variable
simple_prompt = mlflow.genai.register_prompt(
    name="mycatalog.myschema.greeting",
    template="Hello {{name}}, how can I help you today?",
    commit_message="Simple greeting"
)

# Multiple variables with formatting
complex_prompt = mlflow.genai.register_prompt(
    name="mycatalog.myschema.analysis",
    template="""Analyze the following {{data_type}} data:

{{data}}

Consider these factors:
{{factors}}

Output format: {{output_format}}""",
    commit_message="Multi-variable analysis template"
)

# Use the prompt
rendered = complex_prompt.render(
    data_type="sales",
    data="Q1: $1.2M, Q2: $1.5M...",
    factors="seasonality, market trends",
    output_format="bullet points"
)

版本管理工作流

# Development workflow
def develop_prompt(base_name: str, changes: str):
    """Iterate on prompts during development"""

    # Register new version
    new_version = mlflow.genai.register_prompt(
        name=base_name,
        template=changes,
        commit_message=f"Dev iteration: {datetime.now()}"
    )

    # Update dev alias
    mlflow.genai.set_prompt_alias(
        name=base_name,
        alias="dev",
        version=new_version.version
    )

    return new_version

# Promotion workflow
def promote_prompt(name: str, from_env: str, to_env: str):
    """Promote prompt from one environment to another"""

    # Get current version in source environment
    source = mlflow.genai.load_prompt(f"prompts:/{name}@{from_env}")

    # Point target environment to same version
    mlflow.genai.set_prompt_alias(
        name=name,
        alias=to_env,
        version=source.version
    )

    print(f"Promoted {name} v{source.version} from {from_env} to {to_env}")

别名策略

# Standard environment aliases
ENVIRONMENT_ALIASES = ["dev", "staging", "production"]

# Feature branch aliases
def create_feature_alias(prompt_name: str, feature: str, version: int):
    """Create alias for feature development"""
    mlflow.genai.set_prompt_alias(
        name=prompt_name,
        alias=f"feature-{feature}",
        version=version
    )

# Regional aliases
REGIONAL_ALIASES = {
    "us": "production-us",
    "eu": "production-eu",
    "asia": "production-asia"
}

# Rollback-ready aliases
def safe_production_update(name: str, new_version: int):
    """Update production with rollback capability"""
    try:
        # Save current production
        current = mlflow.genai.load_prompt(f"prompts:/{name}@production")
        mlflow.genai.set_prompt_alias(name, "production-previous", current.version)
    except:
        pass  # No current production

    # Update production
    mlflow.genai.set_prompt_alias(name, "production", new_version)

框架集成

LangChain

from langchain_core.prompts import ChatPromptTemplate

# Load from registry
mlflow_prompt = mlflow.genai.load_prompt("prompts:/mycatalog.myschema.chat@production")

# Convert to LangChain format
langchain_template = mlflow_prompt.to_single_brace_format()
chat_prompt = ChatPromptTemplate.from_template(langchain_template)

# Use in chain
chain = chat_prompt | llm | output_parser

LlamaIndex

from llama_index.core import PromptTemplate

# Load and convert
mlflow_prompt = mlflow.genai.load_prompt("prompts:/mycatalog.myschema.rag@production")
llama_template = PromptTemplate(mlflow_prompt.to_single_brace_format())

# Use in query engine
query_engine.update_prompts({"response_synthesizer:text_qa_template": llama_template})

后续步骤

• 创建和编辑提示 - 从第一个提示开始
• 评估提示 - 系统地比较提示版本
• 使用在已部署应用中的提示——利用别名将提示部署到生产环境