2026 年掌握 LLMOps 的路线图

2026年掌握LLMOps的路线图

URL 来源：https://machinelearningmastery.com/the-roadmap-for-mastering-llmops-in-2026/

发布时间：2026-06-01T12:00:18+00:00

在本文中，你将学习如何通过一个结构化的六步 LLMOps 路线图来构建生产级的大语言模型（LLM）系统，该路线图涵盖可观测性、评估、成本控制和代理编排。

我们将涵盖的主题包括：

• LLMOps 与传统 MLOps 的区别，以及在使用任何 LLMOps 工具之前你需要掌握的基础技能。
• 如何对 LLM 调用进行完整追踪，使用 RAGAS 构建和评估 RAG 管道，并通过模型路由实现成本控制。
• 一个循序渐进的学习计划，带你从第一个 LLM API 项目开始，逐步过渡到部署和评估生产级代理系统。

需要覆盖的内容很多，让我们开始吧。

引言

据预测，LLMOps 市场将从 2024 年的 19.7 亿美元 增长至 2028 年的 49 亿美元，年复合增长率（CAGR）达 42%。与此同时，2026 年有 72% 的企业正在采用 AI 自动化工具，但大多数企业尚未在其 LLM 基础设施中建立成本控制机制。这两个数据共同揭示了真正的机遇：需求巨大，而绝大多数构建这些系统的人却缺乏使系统可靠、可审计且成本高效的运营纪律。

LLMOps 正是填补这一空白的工程实践。它不是一个单一工具或一次性配置，而是一种构建基于 LLM 的系统的学科——让这些系统像生产级软件一样运行：具备版本控制、监控、评估，并能随时间持续改进。本路线图是一条从基础到生产级系统的分阶段路径，包含关键工具、按顺序构建的技能、两个完整的可运行代码示例，以及你可以从今天就开始遵循的详细步骤计划。

LLMOps 与 MLOps 的对比

传统的 MLOps 围绕一个明确的对象展开：模型。你训练它、版本化它、部署它、监控其预测是否存在漂移，并在性能下降时重新训练。

而在 LLMOps 中，模型通常是变化频率最低的组件。你不会频繁地版本化模型权重，而是频繁地版本化提示词（prompt）。上周还能正常工作的提示词，在模型提供商悄悄更新其基础模型后，可能产生更差的输出。在测试中看起来更简洁的系统提示词重写，在生产环境中可能在边缘案例上导致性能下降。每一次提示词变更都是一次部署，每一次部署都需要被追踪、测试并可回滚。

第二个主要区别在于 LLM 的输出具有非确定性。相同的输入在不同调用中可能产生不同的输出，这意味着传统的监控方式——“模型是否返回了正确的类别标签？”——不再适用。你需要的是能够以连续评分方式衡量质量的评估基础设施，而不是简单的二元正确性判断。这要求你构建黄金测试集、运行评估管道，并利用“LLM 作为评判者”（LLM-as-judge）在无需人工逐条审核的情况下大规模评分输出。

通常情况下，令牌优化实践可以节省 30–50% 的 API 成本，往往足以覆盖整个工具预算。当每日用户数为 1,000 时看似可控的推理成本，在用户增长到 100,000 时就会演变为预算危机。在 LLMOps 中，成本是一个首要指标，这一点在传统 MLOps 中从未如此突出。如果将其视为事后考虑，工程团队最终将不得不向财务部门解释意外的账单。

在开始 LLMOps 之前你需要什么

在没有以下基础的前提下，请不要直接开始使用 LLMOps 工具。试图在一个你还不了解如何构建的系统上进行仪器化，只会浪费时间和精力。

1. Python 熟练度：核心软件工程技能仍然至关重要。Python 的熟练使用、对分布式系统的理解、熟悉云平台以及强大的调试能力，构成了其他一切的基础。你需要掌握的 Python 技能包括：使用 async/await 实现非阻塞 API 调用、错误处理与重试逻辑、处理 JSON 和结构化数据、将代码打包为可安装模块、编写测试等。不需要精通高级 Python，但必须具备足够的能力来构建和维护一个他人依赖的服务。

1. 大语言模型（LLM）基础：在能够良好地运行 LLM 系统之前，你必须了解它们为什么会失败。这意味着要理解 token 和上下文窗口（为什么长输入成本更高且表现不同）、温度和采样机制（为什么输出会变化以及如何控制）、基础模型与指令微调模型的区别、API 层面的工具调用是什么样子，以及“幻觉”在机制上究竟是什么——而不仅仅是作为一个术语。在接触任何 LLMOps 工具之前，先完成三到五个小型项目：一个摘要器、一个文档分类器、一个简单的 RAG 流水线。通过实际操作体验失败模式，才能让后续的运维工作变得有意义。

1. 云与基础设施基础：你将部署服务，而不仅仅是运行脚本。至少熟悉一种云服务商 —— AWS、GCP 或 Azure —— 并掌握 Docker 容器化技术，以及基本的 CI/CD 概念。你不必成为 DevOps 工程师，但需要理解容器是什么、环境变量如何工作，以及如何运行一个不会因关闭笔记本电脑而崩溃的服务。

1. 版本控制规范：提示词需要存入 Git。配置文件需要存入 Git。评估数据集需要存入 Git。所有会发生变化的内容都应有历史记录。这一习惯是整个运维层的基础——如果某项内容未被版本化，你就无法调试它、回滚它，也无法在性能下降时弄清楚发生了什么改变。

一张清晰的向上“学习栈”示意图，包含从下到上堆叠的四个标注层级（点击放大）

图片作者提供

第一阶段：构建你的第一个生产级 LLM 系统

本阶段的目标不是构建令人印象深刻的东西，而是构建真正可用的东西。在你机器上运行良好的演示程序并不是生产系统。生产系统具备日志记录、错误处理、成本可见性，并且当凌晨两点系统出问题时，有人可以进行调试。

构建什么

构建一个聊天机器人、一个文档问答工具，或一个接收用户查询并返回 LLM 响应的 API 接口。具体应用类型并不重要，关键是你对自己施加的操作要求：每次调用都必须记录日志，每个响应都必须可追溯，而且在进入下一阶段前，你必须清楚每次请求消耗了多少 token 和美元成本。

本阶段需掌握的技能

1. 提示词版本控制：将每个提示词视为生产代码。将其保存在文件中，提交到 Git 时附带描述性信息，不要直接在 API 调用中修改。当出现问题时，你需要知道是什么发生了变化。
2. 结构化输出：使用 JSON 模式或函数调用来获取格式一致、应用程序可可靠解析的响应。非结构化文本输出适用于聊天界面。对于任何需要代码处理的内容，结构化输出是必不可少的。
3. 基础可观测性：记录每一次 LLM 调用：输入内容、输出内容、使用的模型、token 数量、延迟时间以及计算出的成本。这些数据使你能够调试、评估和优化系统。

安装前置依赖：

pip install langfuse anthropic python-dotenv

你还需准备：

• 一个免费的 Langfuse 账户（或自托管实例）——从项目设置中获取你的 LANGFUSE_PUBLIC_KEY 和 LANGFUSE_SECRET_KEY。
• 一个 Anthropic API 密钥 或任意 LLM 提供商的密钥。
• 在项目根目录下创建一个 .env 文件，存放上述密钥。

代码：使用 Langfuse 追踪的仪器化 LLM 调用

# llm_with_tracing.py

# 目的：一个具备完整可观测性的生产级 LLM 调用封装。

# 每次调用都会在 Langfuse 中追踪：输入、输出、token 数、成本、延迟。

#

# 前置条件：

# pip install langfuse anthropic python-dotenv

#

# 配置步骤：

# 1. 在 https://cloud.langfuse.com 注册免费账户

# 2. 从 设置 > API 密钥 获取你的密钥

import os
import json
from dotenv import load_dotenv
from anthropic import Anthropic
from langfuse import Langfuse
from langfuse.model import CreateTraceParams, CreateGenerationParams

# 加载环境变量
load_dotenv()

# 初始化客户端
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
langfuse = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY")
)

def call_llm_with_tracing(prompt: str, model: str = "claude-3-haiku-20240307") -> dict:
    """
    使用 Langfuse 追踪的 LLM 调用包装器。
    
    Args:
        prompt: 输入提示
        model: 使用的模型名称
    
    Returns:
        包含响应和元数据的字典
    """
    # 创建追踪
    trace = langfuse.trace(
        name="llm-call",
        input={"prompt": prompt},
        metadata={"model": model}
    )
    
    try:
        # 执行 LLM 调用
        response = client.messages.create(
            model=model,
            max_tokens=1024,
            messages=[
                {"role": "user", "content": prompt}
            ],
            temperature=0.7
        )
        
        # 计算成本（示例：基于 token 数估算）
        input_tokens = response.usage.input_tokens
        output_tokens = response.usage.output_tokens
        total_tokens = input_tokens + output_tokens
        
        # 示例成本计算（假设每 1000 tokens $0.001）
        cost = (total_tokens / 1000) * 0.001
        
        # 创建生成事件
        generation = langfuse.generation(
            name="llm-response",
            model=model,
            prompt=prompt,
            response=response.content[0].text,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost=cost,
            trace_id=trace.id
        )
        
        # 返回结果
        return {
            "response": response.content[0].text,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": total_tokens,
            "cost": cost,
            "latency": response.latency,
            "model": model
        }
        
    except Exception as e:
        # 记录异常
        trace.error(e)
        raise
    finally:
        # 结束追踪
        trace.end()

# 示例使用
if __name__ == "__main__":
    prompt = "请总结一下人工智能的发展历程。"
    result = call_llm_with_tracing(prompt)
    print(json.dumps(result, indent=2, ensure_ascii=False))

此代码实现了对 LLM 调用的全面追踪，确保每次调用都被记录，便于后续分析与优化。

3. 创建一个包含以下变量的 .env 文件

#

运行：

python llm_with_tracing.py

import os

import time

from dotenv import load_dotenv

import anthropic

from langfuse import Langfuse

从 .env 文件加载环境变量

load_dotenv()

你的 .env 文件中需要包含以下环境变量：

LANGFUSE_PUBLIC_KEY=pk-lf-...

LANGFUSE_SECRET_KEY=sk-lf-...

LANGFUSE_HOST=https://cloud.langfuse.com (或你的自托管 URL)

ANTHROPIC_API_KEY=sk-ant-...

初始化客户端

langfuse_client = Langfuse() # 自动从环境读取密钥

anthropic_client = anthropic.Anthropic() # 从环境读取 ANTHROPIC_API_KEY

── 配置 ─────────────────────────────────────────────────────────────

将提示词存储在此处，而不是直接写在 API 调用中。

这样可以实现版本控制并独立测试。

SYSTEM_PROMPT = """你是一个乐于助人的客服助手。

请清晰简洁地回答问题。

如果你不知道某件事，请直接说明——不要猜测。"""

MODEL = "claude-sonnet-4-20250514"

截至 2026 年中期的 Anthropic 定价（价格变动时请更新）

用于计算每次调用的成本以进行成本追踪

COST_PER_INPUT_TOKEN = 3.00 / 1_000_000 # 每百万输入 token 3.00 美元

COST_PER_OUTPUT_TOKEN = 15.00 / 1_000_000 # 每百万输出 token 15.00 美元

def call_llm_with_tracing( user_message: str, session_id: str = "default-session", user_id: str = "anonymous" ) -> str: """ 执行带追踪的 LLM 调用。每次调用都会在 Langfuse 中创建一条追踪记录，包含：

• 完整的输入和输出
• Token 使用情况（输入、输出、总数）
• 计算出的美元成本
• 延迟（毫秒）
• 使用的模型和会话上下文

参数： user_message : 用户的消息 session_id : 在 Langfuse 中将相关调用分组到同一对话中 user_id : 与特定用户关联，用于分析

返回： LLM 的响应字符串 """

为本次用户交互创建顶层追踪

该追踪将在 Langfuse 控制台中显示为一个工作单元

trace = langfuse_client.trace( name="customer-support-call", session_id=session_id, user_id=user_id, input={"user_message": user_message, "system_prompt": SYSTEM_PROMPT} )

在追踪中创建一个生成跨度

此处捕获模型特定信息：模型名称、token 数量、成本等

generation = trace.generation( name="claude-completion", model=MODEL, input={ "system": SYSTEM_PROMPT, "messages": [{"role": "user", "content": user_message}] } )

start_time = time.time()

try:

执行 API 调用

response = anthropic_client.messages.create( model=MODEL, max_tokens=1024, system=SYSTEM_PROMPT, messages=[{"role": "user", "content": user_message}] )

latency_ms = int((time.time() - start_time) * 1000)

提取响应文本

response_text = response.content[0].text

从响应中提取 token 使用情况

input_tokens = response.usage.input_tokens

output_tokens = response.usage.output_tokens

total_tokens = input_tokens + output_tokens

计算本次调用的成本

cost_usd = ( input_tokens * COST_PER_INPUT_TOKEN + output_tokens * COST_PER_OUTPUT_TOKEN )

更新生成跨度以包含结果

这些数据将填充 Langfuse 的成本和 token 仪表板

generation.end( output=response_text, usage={ "input": input_tokens, "output": output_tokens, "total": total_tokens, "unit": "TOKENS" }, metadata={ "latency_ms": latency_ms, "cost_usd": round(cost_usd, 6), "model": MODEL } )

更新追踪以包含最终输出

trace.update( output={"response": response_text}, metadata={"total_cost_usd": round(cost_usd, 6)} )

打印摘要到 stdout，便于本地查看

print(f"\n{'─' * 60}") print(f"用户: {user_message}") print(f"Claude: {response_text}") print(f"Token: {input_tokens} in / {output_tokens} out / {total_tokens} total") print(f"成本: ${cost_usd:.6f}") print(f"延迟: {latency_ms}ms") print(f"追踪: {langfuse_client.base_url}/trace/{trace.id}") print(f"{'─' * 60}\n")

return response_text

except Exception as e:

在追踪中记录错误，以便在 Langfuse 中显示

generation.end( output=None, metadata={"error": str(e), "latency_ms": int((time.time() - start_time) * 1000)} )

trace.update(output={"error": str(e)})

在抛出异常前始终刷新 —— 确保错误追踪被发送

langfuse_client.flush()

raise

finally:

刷新会将所有缓冲事件发送到 Langfuse

在长时间运行的服务中，Langfuse 会自动刷新。

在脚本中，必须在进程退出前手动刷新。

langfuse_client.flush()

── 运行演示 ────────────────────────────────────────────────────────

if __name__ == "__main__":

模拟两次客服对话轮次

test_messages = [ "电子产品退货政策是什么？", "我45天前购买的商品可以退货吗？" ]

session = "demo-session-001"

for i, message in enumerate(test_messages): print(f"\n第 {i + 1}/{len(test_messages)} 次调用")

try: call_llm_with_tracing( user_message=message, session_id=session, user_id="test-user-42" ) except Exception as e: print(f"第 {i + 1} 次调用出错: {e}")

如何运行：

1

2

3

4

5

6

7

8

9

10

11

12

13

1. 创建你的 .env 文件

cat > .env << 'EOF' LANGFUSE_PUBLIC_KEY=pk-lf-your-key-here LANGFUSE_SECRET_KEY=sk-lf-your-key-here LANGFUSE_HOST=https://cloud.langfuse.com ANTHROPIC_API_KEY=sk-ant-your-key-here EOF

2. 安装依赖

pip install langfuse anthropic python-dotenv

3. 运行脚本

python llm_with_tracing.py

这段代码的作用：

1. 每次调用都会在 Langfuse 中创建一个两层结构：一个表示完整用户交互的 trace，以及其中表示具体模型调用的 generation 跨度。这种分离很重要，因为复杂的应用程序最终每个用户交互会包含多个模型调用——trace 将它们分组在一起，而单独的 generation 跨度则记录每次调用的成本和 token 数量。

1. generation.end() 调用会以 Langfuse 期望的格式提交 token 使用情况，从而自动填充其成本仪表板。

1. 在 finally 块中的 langfuse_client.flush() 是脚本中必需的——如果没有它，缓冲的事件将在 Python 退出前永远不会离开进程。

运行完上述代码后，打开你的 Langfuse 仪表板。你将看到两个调用都被记录下来，包括完整的输入、输出、token 数量以及每次调用的成本。这个“每次调用的成本”数字是你后续所有工作的基准。

第二阶段：RAG 管道与评估

大多数生产环境中的大语言模型（LLM）应用并不是简单的聊天机器人，而是 RAG（检索增强生成）系统。用户提出问题，相关文档从向量存储中检索出来，然后模型基于这些文档合成答案。构建这样的系统相对简单，但判断它是否真正有效才是难点。

需要构建的内容

一个文档问答系统：导入 PDF 或文本文件，将其分块，嵌入到向量数据库中，并在查询时检索相关片段。将该检索步骤连接到第一阶段中已追踪的 LLM 调用上。然后构建评估层，告诉你系统是否真的正确回答了问题。

对 RAG 评估至关重要的指标

RAGAS 提供了四个指标，涵盖了 RAG 系统的主要失败模式：

1. Faithfulness（忠实性）：答案是否基于检索到的上下文？还是模型凭空捏造？这可以捕捉模型超出文档内容进行“幻觉”的情况。
2. Answer relevance（答案相关性）：回复是否真正回应了所提的问题？即使答案忠实，也可能偏离重点。
3. Context precision（上下文精确性）：检索到的片段是否确实与问题相关？还是检索引入了噪声？
4. Context recall（上下文召回率）：检索到的上下文是否包含足够信息来回答问题？如果召回率低，说明你的分块或检索策略丢失了重要信息。

安装依赖项：

pip install ragas langchain-openai chromadb datasets python-dotenv

你需要：

• 一个 OpenAI API 密钥，用于 RAGAS 默认的“LLM 作为评判者”评估。
• 第一阶段中的 .env 文件，并添加 OPENAI_API_KEY=sk-…。

代码：RAGAS 评估管道

# rag_evaluation.py

# 目的：使用 RAGAS 指标评估 RAG 管道。
# 测量忠实性、答案相关性、上下文精确性和召回率。
# 在任何变更部署到生产环境之前，使用此脚本建立基准。

#
# 依赖项：
# pip install ragas langchain-openai chromadb datasets python-dotenv
#
# 设置：
# 将 OPENAI_API_KEY 添加到你的 .env 文件中
# （RAGAS 默认使用 GPT-4 作为评判模型）
#
# 运行：
# python rag_evaluation.py

import os
from dotenv import load_dotenv
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

load_dotenv()

# ── 示例评估数据集 ──────────────────────────────────────────────────
# 在实际项目中，这是你的“黄金数据集”——50-100 个带有真实答案的问题，
# 来自你的真实使用场景。
# 团队中的每个人都应同意这些答案是正确的。
# 这个数据集是你每次生产部署前都要运行的测试集。

#
# RAGAS 所需格式：
# question : 用户的问题
# answer : 你的 RAG 系统实际返回的答案
# contexts : 检索到的文档片段列表
# ground_truth : 正确答案（用于召回率和相关性评分）

EVALUATION_DATASET = {
    "question": [
        "电子产品退货期限是多久？",
        "标准配送需要多长时间？",
        "我可以退回已使用过的商品吗？",
    ],
    "answer": [
        # 这些是你的 RAG 系统返回的答案 —— 替换为真实输出
        "电子产品必须在购买后 15 天内原包装退回。",
        "标准配送对大多数地区需要 5-7 个工作日。",
        "商品必须处于原始未使用状态才能办理退货。",
    ],
    "contexts": [
        # 这些是检索器为每个问题返回的文档片段
        # 每个问题对应一个片段列表（每次查询可能检索多个片段）
        [
            "电子产品及外设的退货期限较短，为 15 天，"
            "且必须以原始未开封包装退回才符合条件。",
            "大多数标准商品可在购买后 30 天内退货。",
        ],
        [
            "标准配送通常需要 5-7 个工作日。"
            "结账时可选择快递选项以加快送达速度。",
        ],
        [
            "要符合退货条件，商品必须未使用且保持收到时的相同状态，"
            "并处于原始包装中。",
        ],
    ],
    "ground_truth": [
        # 这些是真实答案，用于评估
        "电子产品必须在购买后 15 天内原包装退回。",
        "标准配送对大多数地区需要 5-7 个工作日。",
        "商品必须处于原始未使用状态才能办理退货。",
    ],
}

# ── 创建评估数据集 ──────────────────────────────────────────────────
dataset = Dataset.from_dict(EVALUATION_DATASET)

# ── 定义评估指标 ──────────────────────────────────────────────────
metrics = [
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
]

# ── 执行评估 ──────────────────────────────────────────────────
results = evaluate(dataset, metrics=metrics)

# ── 输出结果 ──────────────────────────────────────────────────
print("RAG 评估结果：")
for metric_name, result in results.items():
    print(f"{metric_name}: {result:.4f}")

运行此脚本后，你会得到一组量化指标，展示当前 RAG 系统的表现。这些数值将成为你未来优化和迭代的基准线。

], "ground_truth": [

根据您的文档，正确答案是

"电子产品必须在15天内以原包装退回。", "标准配送需要5-7个工作日。", "退货商品必须未使用且保持原始状态。" ] }

def run_ragas_evaluation(dataset_dict: dict) -> dict: """ 对 RAG 流水线的输出运行 RAGAS 评估。

参数: dataset_dict : 包含 keys: question, answer, contexts, ground_truth 的字典

返回: 包含指标得分（faithfulness, answer_relevancy 等）的字典 """

将字典转换为 HuggingFace Dataset —— RAGAS 所需的格式

dataset = Dataset.from_dict(dataset_dict)

配置 RAGAS 用于判断输出的 LLM 和嵌入模型

RAGAS 使用 LLM-as-judge：它会提示 GPT-4 来评分每个答案

judge_llm = ChatOpenAI(model="gpt-4o-mini") # gpt-4o-mini 更便宜，但仍可靠 embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

print("正在运行 RAGAS 评估...") print(f"正在评估 {len(dataset)} 组问答对\n")

一次性运行全部四个指标

RAGAS 会对每个样本的每个指标发送多个 LLM 请求进行评分

results = evaluate( dataset=dataset, metrics=[ faithfulness, # 答案是否基于检索到的上下文？ answer_relevancy, # 答案是否回答了问题？ context_precision, # 检索到的片段是否与问题相关？ context_recall, # 上下文是否包含足够的信息来回答问题？ ], llm=judge_llm, embeddings=embeddings, ) return results

def print_evaluation_report(results) -> None: """ 打印可读的评估报告，包含分数和解释。 在生产环境中，应将这些分数写入数据库或仪表板。 """

将结果转换为 pandas DataFrame 以便于显示

df = results.to_pandas()

print("=" * 60) print("RAGAS 评估报告") print("=" * 60)

聚合所有样本的得分

metrics = ["faithfulness", "answer_relevancy", "context_precision", "context_recall"] thresholds = { "faithfulness": 0.85, # 低于此值：存在幻觉风险 "answer_relevancy": 0.80, # 低于此值：答案偏离问题 "context_precision": 0.75, # 低于此值：检索引入无关噪声 "context_recall": 0.80, # 低于此值：检索遗漏关键信息 }

print("\n聚合得分:")

all_pass = True for metric in metrics: if metric in df.columns: score = df[metric].mean() threshold = thresholds[metric] status = "PASS" if score >= threshold else "FAIL" if status == "FAIL": all_pass = False print(f"{metric:<22}: {score:.3f}[{status}] (阈值: {threshold})")

print("\n每题详细分析:")

for i, row in df.iterrows(): print(f"\n Q{i+1}: {row['question']}") print(f"答案: {row['answer'][:80]}...") for metric in metrics: if metric in df.columns: print(f"{metric:<22}: {row[metric]:.3f}")

print("\n" + "=" * 60)

if all_pass: print("结果: 所有指标均高于阈值 —— 可安全部署") else: print("结果: 一个或多个指标低于阈值 —— 请勿部署") print("在上线前请检查失败的问题")

print("=" * 60) return all_pass

if __name__ == "__main__": try: results = run_ragas_evaluation(EVALUATION_DATASET) all_pass = print_evaluation_report(results)

在 CI/CD 流水线中，当评估失败时以非零退出码阻止部署

你的流水线会检查这个退出码

if not all_pass: exit(1) except Exception as e: print(f"评估失败，错误信息: {e}") print("请检查你的 OPENAI_API_KEY 和数据集格式。") exit(1)

如何运行：

1 2 3 4 5 6 7 8

将你的 OpenAI 密钥添加到 .env 文件

echo "OPENAI_API_KEY=sk-your-key-here" >> .env

安装 RAGAS 及其依赖项

pip install ragas langchain-openai chromadb datasets python-dotenv

运行评估

python rag_evaluation.py

这段代码的作用：

• EVALUATION_DATASET 是你的黄金测试集——定义“正常工作”的 50–100 个问题及其正确答案。本示例中仅包含三条记录以保证演示可运行；在生产环境中建议至少使用 50 条。
• run_ragas_evaluation 函数将该数据集转换为 RAGAS 所需的 HuggingFace 格式，并一次性运行全部四项指标。
• RAGAS 内部使用 GPT-4 作为评判模型——它将每组答案和上下文发送给 LLM，并要求其从特定维度评分质量。
• print_evaluation_report 函数增加了阈值检查：如果任一指标低于其阈值，脚本将以非零退出码结束。在 GitHub Actions 或 CI/CD 流水线中，该退出码将阻止部署。这就是防止提示词退化进入生产环境的方式。

第三阶段：防护机制、成本控制与生产加固

到目前为止，你已经拥有一个可用且经过评估的 RAG 系统。第三阶段的目标是使其在真实流量下既安全又经济可行。

防护机制

输入防护机制在请求到达模型之前检测提示注入攻击、PII 和恶意意图。输出防护机制在将响应交付给用户前检查是否存在 PII 泄露、幻觉、有毒内容以及格式合规性。它们是防止有害输出的主要工具，也是任何面向客户的 LLM 部署的最低要求。

Guardrails AI 和 NeMo Guardrails 是两个主要选项。Guardrails AI 更加灵活且以代码为中心——你可以在 Python 中定义验证器。NeMo Guardrails 更侧重于对话流程，更适合希望控制模型参与话题范围的系统。

成本控制

LLMOps 已经成为一个完整的生产栈。成本是其中最具可操作性的层面之一。以下三种模式已被证明有效：

1. 语义缓存： 为语义相似的查询缓存响应。“_你们的营业时间是什么？_” 和 “_你们什么时候开门？_” 应返回相同的缓存响应，而不是发起两次 API 调用。LiteLLM 仅需一个配置选项即可实现语义缓存。
2. 模型路由： 将简单、短的查询路由到更便宜且更快的模型；将复杂、多步骤的查询路由到前沿模型。一个支持机器人同时处理“你们的电话号码是多少？”和“解释一下为什么我的发票有误”这类问题时，并不需要对两者使用相同的模型。
3. 令牌审计： 每周拉取你的 Langfuse 成本数据。找出令牌消耗最高的查询。十次中有九次，存在不必要的上下文传递——例如系统提示中包含永远无关的内容、检索的文本块过大，或对话历史未被修剪。

使用模型路由配置 LiteLLM

1

2# 安装 LiteLLM

pip install litellm python-dotenv

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101# cost_control.py

目的：演示如何使用 LiteLLM 实现模型路由和语义缓存。

将简单查询路由到廉价模型，复杂查询路由到前沿模型。

#

前提条件：

pip install litellm python-dotenv

#

运行方式：

python cost_control.py

import os

from dotenv import load_dotenv

import litellm

from litellm import completion

load_dotenv()

── 模型路由逻辑 ────────────────────────────────────────────────────────

简单启发式规则：将短查询路由到更便宜的模型。

在生产环境中，应替换为基于你数据训练的轻量级分类器，

或使用 LiteLLM Router 内置的负载均衡和降级配置。

CHEAP_MODEL = "claude-haiku-4-5-20251001" # 快速、便宜 —— 适合简单查询

FRONTIER_MODEL = "claude-sonnet-4-20250514" # 较慢、较贵 —— 用于复杂查询

令牌阈值：低于该估计令牌数的查询将路由到廉价模型

根据你的成本/质量权衡分析进行调整

ROUTING_THRESHOLD_CHARS = 200 # 粗略代理：约 200 字符 ≈ 50 个令牌

def route_query(user_message: str) -> str:

"""

根据查询复杂度将其路由到合适的模型。

返回本次查询应使用的模型字符串。

"""

基于长度的简单路由 —— 生产环境应替换为训练好的分类器

if len(user_message) < ROUTING_THRESHOLD_CHARS:

print(f"→ 路由至廉价模型（消息长度：{len(user_message)} 字符）")

return CHEAP_MODEL

else:

print(f"→ 路由至前沿模型（消息长度：{len(user_message)} 字符）")

return FRONTIER_MODEL

def call_with_routing(

user_message: str,

system_prompt: str = "你是一个乐于助人的客户服务助手。"

) -> dict:

"""

执行带有自动模型路由的 LLM 调用。

返回一个字典，包含响应文本、所用模型以及令牌数量。

"""

model = route_query(user_message)

response = completion(

model=model,

messages=[

{"role": "system", "content": system_prompt},

{"role": "user", "content": user_message}

],

max_tokens=512

)

result = {

"model_used": model,

"response": response.choices[0].message.content,

"input_tokens": response.usage.prompt_tokens,

"output_tokens": response.usage.completion_tokens,

"total_tokens": response.usage.total_tokens,

}

print(f"模型：{result['model_used']}")

print(f"令牌：{result['input_tokens']} 输入 / {result['output_tokens']} 输出")

print(f"响应：{result['response'][:100]}...")

return result

── 演示路由功能 ────────────────────────────────────────────────────────

if __name__ == "__main__":

queries = [

短查询 —— 路由至廉价模型

"你们的营业时间是什么？",

长而复杂的查询 —— 路由至前沿模型

"""我三周前购买了一款产品，收到后发现电源适配器有缺陷。

我已尝试过文档中的标准故障排除步骤，但问题仍然存在。

我需要了解更换设备或退款的选项，以及在超出电子产品标准 15 天退货窗口的情况下，具体流程是怎样的。"""

]

for i, query in enumerate(queries, 1):

print(f"\n{'─' * 60}")

print(f"查询 {i}: {query[:60]}...")

call_with_routing(query)

print(f"\n{'─' * 60}")

print("成本控制已配置。下一步建议：")

print("1. 在 Langfuse 中设置成本警报，当成本达到每周基准的 120% 时触发")

print("2. 为最常见的查询类型在 LiteLLM 中启用语义缓存")

print("3. 每周审查最高令牌消耗的调用，并修剪不必要的上下文")

第四阶段：智能体与高级评估（第 6 步及以后）

如今的机器学习工程师必须成为智能系统的架构师、复杂推理管道的编排者，以及人工智能可靠性与安全的守护者。智能体系统——即模型决定调用哪些工具以及调用顺序——正是这种复杂性叠加的地方。

一个单次调用的 RAG 系统只有一个主要失败模式：检索质量。而使用工具的智能体则有多种失败可能：它可能调用了错误的工具、传入了错误参数、陷入无限循环、无法识别任务已完成，或在第七步将第三步的错误累积成灾难性后果。评估变得更加困难，因为你不再只是评分一个响应，而是要评分一条完整的执行轨迹。

智能体评估方法

大多数生产团队最终采用的实际方法是：对 100% 的生产日志运行启发式评估，以低成本捕获明显失败；对 10–20% 的样本运行“大语言模型作为评判者”（LLM-as-judge）来以合理成本评估语义质量；并定期使用人工标注来重建和验证真实评估数据集。

启发式评估是可以程序化检查的内容：代理是否在迭代限制内完成？是否调用了正确的工具？输出是否符合预期格式？这些评估成本低且速度快，建议对每条追踪记录都运行。LLM作为评判者（LLM-as-judge）则更昂贵：你将完整的代理执行轨迹发送给更强的模型，让它判断代理是否达成了目标。这种评估只需在样本上运行。人工标注是最昂贵但最准确的方式：由人类审查一组代理运行结果，并标记出自动化评估所遗漏的问题。随着系统演进，使用它来保持评估数据集的准确性。

本阶段的工具

LangGraph 用于构建具有持久状态的有状态、多步骤代理工作流，支持失败后恢复。DeepEval 提供超越 RAGAS 的全面 LLM 评估功能，涵盖代理评估、工具调用正确性以及多轮对话质量。LangSmith Fleet 用于在生产环境中部署和追踪代理运行。

生产级 LLMOps 技术栈

真正的问题不在于哪个平台拥有最漂亮的仪表盘，而在于哪个平台满足驱动你技术栈的实际约束条件。以下是按类别划分的技术栈，列出关键选项。

对于早期团队，低成本的最小可行技术栈：

• 追踪： Langfuse 自托管（MIT 许可，功能完整，数据保留在你的基础设施中）
• 评估： RAGAS 用于 RAG 质量评估 + DeepEval 用于通用 LLM 质量评估
• 成本与路由： LiteLLM（开源，支持跨主流提供商的路由和缓存）
• 安全： Guardrails AI 用于输入/输出验证

各分类下的完整生产级技术栈：

| 类别 | 领先选项 | | --- | --- | | 追踪与可观测性 | Langfuse，LangSmith，Arize Phoenix | | 提示词管理 | LangSmith，Humanloop，PromptLayer | | 评估 | RAGAS，DeepEval，Braintrust | | 成本与路由 | LiteLLM，Portkey，Helicone | | 模型服务 | vLLM，BentoML，Baseten | | 安全与防护机制 | Guardrails AI，NeMo Guardrails，Lakera Guard | | 实验追踪 | MLflow，Weights and Biases |

如果你大量使用 LangChain 或 LangGraph，LangSmith 提供最深入的原生集成。如果开源和数据主权是不可妥协的要求，Langfuse 自托管是最佳选择。对于同时运行传统机器学习模型和 LLM 应用的团队，Arize AI 可提供跨模型类型的统一监控。

分步学习计划

这是具体的学习顺序。每一步都建立在前一步的基础上。不要跳步——后续步骤所需的技能若没有前期基础将难以理解。

• 第1步： 掌握 Python 和云基础，然后完成第一个 LLM API 项目。构建一个摘要器、一个分类器和一个简单的聊天机器人。目标是熟练使用 API——理解 token 的工作方式、temperature 的作用、错误如何暴露，以及如何清晰地组织调用。此时暂不引入任何 LLMOps 工具。
• 第2步： 为第1步中构建的应用添加可观测性。使用第1阶段的代码将 Langfuse 集成到项目中，记录每次调用。构建成本仪表盘。在继续之前，明确每个请求的基准成本和平均延迟。
• 第3步： 构建一个 RAG 系统。导入 20–50 个文档，进行分块处理，嵌入到 ChromaDB 或 Pinecone 中，并将检索功能接入已追踪的 LLM 调用。先让系统能正常工作，再考虑评估。理解检索失败的原因是你在本阶段最重要的收获。
• 第4步： 为 RAG 系统添加评估功能。创建包含 50 个问题及其正确答案的“黄金数据集”。使用第2阶段的代码运行 RAGAS，建立基准得分。然后故意破坏某些部分——更改分块策略、修改提示词——并确认评估能否捕捉到性能下降。这一步让评估真正变得有意义。
• 第5步： 为生产环境加固系统。使用 Guardrails AI 添加输入和输出防护机制。通过 LiteLLM 配置语义缓存和模型路由。设置成本告警，当周成本超过基准值 120% 时触发。对你的 API 端点进行压力测试，找出系统崩溃点，确保在用户遇到问题前发现。
• 第6步及以后： 使用 LangGraph 实现 ReAct 循环。构建一个能够搜索、检索并整合多个来源信息的工具使用代理。加入 LLM-as-judge 评估机制。将你的评估套件集成到 CI/CD 流程中，当评分下降时自动阻止部署。这才是区分实践者与初学者的关键工作。

结论

LLMOps 并非与 AI 工程分离的职业路径，而是决定 AI 工程师构建的 AI 系统能否在大规模场景下真正有效运行的生产性学科。本路线图涵盖的技能并非仅适用于“已完成开发”的团队的可选附加项，而是使所有其他工作变得可靠、可审计、可改进的基础。

路径之所以是顺序的，是有原因的。你无法评估一个未进行可观测性建设的系统；你无法加固一个未经评估的系统；你也无法构建那些你不知道如何调试的代理。上述循序渐进的计划正是围绕这种依赖关系设计的。从第一步开始，构建真实可用的系统，并在每个里程碑真正完成后再继续推进。到 2026 年，相关工具已足够成熟，足以从第一天起支持这一路径——唯一能培养出真正的 LLMOps 工程师的方式，就是构建并交付真实的系统。

##### 尚无评论。