20 | 实战——用 LangFuse + Ragas 构建 Agent 评估与监控体系

version: "3.8"
services:
  langfuse:
    image: langfuse/langfuse:2
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/langfuse
      - NEXTAUTH_SECRET=mysecret
      - SALT=mysalt
      - NEXTAUTH_URL=http://localhost:3000
      - TELEMETRY_ENABLED=false
    depends_on:
      - db

  db:
    image: postgres:16
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=langfuse
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

docker compose up -d

curl http://localhost:3000/api/public/health
# 返回 {"status": "OK"} 就说明跑起来了

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langfuse.callback import CallbackHandler

# 设置 LangFuse 环境变量
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxxx"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"

# 创建 LangFuse callback handler
langfuse_handler = CallbackHandler(
    session_id="rag-session-001",
    user_id="user-42",
    tags=["production", "rag-v2"]
)

# 构建一个简单的 RAG chain
prompt = ChatPromptTemplate.from_template("""
你是一个行业分析助手。根据以下参考资料回答用户问题。
如果参考资料中没有相关信息，请明确说明。

参考资料：
{context}

用户问题：{question}
""")

llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = prompt | llm | StrOutputParser()

# 调用时传入 callback
result = chain.invoke(
    {"context": "量子计算利用量子比特进行并行计算...", "question": "量子计算有哪些应用场景？"},
    config={"callbacks": [langfuse_handler]}
)
print(result)

from langfuse.decorators import observe, langfuse_context
import numpy as np

# 初始化
langfuse_context.configure(
    public_key="pk-lf-xxxx",
    secret_key="sk-lf-xxxx",
    host="http://localhost:3000"
)

@observe()
def retrieve_documents(query: str, top_k: int = 3) -> list:
    """模拟向量检索"""
    # 实际项目中这里会调用向量数据库
    documents = [
        {"text": "量子计算利用量子比特的叠加和纠缠特性...", "score": 0.92},
        {"text": "量子退火算法在组合优化问题上...", "score": 0.87},
        {"text": "IBM 在 2024 年发布了 1000+ 量子比特的处理器...", "score": 0.81},
    ]
    return documents[:top_k]

@observe()
def generate_answer(query: str, context: str) -> str:
    """调用 LLM 生成回答"""
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "根据参考资料回答问题。"},
            {"role": "user", "content": f"资料：{context}\n问题：{query}"}
        ],
        temperature=0
    )
    # 手动记录 token 使用量
    langfuse_context.update_current_observation(
        usage={
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
        }
    )
    return response.choices[0].message.content

@observe(name="rag-pipeline", as_type="span")
def rag_pipeline(query: str) -> dict:
    """完整的 RAG 流程"""
    docs = retrieve_documents(query)
    context = "\n".join([d["text"] for d in docs])
    answer = generate_answer(query, context)

    return {
        "query": query,
        "answer": answer,
        "contexts": [d["text"] for d in docs],
        "retrieval_scores": [d["score"] for d in docs],
    }

# 运行
result = rag_pipeline("量子计算在密码学领域有什么影响？")
langfuse_context.flush()  # 确保数据上报
print(result["answer"][:200])

from datasets import Dataset

eval_data = {
    "question": [
        "量子计算的基本原理是什么？",
        "量子计算在密码学领域有什么影响？",
        "量子纠错码为什么重要？",
        "量子计算和经典计算的本质区别是什么？",
        "目前量子计算的主要挑战有哪些？",
        "量子退火算法适用于什么类型的问题？",
        "IBM 和 Google在量子计算上的进展如何？",
        "量子计算对药物研发有什么潜在影响？",
        "量子互联网的概念是什么？",
        "量子计算什么时候能实现商业化？",
    ],
    "ground_truth": [
        "量子计算利用量子力学的叠加和纠缠特性，使量子比特可以同时处于多个状态，实现指数级的并行计算能力。",
        "量子计算的Shor算法可以在多项式时间内分解大整数，对当前基于RSA和ECC的公钥密码体系构成根本性威胁。后量子密码学正在开发抗量子攻击的替代方案。",
        "量子纠错码用于保护脆弱的量子信息免受退相干和噪声的影响。由于量子不可克隆定理，量子纠错比经典纠错更复杂，需要多个物理量子比特编码一个逻辑量子比特。",
        "经典计算使用确定性的0或1比特，量子计算使用可叠加的量子比特。量子计算在特定问题上（如整数分解、搜索、模拟量子系统）具有指数级或多项式级加速。",
        "主要挑战包括：量子退相干时间短、量子门错误率高、需要极低温运行环境、量子比特扩展困难、缺乏有效的量子纠错方案、编程工具不成熟。",
        "量子退火适用于组合优化问题，如旅行商问题、投资组合优化、蛋白质折叠等。它通过量子隧穿效应来寻找全局最优解，在特定问题上比经典模拟退火更高效。",
        "IBM在2024年发布了超过1000量子比特的Condor处理器，并推进模块化量子计算架构。Google在2023年实现了70量子比特的Sycamore处理器，并在量子纠错方面取得重要突破。",
        "量子计算可以精确模拟分子和化学反应的量子行为，加速药物分子的设计和优化。传统计算机模拟大分子时计算量指数增长，量子计算可以避免这一瓶颈。",
        "量子互联网利用量子纠缠和量子隐形传态实现安全的量子信息传输网络。它可以提供理论上不可破解的通信安全，并连接分布式量子计算资源。",
        "业界预计2027-2030年会出现具有实际商业价值的量子计算应用，但通用容错量子计算机可能需要到2035年之后。NISQ（含噪声中等规模量子）设备目前已在特定领域开始试用。",
    ],
}

dataset = Dataset.from_dict(eval_data)
print(f"评估数据集：{len(dataset)} 个测试用例")

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from ragas.llms import LangchainLLMWrapper
from langchain_openai import ChatOpenAI

# 包装 LLM
eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o", temperature=0))

# 让 RAG 系统回答所有测试问题
results = []
for item in eval_data["question"]:
    result = rag_pipeline(item)
    results.append(result)

# 组装评估输入
eval_input = {
    "question": eval_data["question"],
    "answer": [r["answer"] for r in results],
    "contexts": [r["contexts"] for r in results],
    "ground_truth": eval_data["ground_truth"],
}
eval_dataset = Dataset.from_dict(eval_input)

# 运行评估
scores = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=eval_llm,
)

# 打印结果
print(f"整体评分：")
print(f"  Faithfulness:       {scores['faithfulness']:.3f}")
print(f"  Answer Relevancy:   {scores['answer_relevancy']:.3f}")
print(f"  Context Precision:  {scores['context_precision']:.3f}")
print(f"  Context Recall:     {scores['context_recall']:.3f}")

整体评分：
  Faithfulness:       0.847
  Answer Relevancy:   0.912
  Context Precision:  0.783
  Context Recall:     0.825

import json
from datetime import datetime

baseline = {
    "version": "v1.0-baseline",
    "timestamp": datetime.now().isoformat(),
    "config": {
        "model": "gpt-4o",
        "temperature": 0,
        "top_k": 3,
        "chunk_size": 500,
    },
    "scores": {
        "faithfulness": scores["faithfulness"],
        "answer_relevancy": scores["answer_relevancy"],
        "context_precision": scores["context_precision"],
        "context_recall": scores["context_recall"],
    }
}

with open("eval_baselines.jsonl", "a") as f:
    f.write(json.dumps(baseline, ensure_ascii=False) + "\n")

print(f"Baseline 已保存: {baseline['version']}")

import asyncio
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    public_key="pk-lf-xxxx",
    secret_key="sk-lf-xxxx",
    host="http://localhost:3000"
)

async def compute_faithfulness(answer: str, contexts: list) -> float:
    """异步计算 Faithfulness 分数（简化版）"""
    from ragas.metrics import faithfulness as faithfulness_metric
    from ragas.llms import LangchainLLMWrapper
    from langchain_openai import ChatOpenAI

    eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini", temperature=0))

    # 构造单条评估数据
    single_eval = Dataset.from_dict({
        "answer": [answer],
        "contexts": [contexts],
    })

    score = await faithfulness_metric.ascore(
        row={"answer": answer, "contexts": contexts},
        llm=eval_llm,
    )
    return score

@observe(name="monitored-rag")
async def monitored_rag(query: str) -> dict:
    """带自动评估的 RAG pipeline"""
    # 正常的 RAG 流程
    docs = retrieve_documents(query)
    context = "\n".join([d["text"] for d in docs])
    answer = generate_answer(query, context)

    # 异步计算 Faithfulness
    faithfulness_score = await compute_faithfulness(answer, [d["text"] for d in docs])

    # 上报分数到 LangFuse
    langfuse.score(
        trace_id=langfuse_context.get_current_trace_id(),
        name="faithfulness",
        value=faithfulness_score,
    )

    return {
        "answer": answer,
        "faithfulness": faithfulness_score,
    }

import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class AlertRule:
    metric: str
    threshold: float
    direction: str  # "above" 或 "below"
    message: str

ALERT_RULES = [
    AlertRule("faithfulness", 0.7, "below", "Faithfulness 低于 0.7，可能存在幻觉"),
    AlertRule("cost_usd", 0.10, "above", "单次请求成本超过 $0.10"),
    AlertRule("latency_ms", 10000, "above", "请求延迟超过 10 秒"),
]

class AlertManager:
    def __init__(self):
        self.alerts = []

    def check(self, metrics: dict) -> list:
        triggered = []
        for rule in ALERT_RULES:
            value = metrics.get(rule.metric)
            if value is None:
                continue
            if rule.direction == "below" and value < rule.threshold:
                triggered.append(rule)
            elif rule.direction == "above" and value > rule.threshold:
                triggered.append(rule)

        for rule in triggered:
            alert = {
                "timestamp": time.time(),
                "metric": rule.metric,
                "value": metrics[rule.metric],
                "threshold": rule.threshold,
                "message": rule.message,
            }
            self.alerts.append(alert)
            self._send_notification(alert)

        return triggered

    def _send_notification(self, alert: dict):
        """发送告警通知（接入钉钉/飞书/邮件等）"""
        msg = (
            f"[Agent 告警] {alert['message']}\n"
            f"  指标: {alert['metric']} = {alert['value']:.3f}\n"
            f"  阈值: {alert['threshold']}\n"
            f"  时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
        )
        print(f"⚠ {msg}")
        # 实际项目中接入 webhook:
        # requests.post(WEBHOOK_URL, json={"text": msg})

alert_manager = AlertManager()

# 在请求结束后调用
def after_request(trace_data: dict):
    metrics = {
        "faithfulness": trace_data.get("faithfulness_score", 1.0),
        "cost_usd": trace_data.get("total_cost", 0),
        "latency_ms": trace_data.get("latency", 0),
    }
    triggered = alert_manager.check(metrics)
    if triggered:
        print(f"本次请求触发了 {len(triggered)} 条告警")

from langfuse import Langfuse

langfuse = Langfuse(
    public_key="pk-lf-xxxx",
    secret_key="sk-lf-xxxx",
    host="http://localhost:3000"
)

def export_high_quality_traces(
    min_faithfulness: float = 0.85,
    limit: int = 100
) -> list:
    """从 LangFuse 导出高 Faithfulness 分数的 Trace"""
    traces = langfuse.fetch_traces(
        limit=limit,
        order_by="timestamp",
    )

    high_quality = []
    for trace in traces.data:
        # 获取该 trace 的 faithfulness 分数
        scores = [s for s in trace.scores if s.name == "faithfulness"]
        if not scores:
            continue
        if scores[0].value < min_faithfulness:
            continue

        # 提取输入输出
        observations = langfuse.fetch_observations(trace_id=trace.id)
        # 找到 generate_answer 这一步的输出
        gen_obs = [o for o in observations.data if o.name == "generate_answer"]
        if not gen_obs:
            continue

        high_quality.append({
            "trace_id": trace.id,
            "input": trace.input,
            "output": gen_obs[0].output,
            "faithfulness": scores[0].value,
        })

    print(f"导出 {len(high_quality)} 条高质量 Trace (共 {len(traces.data)} 条)")
    return high_quality

traces = export_high_quality_traces(min_faithfulness=0.85, limit=200)

import json

def convert_to_sft_format(traces: list) -> list:
    """将 Trace 数据转成 SFT 训练格式（OpenAI 格式）"""
    sft_data = []
    for trace in traces:
        sft_data.append({
            "messages": [
                {
                    "role": "system",
                    "content": "你是一个专业的行业分析助手。根据提供的参考资料，准确、全面地回答用户问题。"
                },
                {
                    "role": "user",
                    "content": trace["input"]
                },
                {
                    "role": "assistant",
                    "content": trace["output"]
                }
            ]
        })
    return sft_data

sft_dataset = convert_to_sft_format(traces)

# 保存为 JSONL
with open("sft_dataset.jsonl", "w", encoding="utf-8") as f:
    for item in sft_dataset:
        f.write(json.dumps(item, ensure_ascii=False) + "\n")

print(f"SFT 数据集: {len(sft_dataset)} 条")
print(f"示例:\n{json.dumps(sft_dataset[0], ensure_ascii=False, indent=2)[:500]}")

Trace: monitored-rag (总耗时 2.34s, 成本 $0.014)
├── Span: retrieve_documents (耗时 0.18s)
│   ├── Input: "量子计算行业目前的竞争格局如何？"
│   └── Output: 3 篇文档 (relevance scores: 0.94, 0.88, 0.82)
├── Span: generate_answer (耗时 1.96s, tokens: 1380+420)
│   ├── Input: system prompt + context + question
│   └── Output: 420 字的分析回答
└── Score: faithfulness = 0.91

Stage 1（文章 01-04）: LLM 基础 + Prompt Engineering
  理解了 Transformer 架构、Prompt 设计原则、推理模型的区别、动手做了第一个项目

Stage 2（文章 05-08）: 单 Agent + Loop Engineering
  掌握了 Agent Loop 核心循环、Function Calling 机制、工具集成（MCP）、用 LangChain 构建完整 Agent

Stage 3（文章 09-12）: Context Engineering + RAG
  学会了上下文工程策略、RAG 基础与进阶、记忆系统、用 LlamaIndex 构建生产级 RAG

Stage 4（文章 13-16）: 多 Agent + Harness Engineering
  探索了多 Agent 协作架构、A2A 协议、Harness 编排、用 CrewAI 落地多 Agent 项目

Stage 5（文章 17-20）: 评估 + 安全 + 数据飞轮
  建立了系统化的评估体系、安全防护策略、可观测性基础设施，并用 LangFuse + Ragas 全部落地