feat: 修复数据库持久化,完善服务降级机制
All checks were successful
构建并部署 AI Agent 服务 / deploy (push) Successful in 5m37s

- 恢复使用 AsyncPostgresSaver 持久化短期记忆
- 添加 LLM 作为 Rerank 服务的最后降级方案
- 完善降级链:Local llama.cpp → Zhipu Rerank → LLM Fallback
This commit is contained in:
2026-04-30 17:45:06 +08:00
parent 7a6869ad62
commit 3bf0446ef8
3 changed files with 148 additions and 371 deletions

View File

@@ -4,8 +4,7 @@ FastAPI 后端 - 支持动态模型切换,使用 PostgreSQL 持久化记忆
"""
import os
# from app.config import DB_URI, BACKEND_PORT
from app.config import BACKEND_PORT
from app.config import DB_URI, BACKEND_PORT
import uuid
import json
from contextlib import asynccontextmanager
@@ -15,7 +14,7 @@ from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Depe
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from .agent.service import AIAgentService
from .agent.history import ThreadHistoryService
from app.core.human_review import (
@@ -27,42 +26,56 @@ from app.core.human_review import (
from app.subgraphs.contact.api_client import ContactAPIClient
from app.subgraphs.dictionary.api_client import DictionaryAPIClient
from app.subgraphs.news_analysis.api_client import NewsAPIClient
# from .db.init_db import init_subgraph_tables
# from .db.models import ContactRepository, DictionaryRepository, NewsRepository
from .db.init_db import init_subgraph_tables
from .db.models import ContactRepository, DictionaryRepository, NewsRepository
from app.logger import info, error
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理:创建并注入全局服务(临时用内存 checkpoint"""
# 1. 创建内存 checkpointer(临时测试
checkpointer = MemorySaver()
"""应用生命周期管理:创建并注入全局服务"""
# 1. 创建数据库连接池并初始化表(仅 checkpointer
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
await checkpointer.setup()
# 2. 构建 AI Agent 服务
agent_service = AIAgentService(checkpointer)
await agent_service.initialize()
# 1.5 初始化子图表
await init_subgraph_tables(checkpointer.conn)
# 3. 创建历史查询服务(保持原有的 checkpointer 参数)
history_service = ThreadHistoryService(checkpointer)
# 2. 构建 AI Agent 服务
agent_service = AIAgentService(checkpointer)
await agent_service.initialize()
# 4. 创建审核管理器
review_manager = ReviewManager(InMemoryReviewStore())
# 3. 创建历史查询服务(保持原有的 checkpointer 参数)
history_service = ThreadHistoryService(checkpointer)
# 5. 将服务实例存入 app.state
app.state.agent_service = agent_service
app.state.history_service = history_service
app.state.review_manager = review_manager
app.state.contact_api = ContactAPIClient()
app.state.dictionary_api = DictionaryAPIClient()
app.state.news_api = NewsAPIClient()
app.state.contact_repo = None
app.state.dictionary_repo = None
app.state.news_repo = None
# 3.5 创建子图 Repositories
contact_repo = ContactRepository(checkpointer.conn)
dictionary_repo = DictionaryRepository(checkpointer.conn)
news_repo = NewsRepository(checkpointer.conn)
# 应用运行中...
yield
# 3.6 创建子图 API 客户端(真实数据库模式)
contact_api = ContactAPIClient(checkpointer.conn)
dictionary_api = DictionaryAPIClient(word_repository=dictionary_repo)
news_api = NewsAPIClient(news_repository=news_repo)
# 6. 关闭时清理
info("🛑 应用关闭")
# 4. 创建审核管理器
review_manager = ReviewManager(InMemoryReviewStore())
# 5. 将服务实例存入 app.state
app.state.agent_service = agent_service
app.state.history_service = history_service
app.state.review_manager = review_manager
app.state.contact_api = contact_api
app.state.dictionary_api = dictionary_api
app.state.news_api = news_api
app.state.contact_repo = contact_repo
app.state.dictionary_repo = dictionary_repo
app.state.news_repo = news_repo
# 应用运行中...
yield
# 6. 关闭时自动清理数据库连接async with 负责)
info("🛑 应用关闭,数据库连接池已释放")
app = FastAPI(lifespan=lifespan)

View File

@@ -136,6 +136,92 @@ class ZhipuRerankService(BaseRerankService):
raise
class LLMFallbackRerankService(BaseRerankService):
"""
使用 LLM 作为最后的降级方案进行重排
通过让 LLM 评估文档相关性并给出分数
"""
def __init__(self, llm=None):
from .chat_services import get_chat_service
self.llm = llm or get_chat_service()
def compute_scores(self, query: str, documents: List[str]) -> List[float]:
"""
使用 LLM 评估文档相关性并打分
"""
if not documents:
return []
scores = []
for doc in documents:
score = self._score_single_document(query, doc)
scores.append(score)
return scores
def _score_single_document(self, query: str, document: str) -> float:
"""
让 LLM 为单个文档的相关性打分 (0.0-1.0)
"""
prompt = f"""你是一个文档相关性评分专家。请评估以下文档与查询的相关性返回一个0到1之间的分数
- 1.0表示完全相关
- 0.0表示完全不相关
查询: {query}
文档: {document}
请只返回一个数字,不要解释。"""
try:
result = self.llm.invoke(prompt)
content = result.content if hasattr(result, 'content') else str(result)
# 尝试提取数字
import re
match = re.search(r'(\d+\.?\d*)', content)
if match:
score = float(match.group(1))
# 确保在 0-1 之间
return max(0.0, min(1.0, score))
# 如果没有找到数字返回0.5作为默认值
return 0.5
except Exception as e:
logger.warning(f"LLM 打分失败,返回默认分数 0.5: {e}")
return 0.5
class LLMFallbackRerankProvider(BaseServiceProvider[BaseRerankService]):
"""
LLM 降级重排服务提供者
"""
def __init__(self, llm=None):
super().__init__("llm_fallback_rerank")
self._llm = llm
def is_available(self) -> bool:
"""
LLM 降级方案总是可用(只要 LLM 服务可用)
"""
try:
from .chat_services import get_chat_service
get_chat_service()
logger.info("LLM 降级重排服务可用")
return True
except Exception as e:
logger.warning(f"LLM 降级重排服务不可用: {e}")
return False
def get_service(self) -> BaseRerankService:
"""
获取 LLM 降级重排服务
"""
if self._service_instance is None:
self._service_instance = LLMFallbackRerankService(self._llm)
return self._service_instance
class LocalLlamaCppRerankProvider(BaseServiceProvider[BaseRerankService]):
"""
本地 llama.cpp 重排服务提供者
@@ -221,13 +307,15 @@ def get_rerank_service() -> BaseRerankService:
"""
获取重排服务(带自动降级)- 纯服务层
降级链: Local llama.cpp -> Zhipu Rerank -> LLM Fallback
Returns:
BaseRerankService: 重排服务实例
"""
def _create_chain():
primary = LocalLlamaCppRerankProvider()
fallback = ZhipuRerankProvider()
return FallbackServiceChain(primary, [fallback])
fallbacks = [ZhipuRerankProvider(), LLMFallbackRerankProvider()]
return FallbackServiceChain(primary, fallbacks)
chain = SingletonServiceManager.get_or_create("rerank_service_chain", _create_chain)
return chain.get_available_service()

View File

@@ -1,7 +1,7 @@
#!/bin/bash
# =============================================================================
# AI Agent 启动与管理脚本
# 用法: ./backend/scripts/start.sh both [check|backend|frontend|both|docker-up|docker-down]
# AI Agent 启动与管理脚本 - 简化版
# 用法: ./scripts/start.sh [check|backend|frontend|both]
# =============================================================================
set -e
@@ -14,284 +14,18 @@ YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 项目根目录
PROJECT_DIR="/home/huang/Study/AIProject/Agent1"
PROJECT_DIR="/root/projects/ailine"
echo -e "${BLUE}========================================${NC}"
echo -e "${BLUE} AI Agent - 个人生活助手${NC}"
echo -e "${BLUE}========================================${NC}"
echo ""
# =============================================================================
# 配置检查函数
# =============================================================================
check_config() {
echo -e "${BLUE}📋 开始环境配置检查...${NC}"
echo ""
# 加载 .env 文件
set -a
source "$PROJECT_DIR/.env" 2>/dev/null || true
set +a
PASS=0
FAIL=0
WARN=0
# 辅助函数
check_pass() {
echo -e "${GREEN}${NC} $1"
((PASS++))
}
check_fail() {
echo -e "${RED}${NC} $1"
((FAIL++))
}
check_warn() {
echo -e "${YELLOW}${NC} $1"
((WARN++))
}
# 1. 检查 .env 文件
echo "🔍 检查配置文件..."
if [ -f "$PROJECT_DIR/.env" ]; then
check_pass ".env 文件存在"
# 检查文件格式
if grep -q "^EOF" .env 2>/dev/null; then
check_fail ".env 文件格式错误:发现多余的 EOF 标记"
else
check_pass ".env 文件格式正确"
fi
else
check_fail ".env 文件不存在"
echo " 提示: 请创建 .env 文件并配置环境变量"
return 1
fi
# 2. 检查必需的环境变量
echo ""
echo "🔑 检查环境变量..."
# 检查 ZHIPUAI_API_KEY
if grep -q "^ZHIPUAI_API_KEY=" "$PROJECT_DIR/.env" 2>/dev/null; then
API_KEY=$(grep "^ZHIPUAI_API_KEY=" "$PROJECT_DIR/.env" | head -1 | cut -d'=' -f2- | tr -d '[:space:]')
if [ ${#API_KEY} -gt 10 ]; then
check_pass "ZHIPUAI_API_KEY 已配置(长度: ${#API_KEY}"
else
check_fail "ZHIPUAI_API_KEY 配置可能无效(过短)"
fi
else
check_fail "ZHIPUAI_API_KEY 未配置或格式错误"
fi
# 检查 LLAMACPP_API_KEY
if grep -q "^LLAMACPP_API_KEY=" "$PROJECT_DIR/.env" 2>/dev/null; then
check_pass "LLAMACPP_API_KEY 已配置"
else
check_warn "LLAMACPP_API_KEY 未配置(如不使用本地模型可忽略)"
fi
# 检查 DB_URI (远程服务器)
if grep -q "^DB_URI=.*115.190.121.151" "$PROJECT_DIR/.env" 2>/dev/null; then
check_pass "DB_URI 已配置(远程服务器)"
elif grep -q "^DB_URI=" "$PROJECT_DIR/.env" 2>/dev/null; then
check_warn "DB_URI 已配置(非远程服务器地址)"
else
check_fail "DB_URI 未配置"
fi
# 检查 QDRANT_URL (远程服务器)
if grep -q "^QDRANT_URL=.*115.190.121.151" "$PROJECT_DIR/.env" 2>/dev/null; then
check_pass "QDRANT_URL 已配置(远程服务器)"
elif grep -q "^QDRANT_URL=" "$PROJECT_DIR/.env" 2>/dev/null; then
check_warn "QDRANT_URL 已配置(非远程服务器地址)"
else
check_fail "QDRANT_URL 未配置"
fi
# 3. 检查 Docker 环境
echo ""
echo "🐳 检查 Docker 环境..."
if command -v docker &> /dev/null; then
check_pass "Docker 已安装"
if docker info &> /dev/null; then
check_pass "Docker 守护进程正在运行"
else
check_fail "Docker 守护进程未运行"
echo " 提示: sudo systemctl start docker"
fi
else
check_fail "Docker 未安装"
fi
if command -v docker compose version &> /dev/null || command -v docker-compose &> /dev/null; then
check_pass "Docker Compose 已安装"
else
check_fail "Docker Compose 未安装"
fi
# 4. 检查端口占用
echo ""
echo "🔌 检查端口占用..."
for port in 8081 8082 8083 8501; do
if lsof -i :$port &> /dev/null; then
check_warn "端口 $port 已被占用"
else
check_pass "端口 $port 可用"
fi
done
# 5. 检查远程服务连接
echo ""
echo "🌐 检查远程服务连接..."
# 测试 PostgreSQL 连接(从环境变量读取密码)
if command -v psql &> /dev/null; then
if [ -n "$DB_PASSWORD" ]; then
if PGPASSWORD="$DB_PASSWORD" psql -h 115.190.121.151 -U postgres -d langgraph_db -c "SELECT 1;" &> /dev/null; then
check_pass "PostgreSQL 远程连接正常 (115.190.121.151:5432)"
else
check_fail "PostgreSQL 远程连接失败"
echo " 提示: 检查网络连接和防火墙设置"
fi
else
check_warn "DB_PASSWORD 未配置,跳过 PostgreSQL 连接测试"
fi
else
check_warn "psql 客户端未安装,跳过 PostgreSQL 连接测试"
fi
# 测试 Qdrant 连接
if curl -s http://115.190.121.151:6333/collections &> /dev/null; then
check_pass "Qdrant 远程连接正常 (115.190.121.151:6333)"
else
check_fail "Qdrant 远程连接失败"
echo " 提示: 检查网络连接和防火墙设置"
fi
# 总结
echo ""
echo "=========================================="
echo " 检查结果汇总"
echo "=========================================="
echo -e "${GREEN}通过: $PASS${NC}"
echo -e "${RED}失败: $FAIL${NC}"
echo -e "${YELLOW}警告: $WARN${NC}"
echo ""
if [ $FAIL -eq 0 ]; then
echo -e "${GREEN}✅ 配置检查通过!${NC}"
return 0
else
echo -e "${RED}❌ 发现 $FAIL 个错误,请修复后重试${NC}"
return 1
fi
}
# =============================================================================
# Docker 容器检查函数(仅检查 llama.cpp 服务)
# =============================================================================
check_llamacpp() {
echo -e "${BLUE}🔍 检查 llama.cpp LLM 容器...${NC}"
if ! docker ps --format '{{.Names}}' | grep -q "^gemma4-llamacpp-server$"; then
echo -e "${YELLOW}⚠️ llama.cpp LLM 容器未运行${NC}"
return 1
else
echo -e "${GREEN}✓ llama.cpp LLM 容器正在运行 (gemma4-llamacpp-server)${NC}"
return 0
fi
}
check_embedding() {
echo -e "${BLUE}🔍 检查 llama.cpp Embedding 容器...${NC}"
if ! docker ps --format '{{.Names}}' | grep -q "^embedding-server$"; then
echo -e "${YELLOW}⚠️ llama.cpp Embedding 容器未运行${NC}"
return 1
else
echo -e "${GREEN}✓ llama.cpp Embedding 容器正在运行 (embedding-server)${NC}"
return 0
fi
}
# =============================================================================
# 启动 Docker 依赖服务llama.cpp
# =============================================================================
start_llamacpp() {
echo -e "${BLUE}🚀 启动 llama.cpp LLM 容器...${NC}"
# 检查模型文件
if [ ! -f "/home/huang/Study/AIModel/GGUF/Gemma-4-E2B-Uncensored-HauhauCS-Aggressive-Q6_K_P.gguf" ]; then
echo -e "${RED}✗ 错误LLM 模型文件不存在${NC}"
exit 1
fi
if [ ! -f "/home/huang/Study/AIModel/GGUF/mmproj-Gemma-4-E2B-Uncensored-HauhauCS-Aggressive-f16.gguf" ]; then
echo -e "${RED}✗ 错误:多模态投影文件不存在${NC}"
exit 1
fi
docker run -d \
--name gemma4-llamacpp-server \
--restart=unless-stopped \
--group-add=video \
--device=/dev/kfd \
--device=/dev/dri \
-v /home/huang/Study/AIModel/GGUF:/models \
-p 8081:8080 \
ghcr.io/ggml-org/llama.cpp:server-rocm \
-m /models/Gemma-4-E2B-Uncensored-HauhauCS-Aggressive-Q6_K_P.gguf \
--mmproj /models/mmproj-Gemma-4-E2B-Uncensored-HauhauCS-Aggressive-f16.gguf \
--host 0.0.0.0 \
--port 8080 \
-ngl 99
echo -e "${GREEN}✓ llama.cpp LLM 容器已启动 (端口 8081)${NC}"
echo -e "${YELLOW}⏳ 等待模型加载(可能需要几分钟)...${NC}"
sleep 15
}
start_embedding() {
echo -e "${BLUE}🚀 启动 llama.cpp Embedding 容器...${NC}"
# 检查模型文件
if [ ! -f "/home/huang/Study/AIModel/GGUF/Qwen3-Embedding-0.6B-Q8_0.gguf" ]; then
echo -e "${RED}✗ 错误Embedding 模型文件不存在${NC}"
exit 1
fi
docker run -d \
--name embedding-server \
--restart=unless-stopped \
--group-add=video \
--device=/dev/kfd \
--device=/dev/dri \
-v /home/huang/Study/AIModel/GGUF:/models \
-p 8082:8080 \
-e LLAMA_ARG_CTX_SIZE=16384 \
-e LLAMA_ARG_N_PARALLEL=1 \
-e LLAMA_ARG_BATCH=512 \
-e LLAMA_ARG_N_GPU_LAYERS=99 \
-e LLAMA_ARG_API_KEY="$LLAMACPP_API_KEY" \
ghcr.io/ggml-org/llama.cpp:server-rocm \
-m /models/Qwen3-Embedding-0.6B-Q8_0.gguf \
--host 0.0.0.0 \
--port 8080 \
--embeddings
echo -e "${GREEN}✓ llama.cpp Embedding 容器已启动 (端口 8082)${NC}"
sleep 5
}
# =============================================================================
# 启动 Python 服务
# =============================================================================
start_backend() {
echo -e "\n${BLUE}🚀 启动后端服务 (端口 8079)...${NC}"
echo -e "\n${BLUE}🚀 启动后端服务 (端口 10079)...${NC}"
cd "$PROJECT_DIR"
# 加载 .env 文件中的环境变量
@@ -300,6 +34,7 @@ start_backend() {
set +a
export PYTHONPATH="$PROJECT_DIR/backend"
export BACKEND_PORT=10079
python -m app.backend &
BACKEND_PID=$!
echo -e "${GREEN}✓ 后端服务已启动 (PID: $BACKEND_PID)${NC}"
@@ -307,7 +42,7 @@ start_backend() {
}
start_frontend() {
echo -e "\n${BLUE}🎨 启动前端界面 (端口 8501)...${NC}"
echo -e "\n${BLUE}🎨 启动前端界面 (端口 10501)...${NC}"
cd "$PROJECT_DIR"
# 加载 .env 文件中的环境变量
@@ -316,41 +51,12 @@ start_frontend() {
set +a
export PYTHONPATH="$PROJECT_DIR/frontend/src"
streamlit run frontend/src/frontend_main.py --server.port 8501 --server.address 0.0.0.0 &
export API_URL="http://127.0.0.1:10079/chat"
streamlit run frontend/src/frontend_main.py --server.port 10501 --server.address 0.0.0.0 &
FRONTEND_PID=$!
echo -e "${GREEN}✓ 前端服务已启动 (PID: $FRONTEND_PID)${NC}"
echo -e "${GREEN}✓ 访问地址:${NC}"
echo -e " 本地开发: http://127.0.0.1:8501"
}
# =============================================================================
# Docker Compose 管理
# =============================================================================
docker_up() {
echo -e "${BLUE}🐳 使用 Docker Compose 启动所有服务...${NC}"
cd "$PROJECT_DIR/docker"
# 检查 .env 文件
if [ ! -f "../.env" ]; then
echo -e "${RED}✗ 错误:.env 文件不存在${NC}"
echo " 请先复制配置文件:"
echo " cp .env.docker .env # 服务器部署"
exit 1
fi
docker compose up -d --build
echo -e "\n${GREEN}✓ Docker Compose 服务已启动${NC}"
echo -e "${BLUE}📊 查看服务状态:${NC} docker compose ps"
echo -e "${BLUE}📝 查看日志:${NC} docker compose logs -f"
echo -e "${BLUE}🌐 访问应用:${NC} http://127.0.0.1:8501"
}
docker_down() {
echo -e "${BLUE}🛑 停止 Docker Compose 服务...${NC}"
cd "$PROJECT_DIR/docker"
docker compose down
echo -e "${GREEN}✓ 服务已停止${NC}"
echo -e " 本地开发: http://127.0.0.1:10501"
}
# =============================================================================
@@ -366,10 +72,6 @@ cleanup() {
kill $FRONTEND_PID 2>/dev/null || true
echo -e "${GREEN}✓ 前端服务已停止${NC}"
fi
echo -e "${YELLOW}💡 提示Docker 容器需要手动停止${NC}"
echo -e " 停止 llama.cpp LLM: docker stop gemma4-llamacpp-server"
echo -e " 停止 llama.cpp Embedding: docker stop embedding-server"
echo -e " 或使用: $0 docker-down"
exit 0
}
@@ -380,62 +82,36 @@ trap cleanup SIGINT SIGTERM
# 主逻辑
# =============================================================================
case "${1:-help}" in
check)
check_config
;;
backend)
check_config || exit 1
check_llamacpp || start_llamacpp
check_embedding || start_embedding
start_backend
echo -e "\n${GREEN}后端服务正在运行,按 Ctrl+C 停止${NC}"
wait $BACKEND_PID
;;
frontend)
check_config || exit 1
start_frontend
echo -e "\n${GREEN}前端服务正在运行,按 Ctrl+C 停止${NC}"
wait $FRONTEND_PID
;;
both)
check_config || exit 1
check_llamacpp || start_llamacpp
check_embedding || start_embedding
start_backend
sleep 3
start_frontend
echo -e "\n${GREEN}所有服务正在运行,按 Ctrl+C 停止 Python 服务${NC}"
echo -e "${YELLOW}注意Docker 容器会在后台继续运行${NC}"
echo -e "\n${GREEN}所有服务正在运行,按 Ctrl+C 停止${NC}"
wait
;;
docker-up)
check_config || exit 1
docker_up
;;
docker-down)
docker_down
;;
help|*)
echo -e "${BLUE}用法:${NC} $0 [command]"
echo ""
echo -e "${BLUE}命令:${NC}"
echo " check 检查环境配置"
echo " backend 仅启动后端服务"
echo " frontend 仅启动前端服务"
echo " both 启动前后端服务(默认)"
echo " docker-up 使用 Docker Compose 启动所有服务"
echo " docker-down 停止 Docker Compose 服务"
echo " help 显示此帮助信息"
echo ""
echo -e "${BLUE}示例:${NC}"
echo " $0 check # 检查配置"
echo " $0 both # 启动本地开发环境"
echo " $0 docker-up # 启动 Docker 部署环境"
;;
esac