监控告警、LangGraph 集成与 CI 回归测试

可观测的生产闭环不是「看日志」，而是「定义信号 → 设阈值 → 触发告警 → 回溯 Trace 根因 → 沉淀为回归用例」。监控负责发现异常，Trace 负责定位根因，CI 回归负责防止异常复发。三者缺一，可观测就退化成事后救火。

一、监控 Dashboard：四个生产核心信号

LangSmith 的 Monitor 视图按 project（也就是 tracing project）聚合 Run 级时序指标。打开任意 project 的 Monitor tab，默认就能看到随时间变化的折线/直方图。真正需要盯的生产信号收敛为四类，下表给出各信号的来源字段与典型告警动作。

前三个信号 LangSmith 自动从 Run 数据计算，无需额外代码。第四个「反馈」需要你主动上报——线上既可以接用户的赞/踩，也可以挂一个在线 LLM-as-judge 周期性打分。下面是把反馈写回某条 Run 的真实 API：

# pip install -U langsmith
from langsmith import Client

client = Client()  # 自动读取 LANGSMITH_API_KEY / LANGSMITH_ENDPOINT

# 假设你在主路径里拿到了某次调用的 run_id（见下文 @traceable / LangGraph 如何取得）
# 用户点了「踩」，或离线 judge 给出 0.3 分，都通过 create_feedback 写回
client.create_feedback(
    run_id="<your-run-id>",          # 目标 Run 的 UUID
    key="user_thumbs",               # 反馈维度名，Dashboard 按 key 聚合分布
    score=0,                          # 数值分（0/1 或 0~1 连续分）；Monitor 按 score 画分布
    value="down",                    # 可选的分类值（如 up/down）
    comment="答案与问题无关",        # 可选文本说明
)

# 想批量给反馈，遍历某个 project 最近的 Run 即可
for run in client.list_runs(project_name="my-prod-app", limit=20):
    # 这里可以接入你的在线 judge，对 run.outputs 打分后写回
    pass

配置基于阈值的告警

在 project 的 Monitor 页面，每个图表（latency、error rate、feedback 等）右上角可以 Create rule / Add alert。规则三要素：选指标 → 设比较算子与阈值 → 配通知渠道（Webhook / PagerDuty / Email）。告警触发后，点开规则能直接跳转到触发时间窗内的 Run 列表，从 Dashboard 一键下钻到 Trace 做根因分析。

二、LangGraph 集成：零埋点的分层 Trace

LangGraph 构建的是有状态的多节点图。只要设置了 tracing 环境变量，编译后的图在 invoke 时会自动产生 一个父 Run + 每个节点一个子 Run 的嵌套结构，无需任何手工埋点。这让你能在 Trace 视图里逐节点下钻，看到每一步的 state 输入输出、耗时与 token。

# 安装依赖：langgraph 自带 tracing 集成，langchain-openai 提供 LLM 节点
pip install -U langgraph langchain-openai langsmith

# 三个环境变量即开启自动 tracing（核心是前两个）
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="lsv2_pt_xxx"
export LANGSMITH_PROJECT="langgraph-demo"   # 不设则归到 default project
export OPENAI_API_KEY="sk-xxx"

# 一个最小的两节点 LangGraph：分类节点 -> 回答节点
# 运行后去 LangSmith 的 langgraph-demo project，会看到分层 Trace
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

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

# 1) 定义在节点间流转的状态
class State(TypedDict):
    question: str
    category: str
    answer: str

# 2) 节点一：给问题分类（每个节点会成为一个子 Run）
def classify(state: State) -> dict:
    resp = llm.invoke(
        f"用一个词分类这个问题（技术/闲聊/其他）：{state['question']}"
    )
    return {"category": resp.content.strip()}

# 3) 节点二：根据分类作答
def answer(state: State) -> dict:
    resp = llm.invoke(
        f"这是一个[{state['category']}]问题，请回答：{state['question']}"
    )
    return {"answer": resp.content}

# 4) 编译图：START -> classify -> answer -> END
builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("answer", answer)
builder.add_edge(START, "classify")
builder.add_edge("classify", "answer")
builder.add_edge("answer", END)
graph = builder.compile()

# 5) invoke 会自动上报：1 个父 Run（整图）+ classify/answer 两个子 Run + 内部 LLM 调用
result = graph.invoke({"question": "LangGraph 和 LangChain 有什么区别？"})
print(result["answer"])

三、CI 自动化回归：用 evaluate 守住评分基线

回归测试的本质是「锁住质量下界」。每次改提示词、换模型、调链路，都跑同一个 dataset 过同一组 evaluator，把聚合分数和基线对比；只要低于基线就 fail build。这把「凭感觉觉得没变差」变成「有数字证明没变差」——这是 verifiability_first 在 LLM 工程里的落地。

下面是一个能直接放进 tests/ 目录、被 pytest 收集的回归测试。它调用 evaluate 在指定 dataset 上跑目标函数，用内置/自定义 evaluator 打分，聚合平均分后用 assert 守住基线。

# tests/test_regression.py
# pip install -U langsmith pytest
import statistics
from langsmith import Client, evaluate

client = Client()
DATASET_NAME = "qa-regression-set"   # 事先在 LangSmith 创建好的评估数据集
SCORE_BASELINE = 0.8                  # 质量下界：平均 correctness 不得低于此值

# 1) 被测系统：把它包成一个接收 inputs(dict) 返回 outputs(dict) 的函数
def target(inputs: dict) -> dict:
    # 真实场景这里调用你的链路 / LangGraph 图
    question = inputs["question"]
    answer = f"针对「{question}」的回答"  # 占位，替换为真实推理
    return {"answer": answer}

# 2) 自定义 evaluator：返回 dict，含 key 与 score
#    run = 目标函数的输出 Run；example = dataset 中的标准样本
def correctness(run, example) -> dict:
    predicted = (run.outputs or {}).get("answer", "")
    expected = (example.outputs or {}).get("answer", "")
    score = 1.0 if expected and expected in predicted else 0.0
    return {"key": "correctness", "score": score}

def test_qa_regression():
    # 3) 在数据集上跑评估，evaluate 自动把每条结果上报到 LangSmith
    results = evaluate(
        target,
        data=DATASET_NAME,
        evaluators=[correctness],
        experiment_prefix="ci-regression",   # 实验前缀，便于在 UI 里区分每次 CI
        max_concurrency=4,
    )

    # 4) 聚合分数：遍历每条结果，取 correctness 分
    scores = []
    for r in results:
        for res in r["evaluation_results"]["results"]:
            if res.key == "correctness":
                scores.append(res.score)

    assert scores, "没有产生任何评分，检查 dataset 与 evaluator"
    avg = statistics.mean(scores)
    print(f"平均 correctness = {avg:.3f} (baseline={SCORE_BASELINE})")

    # 5) 守基线：低于下界直接 fail，CI 红灯
    assert avg >= SCORE_BASELINE, (
        f"质量回退！平均分 {avg:.3f} < 基线 {SCORE_BASELINE}"
    )

# .github/workflows/langsmith-regression.yml
# 在 PR 与 main 推送时跑回归，评分跌破基线则 fail
name: LangSmith Regression

on:
  pull_request:
  push:
    branches: [main]

jobs:
  regression:
    runs-on: ubuntu-latest
    env:
      # 关键：在 CI 里注入 LangSmith 与模型凭证（存到 repo Secrets）
      LANGSMITH_TRACING: "true"
      LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
      LANGSMITH_PROJECT: "ci-regression"
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install deps
        run: pip install -U langsmith pytest langchain-openai
      - name: Run regression eval
        # pytest 收集 test_qa_regression；assert 失败 -> 退出码非 0 -> build 红灯
        run: pytest tests/test_regression.py -s -v

四、生产踩坑汇总

瓶颈是验证，不是生成。能被度量的问题才能被完成；能消失的分支胜过能工作的分支。

— 本 Wiki 的工程审美