LangGraph Deployment 详细解读 🚀

📚 概述

本文档详细解读 LangGraph 的**部署（Deployment）**流程。部署是将我们构建的 Agent 从本地开发环境推向生产环境的关键步骤。通过 LangGraph Cloud 和 LangGraph Studio，我们可以轻松地测试、监控和管理我们的 AI 应用。

🎯 核心概念

为什么需要部署？

在前面的课程中，我们已经学会了：

✅ 构建带工具的 Agent（act）
✅ 处理工具输出（observe）
✅ 让 Agent 进行推理（reason）
✅ 使用内存保持状态（persist state）

但这些都是在本地 Jupyter Notebook 中运行的。要让 Agent 真正为用户服务，我们需要：

🌐 可访问性：通过 API 让其他应用调用
📊 可观测性：监控运行状态和性能
🔄 可扩展性：支持多用户并发访问
🛡️ 可靠性：处理错误、重试、持久化

这就是部署的意义！

🏗️ LangGraph 生态架构

核心组件关系图

┌─────────────────────────────────────────────────────────┐
│                    开发者 (You)                          │
└─────────────────────────────────────────────────────────┘
                        │
        ┌───────────────┼───────────────┐
        ↓               ↓               ↓
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│  LangGraph   │ │  LangGraph   │ │  LangGraph   │
│   Library    │ │    Studio    │ │     SDK      │
│ (构建图代码)   │ │  (可视化IDE)  │ │  (程序化调用)  │
└──────────────┘ └──────────────┘ └──────────────┘
        │               │               │
        └───────────────┼───────────────┘
                        ↓
┌─────────────────────────────────────────────────────────┐
│                  LangGraph API                           │
│        (任务队列 + 持久化 + 异步处理)                      │
└─────────────────────────────────────────────────────────┘
                        │
        ┌───────────────┴───────────────┐
        ↓                               ↓
┌──────────────┐                ┌──────────────┐
│ 本地运行      │                │ LangGraph    │
│ (开发测试)    │                │   Cloud      │
└──────────────┘                │  (生产部署)   │
                               └──────────────┘

1. LangGraph Library

作用： 核心 Python/JavaScript 库，用于构建 Agent 工作流

python

from langgraph.graph import StateGraph, START, END

# 使用 LangGraph 定义图
graph = StateGraph(State)
graph.add_node("agent", call_model)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
app = graph.compile()

特点：

提供图构建 API
定义状态管理
支持条件边、循环等高级功能

2. LangGraph API

作用： 将图代码打包为可部署的服务

核心功能：

🔄 任务队列：管理异步操作，处理并发请求
💾 持久化：保存对话历史和状态（跨会话）
🔌 HTTP API：提供 RESTful 接口

为什么需要它？

直接运行图代码有限制：

python

# 本地直接运行 - 只能单次执行
app = graph.compile()
result = app.invoke({"messages": [HumanMessage("Hello")]})

使用 LangGraph API：

python

# 通过 API - 支持流式、异步、多线程
async for chunk in client.runs.stream(
    thread_id="user-123",
    assistant_id="agent",
    input={"messages": [HumanMessage("Hello")]}
):
    print(chunk)

3. LangGraph Cloud

作用： LangGraph API 的托管服务

核心特性：

☁️ 从 GitHub 部署：推送代码自动部署
🔍 监控和追踪：查看每次运行的详细日志
🌐 唯一 URL：每个部署获得专属访问地址
🔑 API 密钥管理：安全管理环境变量

部署流程：

GitHub Repo → LangSmith → LangGraph Cloud → 生产 URL

4. LangGraph Studio

作用： 可视化 IDE，用于测试和调试图

两种运行模式：

本地模式（推荐用于开发）

bash

# 在项目目录运行
langgraph dev

# 输出：
# 🚀 API: http://127.0.0.1:2024
# 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
# 📚 API Docs: http://127.0.0.1:2024/docs

特点：

实时预览图结构
交互式测试对话
查看每步的状态变化
后端使用本地 LangGraph API

云模式（用于测试生产部署）

连接到 LangGraph Cloud 部署
在线协作和分享
访问生产环境数据

5. LangGraph SDK

作用： Python 客户端库，用于程序化交互

python

from langgraph_sdk import get_client

# 创建客户端（本地或云端）
client = get_client(url="http://127.0.0.1:2024")

# 列出所有助手
assistants = await client.assistants.search()

# 创建对话线程
thread = await client.threads.create()

# 运行图
async for chunk in client.runs.stream(
    thread['thread_id'],
    "agent",
    input={"messages": [...]},
    stream_mode="values"
):
    process(chunk)

主要 API：

client.assistants - 管理助手（图定义）
client.threads - 管理对话线程（状态容器）
client.runs - 执行图运行（调用逻辑）

🛠️ 本地部署实战

前置准备

1. 项目结构

确保你的项目包含以下文件：

module-1/
├── studio/
│   ├── agent.py              # 图定义代码
│   ├── langgraph.json        # 配置文件
│   └── .env                  # 环境变量（不要提交到 Git！）
└── deployment.ipynb          # 测试代码

2. langgraph.json 配置文件

json

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:graph"
  },
  "env": ".env"
}

字段说明：

dependencies：Python 包依赖路径
graphs：图定义，格式为 "图ID": "文件路径:变量名"
env：环境变量文件路径

3. agent.py 示例

python

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 定义工具
@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 定义节点
def call_model(state: State):
    llm = ChatOpenAI(model="gpt-4o")
    llm_with_tools = llm.bind_tools([multiply])
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("agent", call_model)
graph_builder.add_edge(START, "agent")
graph_builder.add_edge("agent", END)
graph = graph_builder.compile()

启动本地开发服务器

步骤 1：进入项目目录

bash

cd /Users/brycewang/langchain-academy/module-1/studio

步骤 2：启动服务

bash

langgraph dev

输出示例：

✓ Configuration loaded from langgraph.json
✓ Environment variables loaded from .env
✓ Graph 'agent' loaded successfully

- 🚀 API: http://127.0.0.1:2024
- 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
- 📚 API Docs: http://127.0.0.1:2024/docs

Watching for changes...

重要提示：

保持终端运行（不要关闭！）
代码修改会自动重载
访问 API Docs 查看完整 API 文档

步骤 3：在浏览器打开 Studio

访问：https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

Studio 界面功能：

📊 图可视化：查看节点和边的连接
💬 交互式测试：输入消息，查看 Agent 响应
🔍 状态检查：每步执行后查看状态变化
🐛 调试工具：查看错误堆栈和日志

💻 使用 SDK 调用本地 API

安装依赖

python

%pip install --quiet -U langgraph_sdk langchain_core

代码示例

1. 连接到本地服务器

python

from langgraph_sdk import get_client

# 本地开发服务器地址
URL = "http://127.0.0.1:2024"
client = get_client(url=URL)

Python 知识点：async/await

LangGraph SDK 使用异步编程：

python

# ❌ 错误 - 同步调用会报错
assistants = client.assistants.search()

# ✅ 正确 - 异步调用
assistants = await client.assistants.search()

为什么使用异步？

非阻塞 I/O，提高并发性能
支持流式响应
更好地处理长时间运行的任务

如何在 Jupyter 中使用？ Jupyter Notebook 默认支持顶层 await：

python

# 直接使用 await（仅在 Notebook 中）
result = await client.runs.stream(...)

# 在普通 Python 脚本中需要：
import asyncio
asyncio.run(main())

2. 搜索可用的助手（图）

python

# 搜索所有已部署的图
assistants = await client.assistants.search()

# 查看第一个助手
print(assistants[0])

输出示例：

python

{
    'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca',
    'graph_id': 'agent',
    'config': {},
    'metadata': {'created_by': 'system'},
    'name': 'agent',
    'created_at': '2025-03-04T22:57:28.424565+00:00',
    'updated_at': '2025-03-04T22:57:28.424565+00:00',
    'version': 1
}

字段说明：

assistant_id：唯一标识符（UUID）
graph_id：图的名称（来自 langgraph.json）
config：配置选项（如模型参数）
version：版本号（支持多版本部署）

3. 创建对话线程

python

# 创建新线程用于跟踪状态
thread = await client.threads.create()

print(thread['thread_id'])
# 输出：'1ef6c123-4567-89ab-cdef-0123456789ab'

LangGraph 知识点：Thread（线程）

Thread 是 LangGraph 中的状态容器：

Thread = 独立的对话会话 = 独立的状态快照

为什么需要 Thread？

支持多用户并发（每个用户一个 Thread）
持久化对话历史（跨请求保持状态）
隔离不同会话（避免状态混淆）

示例场景：

python

# 用户 A 的对话
thread_a = await client.threads.create()
client.runs.stream(thread_a['thread_id'], "agent", input={"messages": [...]})

# 用户 B 的对话（完全独立）
thread_b = await client.threads.create()
client.runs.stream(thread_b['thread_id'], "agent", input={"messages": [...]})

4. 运行图并流式输出

python

from langchain_core.messages import HumanMessage

# 输入消息
input = {"messages": [HumanMessage(content="Multiply 3 by 2.")]}

# 流式运行图
async for chunk in client.runs.stream(
    thread['thread_id'],     # 线程 ID
    "agent",                 # 图 ID
    input=input,             # 输入数据
    stream_mode="values",    # 流模式
):
    if chunk.data and chunk.event != "metadata":
        print(chunk.data['messages'][-1])

参数详解：

参数	说明	示例
`thread_id`	线程标识符	`'1ef6c123-...'`
`assistant_id`	助手 ID（图名称）	`'agent'`
`input`	输入状态	`{"messages": [...]}`
`stream_mode`	流模式	`"values"` / `"updates"`

LangGraph 知识点：Stream Modes

LangGraph 支持多种流模式：

stream_mode="values" - 完整状态流

python

# 每步后返回完整状态
{'messages': [msg1]}
{'messages': [msg1, msg2]}
{'messages': [msg1, msg2, msg3]}

stream_mode="updates" - 增量更新流

python

# 每步只返回新增的状态
{'messages': [msg1]}
{'messages': [msg2]}
{'messages': [msg3]}

stream_mode="messages" - 消息流（仅消息）

python

# 直接返回消息对象
AIMessage(content="...")
ToolMessage(content="...")

运行输出示例：

python

# 第 1 步：用户消息
{'content': 'Multiply 3 by 2.', 'type': 'human', 'id': 'cdbd7bd8-...'}

# 第 2 步：AI 调用工具
{'content': '', 'type': 'ai', 'tool_calls': [{'name': 'multiply', 'args': {'a': 3, 'b': 2}}]}

# 第 3 步：工具执行结果
{'content': '6', 'type': 'tool', 'name': 'multiply', 'tool_call_id': 'call_iIPry...'}

# 第 4 步：AI 最终响应
{'content': 'The result of multiplying 3 by 2 is 6.', 'type': 'ai'}

代码解析：

python

async for chunk in client.runs.stream(...):
    # chunk 是流式响应的片段

    if chunk.data and chunk.event != "metadata":
        # chunk.data：包含状态数据
        # chunk.event：事件类型（如 "values", "metadata"）

        print(chunk.data['messages'][-1])
        # 打印最新消息（列表最后一项）

☁️ 云端部署实战

部署流程概览

1. GitHub 创建仓库
         ↓
2. 推送代码到 GitHub
         ↓
3. LangSmith 连接仓库
         ↓
4. 配置部署参数
         ↓
5. 自动部署到 LangGraph Cloud
         ↓
6. 获得生产 URL

步骤 1：创建 GitHub 仓库

1.1 在 GitHub 创建新仓库

登录 GitHub
点击右上角 "+" → "New repository"
仓库名称：langchain-academy
设置为 Private（推荐，避免泄露 API 密钥）
不要初始化 README、.gitignore 或 License

1.2 添加远程仓库

bash

# 进入本地项目目录
cd /Users/brycewang/langchain-academy

# 添加远程仓库
git remote add origin https://github.com/your-username/langchain-academy.git

# 查看远程仓库
git remote -v

Git 知识点：remote

bash

git remote add <名称> <URL>
#              ^^^^^  ^^^^^^
#              别名    仓库地址

# 常见别名：
# origin - 默认主仓库
# upstream - 上游仓库（fork 时使用）

1.3 推送代码

bash

# 确保所有文件已提交
git add .
git commit -m "Add LangGraph agent"

# 推送到 GitHub
git push -u origin main

重要警告：不要提交 .env 文件！

确保 .gitignore 包含：

.env
*.env
.env.local

验证：

bash

# 检查是否误提交了敏感文件
git log --all --full-history -- "*.env"

步骤 2：在 LangSmith 配置部署

2.1 访问 LangSmith

打开 LangSmith
登录（使用与 LangChain 相同的账号）

2.2 创建新部署

点击左侧导航栏的 "Deployments" 标签
点击 "+ New Deployment" 按钮
选择 "Deploy from GitHub"

2.3 连接 GitHub 仓库

授权 GitHub：
- 首次使用需要授权 LangSmith 访问 GitHub
- 选择授权所有仓库或特定仓库
选择仓库：
- Repository：your-username/langchain-academy
- Branch：main
配置图路径：
- LangGraph API config file：module-1/studio/langgraph.json
- 这是 langgraph.json 文件在仓库中的路径

2.4 设置环境变量

为什么需要？

云端不能访问你的本地 .env 文件
需要手动配置 API 密钥和其他环境变量

如何设置？

方法 1：从本地 .env 复制

bash

# 查看本地环境变量
cat module-1/studio/.env

# 输出示例：
# OPENAI_API_KEY=sk-proj-...
# TAVILY_API_KEY=tvly-...

方法 2：在 LangSmith UI 中添加

点击 "Environment Variables"

添加每个变量：

Name: OPENAI_API_KEY
Value: sk-proj-...

Name: TAVILY_API_KEY
Value: tvly-...

安全提示：

✅ 使用专门的生产环境密钥
✅ 定期轮换密钥
❌ 不要在代码中硬编码密钥
❌ 不要在日志中打印密钥

2.5 部署

检查配置无误
点击 "Deploy" 按钮
等待部署完成（通常 1-2 分钟）

部署过程：

[1/5] 克隆 GitHub 仓库
[2/5] 安装 Python 依赖
[3/5] 加载图定义
[4/5] 运行健康检查
[5/5] 分配 URL

2.6 获取部署 URL

部署成功后，你会获得一个唯一的 URL：

https://langchain-academy-8011c561878d50b1883f7ed11b32d720.default.us.langgraph.app

URL 格式：

https://<project-name>-<deployment-id>.<region>.langgraph.app
        ^^^^^^^^^^^^^^  ^^^^^^^^^^^^^^^  ^^^^^^^^
        项目名称         唯一标识符        区域

步骤 3：使用 SDK 调用云端部署

3.1 设置 LangSmith API 密钥

python

import os
import getpass

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

# 设置 LangSmith API 密钥（用于身份验证）
_set_env("LANGSMITH_API_KEY")

Python 知识点：getpass 模块

getpass 用于安全输入密码：

python

import getpass

# 普通 input - 密码会显示
password = input("Password: ")  # 输入: abc123 → 屏幕显示 abc123

# getpass - 密码隐藏
password = getpass.getpass("Password: ")  # 输入: abc123 → 屏幕显示 ******

为什么需要 LANGSMITH_API_KEY？

云端部署需要身份验证
防止未授权访问
关联到你的 LangSmith 账户

如何获取？

访问 LangSmith Settings
点击 "API Keys"
创建新密钥或复制现有密钥

3.2 连接到云端服务

python

from langgraph_sdk import get_client

# 替换为你的部署 URL
URL = "https://langchain-academy-8011c561878d50b1883f7ed11b32d720.default.us.langgraph.app"
client = get_client(url=URL)

# 搜索助手
assistants = await client.assistants.search()

重要提示：

云端 URL 和本地 URL 使用相同的 SDK
切换环境只需更改 url 参数
API 调用方式完全一致

3.3 运行云端 Agent

python

from langchain_core.messages import HumanMessage

# 选择助手
agent = assistants[0]

# 创建线程
thread = await client.threads.create()

# 输入消息
input = {"messages": [HumanMessage(content="Multiply 3 by 2.")]}

# 流式运行（与本地完全相同！）
async for chunk in client.runs.stream(
    thread['thread_id'],
    "agent",
    input=input,
    stream_mode="values",
):
    if chunk.data and chunk.event != "metadata":
        print(chunk.data['messages'][-1])

输出示例：

python

{'content': 'Multiply 3 by 2.', 'type': 'human', 'id': '8ea04559-...'}
{'content': '', 'type': 'ai', 'tool_calls': [{'name': 'multiply', 'args': {'a': 3, 'b': 2}}]}
{'content': '6', 'type': 'tool', 'name': 'multiply', 'tool_call_id': 'call_EQool...'}
{'content': '3 multiplied by 2 equals 6.', 'type': 'ai'}

步骤 4：在 Studio 中访问云端部署

4.1 打开 LangGraph Studio

访问 LangGraph Studio
登录你的 LangSmith 账户

4.2 选择云端部署

在顶部切换到 "Cloud" 模式
选择你的部署（如 langchain-academy）

4.3 交互式测试

Studio 提供与本地相同的功能：

可视化图结构
输入消息测试
查看状态变化
监控性能指标

云端 Studio 的额外功能：

📊 运行历史：查看所有历史运行记录
🔍 追踪详情：每个 LLM 调用的 token 使用、延迟等
👥 团队协作：分享给团队成员
📈 使用统计：API 调用量、费用等

🎓 核心知识点总结

LangGraph 特有概念

1. Deployment（部署）

定义： 将 LangGraph 应用从开发环境迁移到生产环境的过程

关键组件：

langgraph.json：配置文件，定义图路径和依赖
LangGraph API：提供 HTTP 接口和任务队列
Thread：状态容器，支持多用户会话
Assistant：图的运行时实例

2. Thread vs Assistant

概念	作用	类比	持久化
Assistant	图定义（代码）	程序的可执行文件	是
Thread	状态实例（数据）	程序的运行进程	是

python

# 一个 Assistant（图）可以有多个 Thread（会话）
assistant_id = "agent"

thread1 = await client.threads.create()  # 用户 A 的会话
thread2 = await client.threads.create()  # 用户 B 的会话

# 同一个 Assistant，不同的 Thread
client.runs.stream(thread1['thread_id'], assistant_id, ...)
client.runs.stream(thread2['thread_id'], assistant_id, ...)

3. Stream Modes 对比

python

# 场景 1：需要完整状态（如 UI 渲染）
stream_mode="values"
# → {'messages': [msg1, msg2, msg3]}

# 场景 2：只关心新增数据（如增量更新）
stream_mode="updates"
# → {'messages': [msg3]}

# 场景 3：只处理消息（如聊天应用）
stream_mode="messages"
# → AIMessage(content="...")

4. 本地 vs 云端部署

特性	本地开发 (`langgraph dev`)	云端部署 (LangGraph Cloud)
用途	开发和测试	生产环境
URL	`http://127.0.0.1:2024`	`https://xxx.langgraph.app`
环境变量	`.env` 文件	LangSmith UI 配置
热重载	✅ 支持	❌ 需要重新部署
持久化	内存（重启丢失）	数据库（永久）
监控	基础日志	完整追踪和分析
成本	免费	按使用量计费

Python 特有知识点

1. 异步编程（async/await）

为什么使用异步？

同步调用（阻塞）：

python

# 调用 API → 等待响应 → 继续执行
result1 = call_api_1()  # 等待 2 秒
result2 = call_api_2()  # 等待 2 秒
# 总耗时：4 秒

异步调用（非阻塞）：

python

# 同时发起多个请求，并发等待
result1 = await call_api_1()  # 异步等待 2 秒
result2 = await call_api_2()  # 异步等待 2 秒
# 总耗时：2 秒（并发执行）

基础语法：

python

# 定义异步函数
async def fetch_data():
    result = await some_async_operation()
    return result

# 调用异步函数
# 方式 1：在 Jupyter Notebook 中
data = await fetch_data()

# 方式 2：在普通 Python 脚本中
import asyncio
data = asyncio.run(fetch_data())

LangGraph SDK 中的应用：

python

# ✅ 所有 SDK 方法都是异步的
await client.assistants.search()
await client.threads.create()
async for chunk in client.runs.stream(...):
    process(chunk)

2. 异步迭代器（async for）

普通迭代器：

python

# 同步迭代
for item in [1, 2, 3]:
    print(item)

异步迭代器：

python

# 异步迭代
async for chunk in async_generator():
    print(chunk)

在 LangGraph 中的使用：

python

# client.runs.stream() 返回异步生成器
async for chunk in client.runs.stream(thread_id, assistant_id, input=...):
    # 每次 yield 一个新的状态片段
    if chunk.data:
        print(chunk.data)

等价的同步写法（理解用）：

python

# 如果 stream() 是同步的，代码会是这样：
for chunk in client.runs.stream_sync(...):  # 假设有同步版本
    print(chunk.data)

3. getpass 安全输入

python

import getpass

# 安全输入密码（不显示）
api_key = getpass.getpass("API Key: ")

# 常见使用模式
def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")

安全性优势：

输入不会显示在屏幕上
不会被日志记录
防止肩窥攻击（Shoulder Surfing）

4. 环境变量管理

读取环境变量：

python

import os

# 方式 1：直接获取（可能返回 None）
api_key = os.environ.get("OPENAI_API_KEY")

# 方式 2：必须存在（不存在会报错）
api_key = os.environ["OPENAI_API_KEY"]

# 方式 3：提供默认值
api_key = os.environ.get("OPENAI_API_KEY", "default-key")

设置环境变量：

python

# 方式 1：代码中设置（临时，仅当前进程）
os.environ["OPENAI_API_KEY"] = "sk-proj-..."

# 方式 2：.env 文件（推荐）
# 文件：.env
# 内容：OPENAI_API_KEY=sk-proj-...

# 加载 .env 文件
from dotenv import load_dotenv
load_dotenv()

最佳实践：

python

# .env 文件
OPENAI_API_KEY=sk-proj-...
TAVILY_API_KEY=tvly-...
LANGSMITH_API_KEY=lsv2_...

# Python 代码
from dotenv import load_dotenv
import os

load_dotenv()  # 加载 .env

# 验证必需的环境变量
required_vars = ["OPENAI_API_KEY", "TAVILY_API_KEY"]
for var in required_vars:
    if not os.environ.get(var):
        raise ValueError(f"Missing required environment variable: {var}")

💡 最佳实践

1. 项目结构规范

推荐结构：

my-langgraph-project/
├── .gitignore              # 忽略敏感文件
├── README.md               # 项目说明
├── requirements.txt        # Python 依赖
├── .env.example            # 环境变量模板（不含真实密钥）
├── src/
│   ├── agent.py           # 图定义
│   ├── tools.py           # 工具定义
│   └── utils.py           # 工具函数
├── tests/
│   └── test_agent.py      # 单元测试
└── deployment/
    ├── langgraph.json     # 部署配置
    └── .env               # 环境变量（不提交！）

.gitignore 必备内容：

# 环境变量
.env
.env.local
*.env

# Python
__pycache__/
*.pyc
.venv/
venv/

# LangGraph
.langraph/

2. 环境变量管理

.env.example 模板

bash

# OpenAI API
OPENAI_API_KEY=your_openai_key_here

# Tavily Search API
TAVILY_API_KEY=your_tavily_key_here

# LangSmith (for tracing)
LANGSMITH_API_KEY=your_langsmith_key_here
LANGSMITH_PROJECT=my-project

使用流程：

复制 .env.example 为 .env
填入真实 API 密钥
.env 不提交到 Git（通过 .gitignore）

多环境配置

.env.development    # 开发环境
.env.staging        # 预发布环境
.env.production     # 生产环境

加载指定环境：

python

import os
from dotenv import load_dotenv

env = os.getenv("ENV", "development")
load_dotenv(f".env.{env}")

3. 版本管理

langgraph.json 版本控制

json

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:graph"
  },
  "env": ".env",
  "version": "1.0.0"
}

Git 标签（Tag）

bash

# 创建版本标签
git tag -a v1.0.0 -m "Release version 1.0.0"

# 推送标签到 GitHub
git push origin v1.0.0

# LangSmith 部署时可以指定标签
# 部署配置 → Git Ref → v1.0.0

4. 监控和调试

启用 LangSmith 追踪

python

import os

# 启用追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-agent-deployment"

# 运行 Agent
# 所有调用会自动记录到 LangSmith

查看追踪：

访问 LangSmith
选择项目 my-agent-deployment
查看每次运行的详细日志：
- LLM 调用（提示词、响应、token 使用）
- 工具调用（参数、结果）
- 状态变化（每步的状态快照）

本地调试技巧

python

# 添加详细日志
import logging
logging.basicConfig(level=logging.DEBUG)

# 在节点中打印状态
def call_model(state: State):
    print(f"Current state: {state}")  # 调试日志
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

5. 错误处理

优雅的错误恢复

python

def call_model(state: State):
    try:
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    except Exception as e:
        # 记录错误
        logging.error(f"Error in call_model: {e}")

        # 返回错误消息
        return {"messages": [AIMessage(content=f"Sorry, an error occurred: {str(e)}")]}

重试机制

python

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_model_with_retry(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

6. 性能优化

使用流式响应

python

# ❌ 等待完整响应（慢）
response = await client.runs.invoke(thread_id, assistant_id, input=...)

# ✅ 流式获取（快，用户体验好）
async for chunk in client.runs.stream(thread_id, assistant_id, input=...):
    display(chunk)  # 实时显示

缓存常用结果

python

from functools import lru_cache

@lru_cache(maxsize=100)
def expensive_computation(input_data):
    # 昂贵的计算
    return result

🚀 进阶技巧

1. 多版本部署

场景： 同时运行多个版本的 Agent（如 A/B 测试）

实现：

python

# 部署 v1
await client.assistants.create(
    graph_id="agent",
    config={"model": "gpt-4o"},
    version="v1"
)

# 部署 v2
await client.assistants.create(
    graph_id="agent",
    config={"model": "gpt-4o-mini"},
    version="v2"
)

# 用户随机分配版本
import random
version = random.choice(["v1", "v2"])
await client.runs.stream(thread_id, f"agent-{version}", input=...)

2. 自定义配置

langgraph.json 高级配置：

json

{
  "dependencies": ["."],
  "graphs": {
    "agent": {
      "module": "./agent.py",
      "graph": "graph",
      "config": {
        "model": "gpt-4o",
        "temperature": 0.7,
        "max_tokens": 1000
      }
    }
  },
  "env": ".env",
  "python_version": "3.11"
}

在代码中使用配置：

python

from langgraph.graph import StateGraph
from langgraph.config import get_config

def call_model(state: State):
    config = get_config()
    model_name = config.get("model", "gpt-4o")
    temperature = config.get("temperature", 0)

    llm = ChatOpenAI(model=model_name, temperature=temperature)
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

3. 中断和恢复

使用场景： 需要人工审批的工作流

python

from langgraph.checkpoint.memory import MemorySaver

# 编译时添加 checkpointer
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 运行到中断点
config = {"configurable": {"thread_id": "thread-1"}}
result = app.invoke(input, config=config)

# 稍后恢复（从上次状态继续）
result = app.invoke(None, config=config)

4. 集成到生产应用

FastAPI 集成

python

from fastapi import FastAPI
from langgraph_sdk import get_client

app = FastAPI()
client = get_client(url="http://127.0.0.1:2024")

@app.post("/chat")
async def chat(message: str, thread_id: str):
    input_data = {"messages": [HumanMessage(content=message)]}

    response = ""
    async for chunk in client.runs.stream(
        thread_id,
        "agent",
        input=input_data,
        stream_mode="values"
    ):
        if chunk.data:
            response = chunk.data['messages'][-1]['content']

    return {"response": response}

Streamlit 前端

python

import streamlit as st
from langgraph_sdk import get_client

st.title("My AI Assistant")

# 初始化客户端
client = get_client(url="http://127.0.0.1:2024")

# 创建或获取 thread
if "thread_id" not in st.session_state:
    thread = asyncio.run(client.threads.create())
    st.session_state.thread_id = thread['thread_id']

# 聊天界面
user_input = st.text_input("You:", "")

if user_input:
    input_data = {"messages": [HumanMessage(content=user_input)]}

    async for chunk in client.runs.stream(
        st.session_state.thread_id,
        "agent",
        input=input_data,
        stream_mode="values"
    ):
        if chunk.data:
            st.write("AI:", chunk.data['messages'][-1]['content'])

🔍 常见问题

Q1: 本地开发时，修改代码后需要重启服务器吗？

答：不需要！langgraph dev 支持热重载。

bash

# 启动服务器
langgraph dev

# 修改 agent.py
# 保存文件后，终端会显示：
✓ Detected changes in agent.py
✓ Reloading graph...
✓ Graph reloaded successfully

注意： 如果修改了 langgraph.json 或 .env，需要手动重启。

Q2: 如何在本地和云端之间切换？

答：只需修改 URL 参数。

python

# 本地
client = get_client(url="http://127.0.0.1:2024")

# 云端
client = get_client(url="https://xxx.langgraph.app")

# 其他代码完全相同！

最佳实践： 使用环境变量

python

import os

URL = os.getenv("LANGGRAPH_URL", "http://127.0.0.1:2024")
client = get_client(url=URL)

Q3: Thread 的状态会持久化多久？

本地：

使用内存存储（MemorySaver）
重启服务器后丢失
适合开发测试

云端：

使用数据库存储
默认保留 30 天
可配置保留时间

查询 Thread 状态：

python

# 获取 Thread 的历史记录
thread_state = await client.threads.get_state(thread_id)
print(thread_state.values)  # 当前状态
print(thread_state.next)    # 下一步节点

Q4: 如何调试云端部署的错误？

方法 1：查看 LangSmith 日志

访问 LangSmith → Deployments
选择你的部署
点击 "Logs" 标签
查看错误堆栈和调用详情

方法 2：启用详细追踪

python

# 在 langgraph.json 中添加
{
  "env": ".env",
  "tracing": {
    "enabled": true,
    "project": "my-project"
  }
}

方法 3：本地复现

python

# 使用相同的输入在本地测试
async for chunk in client.runs.stream(
    thread_id,
    "agent",
    input=problematic_input,  # 导致错误的输入
    stream_mode="values"
):
    print(chunk)

Q5: 部署后如何更新代码？

方法 1：推送到 GitHub（推荐）

bash

# 修改代码
vim agent.py

# 提交并推送
git add .
git commit -m "Update agent logic"
git push origin main

# LangGraph Cloud 会自动检测并重新部署

方法 2：手动触发部署

访问 LangSmith → Deployments
选择你的部署
点击 "Redeploy" 按钮

注意： 重新部署会有几分钟的停机时间。

Q6: 如何限制 API 调用速率？

场景： 防止恶意用户滥用或超出 OpenAI 配额

方法 1：在图中添加速率限制

python

from ratelimit import limits, sleep_and_retry

# 每分钟最多 10 次调用
@sleep_and_retry
@limits(calls=10, period=60)
def call_model(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

方法 2：使用 LangGraph Cloud 配置

json

{
  "rate_limits": {
    "per_user": {
      "requests": 100,
      "window": "1h"
    }
  }
}

Q7: stream_mode="values" 和 "updates" 有什么实际区别？

示例场景： 聊天应用显示 Agent 思考过程

使用 "values"：

python

async for chunk in client.runs.stream(thread_id, "agent", input=..., stream_mode="values"):
    # 每次获得完整状态
    all_messages = chunk.data['messages']

    # UI 需要每次重新渲染整个对话
    display_all_messages(all_messages)

使用 "updates"：

python

async for chunk in client.runs.stream(thread_id, "agent", input=..., stream_mode="updates"):
    # 只获得新增消息
    new_messages = chunk.data['messages']

    # UI 只需要追加新消息
    append_messages(new_messages)

选择建议：

"values"：适合简单 UI，每次全量更新
"updates"：适合复杂 UI，需要增量更新（更高效）

📊 本地 vs 云端对比总结

维度	本地开发	云端部署
启动命令	`langgraph dev`	通过 LangSmith UI
URL	`http://127.0.0.1:2024`	`https://xxx.langgraph.app`
环境变量	`.env` 文件	LangSmith UI 配置
代码更新	保存文件即可（热重载）	推送 Git → 自动部署
状态持久化	内存（重启丢失）	数据库（永久）
适用场景	开发、测试、调试	生产、多用户、长期运行
成本	免费（仅本地资源）	按使用量计费
可观测性	基础日志	完整追踪、监控、告警

🎯 实战练习

练习 1：本地部署一个简单 Agent

任务： 创建一个计算器 Agent，支持加减乘除

创建项目结构
定义工具和图
配置 langgraph.json
启动 langgraph dev
使用 SDK 测试

提示：

python

@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@tool
def subtract(a: int, b: int) -> int:
    """Subtract b from a."""
    return a - b

# 绑定工具到 LLM
tools = [add, subtract]
llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools(tools)

练习 2：部署到 LangGraph Cloud

任务： 将练习 1 的 Agent 部署到云端

创建 GitHub 仓库
推送代码
在 LangSmith 配置部署
获取生产 URL
使用 SDK 调用云端 API

练习 3：多线程对话

任务： 创建 3 个独立的对话线程，测试状态隔离

python

# 创建 3 个线程
threads = [await client.threads.create() for _ in range(3)]

# 每个线程发送不同消息
messages = [
    "What is 3 + 5?",
    "What is 10 - 2?",
    "What is 4 * 6?"
]

# 并发运行
for thread, message in zip(threads, messages):
    asyncio.create_task(
        client.runs.stream(thread['thread_id'], "agent", input={"messages": [HumanMessage(message)]})
    )

📖 扩展阅读

总结： 通过本教程，你已经掌握了 LangGraph 的完整部署流程！从本地开发测试到云端生产部署，再到使用 SDK 进行程序化调用。这些技能是构建可扩展、可靠的 AI 应用的基石。继续实践，探索更多高级功能！🚀

LangGraph Deployment 详细解读 🚀 ​

📚 概述 ​

🎯 核心概念 ​

为什么需要部署？ ​

🏗️ LangGraph 生态架构 ​

核心组件关系图 ​

1. LangGraph Library ​

2. LangGraph API ​

3. LangGraph Cloud ​

4. LangGraph Studio ​

本地模式（推荐用于开发） ​

云模式（用于测试生产部署） ​

5. LangGraph SDK ​

🛠️ 本地部署实战 ​

前置准备 ​

1. 项目结构 ​

2. langgraph.json 配置文件 ​

3. agent.py 示例 ​

启动本地开发服务器 ​

步骤 1：进入项目目录 ​

步骤 2：启动服务 ​

步骤 3：在浏览器打开 Studio ​

💻 使用 SDK 调用本地 API ​

安装依赖 ​

代码示例 ​

1. 连接到本地服务器 ​

2. 搜索可用的助手（图） ​

3. 创建对话线程 ​

4. 运行图并流式输出 ​

☁️ 云端部署实战 ​

部署流程概览 ​

步骤 1：创建 GitHub 仓库 ​

1.1 在 GitHub 创建新仓库 ​

1.2 添加远程仓库 ​

1.3 推送代码 ​

步骤 2：在 LangSmith 配置部署 ​

2.1 访问 LangSmith ​

2.2 创建新部署 ​

2.3 连接 GitHub 仓库 ​

2.4 设置环境变量 ​

2.5 部署 ​

2.6 获取部署 URL ​

步骤 3：使用 SDK 调用云端部署 ​

3.1 设置 LangSmith API 密钥 ​

3.2 连接到云端服务 ​

3.3 运行云端 Agent ​

步骤 4：在 Studio 中访问云端部署 ​

4.1 打开 LangGraph Studio ​

4.2 选择云端部署 ​

4.3 交互式测试 ​

🎓 核心知识点总结 ​

LangGraph 特有概念 ​

1. Deployment（部署） ​

2. Thread vs Assistant ​

3. Stream Modes 对比 ​

4. 本地 vs 云端部署 ​

Python 特有知识点 ​

1. 异步编程（async/await） ​

2. 异步迭代器（async for） ​

3. getpass 安全输入 ​

4. 环境变量管理 ​

💡 最佳实践 ​

1. 项目结构规范 ​

2. 环境变量管理 ​

.env.example 模板 ​

多环境配置 ​

3. 版本管理 ​

langgraph.json 版本控制 ​

Git 标签（Tag） ​

4. 监控和调试 ​

启用 LangSmith 追踪 ​

本地调试技巧 ​

5. 错误处理 ​

优雅的错误恢复 ​

重试机制 ​

6. 性能优化 ​

使用流式响应 ​

缓存常用结果 ​

🚀 进阶技巧 ​

1. 多版本部署 ​

LangGraph Deployment 详细解读 🚀

📚 概述

🎯 核心概念

为什么需要部署？

🏗️ LangGraph 生态架构

核心组件关系图

1. LangGraph Library

2. LangGraph API

3. LangGraph Cloud

4. LangGraph Studio

本地模式（推荐用于开发）

云模式（用于测试生产部署）

5. LangGraph SDK

🛠️ 本地部署实战

前置准备

1. 项目结构

2. langgraph.json 配置文件

3. agent.py 示例

启动本地开发服务器

步骤 1：进入项目目录

步骤 2：启动服务

步骤 3：在浏览器打开 Studio

💻 使用 SDK 调用本地 API

安装依赖

代码示例

1. 连接到本地服务器

2. 搜索可用的助手（图）

3. 创建对话线程

4. 运行图并流式输出

☁️ 云端部署实战

部署流程概览

步骤 1：创建 GitHub 仓库

1.1 在 GitHub 创建新仓库

1.2 添加远程仓库

1.3 推送代码

步骤 2：在 LangSmith 配置部署

2.1 访问 LangSmith

2.2 创建新部署

2.3 连接 GitHub 仓库

2.4 设置环境变量

2.5 部署

2.6 获取部署 URL

步骤 3：使用 SDK 调用云端部署

3.1 设置 LangSmith API 密钥

3.2 连接到云端服务

3.3 运行云端 Agent

步骤 4：在 Studio 中访问云端部署

4.1 打开 LangGraph Studio

4.2 选择云端部署

4.3 交互式测试

🎓 核心知识点总结

LangGraph 特有概念

1. Deployment（部署）

2. Thread vs Assistant

3. Stream Modes 对比

4. 本地 vs 云端部署

Python 特有知识点

1. 异步编程（async/await）

2. 异步迭代器（async for）

3. getpass 安全输入

4. 环境变量管理

💡 最佳实践

1. 项目结构规范

2. 环境变量管理

.env.example 模板

多环境配置

3. 版本管理

langgraph.json 版本控制

Git 标签（Tag）

4. 监控和调试

启用 LangSmith 追踪

本地调试技巧

5. 错误处理

优雅的错误恢复

重试机制

6. 性能优化

使用流式响应

缓存常用结果

🚀 进阶技巧

1. 多版本部署