LangGraph 生产级部署全解：FastAPI + Docker

一、部署架构总览我们将基于你之前的带人工干预的双智能体系统构建一个完整的生产级部署方案包含三个核心部分FastAPI 接口层封装 Agent 为标准 HTTP 接口支持任务启动、人工干预、状态查询Redis 持久化层替代 MemorySaver实现状态持久化和服务重启恢复Docker 容器化打包整个应用为镜像实现一键部署和环境一致性标准部署架构图plaintext客户端 → FastAPI 接口 → LangGraph Agent → Redis(状态持久化)二、第一步项目结构标准化先将你的代码整理为生产级项目结构plaintextlanggraph-agent-deploy/ ├── agent/ # Agent核心逻辑 │ ├── __init__.py │ ├── multi_agent.py # 双智能体系统代码 │ └── state.py # 状态定义 ├── main.py # FastAPI入口 ├── requirements.txt # 依赖清单 ├── Dockerfile # 镜像构建文件 ├── docker-compose.yml # 服务编排 └── .env # 环境变量配置三、第二步FastAPI 接口封装3.1 核心依赖安装bash运行pip install fastapi uvicorn python-multipart langgraph-checkpoint-redis3.2 完整 FastAPI 代码# main.py import asyncio from typing import Optional from fastapi import FastAPI, HTTPException, CORS from pydantic import BaseModel from langgraph.types import Command from agent.multi_agent import build_multi_agent_system from agent.state import MultiAgentState # 初始化FastAPI app FastAPI(titleLangGraph 双智能体写作系统 API, version1.0.0) # 配置CORS允许前端跨域访问 app.add_middleware( CORS, allow_origins[*], # 生产环境改为具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 初始化Agent系统全局单例避免重复初始化 agent build_multi_agent_system() # 请求响应模型 class StartTaskRequest(BaseModel): thread_id: str task: str topic: str class ResumeTaskRequest(BaseModel): thread_id: str human_input: str class TaskStatusResponse(BaseModel): thread_id: str status: str # running, interrupted, completed, failed current_step: Optional[str] None interrupt_info: Optional[dict] None final_article: Optional[str] None error: Optional[str] None # 核心接口 app.post(/api/task/start, response_modelTaskStatusResponse) async def start_task(request: StartTaskRequest): 启动一个新的写作任务 try: config {configurable: {thread_id: request.thread_id}} # 初始化任务状态 initial_state: MultiAgentState { task: request.task, topic: request.topic, research_notes: , draft: , final_article: , current_agent: researcher, turn_count: 0, max_turns: 5, messages: [], error: None, human_feedback: } # 异步执行Agent避免阻塞事件循环 result await asyncio.to_thread( agent.invoke, initial_state, configconfig ) # 构建响应 return await _build_status_response(request.thread_id, result) except Exception as e: raise HTTPException(status_code500, detailf任务启动失败: {str(e)}) app.post(/api/task/resume, response_modelTaskStatusResponse) async def resume_task(request: ResumeTaskRequest): 恢复被中断的任务传入人类反馈 try: config {configurable: {thread_id: request.thread_id}} # 用Command恢复执行 result await asyncio.to_thread( agent.invoke, Command(resumerequest.human_input), configconfig ) return await _build_status_response(request.thread_id, result) except Exception as e: raise HTTPException(status_code500, detailf任务恢复失败: {str(e)}) app.get(/api/task/status/{thread_id}, response_modelTaskStatusResponse) async def get_task_status(thread_id: str): 查询任务当前状态 try: config {configurable: {thread_id: thread_id}} # 获取最新状态 state await asyncio.to_thread(agent.get_state, config) result state.values return await _build_status_response(thread_id, result) except Exception as e: raise HTTPException(status_code500, detailf状态查询失败: {str(e)}) # 辅助函数 async def _build_status_response(thread_id: str, result: dict) - TaskStatusResponse: 构建统一的状态响应 # 检查是否有中断 if __interrupt__ in result: interrupt_info result[__interrupt__][0][value] return TaskStatusResponse( thread_idthread_id, statusinterrupted, current_stepinterrupt_info[type], interrupt_infointerrupt_info, errorresult.get(error) ) # 检查是否完成 if result.get(final_article): return TaskStatusResponse( thread_idthread_id, statuscompleted, final_articleresult[final_article], errorresult.get(error) ) # 检查是否出错 if result.get(error): return TaskStatusResponse( thread_idthread_id, statusfailed, errorresult[error] ) # 运行中 return TaskStatusResponse( thread_idthread_id, statusrunning, current_stepresult.get(current_agent), errorresult.get(error) ) # 启动服务 if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)3.3 修改 Agent 代码使用 Redis 持久化将之前的MemorySaver替换为RedisSaver实现状态持久化# agent/multi_agent.py import os from langgraph.checkpoint.redis import RedisSaver def build_multi_agent_system(): # ... 其他代码不变 ... # 从环境变量获取Redis连接地址 redis_url os.getenv(REDIS_URI, redis://localhost:6379/0) checkpointer RedisSaver.from_url(redis_url) return builder.compile(checkpointercheckpointer)四、第三步Docker 容器化4.1 编写 requirements.txt# requirements.txt fastapi0.115.0 uvicorn0.30.6 python-multipart0.0.12 langgraph0.2.39 langchain0.2.16 langchain-openai0.1.25 pydantic-settings2.5.2 python-dotenv1.0.1 langgraph-checkpoint-redis1.0.11 redis5.2.14.2 编写 Dockerfile多阶段构建优化镜像体积dockerfile# Dockerfile # 第一阶段构建阶段 FROM python:3.11-slim AS builder # 设置工作目录 WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y --no-install-recommends \ build-essential \ rm -rf /var/lib/apt/lists/* # 创建虚拟环境 RUN python -m venv /opt/venv ENV PATH/opt/venv/bin:$PATH # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir --upgrade pip \ pip install --no-cache-dir -r requirements.txt # 第二阶段运行阶段 FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 复制虚拟环境 COPY --frombuilder /opt/venv /opt/venv ENV PATH/opt/venv/bin:$PATH # 创建非root用户增强安全性 RUN useradd -m appuser USER appuser # 复制应用代码 COPY --chownappuser:appuser . . # 暴露端口 EXPOSE 8000 # 启动命令 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]4.3 编写 docker-compose.ymlyaml# docker-compose.yml version: 3.8 volumes: redis-data: driver: local services: # Redis服务状态持久化 redis: image: redis:7-alpine container_name: langgraph-redis restart: unless-stopped ports: - 6379:6379 volumes: - redis-data:/data healthcheck: test: [CMD, redis-cli, ping] interval: 5s timeout: 1s retries: 5 # LangGraph Agent API服务 langgraph-agent: build: . container_name: langgraph-agent restart: unless-stopped ports: - 8000:8000 depends_on: redis: condition: service_healthy env_file: - .env environment: - REDIS_URIredis://redis:6379/0 - PYTHONUNBUFFERED1 # 禁用Python输出缓冲确保日志实时显示4.4 编写 .env 配置文件env# .env # 智谱API配置 ZHIPU_API_KEY你的智谱API密钥 ZHIPU_BASE_URLhttps://open.bigmodel.cn/api/paas/v4 LLM_MODELglm-4.6 LLM_BACKUP_MODELglm-4-flash LLM_TIMEOUT30 # Redis配置docker-compose中会覆盖这个值 REDIS_URIredis://localhost:6379/0五、第四步部署与运行5.1 一键启动所有服务# 构建镜像并启动服务 docker-compose up -d --build # 查看服务状态 docker-compose ps # 查看日志 docker-compose logs -f langgraph-agent5.2 测试 API 接口1. 启动任务curl -X POST http://localhost:8000/api/task/start \ -H Content-Type: application/json \ -d { thread_id: test_task_001, task: 写一篇关于LangGraph多智能体系统的技术文章, topic: LangGraph多智能体系统的原理、架构与实现 }2. 查询任务状态bashcurl http://localhost:8000/api/task/status/test_task_0013. 恢复中断任务当任务等待人工确认时bashcurl -X POST http://localhost:8000/api/task/resume \ -H Content-Type: application/json \ -d { thread_id: test_task_001, human_input: 通过 }4. 访问自动生成的 API 文档Swagger UIhttp://localhost:8000/docsReDochttp://localhost:8000/redoc六、生产环境优化最佳实践6.1 性能优化增加 Worker 进程数在 Dockerfile 的 CMD 中调整--workers参数通常设置为 CPU 核心数的 2-4 倍使用 Gunicorn Uvicorn生产环境推荐用 Gunicorn 管理 Uvicorn Worker提高并发能力dockerfileCMD [gunicorn, main:app, -w, 4, -k, uvicorn.workers.UvicornWorker, -b, 0.0.0.0:8000]添加 Nginx 反向代理处理静态资源、SSL 终止、负载均衡6.2 安全优化限制 CORS 来源将allow_origins[*]改为具体的前端域名添加 API 密钥认证给接口添加 API 密钥验证防止未授权访问使用 Docker Secrets生产环境不要用.env 文件存储敏感信息改用 Docker Secrets 或 Kubernetes Secrets定期更新基础镜像及时修复安全漏洞6.3 可观测性添加日志配置配置结构化日志方便日志收集和分析添加健康检查接口pythonapp.get(/health) async def health_check(): return {status: healthy, timestamp: datetime.now().isoformat()}集成监控系统用 Prometheus Grafana 监控服务指标6.4 高可用部署多实例部署启动多个 Agent 实例用 Nginx 或 Kubernetes 进行负载均衡Redis 集群生产环境使用 Redis 集群保证数据高可用自动扩缩容根据请求量自动调整实例数量七、常见问题排查Redis 连接失败检查 docker-compose 中的 Redis 服务是否正常启动REDIS_URI 是否正确接口超时增加 Uvicorn 的超时时间或优化 Agent 执行逻辑状态丢失确保使用了 RedisSaver并且 Redis 数据卷配置正确镜像体积过大使用多阶段构建删除不必要的依赖和文件按照这个方案部署后你的 LangGraph Agent 系统就具备了生产级的稳定性、可扩展性和可维护性可以安全地对外提供服务。