21d - Langfuse 设置指南

┌─────────────────────────────────────────────────────────┐
│                    Langfuse v3 架构                       │
│                                                         │
│  ┌──────────────┐    ┌──────────────┐                   │
│  │  Langfuse Web │    │ Langfuse     │                   │
│  │  (UI + API)   │    │ Worker       │                   │
│  └──────┬───────┘    └──────┬───────┘                   │
│         │                   │                           │
│    ┌────▼───────────────────▼────┐                      │
│    │        Redis / Valkey       │  ← 队列 + 缓存       │
│    └────┬───────────────────┬───┘                       │
│         │                   │                           │
│  ┌──────▼───────┐   ┌──────▼───────┐  ┌─────────────┐  │
│  │  PostgreSQL   │   │  ClickHouse  │  │  S3 / Blob  │  │
│  │  (事务数据)   │   │  (分析数据)   │  │  (事件存储)  │  │
│  └──────────────┘   └──────────────┘  └─────────────┘  │
└─────────────────────────────────────────────────────────┘

# 验证 Docker 和 Docker Compose 版本
docker --version
docker compose version

# 克隆 Langfuse 仓库
git clone https://github.com/langfuse/langfuse.git
cd langfuse
 
# 生成加密密钥（必须）
export ENCRYPTION_KEY=$(openssl rand -hex 32)
echo "ENCRYPTION_KEY=$ENCRYPTION_KEY" >> .env
 
# 启动所有服务
docker compose up -d
 
# 查看服务状态
docker compose ps

# ===== 必需配置 =====
# 加密密钥（用于加密 API Key 等敏感数据）
ENCRYPTION_KEY=your_generated_hex_key_here

# 数据库连接
DATABASE_URL=postgresql://postgres:postgres@db:5432/postgres

# ClickHouse 连接
CLICKHOUSE_URL=http://clickhouse:8123
CLICKHOUSE_USER=default
CLICKHOUSE_PASSWORD=your_clickhouse_password

# Redis 连接
REDIS_CONNECTION_STRING=redis://redis:6379

# S3/MinIO 对象存储
LANGFUSE_S3_EVENT_UPLOAD_BUCKET=langfuse
LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT=http://minio:9000
LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID=minioadmin
LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY=minioadmin
LANGFUSE_S3_EVENT_UPLOAD_REGION=us-east-1
LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE=true

# ===== 可选配置 =====
# 应用 URL（生产环境必须设置）
NEXTAUTH_URL=https://langfuse.your-domain.com
NEXTAUTH_SECRET=$(openssl rand -base64 32)

# 邮件配置（用于邀请和通知）
SMTP_CONNECTION_URL=smtp://user:password@smtp.example.com:587
EMAIL_FROM_ADDRESS=langfuse@your-domain.com

# 禁用注册（生产环境推荐）
AUTH_DISABLE_SIGNUP=false

# 日志级别
LOG_LEVEL=info

# 检查所有容器是否正常运行
docker compose ps
 
# 查看 Web 容器日志
docker compose logs langfuse-web --tail 50
 
# 查看 Worker 容器日志
docker compose logs langfuse-worker --tail 50
 
# 健康检查
curl http://localhost:3000/api/public/health

你是一个 DevOps 工程师，请帮我生成 Langfuse 自托管部署的配置方案：

部署环境：[本地开发 / AWS EC2 / GCP GCE / Azure VM / Kubernetes]
团队规模：[1-3 人 / 5-10 人 / 10+ 人]
日均 trace 量：[1,000 / 10,000 / 100,000+]
数据保留期：[7 天 / 30 天 / 90 天 / 永久]
安全要求：[基础 / 需要 HTTPS / 需要 SSO / 需要网络隔离]

请输出：
1. 完整的 docker-compose.yml 配置
2. 环境变量配置文件 .env
3. Nginx 反向代理配置（如需 HTTPS）
4. 备份策略建议
5. 监控告警配置建议

# 添加 Langfuse Helm 仓库
helm repo add langfuse https://langfuse.github.io/langfuse-k8s
helm repo update

# values.yaml - Langfuse Kubernetes 部署配置
langfuse:
  web:
    replicas: 2
    resources:
      requests:
        cpu: "500m"
        memory: "1Gi"
      limits:
        cpu: "2"
        memory: "4Gi"
  worker:
    replicas: 2
    resources:
      requests:
        cpu: "500m"
        memory: "1Gi"
      limits:
        cpu: "2"
        memory: "4Gi"
 
# 使用内置 PostgreSQL（测试用）
postgresql:
  enabled: true
  auth:
    password: "your-secure-password"
 
# 使用内置 ClickHouse（测试用）
clickhouse:
  enabled: true
  auth:
    password: "your-clickhouse-password"
 
# 使用内置 Redis
redis:
  enabled: true
 
# 使用内置 MinIO（S3 兼容存储）
minio:
  enabled: true
 
# Ingress 配置
ingress:
  enabled: true
  className: nginx
  hosts:
    - host: langfuse.your-domain.com
      paths:
        - path: /
          pathType: Prefix
  tls:
    - secretName: langfuse-tls
      hosts:
        - langfuse.your-domain.com

# 创建命名空间
kubectl create namespace langfuse
 
# 创建加密密钥 Secret
kubectl create secret generic langfuse-secrets \
  --namespace langfuse \
  --from-literal=ENCRYPTION_KEY=$(openssl rand -hex 32) \
  --from-literal=NEXTAUTH_SECRET=$(openssl rand -base64 32)
 
# 安装 Langfuse
helm install langfuse langfuse/langfuse \
  --namespace langfuse \
  --values values.yaml
 
# 查看部署状态
kubectl get pods -n langfuse

# values-production.yaml
# 使用外部 PostgreSQL（如 AWS RDS、Cloud SQL）
postgresql:
  enabled: false
langfuse:
  extraEnv:
    - name: DATABASE_URL
      valueFrom:
        secretKeyRef:
          name: langfuse-db-secrets
          key: database-url
 
# 使用外部 ClickHouse（如 ClickHouse Cloud）
clickhouse:
  enabled: false
langfuse:
  extraEnv:
    - name: CLICKHOUSE_URL
      value: "https://your-instance.clickhouse.cloud:8443"
    - name: CLICKHOUSE_USER
      value: "default"
    - name: CLICKHOUSE_PASSWORD
      valueFrom:
        secretKeyRef:
          name: langfuse-ch-secrets
          key: password
 
# 使用外部 Redis（如 ElastiCache、Memorystore）
redis:
  enabled: false
langfuse:
  extraEnv:
    - name: REDIS_CONNECTION_STRING
      valueFrom:
        secretKeyRef:
          name: langfuse-redis-secrets
          key: connection-string

# 使用 pip（推荐 v3，基于 OpenTelemetry）
pip install langfuse
 
# 使用 poetry
poetry add langfuse
 
# 安装特定集成
pip install langfuse langchain langchain-openai  # LangChain 集成
pip install langfuse llama-index                  # LlamaIndex 集成

# .env 文件
LANGFUSE_PUBLIC_KEY=pk-lf-xxx        # 项目公钥
LANGFUSE_SECRET_KEY=sk-lf-xxx        # 项目密钥
LANGFUSE_HOST=https://cloud.langfuse.com  # 云版地址（自托管改为你的地址）

# 或在代码中配置
import os
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"  # 自托管地址

from langfuse.decorators import observe
from langfuse.openai import openai  # Langfuse 包装的 OpenAI 客户端
 
@observe()
def retrieve_context(query: str) -> str:
    """检索相关上下文——自动记录为子 span"""
    # 实际项目中调用向量数据库
    return "Langfuse 是一个开源 LLM 可观测性平台..."
 
@observe()
def generate_answer(query: str, context: str) -> str:
    """生成回答——OpenAI 调用自动追踪"""
    response = openai.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": f"基于以下上下文回答：\n{context}"},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content
 
@observe()
def rag_pipeline(query: str) -> str:
    """RAG 管线——自动创建完整的 trace 树"""
    context = retrieve_context(query)    # 子 span 1
    answer = generate_answer(query, context)  # 子 span 2
    return answer
 
# 执行——trace 自动上报到 Langfuse
result = rag_pipeline("什么是 Langfuse？")
print(result)

📊 Trace: rag_pipeline
├── 🔍 retrieve_context (耗时: 0.1s)
└── 🤖 generate_answer (耗时: 2.3s)
    └── 💬 ChatCompletion (gpt-4.1-mini, Token: 523)

from langfuse.decorators import observe, langfuse_context
 
@observe()
def complex_agent(query: str) -> str:
    # 获取当前观测对象，添加元数据
    langfuse_context.update_current_observation(
        metadata={"user_id": "user_123", "session_id": "sess_456"},
        tags=["production", "customer-service"]
    )
    
    # 手动记录自定义事件
    langfuse_context.update_current_trace(
        user_id="user_123",
        session_id="sess_456",
        metadata={"environment": "production"}
    )
    
    # 使用 start_as_current_observation 创建手动 span
    with langfuse_context.start_as_current_observation(
        name="custom-processing",
        metadata={"step": "preprocessing"}
    ):
        # 这里的代码会被记录在 custom-processing span 下
        processed_query = query.strip().lower()
    
    return generate_answer(processed_query, "context...")
 
# 手动记录分数（用户反馈）
langfuse_context.score_current_trace(
    name="user-feedback",
    value=1.0,
    comment="用户点击了👍"
)

# 方法 1：导入替换（推荐）
from langfuse.openai import openai
 
# 所有 openai 调用自动追踪
response = openai.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "你好"}],
    # Langfuse 特有参数
    langfuse_prompt=my_prompt,  # 关联 Prompt 版本
    name="greeting",            # 自定义 span 名称
    metadata={"feature": "chat"}
)
 
# 方法 2：包装现有客户端
from langfuse.openai import OpenAI
client = OpenAI()  # 自动追踪

# npm（v4 基于 OpenTelemetry JS v2）
npm install langfuse
 
# pnpm
pnpm add langfuse
 
# yarn
yarn add langfuse

LANGFUSE_PUBLIC_KEY=pk-lf-xxx
LANGFUSE_SECRET_KEY=sk-lf-xxx
LANGFUSE_BASEURL=https://cloud.langfuse.com

import { Langfuse } from "langfuse";
import { observeOpenAI } from "langfuse";
import OpenAI from "openai";
 
// 初始化 Langfuse
const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  baseUrl: process.env.LANGFUSE_BASEURL,
});
 
// 包装 OpenAI 客户端
const openai = observeOpenAI(new OpenAI(), {
  clientInitParams: {
    publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
    secretKey: process.env.LANGFUSE_SECRET_KEY!,
  },
});
 
// 所有 OpenAI 调用自动追踪
const response = await openai.chat.completions.create({
  model: "gpt-4.1-mini",
  messages: [
    { role: "system", content: "你是一个有帮助的助手。" },
    { role: "user", content: "什么是 Langfuse？" },
  ],
});
 
console.log(response.choices[0].message.content);

import { Langfuse } from "langfuse";
 
const langfuse = new Langfuse();
 
async function ragPipeline(query: string): Promise<string> {
  // 创建 trace
  const trace = langfuse.trace({
    name: "rag-pipeline",
    userId: "user_123",
    sessionId: "sess_456",
    metadata: { environment: "production" },
    tags: ["rag", "customer-service"],
  });
 
  // 创建检索 span
  const retrievalSpan = trace.span({
    name: "knowledge-retrieval",
    input: { query },
  });
  const context = await retrieveDocuments(query);
  retrievalSpan.end({ output: { context } });
 
  // 创建生成 span（记录为 generation 类型）
  const generation = trace.generation({
    name: "answer-generation",
    model: "gpt-4.1-mini",
    input: [
      { role: "system", content: `上下文：${context}` },
      { role: "user", content: query },
    ],
  });
 
  const answer = await callLLM(query, context);
 
  generation.end({
    output: answer,
    usage: { input: 150, output: 80, unit: "TOKENS" },
  });
 
  // 记录用户反馈
  trace.score({
    name: "helpfulness",
    value: 0.9,
    comment: "回答准确且有帮助",
  });
 
  return answer;
}

import { Langfuse } from "langfuse";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";
 
const langfuse = new Langfuse();
 
const trace = langfuse.trace({ name: "vercel-ai-chat" });
const generation = trace.generation({
  name: "stream-response",
  model: "gpt-4.1-mini",
});
 
const result = await streamText({
  model: openai("gpt-4.1-mini"),
  prompt: "用一句话解释 AgentOps",
});
 
generation.end({
  output: await result.text,
});
 
// 确保所有事件发送完毕
await langfuse.flushAsync();

# 安装依赖
# pip install langfuse opentelemetry-sdk
 
from langfuse import get_client
 
# Langfuse SDK v3 自动基于 OTel 运行
# 只需配置环境变量即可
import os
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"
 
# @observe 装饰器内部使用 OTel span
from langfuse.decorators import observe
 
@observe()
def my_function():
    # 自动创建 OTel span 并转换为 Langfuse observation
    pass

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 
# 配置 Langfuse 作为 OTel 后端
exporter = OTLPSpanExporter(
    endpoint="http://localhost:3000/api/public/otel/v1/traces",
    headers={
        "Authorization": "Basic <base64(public_key:secret_key)>"
    }
)
 
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
 
# 使用标准 OTel API 创建 span
tracer = trace.get_tracer("my-app")
with tracer.start_as_current_span("my-operation") as span:
    span.set_attribute("langfuse.type", "generation")
    span.set_attribute("gen_ai.usage.input_tokens", 150)
    span.set_attribute("gen_ai.usage.output_tokens", 80)
    # 你的业务逻辑
    pass

// TypeScript SDK v4 同样基于 OTel JS v2
import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseExporter } from "langfuse";
 
const sdk = new NodeSDK({
  traceExporter: new LangfuseExporter({
    publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
    secretKey: process.env.LANGFUSE_SECRET_KEY!,
    baseUrl: process.env.LANGFUSE_BASEURL,
  }),
});
 
sdk.start();

1. 登录 Langfuse UI → 选择项目
2. 进入 "Dashboard" 页面
3. 查看以下核心成本指标：
   - 总成本趋势（按天/周/月）
   - 按模型的成本分布
   - 按用户的成本分布
   - 按 trace 名称的成本分布
   - 平均每次交互成本

from langfuse.decorators import observe, langfuse_context
 
@observe(as_type="generation")
def custom_model_call(prompt: str) -> str:
    # 调用自定义模型
    result = call_my_model(prompt)
    
    # 手动记录用量
    langfuse_context.update_current_observation(
        model="my-custom-model-v2",
        usage={
            "input": 250,       # 输入 token 数
            "output": 120,      # 输出 token 数
            "unit": "TOKENS",
            "input_cost": 0.0025,   # 输入成本（美元）
            "output_cost": 0.0036,  # 输出成本（美元）
            "total_cost": 0.0061    # 总成本（美元）
        }
    )
    return result

// TypeScript 版本
const generation = trace.generation({
  name: "custom-model-call",
  model: "my-custom-model-v2",
  input: prompt,
});
 
const result = await callMyModel(prompt);
 
generation.end({
  output: result,
  usage: {
    input: 250,
    output: 120,
    unit: "TOKENS",
    inputCost: 0.0025,
    outputCost: 0.0036,
    totalCost: 0.0061,
  },
});

1. 进入 Settings → Model Prices
2. 点击 "Add Model Price"
3. 配置：
   - Model Name: my-custom-model-v2
   - Match Pattern: my-custom-model-*（支持通配符）
   - Input Price: $0.01 / 1K tokens
   - Output Price: $0.03 / 1K tokens
4. 保存——后续所有匹配的 generation 自动计算成本

from langfuse import Langfuse
 
langfuse = Langfuse()
 
# 获取项目的每日成本统计
# 通过 Langfuse UI 的 Dashboard 查看更直观
# 或使用 API 导出数据进行自定义分析
 
# 按用户统计成本（通过 trace 的 user_id 字段）
traces = langfuse.get_traces(
    user_id="user_123",
    limit=100
)
 
total_cost = sum(
    t.total_cost for t in traces.data 
    if t.total_cost is not None
)
print(f"用户 user_123 的总成本: ${total_cost:.4f}")

你是一个 AI 成本分析师。请根据以下 Langfuse 成本数据，生成优化建议：

## 当前成本数据
- 日均 LLM 调用次数：[数量]
- 月度总成本：$[金额]
- 主要模型使用分布：[GPT-4o: X%, GPT-4o-mini: Y%, Claude 3.5: Z%]
- 平均每次交互 Token 数：输入 [N] / 输出 [M]
- 成本最高的 trace 类型：[类型名称]

## 请分析
1. 成本异常点识别（哪些调用成本不合理？）
2. 模型降级建议（哪些场景可以用更便宜的模型？）
3. Prompt 优化建议（如何减少 Token 消耗？）
4. 缓存策略建议（哪些查询适合缓存？）
5. 预计优化后的月度成本

1. 左侧菜单选择 "Prompts"
2. 点击 "New Prompt"
3. 选择类型：Text（纯文本）或 Chat（消息数组）
4. 输入名称（如 "customer-service-v1"）
5. 编写 Prompt 内容，使用 {{变量名}} 标记占位符
6. 配置模型参数（temperature、max_tokens 等）
7. 添加标签（如 "production"、"staging"）
8. 点击 "Save" 保存

from langfuse import Langfuse
 
langfuse = Langfuse()
 
# 获取最新版本的 Prompt
prompt = langfuse.get_prompt("customer-service-v1")
 
# 获取特定标签的 Prompt（如生产版本）
prompt = langfuse.get_prompt("customer-service-v1", label="production")
 
# 获取特定版本
prompt = langfuse.get_prompt("customer-service-v1", version=3)
 
# 编译 Prompt（替换变量）
compiled = prompt.compile(
    role="客服专家",
    language="中文",
    knowledge_base="电商退货政策..."
)
 
# 在 LLM 调用中使用，并关联 Prompt 版本
from langfuse.openai import openai
 
response = openai.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": compiled}],
    langfuse_prompt=prompt  # 关联 Prompt 版本到 trace
)

// TypeScript 版本
import { Langfuse } from "langfuse";
 
const langfuse = new Langfuse();
 
// 获取 Chat 类型的 Prompt
const prompt = await langfuse.getPrompt("customer-service-v1", undefined, {
  label: "production",
});
 
// 编译并使用
const messages = prompt.compile({
  role: "客服专家",
  language: "中文",
});
 
// 在 generation 中关联 Prompt
const generation = trace.generation({
  name: "customer-service",
  prompt: prompt,  // 自动关联版本
  input: messages,
});

推荐的 Prompt 版本管理流程：

1. 开发阶段：
   - 在 Playground 中迭代测试 Prompt
   - 每次修改自动创建新版本（v1, v2, v3...）
   
2. 测试阶段：
   - 给候选版本添加 "staging" 标签
   - 在测试环境中使用 label="staging" 获取
   - 运行评估数据集验证质量
   
3. 发布阶段：
   - 验证通过后，将标签改为 "production"
   - 生产环境使用 label="production" 获取
   - SDK 自动缓存，零停机切换
   
4. 回滚：
   - 将 "production" 标签移回上一个版本
   - SDK 后台刷新缓存，自动回滚

from langfuse.decorators import observe, langfuse_context
 
@observe()
def chatbot(query: str) -> str:
    answer = generate_answer(query)
    
    # 记录用户反馈分数
    langfuse_context.score_current_trace(
        name="user-feedback",
        value=1,           # 1 = 正面，0 = 负面
        comment="用户点击了👍"
    )
    
    # 记录多维度评分
    langfuse_context.score_current_trace(
        name="helpfulness",
        value=0.8,
        comment="回答基本有帮助但缺少细节"
    )
    langfuse_context.score_current_trace(
        name="accuracy",
        value=1.0,
        comment="事实准确"
    )
    
    return answer

from langfuse import Langfuse
 
langfuse = Langfuse()
 
# 创建评估数据集
dataset = langfuse.create_dataset(
    name="customer-service-eval",
    description="客服机器人评估数据集"
)
 
# 添加评估样本
langfuse.create_dataset_item(
    dataset_name="customer-service-eval",
    input={"question": "如何退款？"},
    expected_output={"answer": "您可以在订单页面点击'申请退款'按钮，7天内无理由退货。"}
)
 
langfuse.create_dataset_item(
    dataset_name="customer-service-eval",
    input={"question": "配送需要多久？"},
    expected_output={"answer": "标准配送3-5个工作日，加急配送1-2个工作日。"}
)
 
# 运行评估
dataset = langfuse.get_dataset("customer-service-eval")
 
for item in dataset.items:
    # 调用你的 Agent
    with item.observe(run_name="eval-v2.1") as trace_id:
        result = my_agent(item.input["question"])
    
    # 计算评分
    expected = item.expected_output["answer"]
    # 简单的关键词匹配评估（实际项目中用 LLM-as-Judge）
    score = 1.0 if any(kw in result for kw in ["退款", "退货", "配送", "工作日"]) else 0.0
    
    # 记录评分
    langfuse.score(
        trace_id=trace_id,
        name="keyword-match",
        value=score
    )

from langfuse.decorators import observe
from langfuse.openai import openai
 
@observe()
def llm_judge_evaluate(question: str, expected: str, actual: str) -> float:
    """使用 LLM 作为评判者"""
    response = openai.chat.completions.create(
        model="gpt-4.1-mini",
        temperature=0,
        messages=[{
            "role": "user",
            "content": f"""请评估以下回答的质量（返回 0-1 之间的数字）：
 
问题：{question}
参考答案：{expected}
实际回答：{actual}
 
评分标准：
- 1.0：完全正确且有帮助
- 0.7：基本正确但有遗漏
- 0.3：部分正确但有误导
- 0.0：完全错误
 
请只返回一个数字。"""
        }]
    )
    return float(response.choices[0].message.content.strip())
 
# 在评估循环中使用
for item in dataset.items:
    with item.observe(run_name="eval-llm-judge") as trace_id:
        result = my_agent(item.input["question"])
        score = llm_judge_evaluate(
            item.input["question"],
            item.expected_output["answer"],
            result
        )
        langfuse.score(trace_id=trace_id, name="llm-judge", value=score)

1. 进入 "Datasets" → 选择数据集
2. 查看各 run 的评分对比
3. 点击具体 run 查看每个样本的详细评分
4. 对比不同版本（如 v2.0 vs v2.1）的评分趋势
5. 导出评估报告

# 安装
# pip install langfuse langchain langchain-openai
 
import os
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"
 
from langfuse.langchain import CallbackHandler
 
# 创建 Langfuse 回调处理器
langfuse_handler = CallbackHandler()
 
# 在 LangChain 调用中使用
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
llm = ChatOpenAI(model="gpt-4.1-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    ("user", "{input}")
])
chain = prompt | llm
 
# 传入回调——自动追踪整个 chain
response = chain.invoke(
    {"input": "什么是 AgentOps？"},
    config={"callbacks": [langfuse_handler]}
)

from langgraph.graph import StateGraph, START, END
from langfuse.langchain import CallbackHandler
 
langfuse_handler = CallbackHandler()
 
# 构建 LangGraph（省略节点定义）
app = graph.compile()
 
# 执行时传入回调——每个节点自动记录为子 span
result = app.invoke(
    {"messages": ["分析 AgentOps 趋势"]},
    config={"callbacks": [langfuse_handler]}
)

# 安装
# pip install langfuse llama-index
 
from langfuse.llama_index import LlamaIndexInstrumentor
 
# 初始化插桩器
instrumentor = LlamaIndexInstrumentor()
instrumentor.start()
 
# 所有 LlamaIndex 操作自动追踪
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
 
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
 
# 这个查询的完整链路（检索 + 生成）自动记录到 Langfuse
response = query_engine.query("什么是 RAG？")
 
# 停止插桩
instrumentor.stop()

# Python：使用 Langfuse 包装的 OpenAI
from langfuse.openai import openai
 
# 所有调用自动追踪，包括：
# - Token 用量
# - 延迟
# - 流式响应的首 Token 时间
# - API 错误
# - 模型参数
 
# 普通调用
response = openai.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "你好"}]
)
 
# 流式调用（自动追踪 time-to-first-token）
stream = openai.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "写一首诗"}],
    stream=True
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")

// TypeScript：使用 observeOpenAI 包装
import { observeOpenAI } from "langfuse";
import OpenAI from "openai";
 
const openai = observeOpenAI(new OpenAI());
 
const response = await openai.chat.completions.create({
  model: "gpt-4.1-mini",
  messages: [{ role: "user", content: "你好" }],
});

from langfuse.decorators import observe
from anthropic import Anthropic
 
client = Anthropic()
 
@observe(as_type="generation")
def call_claude(prompt: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    
    # 手动记录用量（Anthropic SDK 不自动捕获）
    from langfuse.decorators import langfuse_context
    langfuse_context.update_current_observation(
        model="claude-sonnet-4-20250514",
        usage={
            "input": response.usage.input_tokens,
            "output": response.usage.output_tokens,
            "unit": "TOKENS"
        }
    )
    
    return response.content[0].text

# 生产环境数据保留策略
# 在 Langfuse 配置中设置
 
# ClickHouse 数据保留（traces、observations、scores）
# 通过 ClickHouse TTL 配置
# 建议：生产环境保留 90 天，归档到 S3
 
# PostgreSQL 备份（用户、项目、Prompt 等事务数据）
# 使用 pg_dump 定期备份
# 建议：每日全量备份 + 持续 WAL 归档

#!/bin/bash
# langfuse-backup.sh
 
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_DIR="/backups/langfuse"
 
# PostgreSQL 备份
pg_dump -h localhost -U postgres -d langfuse \
  | gzip > "$BACKUP_DIR/pg_$DATE.sql.gz"
 
# ClickHouse 备份（使用 clickhouse-backup 工具）
clickhouse-backup create "backup_$DATE"
clickhouse-backup upload "backup_$DATE"
 
# S3 事件数据已自动持久化，无需额外备份
 
# 清理 30 天前的本地备份
find "$BACKUP_DIR" -name "*.gz" -mtime +30 -delete
 
echo "备份完成: $DATE"

# Kubernetes HPA 配置示例
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: langfuse-web-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: langfuse-web
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

# 生产环境安全配置

# 禁用公开注册
AUTH_DISABLE_SIGNUP=true

# 强制 HTTPS
NEXTAUTH_URL=https://langfuse.your-domain.com

# 配置 SSO（Enterprise 功能）
AUTH_GOOGLE_CLIENT_ID=xxx
AUTH_GOOGLE_CLIENT_SECRET=xxx

# 或 OIDC 通用配置
AUTH_CUSTOM_CLIENT_ID=xxx
AUTH_CUSTOM_CLIENT_SECRET=xxx
AUTH_CUSTOM_ISSUER=https://your-idp.com

# API 速率限制
LANGFUSE_RATE_LIMIT_ENABLED=true

# 加密密钥（务必安全存储）
ENCRYPTION_KEY=your_hex_key_stored_in_vault

import os
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from langfuse.openai import openai
 
# ===== 第 1 步：配置 Langfuse（指向自托管实例）=====
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "https://langfuse.internal.company.com"
 
langfuse = Langfuse()
 
# ===== 第 2 步：创建 Prompt 并管理版本 =====
# 在 UI 中创建 Prompt "customer-service-v1"，内容如下：
# 你是{{company_name}}的客服助手。
# 基于以下知识库内容回答用户问题：
# {{knowledge_base}}
# 要求：准确、简洁、友好。如果不确定，请说"我需要转接人工客服"。
 
# ===== 第 3 步：带追踪的 Agent 实现 =====
@observe()
def search_knowledge_base(query: str) -> str:
    """知识库检索——自动记录为子 span"""
    # 实际项目中调用向量数据库
    knowledge = {
        "退货": "7天无理由退货政策，请在订单页面点击'申请退款'...",
        "发票": "支持电子发票，订单完成后在订单详情页申请...",
        "会员": "会员享受9折优惠和免运费服务..."
    }
    for key, value in knowledge.items():
        if key in query:
            return value
    return "未找到相关信息，建议联系人工客服。"
 
@observe()
def customer_service_agent(query: str, user_id: str, session_id: str) -> str:
    """客服 Agent 主流程"""
    # 设置 trace 级别的元数据
    langfuse_context.update_current_trace(
        user_id=user_id,
        session_id=session_id,
        tags=["production", "customer-service"],
        metadata={"channel": "web-chat"}
    )
    
    # 获取生产版本的 Prompt
    prompt = langfuse.get_prompt("customer-service-v1", label="production")
    
    # 检索知识库
    context = search_knowledge_base(query)
    
    # 编译 Prompt
    system_message = prompt.compile(
        company_name="示例科技",
        knowledge_base=context
    )
    
    # 调用 LLM（自动追踪 Token、延迟、成本）
    response = openai.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": query}
        ],
        langfuse_prompt=prompt  # 关联 Prompt 版本
    )
    
    answer = response.choices[0].message.content
    return answer
 
# ===== 第 4 步：创建评估数据集 =====
dataset = langfuse.create_dataset(
    name="cs-golden-set-v1",
    description="客服机器人黄金测试集"
)
 
test_cases = [
    {"input": {"q": "怎么退货？"}, "expected": {"a": "7天无理由退货"}},
    {"input": {"q": "发票怎么开？"}, "expected": {"a": "电子发票"}},
    {"input": {"q": "会员有什么优惠？"}, "expected": {"a": "9折优惠"}},
    {"input": {"q": "你们公司在哪？"}, "expected": {"a": "人工客服"}},
]
 
for tc in test_cases:
    langfuse.create_dataset_item(
        dataset_name="cs-golden-set-v1",
        input=tc["input"],
        expected_output=tc["expected"]
    )
 
# ===== 第 5 步：运行评估 =====
dataset = langfuse.get_dataset("cs-golden-set-v1")
 
for item in dataset.items:
    with item.observe(run_name="v1.0-gpt4mini") as trace_id:
        result = customer_service_agent(
            item.input["q"], 
            user_id="eval-user",
            session_id="eval-session"
        )
    
    # 关键词匹配评分
    expected_keywords = item.expected_output["a"]
    score = 1.0 if expected_keywords in result else 0.0
    langfuse.score(trace_id=trace_id, name="keyword-match", value=score)
 
print("评估完成，请在 Langfuse UI 查看结果")