写在README里的LangGraph、LangSmith、Langfuse，怎样真正跑起来？
大家好，我是云萱。
最近我在忙着给云萱 AI 工程平台补”欠账”。
什么欠账呢？就是那种写在 README 技术栈表里、看起来很厉害、但点进去发现根本没实现的东西。
说的就是你——LangGraph、LangSmith、Langfuse。
我一直很好奇，号称”LangChain 全家桶”的项目，有多少是真的把这些东西用起来了，又有多少只是把名字列上去撑场面的。
这篇文章，我用自己踩过的坑，彻底解答一个问题：
一个 AI 工程平台，怎样把 LangChain 生态从”写在文档里”变成”真正跑在生产里”？
看完之后，相信你对 LangGraph、LangSmith、Langfuse 的落地路径会有更清晰的认识，在做自己的 AI 项目时，也就更有底了。
1. 先泼一盆冷水：你真的需要 LangGraph 吗？
在讲怎么接入之前，我想先劝退。
没错，劝退。
很多人一上来就说”我要用 LangGraph 做 Agent 编排”，但你问他为什么要用，得到的回答往往是”因为大家都在用”。
这和十几年前大家一窝蜂上微服务的场景，简直一模一样。
什么时候不需要 LangGraph
如果你满足下面这些条件，真心建议别用：
- 你的工作流是线性的，A 做完做 B，B 做完做 C
- 状态很简单，一个字典就能装下
- 不需要中断恢复、不需要持久化、不需要并行分支
- 团队里没人熟悉 LangChain 生态
手写状态机，往往比 LangGraph 更快、更简单、更好调试。
我们平台的第一版就是纯 Python 手写状态机，200 多行代码，逻辑一目了然，出了 bug 一找一个准。
那为什么我们还是接入了？
两个原因。
第一个，生态扩展性。
LangGraph 内置了 persistence（检查点持久化）、human-in-the-loop（人工介入）、streaming（流式输出）、并行节点这些能力。现在用不上，不代表以后用不上。等你需要的时候再换，代价就大了。
第二个，人才市场。
说个很现实的问题：你招 AI 工程师的时候，“会 LangGraph” 和 “会我们自研的状态机”，哪个更有吸引力？答案不言而喻。
但我们做了一个非常关键的设计决策：
LangGraph 是可选的，不是必需的。
什么意思？就是没有 LangGraph 的环境，系统也能 100% 正常运行。LangGraph 只是一个加速包，不是发动机。
2. 双模式设计：LangGraph 能用就用，不能用就算了
具体怎么实现的？很简单。
class SupervisorAgent:
def __init__(self, use_langgraph: bool | None = None):
# 自动检测：装了 langgraph 就用，没装就 fallback
self.use_langgraph = (
use_langgraph if use_langgraph is not None
else _langgraph_available()
)
def run(self, user_request: str, **kwargs) -> AgentState:
if self.use_langgraph:
return self._run_langgraph(state)
else:
return self._run_fallback(state)
_langgraph_available() 就是个简单的 import 检查：
def _langgraph_available() -> bool:
try:
import langgraph
return True
except ImportError:
return False
这种渐进式增强的思路，贯穿了我们整个技术选型：
基础功能零依赖，高级能力有就用，没有也不影响。
我个人非常推崇这种方式。它的好处是：
- 新人上手快，不装 LangGraph 也能跑
- 出问题可以快速切回 fallback，定位是不是 LangGraph 的锅
- 不把项目绑定在某一个框架上，未来换框架也容易
3. 接入 LangGraph 踩过的三个坑
说起来简单，真接进去的时候还是踩了不少坑。挑三个最典型的说说。
坑一：StateGraph 的 state 更新语义
LangGraph 的节点函数返回的是 state 的更新字段（partial update），不是完整的 state。
这个设计本身没问题。但如果你的 state 里有嵌套的 Pydantic 对象，更新的时候就要特别小心了。
# ❌ 直接返回 dict，LangGraph 会把整个 tester 字段覆盖掉
# 原来 tester 里的其他字段就没了
return {"tester": {"failed_count": 3}}
我们的解决方案是自己实现了一个 merge 逻辑：
@staticmethod
def _apply_update(state: AgentState, update: dict) -> None:
for key, value in update.items():
# 如果是子状态对象且 value 是 dict，合并而不是覆盖
if key in state_classes and isinstance(value, dict):
existing = getattr(state, key)
for k, v in value.items():
if hasattr(existing, k):
setattr(existing, k, v)
else:
setattr(state, key, value)
这个 bug 藏得挺深的——一开始测试的时候没问题，因为每个阶段只写自己的字段。但到了 debugger 阶段，需要回头修改 tester 的状态时，数据就丢了。查了好半天才定位到。
坑二：条件边 + 循环 = 递归深度限制
我们的工作流里有一个 tester → debugger → tester 的循环（测试失败了就去修，修完了回来再测）。
在 LangGraph 里这个是靠 conditional edge 实现的：
graph.add_conditional_edges(
"tester",
self._router_after_tester,
{"debugger": "debugger", "evaluator": "evaluator"},
)
看起来没问题对吧？但 LangGraph 默认的 recursion_limit 是 25。
什么意思？就是整个图的执行步数不能超过 25 步。我们一个 tester→debugger→tester 循环就占了 3 步，循环 3 次就是 9 步，再加上其他 5 个阶段，总共十几个节点，还没到 25。
但如果你的逻辑更复杂，或者循环更多，一定要记得调这个参数。不然跑到一半给你报个 GraphRecursionError，你还以为是自己写了死循环。
坑三：get_graph() 报的错可以忽略
这个纯属小问题，但还是提一下。
调用 compiled_graph.get_graph() 想看看图的结构，结果在 __root__ key 上报了个 InvalidUpdateError。
一开始还以为是自己图建错了，查了半天发现 invoke 和 stream 都正常跑，就是可视化工具有点小 bug。
不影响执行，忽略就行。
4. 比 LangGraph 更重要的，是可观测性
聊完了 LangGraph，来说说我认为更重要的东西——可观测性。
为什么说它更重要？
因为 AI 系统的调试难度，和传统软件根本不是一个量级的。传统软件你可以打断点、单步跟踪，AI 系统呢？你能单步跟踪 LLM 的思考过程吗？你能稳定复现某一个输出吗？
都很难。
没有好的追踪系统，出了问题你根本无从下手。你不知道是 prompt 写得不好，还是模型抽风，还是工具调用出了问题。
我们的设计目标
在动手之前，先定了三个原则：
第一，后端无关。
业务代码绝对不能直接依赖 LangSmith 或者 Langfuse 的 SDK。不然哪天想换平台，就要改一堆业务代码。
第二，自动降级。
配置了 LangSmith，但 API Key 失效了怎么办？网络不通了怎么办？
答案是：自动退回到 console 模式，绝不影响主流程。
没有什么比”追踪系统把主流程搞挂了”更尴尬的事了。
第三，渐进增强。
console 模式就能用，而且够用。上生产了再接 LangSmith 或者 Langfuse，能力升级但代码不改。
5. 五种后端，一套接口
最终我们实现了 5 种追踪后端：
| 后端 | 适用场景 |
|---|---|
noop | 性能测试、生产环境完全静默 |
console | 本地开发调试（默认） |
langsmith | LangChain 生态项目 |
langfuse | 开源、私有化部署 |
otel | 企业级 APM 集成（预留） |
怎么做到的？核心就是一个抽象基类：
class _BaseBackend:
def start_span(self, span: Span) -> None: ...
def end_span(self, span: Span) -> None: ...
def get_spans(self, trace_id: str) -> list[Span]: ...
def flush(self) -> None: ...
然后 Tracer 初始化的时候，根据配置选择具体的后端：
def _create_backend(backend_type: str) -> _BaseBackend:
try:
if backend_type == "noop":
return _NoOpBackend()
if backend_type == "console":
return _ConsoleBackend()
if backend_type == "langsmith":
return _LangSmithBackend()
if backend_type == "langfuse":
return _LangfuseBackend()
# ...
except Exception as e:
logger.warning(f"追踪后端 {backend_type} 初始化失败，降级为 console: {e}")
return _ConsoleBackend()
注意这个 try/except。它的存在，就是第二条原则的具体体现——任何后端初始化失败，都自动降级到 console。
6. LangSmith 后端怎么实现的
LangSmith 的核心概念是 Run，有父子嵌套关系。我们用 RunTree 来构建 span 树。
class _LangSmithBackend(_BaseBackend):
def __init__(self):
from langsmith import Client, RunTree
self.client = Client()
self._run_trees: dict[str, RunTree] = {}
def start_span(self, span: Span):
run_type = _map_span_type_to_run_type(span.span_type)
parent = self._run_trees.get(span.parent_span_id) if span.parent_span_id else None
if parent:
run = parent.create_child(
name=span.name,
run_type=run_type,
inputs={"input": span.input_snapshot, **span.metadata},
)
else:
run = RunTree(
name=span.name,
run_type=run_type,
inputs={"input": span.input_snapshot, **span.metadata},
trace_id=span.trace_id,
)
self._run_trees[span.span_id] = run
def end_span(self, span: Span):
run = self._run_trees.pop(span.span_id, None)
if run:
run.end(outputs={"output": span.output_snapshot, "status": span.status})
if span.parent_span_id is None:
run.post() # 根 span 需要主动 post
几个 span 类型到 LangSmith run_type 的映射：
llm_call→run_type="llm"tool_call→run_type="tool"rag_query→run_type="retriever"agent_step→run_type="chain"- 其他 →
run_type="span"
接入完成后，在 LangSmith 的 Dashboard 里就能看到每一次 LLM 调用、每一个工作流阶段，还有它们之间的父子关系。调试起来方便太多了。
7. Langfuse 后端怎么实现的
Langfuse 是另一个很火的开源 LLM 可观测性平台。它的 API 设计更贴合 LLM 追踪场景，有 trace → span → generation 三层结构。
class _LangfuseBackend(_BaseBackend):
def __init__(self):
from langfuse import Langfuse
self.client = Langfuse() # 自动从环境变量读配置
self._traces: dict[str, Any] = {}
self._spans: dict[str, Any] = {}
def start_span(self, span: Span):
# 根 span 对应一个 trace
if span.parent_span_id is None:
trace = self.client.trace(
id=span.trace_id,
name=span.name,
metadata=span.metadata,
)
self._traces[span.trace_id] = trace
parent_obj = trace
else:
parent_obj = self._spans.get(span.parent_span_id)
langfuse_span = parent_obj.span(
name=span.name,
input=span.input_snapshot,
metadata=span.metadata,
)
self._spans[span.span_id] = langfuse_span
def end_span(self, span: Span):
s = self._spans.pop(span.span_id, None)
if s:
s.end(output=span.output_snapshot)
def flush(self):
self.client.flush()
Langfuse 有一点我特别喜欢：它的 SDK 在缺少配置的时候，不会抛异常，只是打个 warning，然后静默工作。这和我们”自动降级”的理念完全一致。
8. 目前接了哪些埋点？
说了这么多架构，那实际代码里到底哪些地方接了追踪？
目前是两个核心模块。
LLM 客户端：6 个方法全埋点
invoke、ainvoke、call、acall、call_json、acall_json，所有调用方法都接了。
记录了哪些信息？
- 模型名、消息数量、temperature
- 输入输出的截断快照（500 字符，避免存太多数据）
- 粗略估算的 token 数（中文按字数，英文按词数 × 0.75）
def invoke(self, messages: list[BaseMessage]) -> BaseMessage:
tracer = get_tracer()
with tracer.span(
f"llm_invoke:{self._config.model}",
span_type="llm_call",
metadata={
"model": self._config.model,
"message_count": len(messages),
"temperature": self._config.temperature,
},
input_snapshot=_truncate(str(messages[-1].content), 500),
) as span:
result = self._get_model().invoke(messages)
span.metadata["input_tokens"] = _estimate_tokens(...)
span.metadata["output_tokens"] = _estimate_tokens(...)
return result
工作流引擎：每个阶段一个 span
工作流引擎每次推进到下一个阶段，都会包一个 stage_transition span：
def advance(self):
tracer = get_tracer()
with tracer.span(
f"stage:{next_stage}",
span_type="stage_transition",
metadata={"from_stage": current, "to_stage": next_stage},
):
# 执行阶段逻辑...
这样一来，你就能清楚地看到：整个工作流花了多久，每个阶段各占多少时间，哪个阶段是瓶颈。
9. 几个可能反直觉的思考
做这个事情的过程中，有一些想法和我最开始的预期不太一样，分享给大家。
“完整” 是敌人，“能用” 是朋友
我们的 tracer 一开始设计得特别宏大：
12 种 span 类型、10 个可回答的问题、完美的 OTel 兼容、流式追踪、采样率控制……
结果呢？写了一周还没落地。
后来我把所有不立刻要用的东西全砍了，只保留：
- 2 种 span（llm_call, stage_transition）
- 3 种后端（noop, console, langsmith）
- 1 个查询方法（get_spans）
半天就跑通了。然后边用边加，用到什么加什么。
做基础设施类的东西，先有 60 分的版本能用起来，比花三个月做 100 分的设计重要得多。
这个道理说起来都懂，但做的时候总忍不住想”一步到位”。
追踪系统本身不能引入新的不确定性
这一点怎么强调都不过分。
- 追踪失败不能导致主流程失败 → try/except + 自动降级
- 追踪开销不能太大 → 快照截断、异步发送
- 配置缺失要有合理默认 → 默认 console 后端，开箱即用
追踪系统是辅助工具，它的优先级永远在主业务之后。如果它会影响主业务，那它就失败了。
LangChain 是”胶水”，不是”框架”
用了一段时间 LangChain 生态后，我最大的感受是：
LangChain Core 很好用。 BaseMessage、ChatOpenAI、JsonOutputParser 这些基础抽象，节省了大量重复劳动。
LangGraph 在特定场景下有价值。 比如复杂的状态机、需要持久化和中断恢复的场景。但如果你的流程很简单，真没必要硬上。
LangSmith 和 Langfuse 很有价值。 AI 系统的调试和监控是刚需，这两个工具能帮你省很多事。
但你不需要”全部都用”。
挑你需要的，剩下的自己实现，往往是更务实的选择。
文档诚实比好看重要
这一点是我个人感触最深的。
最开始的 README 把所有计划中的技术都列进去了，看起来特别厉害。LangGraph、LangSmith、Langfuse、OTel…… 技术栈表写得满满当当。
但真正有人用的时候，发现”哎？怎么 LangSmith 配置了没反应？”
信任感一下就掉了。
现在我们的做法很简单：
- 已实现的：写清楚怎么用、有什么限制
- 未实现的：要么不写，要么明确标注”规划中”
- 部分实现的：标注接入进度（✅ 已接入 / ⏳ 待接入）
技术栈表格”看起来弱一点”没关系，文档的可信度才是最宝贵的资产。
10. 接下来打算怎么做？
目前这套可观测性体系还处于早期阶段，接下来有几个方向：
- 接入更多模块：工具调用、Agent 步骤、RAG 查询、安全事件，这些都要逐步接进来
- Token 用量精确统计：现在是粗略估算，后面要接入 tiktoken 做精确计数
- 成本分析看板：按项目、按用户、按阶段的成本统计
- 质量评估：结合追踪数据，自动分析哪些环节质量不稳定
- LangGraph 持久化：用 checkpointer 实现工作流中断恢复
一步一步来，不急。
最后还想说几句
看到这里，不知道你对我们这套 LangChain 生态落地方案感觉怎么样。
我知道有人会说：“搞这么复杂干嘛？直接用 LangChain 自带的不就行了？”
确实，如果站在快速原型的角度，完全可以更简单。LangChain 自带了 LangSmith 集成，Langfuse 也有 LangChain 集成，几行代码就能接上。
但如果我们换个角度——如果你的项目计划长期维护，不希望绑定在某一个供应商上，那多一层抽象带来的自由度，是非常有价值的。
今天你想用 LangSmith，明天你想换成 Langfuse，后天你想自建 OTel，难道每次都去改业务代码吗？
作为一个程序员，就算有人说这是过度设计，我也要说一句：这层抽象，值。
它让我们的追踪系统变成了一个可以自由插拔的模块，而不是绑死在某个平台上的黑盒。
得益于这套设计，我们切换追踪后端只需要改一行配置。
只是在当下这个时间点来说，对很多小项目而言，这样做可能确实有点超前。这条路值不值得走，每个人心里都有自己的答案。
如果你还想进一步了解云萱 AI 工程平台的技术细节，可以看看我之前写的这篇文章：
👉️ 云萱 AI 原生工程平台：从需求到上线的全链路 AI 自动化实践
好了，今天就分享到这里。后续我还会继续写 AI 工程化相关的实战经验，欢迎持续关注。
本文由云萱 AI 原生工程平台辅助生成，人工审校发布。