LangSmith与LLM应用可观测性完全教程
从概念理解到企业级LLMOps平台搭建，构建生产级LLM应用的可观测性体系
目录
- LLM可观测性概述
- LangSmith架构与核心概念
- 快速接入与环境配置
- Trace追踪机制
- Prompt版本管理
- 评估体系：自动与人工
- 成本监控与优化
- LangFuse开源替代方案
- 生产级监控与告警
- CI/CD集成
- 企业级LLMOps平台搭建
- 最佳实践总结
1. LLM可观测性概述
1.1 为什么LLM应用需要可观测性
传统软件的可观测性建立在确定性逻辑之上——相同的输入永远产生相同的输出。LLM应用打破了这一假设：
| 维度 | 传统应用 | LLM应用 |
|---|---|---|
| 输出确定性 | 确定性 | 概率性 |
| 错误类型 | 逻辑错误、异常 | 幻觉、偏见、格式错误 |
| 性能瓶颈 | CPU/内存/IO | Token延迟、API限流 |
| 成本模型 | 固定基础设施 | 按Token计费，不可预测 |
| 质量评估 | 单元测试/集成测试 | 主观评估 + 自动指标 |
| 调试方式 | 堆栈跟踪 | 完整的推理链追踪 |
LLM应用面临的核心挑战：
1. 黑盒推理：用户问了什么 → 模型"想"了什么 → 为什么给出这个回答？
2. 质量漂移：同一提示词在不同模型版本下表现可能截然不同
3. 成本失控：一次RAG调用可能消耗数千Token，成本难以预测
4. 延迟波动：模型响应时间从100ms到30s不等
5. 幻觉检测：模型"自信地"给出错误信息
1.2 可观测性三大支柱在LLM中的映射
传统可观测性 LLM可观测性
───────────── ─────────────
Logs (日志) → Traces (推理链追踪)
- 完整的调用链路
- 每一步的输入/输出
- Token使用详情
Metrics (指标) → Metrics (LLM指标)
- 延迟分布 (P50/P95/P99)
- Token消耗趋势
- 成本仪表板
- 质量评分趋势
Traces (链路) → Evaluations (评估)
- 自动评估（相似度、正确性）
- 人工标注与反馈
- A/B测试对比
1.3 主流LLM可观测性工具对比
| 工具 | 类型 | 核心优势 | 适用场景 |
|---|---|---|---|
| LangSmith | 商业SaaS | LangChain深度集成，评估体系完善 | LangChain生态用户 |
| LangFuse | 开源自托管 | 完全控制数据，社区活跃 | 数据敏感/自定义需求 |
| Helicone | 商业SaaS | 简单易用，代理模式接入 | 快速上手，轻量需求 |
| Arize Phoenix | 商业+开源 | 可视化强大，嵌入向量分析 | RAG应用调试 |
| Weights & Biases | 商业SaaS | 实验跟踪，与ML流程统一 | ML团队统一平台 |
| Datadog LLM | 商业SaaS | 与现有APM集成 | 已用Datadog的企业 |
2. LangSmith架构与核心概念
2.1 LangSmith架构概览
┌─────────────────────────────────────────────────────────┐
│ LangSmith Platform │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Traces │ │ Datasets │ │Evaluations│ │ Prompts │ │
│ │ 追踪引擎 │ │ 数据管理 │ │ 评估系统 │ │ 版本管理 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Monitor │ │ Rules │ │ Play- │ │
│ │ 监控告警 │ │ 规则引擎 │ │ ground │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────┘
↑ ↑ ↑
┌────────┐ ┌────────┐ ┌────────┐
│Python │ │ JS/TS │ │ REST │
│ SDK │ │ SDK │ │ API │
└────────┘ └────────┘ └────────┘
↑ ↑ ↑
┌────────┐ ┌────────┐ ┌────────┐
│LangChain│ │LlamaIndex│ │ 自定义 │
│ 应用 │ │ 应用 │ │ 应用 │
└────────┘ └────────┘ └────────┘
2.2 核心概念
Trace（追踪）：一次完整的用户请求链路，包含所有子步骤。
Trace: "用户询问天气"
├── Chain: 处理用户输入
│ ├── LLM: 意图识别 (200ms, 150 tokens)
│ └── Tool: 天气API调用 (500ms)
├── LLM: 生成回答 (1.2s, 300 tokens)
└── 总计: 1.9s, 450 tokens, $0.003
Run（运行）：Trace中的单个步骤，类型包括：
llm：模型调用chain：链式调用tool：工具调用retriever：文档检索parser：输出解析
Dataset（数据集）：用于评估的输入/输出对集合。
Evaluation（评估）：对模型输出质量的量化衡量。
Experiment（实验）：在数据集上运行的评估批次。
3. 快速接入与环境配置
3.1 注册与API密钥
# 1. 访问 https://smith.langchain.com 注册账户
# 2. 在 Settings → API Keys 创建密钥
# 设置环境变量
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="lsv2_pt_xxxxxxxxxxxxxxxxxxxx"
export LANGCHAIN_PROJECT="my-project" # 项目名称
export LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
3.2 Python SDK安装
pip install langsmith
# 如果使用LangChain生态
pip install langchain langchain-openai langchain-community
3.3 最小化接入示例
import os
from langsmith import Client
from langsmith.run_helpers import traceable
# 初始化客户端
client = Client()
# 方式1：使用装饰器自动追踪
@traceable(
name="My LLM Call",
run_type="chain",
project_name="my-project"
)
def ask_question(question: str) -> str:
from openai import OpenAI
oai_client = OpenAI() # 自动读取OPENAI_API_KEY
response = oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content
# 调用函数 - 自动创建Trace并上报
answer = ask_question("什么是LLM可观测性？")
print(answer)
3.4 LangChain集成
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 设置环境变量后，LangChain自动上报所有调用到LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_pt_xxx"
# 创建链
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的技术文档翻译助手。"),
("user", "请将以下内容翻译成英文：{text}")
])
chain = prompt | llm | StrOutputParser()
# 执行 - 自动追踪到LangSmith
result = chain.invoke({"text": "LangSmith是一个LLM应用可观测性平台"})
print(result)
3.5 LlamaIndex集成
import os
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
# 启用LangSmith追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_pt_xxx"
# LlamaIndex也支持自动追踪
Settings.llm = OpenAI(model="gpt-4o-mini")
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
# 所有检索和LLM调用都会被追踪
response = query_engine.query("这个文档的主要内容是什么？")
print(response)
4. Trace追踪机制
4.1 手动创建Trace
from langsmith import Client, traceable
from langsmith.run_helpers import tracing_context
client = Client()
# 方式1：使用上下文管理器
with tracing_context(project_name="my-project") as run:
# 所有在此上下文中的@traceable调用都归属到这个trace
result = my_function("input")
# 方式2：手动创建和管理Run
def manual_trace_example():
# 创建根Run
root_run = client.create_run(
name="Root Chain",
run_type="chain",
inputs={"question": "什么是RAG?"},
project_name="my-project"
)
# 创建子Run - LLM调用
llm_run = client.create_run(
name="GPT-4o Call",
run_type="llm",
inputs={
"messages": [{"role": "user", "content": "解释RAG"}],
"model": "gpt-4o"
},
parent_run_id=root_run.id
)
# ... 执行LLM调用 ...
# 更新子Run结果
client.update_run(
run_id=llm_run.id,
outputs={
"content": "RAG是检索增强生成...",
"usage": {"prompt_tokens": 50, "completion_tokens": 200}
},
end_time=datetime.now()
)
# 更新根Run结果
client.update_run(
run_id=root_run.id,
outputs={"answer": "RAG是检索增强生成..."},
end_time=datetime.now()
)
4.2 复杂链路追踪
from langsmith import traceable
from typing import Dict, Any
import time
@traceable(run_type="chain", name="RAG Pipeline")
def rag_pipeline(question: str) -> str:
"""完整的RAG管道追踪"""
# 步骤1：查询改写
rewritten = rewrite_query(question)
# 步骤2：文档检索
docs = retrieve_documents(rewritten)
# 步骤3：上下文构建
context = build_context(docs)
# 步骤4：生成回答
answer = generate_answer(question, context)
return answer
@traceable(run_type="llm", name="Query Rewriting")
def rewrite_query(question: str) -> str:
"""查询改写 - 被自动追踪为子步骤"""
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "将用户问题改写为更适合搜索的形式。"},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content
@traceable(run_type="retriever", name="Document Retrieval")
def retrieve_documents(query: str) -> list:
"""文档检索 - 被自动追踪为retriever类型"""
# 模拟向量检索
from openai import OpenAI
client = OpenAI()
# 获取embedding
embedding = client.embeddings.create(
model="text-embedding-3-small",
input=query
)
# 模拟检索结果
return [
{"content": "文档片段1...", "score": 0.92},
{"content": "文档片段2...", "score": 0.87},
{"content": "文档片段3...", "score": 0.81}
]
@traceable(run_type="llm", name="Answer Generation")
def generate_answer(question: str, context: str) -> str:
"""生成回答"""
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": f"基于以下上下文回答问题：\n{context}"},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content
# 调用 - 完整的调用链会被追踪
result = rag_pipeline("LangSmith的主要功能是什么？")
4.3 Trace元数据与标签
from langsmith import traceable
@traceable(
name="Product Search",
run_type="chain",
metadata={
"environment": "production",
"version": "1.2.0",
"feature_flag": "new_search_v2"
},
tags=["search", "production", "v1.2"]
)
def product_search(query: str, user_id: str) -> dict:
# 可以在运行时动态添加元数据
from langsmith.context import get_current_run_tree
run_tree = get_current_run_tree()
if run_tree:
run_tree.metadata["user_id"] = user_id
run_tree.metadata["query_length"] = len(query)
# ... 执行搜索逻辑 ...
return {"results": [], "total": 0}
4.4 错误追踪
from langsmith import traceable
from langsmith.context import get_current_run_tree
@traceable(run_type="chain", name="Fragile API Call")
def call_external_api(prompt: str) -> str:
try:
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
except Exception as e:
# 错误会被自动记录到LangSmith
run_tree = get_current_run_tree()
if run_tree:
run_tree.metadata["error_type"] = type(e).__name__
run_tree.metadata["error_message"] = str(e)
raise # 重新抛出，LangSmith会记录错误状态
5. Prompt版本管理
5.1 LangSmith Hub
LangSmith提供了Prompt Hub功能，用于集中管理和版本化提示词：
from langsmith import Client
client = Client()
# 推送提示词到Hub
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的代码审查助手。请用中文回答，"
"并遵循以下原则：\n1. 指出潜在的bug\n"
"2. 建议性能优化\n3. 评估代码可读性"),
("user", "请审查以下代码：\n```{language}\n{code}\n```")
])
# 创建或更新提示词
client.push_prompt(
"code-review-assistant",
object=prompt,
description="代码审查助手提示词",
tags=["code", "review", "chinese"],
is_public=False # 私有提示词
)
5.2 提示词版本控制
from langsmith import Client
from langchain_core.prompts import ChatPromptTemplate
client = Client()
# 获取特定版本的提示词
prompt_v1 = client.pull_prompt("code-review-assistant", version=1)
prompt_v2 = client.pull_prompt("code-review-assistant", version=2)
# 获取最新版本
prompt_latest = client.pull_prompt("code-review-assistant")
# 列出所有版本
versions = client.list_prompt_versions("code-review-assistant")
for v in versions:
print(f"Version {v.version}: {v.created_at} - {v.description}")
5.3 提示词A/B测试
from langsmith import Client, traceable
import random
client = Client()
# 拉取两个版本的提示词
prompt_v1 = client.pull_prompt("customer-service-v1")
prompt_v2 = client.pull_prompt("customer-service-v2")
@traceable(name="A/B Test Router")
def route_to_prompt(user_message: str, user_id: str) -> str:
"""基于用户ID的A/B测试路由"""
# 使用用户ID的哈希确保同一用户始终看到同一版本
variant = hash(user_id) % 2
if variant == 0:
prompt = prompt_v1
variant_name = "v1"
else:
prompt = prompt_v2
variant_name = "v2"
from openai import OpenAI
oai_client = OpenAI()
messages = prompt.format_messages(user_message=user_message)
response = oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": m.type, "content": m.content} for m in messages]
)
result = response.choices[0].message.content
# 记录A/B测试元数据
from langsmith.context import get_current_run_tree
run_tree = get_current_run_tree()
if run_tree:
run_tree.metadata["ab_variant"] = variant_name
run_tree.metadata["prompt_version"] = variant + 1
return result
5.4 提示词与代码分离的最佳实践
# project/prompts.py - 集中管理所有提示词
from langsmith import Client
from functools import lru_cache
_client = None
def get_client():
global _client
if _client is None:
_client = Client()
return _client
@lru_cache(maxsize=32)
def get_prompt(name: str, version: int = None):
"""获取提示词，带缓存"""
client = get_client()
if version:
return client.pull_prompt(f"{name}:{version}")
return client.pull_prompt(name)
# 在业务代码中使用
def answer_customer_question(question: str) -> str:
prompt = get_prompt("customer-service")
chain = prompt | get_llm()
return chain.invoke({"question": question})
6. 评估体系：自动与人工
6.1 评估框架概述
LangSmith的评估体系分为三个层次：
┌─────────────────────────────────────────────┐
│ Layer 3: 人工评估 │
│ - 众包标注 - 专家评审 - 用户反馈 │
├─────────────────────────────────────────────┤
│ Layer 2: LLM-as-Judge │
│ - 模型评分 - 对比评估 - 多维度打分 │
├─────────────────────────────────────────────┤
│ Layer 1: 自动指标 │
│ - 字符匹配 - 语义相似度 - 格式校验 │
└─────────────────────────────────────────────┘
6.2 创建评估数据集
from langsmith import Client
client = Client()
# 创建数据集
dataset = client.create_dataset(
dataset_name="customer-service-eval-v2",
description="客服助手评估数据集 - 2024Q4"
)
# 添加测试用例
test_cases = [
{
"inputs": {"question": "如何重置密码？"},
"outputs": {
"expected": "您可以通过以下步骤重置密码：1. 点击登录页面的'忘记密码'；2. 输入注册邮箱；3. 查收重置邮件；4. 设置新密码。"
},
"metadata": {"difficulty": "easy", "category": "account"}
},
{
"inputs": {"question": "我的订单显示已发货但物流信息三天没更新"},
"outputs": {
"expected": "很抱歉给您带来不便。物流信息延迟更新可能有以下原因：1. 快递公司系统同步延迟；2. 包裹正在中转站；3. 天气等不可抗力因素。建议您先等待24小时，如仍未更新，请提供订单号，我为您联系快递公司查询。"
},
"metadata": {"difficulty": "medium", "category": "order"}
},
{
"inputs": {"question": "你们的产品质量太差了，我要投诉！"},
"outputs": {
"expected": "非常抱歉给您带来不好的体验。我理解您的感受，我们会认真对待每一位客户的反馈。请您提供具体的订单号和遇到的问题，我会立即为您记录并转交给专门的售后团队处理。我们承诺会在24小时内给您回复。"
},
"metadata": {"difficulty": "hard", "category": "complaint"}
}
]
for case in test_cases:
client.create_example(
inputs=case["inputs"],
outputs=case["outputs"],
dataset_id=dataset.id,
metadata=case.get("metadata")
)
print(f"数据集创建完成，共 {len(test_cases)} 条用例")
6.3 自动评估器
from langsmith import Client, evaluate
from langsmith.evaluation import EvaluationResult
client = Client()
# 评估器1：字符串精确匹配
def exact_match_evaluator(run, example):
"""精确匹配评估"""
prediction = run.outputs.get("output", "")
expected = example.outputs.get("expected", "")
score = 1.0 if prediction.strip() == expected.strip() else 0.0
return EvaluationResult(
key="exact_match",
score=score,
comment="完全匹配" if score == 1.0 else "不匹配"
)
# 评估器2：语义相似度
def semantic_similarity_evaluator(run, example):
"""语义相似度评估（使用embedding）"""
from openai import OpenAI
oai_client = OpenAI()
prediction = run.outputs.get("output", "")
expected = example.outputs.get("expected", "")
# 获取两段文本的embedding
emb_pred = oai_client.embeddings.create(
model="text-embedding-3-small",
input=prediction
).data[0].embedding
emb_expected = oai_client.embeddings.create(
model="text-embedding-3-small",
input=expected
).data[0].embedding
# 计算余弦相似度
import numpy as np
similarity = np.dot(emb_pred, emb_expected) / (
np.linalg.norm(emb_pred) * np.linalg.norm(emb_expected)
)
return EvaluationResult(
key="semantic_similarity",
score=float(similarity),
comment=f"语义相似度: {similarity:.3f}"
)
# 评估器3：LLM-as-Judge
def llm_judge_evaluator(run, example):
"""使用LLM评估回答质量"""
from openai import OpenAI
oai_client = OpenAI()
question = example.inputs.get("question", "")
prediction = run.outputs.get("output", "")
expected = example.outputs.get("expected", "")
judge_prompt = f"""请评估以下AI回答的质量。
用户问题：{question}
AI回答：{prediction}
参考答案：{expected}
请从以下维度评分（1-5分）：
1. 准确性：信息是否正确
2. 完整性：是否涵盖关键点
3. 友好度：语气是否专业友善
请以JSON格式返回：{{"accuracy": X, "completeness": X, "friendliness": X, "overall": X, "reason": "..."}}"""
response = oai_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": judge_prompt}],
response_format={"type": "json_object"}
)
import json
scores = json.loads(response.choices[0].message.content)
return EvaluationResult(
key="llm_judge",
score=scores.get("overall", 0) / 5.0, # 归一化到0-1
comment=json.dumps(scores, ensure_ascii=False)
)
# 运行评估
def target_function(inputs):
"""被评估的函数"""
from openai import OpenAI
oai_client = OpenAI()
response = oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "你是一个专业的客服助手。"},
{"role": "user", "content": inputs["question"]}
]
)
return {"output": response.choices[0].message.content}
# 执行评估实验
experiment_results = evaluate(
target_function,
data="customer-service-eval-v2",
evaluators=[
exact_match_evaluator,
semantic_similarity_evaluator,
llm_judge_evaluator
],
experiment_prefix="gpt4o-mini-baseline",
metadata={
"model": "gpt-4o-mini",
"temperature": 0,
"version": "1.0"
}
)
print(f"评估完成！平均分数：{experiment_results}")
6.4 人工标注与反馈
from langsmith import Client
client = Client()
# 创建标注队列
queue = client.create_annotation_queue(
name="客服回答质量审核",
description="人工审核AI客服回答的质量",
dataset_id="customer-service-eval-v2"
)
# 提交需要人工标注的Run
def submit_for_review(run_id: str):
client.create_annotation(
run_id=run_id,
annotation_queue_id=queue.id,
metadata={"priority": "high", "reason": "低置信度回答"}
)
# 通过API收集用户反馈
@traceable(name="Collect Feedback")
def collect_user_feedback(run_id: str, feedback_type: str, comment: str = ""):
"""收集终端用户的反馈"""
score_map = {"positive": 1.0, "neutral": 0.5, "negative": 0.0}
client.create_feedback(
run_id=run_id,
key="user_feedback",
score=score_map.get(feedback_type, 0.5),
comment=comment,
source_info={"source": "user_widget", "version": "1.0"}
)
# 在应用中嵌入反馈收集
@traceable(name="Chat Response")
def chat_response(user_message: str) -> dict:
from openai import OpenAI
oai_client = OpenAI()
response = oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": user_message}]
)
answer = response.choices[0].message.content
return {
"answer": answer,
"run_id": "从上下文获取当前run_id" # 实际从traceable上下文获取
}
7. 成本监控与优化
7.1 Token使用追踪
from langsmith import Client, traceable
from openai import OpenAI
client = Client()
oai_client = OpenAI()
@traceable(name="Cost Tracked LLM Call")
def cost_tracked_call(prompt: str, model: str = "gpt-4o") -> dict:
"""带有详细成本追踪的LLM调用"""
response = oai_client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
usage = response.usage
# 价格表（美元/1M tokens）
pricing = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gpt-4-turbo": {"input": 10.00, "output": 30.00},
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
}
model_pricing = pricing.get(model, {"input": 0, "output": 0})
input_cost = (usage.prompt_tokens / 1_000_000) * model_pricing["input"]
output_cost = (usage.completion_tokens / 1_000_000) * model_pricing["output"]
total_cost = input_cost + output_cost
# 记录到LangSmith元数据
from langsmith.context import get_current_run_tree
run_tree = get_current_run_tree()
if run_tree:
run_tree.metadata.update({
"model": model,
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 6),
"total_cost_usd": round(total_cost, 6)
})
return {
"content": response.choices[0].message.content,
"usage": {
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
"cost_usd": round(total_cost, 6)
}
}
7.2 成本仪表板查询
from langsmith import Client
from datetime import datetime, timedelta
client = Client()
def get_cost_report(project_name: str, days: int = 7) -> dict:
"""获取项目成本报告"""
start_time = datetime.now() - timedelta(days=days)
# 获取所有Run
runs = client.list_runs(
project_name=project_name,
start_time=start_time.isoformat(),
run_type="llm"
)
total_cost = 0
total_tokens = 0
model_costs = {}
daily_costs = {}
for run in runs:
if run.metadata:
cost = run.metadata.get("total_cost_usd", 0)
tokens = run.metadata.get("total_tokens", 0)
model = run.metadata.get("model", "unknown")
total_cost += cost
total_tokens += tokens
# 按模型统计
if model not in model_costs:
model_costs[model] = {"cost": 0, "calls": 0, "tokens": 0}
model_costs[model]["cost"] += cost
model_costs[model]["calls"] += 1
model_costs[model]["tokens"] += tokens
# 按天统计
day = run.start_time.strftime("%Y-%m-%d") if run.start_time else "unknown"
if day not in daily_costs:
daily_costs[day] = 0
daily_costs[day] += cost
return {
"period": f"Last {days} days",
"total_cost_usd": round(total_cost, 4),
"total_tokens": total_tokens,
"total_calls": sum(m["calls"] for m in model_costs.values()),
"avg_cost_per_call": round(total_cost / max(sum(m["calls"] for m in model_costs.values()), 1), 6),
"by_model": model_costs,
"daily_breakdown": daily_costs
}
# 生成报告
report = get_cost_report("my-project", days=30)
print(f"过去30天总成本: ${report['total_cost_usd']}")
print(f"总调用次数: {report['total_calls']}")
print(f"平均每次调用: ${report['avg_cost_per_call']}")
7.3 成本优化策略
# 策略1：智能模型路由
def smart_model_router(query: str, complexity_threshold: float = 0.7) -> str:
"""根据查询复杂度选择模型"""
from openai import OpenAI
oai_client = OpenAI()
# 使用小模型评估复杂度
eval_response = oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"评估以下问题的复杂度（0-1的JSON）：{query}"
}],
response_format={"type": "json_object"}
)
complexity = json.loads(eval_response.choices[0].message.content).get("score", 0.5)
# 根据复杂度选择模型
if complexity < 0.3:
return "gpt-4o-mini" # 简单问题用便宜模型
elif complexity < 0.7:
return "gpt-4o-mini" # 中等问题用中等模型
else:
return "gpt-4o" # 复杂问题用强模型
# 策略2：语义缓存
import hashlib
import numpy as np
from typing import Optional
class SemanticCache:
"""基于语义相似度的缓存"""
def __init__(self, similarity_threshold: float = 0.95):
self.cache = [] # [(embedding, response, query)]
self.threshold = similarity_threshold
self.openai_client = OpenAI()
def _get_embedding(self, text: str) -> list:
response = self.openai_client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def get(self, query: str) -> Optional[str]:
"""查找语义相似的缓存"""
if not self.cache:
return None
query_emb = self._get_embedding(query)
for cached_emb, cached_response, cached_query in self.cache:
similarity = np.dot(query_emb, cached_emb) / (
np.linalg.norm(query_emb) * np.linalg.norm(cached_emb)
)
if similarity >= self.threshold:
return cached_response
return None
def set(self, query: str, response: str):
"""缓存新的问答对"""
embedding = self._get_embedding(query)
self.cache.append((embedding, response, query))
# 限制缓存大小
if len(self.cache) > 1000:
self.cache = self.cache[-500:]
8. LangFuse开源替代方案
8.1 LangFuse概述
LangFuse是一个完全开源的LLM可观测性平台，可以自托管，提供与LangSmith类似的功能：
| 特性 | LangSmith | LangFuse |
|---|---|---|
| 部署方式 | SaaS | 自托管 / Cloud |
| 开源 | 否 | 是 (MIT) |
| 数据控制 | 第三方托管 | 完全自控 |
| LangChain集成 | 原生 | 优秀 |
| LlamaIndex集成 | 原生 | 优秀 |
| 评估系统 | 完善 | 完善 |
| 社区 | LangChain官方 | 活跃开源社区 |
8.2 自托管部署
# 使用Docker Compose部署LangFuse
# docker-compose.yml
version: '3.8'
services:
langfuse:
image: langfuse/langfuse:latest
ports:
- "3000:3000"
environment:
- DATABASE_URL=postgresql://postgres:postgres@db:5432/langfuse
- NEXTAUTH_URL=http://localhost:3000
- NEXTAUTH_SECRET=your-secret-key-here
- SALT=your-salt-here
depends_on:
- db
db:
image: postgres:15
environment:
- POSTGRES_DB=langfuse
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
volumes:
- langfuse_data:/var/lib/postgresql/data
volumes:
langfuse_data:
# 启动
docker-compose up -d
# 访问 http://localhost:3000 创建账户
8.3 LangFuse Python SDK
pip install langfuse
import os
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
# 初始化
langfuse = Langfuse(
public_key="pk-lf-xxx",
secret_key="sk-lf-xxx",
host="http://localhost:3000" # 自托管地址
)
# 使用装饰器追踪
@observe(as_type="generation")
def call_llm(prompt: str, model: str = "gpt-4o-mini") -> str:
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
# 记录到LangFuse
langfuse_context.update_current_observation(
usage={
"input": response.usage.prompt_tokens,
"output": response.usage.completion_tokens,
"total": response.usage.total_tokens
},
model=model,
metadata={"temperature": 0}
)
return response.choices[0].message.content
@observe(as_type="chain")
def rag_pipeline(question: str) -> str:
"""完整的RAG管道"""
# 检索
docs = retrieve_docs(question)
# 生成
context = "\n".join(docs)
answer = call_llm(
f"基于以下上下文回答：\n{context}\n\n问题：{question}"
)
return answer
# 调用
result = rag_pipeline("什么是LLMOps？")
# 确保所有事件都已发送
langfuse.flush()
8.4 LangFuse评估
from langfuse import Langfuse
from langfuse.decorators import observe
langfuse = Langfuse()
# 创建评估数据集
dataset = langfuse.create_dataset(
name="qa-evaluation",
description="问答系统评估"
)
# 添加数据点
langfuse.create_dataset_item(
dataset_name="qa-evaluation",
input={"question": "什么是机器学习？"},
expected_output={"answer": "机器学习是人工智能的一个分支..."}
)
# 运行评估
@observe()
def run_evaluation():
dataset = langfuse.get_dataset("qa-evaluation")
for item in dataset.items:
# 在每个数据点上运行你的应用
with item.observe(run_name="gpt4o-eval") as trace:
answer = call_llm(item.input["question"])
# 记录评估分数
langfuse.score(
trace_id=trace.id,
name="accuracy",
value=1.0 if "机器学习" in answer else 0.0
)
run_evaluation()
langfuse.flush()
9. 生产级监控与告警
9.1 监控指标体系
from dataclasses import dataclass
from typing import Dict, List
from datetime import datetime
@dataclass
class LLMMetric:
"""LLM应用核心监控指标"""
# 性能指标
latency_p50_ms: float # 中位延迟
latency_p95_ms: float # P95延迟
latency_p99_ms: float # P99延迟
throughput_rps: float # 每秒请求数
# 质量指标
avg_score: float # 平均质量评分
low_score_rate: float # 低分回答比例
hallucination_rate: float # 幻觉率
error_rate: float # 错误率
# 成本指标
total_cost_usd: float # 总成本
avg_cost_per_request: float # 单请求平均成本
total_tokens: int # 总Token数
# 业务指标
user_satisfaction: float # 用户满意度
task_completion_rate: float # 任务完成率
# 监控阈值配置
ALERT_THRESHOLDS = {
"latency_p95_ms": {"warning": 5000, "critical": 10000},
"error_rate": {"warning": 0.05, "critical": 0.10},
"low_score_rate": {"warning": 0.15, "critical": 0.30},
"hallucination_rate": {"warning": 0.10, "critical": 0.20},
"daily_cost_usd": {"warning": 100, "critical": 500},
"avg_cost_per_request": {"warning": 0.05, "critical": 0.20}
}
9.2 实时监控实现
import time
from collections import deque
from dataclasses import dataclass, field
from threading import Lock
from typing import Optional
import json
class LLMMonitor:
"""实时LLM监控系统"""
def __init__(self, window_size: int = 1000):
self.window_size = window_size
self.latencies = deque(maxlen=window_size)
self.costs = deque(maxlen=window_size)
self.errors = deque(maxlen=window_size)
self.scores = deque(maxlen=window_size)
self.lock = Lock()
# 告警历史
self.alerts = []
def record_request(
self,
latency_ms: float,
cost_usd: float,
tokens: int,
success: bool,
score: Optional[float] = None,
metadata: Optional[dict] = None
):
"""记录一次请求"""
with self.lock:
self.latencies.append(latency_ms)
self.costs.append({"cost": cost_usd, "tokens": tokens})
self.errors.append(0 if success else 1)
if score is not None:
self.scores.append(score)
# 检查告警
self._check_alerts()
def _check_alerts(self):
"""检查是否需要触发告警"""
metrics = self.get_metrics()
# 延迟告警
if metrics["latency_p95"] > ALERT_THRESHOLDS["latency_p95_ms"]["critical"]:
self._trigger_alert("CRITICAL", f"P95延迟过高: {metrics['latency_p95']:.0f}ms")
# 错误率告警
if metrics["error_rate"] > ALERT_THRESHOLDS["error_rate"]["critical"]:
self._trigger_alert("CRITICAL", f"错误率过高: {metrics['error_rate']:.2%}")
# 成本告警
daily_cost = sum(c["cost"] for c in self.costs)
if daily_cost > ALERT_THRESHOLDS["daily_cost_usd"]["warning"]:
self._trigger_alert("WARNING", f"日成本过高: ${daily_cost:.2f}")
def _trigger_alert(self, level: str, message: str):
"""触发告警"""
alert = {
"level": level,
"message": message,
"timestamp": datetime.now().isoformat()
}
self.alerts.append(alert)
# 发送通知（邮件、Slack、PagerDuty等）
self._send_notification(alert)
def _send_notification(self, alert: dict):
"""发送告警通知"""
# Slack webhook示例
import requests
webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
payload = {
"text": f"🚨 LLM告警 [{alert['level']}]\n{alert['message']}",
"username": "LLM Monitor"
}
try:
requests.post(webhook_url, json=payload, timeout=5)
except:
pass # 告警发送失败不影响主流程
def get_metrics(self) -> dict:
"""获取当前指标"""
with self.lock:
if not self.latencies:
return {}
sorted_latencies = sorted(self.latencies)
n = len(sorted_latencies)
return {
"latency_p50": sorted_latencies[int(n * 0.5)],
"latency_p95": sorted_latencies[int(n * 0.95)],
"latency_p99": sorted_latencies[int(n * 0.99)],
"error_rate": sum(self.errors) / len(self.errors),
"avg_cost": sum(c["cost"] for c in self.costs) / len(self.costs),
"total_tokens": sum(c["tokens"] for c in self.costs),
"avg_score": sum(self.scores) / len(self.scores) if self.scores else None,
"request_count": n
}
# 全局监控实例
monitor = LLMMonitor()
# 在应用中使用
@traceable(name="Monitored LLM Call")
def monitored_llm_call(prompt: str) -> str:
start_time = time.time()
try:
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
latency_ms = (time.time() - start_time) * 1000
# 记录指标
monitor.record_request(
latency_ms=latency_ms,
cost_usd=calculate_cost(response.usage, "gpt-4o-mini"),
tokens=response.usage.total_tokens,
success=True
)
return response.choices[0].message.content
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
monitor.record_request(
latency_ms=latency_ms,
cost_usd=0,
tokens=0,
success=False
)
raise
9.3 Grafana集成
# 使用Prometheus指标暴露监控数据
from prometheus_client import Counter, Histogram, Gauge, start_http_server
# 定义指标
REQUEST_COUNT = Counter(
'llm_requests_total',
'Total LLM requests',
['model', 'status', 'endpoint']
)
REQUEST_LATENCY = Histogram(
'llm_request_duration_seconds',
'LLM request latency',
['model', 'endpoint'],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
TOKEN_USAGE = Counter(
'llm_tokens_total',
'Total tokens used',
['model', 'type'] # type: input/output
)
COST_TOTAL = Counter(
'llm_cost_dollars_total',
'Total cost in dollars',
['model']
)
ACTIVE_REQUESTS = Gauge(
'llm_active_requests',
'Number of active LLM requests'
)
# 在应用中使用
@traceable(name="Prometheus Monitored Call")
def prometheus_monitored_call(prompt: str, model: str = "gpt-4o-mini") -> str:
ACTIVE_REQUESTS.inc()
try:
with REQUEST_LATENCY.labels(model=model, endpoint="/chat").time():
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
# 记录指标
REQUEST_COUNT.labels(model=model, status="success", endpoint="/chat").inc()
TOKEN_USAGE.labels(model=model, type="input").inc(response.usage.prompt_tokens)
TOKEN_USAGE.labels(model=model, type="output").inc(response.usage.completion_tokens)
cost = calculate_cost(response.usage, model)
COST_TOTAL.labels(model=model).inc(cost)
return response.choices[0].message.content
except Exception as e:
REQUEST_COUNT.labels(model=model, status="error", endpoint="/chat").inc()
raise
finally:
ACTIVE_REQUESTS.dec()
# 启动Prometheus指标服务器
start_http_server(8000)
# Grafana Dashboard JSON 片段
{
"panels": [
{
"title": "LLM请求延迟分布",
"type": "heatmap",
"targets": [
{
"expr": "rate(llm_request_duration_seconds_bucket[5m])",
"legendFormat": "{{le}}"
}
]
},
{
"title": "Token使用趋势",
"type": "timeseries",
"targets": [
{
"expr": "rate(llm_tokens_total[5m])",
"legendFormat": "{{model}} - {{type}}"
}
]
},
{
"title": "累计成本",
"type": "stat",
"targets": [
{
"expr": "llm_cost_dollars_total",
"legendFormat": "{{model}}"
}
]
}
]
}
10. CI/CD集成
10.1 评估作为测试
# .github/workflows/llm-eval.yml
name: LLM Quality Gate
on:
pull_request:
paths:
- 'prompts/**'
- 'chains/**'
- 'eval/**'
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.10'
- name: Install dependencies
run: pip install -r requirements.txt
- name: Run evaluation
env:
LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python eval/run_evaluation.py
- name: Check quality gate
run: |
python -c "
import json
with open('eval/results.json') as f:
results = json.load(f)
# 质量门禁
min_score = 0.7
max_error_rate = 0.1
assert results['avg_score'] >= min_score, \
f'Score {results[\"avg_score\"]} below minimum {min_score}'
assert results['error_rate'] <= max_error_rate, \
f'Error rate {results[\"error_rate\"]} above maximum {max_error_rate}'
print('✅ Quality gate passed!')
"
- name: Comment on PR
if: always()
uses: actions/github-script@v7
with:
script: |
const fs = require('fs');
const results = JSON.parse(fs.readFileSync('eval/results.json', 'utf8'));
const body = `## LLM Evaluation Results
| Metric | Value | Threshold | Status |
|--------|-------|-----------|--------|
| Avg Score | ${results.avg_score.toFixed(3)} | ≥0.7 | ${results.avg_score >= 0.7 ? '✅' : '❌'} |
| Error Rate | ${(results.error_rate * 100).toFixed(1)}% | ≤10% | ${results.error_rate <= 0.1 ? '✅' : '❌'} |
| Avg Latency | ${results.avg_latency_ms.toFixed(0)}ms | ≤5000ms | ${results.avg_latency_ms <= 5000 ? '✅' : '❌'} |
[View full results in LangSmith](${results.langsmith_url})
`;
github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: body
});
10.2 评估脚本
# eval/run_evaluation.py
import json
import os
from langsmith import Client, evaluate
def run_evaluation():
client = Client()
# 定义评估目标
def target(inputs):
# 这里调用你的应用
from chains.customer_service import get_chain
chain = get_chain()
result = chain.invoke(inputs)
return {"output": result}
# 定义评估器
def accuracy_eval(run, example):
prediction = run.outputs.get("output", "")
expected = example.outputs.get("expected", "")
# 使用简单的关键词匹配作为基线
keywords = expected.lower().split()
matches = sum(1 for kw in keywords if kw in prediction.lower())
score = matches / len(keywords) if keywords else 0
return {"key": "accuracy", "score": score}
# 运行评估
results = evaluate(
target,
dataset="customer-service-eval",
evaluators=[accuracy_eval],
metadata={"pr": os.environ.get("GITHUB_PR_NUMBER", "local")}
)
# 保存结果
summary = {
"avg_score": results["accuracy"],
"error_rate": 0, # 从results中提取
"avg_latency_ms": 0, # 从results中提取
"langsmith_url": f"https://smith.langchain.com/projects/{results.get('project_id', '')}"
}
with open("eval/results.json", "w") as f:
json.dump(summary, f, indent=2)
print(f"评估完成: {summary}")
if __name__ == "__main__":
run_evaluation()
10.3 Prompt版本发布流程
# scripts/publish_prompt.py
"""提示词发布脚本 - 在CI/CD中使用"""
import sys
from langsmith import Client
def publish_prompt(prompt_name: str, version: int, environment: str = "production"):
client = Client()
# 1. 验证评估结果
# 获取最新实验结果
experiments = client.list_experiments(
dataset_name=f"{prompt_name}-eval"
)
latest = next(iter(experiments), None)
if not latest:
print("❌ No evaluation results found")
sys.exit(1)
# 2. 检查质量门禁
# ... 检查各项指标 ...
# 3. 标记版本为已发布
client.update_prompt_version(
prompt_name=prompt_name,
version=version,
metadata={
"environment": environment,
"released_at": datetime.now().isoformat(),
"evaluation_score": latest.results.get("avg_score", 0)
}
)
print(f"✅ Prompt '{prompt_name}' v{version} published to {environment}")
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--name", required=True)
parser.add_argument("--version", required=True, type=int)
parser.add_argument("--env", default="production")
args = parser.parse_args()
publish_prompt(args.name, args.version, args.env)
11. 企业级LLMOps平台搭建
11.1 平台架构设计
┌─────────────────────────────────────────────────────────────────┐
│ LLMOps Platform Architecture │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ API Gateway (Kong/Nginx) │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Trace │ │ Eval │ │ Prompt │ │ Cost │ │
│ │ Service │ │ Service │ │ Service │ │ Service │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Message Queue (Kafka/RabbitMQ) │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ClickHouse│ │ Postgres │ │ Redis │ │ S3 │ │
│ │(Traces) │ │(Metadata)│ │ (Cache) │ │(Artifacts)│ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Monitoring (Grafana + Prometheus) │ │
│ └─────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
11.2 数据存储设计
-- PostgreSQL: 元数据存储
CREATE TABLE projects (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(255) NOT NULL,
description TEXT,
created_at TIMESTAMP DEFAULT NOW(),
settings JSONB DEFAULT '{}'
);
CREATE TABLE prompts (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID REFERENCES projects(id),
name VARCHAR(255) NOT NULL,
description TEXT,
created_at TIMESTAMP DEFAULT NOW(),
UNIQUE(project_id, name)
);
CREATE TABLE prompt_versions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
prompt_id UUID REFERENCES prompts(id),
version INTEGER NOT NULL,
template TEXT NOT NULL,
config JSONB DEFAULT '{}',
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT NOW(),
UNIQUE(prompt_id, version)
);
CREATE TABLE evaluations (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID REFERENCES projects(id),
dataset_name VARCHAR(255),
prompt_version_id UUID REFERENCES prompt_versions(id),
model VARCHAR(100),
results JSONB,
created_at TIMESTAMP DEFAULT NOW()
);
-- ClickHouse: 高性能追踪存储
CREATE TABLE traces (
trace_id UUID,
run_id UUID,
parent_run_id Nullable(UUID),
project_id UUID,
name String,
run_type String, -- llm, chain, tool, retriever
inputs String, -- JSON
outputs String, -- JSON
metadata String, -- JSON
status String, -- success, error, pending
start_time DateTime64(3),
end_time Nullable(DateTime64(3)),
latency_ms Nullable(UInt32),
tokens_input Nullable(UInt32),
tokens_output Nullable(UInt32),
cost_usd Nullable(Float64),
error_message Nullable(String)
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(start_time)
ORDER BY (project_id, start_time, trace_id)
TTL start_time + INTERVAL 90 DAY;
-- 物化视图：自动聚合统计
CREATE MATERIALIZED VIEW trace_stats_hourly
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (project_id, hour, model)
AS SELECT
project_id,
toStartOfHour(start_time) AS hour,
metadata:model AS model,
count() AS request_count,
sum(tokens_input) AS total_input_tokens,
sum(tokens_output) AS total_output_tokens,
sum(cost_usd) AS total_cost,
avg(latency_ms) AS avg_latency,
quantile(0.95)(latency_ms) AS p95_latency
FROM traces
GROUP BY project_id, hour, model;
11.3 统一SDK封装
# llmops_sdk/client.py
"""企业级LLMOps统一SDK"""
from typing import Optional, Dict, Any, Callable
from contextlib import contextmanager
from dataclasses import dataclass
from abc import ABC, abstractmethod
import time
import uuid
@dataclass
class LLMCall:
"""LLM调用记录"""
trace_id: str
run_id: str
model: str
prompt: str
response: str
latency_ms: float
tokens_input: int
tokens_output: int
cost_usd: float
metadata: Dict[str, Any]
class LLMOpsBackend(ABC):
"""后端接口抽象"""
@abstractmethod
def log_trace(self, call: LLMCall) -> None:
pass
@abstractmethod
def get_prompt(self, name: str, version: Optional[int] = None) -> str:
pass
@abstractmethod
def log_evaluation(self, run_id: str, scores: Dict[str, float]) -> None:
pass
class LLMOpsClient:
"""统一LLMOps客户端"""
def __init__(
self,
backend: LLMOpsBackend,
project: str,
enable_tracing: bool = True,
enable_cost_tracking: bool = True
):
self.backend = backend
self.project = project
self.enable_tracing = enable_tracing
self.enable_cost_tracking = enable_cost_tracking
self._pricing = self._load_pricing()
def _load_pricing(self) -> Dict[str, Dict[str, float]]:
return {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
}
def track(self, func: Callable = None, *, name: str = None):
"""装饰器：自动追踪函数调用"""
def decorator(fn):
def wrapper(*args, **kwargs):
if not self.enable_tracing:
return fn(*args, **kwargs)
run_id = str(uuid.uuid4())
start = time.time()
try:
result = fn(*args, **kwargs)
latency = (time.time() - start) * 1000
# 记录调用
call = LLMCall(
trace_id=str(uuid.uuid4()),
run_id=run_id,
model=kwargs.get("model", "unknown"),
prompt=str(args),
response=str(result),
latency_ms=latency,
tokens_input=0,
tokens_output=0,
cost_usd=0,
metadata={"function": name or fn.__name__}
)
self.backend.log_trace(call)
return result
except Exception as e:
# 记录错误
raise
return wrapper
if func is not None:
return decorator(func)
return decorator
@contextmanager
def trace_context(self, name: str):
"""上下文管理器：追踪代码块"""
trace_id = str(uuid.uuid4())
start = time.time()
try:
yield {"trace_id": trace_id}
finally:
latency = (time.time() - start) * 1000
# 记录到后端
# 使用示例
class LangSmithBackend(LLMOpsBackend):
"""LangSmith后端实现"""
def __init__(self, api_key: str):
from langsmith import Client
self.client = Client(api_key=api_key)
def log_trace(self, call: LLMCall) -> None:
self.client.create_run(
name="LLM Call",
run_type="llm",
inputs={"prompt": call.prompt},
outputs={"response": call.response},
trace_id=call.trace_id,
run_id=call.run_id
)
def get_prompt(self, name: str, version: Optional[int] = None) -> str:
return self.client.pull_prompt(name, version=version)
def log_evaluation(self, run_id: str, scores: Dict[str, float]) -> None:
for key, value in scores.items():
self.client.create_feedback(
run_id=run_id,
key=key,
score=value
)
# 初始化
ops = LLMOpsClient(
backend=LangSmithBackend(api_key="xxx"),
project="production"
)
# 使用
@ops.track(name="Chat Service")
def chat(message: str, model: str = "gpt-4o-mini") -> str:
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": message}]
)
return response.choices[0].message.content
11.4 部署配置
# k8s/llmops-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: llmops-api
namespace: llmops
spec:
replicas: 3
selector:
matchLabels:
app: llmops-api
template:
metadata:
labels:
app: llmops-api
spec:
containers:
- name: llmops-api
image: your-registry/llmops-api:latest
ports:
- containerPort: 8000
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: llmops-secrets
key: database-url
- name: LANGCHAIN_API_KEY
valueFrom:
secretKeyRef:
name: llmops-secrets
key: langchain-api-key
resources:
requests:
memory: "512Mi"
cpu: "250m"
limits:
memory: "1Gi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: llmops-api
namespace: llmops
spec:
selector:
app: llmops-api
ports:
- port: 80
targetPort: 8000
type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llmops-api-hpa
namespace: llmops
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llmops-api
minReplicas: 2
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
12. 最佳实践总结
12.1 接入最佳实践
- 渐进式接入：从装饰器自动追踪开始，逐步添加自定义元数据
- 命名规范：建立统一的Trace/Run命名规范，便于搜索和过滤
- 元数据标准化：定义统一的元数据schema（模型、版本、环境等）
- 采样策略：生产环境可按比例采样（如10%），避免存储爆炸
- 敏感数据脱敏：在上报前对PII信息进行脱敏处理
12.2 评估最佳实践
- 评估先行：在修改提示词之前，先建立评估基线
- 多维度评估：同时使用自动指标和LLM-as-Judge
- 黄金数据集：维护一套高质量的评估数据集，定期更新
- 持续评估：将评估集成到CI/CD，每次变更自动验证
- 人工校准：定期用人工标注结果校准自动评估器
12.3 成本最佳实践
- 预算告警：设置日/周/月成本预算，超支自动告警
- 模型路由：根据任务复杂度自动选择合适成本的模型
- 语义缓存：对相似查询使用缓存，减少重复调用
- Token优化：精简系统提示词，使用更高效的格式
- 定期审计：每月分析成本分布，识别优化机会
12.4 监控最佳实践
- 分层告警：区分Warning和Critical级别，避免告警疲劳
- SLA定义：明确LLM应用的SLA（延迟、可用性、质量）
- 根因分析：建立从告警到根因的快速定位流程
- 仪表板标准化：为不同角色（研发/产品/运维）定制仪表板
- 演练机制：定期进行故障演练，验证告警和恢复流程
12.5 安全与合规
# PII脱敏示例
import re
from typing import str
class PIISanitizer:
"""个人信息脱敏工具"""
PATTERNS = {
"phone": (r'\b1[3-9]\d{9}\b', "***手机号***"),
"id_card": (r'\b\d{17}[\dXx]\b', "***身份证号***"),
"email": (r'\b[\w.-]+@[\w.-]+\.\w+\b', "***邮箱***"),
"bank_card": (r'\b\d{16,19}\b', "***银行卡号***"),
}
@classmethod
def sanitize(cls, text: str) -> str:
"""脱敏文本中的PII信息"""
result = text
for name, (pattern, replacement) in cls.PATTERNS.items():
result = re.sub(pattern, replacement, result)
return result
@classmethod
def sanitize_dict(cls, data: dict) -> dict:
"""递归脱敏字典中的PII信息"""
result = {}
for key, value in data.items():
if isinstance(value, str):
result[key] = cls.sanitize(value)
elif isinstance(value, dict):
result[key] = cls.sanitize_dict(value)
elif isinstance(value, list):
result[key] = [
cls.sanitize_dict(v) if isinstance(v, dict)
else cls.sanitize(v) if isinstance(v, str)
else v
for v in value
]
else:
result[key] = value
return result
# 在追踪中使用
@traceable(name="Secure Pipeline")
def secure_pipeline(user_input: str) -> str:
# 脱敏后再上报
sanitized_input = PIISanitizer.sanitize(user_input)
from langsmith.context import get_current_run_tree
run_tree = get_current_run_tree()
if run_tree:
run_tree.inputs = {"question": sanitized_input}
# ... 正常处理 ...
return response
参考资源
- LangSmith官方文档
- LangSmith Python SDK
- LangFuse官方文档
- LangFuse GitHub
- LangChain评估指南
- LLMOps最佳实践
- Prometheus监控
- Grafana可视化
本教程最后更新：2025年1月。LLM可观测性领域发展迅速，建议关注各工具的官方文档获取最新信息。
