MLflow 中的 LangChain（实验性）

信息

langchain 风格目前正在积极开发中，并标记为实验性。公共 API 正在不断发展，并且正在添加新功能以增强其功能。

概述

LangChain 是一个 Python 框架，用于创建由语言模型驱动的应用程序。它为开发上下文感知应用程序提供了独特的功能，这些应用程序利用语言模型进行推理和生成响应。与 MLflow 的集成简化了复杂 NLP 应用程序的开发和部署。

LangChain 的技术本质

上下文感知应用程序：LangChain 专门将语言模型连接到各种上下文源，使它们能够生成更相关和准确的输出。
推理能力：它利用语言模型的力量来推理给定的上下文，并根据它采取适当的行动。
灵活的链式组合：LangChain 表达式语言 (LCEL) 允许从基本组件轻松构建复杂的链，支持流式传输、并行处理和日志记录等功能。

使用 LangChain 构建链

基本组件：LangChain 方便地将提示模板、模型和输出解析器等组件链接在一起，以创建复杂的工作流程。
示例 - 笑话生成器:
- 一个基本的链可以接受一个主题，并使用提示模板、ChatOpenAI 模型和输出解析器的组合来生成一个笑话。
- 这些组件使用 | 运算符链接在一起，类似于 Unix 管道，允许一个组件的输出馈送到下一个组件。
高级用例:
- LangChain 还支持更复杂的设置，例如检索增强生成 (RAG) 链，该链可以在响应问题时添加上下文。

与 MLflow 集成

简化的日志记录和加载：MLflow 的 langchain 风格提供了诸如 log_model() 和 load_model() 之类的函数，可以轻松地在 MLflow 生态系统中记录和检索 LangChain 模型。
简化的部署：在 MLflow 中记录的 LangChain 模型可以解释为通用的 Python 函数，从而简化了它们在各种应用程序中的部署和使用。通过直接集成到您记录的模型中的依赖管理，您可以部署您的应用程序，因为您知道用于训练模型的环境将用于为其提供服务。
多功能模型交互：该集成允许开发人员结合 MLflow 强大的模型跟踪和管理功能来利用 LangChain 的独特功能。
自动日志记录：MLflow 的 langchain 风格提供 LangChain 模型的自动日志记录，这会自动记录用于推理的工件、指标和模型。

langchain 模型风格可以通过 mlflow.langchain.save_model() 和 mlflow.langchain.log_model() 函数在 MLflow 格式中记录 LangChain 模型。使用这些函数还会将 python_function 风格添加到它们生成的 MLflow 模型中，从而允许通过 mlflow.pyfunc.load_model() 将模型解释为用于推理的通用 Python 函数。

您还可以使用 mlflow.langchain.load_model() 函数将保存或记录的具有 langchain 风格的 MLflow 模型加载为模型属性的字典。

基本示例：在 MLflow 中记录 LangChain `LLMChain`

import os

from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

import mlflow

# Ensure the OpenAI API key is set in the environment
assert (
    "OPENAI_API_KEY" in os.environ
), "Please set the OPENAI_API_KEY environment variable."

# Initialize the OpenAI model and the prompt template
llm = OpenAI(temperature=0.9)
prompt = PromptTemplate(
    input_variables=["product"],
    template="What is a good name for a company that makes {product}?",
)

# Create the LLMChain with the specified model and prompt
chain = LLMChain(llm=llm, prompt=prompt)

# Log the LangChain LLMChain in an MLflow run
with mlflow.start_run():
    logged_model = mlflow.langchain.log_model(chain, name="langchain_model")

# Load the logged model using MLflow's Python function flavor
loaded_model = mlflow.pyfunc.load_model(logged_model.model_uri)

# Predict using the loaded model
print(loaded_model.predict([{"product": "colorful socks"}]))

该示例的输出如下所示

["\n\nColorful Cozy Creations."]

简单 LLMChain 示例展示的内容

集成灵活性：该示例突出显示了如何轻松地在 MLflow 中记录由 OpenAI 模型和自定义提示模板组成的 LangChain 的 LLMChain。
简化的模型管理：通过 MLflow 的 langchain 风格，链被记录下来，从而可以进行版本控制、跟踪和轻松检索。
易于部署：使用 MLflow 的 pyfunc 模块加载记录的 LangChain 模型，说明了 LangChain 模型在 MLflow 中的直接部署过程。
实际应用：最后的预测步骤展示了模型在现实场景中的功能，即基于给定的产品生成公司名称。

使用 MLflow 记录 LangChain 代理

什么是代理？

LangChain 中的代理利用语言模型来动态确定和执行一系列操作，这与链中硬编码的序列形成对比。要了解有关代理的更多信息并查看 LangChain 中的其他示例，您可以阅读 LangChain 关于代理的文档。

代理的关键组成部分

代理

驱动决策的核心链，利用语言模型和提示。
接收工具描述、用户目标和先前执行的步骤等输入。
输出下一个操作集 (AgentActions) 或最终响应 (AgentFinish)。

工具

代理调用以完成任务的函数。
提供适当的工具并准确描述它们以供有效使用至关重要。

工具包

为特定任务量身定制的工具集合。
LangChain 提供了一系列内置工具包并支持自定义工具包创建。

AgentExecutor

执行代理决策的运行时环境。
处理工具错误和代理输出解析等复杂性。
确保全面的日志记录和可观察性。

其他代理运行时

除了 AgentExecutor 之外，LangChain 还支持实验性运行时，例如 Plan-and-execute Agent、Baby AGI 和 Auto GPT。
还促进了自定义运行时逻辑创建。

记录 LangChain 代理的示例

此示例说明了在 MLflow 中记录 LangChain 代理的过程，突出显示了 LangChain 复杂的代理功能与 MLflow 强大的模型管理的集成。

import os

from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.llms import OpenAI

import mlflow

# Note: Ensure that the package 'google-search-results' is installed via pypi to run this example
# and that you have a accounts with SerpAPI and OpenAI to use their APIs.

# Ensuring necessary API keys are set
assert (
    "OPENAI_API_KEY" in os.environ
), "Please set the OPENAI_API_KEY environment variable."
assert (
    "SERPAPI_API_KEY" in os.environ
), "Please set the SERPAPI_API_KEY environment variable."

# Load the language model for agent control
llm = OpenAI(temperature=0)

# Next, let's load some tools to use. Note that the `llm-math` tool uses an LLM, so we need to pass that in.
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# Finally, let's initialize an agent with the tools, the language model, and the type of agent we want to use.
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

# Log the agent in an MLflow run
with mlflow.start_run():
    logged_model = mlflow.langchain.log_model(agent, name="langchain_model")

# Load the logged agent model for prediction
loaded_model = mlflow.pyfunc.load_model(logged_model.model_uri)

# Generate an inference result using the loaded model
question = "What was the high temperature in SF yesterday in Fahrenheit? What is that number raised to the .023 power?"

answer = loaded_model.predict([{"input": question}])

print(answer)

上面示例的输出如下所示

["1.1044000282035853"]

简单代理示例展示的内容

复杂的代理日志记录：演示了如何在 MLflow 中记录 LangChain 的复杂代理，该代理利用多个工具和语言模型。
高级工具的集成：展示了将“serpapi”和“llm-math”等其他工具与 LangChain 代理一起使用，强调了该框架集成复杂功能的能力。
代理初始化和使用：详细介绍了使用特定工具和模型设置初始化 LangChain 代理的过程，以及如何使用它来执行复杂的查询。
高效的模型管理和部署：说明了使用 MLflow 管理和部署复杂 LangChain 代理的简易性，从日志记录到预测。

使用 LangChain 和 GenAI LLM 进行实时流式输出

注意

通过 predict_stream API 的流式响应仅在 MLflow 版本 >= 2.12.2 中可用。以前版本的 MLflow 不支持流式响应。

流式输出功能概述

MLflow 内的 LangChain 集成支持来自支持此类功能的各种 GenAI 语言模型 (LLM) 的实时流式输出。此功能对于需要立即、增量响应的应用程序至关重要，从而促进动态交互，例如对话代理或实时内容生成。

支持的流式模型

LangChain 旨在与任何提供流式输出功能的 LLM 无缝协作。这包括来自 OpenAI 等提供商的某些模型（例如，ChatGPT 的特定版本），以及来自支持类似功能的不同供应商的其他 LLM。

使用 `predict_stream` 进行流式输出

MLflow pyfunc LangChain 风格中的 predict_stream 方法旨在处理同步输入并以流式方式提供输出。此方法对于通过在模型的响应可用时传递部分模型响应（而不是等待整个响应生成完成）来保持引人入胜的用户体验特别有用。

示例用法

以下示例演示了如何设置和使用 predict_stream 函数以及在 MLflow 中管理的 LangChain 模型，突出显示了实时响应生成

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI
import mlflow


template_instructions = "Provide brief answers to technical questions about {topic} and do not answer non-technical questions."
prompt = PromptTemplate(
    input_variables=["topic"],
    template=template_instructions,
)
chain = LLMChain(llm=OpenAI(temperature=0.05), prompt=prompt)

with mlflow.start_run():
    model_info = mlflow.langchain.log_model(chain, name="tech_chain")

# Assuming the model is already logged in MLflow and loaded
loaded_model = mlflow.pyfunc.load_model(model_uri=model_info.model_uri)

# Simulate a single synchronous input
input_data = "Hello, can you explain streaming outputs?"

# Generate responses in a streaming fashion
response_stream = loaded_model.predict_stream(input_data)
for response_part in response_stream:
    print("Streaming Response Part:", response_part)
    # Each part of the response is handled as soon as it is generated

与回调的高级集成

LangChain 的架构还支持在流式输出上下文中使用回调。这些回调可用于通过允许在流式传输过程中触发操作（例如记录中间响应或在传递之前修改它们）来增强功能。

注意

回调处理程序的大多数用途都涉及记录链或检索器中各种服务和工具调用的跟踪。为了简单起见，下面显示了一个简单的 stdout 回调处理程序。真实的回调处理程序必须是 LangChain 中 BaseCallbackHandler 类的子类。

from langchain_core.callbacks import StdOutCallbackHandler

handler = StdOutCallbackHandler()

# Attach callback to enhance the streaming process
response_stream = loaded_model.predict_stream(input_data, callback_handlers=[handler])
for enhanced_response in response_stream:
    print("Enhanced Streaming Response:", enhanced_response)

这些示例和说明展示了开发人员如何在 MLflow 中利用 LangChain 模型的实时流式输出功能，从而能够创建高度响应和交互式应用程序。

使用 MLflow 增强 RetrievalQA 链的管理

LangChain 与 MLflow 的集成引入了一种更有效的方式来管理和利用 RetrievalQA 链，这是 LangChain 功能的关键方面。这些链巧妙地将数据检索与问答过程结合起来，从而利用了语言模型的优势。

RetrievalQA 链的关键见解

RetrievalQA 链功能：这些链代表了一种复杂的 LangChain 功能，其中信息检索与基于语言模型的问答无缝融合。它们擅长于需要语言模型查阅特定数据或文档以获得准确响应的场景。
检索对象的作用：RetrievalQA 链的核心在于检索器对象，其任务是获取响应查询的相关文档或数据。

RAG 流程的详细概述

文档加载器：方便从各种来源加载文档，拥有 100 多个加载器和集成。
文档转换器：通过将文档转换和分割成可管理的部分来准备文档以进行检索。
文本嵌入模型：生成文本的语义嵌入，从而提高数据检索的相关性和效率。
向量存储：专门的数据库，用于存储和促进文本嵌入的搜索。
检索器：采用各种检索技术，从简单的语义搜索到更复杂的方法，如父文档检索器和集成检索器。

使用 MLflow 阐明向量数据库管理

传统的 LangChain 序列化：LangChain 通常需要手动管理检索器对象的序列化，包括处理向量数据库。
MLflow 的简化：MLflow 中的 langchain 风格大大简化了这个过程。它会自动执行序列化，管理 persist_dir 的内容和 loader_fn 函数的 Pickle 序列化。

关键 MLflow 组件和 VectorDB 日志记录

persist_dir：存储检索器对象（包括向量数据库）的目录。
loader_fn：用于从其存储位置加载检索器对象的函数。

重要注意事项

VectorDB 日志记录：MLflow 通过其 langchain 风格，确实将向量数据库作为检索器对象的一部分进行管理。但是，向量数据库本身不会在 MLflow 中显式记录为单独的实体。
运行时 VectorDB 维护：在训练和运行时环境之间保持向量数据库的一致性至关重要。虽然 MLflow 管理检索器对象的序列化，但确保在运行时可以访问相同的向量数据库对于保持一致的性能仍然至关重要。

记录 LangChain RetrievalQA 链的示例

import os
import tempfile

from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.llms import OpenAI
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS

import mlflow

assert (
    "OPENAI_API_KEY" in os.environ
), "Please set the OPENAI_API_KEY environment variable."

with tempfile.TemporaryDirectory() as temp_dir:
    persist_dir = os.path.join(temp_dir, "faiss_index")

    # Create the vector db, persist the db to a local fs folder
    loader = TextLoader("tests/langchain/state_of_the_union.txt")
    documents = loader.load()
    text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
    docs = text_splitter.split_documents(documents)
    embeddings = OpenAIEmbeddings()
    db = FAISS.from_documents(docs, embeddings)
    db.save_local(persist_dir)

    # Create the RetrievalQA chain
    retrievalQA = RetrievalQA.from_llm(llm=OpenAI(), retriever=db.as_retriever())

    # Log the retrievalQA chain
    def load_retriever(persist_directory):
        embeddings = OpenAIEmbeddings()
        vectorstore = FAISS.load_local(persist_directory, embeddings)
        return vectorstore.as_retriever()

    with mlflow.start_run() as run:
        logged_model = mlflow.langchain.log_model(
            retrievalQA,
            name="retrieval_qa",
            loader_fn=load_retriever,
            persist_dir=persist_dir,
        )

# Load the retrievalQA chain
loaded_model = mlflow.pyfunc.load_model(logged_model.model_uri)
print(
    loaded_model.predict(
        [{"query": "What did the president say about Ketanji Brown Jackson"}]
    )
)

上面示例的输出如下所示

[" The president said..."]

在 MLflow 中记录和评估 LangChain 检索器

MLflow 中的 langchain 风格将其功能扩展到包括检索器对象的日志记录和单独评估。此功能对于评估检索器检索到的文档的质量而无需通过大型语言模型 (LLM) 进行处理尤其有价值。

记录各个检索器的目的

独立评估：允许评估检索器在获取相关文档方面的性能，而无需考虑其后续在 LLM 中的使用情况。
质量保证：方便评估检索器在获取准确和上下文相关的文档方面的有效性。

在 MLflow 中记录检索器的要求

persist_dir：指定存储检索器对象的位置。
loader_fn：详细说明用于从其存储位置加载检索器对象的函数。
这些要求与记录 RetrievalQA 链的要求一致，确保过程的一致性。

记录 LangChain 检索器的示例

import os
import tempfile

from langchain.document_loaders import TextLoader
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS

import mlflow

assert (
    "OPENAI_API_KEY" in os.environ
), "Please set the OPENAI_API_KEY environment variable."

with tempfile.TemporaryDirectory() as temp_dir:
    persist_dir = os.path.join(temp_dir, "faiss_index")

    # Create the vector database and persist it to a local filesystem folder
    loader = TextLoader("tests/langchain/state_of_the_union.txt")
    documents = loader.load()
    text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
    docs = text_splitter.split_documents(documents)
    embeddings = OpenAIEmbeddings()
    db = FAISS.from_documents(docs, embeddings)
    db.save_local(persist_dir)

    # Define a loader function to recall the retriever from the persisted vectorstore
    def load_retriever(persist_directory):
        embeddings = OpenAIEmbeddings()
        vectorstore = FAISS.load_local(persist_directory, embeddings)
        return vectorstore.as_retriever()

    # Log the retriever with the loader function
    with mlflow.start_run() as run:
        logged_model = mlflow.langchain.log_model(
            db.as_retriever(),
            name="retriever",
            loader_fn=load_retriever,
            persist_dir=persist_dir,
        )

# Load the retriever chain
loaded_model = mlflow.pyfunc.load_model(logged_model.model_uri)
print(
    loaded_model.predict(
        [{"query": "What did the president say about Ketanji Brown Jackson"}]
    )
)

上面示例的输出如下所示

[
    [
        {
            "page_content": "Tonight. I call...",
            "metadata": {"source": "/state.txt"},
        },
        {
            "page_content": "A former top...",
            "metadata": {"source": "/state.txt"},
        },
    ]
]

MLflow Langchain 自动日志记录

请参阅 MLflow Langchain 自动日志记录文档，以获取有关如何为 Langchain 模型启用自动日志记录的更多详细信息。

概述​

LangChain 的技术本质​

使用 LangChain 构建链​

与 MLflow 集成​

基本示例：在 MLflow 中记录 LangChain LLMChain​

简单 LLMChain 示例展示的内容​

使用 MLflow 记录 LangChain 代理​

什么是代理？​

代理的关键组成部分​

代理​

工具​

工具包​

AgentExecutor​

其他代理运行时​

记录 LangChain 代理的示例​

简单代理示例展示的内容​

使用 LangChain 和 GenAI LLM 进行实时流式输出​

流式输出功能概述​

支持的流式模型​

使用 predict_stream 进行流式输出​

示例用法​

与回调的高级集成​

使用 MLflow 增强 RetrievalQA 链的管理​

RetrievalQA 链的关键见解​

RAG 流程的详细概述​

使用 MLflow 阐明向量数据库管理​

关键 MLflow 组件和 VectorDB 日志记录​

重要注意事项​

记录 LangChain RetrievalQA 链的示例​

在 MLflow 中记录和评估 LangChain 检索器​

记录各个检索器的目的​

在 MLflow 中记录检索器的要求​

记录 LangChain 检索器的示例​

MLflow Langchain 自动日志记录​

概述

LangChain 的技术本质

使用 LangChain 构建链

与 MLflow 集成

基本示例：在 MLflow 中记录 LangChain `LLMChain`

简单 LLMChain 示例展示的内容

使用 MLflow 记录 LangChain 代理

什么是代理？

代理的关键组成部分

代理

工具

工具包

AgentExecutor

其他代理运行时

记录 LangChain 代理的示例

简单代理示例展示的内容

使用 LangChain 和 GenAI LLM 进行实时流式输出

流式输出功能概述

支持的流式模型

使用 `predict_stream` 进行流式输出

示例用法

与回调的高级集成

使用 MLflow 增强 RetrievalQA 链的管理

RetrievalQA 链的关键见解

RAG 流程的详细概述

使用 MLflow 阐明向量数据库管理

关键 MLflow 组件和 VectorDB 日志记录

重要注意事项

记录 LangChain RetrievalQA 链的示例

在 MLflow 中记录和评估 LangChain 检索器

记录各个检索器的目的

在 MLflow 中记录检索器的要求

记录 LangChain 检索器的示例

MLflow Langchain 自动日志记录