为 LangGraph 智能体构建生产级后端架构

[ 用户客户端 ] <---> [ FastAPI 后端 ] <---> [ LangGraph 引擎 ]
                            |                          |
                            v                          v
                  [ 业务数据库表 ]             [ Postgres 检查点 ]
                            |                          |
                            +---------> [ PostgreSQL ] <+

# database.py
import os
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/agent_db")

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

class Booking(Base):
    __tablename__ = "bookings"
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    service_name = Column(String, nullable=False)
    booking_time = Column(DateTime, default=datetime.utcnow)
    status = Column(String, default="pending")  # pending, confirmed, cancelled

def init_db():
    Base.metadata.create_all(bind=engine)

pip install langgraph-checkpoint-postgres psycopg

# checkpointer.py
from contextlib import contextmanager
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
import os

DB_URI = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/agent_db")

# 为检查点创建连接池
pool = ConnectionPool(conninfo=DB_URI, max_size=10)

@contextmanager
def get_checkpointer():
    with pool.connection() as conn:
        checkpointer = PostgresSaver(conn)
        # 确保数据库中存在检查点所需的表结构
        checkpointer.setup()
        yield checkpointer

# agent.py
from typing import Annotated, Dict, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from database import SessionLocal, Booking
import os

# 配置 LangChain 使用 n1n.ai 的 API 聚合网关
llm = ChatOpenAI(
    model="claude-3-5-sonnet",
    openai_api_key=os.getenv("N1N_API_KEY"),
    openai_api_base="https://api.n1n.ai/v1"
)

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user_id: int

@tool
def create_booking(service_name: str, config: dict) -> str:
    """为当前用户预订服务。"""
    # 从图配置中获取 user_id
    user_id = config["configurable"].get("user_id")
    if not user_id:
        return "Error: 用户未登录。"

    db = SessionLocal()
    try:
        new_booking = Booking(user_id=user_id, service_name=service_name, status="confirmed")
        db.add(new_booking)
        db.commit()
        db.refresh(new_booking)
        return f"成功预订 {service_name} (预订 ID: {new_booking.id})。"
    except Exception as e:
        db.rollback()
        return f"预订失败: {str(e)}"
    finally:
        db.close()

tools = [create_booking]
llm_with_tools = llm.bind_tools(tools)

def call_model(state: AgentState, config: dict):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages, config)
    return {"messages": [response]}

# 构建图结构
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_edge(START, "agent")

# 我们将在 API 层动态编译图，以便注入持久化 Checkpointer

# main.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from database import init_db, SessionLocal, User
from checkpointer import get_checkpointer
from agent import workflow
from langchain_core.messages import HumanMessage

app = FastAPI(title="LangGraph 生产级后端")

@app.on_event("startup")
def startup_event():
    init_db()

class ChatRequest(BaseModel):
    message: str
    thread_id: str
    user_id: int

@app.post("/chat")
def chat_with_agent(payload: ChatRequest):
    # 验证用户是否存在于业务数据库中
    db = SessionLocal()
    user = db.query(User).filter(User.id == payload.user_id).first()
    db.close()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # 使用上下文管理器获取持久化检查点
    with get_checkpointer() as checkpointer:
        # 编译包含 Postgres 检查点的图
        compiled_graph = workflow.compile(checkpointer=checkpointer)

        # 配置线程与元数据
        config = {
            "configurable": {
                "thread_id": payload.thread_id,
                "user_id": payload.user_id
            }
        }

        # 运行智能体
        initial_state = {
            "messages": [HumanMessage(content=payload.message)],
            "user_id": payload.user_id
        }

        events = compiled_graph.stream(initial_state, config, stream_mode="values")

        # 提取最终的响应消息
        final_message = ""
        for event in events:
            if "messages" in event:
                final_message = event["messages"][-1].content

        return {"response": final_message}

# test.py
import requests
from database import SessionLocal, User

# 在数据库中创建一个测试用户
db = SessionLocal()
if not db.query(User).filter(User.email == "[email protected]").first():
    dev_user = User(email="[email protected]", name="Developer")
    db.add(dev_user)
    db.commit()
db.close()

# 向 FastAPI 服务器发送请求
response = requests.post("http://localhost:8000/chat", json={
    "message": "帮我预订一张机票。",
    "thread_id": "session_abc_123",
    "user_id": 1
})
print(response.json())