从零基础到LLM运维高手：生产环境中运行LLM的入门指南

如果您曾经尝试将 Jupyter Notebook 中的 LLM 原型部署到实际环境中，那么您一定知道这并非点击“运行”按钮那么简单。我还记得第一次尝试部署 LLM 端点时的情景：简直一团糟。模型在本地运行良好，但一旦用户开始发送大量请求……一切都崩溃了。

那时我才意识到，我缺少的不是机器学习知识，而是 LLMOps。

本指南旨在帮助您从零基础成为 LLMOps 高手，完整讲解如何使用 LangChain、FastAPI 和 Docker 在生产环境中运行 LLM，并提供 AWS 部署的概念概述。

我们将构建一个小型但实用的 RAG 聊天机器人，作为理解 LLMOps 生命周期的完美切入点。

LLMOps究竟是什么？

简而言之，LLMOps 是 MLOps 的扩展，用于大型语言模型。

它涵盖了模型训练或选择之后的所有环节，包括推理优化、监控、提示编排、数据管道、部署、扩展和治理。

您可以将其视为模型与现实世界之间的桥梁，确保可靠性、可扩展性、可观测性和合规性。

以下是我们将要介绍的 LLMOps 流水线概览：

1. 数据摄取与向量化
2. 理解数据集
3. 构建向量存储
4. 构建并运行 RAG 聊天机器人
5. 使用 LangChain 进行提示编排
6. 通过 FastAPI 提供服务
7. 使用 Streamlit 构建聊天 UI
8. 通过 Ngrok 实现公共访问
9. 使用 Docker 进行容器化
10. AWS 部署概念
11. 监控、版本控制、评估和治理

基本设置

在开始实现之前，让我们先进行基本设置，安装必要的库。本项目使用 Google Colab 作为编程环境。请复制以下代码以设置库：

!mkdir llmops-demo 
!cd llmops-demo 
!python3 -m venv venv 
!source venv/bin/activate 
!pip install langchain openai faiss-cpu fastapi uvicorn python-dotenv langchain-community 
!pip install nest_asyncio pyngrok streamlit

本项目还将使用 OpenAI 和 NGROK。您可以从以下链接获取它们的 API 访问令牌/密钥：

• OpenAI: https://platform.openai.com/api-keys
• NGROK: https://dashboard.ngrok.com/get-started/your-authtoken

请注意，本博客使用免费版本的 API/身份验证令牌即可。请在 Colab 中编写以下代码来设置这些令牌：

import os 
from getpass import getpass 
os.environ['OPENAI_API_KEY'] = getpass("Enter OPENAI KEY:") 
os.environ['NGROK_AUTHTOKEN'] = getpass("ENTER NGROK TOKEN:")

数据摄取和向量化

了解数据集

在构建任何模型之前，让我们先从数据入手，因为数据才是 RAG 系统的核心。我使用了一个名为 Sample RAG Knowledge Item Dataset 的小型 Kaggle 数据集。它简洁明了，非常适合学习。每一行都像一条迷你版的“IT 服务台”笔记，是特定主题下的简短知识点。

你会看到两列：

• ki_text → 实际内容（例如“VPN 连接故障排除步骤”）
• ki_topic → 主题标签，例如“网络”、“硬件”或“安全”

这个数据集特意设计得比较小，我很喜欢这一点，因为它可以让你快速测试不同的 RAG 概念，例如分块大小、嵌入模型和检索策略，而无需等待数小时的索引。它提取自一个更大的 IT 知识库数据集（约 100 篇文章），但这个精简版非常适合实验，因为它速度快、重点突出，而且非常适合演示。

现在我们了解了数据的结构，接下来让我们开始教模型如何“记住”它：将文本转换为嵌入向量，并将其存储在向量数据库中。

构建向量存储

数据集准备就绪后，下一个目标是让模型理解它，而不仅仅是读取它。为此，我们需要将文本转换为嵌入向量，即表示含义的数值向量。

以下是实现此操作的代码：

import os
import pandas as pd
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.docstore.document import Document
# Load dataset
df = pd.read_csv("/content/rag_sample_qas_from_kis.csv")
# Use ki_text as main text and ki_topic as metadata
docs = [
Document(page_content=row["ki_text"], metadata={"topic": row["ki_topic"]})
for _, row in df.iterrows()
]
# Split into chunks for embedding
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
chunks = splitter.split_documents(docs)
# Embed and store in FAISS
embeddings = OpenAIEmbeddings(openai_api_key=os.getenv("OPENAI_API_KEY"))
db = FAISS.from_documents(chunks, embeddings)
db.save_local("vectorstore")
print("Vectorstore created successfully using ki_text as content!")

哇，代码好多啊，我们来看看这里面到底发生了什么！

设置好数据集之后，下一步是帮助模型“理解”它，方法是将文本转换为词嵌入。我们首先加载 CSV 文件，并将每条记录封装成 LangChain Document 对象，将“ki_text”存储为内容，“ki_topic”存储为元数据。这些元数据之后可以帮助进行过滤或特定主题的检索。

接下来，我们将每个文档分割成更小的重叠块（800 个字符，100 个字符重叠），这样分割过程中就不会丢失任何信息。然后，我们使用 OpenAIEmbeddings 将每个块转换为向量：语义含义的密集数值表示。

最后，所有词嵌入都存储在 FAISS 向量存储中，这是一个高效的相似性搜索索引，可以在查询期间快速检索最相关的块。

完成此步骤后，向量存储/文件夹将作为模型的本地“内存”，准备在下一阶段进行查询。

构建和运行RAG聊天机器人

以下是部署后的最终聊天机器人：

现在，让我们介绍实现此功能所需的所有库/工具，并了解它们如何协同工作。

使用LangChain进行提示编排

在这里，我们使用 LangChain 的 RetrievalQA 链将向量存储、检索器和 LLM 连接起来。

之前创建的 FAISS 向量存储将被重新加载到内存中，并连接到 OpenAI 嵌入。然后，检索器充当智能查找引擎，仅从数据集中获取最相关的数据块。

现在，每个查询都将流经此管道：检索 → 增强 → 生成。

通过FastAPI提供服务

为了模拟生产环境，我们添加了一个轻量级的 FastAPI 后端。虽然在 Colab 中此步骤是可选的，但它模拟了您的 RAG 模型在真实环境中的暴露方式，并提供可用于外部集成或 API 请求的端点。

目前，“/”路由仅返回健康消息，但此结构可以轻松扩展，以处理来自您的 UI、Slack 机器人或 Web 客户端的查询。

使用Streamlit构建聊天UI

在后端之上，我们使用 Streamlit 构建了一个交互式聊天界面。这提供了一个简洁的、基于浏览器的体验，用于与您的 RAG 管道进行交互。用户可以输入问题，点击“发送”，然后看到由 LangChain 的检索和推理链支持的、具有上下文感知和文档感知的回复。

每次交流都存储在 st.session_state 中，从而在您和模型之间创建持久的对话流。

通过Ngrok进行公开访问

由于 Colab 不支持直接 Web 托管，我们使用 ngrok 安全地公开 Streamlit 应用。Ngrok 会将您的本地端口隧道连接到一个临时的公共 URL，允许任何人（或仅限您）在真实的浏览器标签页中访问聊天机器人 UI。

一旦 ngrok 生成公共链接，聊天机器人即可立即访问。

整合所有组件

• LangChain 负责编排嵌入、检索和生成
• FastAPI 提供可选的后端层
• Streamlit 作为交互式 UI
• Ngrok 将 Colab 与外部世界连接起来

这样，您就可以在笔记本电脑上实时运行一个功能齐全、端到端的 RAG 聊天机器人。

以下是结合了上述所有工具的 RAG 聊天机器人的完整代码：

from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.llms import OpenAI
from pyngrok import ngrok
import nest_asyncio
import threading
import streamlit as st
import uvicorn
from fastapi import FastAPI
# --- Create FastAPI backend (optional) ---
app = FastAPI(title="RAG Chatbot – Backend")
embeddings = OpenAIEmbeddings(openai_api_key=os.getenv("OPENAI_API_KEY"))
db = FAISS.load_local("vectorstore", embeddings, allow_dangerous_deserialization=True)
retriever = db.as_retriever()
qa_chain = RetrievalQA.from_chain_type(
llm=OpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"), temperature=0),
chain_type="stuff",
retriever=retriever,
)
@app.get("/")
def home():
return {"message": "Backend ready"}
# --- Streamlit Chat UI ---
def run_streamlit():
st.set_page_config(page_title="💬 RAG Chatbot", layout="centered")
st.title("💬 Chat with your RAG-Powered LLM")
if "history" not in st.session_state:
st.session_state.history = []
query = st.text_input("Ask me something:")
if st.button("Send") and query:
with st.spinner("Thinking..."):
answer = qa_chain.run(query)
st.session_state.history.append((query, answer))
for q, a in reversed(st.session_state.history):
st.markdown(f"**You:** {q}")
st.markdown(f"**Bot:** {a}")
st.markdown("---")
# --- Launch ngrok and Streamlit ---
ngrok.kill()
public_url = ngrok.connect(8501)
print(f"Chat UI available at: {public_url}")
# Run Streamlit app in background
nest_asyncio.apply()
def start_streamlit():
!streamlit run app.py &>/dev/null&
# Save UI to file
with open("app.py", "w") as f:
f.write('''
import streamlit as st
from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.llms import OpenAI
import os
os.environ["OPENAI_API_KEY"] = "''' + os.environ["OPENAI_API_KEY"] + '''"
st.set_page_config(page_title="💬 RAG Chatbot", layout="centered")
st.title("💬 Chat with your RAG-Powered LLM")
embeddings = OpenAIEmbeddings(openai_api_key=os.getenv("OPENAI_API_KEY"))
db = FAISS.load_local("vectorstore", embeddings, allow_dangerous_deserialization=True)
retriever = db.as_retriever()
qa_chain = RetrievalQA.from_chain_type(
llm=OpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"), temperature=0),
chain_type="stuff",
retriever=retriever,
)
if "history" not in st.session_state:
st.session_state.history = []
query = st.text_input("Ask me something:")
if st.button("Send") and query:
with st.spinner("Thinking..."):
answer = qa_chain.run(query)
st.session_state.history.append((query, answer))
for q, a in reversed(st.session_state.history):
st.markdown(f"**You:** {q}")
st.markdown(f"**Bot:** {a}")
st.markdown("---")
''')
# Start Streamlit
get_ipython().system_raw('streamlit run app.py --server.port 8501 &')
print("Chatbot is running. Open the ngrok URL above")

运行上述代码后，您将看到如下输出：

聊天界面位于：NgrokTunnel: “https://22f6c6d1ef68.ngrok-free.app” -> “http://localhost:8501” 。聊天机器人正在运行。打开上面的 ngrok URL。

这是您的聊天机器人部署的 URL：https://22f6c6d1ef68.ngrok-free.app/

此链接可从世界任何地方通过互联网访问。您可以与朋友分享，或亲自使用它来测试您的机器人。是不是很棒？

通过使用 Streamlit、FastAPI、NGrok 和 LangChain 等工具，我们仅用几行代码就成功部署了一个基于 RAG 的端到端聊天机器人。想象一下它为您带来的无限可能；您的成就将无可限量。

使用Docker进行容器化

至此，我们已经拥有一个功能齐全的 RAG 聊天机器人：后端、用户界面等等。但如果您曾经尝试过跨环境部署该配置，您就会明白其中的痛苦：缺少依赖项、版本不匹配、路径损坏等等，都是常见的混乱情况。

而 Docker 正是解决这一问题的利器。Docker 提供了一种简洁、可移植的方式，将整个环境以及所有工具（例如 FastAPI、Streamlit、LangChain、FAISS，甚至您的向量存储）打包成一个统一的单元，可以在任何地方运行。

如果您想部署与上述完全相同的 RAG 聊天机器人，但希望通过 Docker 来实现，那么以下是更新后的代码：

import os
import subprocess
from pyngrok import ngrok
# --- Create Streamlit UI ---
streamlit_code = """
import streamlit as st
import requests
st.set_page_config(page_title="RAG Chatbot", page_icon="", layout="centered")
st.title("RAG Chatbot – LLMOps Demo")
st.write("Ask questions from your knowledge base. Backend powered by FastAPI, deployed in Docker.")
query = st.text_input("💬 Your question:")
if query:
with st.spinner("Thinking..."):
try:
response = requests.get(f"http://localhost:8000/ask", params={"query": query})
if response.status_code == 200:
st.success(response.json()["answer"])
else:
st.error("Something went wrong with the backend.")
except Exception as e:
st.error(f"Error: {e}")
"""
with open("streamlit_app.py", "w") as f:
f.write(streamlit_code.strip())
# --- Create requirements.txt ---
with open("requirements.txt", "w") as f:
f.write("fastapi\nuvicorn\nstreamlit\nlangchain\nopenai\nfaiss-cpu\npyngrok\nrequests\n")
# --- Create Dockerfile ---
dockerfile = f"""
FROM python:3.10-slim
WORKDIR /app
COPY . .
RUN pip install --no-cache-dir -r requirements.txt
EXPOSE 8000 8501
ENV OPENAI_API_KEY=${{OPENAI_API_KEY}}
CMD bash -c "uvicorn from_zero_to_llmops_hero_your_101_guide_to_running_llms_in_production:app --host 0.0.0.0 --port 8000 & streamlit run streamlit_app.py --server.port 8501 --server.address 0.0.0.0"
"""
with open("Dockerfile", "w") as f:
f.write(dockerfile.strip())
# --- Install Docker if needed ---
try:
subprocess.run(["docker", "--version"], check=True)
except Exception:
print("Installing Docker...")
subprocess.run(["apt-get", "update", "-qq"], check=True)
subprocess.run(["apt-get", "install", "-qq", "-y", "docker.io"], check=True)
# --- Check if Docker daemon available ---
try:
subprocess.run(["docker", "build", "--version"], check=True)
docker_available = True
except Exception:
docker_available = False
if docker_available:
print("Building Docker image...")
subprocess.run(["docker", "build", "-t", "rag-chatbot-ui", "."], check=True)
print("Running container (FastAPI + Streamlit)...")
subprocess.run(["docker", "run", "-d", "-p", "8000:8000", "-p", "8501:8501",
"-e", f"OPENAI_API_KEY={os.getenv('OPENAI_API_KEY')}",
"rag-chatbot-ui"], check=True)
else:
print("Docker not supported in Colab — running natively instead.")
print("Starting FastAPI + Streamlit locally...")
# Run both apps directly
import threading
def run_fastapi():
os.system("uvicorn from_zero_to_llmops_hero_your_101_guide_to_running_llms_in_production:app --host 0.0.0.0 --port 8000")
def run_streamlit():
os.system("streamlit run streamlit_app.py --server.port 8501 --server.address 0.0.0.0")
threading.Thread(target=run_fastapi).start()
threading.Thread(target=run_streamlit).start()
# --- Expose via ngrok ---
ngrok.kill()
public_url = ngrok.connect(8501)
print(f"Your RAG Chatbot UI is live at: {public_url}")
print("API (FastAPI docs) at: <ngrok_url_for_8000>/docs")

代码的实际变化

仔细查看新的代码块，你会发现以下几个主要变化：

1. 创建Dockerfile

这就像应用程序的配方。我们从一个轻量级的 python:3.10-slim 镜像开始，复制所有项目文件，根据 requirements.txt 安装依赖项，并暴露两个端口：

• 8000 用于 FastAPI 后端，
• 8501 用于 Streamlit UI。

最后，使用一条命令即可在同一个容器中启动这两个服务器：

CMD bash -c "uvicorn ... & streamlit run ..."

只需一行代码，Docker 就能同时运行 FastAPI 和 Streamlit，一个提供 API 服务，另一个提供界面服务。

2. 自动化需求处理

在构建 Docker 镜像之前，我们会以编程方式生成一个干净的 requirements.txt 文件，其中包含我们使用的所有库：fastapi、uvicorn、streamlit、langchain、openai、faiss-cpu、pyngrok 和 requests。

这确保了 Docker 镜像始终安装笔记本中使用的确切软件包，无需手动复制粘贴。

3. 构建和运行容器

Docker 准备就绪后，我们运行以下命令：

subprocess.run(["docker", "build", "-t", "rag-chatbot-ui", "."]) subprocess.run(["docker", "run", "-d", "-p", "8000:8000", "-p", "8501:8501", ...])

这会构建一个名为 rag-chatbot-ui 的镜像，并启动一个运行后端和前端的容器。如果您在本地计算机上，这将立即为您提供两个可用的端点：

• http://localhost:8000/docs → FastAPI
• http://localhost:8501 → Streamlit UI

4. Colab的备用方案

由于 Google Colab 不直接支持 Docker 守护进程，我们采取了优雅的处理方式。如果 Docker 不可用，脚本会自动在两个后台线程中启动 FastAPI 和 Streamlit，从而确保用户体验与本地环境一致。

为什么这很重要？

从 notebook 代码到容器化设置的这种小小的转变，是真正意义上从“玩转 LLM”到“像运营产品一样运营 LLM”的转折点。

现在您拥有：

• 可复现性 → 随时随地运行相同的环境
• 可移植性 → 可部署在 AWS、GCP 甚至您的笔记本电脑上
• 可扩展性 → 稍后可在负载均衡器后部署多个容器

没错，它就是同一个应用程序，只是被巧妙地封装在了 Docker 容器中。

AWS部署概念

在 AWS 上部署 LLM 系统可以实现生产级的可扩展性，但这本身就是一个值得深入探讨的话题。从在 EC2 或 ECS 上托管 FastAPI 应用程序，到在 S3 上存储向量数据库，再到使用 AWS Lambda 进行事件驱动触发，各种可能性层出不穷。您甚至可以集成 CloudFront + API Gateway，以实现安全、全球分布式的推理端点。

目前，您只需知道 AWS 为您提供了从本地 Docker 容器迁移到完全托管、自动扩展设置所需的所有构建模块。我们有机会将会探讨整个工作流程，包括基础设施即代码、CI/CD 流水线和部署脚本。

监控、版本控制、评估和治理

系统部署完成后，真正的工作才刚刚开始：监控和维护。跟踪延迟、检索准确率、故障率和版本漂移对于确保聊天机器人的可靠性和可解释性至关重要。再加上版本控制、评估和治理，就构成了生产级 LLMOps 的基础。

这些领域值得单独探讨。在下一篇文章中，我们将深入研究日志记录、评估仪表板和模型治理，所有这些都直接集成到您的 RAG 流水线中。

小结

我们从零开始，搭建了一个简单的聊天机器人用户界面，并逐步构建了一个完整的 LLMOps 工作流程：

• 数据摄取、创建向量存储、嵌入、提示编排、通过 FastAPI 提供服务、使用 Docker 容器化，以及准备在 AWS 上部署。
• 每一层都增加了结构、可扩展性和可靠性，最终将实验转化为可部署、可维护的产品。
• 但这仅仅是开始。真正的 LLMOps 始于部署之后，那时您需要监控行为、优化检索、对嵌入进行版本控制，并确保模型的安全性和合规性。

在本系列的下一部分中，我们将超越设置，深入探讨 AWS 部署、可观测性、版本控制和治理——这些才是真正让您的 LLM 达到生产就绪状态的关键所在。

请在下方留言分享您的想法/问题！