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性能与成本从何而来 哪些调用最耗时哪些模型消耗了最多 Token4错误发生在哪里为什么会发生 是否调用工具失败提示词构造错误了吗模型响应格式不符吗这些问题看似简单却在大模型项目化中最常让团队陷入困境。LangSmith 平台的可观测性体系提供了一种统一的思考框架让这些问题能够被结构化记录与可视化呈现并且支持深入分析。1. 在大模型应用生命周期中的作用大模型应用生命周期通常包括开发阶段、测试阶段与生产阶段。在每个阶段可观测性承担的角色各不相同。1开发阶段用来调试提示词、分析错误、理解代理的决策路径。2测试阶段确保相同输入在各场景下执行链路一致避免出现隐藏分支。3生产阶段监控延迟、错误率、成本和吞吐量确保系统在真实流量下稳定运行。LangSmith 平台的优势在于让上述三个阶段共享同一套追踪结构与数据模型无论是手动调试调用还是生产环境的大规模流量都能在相同的可视化结构中进行分析从而避免开发日志与生产监控割裂带来的认知断层。2. 核心能力LangSmith 平台的可观测性能力包括以下几个方面。1追踪记录执行中的每个步骤构建层级结构可视化链路。2执行元数据收集模型参数、提示内容、输出、耗时、Token 数、错误等信息。3项目与对话管理按项目、对话或任务运行数据支持不同的开发环境与版本隔离。4性能与成本统计帮助优化链路设计、模型选择与调用策略。5错误分析定位异常节点、调用失败原因和模型不稳定行为。这些能力共同构成了一个覆盖开发、测试与生产环境的完整可观测体系使复杂的模型流程可解释、可回放、可优化。3 核心组件详解LangSmith 平台建立了一套层次清晰的可观测结构。它的基本思路是先记录小步骤再把这些步骤串成链把链组合成对话或任务最后将所有运行数据放在同一个项目里管理。与此同时LangSmith 平台还允许在这些运行记录上添加标签、上下文信息和质量反馈从而让开发者不仅能看到系统做了什么还能理解为什么这样做、是在什么情况下做的、效果如何。1Run最小的可观测单元。无论是模型调用、工具执行还是一段链式逻辑每一步都会生成一个 Run。2Trace一次请求的执行链路。一个 Trace 由多个 Run 组成表示一次请求从开始到结束经历了哪些步骤。Trace 显示了父子调用关系、执行顺序与耗时是理解执行链路的关键。3Thread多轮上下文的聚合视图。Thread 会把多个相关 Trace 组织成一段连续的对话或一次长期任务让开发者不仅能分析单轮执行还能看到整个任务的上下文变化。4Project应用或环境的组织边界。Project 是更高层的组织边界用来区分不同的应用或环境如 dev 和 prod。所有 Trace 和 Thread 都属于某个 Project便于统一管理和对比。5附件信息。① Tag附加在 Run 或 Trace 上的标签用于分类与筛选让开发者可以快速分组、对比不同的实验版本或不同类型的流量。② Metadata结构化上下文信息以键值对形式记录更详细的上下文如地区、模型版本、用户类型、应用配置等。③ Feedback用于对某次 Run 进行人工评分、用户反馈或自动评估。Run→Trace→Thread→Project 的层级结构让开发者能够从细节到全局、从单次操作到完整任务完整复盘智能体的行为。Tag、Metadata 与 Feedback 则分别提供了分类、分析与质量判断能力使可观测性体系真正可用、可扩展。4. 性能指标除了呈现执行链路LangSmith 平台的可观测性还提供了一组关键性能指标用于帮助开发者评估系统效率与成本。例如1延迟Latency记录每一步的执行耗时用于定位性能瓶颈。2Token 使用量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_TRACINGtrue # 是否启用LangSmith平台调试追踪true表示开启 LANGSMITH_API_KEYlsv2_pt_42e276d387874083aeea7f23918dd50e_675982e28e # LangSmith平台的API Key用于认证请求 LANGSMITH_ENDPOINThttps://api.smith.langchain.com # LangSmith平台的服务端接口地址一般保持默认 LANGSMITH_PROJECThongzehu_smith # 在LangSmith平台中用于归类与管理本项目的项目名称如果不添加默认用的是defaultset LANGSMITH_TRACINGtrue # 是否启用LangSmith平台调试追踪true表示开启 set LANGSMITH_API_KEYlsv2_pt_42e276d387874083aeea7f23918dd50e_675982e28e # LangSmith平台的API Key用于认证请求 set LANGSMITH_ENDPOINThttps://api.smith.langchain.com # LangSmith平台的服务端接口地址一般保持默认 set LANGSMITH_PROJECThongzehu_smith先创建一个项目在项目中安装软件pip install -U langsmith代码在我们的 .env 中添加如下内容OPENAI_API_KEYsk-6C24A3NsyvexkWHFsVBCjzlvBYU8NBqkunvIkv971Jog8ZL9 # 个人API密钥 OPENAI_BASE_URLhttps://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(modelgpt-4o-mini) # 例子 examples [ {input: 北京天气怎么样, output: 北京市}, {input: 南京下雨吗, output: 南京市}, {input: 武汉热吗, output: 武汉市} ] # 例子拼装的格式 example_prompt PromptTemplate( input_variables[input, output], templateInput: {input}\nOutput: {output} ) # Prompt模板 prompt FewShotPromptTemplate( examplesexamples, example_promptexample_prompt, suffixInput: {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) # 组装 AgentDeepSeek 模型 # 你也可以通过环境变量配置OPENAI_API_KEY/DEEPSEEK_API_KEY llm ChatOpenAI(modelgpt-4o-mini) agent create_agent( modelllm, 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( modelgpt-4o-mini, temperature0.7, # 控制输出的随机性0-1之间值越大越随机 api_keysk-6C24A3NsyvexkWHFsVBCjzlvBYU8NBqkunvIkv971Jog8ZL9, base_urlhttps://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( modelgpt-4o-mini, temperature0.7 ) for chunk in model.stream(请用一句话总结人工智能的意义): print(chunk.content, end)