五个核心组件、SDK 调用链、本地开发流程,以及如何用统一接口无缝切换本地与云端。
前几讲我们构建了 Simple Graph、Chain、Router、Agent,全都在 Jupyter Notebook 里通过 graph.invoke() 同步调用。这在学习阶段没问题,但在生产环境会遇到几个严重问题:
| 问题 | invoke() 的局限 | LangGraph 部署体系的解决方案 |
|---|---|---|
| 阻塞 | invoke() 是同步的,Agent 运行几分钟时,调用方会一直等待 |
LangGraph API 提供异步任务队列,调用方立即返回,结果通过 streaming 推送 |
| 持久化 | 需要手动配置 MemorySaver,且只在单进程内有效 |
LangGraph API 内置持久化层,跨进程、跨重启保持 Thread State |
| 可访问性 | 只能在同一进程内通过 Python 调用 | HTTP API + SDK,任何语言、任何地方都能调用 |
| 可观测性 | 运行过程不透明 | LangSmith 监控、追踪每一步的输入输出 |
理解 LangGraph 部署体系的 五个组件,以及 SDK 的三个核心概念(Assistant / Thread / Run)。掌握了这些,你就能把本地调试好的 Agent 一键搬上云端,同时保持相同的调用代码。
LangGraph 部署体系由五个组件构成,它们的层次关系如下:从最底层的 Python/JS 库,到 API 服务器,再到三种不同的访问/托管方式。
| 组件 | 本质 | 你用它做什么 |
|---|---|---|
| LangGraph(库) | pip install langgraph Python/JS 库 |
定义 StateGraph,添加节点和边,这是你写业务逻辑的地方 |
| LangGraph API | HTTP 服务器 本地 / 云端都能跑 |
把你的图包装成可以被远程调用的服务,提供持久化和队列 |
| LangSmith Studio (原 LangGraph Studio) |
浏览器 IDE | 可视化地看图的结构,手动测试输入,观察每步 State 变化 |
| LangSmith Deployment (原 LangGraph Cloud) |
托管云服务 | 在 LangSmith 平台一键部署,得到唯一 URL,托管 LangGraph API |
| LangGraph SDK | pip install langgraph-sdk Python 客户端 |
在代码里程序化地调用图(创建 Thread、发起 Run、获取 Stream) |
LangGraph 生态正在与 LangSmith 深度整合。原来叫 LangGraph Cloud 的云服务现在叫 LangSmith Deployment;原来叫 LangGraph Studio 的 IDE 现在叫 LangSmith Studio。功能本质相同,只是品牌统一了。
这是理解整个部署体系的关键问题。直接用 invoke() 不是更简单吗?让我们对比两种方式:
# 直接 invoke —— 简单,但在生产环境有严重问题
result = graph.invoke({"messages": [HumanMessage("帮我研究量子计算的最新进展")]})
# ^ 这里会阻塞 5~10 分钟,直到 Agent 完成所有工具调用
# Web 服务器的请求会超时,用户体验极差
Client 发起请求后立即返回 run_id,不阻塞。API 在后台异步执行图,结果通过 streaming 推送给监听方。适合运行时间长的 Agent。
API 自动为每个 Thread 维护跨交互的状态快照。不需要手动配置 MemorySaver,重启服务也不会丢失对话历史。这是 Module 2 讲 Memory 的底层基础。
无论图运行在本地(langgraph dev)还是云端(LangSmith Deployment),都通过同一套 HTTP API 暴露。SDK 封装了这些 HTTP 调用,让你用 Python 优雅地访问。
# 通过 API 的异步方式 —— 立即返回,streaming 获取结果
async for chunk in client.runs.stream(thread_id, "agent", input=input):
# 每个图步骤完成后立即收到更新,不阻塞
print(chunk.data)
在把 Agent 部署到云端之前,你需要在本地启动 LangGraph API 服务器,这样才能用 Studio 可视化调试,并用 SDK 做集成测试。
每个模块都有一个 studio/ 子目录,这是 LangGraph API 服务器的根目录。结构如下:
// module-1/studio/langgraph.json
{
"dependencies": ["."], // 依赖当前目录(读取 requirements.txt)
"graphs": {
"agent": "./agent.py:graph", // 注册图:graph_id → 文件:变量名
"router": "./router.py:graph" // 可以注册多个图
},
"env": ".env" // 环境变量文件路径
}
langgraph.json 的 graphs 字段中,键名(如 "agent")就是 graph_id。SDK 调用 runs.stream() 时传入的第二个参数就是这个 id。注册多个图时,每个都有自己的 id。
# 在 studio/ 目录下运行
$ langgraph dev
# 成功启动后输出:
🚀 API: http://127.0.0.1:2024
🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
注意 Studio 的 URL 带有参数 ?baseUrl=http://127.0.0.1:2024。这意味着 Studio 是一个运行在 LangSmith 服务器上的前端应用,它通过你提供的 baseUrl 连接你 本机 的 API。你的图代码从未离开本地——Studio 只是一个可视化前端。
LangGraph SDK(pip install langgraph-sdk)是 Python 客户端库,让你在代码中与 LangGraph API 交互。核心入口是 get_client():
from langgraph_sdk import get_client
# 连接本地开发服务器
URL = "http://127.0.0.1:2024"
client = get_client(url=URL)
# 连接云端部署(切换一行代码即可)
# URL = "https://your-deployment-xxxx.us.langgraph.app"
# client = get_client(url=URL, api_key="lsv2_...")
client 是整个 SDK 的操作入口,通过它访问所有资源:
from langgraph_sdk import get_client
from langchain_core.messages import HumanMessage
# 1. 建立连接
URL = "http://127.0.0.1:2024"
client = get_client(url=URL)
# 2. 查看已注册的图(对应 langgraph.json 里的 graphs)
assistants = await client.assistants.search()
print(assistants)
# [{'assistant_id': '...', 'graph_id': 'agent', ...},
# {'assistant_id': '...', 'graph_id': 'router', ...}]
# 3. 创建一个对话线程(对应一个持久化的 State 容器)
thread = await client.threads.create()
print(thread['thread_id']) # 类似 UUID 的字符串
# 4. 在这个线程上发起流式运行
input = {"messages": [HumanMessage(content="Multiply 3 by 2.")]}
async for chunk in client.runs.stream(
thread['thread_id'], # 线程 ID
"agent", # graph_id(对应 langgraph.json 中的键)
input=input,
stream_mode="values", # 每步后返回完整 State
):
if chunk.data and chunk.event != "metadata":
print(chunk.data['messages'][-1])
理解这三个概念是用好 SDK 的基础。它们是 LangGraph API 的资源模型,类似于数据库里的三张表。
对应一个已注册的图(graph_id)。可以理解为图的一个"配置实例",可以有不同版本或不同系统提示词,但底层共享同一个图结构。
每次 langgraph dev 启动后,langgraph.json 里每个图都会自动生成一个 Assistant。
对应一个持久化的对话会话。有唯一的 thread_id,跨多次 Run 共享同一个 State。
类比:Thread 就像微信里的一个聊天窗口——每次发消息(Run)都在同一个上下文里累积。
对应一次具体的图运行。在某个 Thread 上,用特定输入调用某个 Assistant,就产生一个 Run。
Run 可以 stream(流式),也可以 wait(等待完成)。每个 Run 的结果会自动写回 Thread 的 State。
| 概念 | 类比 | 关键属性 | 生命周期 |
|---|---|---|---|
| Assistant | 一个已配置好的 Agent 服务 | assistant_id、graph_id |
随服务存在,长期有效 |
| Thread | 一个对话会话 / 聊天窗口 | thread_id、State 快照 |
手动创建,可跨多个 Run |
| Run | 一次"发送消息"的动作 | run_id、status、输出 chunks |
单次执行,完成后结束 |
一个 Thread 里可以有多个 Run,每个 Run 结束后的 State 会被持久化。下次 Run 开始时,会从上次结束的 State 继续(这就是 Memory 的实现原理)。如果你想开始一个全新的对话,需要创建新 Thread,而不是新 Run。
client.runs.stream() 返回一个异步迭代器,每次迭代得到一个 chunk 对象。chunk 有两个关键字段:chunk.event 和 chunk.data。
chunk.data 包含 run_id。通常跳过(chunk.event != "metadata")。chunk.data 就是完整的 State 字典,如 {"messages": [...]}。chunk.data 为空。收到后可以停止监听。| stream_mode | 每次推送的内容 | 适用场景 |
|---|---|---|
"values" |
每步执行后,推送完整的当前 State(包含所有字段) | 需要展示完整对话历史、监控整体状态 |
"updates" |
每步执行后,只推送本步的 State 变更(delta) | 减少传输数据量,只关心新增内容 |
"messages" |
LLM 生成 token 时逐 token 推送(类似 ChatGPT 的打字效果) | 需要流式展示 LLM 文字输出 |
"debug" |
详细的调试信息,每个内部事件都推送 | 深度调试、排查问题 |
# chunk.event == "values" 时,chunk.data 就是完整 State:
chunk.data = {
"messages": [
HumanMessage(content="Multiply 3 by 2."),
AIMessage(content="", tool_calls=[{
"name": "multiply",
"args": {"a": 3, "b": 2}
}]),
ToolMessage(content="6"),
AIMessage(content="3 multiplied by 2 equals 6.")
]
}
# 每步执行后都会收到一次这样的 chunk,messages 列表逐步增长
# 所以用 chunk.data['messages'][-1] 只看最新的消息
用 stream_mode="values" 时,每步都会收到完整 State,消息列表会累积增长。用 chunk.data['messages'][-1] 获取最新消息,用 chunk.event != "metadata" 过滤掉元数据帧,这是最常见的处理模式。
LangGraph SDK 最重要的设计理念是本地开发和云端生产使用完全相同的 API。唯一的差异是连接时的 URL 和认证 key。
from langgraph_sdk import get_client
# 本地:只需 URL
client = get_client(
url="http://127.0.0.1:2024"
)
# 以下代码完全相同 ↓
thread = await client.threads.create()
async for chunk in client.runs.stream(
thread['thread_id'],
"agent",
input=input,
stream_mode="values",
):
...
from langgraph_sdk import get_client
# 云端:URL + API Key
client = get_client(
url="https://xxx.us.langgraph.app",
api_key="lsv2_pt_..."
)
# 以下代码完全相同 ↓
thread = await client.threads.create()
async for chunk in client.runs.stream(
thread['thread_id'],
"agent",
input=input,
stream_mode="values",
):
...
你可以在本地用 langgraph dev 开发调试,代码写好后,只改一行 URL 就能切换到云端生产环境。所有业务逻辑代码无需修改。这是部署体系设计的精髓。
| 维度 | 本地(langgraph dev) | 云端(LangSmith Deployment) |
|---|---|---|
| URL 格式 | http://127.0.0.1:2024 |
https://xxx.us.langgraph.app |
| 认证 | 无需(本地开发) | 需要 api_key(LangSmith API Key) |
| 代码来源 | 本地文件系统,修改即时生效 | GitHub 仓库,push 后触发部署 |
| 持久化 | 内存或本地 SQLite,重启会丢失 | 云端持久化存储,永久保存 |
| Studio 访问 | 需要本地服务运行中 | 随时可访问,无需本地环境 |
| SDK 调用代码 | 完全相同(只有 URL 和 api_key 不同) | |
当你的 Agent 在本地测试通过后,可以将它部署到 LangSmith 云端,获得稳定的线上服务地址。
将项目 push 到 GitHub。确保 studio/ 目录包含 langgraph.json、图定义文件和 requirements.txt。不要把 .env 文件 commit 进去。
登录 smith.langchain.com,进入 Deployments 页面,点击 "New Deployment"。
连接你的 GitHub 账号,选择仓库,指定 langgraph.json 的路径(如 module-1/studio/langgraph.json)。平台会自动读取图配置。
在部署界面填写 OPENAI_API_KEY 等环境变量。这相当于云端的 .env 文件,安全地存储在 LangSmith 平台。
平台自动拉取代码、构建环境、启动服务。完成后得到唯一 URL,格式类似 https://xxx.us.langgraph.app。用这个 URL 替换 SDK 里的 http://127.0.0.1:2024 即可。
每个 Deployment 都有独立且稳定的 URL,不会因为重新部署而改变。LangSmith 支持多个并行的 Deployment(如 staging、production),每个环境都有自己的 URL。
把所有知识点串联起来,看看一个完整的 LangGraph 开发到部署的流程:
在 studio/agent.py 里定义 StateGraph,添加节点、边,编译为 graph 对象。这是 Module 1 前五讲的内容。
在配置文件里给图起一个 graph_id(如 "agent"),指向 agent.py 里的 graph 变量。
在浏览器里可视化调试图,用 Studio 手动测试各种输入,确保图的行为符合预期。
写 Python 代码,通过 SDK 连接本地 API,测试 Thread 创建、Run 发起、streaming 接收。确认代码逻辑正确。
Push 代码到 GitHub,在 LangSmith 创建 Deployment,获得云端 URL。SDK 代码里只改 URL 和 api_key,其余业务逻辑完全不变。
LangGraph 部署体系的设计核心是分层解耦:你只负责写图逻辑(LangGraph 库),LangGraph API 负责把它变成可访问的服务,SDK 负责提供统一的访问接口,Studio 负责可视化调试,LangSmith Deployment 负责云端托管。每一层职责清晰,可以独立替换,这让 Agent 的开发体验极大提升。
| 知识点 | 要记住的内容 |
|---|---|
| 启动本地服务 | langgraph dev,在 studio/ 目录下运行 |
| 本地 API 地址 | http://127.0.0.1:2024 |
| 注册图的地方 | langgraph.json 的 graphs 字段,键名就是 graph_id |
| SDK 入口 | from langgraph_sdk import get_client |
| 三大资源 | Assistants(图的实例) / Threads(对话会话) / Runs(单次执行) |
| stream_mode 推荐 | "values":每步后返回完整 State,用 [-1] 取最新消息 |
| 跳过元数据帧 | chunk.event != "metadata" 且 chunk.data 不为空 |
| 切换到云端 | 只改 url 和 api_key,业务代码不变 |