在前两篇文章中，我们介绍了 LangGraph 的基础概念和核心特性。本文将聚焦于实际开发流程，包括项目结构、测试策略和部署方案。
目录
应用结构
一个 LangGraph 应用由一个或多个图、配置文件（langgraph.json）、依赖文件和可选的环境变量文件组成。
典型的项目结构
my-app/
├── my_agent/ # 所有项目代码
│ ├── utils/ # 图的工具函数
│ │ ├── __init__.py
│ │ ├── tools.py # 图的工具
│ │ ├── nodes.py # 图的节点函数
│ │ └── state.py # 图的状态定义
│ ├── __init__.py
│ └── agent.py # 构建图的代码
├── .env # 环境变量
├── requirements.txt # 包依赖
└── langgraph.json # LangGraph 配置文件配置文件（langgraph.json）
langgraph.json 是一个 JSON 文件，指定部署 LangGraph 应用所需的依赖、图、环境变量和其他设置。
基本示例：
{
"dependencies": ["langchain_openai", "./your_package"],
"graphs": {
"my_agent": "./your_package/your_file.py:agent"
},
"env": "./.env"
}这个配置表示：
- 依赖包括自定义本地包和
langchain_openai包 - 从
./your_package/your_file.py文件的agent变量加载单个图 - 从
.env文件加载环境变量
依赖管理
LangGraph 应用可能依赖其他 Python 包。需要指定以下信息：
- 依赖文件：
requirements.txt或pyproject.toml - 配置文件中的依赖键：指定运行应用所需的依赖
- 额外的二进制文件或系统库：使用
dockerfile_lines键指定
requirements.txt 示例：
langgraph
langchain-openai
langchain-anthropicpyproject.toml 示例：
[project]
name = "my-agent"
version = "0.1.0"
dependencies = [
"langgraph>=0.2.0",
"langchain-openai>=0.1.0",
]图配置
使用配置文件中的 graphs 键指定部署的 LangGraph 应用中可用的图。可以指定一个或多个图，每个图由名称（应该是唯一的）和路径标识。
{
"graphs": {
"customer_support": "./agents/customer_support.py:graph",
"data_analyst": "./agents/data_analyst.py:graph"
}
}环境变量
对于本地开发，可以在配置文件的 env 键中配置环境变量：
{
"env": "./.env"
}.env 文件示例：
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
DATABASE_URL=postgresql://localhost:5432/mydb对于生产部署，通常在部署环境中配置环境变量。
测试策略
在原型化 LangGraph Agent 后，添加测试是自然的下一步。本节介绍编写单元测试时的一些有用模式。
安装 pytest
pip install pytest基本测试模式
由于许多 LangGraph Agent 依赖于状态，一个有用的模式是在每个测试之前创建图，然后在测试中使用新的检查点实例编译它。
import pytest
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
def create_graph() -> StateGraph:
class MyState(TypedDict):
my_key: str
graph = StateGraph(MyState)
graph.add_node("node1", lambda state: {"my_key": "hello from node1"})
graph.add_node("node2", lambda state: {"my_key": "hello from node2"})
graph.add_edge(START, "node1")
graph.add_edge("node1", "node2")
graph.add_edge("node2", END)
return graph
def test_basic_agent_execution() -> None:
checkpointer = MemorySaver()
graph = create_graph()
compiled_graph = graph.compile(checkpointer=checkpointer)
result = compiled_graph.invoke(
{"my_key": "initial_value"},
config={"configurable": {"thread_id": "1"}}
)
assert result["my_key"] == "hello from node2"测试单个节点
编译的 LangGraph Agent 通过 graph.nodes 暴露对每个单独节点的引用。可以利用这一点来测试 Agent 中的单个节点。
def test_individual_node_execution() -> None:
checkpointer = MemorySaver()
graph = create_graph()
compiled_graph = graph.compile(checkpointer=checkpointer)
# 只调用 node1
result = compiled_graph.nodes["node1"].invoke(
{"my_key": "initial_value"},
)
assert result["my_key"] == "hello from node1"注意：这将绕过编译图时传递的任何检查点。
部分执行测试
对于由较大图组成的 Agent，可能希望测试 Agent 中的部分执行路径，而不是整个端到端流程。可以使用 LangGraph 的持久化机制来模拟 Agent 在所需部分开始之前暂停的状态。
步骤如下：
- 使用检查点编译 Agent（内存检查点
MemorySaver足以用于测试） - 调用 Agent 的
update_state方法，将as_node参数设置为要开始测试的节点之前的节点名称 - 使用相同的
thread_id调用 Agent，并将interrupt_after参数设置为要停止的节点名称
示例：仅执行线性图中的第二和第三个节点：
def create_graph() -> StateGraph:
class MyState(TypedDict):
my_key: str
graph = StateGraph(MyState)
graph.add_node("node1", lambda state: {"my_key": "hello from node1"})
graph.add_node("node2", lambda state: {"my_key": "hello from node2"})
graph.add_node("node3", lambda state: {"my_key": "hello from node3"})
graph.add_node("node4", lambda state: {"my_key": "hello from node4"})
graph.add_edge(START, "node1")
graph.add_edge("node1", "node2")
graph.add_edge("node2", "node3")
graph.add_edge("node3", "node4")
graph.add_edge("node4", END)
return graph
def test_partial_execution_from_node2_to_node3() -> None:
checkpointer = MemorySaver()
graph = create_graph()
compiled_graph = graph.compile(checkpointer=checkpointer)
# 更新状态，模拟在 node1 结束时的状态
compiled_graph.update_state(
config={"configurable": {"thread_id": "1"}},
values={"my_key": "initial_value"},
as_node="node1", # 执行将从 node2 恢复
)
# 恢复执行
result = compiled_graph.invoke(
None, # 传递 None 以恢复执行
config={"configurable": {"thread_id": "1"}},
interrupt_after="node3", # 在 node3 后停止
)
assert result["my_key"] == "hello from node3"测试工具调用
对于使用工具的 Agent，可以模拟工具响应来测试 Agent 的行为：
from unittest.mock import patch
def test_agent_with_mocked_tool():
graph = create_agent_graph()
compiled_graph = graph.compile()
with patch('my_agent.tools.search_tool') as mock_search:
mock_search.return_value = "模拟的搜索结果"
result = compiled_graph.invoke({
"messages": [{"role": "user", "content": "搜索 LangGraph"}]
})
assert "模拟的搜索结果" in str(result)
mock_search.assert_called_once()测试错误处理
测试 Agent 如何处理错误情况：
def test_error_handling():
graph = create_graph_with_error_handling()
compiled_graph = graph.compile()
# 测试无效输入
result = compiled_graph.invoke({"invalid_key": "value"})
assert "error" in result
# 测试工具失败
with patch('my_agent.tools.api_call') as mock_api:
mock_api.side_effect = Exception("API 错误")
result = compiled_graph.invoke({"query": "test"})
assert result["status"] == "failed"本地开发与调试
本地服务器
LangGraph 提供了本地开发服务器，可以在部署前测试应用。
启动本地服务器：
langgraph dev这将启动一个本地服务器，通常在 http://localhost:8123。
使用 LangGraph Studio
LangGraph Studio 是一个可视化调试工具，可以：
- 可视化图结构
- 逐步执行图
- 检查每个步骤的状态
- 测试不同的输入
在 Studio 中打开应用：
langgraph dev --studio调试技巧
1. 使用 debug 流式模式：
for chunk in graph.stream(
inputs,
stream_mode="debug",
):
print(chunk)2. 检查状态历史：
config = {"configurable": {"thread_id": "1"}}
history = list(graph.get_state_history(config))
for state in history:
print(f"步骤 {state.metadata['step']}: {state.values}")3. 添加日志节点：
def log_state(state: State):
print(f"当前状态: {state}")
return state
graph.add_node("log", log_state)部署到生产环境
部署到 LangSmith Cloud
LangSmith Cloud 是专为 Agent 工作负载设计的完全托管托管平台。
前提条件：
- GitHub 账户
- LangSmith 账户（免费注册）
部署步骤：
- 创建 GitHub 仓库
将应用代码推送到 GitHub 仓库（支持公共和私有仓库）。
- 连接到 LangSmith
在 LangSmith 控制台中：
- 导航到 Deployments
- 点击 “New Deployment”
- 选择 GitHub 仓库
- 配置部署设置
- 在 Studio 中测试
部署完成后：
- 选择刚创建的部署
- 点击右上角的 “Studio” 按钮
- Studio 将打开并显示图
- 获取 API URL
在部署详情视图中，点击 API URL 将其复制到剪贴板。
- 测试 API
使用 Python SDK：
pip install langgraph-sdkfrom langgraph_sdk import get_sync_client
client = get_sync_client(
url="your-deployment-url",
api_key="your-langsmith-api-key"
)
for chunk in client.runs.stream(
None, # 无线程运行
"agent", # Agent 名称，在 langgraph.json 中定义
input={
"messages": [{
"role": "human",
"content": "什么是 LangGraph？",
}],
},
stream_mode="updates",
):
print(f"接收到新事件，类型: {chunk.event}")
print(chunk.data)使用 REST API：
curl -s --request POST \
--url <DEPLOYMENT_URL>/runs/stream \
--header 'Content-Type: application/json' \
--header "X-Api-Key: <LANGSMITH_API_KEY>" \
--data '{
"assistant_id": "agent",
"input": {
"messages": [{
"role": "human",
"content": "什么是 LangGraph？"
}]
},
"stream_mode": "updates"
}'自托管部署
如果需要自托管，可以使用 Docker 容器化应用。
Dockerfile 示例：
FROM python:3.11-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 暴露端口
EXPOSE 8000
# 启动应用
CMD ["uvicorn", "my_agent.server:app", "--host", "0.0.0.0", "--port", "8000"]构建和运行：
docker build -t my-langgraph-agent .
docker run -p 8000:8000 --env-file .env my-langgraph-agent使用 Kubernetes
对于大规模部署，可以使用 Kubernetes：
deployment.yaml 示例：
apiVersion: apps/v1
kind: Deployment
metadata:
name: langgraph-agent
spec:
replicas: 3
selector:
matchLabels:
app: langgraph-agent
template:
metadata:
labels:
app: langgraph-agent
spec:
containers:
- name: agent
image: my-langgraph-agent:latest
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: api-keys
key: openai
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "1Gi"
cpu: "1000m"可观测性与监控
使用 LangSmith 进行追踪
LangSmith 提供了强大的追踪和调试功能：
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"
# 现在所有的运行都会被追踪
result = graph.invoke(inputs)自定义指标
添加自定义指标来监控 Agent 性能：
import time
from typing import TypedDict
class State(TypedDict):
query: str
result: str
execution_time: float
def timed_node(state: State):
start_time = time.time()
# 执行节点逻辑
result = process_query(state["query"])
execution_time = time.time() - start_time
return {
"result": result,
"execution_time": execution_time
}错误监控
集成错误监控服务（如 Sentry）：
import sentry_sdk
sentry_sdk.init(
dsn="your-sentry-dsn",
traces_sample_rate=1.0,
)
def monitored_node(state: State):
try:
return process_state(state)
except Exception as e:
sentry_sdk.capture_exception(e)
raise性能优化
并行执行
利用 LangGraph 的并行执行能力：
# 并行执行多个独立节点
graph.add_edge(START, "node1")
graph.add_edge(START, "node2")
graph.add_edge(START, "node3")缓存策略
实现缓存以减少重复计算：
from functools import lru_cache
@lru_cache(maxsize=100)
def expensive_operation(query: str):
# 昂贵的操作
return result
def cached_node(state: State):
result = expensive_operation(state["query"])
return {"result": result}异步执行
使用异步操作提高性能：
async def async_node(state: State):
result = await async_api_call(state["query"])
return {"result": result}
# 使用异步调用
result = await graph.ainvoke(inputs)最佳实践总结
- 结构化项目：使用清晰的目录结构，分离关注点
- 全面测试：编写单元测试、集成测试和端到端测试
- 环境管理：使用环境变量管理配置，不要硬编码敏感信息
- 错误处理：实现适当的错误处理和重试策略
- 监控追踪：使用 LangSmith 或其他工具进行追踪和监控
- 性能优化：利用并行执行、缓存和异步操作
- 文档化：为图、节点和工具编写清晰的文档
- 版本控制：使用 Git 管理代码，使用语义化版本控制
系列总结
通过这三篇文章，我们完整地介绍了 LangGraph 的核心概念和实践：
- 入门指南：介绍了 LangGraph 的基础概念、快速开始和思维方式
- 核心特性：深入探讨了持久化、流式输出、人机协作和记忆管理
- 开发部署：涵盖了项目结构、测试策略、调试方法和部署方案
LangGraph 为构建生产级 Agent 系统提供了强大的基础设施。通过掌握这些概念和实践，你可以构建可靠、可扩展的 AI Agent 应用。
参考资源
内容基于 LangGraph 官方文档改编，遵循内容许可限制