失业一年了，天天想着怎么翻身。前六课我们已经从基础一直学到人机交互，这节课我们要进入真正能上线生产的阶段 —— 生产级特性。本课将重点讲解持久化、部署和监控，让你的LangGraph应用不再只是本地玩具，而是可以稳定运行在服务器上的企业级系统。

本课目标

• 理解LangServe和FastAPI在部署中的作用
• 掌握生产级持久化（PostgreSQL）的正确用法
• 学会使用LangSmith进行监控和调试
• 构建一个可直接部署的编程助手服务

环境准备

pip install -U langgraph langchain langchain-deepseek langchain-core fastapi uvicorn langserve langsmith psycopg
pip install langgraph-checkpoint-postgres

核心概念说明（新增）

什么是FastAPI？
FastAPI 是目前最流行的Python Web框架之一，以高性能、自动生成API文档、类型安全著称。常用于构建RESTful API服务。

什么是LangServe？
LangServe 是LangChain官方推出的工具，专门用来把LangGraph或LangChain链快速包装成REST API。它能自动处理输入输出、流式输出、配置传递等，让你只需几行代码就能把图部署成可通过HTTP调用的服务。

PostgreSQL在LangGraph中存储什么？
PostgreSQL作为生产级Checkpointer，主要存储以下内容：

• 每个thread_id对应的完整状态快照（State）
• 历史checkpoint记录（支持回溯任意历史状态）
• 中断点信息（Human-in-the-loop时保存中断状态）
• 消息历史、自定义字段（如current_code、review_status等）

核心代码实战

代码段1：生产级State定义

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    project_requirements: str
    current_code: str
    review_status: str

代码段2：Agent节点（带生产日志）

from langchain_deepseek import ChatDeepSeek

llm = ChatDeepSeek(model="deepseek-chat", temperature=0.5)

def coding_agent_node(state: State):
    print(f"[生产日志] thread_id: {state.get('__thread_id__')} | 消息数量: {len(state['messages'])}")

    system_prompt = f"""
    你是企业级编程助手。
    项目需求：{state.get('project_requirements', '')}
    当前代码：{state.get('current_code', '尚未开始')}
    """
    response = llm.invoke(system_prompt + "\n\n用户最新要求：" + state["messages"][-1].content)
    return {"messages": [response]}

代码段3：完整生产级图 + FastAPI部署（推荐写法）

from fastapi import FastAPI
from langserve import add_routes
from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

# ==================== PostgreSQL持久化 ====================
conn_string = "postgresql://postgres:password@localhost:5432/langgraph_db"

with psycopg.connect(conn_string) as conn:
    checkpointer = PostgresSaver(conn)
    checkpointer.setup()   # 首次运行时自动创建表

# ==================== 构建图 ====================
workflow = StateGraph(State)
workflow.add_node("agent", coding_agent_node)
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)

graph = workflow.compile(checkpointer=checkpointer)

# ==================== FastAPI + LangServe 部署 ====================
app = FastAPI(
    title="LangGraph 企业级编程助手",
    description="支持持久化、多轮记忆、人机交互的生产级Agent服务",
    version="1.0.0"
)

# 通过LangServe把LangGraph图暴露为API接口
add_routes(
    app,
    graph,
    path="/coding-agent",
    config_keys=["configurable"],   # 支持传递 thread_id
)

# 启动服务命令：
# uvicorn main:app --host 0.0.0.0 --port 8000 --reload

使用示例（生产环境多轮调用）

config = {"configurable": {"thread_id": "prod_project_20260325"}}

# 第一轮调用
graph.invoke({
    "messages": [HumanMessage(content="创建一个用户管理系统")],
    "project_requirements": "企业级FastAPI用户管理系统",
    "current_code": "",
    "review_status": "pending"
}, config=config)

# 第二轮调用（即使服务重启也能继续）
graph.invoke({
    "messages": [HumanMessage(content="增加JWT认证和速率限制")],
}, config=config)

生产级监控技巧

• 开启LangSmith（设置LANGCHAIN_TRACING_V2=true后自动追踪每次调用）
• 查看详细执行链路、token消耗、耗时等
• 使用 graph.get_state(config) 和 graph.get_state_history(config) 查看任意线程的状态快照

小练习（正好2道）

• 练习1（基础）
  修改FastAPI部分，增加一个GET接口 /threads/{thread_id}，用于查询指定thread的当前完整状态（使用graph.get_state实现）。

• 练习2（进阶）
  将Checkpointer从PostgreSQL改回SqliteSaver，并对比两者在多线程并发场景下的差异（可自行思考或简单测试）。

本课小结
本课我们学习了LangGraph的生产级能力：

• 使用FastAPI + LangServe快速把图部署成可供前端/其他服务调用的API；
• 使用PostgreSQL + PostgresSaver实现真正的状态持久化（保存State快照、历史记录和中断信息）；
• 通过LangSmith实现全链路监控。

掌握这些内容后，你的LangGraph应用就已经具备了上线生产的基本条件。

下节预告
第8课：综合大项目 + 未来方向（完结篇）
我们将整合前7课所有知识，完成一个企业级多Agent智能系统（包含RAG、多Agent协作、人机交互、持久化、部署），并展望LangGraph未来的发展方向。

如果觉得这篇有用，欢迎点赞和关注，一起玩转 LangGraph！