Agent 落地

Agent 落地自然要保证整个系统的稳定可观测。

这和其他生产系统其实差别不算很大，基本就是可观测、日志、重试、缓存、Fallback、超时、限流。

重试（Retry）与 Fallback

在调用大模型或业务系统 API（如查询余额接口）时，网络抖动或大模型返回 JSON 格式错误是常态。

可以在定义 Node 时，直接对大模型或 API 调用链加上 .with_retry()。

工程实践：

当外呼机器人调用计费接口偶尔失败时，你可以精细控制重试策略，比如只针对特定异常（如 TimeoutError）重试，最多 3 次，并设置指数退避。

兜底回退：

如果重试 3 次仍然失败，使用 .with_fallbacks() 切换到备选模型或返回预设的默认状态（如："系统繁忙，请稍后再试"），防止整个图崩溃。

def query_billing_node(state):
    # 配置重试和兜底策略
    robust_chain = (
        api_call_chain
        .with_retry(
            stop_after_attempt=3,
            wait_exponential_jitter=True
        )
        .with_fallbacks([fallback_chain])
    )

    result = robust_chain.invoke(state["account_id"])
    return {"account_info": result}

超时控制（Timeout）

LangGraph 的调用是在 graph.invoke() 时统一从外部注入超时控制。

config = {
    "configurable": {
        "thread_id": "call_123"
    },
    "timeout": 15.0  # 15秒超时
}

graph.invoke(state, config=config)

断点（Checkpoint）

实际生产中，直接让 Agent 完全自动化是非常危险的，因此需要在固定的地方设定断点，让人工介入与审核，出现问题也能恢复状态。

Checkpointer 会在每个节点执行完毕后，把当前状态序列化落盘。

在执行高危操作（如“给用户办理退款”或“发送正式邮件”）的节点前设置断点。

架构走到这里会自动挂起并保存状态。

人工审批通过后，调用端带上同一个 thread_id 再次运行：

graph.invoke(None, config)

它就会从挂起的地方继续执行。

如果 Agent 在某个节点（比如下载大文件）因为 OOM 崩溃了，只要 thread_id 还在，重启服务后传入该 ID，它会直接跳过之前已成功的节点，从失败的节点重新开始。

可观测（Observability）

在由状态机驱动的系统里，传统的打 log 方式（查一堆混杂的文本日志）排障效率极低，因为你很难看清状态是怎么流转的。

所以可以选择接入 LangSmith，不过这玩意要钱，所以还是选择其他开源的 Langfuse 吧。

部署并接入后初始化一个 CallbackHandler，并在执行的时候作为参数传进去。

langfuse_handler = CallbackHandler()

# 在 invoke 时的 config 中注入这个 handler
config = {
    "configurable": {
        "thread_id": "user_session_999"
    },
    "callbacks": [langfuse_handler]
}

部署完成后，你可以在 Langfuse 网页端直观看到：

完全本地化的 Trace

点开任意一次用户的会话，你能清晰地看到：

输入是什么 → 进入了哪个 Node → 调了什么 Tool API（耗时多久、返回了什么）→ 最终模型吐出了什么。

哪里卡死、哪里逻辑跑偏，一目了然。

RAG 检索质量

它支持专门的 Generation 和 Span 类型标签。

可以把向量数据库捞出来的 Context 文档和最后大模型的 Prompt 拆开对比，一眼看出是不是检索出来的文档出了差错。

离线数据集与回归评测

可以把线上跑偏的 Case 一键保存为内网测试集。

当你微调了图中的某个 Prompt 后，直接在内网跑批量跑测，使用本地的其他开源大模型（如大模型的本地部署版本）来当裁判打分，全流程数据不出网。

最核心、也是日常排障频率最高的界面就是 Trace 看板。

学习门槛很低，一用就会。