如何使用ZenML与LangChain构建生产级LLM应用：完整指南

ZenML是一个强大的MLOps工具，能够帮助开发者构建可移植、生产就绪的机器学习管道。当与LangChain集成时，它为构建企业级大型语言模型(LLM)应用提供了坚实的基础。本文将详细介绍如何利用ZenML与LangChain的强大组合，创建可靠、可扩展的LLM应用程序。## ZenML与LangChain集成的核心优势ZenML与LangChain的集成将MLOps最佳实践与LLM应用

薛锨宾

930人浏览 · 2026-03-30 16:37:23

薛锨宾 · 2026-03-30 16:37:23 发布

如何使用ZenML与LangChain构建生产级LLM应用：完整指南

【免费下载链接】zenml ZenML 🙏: Build portable, production-ready MLOps pipelines. https://zenml.io. 项目地址: https://gitcode.com/gh_mirrors/ze/zenml

ZenML是一个强大的MLOps工具，能够帮助开发者构建可移植、生产就绪的机器学习管道。当与LangChain集成时，它为构建企业级大型语言模型(LLM)应用提供了坚实的基础。本文将详细介绍如何利用ZenML与LangChain的强大组合，创建可靠、可扩展的LLM应用程序。

ZenML与LangChain集成的核心优势

ZenML与LangChain的集成将MLOps最佳实践与LLM应用开发完美结合，带来以下关键优势：

完整的管道管理：ZenML提供端到端的管道编排，从数据加载、模型训练到部署
可重现性：确保每次LLM应用的执行都有一致的结果和完整的追踪
生产级部署：简化将LangChain应用部署到各种环境的过程
丰富的集成生态：结合ZenML的各种堆栈组件和LangChain的语言模型能力

ZenML系统架构展示了其灵活的组件化设计，非常适合构建复杂的LLM应用管道

环境准备与安装

开始使用ZenML与LangChain构建LLM应用前，需要完成以下准备工作：

安装ZenML：

pip install zenml[server]
zenml up

安装LangChain集成：

zenml integration install langchain -y

ZenML的LangChain集成包含了必要的依赖项，如langchain==0.3.0、langchain-community等，确保了与LangChain生态系统的兼容性。

构建第一个ZenML+LangChain LLM应用

下面我们将构建一个简单但功能完整的LLM应用，该应用能够从网页加载内容并生成摘要。

创建基本管道组件

首先，创建一个使用LangChain加载网页内容并生成摘要的步骤：

from langchain_community.document_loaders import WebBaseLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableLambda
from langchain_openai import ChatOpenAI
from zenml import step

@step
def load_and_summarize_web_content(url: str) -> str:
    """加载网页内容并生成摘要"""
    # 加载网页内容
    loader = WebBaseLoader(url)
    docs = loader.load()
    text = "\n".join([doc.page_content for doc in docs])
    
    # 创建摘要链
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    prompt = ChatPromptTemplate.from_template(
        "Write a concise summary of the following text:\n\n{text}"
    )
    
    chain = prompt | llm | StrOutputParser()
    summary = chain.invoke({"text": text})
    
    return summary

定义ZenML管道

接下来，定义一个完整的ZenML管道来组织这个步骤：

from zenml import pipeline

@pipeline(
    name="llm_web_summarizer",
    enable_cache=True
)
def llm_web_summarizer_pipeline(url: str):
    """LLM网页摘要生成管道"""
    summary = load_and_summarize_web_content(url=url)

运行管道

最后，运行这个管道并查看结果：

if __name__ == "__main__":
    run = llm_web_summarizer_pipeline(url="https://zenml.io/blog/zenml-0-94-0-release")
    print(f"网页摘要: {run.outputs.summary.output}")

高级功能与最佳实践

实验跟踪与版本控制

ZenML提供了强大的实验跟踪能力，可以轻松比较不同LLM模型和参数的性能：

from zenml.integrations.mlflow.experiment_trackers import MLFlowExperimentTracker

# 配置MLflow实验跟踪器
experiment_tracker = MLFlowExperimentTracker(
    name="mlflow_tracker",
    tracking_uri="http://mlflow-server:5000"
)

# 在管道中使用实验跟踪器
@pipeline(
    name="llm_web_summarizer",
    experiment_tracker=experiment_tracker.name,
    enable_cache=True
)
def llm_web_summarizer_pipeline(url: str):
    summary = load_and_summarize_web_content(url=url)

远程执行与部署

ZenML的强大之处在于其能够将管道部署到各种环境。下面是一个使用Kubernetes进行远程执行的示例：

# 创建云堆栈
zenml orchestrator register kubernetes_orchestrator --flavor=kubernetes
zenml stack register cloud_stack -o kubernetes_orchestrator ...
zenml stack set cloud_stack

# 运行管道
python run_pipeline.py

ZenML远程执行流程展示了如何在云环境中运行包含LangChain组件的管道

处理敏感信息

在LLM应用中，安全处理API密钥等敏感信息至关重要。ZenML的秘密管理功能可以帮助安全存储和使用这些信息：

from zenml.client import Client

# 存储OpenAI API密钥
client = Client()
client.create_secret(
    name="openai_secret",
    secrets={"OPENAI_API_KEY": "your-api-key-here"}
)

# 在步骤中使用密钥
@step(secrets=["openai_secret"])
def load_and_summarize_web_content(url: str) -> str:
    # 密钥会自动注入环境变量
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    # ...

实际应用示例：文档分析代理

ZenML与LangChain的集成非常适合构建复杂的LLM应用，如文档分析代理。以下是一个更完整的示例，展示如何构建一个能够分析多个文档并回答问题的系统：

# 文档加载步骤
@step
def load_documents(document_paths: List[str]) -> List[str]:
    """加载多个文档"""
    documents = []
    for path in document_paths:
        if path.startswith("http"):
            loader = WebBaseLoader(path)
        else:
            loader = TextLoader(path)
        documents.extend(loader.load())
    return [doc.page_content for doc in documents]

# 文档分块步骤
@step
def split_documents(documents: List[str]) -> List[str]:
    """将文档分割成小块"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000, chunk_overlap=200
    )
    return text_splitter.split_text("\n\n".join(documents))

# 向量存储创建步骤
@step
def create_vector_store(texts: List[str]) -> None:
    """创建向量存储"""
    embeddings = OpenAIEmbeddings()
    db = Chroma.from_texts(texts, embeddings, persist_directory="./chroma_db")
    db.persist()

# 问答步骤
@step
def answer_question(question: str) -> str:
    """回答用户问题"""
    embeddings = OpenAIEmbeddings()
    db = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
    
    retriever = db.as_retriever()
    qa_chain = RetrievalQA.from_chain_type(
        llm=ChatOpenAI(),
        chain_type="stuff",
        retriever=retriever
    )
    
    return qa_chain.run(question)

# 完整管道
@pipeline
def document_analysis_pipeline(document_paths: List[str], question: str):
    docs = load_documents(document_paths=document_paths)
    chunks = split_documents(documents=docs)
    create_vector_store(texts=chunks)
    answer = answer_question(question=question)