AI智能体详解（四）-- LangSmith的使用

6.1 LangSmith 平台概述

LangSmith是LangChain生态系统中专门用于LLM（大语言模型）应用调试、监控、评估和管理的平台。

前几章的核心聚焦于智能体的开发阶段，而本章的重点将会放在 LangSmith 平台，它聚焦于智能体的可观测性和评估环节。

LangSmith 是大模型项目化平台，覆盖了从构建、评估、部署、监控到迭代的全生命周期。开发者可以把它与任意框架搭配使用，不只局限于 LangChain。它可以把应用运行过程转化为结构化数据，再通过这些数据分析和持续优化。

6.1.1 LangSmith 平台的产生背景与意义

随着大模型在业务场景中深入应用，开发者面临的挑战愈发明显：大模型行为难预测、执行过程不透明、质量评估机制不稳定。传统软件开发中清晰的逻辑链路追踪、错误定位和版本影响评估，在大模型应用中变得困难。链式任务、RAG 工作流和多工具智能体运行时，往往涉及多轮大模型调用、文档检索、工具执行和外部接口交互。这些步骤复杂、动态变化且难复现，导致调试、监控和优化困难。

LangSmith 平台正是在这样的背景下产生的，作为 LangChain 生态中面向生产环境的可观测性与评估平台，为大模型应用构建端到端的项目化支撑体系。无论系统使用 LangChain、LangGraph、Deep Agent 还是其他框架，LangSmith 平台都能可视化大模型的决策过程，提供从开发到部署的全流程调试、监控、评估和协作能力。

基于 LangSmith 平台，大模型应用的构建流程从 “写代码 — 手动测试 — 上线排查” 转变为 “开发 — 调试 — 观测 — 评估 — 优化 — 部署 — 监控 — 再优化”。其项目价值具体体现在以下三个方面。

（1）执行过程可视化。过去，开发者只能查看最终输出，而无法查看每一步模型推理、检索与工具调用的实际情况。LangSmith 平台通过结构化追踪，将请求拆解为清晰的调用序列，为调试和问题定位提供了基础。

（2）建立质量控制统一机制。提示词修改、模型替换或检索策略优化都需要量化评估。LangSmith 平台通过数据集评估、自动评分和人工审批，让改进基于真实数据而非直觉。

（3）构建可持续协作体系。团队规模扩大后，提示词项目工程师、开发项目工程师、数据分析项目工程师协同工作。LangSmith 平台将追踪、评估、版本管理整合到统一平台，让大模型应用以可控和项目化的方式持续迭代。

因此，LangSmith 平台不仅提升调试效率，还为大模型应用提供从构建到生产的项目底座，使智能体、RAG 系统和复杂工作流具备可观察、可评估、可维护的核心能力。

6.2 LangSmith 平台的可观测性

可观测性（Observability）是 LangSmith 平台的第一项核心能力，让开发者能追踪大模型应用内部的运行逻辑，每个步骤、每次调用、每个分支决策都有迹可循。 与传统应用相比，大模型应用的模型输出不可预测、工具链路不确定性大，用户输入差异会导致完全不同的执行路径。所以，没有结构化的可观测能力，开发者很难调试复杂智能体，也很难在生产环境中定位瓶颈、管理成本和监控质量。 基于这一需求，LangSmith 平台设计了一套由 Run、Trace、Thread 到 Project 逐级组织的追踪层级模型，用于对智能体执行过程进行统一观测与管理。 接下来，我们从可观测性的核心理念和关键组件出发，逐步了解 LangSmith 平台如何为复杂的大模型应用提供透明性和项目控制能力。

6.2.1 可观测性综述 LangSmith

平台的可观测性旨在让开发者可以回答以下几个关键问题。

（1）某个响应为什么会产生？ 模型用了哪些提示？中间步骤发生了什么？是否调用了工具？

（2）每一步的执行顺序是什么？ 哪些步骤是串行的？哪些是嵌套的？调用链有多深？

（3）性能与成本从何而来？ 哪些调用最耗时？哪些模型消耗了最多 Token？

（4）错误发生在哪里？为什么会发生？ 是否调用工具失败？提示词构造错误了吗？模型响应格式不符吗？

这些问题看似简单，却在大模型项目化中最常让团队陷入困境。LangSmith 平台的可观测性体系提供了一种统一的思考框架，让这些问题能够被结构化记录与可视化呈现，并且支持深入分析。

1. 在大模型应用生命周期中的作用

大模型应用生命周期通常包括开发阶段、测试阶段与生产阶段。在每个阶段，可观测性承担的角色各不相同。

（1）开发阶段：用来调试提示词、分析错误、理解代理的决策路径。

（2）测试阶段：确保相同输入在各场景下执行链路一致，避免出现隐藏分支。

（3）生产阶段：监控延迟、错误率、成本和吞吐量，确保系统在真实流量下稳定运行。

LangSmith 平台的优势在于让上述三个阶段共享同一套追踪结构与数据模型，无论是手动调试调用，还是生产环境的大规模流量，都能在相同的可视化结构中进行分析，从而避免开发日志与生产监控割裂带来的认知断层。

2. 核心能力

LangSmith 平台的可观测性能力包括以下几个方面。

（1）追踪：记录执行中的每个步骤，构建层级结构可视化链路。

（2）执行元数据：收集模型参数、提示内容、输出、耗时、Token 数、错误等信息。

（3）项目与对话管理：按项目、对话或任务运行数据，支持不同的开发环境与版本隔离。

（4）性能与成本统计：帮助优化链路设计、模型选择与调用策略。

（5）错误分析：定位异常节点、调用失败原因和模型不稳定行为。

这些能力共同构成了一个覆盖开发、测试与生产环境的完整可观测体系，使复杂的模型流程可解释、可回放、可优化。

3 核心组件详解

LangSmith 平台建立了一套层次清晰的可观测结构。它的基本思路是，先记录小步骤，再把这些步骤串成链，把链组合成对话或任务，最后将所有运行数据放在同一个项目里管理。与此同时，LangSmith 平台还允许在这些运行记录上添加标签、上下文信息和质量反馈，从而让开发者不仅能看到系统做了什么，还能理解为什么这样做、是在什么情况下做的、效果如何。

（1）Run：最小的可观测单元。无论是模型调用、工具执行，还是一段链式逻辑，每一步都会生成一个 Run。

（2）Trace：一次请求的执行链路。一个 Trace 由多个 Run 组成，表示一次请求从开始到结束经历了哪些步骤。Trace 显示了父子调用关系、执行顺序与耗时，是理解执行链路的关键。

（3）Thread：多轮上下文的聚合视图。Thread 会把多个相关 Trace 组织成一段连续的对话或一次长期任务，让开发者不仅能分析单轮执行，还能看到整个任务的上下文变化。

（4）Project：应用或环境的组织边界。Project 是更高层的组织边界，用来区分不同的应用或环境（如 dev 和 prod）。所有 Trace 和 Thread 都属于某个 Project，便于统一管理和对比。

（5）附件信息。

① Tag：附加在 Run 或 Trace 上的标签，用于分类与筛选，让开发者可以快速分组、对比不同的实验版本或不同类型的流量。

② Metadata：结构化上下文信息，以键值对形式记录更详细的上下文，如地区、模型版本、用户类型、应用配置等。

③ Feedback：用于对某次 Run 进行人工评分、用户反馈或自动评估。

Run→Trace→Thread→Project 的层级结构，让开发者能够从细节到全局、从单次操作到完整任务，完整复盘智能体的行为。Tag、Metadata 与 Feedback 则分别提供了分类、分析与质量判断能力，使可观测性体系真正可用、可扩展。

4. 性能指标

除了呈现执行链路，LangSmith 平台的可观测性还提供了一组关键性能指标，用于帮助开发者评估系统效率与成本。例如：

（1）延迟（Latency）：记录每一步的执行耗时，用于定位性能瓶颈。

（2）Token 使用量（Tokens）：统计输入与输出的 Token 数，作为计算成本基础。

（3）成本（Cost）：结合模型定价，计算调用的实际开销。

（4）错误率（Error Rate）：按模型、工具或项目维度分析失败情况。

（5）吞吐量（Throughput）：监控调用量与流量趋势，支持容量规划。

这些指标与追踪相结合，构成了优化性能、控制成本、保障系统稳定性的核心数据基础。

6.2.2 本地调试与 LangSmith Studio

在将大模型应用推向生产环境之前，需要本地启动、运行并调试系统。LangSmith 平台支持通过本地服务器模式来运行应用，从而让开发团队在真实和可控的环境中进行调试、观测、优化。

访问Langsmith官网：https://smith.LangChain.com/

注册，登录，获取 key 值

登录的界面在： Log In

登录成功后：

获取 key 值：

在设置中，找到

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

界面中的功能很多，主要关注这两个即可

在项目根目录创建.env 文件，填入必需变量，确保本地服务运行时能够与 LangSmith 后端进行通信。例如：

LANGSMITH_TRACING="true" # 是否启用LangSmith平台调试追踪（true表示开启） LANGSMITH_API_KEY="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" # LangSmith平台的API Key，用于认证请求 LANGSMITH_ENDPOINT="https://api.smith.langchain.com" # LangSmith平台的服务端接口地址（一般保持默认） LANGSMITH_PROJECT="hongzehu_smith" # 在LangSmith平台中用于归类与管理本项目的项目名称，如果不添加默认用的是default

set LANGSMITH_TRACING="true" # 是否启用LangSmith平台调试追踪（true表示开启） set LANGSMITH_API_KEY="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" # LangSmith平台的API Key，用于认证请求 set LANGSMITH_ENDPOINT="https://api.smith.langchain.com" # LangSmith平台的服务端接口地址（一般保持默认） set LANGSMITH_PROJECT="hongzehu_smith"

先创建一个项目：

在项目中安装软件：

pip install -U langsmith

代码：

在我们的 .env 中，添加如下内容：

OPENAI_API_KEY="sk-6C24A3NsyvexkWHFsVBCjzlvBYU8NBqkunvIkv971Jog8ZL9" # 个人API密钥 OPENAI_BASE_URL="https://api.openai-proxy.org/v1" # 固定代理地址，无需修改

如果你使用的是 deepseek 大模型，可能不出结果，运行没有问题，监测端监测不到

from langchain_core.prompts import PromptTemplate from langchain_core.prompts.few_shot import FewShotPromptTemplate import os import dotenv from langchain_openai import ChatOpenAI dotenv.load_dotenv() os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1") os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL") chat_model = ChatOpenAI(model="gpt-4o-mini") # 例子 examples = [ {"input": "北京天气怎么样", "output": "北京市"}, {"input": "南京下雨吗", "output": "南京市"}, {"input": "武汉热吗", "output": "武汉市"} ] # 例子拼装的格式 example_prompt = PromptTemplate( input_variables=["input", "output"], template="Input: {input}\nOutput: {output}" ) # Prompt模板 prompt = FewShotPromptTemplate( examples=examples, example_prompt=example_prompt, suffix="Input: {input}\nOutput:", # 要放在示例后面的提示模板字符串。 input_variables=["input"] # 传入的变量 ) prompt = prompt.format(input="长沙多少度") print("===Prompt===") print(prompt) print("===Response===") response = chat_model.invoke(prompt) print(response.content)

我们把之前的代码修改之后，也是可以的：

from typing import Any, Callable import time import math from langchain.agents import create_agent from langchain.agents.middleware import ( before_agent, after_agent, before_model, after_model, wrap_model_call, dynamic_prompt, AgentState, ModelRequest, ModelResponse, ) from langchain.chat_models import init_chat_model from langchain.messages import AIMessage, HumanMessage from langgraph.runtime import Runtime from langchain_openai import ChatOpenAI import os import dotenv from langchain_openai import ChatOpenAI dotenv.load_dotenv() os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY") os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL") # ========= 便捷装饰器：动态系统提示 ========= @dynamic_prompt def personalized_prompt(req: ModelRequest) -> str: runtime = getattr(req, "runtime", None) user_id = "访客" if runtime and getattr(runtime, "context", None): user_id = runtime.context.get("user_id", "访客") return f"你是一名贴心的中文助手，正在为用户「{user_id}」提供帮助。回答时要简洁、自然。" # ========= 节点式：Agent 级别前置/后置（一次调用各触发一次） ========= @before_agent def log_before_agent(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"[before_agent] 本次会话开始，已有消息数：{len(state.get('messages', []))}") return None @after_agent def log_after_agent(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"[after_agent] 会话结束，最终消息数：{len(state.get('messages', []))}") return None # ========= 节点式：模型调用前/后（每次模型调用都会触发） ========= @before_model def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: print(f"[before_model] 准备进行模型调用，当前消息数：{len(state['messages'])}") return None @after_model(can_jump_to=["end"]) def validate_output(state: AgentState, runtime: Runtime) -> dict[str, Any] | None: """简单的输出校验：若模型输出包含“BLOCKED_CN”，则改写消息并跳转到 end。""" last = state["messages"][-1] if isinstance(last, AIMessage) and "BLOCKED_CN" in (last.content or ""): print("[after_model] 触发安全规则：检测到 BLOCKED_CN，跳转到 end") return { "messages": [AIMessage("该请求触发了安全校验，无法继续。")], "jump_to": "end", } return None # ========= 包裹式：为模型调用加 重试与耗时统计 ========= @wrap_model_call def retry_and_timing( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: max_retries = 2 start = time.time() try: for i in range(max_retries + 1): try: return handler(request) except Exception as e: if i == max_retries: raise # 退避等待（100ms, 200ms） backoff = 0.1 * math.pow(2, i) print(f"[wrap_model_call] 调用失败，将在 {backoff:.2f}s 后重试：{e}") time.sleep(backoff) finally: cost = (time.time() - start) * 1000 print(f"[wrap_model_call] 本次模型调用耗时：{cost:.0f} ms") # ========= 组装 Agent（DeepSeek 模型） ========= # 你也可以通过环境变量配置：OPENAI_API_KEY/DEEPSEEK_API_KEY llm = ChatOpenAI(model="gpt-4o-mini") agent = create_agent( model=llm, tools=[], # 如需可加工具，此处保持最小化 middleware=[ personalized_prompt, # 便捷：动态系统提示 log_before_agent, # 节点：Agent 级前置 log_before_model, # 节点：模型前 retry_and_timing, # 包裹：重试与计时 validate_output, # 节点：模型后（含 jump_to） log_after_agent, # 节点：Agent 级后置 ], ) # ========= 最小演示 ========= if __name__ == "__main__": # 1) 正常问答（不会触发安全跳转） res1 = agent.invoke( {"messages": [HumanMessage("用一句话解释 LangGraph 是什么。")]}, config={"context": {"user_id": "alice"}}, ) print("\n[Result-1]", res1["messages"][-1].content) print("--------------------------------") # 2) 触发 after_model 的阻断（让模型输出包含关键字） res2 = agent.invoke( {"messages": [HumanMessage("请只回复：BLOCKED_CN")]}, config={"context": {"user_id": "bob"}}, ) print("\n[Result-2]", res2["messages"][-1].content)

使用 openAI 进行测试的时候，openai 的 key 必须在.env 里面，在代码中，是不监听的

比如：

from langchain_openai import ChatOpenAI # 创建 ChatGPT 模型实例 model = ChatOpenAI( model="gpt-4o-mini", temperature=0.7, # 控制输出的随机性，0-1之间，值越大越随机 api_key="sk-6C24A3NsyvexkWHFsVBCjzlvBYU8NBqkunvIkv971Jog8ZL9", base_url="https://api.openai-proxy.org/v1" ) # 调用模型 #response = model.invoke("请解释LangChain模型接口的统一性。") #print(response.content) for chunk in model.stream("请用一句话总结人工智能的意义："): print(chunk.content, end="")

第二个，将 open_ai 的 key 写在外面就可以监测

import os import dotenv from langchain_openai import ChatOpenAI dotenv.load_dotenv() os.environ['OPENAI_API_KEY']=os.getenv("OPENAI_API_KEY") os.environ['OPENAI_BASE_URL']=os.getenv("OPENAI_BASE_URL") # 创建 ChatGPT 模型实例 model = ChatOpenAI( model="gpt-4o-mini", temperature=0.7 ) for chunk in model.stream("请用一句话总结人工智能的意义："): print(chunk.content, end="")