监控日志与可观测

「能跑」和「能维护」之间隔着一道鸿沟，跨过去的关键就是可观测性（Observability）——当线上出问题时，你能看到每一步的输入输出、每个节点的耗时、每次调用的 token 成本。本篇讲三件套：LangSmith 链路追踪、结构化日志、指标采集，并给一个把图「点亮」的完整示例。

一、可观测性三层模型

层次	工具	回答的问题
追踪（Tracing）	LangSmith / OpenTelemetry	「这次请求经过了哪些节点，每步输入输出是啥」
日志（Logging）	logging + JSON	「何时发生了什么，异常堆栈是啥」
指标（Metrics）	Prometheus + Grafana	「整体错误率、平均耗时、QPS 趋势如何」

三者互补：追踪看「单次请求」，日志看「事件细节」，指标看「全局趋势」。

mermaid

flowchart TB
    Req[请求] --> Graph[LangGraph 图]
    Graph -->|每步上报| LS[LangSmith Trace]
    Graph -->|logging.info| Log[JSON 日志文件]
    Graph -->|/metrics| Prom[Prometheus]
    LS --> Dashboard1[Trace 面板]
    Log --> Loki[Loki/ELK]
    Prom --> Grafana[Grafana 仪表盘]

二、LangSmith 集成（最省力的追踪）

LangSmith 是 LangChain 官方的可观测平台，对 LangGraph 几乎零侵入——配几个环境变量，图里每次 ainvoke、每个节点执行、每个 LLM 调用自动上报。

1. 配置

到 https://smith.langchain.com 注册，拿到 API Key，配 .env：

bash

LANGSMITH_API_KEY=lsv2_pt_xxxxx
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-langgraph-app
# 端点默认指向官方，国内网络若不通可不动（无官方国内端点）

python

# 代码里加载环境变量即可，无需任何额外代码
from dotenv import load_dotenv
load_dotenv()  # 加载 .env
# 之后正常用图，自动追踪

就这么简单

不需要在节点里手写 trace()。LangChain 的 ChatModel、Tool、Graph 都内置了追踪钩子，只要 LANGSMITH_TRACING=true 就全量上报。

2. 查看 Trace 面板

打开 smith.langchain.com → 你的 Project，能看到每次请求是一条 trace，展开后是树状结构：

text

/langgraph.invoke
├── chatbot 节点
│   └── ChatOpenAI.invoke
│       ├── Prompt (输入 token: 120)
│       └── Generation (输出 token: 45, 耗时 1.2s, 成本 $0.0008)
├── tools 节点
│   └── square 调用 (耗时 2ms)
└── chatbot 节点
    └── ChatOpenAI.invoke (输出 token: 20)

每个节点能看到：输入、输出、耗时、token 数、成本、错误。这正是排查「为什么这次回答不对」的利器。

3. 手动加追踪标签

给不同业务线打 tag，便于在 LangSmith 里过滤：

python

config = {
    "configurable": {"thread_id": thread_id},
    "metadata": {
        "user_id": "u123",
        "business": "客服",
        "version": "v1.2",
    },
}
await graph.ainvoke(inputs, config=config)

metadata 会出现在 trace 上，支持按字段筛选。

三、结构化日志

print 在生产是灾难——没法检索、没法关联请求。用 logging 输出 JSON 行日志，方便 ELK/Loki 采集。

1. 配置 JSON 日志

python

import logging
import json
import sys
from datetime import datetime, timezone

class JsonFormatter(logging.Formatter):
    """把日志格式化成 JSON 行"""
    def format(self, record: logging.LogRecord) -> str:
        log = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        # 自定义字段（通过 extra 传入）
        for key in ("thread_id", "node", "user_id", "trace_id"):
            val = getattr(record, key, None)
            if val is not None:
                log[key] = val
        if record.exc_info:
            log["exc"] = self.formatException(record.exc_info)
        return json.dumps(log, ensure_ascii=False)

def setup_logging():
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    root.addHandler(handler)

setup_logging()
logger = logging.getLogger("app")

2. 在图节点里打日志

python

def chatbot(state: State):
    logger.info("进入 chatbot 节点", extra={"node": "chatbot", "thread_id": state.get("thread_id", "")})
    try:
        resp = llm.invoke(state["messages"])
        logger.info("LLM 返回", extra={"node": "chatbot", "msg_len": len(resp.content)})
        return {"messages": [resp]}
    except Exception as e:
        logger.exception("LLM 调用失败", extra={"node": "chatbot"})
        raise

输出示例（每行一个 JSON）：

text

{"ts":"2026-08-06T10:00:00Z","level":"INFO","logger":"app","msg":"进入 chatbot 节点","node":"chatbot","thread_id":"abc"}
{"ts":"2026-08-06T10:00:01Z","level":"INFO","logger":"app","msg":"LLM 返回","node":"chatbot","msg_len":42}

敏感信息脱敏

千万别把 OPENAI_API_KEY、用户手机号、身份证打日志。对 state 里的敏感字段打日志前手动 mask：

python

safe_msg = msg.replace(phone, phone[:3] + "****" + phone[-4:])
logger.info("用户输入", extra={"input": safe_msg})

四、指标采集

1. 自定义指标中间件

用 prometheus_client 暴露节点耗时、token、错误率：

python

from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi import Response
import time

# 定义指标
REQUEST_COUNT = Counter("graph_requests_total", "总请求数", ["business"])
REQUEST_LATENCY = Histogram("graph_latency_seconds", "请求耗时", ["business"])
NODE_LATENCY = Histogram("graph_node_seconds", "节点耗时", ["node"])
TOKEN_COUNT = Counter("graph_tokens_total", "token 用量", ["type"])  # type=prompt/completion
ERROR_COUNT = Counter("graph_errors_total", "错误数", ["node"])

@app.get("/metrics")
async def metrics():
    """Prometheus 拉取端点"""
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

2. 在 invoke 外层包一层

python

@app.post("/invoke")
async def invoke(req: InvokeRequest):
    business = "default"
    REQUEST_COUNT.labels(business).inc()
    start = time.perf_counter()
    try:
        result = await _do_invoke(req)
        # 记录耗时
        REQUEST_LATENCY.labels(business).observe(time.perf_counter() - start)
        return result
    except Exception:
        ERROR_COUNT.labels("invoke").inc()
        raise

3. Prometheus + Grafana 对接

text

FastAPI :8000/metrics  ──scrape──▶  Prometheus :9090  ──query──▶  Grafana :3000

Prometheus 配置 scrape_configs 加 targets: ['host:8000']，每 15s 拉一次
Grafana 接 Prometheus 数据源，画 P50/P95 延迟、错误率、QPS 曲线

告警示例

Grafana 里设告警：rate(graph_errors_total[5m]) > 0.1 → 触发钉钉/飞书通知。

五、OpenTelemetry 简介

OpenTelemetry（OTel）是 CNCF 的可观测标准，跨语言跨后端。LangChain/LangGraph 已支持 OTel 导出，可以把 trace 同时发到 LangSmith、Jaeger、Datadog 等：

bash

pip install opentelemetry-sdk opentelemetry-exporter-otlp

python

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 初始化 OTel，导出到 Jaeger（本地 docker 起）
provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
)
trace.set_tracer_provider(provider)
# 此后 LangGraph 的 trace 也会发到 OTel 后端

适用场景：公司已有统一可观测平台（ELK + Jaeger），不想另开 LangSmith。Java 同学对 OTel 应该不陌生——和 Spring Boot Sleuth/OTel starter 一脉相承。

六、健康检查端点

部署到 K8s/容器平台必须提供探针端点：

python

from fastapi import FastAPI
import psutil

app = FastAPI()

@app.get("/health")
async def health():
    """存活探针：进程能响应就算活"""
    return {"status": "ok"}

@app.get("/ready")
async def ready():
    """就绪探针：依赖（DB、LLM）都通才算就绪"""
    checks = {"db": await _check_db(), "llm": await _check_llm()}
    ok = all(checks.values())
    return Response(
        status_code=200 if ok else 503,
        content=json.dumps({"ready": ok, "checks": checks}),
        media_type="application/json",
    )

K8s 配置：livenessProbe → /health，readinessProbe → /ready。

七、完整示例：给图加追踪 + 日志

把前面 ReAct 图改成带可观测的版本：

python

import logging, json, sys, os, time
from datetime import datetime, timezone
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi import FastAPI, Response
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from dotenv import load_dotenv

load_dotenv()  # 加载 LANGSMITH_* 环境变量

# ---- 日志 ----
class JsonFormatter(logging.Formatter):
    def format(self, record):
        log = {"ts": datetime.now(timezone.utc).isoformat(),
               "level": record.levelname, "msg": record.getMessage()}
        for k in ("node", "thread_id"):
            if hasattr(record, k): log[k] = getattr(record, k)
        return json.dumps(log, ensure_ascii=False)

h = logging.StreamHandler(sys.stdout)
h.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[h])
logger = logging.getLogger("app")

# ---- 指标 ----
NODE_LATENCY = Histogram("graph_node_seconds", "节点耗时", ["node"])
ERROR_COUNT = Counter("graph_errors_total", "错误数", ["node"])

# ---- 带观测的节点 ----
from graph import builder, State, llm

def chatbot(state: State):
    t = time.perf_counter()
    logger.info("进入 chatbot", extra={"node": "chatbot"})
    try:
        resp = llm.invoke(state["messages"])
        NODE_LATENCY.labels("chatbot").observe(time.perf_counter() - t)
        logger.info("chatbot 完成", extra={"node": "chatbot"})
        return {"messages": [resp]}
    except Exception as e:
        ERROR_COUNT.labels("chatbot").inc()
        logger.exception("chatbot 失败", extra={"node": "chatbot"})
        raise

# 重建带观测节点的图
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
import graph as g
b2 = StateGraph(State)
b2.add_node("chatbot", chatbot)
b2.add_node("tools", ToolNode(g.tools))
b2.add_edge(START, "chatbot")
b2.add_conditional_edges("chatbot", tools_condition, "tools")
b2.add_edge("tools", "chatbot")
b2.add_edge("chatbot", END)
graph = b2.compile(checkpointer=MemorySaver())

app = FastAPI()

@app.post("/invoke")
async def invoke(message: str):
    # LangSmith 自动追踪；这里只管业务
    result = await graph.ainvoke({"messages": [HumanMessage(content=message)]},
                                  config={"configurable": {"thread_id": "demo"}})
    return {"reply": result["messages"][-1].content}

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

跑一次请求后，去 LangSmith 看 trace、/metrics 看指标、stdout 看 JSON 日志——三处信息齐全。

八、常见踩坑

1. 敏感信息泄到 LangSmith 默认 state 全量上报，若 state 含密钥会被记录。用 LangSmith 的 metadata 过滤或在节点里手动剥离敏感字段后再 return。

2. 采样率与成本 LangSmith 按 trace 数收费，高 QPS 场景全量上报成本高。可在调用层加采样：

python

import random
if random.random() < 0.1:  # 10% 采样
    os.environ["LANGSMITH_TRACING"] = "true"
else:
    os.environ.pop("LANGSMITH_TRACING", None)

3. 日志重复输出logging.basicConfig 和手动 addHandler 都加了 handler，导致日志打两遍。统一用一种方式配置，或设 propagate=False。

4. Prometheus 指标冲突 多 worker（多进程）下 prometheus_client 默认指标在每进程独立计数，/metrics 只反映当前进程。生产用 prometheus_multiprocess_mode 或每 worker 独立端口 + Prometheus 服务发现。

5. Trace 看不到工具调用细节 工具函数没加 @tool 或返回值不可序列化，导致 LangSmith 拿不到输入输出。确保工具用 @tool 装饰、返回基础类型。

九、小结

LangSmith 零侵入追踪，配 LANGSMITH_TRACING=true 即自动上报每步
结构化 JSON 日志便于检索，务必脱敏敏感字段
Prometheus + Grafana 看全局趋势，OpenTelemetry 跨后端通用
健康检查端点是容器部署的标配
三层（追踪/日志/指标）配合，才能在出问题时「看得见、查得到、还原得出」

下一篇汇总生产环境最佳实践清单。

监控日志与可观测 ​

一、可观测性三层模型 ​

二、LangSmith 集成（最省力的追踪） ​

1. 配置 ​

2. 查看 Trace 面板 ​

3. 手动加追踪标签 ​

三、结构化日志 ​

1. 配置 JSON 日志 ​

2. 在图节点里打日志 ​

四、指标采集 ​

1. 自定义指标中间件 ​

2. 在 invoke 外层包一层 ​

3. Prometheus + Grafana 对接 ​

五、OpenTelemetry 简介 ​

六、健康检查端点 ​

七、完整示例：给图加追踪 + 日志 ​

八、常见踩坑 ​

九、小结 ​

监控日志与可观测

一、可观测性三层模型

二、LangSmith 集成（最省力的追踪）

1. 配置

2. 查看 Trace 面板

3. 手动加追踪标签

三、结构化日志

1. 配置 JSON 日志

2. 在图节点里打日志

四、指标采集

1. 自定义指标中间件

2. 在 invoke 外层包一层

3. Prometheus + Grafana 对接

五、OpenTelemetry 简介

六、健康检查端点

七、完整示例：给图加追踪 + 日志

八、常见踩坑

九、小结