MLflow LangChain Flavor

信息

langchain Flavor 正在积极开发中，并被标记为“实验性”。公共 API 可能会发生变化，并且随着 Flavor 的发展，可能会添加新功能。

欢迎来到 MLflow 与 LangChain 集成的开发者指南。本指南是理解和利用 LangChain 和 MLflow 的组合功能来开发高级语言模型应用程序的全面资源。

LangChain 是一个用于构建由语言模型驱动的应用程序的通用框架。它在创建利用语言模型进行推理和生成响应的上下文感知应用程序方面表现出色，从而能够开发复杂的 NLP 应用程序。

LangGraph 是 Langchain 创建者提供的互补的基于代理的框架，支持有状态代理和多代理 GenAI 应用程序的创建。LangGraph 使用 LangChain 来与 GenAI 代理组件进行接口。

为什么在 LangChain 中使用 MLflow？

除了使用 MLflow 进行机器学习模型管理和部署的优势之外，LangChain 与 MLflow 的集成还提供了许多与在更广泛的 MLflow 生态系统中 M使用 LangChain 相关的优势。

实验跟踪

LangChain 在试验各种代理、工具和检索器方面的灵活性与 MLflow Tracking 结合后变得更加强大。这种组合可以实现快速的实验和迭代。您可以轻松地比较运行，从而更容易地优化模型并加快从开发到生产部署的进程。

依赖管理

利用 MLflow 自动管理和记录代码及环境依赖项的能力，自信地部署您的 LangChain 应用程序。您还可以显式声明外部资源依赖项，例如 LangChain 应用程序查询的 LLM 服务端点或向量搜索索引。MLflow 会将这些依赖项作为模型元数据进行跟踪，以便下游服务系统能够确保从已部署的 LangChain 应用程序到这些依赖资源的身份验证正常工作。

这些功能确保了开发和生产环境之间的一致性，并通过较少的手动干预降低了部署风险。

MLflow 评估

MLflow Evaluate 在 MLflow 中提供了评估语言模型的原生功能。借助此功能，您可以轻松地对 LangChain 应用程序的推理结果使用自动评估算法。此功能有助于高效评估 LangChain 应用程序的推理结果，确保强大的性能分析。

可观察性

MLflow Tracing 是 MLflow 的一项新功能，可让您跟踪数据如何在 LangChain 链/代理等中流动。此功能提供了数据流的视觉表示，使其更容易理解 LangChain 应用程序的行为并识别潜在的瓶颈或问题。凭借其强大的自动跟踪功能，您只需运行一次 mlflow.langchain.autolog() 命令，即可在不更改任何代码的情况下 instrument 您的 LangChain 应用程序。

自动日志记录

Autologging 是一站式解决方案，只需一行代码 mlflow.langchain.autolog() 即可实现上述所有优势。通过启用自动日志记录，您可以轻松地自动记录 LangChain 应用程序的所有组件，包括链、代理和检索器。此功能简化了跟踪和管理 LangChain 应用程序的过程，让您可以专注于开发和改进模型。有关如何使用此功能的更多信息，请参阅 MLflow LangChain Autologging 文档。

MLflow LangChain 集成中支持的元素

代理 (Agents)
检索器 (Retrievers)
可运行对象 (Runnables)
LangGraph 编译图（仅通过从代码记录模型支持）
LLMChain（已弃用，仅支持 langchain<0.3.0）
RetrievalQA（已弃用，仅支持 langchain<0.3.0）

警告

当使用合作伙伴包（如 langchain-openai）中的 LangChain 组件记录链或代理时，存在一个已知的反序列化问题。如果使用旧版基于日志记录的序列化来记录此类模型，则某些组件可能会从相应的 langchain-community 包而不是合作伙伴包库加载，这可能导致执行代码时出现意外行为或导入错误。为避免此问题，我们强烈建议使用从代码记录模型方法来记录此类模型。此方法允许您绕过模型序列化并稳健地保存模型定义。

信息

记录包含 ChatOpenAI 和 AzureChatOpenAI 的链/代理需要 MLflow>=2.12.0 和 LangChain>=0.0.307。

链、代理和检索器的概述

链 (Chain)
代理 (Agents)
检索器 (Retrievers)

在代码中硬编码的动作或步骤序列。LangChain 中的链将提示、模型和输出解析器等各种组件组合起来，以创建处理步骤的流程。

下图显示了一个示例，其中顶部部分直接通过 API 调用与 SaaS LLM 进行交互，而无需考虑对话历史记录。底部部分显示了相同的查询被提交到 LangChain 链，该链包含对话历史记录状态，以便整个对话历史记录都包含在每个后续输入中。以这种方式保留对话上下文是创建“聊天机器人”的关键。

The importance of stateful storage of conversation history for chat applications

开始使用 MLflow LangChain Flavor - 教程和指南

入门教程

在本入门教程中，您将学习 LangChain 最基本的组件，以及如何利用与 MLflow 的集成来存储、检索和使用链。

LangChain 快速入门

通过探索最简单的链配置——将提示和模型链接起来创建单个用途的实用应用程序——来开始使用 MLflow 和 LangChain。

高级教程

在这些教程中，您可以了解 LangChain 与 MLflow 的更复杂用法。强烈建议在探索这些更高级的用例之前，先阅读入门教程。

RAG 教程（使用 LangChain）

了解如何构建一个集成了 MLflow 的 LangChain RAG，以回答有关商业活动合法性的高度具体的问题。

从代码记录模型

自 MLflow 2.12.2 起，MLflow 引入了直接从代码定义记录 LangChain 模型的功能。

该功能在管理 LangChain 模型方面提供了多项优势

避免序列化复杂性：文件句柄、套接字、外部连接、动态引用、lambda 函数和系统资源是无法 pick 化的。某些 LangChain 组件不支持原生序列化，例如 RunnableLambda。
无 Pickling：在与序列化对象时使用的 Python 版本不同的 Python 版本中加载 pickle 或 cloudpickle 文件，不能保证兼容性。
可读性：序列化对象通常很难被人类阅读。代码中的模型允许您通过代码查看模型定义。

有关此功能的更多信息，请参阅从代码中获取模型功能文档。

要使用此功能，您将使用 mlflow.models.set_model() API 来定义您想要记录为 MLflow 模型的链。在代码中定义链之后，在记录模型时，您将指定定义链的文件的路径。

以下示例演示了如何使用此方法记录一个简单的链

在单独的 Python 文件中定义链

提示

如果您使用的是 Jupyter Notebook，您可以使用 %%writefile 魔术命令将代码单元直接写入文件，而无需离开 Notebook 手动创建它。

# %%writefile chain.py

import os
from operator import itemgetter

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import OpenAI

import mlflow

mlflow.set_experiment("Homework Helper")

mlflow.langchain.autolog()

prompt = PromptTemplate(
    template="You are a helpful tutor that evaluates my homework assignments and provides suggestions on areas for me to study further."
    " Here is the question: {question} and my answer which I got wrong: {answer}",
    input_variables=["question", "answer"],
)


def get_question(input):
    default = "What is your name?"
    if isinstance(input_data[0], dict):
        return input_data[0].get("content").get("question", default)
    return default


def get_answer(input):
    default = "My name is Bobo"
    if isinstance(input_data[0], dict):
        return input_data[0].get("content").get("answer", default)
    return default


model = OpenAI(temperature=0.95)

chain = (
    {
        "question": itemgetter("messages") | RunnableLambda(get_question),
        "answer": itemgetter("messages") | RunnableLambda(get_answer),
    }
    | prompt
    | model
    | StrOutputParser()
)

mlflow.models.set_model(chain)

然后，从主 Notebook 中，通过提供定义链的文件路径来记录模型

from pprint import pprint

import mlflow

chain_path = "chain.py"

with mlflow.start_run():
    info = mlflow.langchain.log_model(lc_model=chain_path, name="chain")

现在，chain.py 中定义的模型已记录到 MLflow。您可以加载模型并运行推理

# Load the model and run inference
homework_chain = mlflow.langchain.load_model(model_uri=info.model_uri)

exam_question = {
    "messages": [
        {
            "role": "user",
            "content": {
                "question": "What is the primary function of control rods in a nuclear reactor?",
                "answer": "To stir the primary coolant so that the neutrons are mixed well.",
            },
        },
    ]
}

response = homework_chain.invoke(exam_question)

pprint(response)

您可以看到模型已作为代码记录在 MLflow UI 中

Logging a LangChain model from a code script file

警告

在从代码记录模型时，请确保您的代码不包含任何敏感信息，例如 API 密钥、密码或其他机密数据。代码将以纯文本形式存储在 MLflow 模型工件中，任何有权访问该工件的人都可以查看代码。

详细文档

要了解有关 MLflow LangChain Flavor 的详细信息，请阅读下面的详细指南。

常见问题

我无法加载我的链！

允许危险的反序列化：LangChain 中的 Pickle 选择加入逻辑将阻止通过 MLflow 加载组件。您可能会看到类似这样的错误
```
ValueError: This code relies on the pickle module. You will need to set allow_dangerous_deserialization=True if you want to opt-in to
allow deserialization of data using pickle. Data can be compromised by a malicious actor if not handled properly to include a malicious
payload that when deserialized with pickle can execute arbitrary code on your machine.
```
LangChain 中的一项更改强制用户选择加入 pickle 反序列化，这可能会导致加载 MLflow 记录的链、向量存储、检索器和代理时出现一些问题。由于选项未针对每个组件暴露以在加载函数上设置此参数，因此您需要确保在记录模型时直接在此定义的加载函数中设置此选项。未设置此值的 LangChain 组件将得以保存，但如果未设置，加载时将引发 ValueError。

要解决此问题，只需重新记录模型，在定义的加载函数中指定选项 allow_dangerous_deserialization=True。有关在记录 FAISS 向量存储实例（在 loader_fn 声明中）时指定此选项的示例，请参阅LangChain 检索器的教程。

我无法将我的链、代理或检索器与 MLflow 一起保存。

提示

如果您在记录或保存 LangChain 组件与 MLflow 时遇到问题，请参阅从代码中获取模型功能文档，以确定是否从脚本文件记录模型是否提供了更简单、更稳健的记录解决方案！

Cloudpickle 的序列化挑战：根据对象的复杂性，使用 cloudpickle 进行序列化可能会遇到限制。

某些对象，特别是那些具有复杂内部状态或依赖于外部系统资源的对象，并非天生就可 pickle。这种限制的出现是因为序列化本质上需要将对象转换为字节流，这对于与系统状态紧密耦合的对象或进行外部 I/O 操作的对象来说可能很复杂。尝试将 PyDantic 升级到 2.x 版本来解决此问题。
验证原生序列化支持：确保 langchain 对象（链、代理或检索器）可以使用 langchain API 原生序列化，如果使用 MLflow 保存或记录不成功。

由于其复杂的结构，并非所有 langchain 组件都易于序列化。如果原生序列化不受支持且 MLflow 不支持保存模型，您可以在LangChain 存储库中提交问题，或在LangChain 讨论板中寻求指导。
跟上 MLflow 的新功能：MLflow 可能不会立即支持最新的 LangChain 功能。

如果 MLflow 不支持新功能，请考虑在MLflow GitHub 问题页面上提交功能请求。随着积极开发中的库（如LangChain 的发布速度）的快速变化，破坏性更改、API 重构以及对现有功能的基本功能支持都可能导致集成问题。如果您希望看到链、代理、检索器或 LangChain 中的任何未来结构得到支持，请告知我们！

保存模型时出现 AttributeError

处理 LangChain 和 MLflow 中的依赖项安装：LangChain 和 MLflow 不会自动安装所有依赖项。

保存或记录模型时，可能需要为特定代理、检索器或工具所需的其他包进行显式定义。如果您的模型依赖于这些不包含在标准 LangChain 包中的外部组件库（特别是对于工具），则这些依赖项不一定会始终作为模型的一部分自动记录（请参阅下文，了解如何包含它们的指南）。
声明额外的依赖项：保存和记录时使用 extra_pip_requirements 参数。

当保存或记录包含核心 langchain 安装之外的外部依赖项的模型时，您将需要这些额外的依赖项。模型 Flavor 包含两个用于声明这些依赖项的选项：extra_pip_requirements 和 pip_requirements。虽然指定 pip_requirements 是完全有效的，但我们推荐使用 extra_pip_requirements，因为它不需要定义使用 langchain 模型进行推理所需的所有核心依赖包（其他核心依赖项将自动推断）。

如何在 LangChain 中使用流式 API？

使用 LangChain 模型进行流式传输：确保 LangChain 模型支持流式响应，并使用 MLflow 版本 >= 2.12.2。

从 MLflow 2.12.2 版本开始，使用 MLflow 2.12.2（或更高版本）保存的、支持流式响应的 LangChain 模型可以加载并用于可流式推理，使用 predict_stream API。确保您正确消耗返回类型，因为这些模型的返回是一个 Generator 对象。要了解更多信息，请参阅predict_stream 指南。

如何将 LangGraph 构建的代理记录到 MLflow？

LangGraph 与 MLflow 的集成旨在利用 MLflow 中的从代码中获取模型功能来拓宽和简化代理序列化的支持。

要记录 LangGraph 代理，您可以将代理代码定义在一个脚本中，如下所示，并保存到文件 langgraph.py

from typing import Literal

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

import mlflow


@tool
def get_weather(city: Literal["seattle", "sf"]):
    """Use this to get weather information."""
    if city == "seattle":
        return "It's probably raining. Again."
    elif city == "sf":
        return "It's always sunny in sf"


llm = ChatOpenAI()
tools = [get_weather]
graph = create_react_agent(llm, tools)

# specify the Agent as the model interface to be loaded when executing the script
mlflow.models.set_model(graph)

当您准备将此代理脚本定义记录到 MLflow 时，您可以在定义模型时直接引用此保存的脚本

import mlflow

input_example = {
    "messages": [{"role": "user", "content": "what is the weather in seattle today?"}]
}

with mlflow.start_run():
    model_info = mlflow.langchain.log_model(
        lc_model="./langgraph.py",  # specify the path to the LangGraph agent script definition
        name="langgraph",
        input_example=input_example,
    )

当代理从 MLflow 加载时，脚本将被执行，定义的代理将可用于调用。

代理可以加载并用于推理，如下所示

agent = mlflow.langchain.load_model(model_info.model_uri)
query = {
    "messages": [
        {
            "role": "user",
            "content": "Should I bring an umbrella today when I go to work in San Francisco?",
        }
    ]
}
agent.invoke(query)

如何在 PyFunc predict 中控制输入是否转换为 List[langchain.schema.BaseMessage]？

默认情况下，MLflow 会将聊天请求格式的输入 {"messages": [{"role": "user", "content": "some_question"}]} 转换为 List[langchain.schema.BaseMessage]，例如 [HumanMessage(content="some_question")]，用于某些模型类型。要强制进行转换，请将环境变量 MLFLOW_CONVERT_MESSAGES_DICT_FOR_LANGCHAIN 设置为 True。要禁用此行为，请将环境变量 MLFLOW_CONVERT_MESSAGES_DICT_FOR_LANGCHAIN 设置为 False，如下所示

import json
import mlflow
import os
from operator import itemgetter
from langchain.schema.runnable import RunnablePassthrough

model = RunnablePassthrough.assign(
    problem=lambda x: x["messages"][-1]["content"]
) | itemgetter("problem")

input_example = {
    "messages": [
        {
            "role": "user",
            "content": "Hello",
        }
    ]
}
# this model accepts the input_example
assert model.invoke(input_example) == "Hello"

# set this environment variable to avoid input conversion
os.environ["MLFLOW_CONVERT_MESSAGES_DICT_FOR_LANGCHAIN"] = "false"
with mlflow.start_run():
    model_info = mlflow.langchain.log_model(
        model, name="model", input_example=input_example
    )

pyfunc_model = mlflow.pyfunc.load_model(model_info.model_uri)
assert pyfunc_model.predict(input_example) == ["Hello"]

为什么在 LangChain 中使用 MLflow？​

实验跟踪​

依赖管理​

MLflow 评估​

可观察性​

自动日志记录​

MLflow LangChain 集成中支持的元素​

链、代理和检索器的概述​

开始使用 MLflow LangChain Flavor - 教程和指南​

入门教程​

高级教程​

从代码记录模型​

详细文档​

常见问题​

我无法加载我的链！​

我无法将我的链、代理或检索器与 MLflow 一起保存。​

保存模型时出现 AttributeError​

如何在 LangChain 中使用流式 API？​

如何将 LangGraph 构建的代理记录到 MLflow？​

如何在 PyFunc predict 中控制输入是否转换为 List[langchain.schema.BaseMessage]？​