fd13607366
使用 psycopg_pool 作为 LangGraph checkpointer 连接池，并在启动时执行 setup() 确保表存在；Windows 下切换为 SelectorEventLoopPolicy 以兼容 psycopg 异步模式。
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
from contextlib import asynccontextmanager
|
||
|
||
from fastapi import FastAPI
|
||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
||
|
||
from src.services.mcp_service import init_mcp_servers
|
||
from src.services.run_queue_service import close_queue_clients, get_redis_client
|
||
from src.services.task_service import tasker
|
||
from src.storage.postgres.manager import pg_manager
|
||
from src.knowledge import knowledge_base
|
||
from src.sandbox import init_sandbox_provider, shutdown_sandbox_provider
|
||
from src.utils import logger
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
"""FastAPI lifespan事件管理器"""
|
||
# 初始化数据库连接
|
||
try:
|
||
pg_manager.initialize()
|
||
await pg_manager.create_business_tables()
|
||
await pg_manager.ensure_business_schema()
|
||
await pg_manager.ensure_knowledge_schema()
|
||
except Exception as e:
|
||
logger.error(f"Failed to initialize database during startup: {e}")
|
||
|
||
# 初始化 MCP 服务器配置
|
||
try:
|
||
await init_mcp_servers()
|
||
except Exception as e:
|
||
logger.error(f"Failed to initialize MCP servers during startup: {e}")
|
||
|
||
# 初始化知识库管理器
|
||
try:
|
||
await knowledge_base.initialize()
|
||
except Exception as e:
|
||
logger.error(f"Failed to initialize knowledge base manager: {e}")
|
||
|
||
# 预热 Redis（run 队列）
|
||
try:
|
||
redis = await get_redis_client()
|
||
await redis.ping()
|
||
except Exception as e:
|
||
logger.warning(f"Run queue redis unavailable on startup: {e}")
|
||
|
||
try:
|
||
init_sandbox_provider()
|
||
except Exception as e:
|
||
logger.error(f"Failed to initialize sandbox provider during startup: {e}")
|
||
|
||
# =========================================================
|
||
# 2. 核心修复：在这里执行一次 setup()，建完表就拉倒
|
||
# =========================================================
|
||
checkpointer = AsyncPostgresSaver(pg_manager.langgraph_pool)
|
||
await checkpointer.setup()
|
||
print("LangGraph Checkpoint tables verified/created!")
|
||
|
||
await tasker.start()
|
||
yield
|
||
await tasker.shutdown()
|
||
shutdown_sandbox_provider()
|
||
await close_queue_clients()
|
||
await pg_manager.close()
|