如何将 LangGraph AI 智能体连接至 PostgreSQL 实现持久化内存

+-------------------------------------------------------------+
|                       Python 应用程序                       |
|                                                             |
|   +------------------+             +--------------------+   |
|   |  LangGraph 智能体 |             |   PostgresSaver    |   |
|   +--------+---------+             +---------+----------+   |
+------------|---------------------------------|--------------+
             |                                 |
             | (LLM API 调用)                   | (状态检查点写入/读取)
             v                                 v
+----------------------------+     +--------------------------+
|      n1n.ai API 网关       |     |   PostgreSQL 数据库      |
|  (Claude / DeepSeek-V3)    |     |   (Docker / 腾讯云/阿里云) |
+----------------------------+     +--------------------------+

version: '3.8'

services:
  postgres:
    image: postgres:16-alpine
    container_name: langgraph_postgres
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: mysecretpassword
      POSTGRES_DB: langgraph_db
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

docker compose up -d

postgresql://username:password@hostname:port/database_name?sslmode=require

python -m venv venv
source venv/bin/activate  # Windows 系统使用: venv\Scripts\activate
pip install langgraph langgraph-checkpoint-postgres psycopg[binary,pool] langchain-openai python-dotenv

N1N_API_KEY=您的_n1n_api_key
DATABASE_URL=postgresql://postgres:mysecretpassword@localhost:5432/langgraph_db

import os
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from psycopg_pool import ConnectionPool

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI

load_dotenv()

# 定义智能体的状态结构
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# 初始化 LLM 客户端，并配置指向 n1n.ai 网关
# 这样我们可以轻松调用包括 DeepSeek-V3 和 Claude 在内的前沿大模型
llm = ChatOpenAI(
    model="deepseek-ai/DeepSeek-V3",
    api_key=os.getenv("N1N_API_KEY"),
    base_url="https://api.n1n.ai/v1"
)

# 定义调用大模型的节点
def call_model(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 构建状态图（State Graph）
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)

# 获取数据库连接 URI
DB_URI = os.getenv("DATABASE_URL")

def run_agent():
    # 创建 PostgreSQL 连接池
    with ConnectionPool(conninfo=DB_URI, max_size=10) as pool:
        # 初始化 Postgres 检查点管理器
        checkpointer = PostgresSaver(pool)
        
        # 自动创建所需的数据库表结构（如果不存在）
        checkpointer.setup()
        
        # 编译工作流，并注入 Postgres 检查点
        app = workflow.compile(checkpointer=checkpointer)
        
        # 配置会话 ID，用于区分不同用户的对话历史
        config = {"configurable": {"thread_id": "session_user_999"}}
        
        # 第一轮对话
        print("--- 第一轮对话 ---")
        user_msg = HumanMessage(content="你好，我叫小明。请记住我的名字。")
        events = app.stream({"messages": [user_msg]}, config, stream_mode="values")
        for event in events:
            event["messages"][-1].pretty_print()
            
        # 第二轮对话（验证数据库中是否持久化保存了记忆）
        print("\n--- 第二轮对话 ---")
        follow_up = HumanMessage(content="我的名字是什么？")
        events = app.stream({"messages": [follow_up]}, config, stream_mode="values")
        for event in events:
            event["messages"][-1].pretty_print()

if __name__ == "__main__":
    run_agent()

docker exec -it langgraph_postgres psql -U postgres -d langgraph_db