Module-7: LangGraph 生产部署（Production Deployment）- 本章介绍

致学习者的一封信
恭喜你走到了这一步!经过前面6个模块的学习,你已经掌握了 LangGraph 的核心技术——从基础的 Graph 构建,到复杂的 Human-in-the-Loop 交互,再到强大的记忆系统。但这些知识的真正价值,只有在部署到生产环境,服务真实用户时才能完全体现。
Module-7 将带你完成从开发到生产的关键跨越。我们不仅要学习如何部署 LangGraph 应用,更要掌握生产环境中的关键挑战——并发控制、多租户管理、配置化设计。这些都是区分玩具项目和企业级应用的关键能力。
在我数十年的工程经验中,我见过太多优秀的原型因为部署问题而胎死腹中。让我们一起确保你的 LangGraph 应用不会是其中之一。准备好了吗?让我们开始这段从开发到生产的旅程!
—— 你的 LangGraph 导师

一、本章学习目标

完成本章学习后,你将能够:

核心能力

部署 LangGraph 应用
- 理解 LangGraph Platform 架构
- 使用 Docker 容器化应用
- 配置 Redis 和 PostgreSQL 存储
- 使用 LangGraph CLI 构建和部署
连接和管理部署
- 使用 LangGraph SDK 连接远程部署
- 使用 Remote Graph 简化迁移
- 管理 Runs、Threads、Store 三大 API
处理并发请求
- 理解 Double Texting 问题
- 掌握 4 种并发策略(Reject/Enqueue/Interrupt/Rollback)
- 根据场景选择合适策略
配置化设计
- 创建和管理 Assistants
- 为同一图配置多个变体
- 实现多租户隔离

工程实践

DevOps 技能:Docker、docker-compose、环境变量管理
API 设计:RESTful API、流式传输、错误处理
并发控制:竞态条件、队列管理、中断恢复
生产思维:监控、日志、扩展性、可靠性

二、核心概念预览

2.1 LangGraph Platform 架构全景

┌──────────────────────────────────────────────────────────────┐
│                  LangGraph Platform 架构                      │
├──────────────────────────────────────────────────────────────┤
│                                                               │
│  ┌──────────────────────────────────────────────────────┐   │
│  │              客户端层(Client Layer)                   │   │
│  │  ┌─────────────┐  ┌──────────────┐  ┌────────────┐  │   │
│  │  │ LangGraph   │  │ Remote       │  │ HTTP       │  │   │
│  │  │ SDK         │  │ Graph        │  │ Client     │  │   │
│  │  │ (Python/JS) │  │ (简化接口)    │  │ (原始API)  │  │   │
│  │  └─────────────┘  └──────────────┘  └────────────┘  │   │
│  └────────────────┬─────────────────────────────────────┘   │
│                   │                                          │
│                   │ HTTP/WebSocket                           │
│                   ↓                                          │
│  ┌──────────────────────────────────────────────────────┐   │
│  │          服务层(Service Layer)                        │   │
│  │  ┌────────────────────────────────────────────────┐  │   │
│  │  │  LangGraph Server (HTTP Workers)              │  │   │
│  │  │  • 接收API请求                                 │  │   │
│  │  │  • 管理流式传输                               │  │   │
│  │  │  • 处理认证授权                               │  │   │
│  │  │  • 提供Swagger文档                            │  │   │
│  │  └────────────────────────────────────────────────┘  │   │
│  │  ┌────────────────────────────────────────────────┐  │   │
│  │  │  Queue Workers                                 │  │   │
│  │  │  • 后台执行图                                  │  │   │
│  │  │  • 并发控制                                    │  │   │
│  │  │  • 任务队列管理                                │  │   │
│  │  └────────────────────────────────────────────────┘  │   │
│  └────────────────┬─────────────────────────────────────┘   │
│                   │                                          │
│                   ↓                                          │
│  ┌──────────────────────────────────────────────────────┐   │
│  │          存储层(Storage Layer)                        │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌───────────┐  │   │
│  │  │  PostgreSQL  │  │    Redis     │  │   LLM     │  │   │
│  │  │              │  │              │  │   APIs    │  │   │
│  │  │ • Checkpoints│  │ • 消息队列   │  │           │  │   │
│  │  │ • Store      │  │ • 缓存       │  │ • OpenAI  │  │   │
│  │  │ • Threads    │  │ • 流式数据   │  │ • Anthropic│ │   │
│  │  │ • Assistants │  │              │  │ • ...     │  │   │
│  │  └──────────────┘  └──────────────┘  └───────────┘  │   │
│  └──────────────────────────────────────────────────────┘   │
│                                                               │
└──────────────────────────────────────────────────────────────┘

2.2 从开发到生产的转变

维度	开发环境	生产环境
运行方式	本地 Python 脚本	Docker 容器 + HTTP API
调用方式	`graph.invoke()`	`client.runs.create()`
存储	InMemoryStore + MemorySaver	PostgreSQL + Redis
并发	单用户,顺序执行	多用户,并发处理
配置	硬编码在代码中	Assistants + 环境变量
访问	localhost:8000	your-domain.com (HTTPS)
监控	print() 调试	LangSmith + 日志系统
扩展	单机运行	水平扩展(多容器)
可靠性	可以重启	高可用(HA)部署

2.3 三大 API 类别

LangGraph Server 提供的 API 分为三类:

API 端点
├── Runs (运行)
│   ├── 创建运行: POST /threads/{thread_id}/runs
│   ├── 获取状态: GET /threads/{thread_id}/runs/{run_id}
│   ├── 流式执行: GET /threads/{thread_id}/runs/stream
│   └── 类型:
│       • Background (后台)
│       • Blocking (阻塞)
│       • Streaming (流式)
│
├── Threads (线程/会话)
│   ├── 创建线程: POST /threads
│   ├── 获取状态: GET /threads/{thread_id}/state
│   ├── 更新状态: POST /threads/{thread_id}/state
│   ├── 列表运行: GET /threads/{thread_id}/runs
│   └── 功能:
│       • 对话历史管理
│       • Human-in-the-Loop
│       • 状态编辑(时间旅行)
│       • 线程复制
│
└── Store (存储)
    ├── 搜索: GET /store/{namespace}
    ├── 添加: POST /store/{namespace}
    ├── 删除: DELETE /store/{namespace}/{key}
    └── 功能:
        • 跨线程数据
        • 用户记忆
        • 持久化存储

三、Docker 容器化基础

3.1 为什么需要 Docker?

问题:「在我机器上能跑」综合症

开发者机器             →  测试服务器           →  生产服务器
Python 3.11              Python 3.9              Python 3.10
macOS                    Ubuntu 20.04            Ubuntu 22.04
依赖版本 A               依赖版本 B              依赖版本 C
        ↓                        ↓                        ↓
   能运行                   部分错误                 完全崩溃

Docker 解决方案:

Docker 镜像(Image)
├── 基础系统: Debian Linux
├── Python 3.11
├── 所有依赖包(固定版本)
├── 应用代码
└── 启动脚本

这个镜像在任何支持 Docker 的机器上都能以完全相同的方式运行

3.2 核心概念

Docker 镜像 vs 容器

镜像(Image)                    容器(Container)
    ↓                              ↓
  模板                            实例
  类(Class)                       对象(Object)
  ISO文件                         运行的系统
  食谱                            做好的菜

一次构建                        多次运行
只读                            可读写
共享                            隔离

实际例子:

python

# 镜像: langgraph-app:v1.0
docker build -t langgraph-app:v1.0 .

# 从同一镜像启动多个容器
docker run langgraph-app:v1.0  # 容器1
docker run langgraph-app:v1.0  # 容器2
docker run langgraph-app:v1.0  # 容器3

docker-compose

作用:一次性管理多个容器

yaml

# docker-compose.yml
services:
  redis:        # 容器1: Redis
  postgres:     # 容器2: PostgreSQL  
  langgraph:    # 容器3: LangGraph Server

# 一行命令启动所有容器
docker compose up

3.3 LangGraph 部署所需文件

deployment/
├── langgraph.json          # LangGraph配置
│   └── 定义图的位置和名称
│
├── task_maistro.py         # 图的实现
│   └── 导出编译后的graph
│
├── requirements.txt        # Python依赖
│   └── 声明所有需要的包
│
├── .env                    # 环境变量
│   └── API密钥等敏感信息
│
└── docker-compose.yml      # 容器编排
    └── 定义3个服务
        • Redis
        • PostgreSQL
        • LangGraph Server

四、并发控制:Double Texting 问题

4.1 什么是 Double Texting?

用户快速连续发送消息,不等待前一个请求完成:

时间轴:
00:00 用户发送:"添加待办:买牛奶"
      ↓ Run 1 开始执行
00:01 用户又发送:"改成买全脂牛奶"
      ↓ Run 2 想要开始
      ↓ 问题:Run 1 还在执行!

4.2 为什么这是个问题?

数据竞争(Race Condition):

Thread State
    ↓
messages: [msg1, msg2]
    ↓
Run 1 读取: [msg1, msg2]
    ↓ (执行中...)
Run 2 也读取: [msg1, msg2]
    ↓
Run 1 完成,写入: [msg1, msg2, ai_reply1]
    ↓
Run 2 完成,写入: [msg1, msg2, ai_reply2]
    ↓
结果:ai_reply1 丢失! ❌

4.3 四种策略对比

策略1: Reject (拒绝)
┌────────────────────────────────┐
│ Run 1 执行中                    │
│    ↓                           │
│ Run 2 请求 → ❌ 409 Conflict   │
│    ↓                           │
│ Run 1 完成                     │
│    ↓                           │
│ 现在可以接受新请求              │
└────────────────────────────────┘
适用:严格顺序要求,资源受限

策略2: Enqueue (排队)
┌────────────────────────────────┐
│ Run 1 执行中                    │
│    ↓                           │
│ Run 2 请求 → ✅ 加入队列        │
│    ↓                           │
│ Run 1 完成                     │
│    ↓                           │
│ Run 2 自动开始                 │
│    ↓                           │
│ Run 2 完成                     │
└────────────────────────────────┘
适用:所有请求都重要,批量操作

策略3: Interrupt (中断)
┌────────────────────────────────┐
│ Run 1 执行中                    │
│    ↓                           │
│ Run 2 请求 → ⚠️ 中断Run 1      │
│    ↓                           │
│ 保存Run 1进度到checkpoint       │
│    ↓                           │
│ Run 2 开始执行                 │
│    ↓                           │
│ Run 2 完成                     │
│    ↓ (可选)                    │
│ Run 1 可以从断点继续            │
└────────────────────────────────┘
适用:用户纠正/补充,新请求优先级高

策略4: Rollback (回滚)
┌────────────────────────────────┐
│ Run 1 执行中                    │
│    ↓                           │
│ Run 2 请求 → 🔄 删除Run 1      │
│    ↓                           │
│ Run 2 重新开始                 │
│    ↓                           │
│ Run 2 完成                     │
└────────────────────────────────┘
适用:用户完全改变主意,最新请求覆盖旧请求

4.4 策略选择决策树

mermaid

graph TD
    A[收到并发请求] --> B{第一个请求还重要吗?}
    B -->|不重要,用户改变主意| C[Rollback]
    B -->|仍然重要| D{两个请求都必须处理?}
    
    D -->|是,都重要| E{能等待吗?}
    E -->|可以按顺序处理| F[Enqueue]
    E -->|新请求更紧急| G[Interrupt]
    
    D -->|否,只处理一个| H{哪个优先?}
    H -->|第一个优先| I[Reject]
    H -->|第二个优先| C
    
    style C fill:#ff6b6b
    style F fill:#51cf66
    style G fill:#ffd43b
    style I fill:#339af0

五、Assistants:配置化设计

5.1 什么是 Assistant?

Assistant = Graph + Config

Base Graph: task_maistro
    ↓
┌──────────────────────────────────────┐
│ Personal Assistant                   │
│ config:                              │
│   • user_id: "alice"                 │
│   • todo_category: "personal"        │
│   • role: "友好鼓励的个人助手..."     │
└──────────────────────────────────────┘
┌──────────────────────────────────────┐
│ Work Assistant                       │
│ config:                              │
│   • user_id: "alice"                 │
│   • todo_category: "work"            │
│   • role: "专业高效的工作助手..."     │
└──────────────────────────────────────┘
┌──────────────────────────────────────┐
│ Family Assistant                     │
│ config:                              │
│   • user_id: "alice"                 │
│   • todo_category: "family"          │
│   • role: "温馨协调的家庭助手..."     │
└──────────────────────────────────────┘

5.2 为什么需要 Assistants?

没有 Assistants:

python

# 每次都要传递完整配置
config = {
    "configurable": {
        "user_id": "alice",
        "todo_category": "work",
        "role": "你是一个专业高效的工作助手,帮助用户管理工作任务..."
    }
}

run = await client.runs.create(
    thread_id,
    "task_maistro",
    input=input,
    config=config  # 每次都要传
)

有了 Assistants:

python

# 创建一次
work_assistant = await client.assistants.create(
    "task_maistro",
    config={"configurable": {...}}
)

# 以后直接使用assistant_id
run = await client.runs.create(
    thread_id,
    work_assistant["assistant_id"],  # 简洁!
    input=input
)

5.3 Assistant 的版本控制

Assistant: work_assistant_id
├── Version 1 (创建时)
│   └── config: {user_id, todo_category}
│
├── Version 2 (添加角色)
│   └── config: {user_id, todo_category, role}
│
└── Version 3 (更新角色描述)
    └── config: {user_id, todo_category, role(updated)}

可以:
• 查看历史版本
• 回滚到旧版本
• A/B测试不同配置

5.4 实际应用场景

场景1:多语言支持

python

# 英文助手
english_assistant = await client.assistants.create(
    "customer_service",
    config={"configurable": {"language": "en"}}
)

# 中文助手
chinese_assistant = await client.assistants.create(
    "customer_service",
    config={"configurable": {"language": "zh"}}
)

场景2:不同行业定制

python

# 医疗助手
medical_assistant = await client.assistants.create(
    "consultation_bot",
    config={"configurable": {
        "domain": "medical",
        "compliance": "HIPAA",
        "terminology": "medical_terms.json"
    }}
)

# 法律助手
legal_assistant = await client.assistants.create(
    "consultation_bot",
    config={"configurable": {
        "domain": "legal",
        "compliance": "attorney-client",
        "terminology": "legal_terms.json"
    }}
)

场景3:多租户SaaS

python

# 为每个公司创建独立助手
for company in companies:
    assistant = await client.assistants.create(
        "crm_agent",
        config={"configurable": {
            "tenant_id": company.id,
            "branding": company.branding,
            "features": company.subscription_features
        }}
    )

六、学习路径建议

路径 1:快速部署 (初学者) ⏱️ 2-3天

目标:能够部署和使用 LangGraph 应用

Day 1: 部署基础
├── 7.1 Creating Deployment
│   ├── 理解部署文件结构
│   ├── 使用 LangGraph CLI
│   ├── Docker 基础概念
│   └── 本地启动部署
└── 实践:部署 task_maistro

Day 2: API 使用
├── 7.2 Connecting
│   ├── SDK vs Remote Graph
│   ├── Runs API (Background/Blocking/Streaming)
│   ├── Threads API
│   └── Store API
└── 实践:通过 API 调用部署

Day 3: 并发和配置
├── 7.3 Double Texting
│   └── 理解4种策略
├── 7.4 Assistants
│   └── 创建多个助手
└── 实践:配置自己的助手

学习重点:

✅ 能够成功部署应用
✅ 理解 Docker 基本概念
✅ 会使用 SDK 连接部署
✅ 了解并发策略

跳过内容:

Docker 内部机制
高级并发场景
生产环境优化

路径 2:生产就绪 (中级开发者) ⏱️ 5-7天

目标:能够部署生产级应用并处理常见问题

Day 1-2: 深入部署
├── Docker 深入
│   ├── Dockerfile 自定义
│   ├── 镜像优化(分层、缓存)
│   ├── docker-compose 高级配置
│   └── 环境变量管理
├── 存储配置
│   ├── PostgreSQL 配置和优化
│   ├── Redis 配置
│   └── 数据持久化
└── 实践:自定义部署配置

Day 3-4: API 精通
├── Runs 深入
│   ├── 流式传输机制
│   ├── 错误处理和重试
│   ├── 超时控制
│   └── 取消运行
├── Threads 高级
│   ├── 状态编辑(时间旅行)
│   ├── 线程复制
│   ├── Human-in-the-Loop 集成
│   └── 状态检查点管理
└── 实践:实现复杂交互流程

Day 5-6: 并发控制
├── Double Texting 深入
│   ├── 4种策略实现原理
│   ├── 自定义并发逻辑
│   ├── 性能影响分析
│   └── 最佳实践
├── 队列管理
│   ├── Redis 队列机制
│   ├── 队列监控
│   └── 队列优化
└── 实践:处理高并发场景

Day 7: 配置和扩展
├── Assistants 高级
│   ├── 动态配置
│   ├── 配置模板
│   ├── 版本管理策略
│   └── 多租户隔离
├── 扩展性设计
│   ├── 水平扩展(多容器)
│   ├── 负载均衡
│   └── 缓存策略
└── 实践:设计可扩展架构

学习重点:

✅ 理解部署的内部机制
✅ 掌握并发控制策略
✅ 能够优化性能
✅ 处理生产环境问题

路径 3:DevOps 专家 (高级工程师) ⏱️ 7-10天

目标:成为 LangGraph 部署和运维专家

Week 1: 架构与实现
├── Day 1-2: 平台架构深入
│   ├── HTTP Workers vs Queue Workers
│   ├── 请求处理流程
│   ├── 流式传输架构
│   ├── 消息队列机制
│   └── 实践:架构分析和优化
│
├── Day 3-4: 存储层优化
│   ├── PostgreSQL 性能调优
│   │   ├── 索引优化
│   │   ├── 连接池配置
│   │   ├── 查询优化
│   │   └── 分区策略
│   ├── Redis 优化
│   │   ├── 内存管理
│   │   ├── 持久化策略
│   │   ├── 集群配置
│   │   └── 缓存策略
│   └── 实践:性能基准测试
│
└── Day 5-7: 高级并发控制
    ├── 分布式锁
    ├── 事务管理
    ├── 幂等性设计
    ├── 失败重试策略
    └── 实践:高并发压力测试

Week 2: 生产运维
├── Day 8-9: 监控与可观测性
│   ├── LangSmith 集成
│   ├── Prometheus + Grafana
│   ├── 日志聚合(ELK/Loki)
│   ├── 告警配置
│   ├── 追踪(Tracing)
│   └── 实践:构建监控系统
│
└── Day 10: 部署和CI/CD
    ├── Kubernetes 部署
    ├── Helm Charts
    ├── CI/CD 流水线
    ├── 蓝绿部署
    ├── 金丝雀发布
    └── 实践:自动化部署流程

学习重点:

✅ 深入理解平台架构
✅ 精通性能优化
✅ 掌握生产运维
✅ 能够设计企业级解决方案

额外挑战:

🎯 设计高可用(HA)架构
🎯 实现多区域部署
🎯 构建自动扩缩容系统
🎯 优化成本和性能平衡

七、关键技术对比

7.1 客户端连接方式

特性	LangGraph SDK	Remote Graph	原始HTTP
易用性	高(Python风格API)	最高(类似本地图)	低(手动HTTP)
功能完整性	完整(所有API)	基础(图执行)	完整(所有端点)
类型安全	✅ 类型提示	✅ 类型提示	❌ 需手动处理
错误处理	✅ 友好异常	✅ 友好异常	⚠️ HTTP错误码
流式支持	✅ async iterator	✅ async iterator	⚠️ SSE/WebSocket
学习曲线	中等	低(熟悉LangGraph即可)	高(需了解HTTP)
适用场景	生产应用	快速迁移,原型	非Python客户端

7.2 Run 类型对比

类型	Background	Blocking	Streaming
执行方式	异步后台	同步等待	实时推送
返回时机	立即返回run_id	等待完成后返回	边执行边返回chunks
客户端阻塞	❌ 不阻塞	✅ 阻塞	✅ 阻塞(可中断)
状态查询	需主动轮询	自动等待	实时更新
用户体验	提交后可离开	等待完成	实时反馈(打字效果)
适用场景	长时间任务,批处理	简单同步操作	聊天,实时交互
API方法	`runs.create()`	`runs.join()`	`runs.stream()`

7.3 并发策略对比

维度	Reject	Enqueue	Interrupt	Rollback
第一个请求	✅ 完成	✅ 完成	⚠️ 中断但保存	❌ 删除
第二个请求	❌ 拒绝(409)	✅ 排队执行	✅ 立即执行	✅ 立即执行
数据一致性	✅ 高	✅ 高	⚠️ 需careful设计	⚠️ 可能丢失数据
用户体验	⚠️ 需处理错误	✅ 好(无感知)	✅ 好(响应快)	✅ 符合预期
资源使用	✅ 低	⚠️ 中(队列)	⚠️ 中(保存状态)	✅ 低(只保留最新)
实现复杂度	✅ 简单	✅ 简单	⚠️ 复杂	✅ 简单
典型场景	银行转账	批量添加任务	用户纠正错误	用户改变主意

八、最佳实践

8.1 部署配置最佳实践

环境变量管理

bash

# ✅ 好的做法
# .env (不提交到git)
OPENAI_API_KEY=sk-...
DATABASE_URI=postgresql://...

# .env.example (提交到git,提供模板)
OPENAI_API_KEY=your_key_here
DATABASE_URI=postgresql://user:pass@host/db

# .gitignore
.env

Requirements 版本固定

txt

# ❌ 不好
langgraph
langchain-openai

# ✅ 好(固定版本)
langgraph==0.2.0
langchain-openai==0.2.1
langchain-core==0.3.0

# ✅ 更好(兼容版本范围)
langgraph>=0.2.0,<0.3.0

Docker 镜像优化

dockerfile

# ✅ 多阶段构建
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user -r requirements.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY . .
ENV PATH=/root/.local/bin:$PATH
CMD ["python", "-m", "langgraph.server"]

8.2 并发控制最佳实践

选择合适策略

python

# 场景1:严格顺序(银行转账)
multitask_strategy="reject"

# 场景2:批量操作(添加多个任务)
multitask_strategy="enqueue"

# 场景3:用户纠正(改正错误输入)
multitask_strategy="interrupt"

# 场景4:用户改变主意(完全替换)
multitask_strategy="rollback"

客户端错误处理

python

import httpx
import asyncio

async def create_run_with_retry(
    client, thread_id, assistant_id, input,
    max_retries=3,
    strategy="reject"
):
    """带重试的运行创建"""
    for attempt in range(max_retries):
        try:
            return await client.runs.create(
                thread_id,
                assistant_id,
                input=input,
                multitask_strategy=strategy
            )
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 409:
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # 指数退避
                    await asyncio.sleep(wait_time)
                    continue
            raise

8.3 Assistant 设计最佳实践

配置模板化

python

# 定义配置模板
ASSISTANT_TEMPLATES = {
    "personal": {
        "todo_category": "personal",
        "role": "You are a friendly personal assistant...",
        "features": ["reminders", "calendar_sync"]
    },
    "work": {
        "todo_category": "work",
        "role": "You are a professional work assistant...",
        "features": ["time_tracking", "priority_management"]
    }
}

# 创建助手
async def create_assistant_from_template(template_name, user_id):
    template = ASSISTANT_TEMPLATES[template_name]
    config = {
        **template,
        "user_id": user_id
    }
    return await client.assistants.create(
        "task_maistro",
        config={"configurable": config}
    )

版本管理策略

python

# 记录助手变更历史
async def update_assistant_with_changelog(
    assistant_id, 
    new_config,
    changelog_message
):
    # 获取当前版本
    current = await client.assistants.get(assistant_id)
    
    # 更新
    updated = await client.assistants.update(
        assistant_id,
        config=new_config,
        metadata={
            "changelog": changelog_message,
            "previous_version": current["version"]
        }
    )
    
    return updated

8.4 监控和日志

结构化日志

python

import logging
import json

logger = logging.getLogger(__name__)

# 结构化日志
def log_run(event, run_id, thread_id, status, **kwargs):
    log_entry = {
        "event": event,
        "run_id": run_id,
        "thread_id": thread_id,
        "status": status,
        "timestamp": datetime.now().isoformat(),
        **kwargs
    }
    logger.info(json.dumps(log_entry))

# 使用
log_run("run_created", run["run_id"], thread["thread_id"], "pending")
log_run("run_completed", run["run_id"], thread["thread_id"], "success", duration_ms=1500)

性能追踪

python

from langsmith import traceable

@traceable
async def create_and_wait_run(client, thread_id, assistant_id, input):
    """可追踪的运行创建"""
    run = await client.runs.create(
        thread_id,
        assistant_id,
        input=input
    )
    await client.runs.join(thread_id, run["run_id"])
    return run

九、常见问题 FAQ

Q1:部署和开发环境的主要区别是什么?

核心区别:

存储:
- 开发:InMemoryStore + MemorySaver (内存,重启丢失)
- 生产:PostgreSQL + Redis (持久化)
访问方式:
- 开发:graph.invoke() (直接调用)
- 生产:HTTP API (client.runs.create())
并发:
- 开发:单用户,顺序执行
- 生产:多用户,并发控制
配置:
- 开发:硬编码在代码中
- 生产:Assistants + 环境变量

迁移步骤:

python

# 1. 开发代码无需修改
# task_maistro.py 保持不变

# 2. 添加部署配置
# langgraph.json
{
  "graphs": {
    "task_maistro": "./task_maistro.py:graph"
  }
}

# 3. 构建镜像
langgraph build -t my-app

# 4. 启动部署
docker compose up

# 5. 使用SDK连接
client = get_client(url="http://localhost:8123")

Q2:什么时候应该使用哪种并发策略?

决策指南:

python

# 1. Reject - 严格顺序要求
if requires_strict_ordering or resource_constrained:
    strategy = "reject"
    # 示例:银行转账,昂贵的API调用

# 2. Enqueue - 所有请求都重要
elif all_requests_important:
    strategy = "enqueue"
    # 示例:批量任务,客服消息

# 3. Interrupt - 用户纠正/补充
elif user_correcting_or_adding_info:
    strategy = "interrupt"
    # 示例:纠正拼写,补充信息

# 4. Rollback - 用户完全改变主意
elif user_completely_changed_mind:
    strategy = "rollback"
    # 示例:取消订单后重新下单

实际场景:

聊天机器人:
用户:"帮我添加待办:买牛奶"
     (AI正在处理...)
用户:"改成买全脂牛奶"
→ 使用 Interrupt (用户在纠正)

批量导入:
用户:上传了100条数据
     (系统处理第1-50条...)
用户:又上传50条数据
→ 使用 Enqueue (都需要处理)

订单系统:
用户:提交订单A
     (系统处理中...)
用户:取消订单A,提交订单B
→ 使用 Rollback (完全替换)

Q3:如何选择 SDK 还是 Remote Graph?

选择矩阵:

需求	推荐方案	原因
生产环境应用	LangGraph SDK	完整功能,生产就绪
本地图快速迁移	Remote Graph	最小改动,API兼容
需要管理threads/store	LangGraph SDK	完整API支持
只需要执行图	Remote Graph	简单够用
非Python客户端	原始HTTP API	语言无关
需要流式传输	两者都可	都有良好支持

代码对比:

python

# ========== Remote Graph(熟悉的API) ==========
from langgraph.pregel.remote import RemoteGraph

remote_graph = RemoteGraph("task_maistro", url="http://localhost:8123")

# 像本地图一样使用
result = remote_graph.invoke(
    {"messages": [HumanMessage("Hi")]},
    config={"configurable": {"user_id": "alice"}}
)

# ========== LangGraph SDK(生产推荐) ==========
from langgraph_sdk import get_client

client = get_client(url="http://localhost:8123")

# 更多控制
thread = await client.threads.create()
run = await client.runs.create(
    thread["thread_id"],
    "task_maistro",
    input={"messages": [HumanMessage("Hi")]},
    config={"configurable": {"user_id": "alice"}}
)
await client.runs.join(thread["thread_id"], run["run_id"])

Q4:Assistant 和直接传config有什么区别?

功能对比:

python

# ========== 不使用Assistant ==========
# 每次都要传递完整配置
config = {
    "configurable": {
        "user_id": "alice",
        "todo_category": "work",
        "role": "你是一个专业的工作助手..."  # 长文本
    }
}

# 每次调用都要传
run1 = await client.runs.create(thread_id, "task_maistro", input1, config=config)
run2 = await client.runs.create(thread_id, "task_maistro", input2, config=config)
run3 = await client.runs.create(thread_id, "task_maistro", input3, config=config)

# ========== 使用Assistant ==========
# 创建一次
work_assistant = await client.assistants.create(
    "task_maistro",
    config={"configurable": {...}}
)

# 以后只用ID
run1 = await client.runs.create(thread_id, work_assistant["assistant_id"], input1)
run2 = await client.runs.create(thread_id, work_assistant["assistant_id"], input2)
run3 = await client.runs.create(thread_id, work_assistant["assistant_id"], input3)

额外好处:

版本控制:

python

# 更新配置
await client.assistants.update(assistant_id, new_config)
# version 1 → 2

# 可以回滚
old_version = await client.assistants.get(assistant_id, version=1)

多变体管理:

python

# 为不同场景创建不同助手
personal = await client.assistants.create("task_maistro", config1)
work = await client.assistants.create("task_maistro", config2)
family = await client.assistants.create("task_maistro", config3)

持久化存储:

python

# Assistant存储在PostgreSQL
# 重启后仍然存在
# 可以跨应用实例共享

十、实践项目建议

项目 1:部署个人任务助手 🔰 初级

时长:4-6小时
难度:⭐⭐

目标:

部署 task_maistro 到本地
通过 SDK 连接和使用
创建多个 Assistants

步骤:

准备部署文件(langgraph.json, requirements.txt, .env)
使用 docker-compose 启动服务
使用 SDK 创建 personal 和 work assistants
通过不同 assistants 添加任务
测试跨会话记忆

评估标准:

[ ] 部署成功启动
[ ] SDK 连接正常
[ ] 创建了至少2个assistants
[ ] 数据持久化有效

项目 2:构建多用户聊天系统 ⭐ 中级

时长:2-3天
难度:⭐⭐⭐

功能要求:

支持多用户同时使用
每个用户有独立的对话历史
实现流式回复
处理并发请求(Double Texting)
用户可选择不同风格的助手

技术栈:

LangGraph Platform
FastAPI(后端)
WebSocket(流式)
React(前端,可选)

架构:

Frontend (React)
    ↓ WebSocket
Backend (FastAPI)
    ↓ SDK
LangGraph Platform
    ↓
PostgreSQL + Redis

评估标准:

[ ] 多用户隔离正确
[ ] 流式传输流畅
[ ] 并发控制有效
[ ] Assistants 切换正常

项目 3:企业级客服系统 🚀 高级

时长:5-7天
难度:⭐⭐⭐⭐⭐

功能要求:

多租户支持(不同公司)
RBAC权限控制
对话分配和转接
实时监控和分析
自动扩缩容
高可用部署

技术栈:

LangGraph Platform
Kubernetes
Prometheus + Grafana
ELK Stack
Redis Cluster
PostgreSQL HA

架构:

Load Balancer
    ↓
LangGraph Servers (多副本)
    ↓
Redis Cluster
    ↓
PostgreSQL Primary/Replica

评估标准:

[ ] 多租户隔离有效
[ ] 权限控制正确
[ ] 监控完整
[ ] 可以水平扩展
[ ] 高可用验证

十一、下一步学习

完成 Module-7 后,你已经具备了端到端的 LangGraph 应用开发和部署能力!

后续方向

深入 LangGraph 生态
- LangSmith(监控和调试)
- LangServe(替代部署方案)
- LangChain Hub(共享Prompts和Chains)
AI工程进阶
- Prompt Engineering 最佳实践
- RAG系统优化
- Agent评估和测试
DevOps深化
- Kubernetes进阶
- Service Mesh(Istio)
- 可观测性(OpenTelemetry)
业务应用
- 构建自己的AI产品
- 行业解决方案(医疗/金融/教育)
- 开源贡献

十二、总结

通过 Module-7 的学习,你已经掌握:

✅ 部署技能:

Docker容器化
LangGraph CLI使用
存储配置(PostgreSQL + Redis)

✅ API精通:

Runs/Threads/Store三大API
流式传输
错误处理

✅ 并发控制:

Double Texting 4种策略
策略选择决策
客户端重试逻辑

✅ 配置化设计:

Assistants创建和管理
版本控制
多租户支持

你现在可以:

🚀 部署生产级 LangGraph 应用
🛠️ 处理真实世界的并发场景
📊 监控和优化应用性能
🏗️ 设计可扩展的架构

恭喜你完成 LangGraph 完整学习之旅!

继续探索,持续创新,用 AI 改变世界! 🌟

本文档最后更新:2024-11
LangGraph Platform 版本:0.2.0+
作者:LangGraph 教学团队

Module-7: LangGraph 生产部署（Production Deployment）- 本章介绍 ​

一、本章学习目标 ​

核心能力 ​

工程实践 ​

二、核心概念预览 ​

2.1 LangGraph Platform 架构全景 ​

2.2 从开发到生产的转变 ​

2.3 三大 API 类别 ​

三、Docker 容器化基础 ​

3.1 为什么需要 Docker? ​

3.2 核心概念 ​

Docker 镜像 vs 容器 ​

docker-compose ​

3.3 LangGraph 部署所需文件 ​

四、并发控制:Double Texting 问题 ​

4.1 什么是 Double Texting? ​

4.2 为什么这是个问题? ​

4.3 四种策略对比 ​

4.4 策略选择决策树 ​

五、Assistants:配置化设计 ​

5.1 什么是 Assistant? ​

5.2 为什么需要 Assistants? ​

5.3 Assistant 的版本控制 ​

5.4 实际应用场景 ​

六、学习路径建议 ​

路径 1:快速部署 (初学者) ⏱️ 2-3天 ​

路径 2:生产就绪 (中级开发者) ⏱️ 5-7天 ​

路径 3:DevOps 专家 (高级工程师) ⏱️ 7-10天 ​

七、关键技术对比 ​

7.1 客户端连接方式 ​

7.2 Run 类型对比 ​

7.3 并发策略对比 ​

八、最佳实践 ​

8.1 部署配置最佳实践 ​

环境变量管理 ​

Requirements 版本固定 ​

Docker 镜像优化 ​

8.2 并发控制最佳实践 ​

选择合适策略 ​

客户端错误处理 ​

8.3 Assistant 设计最佳实践 ​

配置模板化 ​

版本管理策略 ​

8.4 监控和日志 ​

结构化日志 ​

性能追踪 ​

九、常见问题 FAQ ​

Q1:部署和开发环境的主要区别是什么? ​

Q2:什么时候应该使用哪种并发策略? ​

Q3:如何选择 SDK 还是 Remote Graph? ​

Q4:Assistant 和直接传config有什么区别? ​

十、实践项目建议 ​

项目 1:部署个人任务助手 🔰 初级 ​

项目 2:构建多用户聊天系统 ⭐ 中级 ​

项目 3:企业级客服系统 🚀 高级 ​

十一、下一步学习 ​

后续方向 ​

十二、总结 ​

Module-7: LangGraph 生产部署（Production Deployment）- 本章介绍

一、本章学习目标

核心能力

工程实践

二、核心概念预览

2.1 LangGraph Platform 架构全景

2.2 从开发到生产的转变

2.3 三大 API 类别

三、Docker 容器化基础

3.1 为什么需要 Docker?

3.2 核心概念

Docker 镜像 vs 容器

docker-compose

3.3 LangGraph 部署所需文件

四、并发控制:Double Texting 问题

4.1 什么是 Double Texting?

4.2 为什么这是个问题?

4.3 四种策略对比

4.4 策略选择决策树

五、Assistants:配置化设计

5.1 什么是 Assistant?

5.2 为什么需要 Assistants?

5.3 Assistant 的版本控制

5.4 实际应用场景

六、学习路径建议

路径 1:快速部署 (初学者) ⏱️ 2-3天

路径 2:生产就绪 (中级开发者) ⏱️ 5-7天

路径 3:DevOps 专家 (高级工程师) ⏱️ 7-10天

七、关键技术对比

7.1 客户端连接方式

7.2 Run 类型对比

7.3 并发策略对比

八、最佳实践

8.1 部署配置最佳实践

环境变量管理

Requirements 版本固定

Docker 镜像优化

8.2 并发控制最佳实践

选择合适策略

客户端错误处理

8.3 Assistant 设计最佳实践

配置模板化

版本管理策略

8.4 监控和日志

结构化日志

性能追踪

九、常见问题 FAQ

Q1:部署和开发环境的主要区别是什么?

Q2:什么时候应该使用哪种并发策略?

Q3:如何选择 SDK 还是 Remote Graph?

Q4:Assistant 和直接传config有什么区别?

十、实践项目建议

项目 1:部署个人任务助手 🔰 初级

项目 2:构建多用户聊天系统 ⭐ 中级

项目 3:企业级客服系统 🚀 高级

十一、下一步学习

后续方向

十二、总结