LangSmith

LangChain 2026：从链到图的范式跃迁 2024 年初，LangChain 创始人 Harrison Chase 做了一个大胆的决定：将框架核心从"链"（Chain）转向"图"（Graph）。两年后的今天，这个决定被证明是极具前瞻性的。LangChain 2026 生态已经发展为一个包含 LangGraph、LangSmith、LangServe、LangChain CLI 在内的完整工具矩阵，月活开发者超过 120 万，成为 Agent 开发领域事实上的标准。 核心组件架构 LangGraph：状态机驱动的 Agent 编排 LangGraph 是 2026 年 LangChain 生态中最核心的组件。与传统的链式调用不同，LangGraph 采用有向有环图（Directed Cyclic Graph）来建模 Agent 工作流，原生支持循环、分支、并行和人工干预。 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] context: dict iteration: int def research_node(state: AgentState): # 调用搜索工具获取信息 result = search_tool.invoke(state["messages"][-1].content) return {"messages": [{"role": "tool", "content": result}]} def reasoning_node(state: AgentState): # LLM 推理节点 response = llm.invoke(state["messages"]) return {"messages": [response]} def should_continue(state: AgentState) -> str: last_msg = state["messages"][-1] if state["iteration"] >= 5: return END if last_msg.tool_calls: return "tools" return END # 构建工作流图 workflow = StateGraph(AgentState) workflow.add_node("research", research_node) workflow.add_node("reasoning", reasoning_node) workflow.add_node("tools", tool_executor) workflow.set_entry_point("reasoning") workflow.add_conditional_edges("reasoning", should_continue) workflow.add_edge("tools", "reasoning") app = workflow.compile(checkpointer=MemorySaver()) LangGraph 2026 关键特性 特性 2024 版本 2026 版本 状态管理 基础状态字典 类型安全的 TypedDict + Reducer 并行执行 不支持 原生 Fan-out/Fan-in 持久化 基础 Memory SQLite/PostgreSQL/Redis 多后端 人工干预 基础中断 细粒度审批流 + 超时机制 流式输出 仅支持文本 事件流 + Token 级流式 子图嵌套 不支持 多级子图 + 状态隔离 时间旅行 不支持 完整状态回放 + 分支执行 LangSmith：全链路可观测性平台 LangSmith 在 2026 年已经从单纯的调试工具进化为完整的 LLM 可观测性平台。它提供： ...

为什么智能体需要专属的可观测性？ 传统软件的可观测性聚焦于 CPU、内存、延迟等系统指标。但 AI 智能体引入了全新的可观测维度： Token 消耗：每次 LLM 调用都有成本，需要精确追踪 推理链路：Agent 可能经历多轮 Thought → Action → Observation 循环，需要完整 Trace 工具调用质量：工具是否被正确调用？返回结果是否有效？ Prompt 效果：不同 Prompt 版本对输出质量的影响如何？ 幻觉检测：模型输出是否与事实相符？ 缺乏可观测性的智能体就像一个黑盒——你不知道它在想什么，也不知道它为什么出错。本文将带你从零搭建一套生产级的智能体可观测性平台。 可观测性三支柱在 AI 场景下的重构 传统可观测性的三支柱是 Metrics、Logs、Traces。在智能体场景下，我们需要将其扩展为五支柱： 支柱 传统场景 智能体场景 Traces 请求链路追踪 Thought-Action-Observation 链路 Metrics QPS、延迟、错误率 Token 用量、工具调用成功率、幻觉率 Logs 结构化日志 Prompt/Completion 完整记录 Evaluations N/A 输出质量自动评估 Cost N/A Token 成本与预算控制 架构设计 ┌────────────────────────────────────────────────────────┐ │ Agent Application │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │ │ │ LLM Call │ │ Tool Call │ │ Observation Callback │ │ │ └─────┬─────┘ └─────┬────┘ └──────────┬───────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Observability SDK │ │ │ │ (OpenTelemetry + Custom Spans + Token Counter) │ │ │ └──────────────────────┬──────────────────────────┘ │ └─────────────────────────┼──────────────────────────────┘ │ ┌───────────────┼───────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ Jaeger │ │ Prometheus │ │ Postgres │ │ (Traces) │ │ (Metrics) │ │ (Logs) │ └─────────────┘ └─────────────┘ └───────────┘ │ │ │ └───────────────┼───────────────┘ ▼ ┌─────────────┐ │ Grafana │ │ (Dashboard) │ └─────────────┘ 核心组件实现 1. Trace 追踪：基于 OpenTelemetry 扩展 智能体的 Trace 与传统微服务不同，需要记录 LLM 特有的 Span 属性： ...

为什么 AI 应用需要可观测性 传统应用监控关注 CPU、内存、QPS。AI 应用需要关注：LLM 调用了几次？每次延迟多少？Token 消耗多少？检索结果是否相关？Agent 的推理链路是什么？没有可观测性，AI 应用就是黑盒，出问题只能猜。 三支柱模型 支柱 传统应用 AI 应用 工具 Tracing 请求链路 LLM 调用链 + Agent 推理步骤 + 检索过程 LangSmith/Langfuse Logging 日志 Prompt/Response/中间状态 结构化日志 Metrics QPS/延迟 Token消耗/检索召回率/幻觉率 Prometheus+Grafana 关键指标定义 from dataclasses import dataclass from typing import Optional @dataclass class LLMCallMetrics: # 延迟指标 time_to_first_token: float # 首 token 延迟(ms) total_latency: float # 总延迟(ms) # 成本指标 prompt_tokens: int completion_tokens: int total_cost: float # 质量指标 model_name: str temperature: float finish_reason: str # 检索指标(RAG场景) retrieval_count: Optional[int] = None retrieval_relevance: Optional[float] = None context_utilization: Optional[float] = None @dataclass class AgentTrace: trace_id: str spans: list[dict] # 每个步骤一个 span # span 结构: name, start, end, input, output, metadata Langfuse 集成实现 from langfuse import Langfuse from langfuse.decorators import langfuse_context, observe import functools, time langfuse = Langfuse( public_key="pk-lf-xxx", secret_key="sk-lf-xxx", host="https://your-langfuse.com" ) def trace_llm(name="llm_call"): def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): start = time.time() span = langfuse_context.start_span(name=name, input=kwargs) try: result = await func(*args, **kwargs) ttft = time.time() - start span.end( output=result, metadata={"ttft_ms": ttft * 1000, "model": kwargs.get("model", "unknown")} ) langfuse_context.update_current_observation( usage={ "prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0), "completion_tokens": result.get("usage", {}).get("completion_tokens", 0), "total_tokens": result.get("usage", {}).get("total_tokens", 0), }, metadata={"latency_ms": (time.time() - start) * 1000} ) return result except Exception as e: span.end(level="ERROR", status_message=str(e)) raise return wrapper return decorator @observe(name="rag_pipeline") async def rag_pipeline(query: str): docs = await retrieve(query) answer = await generate(query, docs) return answer @observe(name="retrieve") async def retrieve(query: str): results = await vector_search(query) return {"query": query, "docs": results} @trace_llm(name="generate") async def generate(query: str, docs: list): resp = await llm_client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": f"Q: {query}\nContext: {docs}"}] ) return resp 工具对比 工具 部署方式 核心优势 适用场景 价格 LangSmith SaaS LangChain 深度集成 LangChain 用户 $39/月起 Langfuse 自部署/SaaS 开源，框架无关 通用 AI 应用 免费/自部署 Phoenix 自部署 Arize 出品，评估强 需要评估体系 开源 OpenTelemetry 自部署 标准化，通用 已有 OTel 体系 免费 选型建议：不用 LangChain 选 Langfuse（开源、框架无关），用 LangChain 选 LangSmith（集成最好），需要深度评估选 Phoenix。 ...