7.2 可观测性（Observability）

本节介绍如何使用 LangSmith 实现 LangChain 应用的可观测性监控。

---

可观测性概述

LangChain Agent 的行为具有非确定性，需要深入了解其执行过程。LangSmith 提供了专为 LLM 应用设计的可观测性平台，支持：

• 追踪（Tracing）：记录每个执行步骤
• 调试（Debugging）：分析工具调用和决策点
• 评估（Evaluation）：评估不同输入的性能
• 监控（Monitoring）：追踪生产环境的使用模式

---

核心概念

Trace（追踪）

Trace 是应用从输入到输出的完整执行序列，记录了：

• 用户输入
• 模型交互
• 工具调用
• 决策点
• 最终响应

Run（运行）

Run 是 Trace 中的单个执行步骤，可以是：

• LLM 调用
• Chain 执行
• Tool 调用
• Retriever 查询

Project（项目）

Project 是组织 Trace 的逻辑分组，便于：

• 按环境分组（开发/测试/生产）
• 按功能分组（聊天/搜索/分析）
• 按版本分组（v1.0/v2.0）

---

快速开始

1. 获取 API Key

1. 访问 LangSmith
2. 创建免费账户
3. 在 Settings 中生成 API Key

2. 配置环境变量

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=<your-api-key>

3. 运行应用

无需修改代码，追踪自动启用：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 创建 Chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
model = ChatOpenAI(model="gpt-4o")
chain = prompt | model | StrOutputParser()

# 自动追踪
result = chain.invoke({"question": "What is LangChain?"})

追踪结果将自动显示在 LangSmith 控制台。

---

追踪功能详解

自动追踪

使用 create_agent() 创建的 Agent 自动支持追踪：

from langchain.agents import create_agent

agent = create_agent(
    "gpt-4o",
    tools=[search_tool, calculator_tool]
)

# 自动追踪所有步骤
response = agent.invoke({
    "messages": [{"role": "user", "content": "What is 25 * 4?"}]
})

选择性追踪

使用 tracing_context 控制追踪范围：

import langsmith as ls

# 启用追踪
with ls.tracing_context(enabled=True):
    response = agent.invoke({
        "messages": [{"role": "user", "content": "Hello"}]
    })

# 禁用追踪（用于敏感操作）
with ls.tracing_context(enabled=False):
    response = agent.invoke({
        "messages": [{"role": "user", "content": "Private data"}]
    })

---

项目配置

静态配置

通过环境变量设置默认项目：

export LANGSMITH_PROJECT=my-agent-project

动态配置

在代码中指定项目：

import langsmith as ls

# 为特定操作指定项目
with ls.tracing_context(project_name="email-agent-test", enabled=True):
    response = agent.invoke({
        "messages": [{"role": "user", "content": "Send email"}]
    })

多项目管理

# 开发环境
with ls.tracing_context(project_name="agent-dev"):
    dev_response = agent.invoke(...)

# 测试环境
with ls.tracing_context(project_name="agent-test"):
    test_response = agent.invoke(...)

# 生产环境
with ls.tracing_context(project_name="agent-prod"):
    prod_response = agent.invoke(...)

---

元数据和标签

为追踪添加自定义注释，便于筛选和分析：

添加标签

response = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}]},
    config={
        "tags": ["production", "v1.0", "high-priority"]
    }
)

添加元数据

response = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}]},
    config={
        "metadata": {
            "user_id": "user_123",
            "session_id": "session_456",
            "request_id": "req_789",
            "source": "web_app",
            "version": "1.0.0"
        }
    }
)

组合使用

response = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}]},
    config={
        "tags": ["production", "chatbot"],
        "metadata": {
            "user_id": "user_123",
            "session_id": "session_456",
            "feature_flags": {
                "new_model": True,
                "streaming": False
            }
        }
    }
)

---

监控仪表板

关键指标

LangSmith 提供以下监控指标：

指标	说明
延迟	端到端响应时间
Token 使用	输入/输出 Token 数量
成功率	请求成功百分比
错误率	错误类型分布
成本	估算的 API 成本

筛选和搜索

# 在 LangSmith UI 中可以：
# 1. 按标签筛选
# 2. 按元数据筛选
# 3. 按时间范围筛选
# 4. 按项目筛选
# 5. 按状态筛选（成功/失败）

---

洞察分析

性能分析

• 延迟分布：识别慢请求
• Token 趋势：追踪使用量变化
• 错误模式：发现常见失败原因

对比分析

在 LangSmith 中可以：

1. 选择多个 Trace
2. 并排对比执行步骤
3. 识别差异和问题

---

自定义追踪

使用 @traceable 装饰器

from langsmith import traceable

@traceable
def process_document(doc: str) -> dict:
    """处理文档的自定义函数"""
    # 你的逻辑
    result = {"processed": True, "length": len(doc)}
    return result

@traceable
def analyze_sentiment(text: str) -> str:
    """情感分析"""
    # 调用模型
    response = model.invoke(f"Analyze sentiment: {text}")
    return response.content

手动创建 Run

import langsmith as ls

# 手动追踪
with ls.trace("custom-operation") as run:
    # 你的逻辑
    result = do_something()

    # 记录输出
    run.end(outputs={"result": result})

嵌套追踪

from langsmith import traceable

@traceable
def main_pipeline(input_data):
    """主管道"""
    # Step 1: 预处理
    processed = preprocess(input_data)

    # Step 2: 分析
    analysis = analyze(processed)

    # Step 3: 生成
    output = generate(analysis)

    return output

@traceable
def preprocess(data):
    """预处理步骤"""
    return data.strip().lower()

@traceable
def analyze(data):
    """分析步骤"""
    return {"word_count": len(data.split())}

@traceable
def generate(analysis):
    """生成步骤"""
    return f"Analysis complete: {analysis}"

---

与 Agent 集成

Agent 中间件追踪

from langchain.agents import AgentMiddleware, create_agent

class ObservabilityMiddleware(AgentMiddleware):
    """可观测性中间件"""

    def before_agent(self, state, runtime):
        print(f"Agent starting with {len(state['messages'])} messages")
        return None

    def after_model(self, state, response, runtime):
        # 记录模型响应元数据
        if hasattr(response, "response_metadata"):
            usage = response.response_metadata.get("usage", {})
            print(f"Tokens used: {usage}")
        return None

    def wrap_tool_call(self, state, tool_name, tool_args, runtime, call_next):
        import time
        start = time.time()
        result = call_next()
        duration = time.time() - start
        print(f"Tool '{tool_name}' took {duration:.2f}s")
        return result

agent = create_agent(
    "gpt-4o",
    tools=[my_tools],
    middleware=[ObservabilityMiddleware()]
)

流式追踪

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", streaming=True)

# 流式输出也会被追踪
for chunk in model.stream("Tell me a story"):
    print(chunk.content, end="", flush=True)

---

评估和测试

创建数据集

from langsmith import Client

client = Client()

# 创建数据集
dataset = client.create_dataset("qa-test-set")

# 添加示例
client.create_examples(
    inputs=[
        {"question": "What is Python?"},
        {"question": "How to install pip?"},
    ],
    outputs=[
        {"answer": "Python is a programming language."},
        {"answer": "Use: curl https://bootstrap.pypa.io/get-pip.py | python"},
    ],
    dataset_id=dataset.id
)

运行评估

from langsmith.evaluation import evaluate

def my_agent(inputs):
    """要评估的 Agent"""
    return agent.invoke(inputs)

# 运行评估
results = evaluate(
    my_agent,
    data="qa-test-set",
    evaluators=[
        "correctness",
        "helpfulness",
    ]
)

---

告警配置

设置告警规则

在 LangSmith 控制台中配置：

1. 延迟告警：响应时间超过阈值
2. 错误率告警：错误率超过百分比
3. 成本告警：日/月成本超过预算

Webhook 通知

# LangSmith 支持 Webhook 通知
# 配置示例：
# 1. 在 Settings 中添加 Webhook URL
# 2. 选择触发事件
# 3. 配置告警条件

---

最佳实践

1. 结构化项目命名

# 推荐命名规范
<team>-<service>-<environment>

# 示例
ml-chatbot-dev
ml-chatbot-staging
ml-chatbot-prod

2. 有意义的标签

config = {
    "tags": [
        "v2.0",           # 版本
        "gpt-4o",         # 模型
        "qa-task",        # 任务类型
        "high-priority",  # 优先级
    ]
}

3. 丰富的元数据

config = {
    "metadata": {
        # 用户信息
        "user_id": user_id,
        "user_tier": "premium",

        # 请求信息
        "request_id": request_id,
        "source": "mobile_app",

        # 实验信息
        "experiment": "new-prompt-v2",
        "variant": "A",
    }
}

4. 敏感数据处理

# 对敏感数据禁用追踪
with ls.tracing_context(enabled=False):
    result = process_sensitive_data(pii_data)

# 或使用数据脱敏
def sanitize_input(input_data):
    # 脱敏处理
    return redact_pii(input_data)

---

常见问题

Q: 追踪数据存储在哪里？

追踪数据存储在 LangSmith 云端服务器。企业版支持自托管。

Q: 追踪会影响性能吗？

追踪使用后台线程异步提交，对应用性能影响极小。

Q: 如何控制追踪成本？

1. 使用采样减少追踪量
2. 只追踪关键路径
3. 设置数据保留策略

Q: 支持 OpenTelemetry 吗？

是的，LangSmith 支持 OpenTelemetry 协议，可以与现有的可观测性基础设施集成。

---

总结

功能	用途
追踪	记录完整执行流程
调试	分析问题根因
监控	实时性能监控
评估	质量评估和回归测试
告警	异常情况通知

通过 LangSmith 的可观测性功能，你可以深入了解 LangChain 应用的行为，快速定位问题，持续优化性能。

---

上一节：7.1 Deployment

返回目录：1.0 Overview