部署

前置阅读：Agent 实战指南 · 短期记忆 · 流式响应

本节解决什么问题

create_agent 在本地开发时只需 agent.invoke() 即可运行。但推向生产时，你需要解决：如何对外提供 HTTP API、如何做流式响应、如何持久化 Agent 状态（让多轮对话和中断恢复跨重启存活）、如何多实例水平扩展、如何管理密钥和环境变量。本节覆盖三条当前推荐的部署路径，并说明各自的持久化策略。

核心概念

必须深刻理解，不能跳过：有状态的 Agent 需要 Checkpointer，多实例需要共享存储。

create_agent 返回的 CompiledStateGraph 默认是无状态的--每次 invoke 都是独立的。要让 Agent 记住对话历史、支持 HITL 中断恢复（详见 HITL），必须配置 checkpointer=。生产环境中：

• 单实例：InMemorySaver 可用，但重启后状态丢失。
• 多实例：必须使用 PostgresSaver 等持久化 Checkpointer，否则不同实例看不到彼此的对话状态。
• LangSmith Agent Server：自动提供持久化，无需手动配置 checkpointer 和 store。
• langgraph up：自带 Postgres 容器，自动配置持久化。

前端类比

部署 Agent 的选择逻辑与部署 Next.js 应用类似：

• 自建 FastAPI ≈ 自行搭建 Express/Koa 服务器，完全可控但需要自己管理一切。
• LangSmith Agent Server ≈ Vercel Functions--托管平台自动处理扩缩容、持久化和监控，你只需上传代码。
• langgraph up/build ≈ docker-compose up--用官方镜像自托管，环境一致性好，适合私有云。

核心考量都是：运维复杂度 vs 控制力度。

原生语义：Agent 的"状态"是 LangGraph 的 checkpoint--每一步执行后，整个 {"messages": [...]} 状态被序列化保存。恢复时从最后一个 checkpoint 反序列化继续执行。多实例部署时，所有实例共享同一个 Checkpointer（通常是 Postgres），通过 thread_id 路由到同一份对话状态。

部署方案概览

方案	适用场景	持久化	运维复杂度	控制力度
自建 FastAPI	需要完全定制 API、已有 FastAPI 基础设施	手动配置 Checkpointer	高	最高
LangSmith Agent Server	快速上线、不想运维基础设施	自动持久化	最低	中
自托管 langgraph up/build	私有云、数据合规、需要环境一致性	自带 Postgres	中	高

路径一：自建 FastAPI + create_agent

最小 API 示例

# app.py
import os

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver


@tool
def lookup_factor(keyword: str) -> str:
    """根据关键词查询环保因子信息。

    Args:
        keyword: 因子名称或别名（如 COD、化学需氧量）
    """
    factors = {
        "COD": {"unit": "mg/L", "standard": "GB 11914-89"},
        "二氧化硫": {"unit": "mg/m³", "standard": "HJ 482-2009"},
    }
    alias_map = {"化学需氧量": "COD", "SO2": "二氧化硫"}
    normalized = alias_map.get(keyword, keyword)
    result = factors.get(normalized)
    return str(result) if result else f"未找到因子: {keyword}"


# 模型配置使用环境变量
model = init_chat_model(os.environ["LLM_MODEL"])

# 开发环境用 InMemorySaver；生产环境用 PostgresSaver（见下文）
checkpointer = InMemorySaver()

agent = create_agent(
    model=model,
    tools=[lookup_factor],
    system_prompt=(
        "你是 EnviroNexus 环保知识库助手。"
        "用户询问因子时使用 lookup_factor 工具查询。"
    ),
    checkpointer=checkpointer,
)

# server.py
import uuid

from fastapi import FastAPI
from pydantic import BaseModel

from app import agent

app = FastAPI(title="EnviroNexus Agent API", version="1.0.0")


class ChatRequest(BaseModel):
    message: str
    thread_id: str | None = None


class ChatResponse(BaseModel):
    thread_id: str
    reply: str


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    """同步调用 Agent。"""
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    result = agent.invoke(
        {"messages": [{"role": "user", "content": request.message}]},
        config=config,
    )

    return ChatResponse(
        thread_id=thread_id,
        reply=result["messages"][-1].content,
    )

启动：

uvicorn server:app --host 0.0.0.0 --port 8000

SSE 流式端点

流式响应让用户尽快看到输出，而不是等待完整回复。详见 流式响应。

# server.py（追加）
import json

from fastapi.responses import StreamingResponse


@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """SSE 流式调用 Agent。"""
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    async def event_generator():
        async for chunk in agent.astream(
            {"messages": [{"role": "user", "content": request.message}]},
            config=config,
            stream_mode="messages",
        ):
            # stream_mode="messages" 返回 (message_chunk, metadata)
            msg_chunk, metadata = chunk
            content = getattr(msg_chunk, "content", "")
            if content:
                data = json.dumps(
                    {"content": content, "thread_id": thread_id},
                    ensure_ascii=False,
                )
                yield f"data: {data}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

生产环境推荐使用 astream_events(version="v3") 获取更细粒度的事件（模型 token 流、工具调用生命周期等），详见 流式响应。

生产环境：PostgresSaver

InMemorySaver 重启后状态丢失，且不支持多实例共享。生产环境使用 PostgresSaver：

# app.py（生产配置）
import os

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

# 从环境变量读取数据库连接字符串，不硬编码密码
db_uri = os.environ["LANGGRAPH_DB_URI"]  # postgresql://user:pass@host:5432/dbname

# 使用连接池
pool = ConnectionPool(
    conninfo=db_uri,
    max_size=20,
    kwargs={"autocommit": True, "prepare_threshold": 0},
)

checkpointer = PostgresSaver(pool)
# 首次运行时创建数据库表
checkpointer.setup()

agent = create_agent(
    model=init_chat_model(os.environ["LLM_MODEL"]),
    tools=[lookup_factor],
    system_prompt="你是 EnviroNexus 环保知识库助手。",
    checkpointer=checkpointer,
)

WARNING

PostgresSaver(conn_string=...) 构造方式不推荐。使用 PostgresSaver.from_conn_string(uri) 上下文管理器或连接池方式，并调用 .setup() 创建表。详见 短期记忆。

健康检查端点

# server.py（追加）
from datetime import datetime

_start_time = datetime.now()


@app.get("/health")
def health_check():
    """存活探针。"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}


@app.get("/ready")
async def readiness_check():
    """就绪探针 -- 检查依赖服务连通性。"""
    checks = {}
    try:
        # 检查 LLM API（轻量请求）
        from app import agent
        await agent.app.nodes  # 确保图已编译
        checks["agent"] = "ok"
    except Exception as e:
        checks["agent"] = f"error: {e}"

    all_ok = all(v == "ok" for v in checks.values())
    return {
        "status": "ready" if all_ok else "not_ready",
        "checks": checks,
        "uptime_seconds": (datetime.now() - _start_time).total_seconds(),
    }

多实例水平扩展

多实例部署时，所有实例必须共享同一个 Postgres Checkpointer。通过 thread_id 路由：

负载均衡器可以使用一致性哈希（按 thread_id 路由到同一实例）减少缓存未命中，但非必须--因为状态在 Postgres 中共享，任何实例都能恢复对话。

环境变量与密钥管理

# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    """应用配置 -- 全部从环境变量读取。"""

    # LLM 配置
    llm_model: str                    # 模型标识符，如 anthropic:claude-sonnet-4-6
    llm_api_key: str                  # LLM API Key
    llm_base_url: str | None = None   # 自定义 API 端点（可选）

    # 数据库（Checkpointer）
    langgraph_db_uri: str             # postgresql://user:pass@host:5432/dbname

    # 应用
    api_key: str                      # 客户端认证 Key
    log_level: str = "INFO"

    # LangSmith（可选）
    langsmith_api_key: str | None = None
    langsmith_tracing: bool = False
    langsmith_project: str = "default"

    model_config = {"env_file": ".env"}

# .env.example -- 复制为 .env 并填入实际值
# LLM 配置（示例名称，可能随时间变化）
LLM_MODEL=anthropic:claude-sonnet-4-6
LLM_API_KEY=sk-ant-xxxxxxxxxxxx
# LLM_BASE_URL=https://api.anthropic.com  # 可选

# 数据库
LANGGRAPH_DB_URI=postgresql://user:password@localhost:5432/langgraph

# 客户端认证
API_KEY=your-client-api-key

# LangSmith（可选）
LANGSMITH_API_KEY=lsv2_pt_xxxxxxxxxxxx
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=environeus-prod

WARNING

不得将密码、API Key 或连接字符串硬编码到代码中。 所有敏感配置通过环境变量或 Secrets Manager 注入。.env 文件加入 .gitignore，不提交到版本控制。

API Key 认证

# server.py（追加）
from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from config import Settings

settings = Settings()
security = HTTPBearer()


def verify_api_key(
    credentials: HTTPAuthorizationCredentials = Security(security),
) -> str:
    """验证客户端 API Key。"""
    if credentials.credentials != settings.api_key:
        raise HTTPException(status_code=401, detail="无效的 API Key")
    return credentials.credentials


@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    _api_key: str = Depends(verify_api_key),
) -> ChatResponse:
    """受保护的 Agent 调用端点。"""
    ...

任务恢复

配置了 Checkpointer 的 Agent 支持任务恢复。如果 Agent 在执行过程中崩溃（如服务器重启），可以从最后一个 checkpoint 恢复：

# 恢复某个 thread 的执行
config = {"configurable": {"thread_id": "user-session-123"}}

# 获取当前状态
state = agent.get_state(config)
if state:
    print(f"当前状态: {state.values}")
    print(f"下一步: {state.next}")

    # 如果被中断（如 HITL 审批），可以恢复
    if state.next:
        # 使用 None 作为输入，从 checkpoint 继续
        result = agent.invoke(None, config=config)

这对 EnviroNexus 的 interrupt 人工审核流程尤其重要：Agent 在等待审核时暂停，审核通过后即使过了几个小时也能恢复执行。详见 HITL。

路径二：LangSmith Agent Server

LangSmith Agent Server 是 LangChain 官方的托管部署平台。它的核心优势是 自动持久化--无需手动配置 Checkpointer 和 Store，平台自动管理。

前端类比

Agent Server 类似于 Vercel Functions：你推送代码，平台自动构建、部署、扩缩容，并提供持久化存储。你不需要管理服务器、数据库或容器编排。

部署步骤

1. 准备 langgraph.json：

{
  "graphs": {
    "enviro_agent": "./app:agent"
  },
  "dependencies": ["."],
  "env": ".env"
}

2. 通过 LangSmith CLI 或 Web UI 部署：

# 安装 CLI
pip install "langgraph-cli[inmem]"

# 登录 LangSmith
langgraph auth

# 部署到 Agent Server
langgraph deploy

3. 调用部署后的 Agent：

from langgraph_sdk import get_client

# 连接到 Agent Server
client = get_client(url="https://your-agent-server.langchain.com")

# 创建线程（自动持久化）
thread = await client.threads.create()

# 发送消息
await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_name="enviro_agent",
    input={"messages": [{"role": "user", "content": "查询 COD 标准号"}]},
)

Agent Server 的自动持久化

能力	说明
自动 Checkpointer	每个 thread 的状态自动持久化，无需手动配置 checkpointer=
自动 Store	长期记忆自动存储，无需手动配置 store=
多实例	平台自动处理水平扩展和负载均衡
HITL 恢复	中断的 Agent 自动暂停，审核后自动恢复
Cron 任务	支持定时触发 Agent 执行

TIP

使用 Agent Server 时，create_agent 不需要 传入 checkpointer= 和 store= 参数。平台会在运行时自动注入。如果你在本地开发时配置了 checkpointer=InMemorySaver()，部署时可以移除或保留--Agent Server 会覆盖本地配置。

适用场景

• 快速上线，不想运维基础设施
• 需要 HITL 中断恢复，但不想自己管理 Postgres
• 团队深度使用 LangSmith 生态（追踪、评估、部署一体化）
• 流量波动大，需要自动扩缩容

不适用场景

• 数据合规要求必须自托管（金融、医疗等）
• 需要完全控制 API 行为和中间件
• 已有自建 FastAPI 基础设施

路径三：自托管 langgraph up / build

langgraph up 和 langgraph build 是 LangGraph CLI 提供的自托管部署方案，基于 Docker。

langgraph up（本地/测试环境）

langgraph up 启动一组 Docker 容器（API 服务器 + Postgres），适合本地测试和私有云部署。

# 安装 CLI（含 Docker 依赖）
pip install "langgraph-cli[inmem]"

# 准备 langgraph.json
cat > langgraph.json << 'EOF'
{
  "graphs": {
    "enviro_agent": "./app:agent"
  },
  "dependencies": ["."],
  "env": ".env"
}
EOF

# 启动（端口 8123）
langgraph up

启动后：

• API 服务器在 http://localhost:8123
• Studio UI 在 https://smith.langchain.com/studio/?baseUrl=http://localhost:8123
• Postgres 自动配置为 Checkpointer 和 Store

langgraph build（生产 Docker 镜像）

langgraph build 构建一个可分发的 Docker 镜像，适合在自有集群（K8s、ECS 等）中部署。

# 构建镜像
langgraph build -t environeus-agent:latest

# 运行（需要外部 Postgres）
docker run -d \
    --name enviro-agent \
    -p 8123:8000 \
    -e LLM_MODEL=anthropic:claude-sonnet-4-6 \
    -e LLM_API_KEY=sk-ant-xxxx \
    -e LANGGRAPH_DB_URI=postgresql://user:pass@db:5432/langgraph \
    environeus-agent:latest

Docker Compose 部署

# docker-compose.yml
version: '3.8'

services:
  agent-api:
    image: environeus-agent:latest
    ports:
      - '8123:8000'
    env_file:
      - .env
    environment:
      - LANGSMITH_TRACING=true
      - LOG_LEVEL=INFO
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 1G
          cpus: '2.0'
    depends_on:
      - postgres
    healthcheck:
      test: ['CMD', 'curl', '-f', 'http://localhost:8000/health']
      interval: 30s
      timeout: 10s
      retries: 3

  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: langgraph
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: langgraph
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U langgraph']
      interval: 10s
      timeout: 5s
      retries: 5

volumes:
  postgres_data:

.dockerignore

__pycache__
*.pyc
.env
.git
.gitignore
tests/
*.md
.vscode/
.idea/

云平台部署

自建 FastAPI 或 langgraph build 产出的 Docker 镜像可以部署到任何云平台：

AWS ECS Fargate

# 使用 AWS Copilot CLI
copilot init \
    --app environeus \
    --name api-service \
    --type "Load Balanced Web Service" \
    --dockerfile ./Dockerfile \
    --deploy

# 注入密钥（不硬编码）
copilot secret init --name LLM_API_KEY
copilot secret init --name LANGGRAPH_DB_URI

GCP Cloud Run

gcloud run deploy enviro-agent \
    --image gcr.io/PROJECT_ID/environeus-agent \
    --port 8000 \
    --region asia-east1 \
    --memory 1Gi \
    --cpu 2 \
    --min-instances 0 \
    --max-instances 10 \
    --set-env-vars LANGSMITH_TRACING=true \
    --set-secrets LLM_API_KEY=llm-key:latest

注意：Cloud Run 有请求超时限制（默认 60s，最大 60 分钟）。长时间运行的 Agent 任务建议使用流式响应或异步处理。

Azure Container Apps

az containerapp up \
    --name enviro-agent \
    --source . \
    --resource-group my-rg \
    --environment langgraph-env \
    --ingress external \
    --target-port 8000 \
    --min-replicas 0 \
    --max-replicas 10 \
    --env-vars LANGSMITH_TRACING=true \
    --secrets llm-key=YOUR_KEY \
    --secret-env-vars LLM_API_KEY=llm-key

生产环境清单

安全性

• [ ] 使用 HTTPS（配置 SSL/TLS 证书）
• [ ] 实现 API Key / JWT 认证
• [ ] 配置 CORS 白名单（禁止使用 *）
• [ ] 密钥通过环境变量或 Secrets Manager 注入，不硬编码
• [ ] 定期轮换 API Key
• [ ] 输入验证和清洗（防止 prompt injection）

性能

• [ ] 配置适当的请求超时（Agent 通常需要 30-120 秒）
• [ ] 实现速率限制
• [ ] 使用连接池（数据库、HTTP 客户端）
• [ ] 启用 SSE 流式响应减少首字延迟
• [ ] 配置 recursion_limit 防止无限循环（在 invoke 时通过 config={"recursion_limit": 25} 传入）

可观测性

• [ ] 结构化日志（JSON 格式）
• [ ] 集成 LangSmith 追踪
• [ ] 健康检查端点（/health + /ready）
• [ ] 配置告警（错误率、延迟、Token 用量）

可靠性

• [ ] 使用 PostgresSaver 替代 InMemorySaver（生产环境）
• [ ] 实现优雅关闭（lifespan）
• [ ] 配置自动重启策略
• [ ] 设置资源限制（CPU、内存）
• [ ] 配置水平自动扩缩
• [ ] 异常重试与断路器

速率限制

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.post("/chat")
@limiter.limit("10/minute")
async def chat(request: Request, body: ChatRequest):
    ...

优雅关闭

from contextlib import asynccontextmanager
import logging

logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动阶段
    logger.info("Agent 服务启动...")
    yield
    # 关闭阶段：清理连接池等资源
    logger.info("Agent 服务关闭，清理资源...")


app = FastAPI(lifespan=lifespan)

常见错误与旧版 API 对照

旧版（仅用于读旧项目，新项目不得采用）

LangServe 于 2024-11-18 弃用。以下代码仅用于理解旧项目，新项目 不得采用：

# ❌ LangServe 写法 -- 已弃用，新项目不得采用
from langserve import add_routes  # LangServe 2024-11 弃用
from langchain.agents import create_tool_calling_agent, AgentExecutor  # 1.0 已删除

agent = create_tool_calling_agent(llm, tools, prompt)  # ImportError
executor = AgentExecutor(agent=agent, tools=tools)      # ImportError
add_routes(app, executor, path="/agent")                # LangServe 弃用

# 旧版 CLI 也已弃用
# langchain serve    # 弃用
# langchain app new  # 弃用
# langchain deploy   # 弃用

迁移方式：

• langchain serve -> langgraph dev（本地开发）或 langgraph up（Docker）
• langchain app new -> langgraph new
• langchain deploy -> langgraph deploy（Agent Server）
• add_routes(app, executor) -> 自建 FastAPI 端点 或 Agent Server
• create_tool_calling_agent + AgentExecutor -> create_agent

部署命令对照

旧版（已弃用）	当前推荐	说明
langchain serve	langgraph dev	本地开发服务器（端口 2024）
langchain app new	langgraph new	创建新项目
langchain deploy	langgraph deploy	部署到 Agent Server
无	langgraph up	启动 Docker 容器组（端口 8123）
无	langgraph build	构建 Docker 镜像
langserve.add_routes()	自建 FastAPI 端点	自定义 API

层级边界与 Deep Agents 交叉引用

本节覆盖的是 create_agent 产生的 Agent 的部署。如果你使用 Deep Agents 构建长周期研究、多文件多 Subagent 的高级 Agent，部署方式类似（Deep Agents 也基于 LangGraph 运行时），但有额外的文件系统、沙箱和子 Agent 编排需求。详见 Deep Agents 部署 和 Deep Agents 生态定位。

选型参考：

• 固定/强约束/可审计流程 -> 显式 LangGraph + 自建 FastAPI
• 普通工具 Agent -> create_agent + 任一部署路径
• 长周期研究/复杂规划/多文件多 Subagent -> Deep Agents（见上述链接）

常见问题

Q: 自建 FastAPI 和 Agent Server 如何选择？

自建 FastAPI 适合需要完全控制 API 行为、已有 FastAPI 基础设施、或有数据合规要求的场景。Agent Server 适合快速上线、不想运维基础设施、需要自动持久化和扩缩容的场景。

Q: 如何处理长时间运行的 Agent 请求？

1. 优先使用 SSE 流式响应（/chat/stream），让用户尽快看到输出
2. 配置 Checkpointer，让 Agent 状态可持久化和恢复
3. 对于超过 5 分钟的任务，考虑异步处理 + Webhook 通知
4. Cloud Run / Lambda 有超时限制，长任务建议用 ECS Fargate 或自托管

Q: 多实例部署时 InMemorySaver 会出什么问题？

不同实例的内存不共享，用户在实例 A 的对话状态在实例 B 看不到。表现为：多轮对话上下文丢失、HITL 中断无法恢复。必须使用 PostgresSaver 等共享存储。

下一步

• 学习 流式响应 优化用户体验
• 了解 可观测性 在生产环境中监控 Agent
• 掌握 测试 确保部署前的质量关
• 阅读 LangSmith Studio 调试部署中的问题
• 了解 HITL 在生产环境中配置人工审批流程

参考资源

• LangGraph CLI 文档
• LangSmith Deployment 文档
• create_agent API 参考
• PostgresSaver API 参考
• FastAPI 官方文档