1100 lines
36 KiB
Python
Executable File
1100 lines
36 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
RAG搜索路由 - 支持新的索引格式
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import pickle
|
||
from pathlib import Path
|
||
from typing import Optional, List
|
||
from fastapi import APIRouter, HTTPException
|
||
from pydantic import BaseModel
|
||
|
||
# 添加项目路径
|
||
PROJECT_ROOT = Path(os.environ.get("PLANTAI_PROJECT_ROOT", "") or Path(__file__).resolve().parents[1])
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
extra_site_packages = os.environ.get("GF_PYTHON_SITE_PACKAGES", "").strip()
|
||
if extra_site_packages:
|
||
sys.path.insert(0, extra_site_packages)
|
||
|
||
from src.rag.base import SearchResult
|
||
from src.rag.vector_store import InMemoryStore
|
||
from src.rag.embeddings import BGEEmbedding
|
||
from src.rag.agentic.query_analyzer import QueryAnalyzer, QueryType
|
||
from src.rag.agentic.retrieval_strategy import RetrievalStrategyAgent, RetrievalMode
|
||
|
||
router = APIRouter()
|
||
|
||
# 初始化Agent
|
||
query_analyzer = QueryAnalyzer()
|
||
strategy_agent = RetrievalStrategyAgent()
|
||
|
||
# 初始化LLM客户端
|
||
_llm_manager = None
|
||
def get_llm_manager():
|
||
global _llm_manager
|
||
if _llm_manager is None:
|
||
try:
|
||
from src.llm.manager import LLMManager
|
||
import yaml
|
||
|
||
# 读取配置文件
|
||
config_path = PROJECT_ROOT / "config" / "llm.yaml"
|
||
if config_path.exists():
|
||
with open(config_path, 'r') as f:
|
||
config = yaml.safe_load(f)
|
||
_llm_manager = LLMManager(config)
|
||
provider = config.get("default", {}).get("provider", "unknown")
|
||
print(f"LLMManager初始化成功（使用{provider}）")
|
||
else:
|
||
# 配置文件不存在，使用Ollama作为fallback
|
||
_llm_manager = LLMManager({
|
||
"default": {"provider": "ollama", "model": "qwen3:0.6b"}
|
||
})
|
||
print("LLMManager初始化成功（使用Ollama fallback）")
|
||
except Exception as e:
|
||
print(f"LLM客户端初始化失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return _llm_manager
|
||
|
||
# 全局变量存储vector store
|
||
_vector_store = None
|
||
_embedding_model = None
|
||
|
||
def get_vector_store():
|
||
"""懒加载vector store"""
|
||
global _vector_store, _embedding_model
|
||
|
||
if _vector_store is None:
|
||
index_path = PROJECT_ROOT / "data" / "rag_index_final" / "vector_index.pkl"
|
||
|
||
if not index_path.exists():
|
||
raise HTTPException(status_code=500, detail="索引文件不存在，请先运行索引构建")
|
||
|
||
# 创建embedding model
|
||
_embedding_model = BGEEmbedding()
|
||
|
||
# 创建store
|
||
_vector_store = InMemoryStore(embedding_model=_embedding_model)
|
||
|
||
# 加载索引
|
||
with open(index_path, 'rb') as f:
|
||
data = pickle.load(f)
|
||
|
||
_vector_store._chunks = data['chunks']
|
||
_vector_store._embeddings = data['embeddings']
|
||
|
||
print(f"RAG索引加载完成: {len(_vector_store._chunks)} chunks")
|
||
|
||
return _vector_store
|
||
|
||
|
||
class RAGQueryRequest(BaseModel):
|
||
query: str
|
||
mode: str = "hybrid" # hybrid, normal, exact, alias, substring, bm25
|
||
top_k: int = 10
|
||
include_sources: bool = True
|
||
use_llm: bool = False # 是否使用LLM生成总结
|
||
|
||
|
||
class PhylogenyExtractionRequest(BaseModel):
|
||
"""系统发育提取请求"""
|
||
pdf_path: str # PDF文件路径
|
||
doi: Optional[str] = None # DOI（可选，用于获取补充材料）
|
||
pmcid: Optional[str] = None # PMCID（可选）
|
||
auto_fetch_supplementary: bool = True # 是否自动获取补充材料
|
||
|
||
|
||
class RAGSource(BaseModel):
|
||
doc_id: str
|
||
content: str
|
||
score: float
|
||
metadata: dict
|
||
|
||
|
||
class KnowledgeGraphData(BaseModel):
|
||
nodes: List[dict] = []
|
||
edges: List[dict] = []
|
||
|
||
|
||
class RAGQueryResponse(BaseModel):
|
||
response: str
|
||
sources: List[RAGSource] = []
|
||
mode: str
|
||
total_chunks: int
|
||
knowledge_graph: Optional[KnowledgeGraphData] = None # 动态生成的知识图谱数据
|
||
|
||
|
||
@router.post("/chat")
|
||
async def rag_chat(request: RAGQueryRequest):
|
||
"""RAG搜索端点"""
|
||
try:
|
||
store = get_vector_store()
|
||
|
||
# 使用Agent编排进行智能检索
|
||
if request.mode == "agentic":
|
||
# Agent编排模式
|
||
results = await _agentic_search(store, request.query, request.top_k)
|
||
elif request.mode == "exact":
|
||
results = await _exact_search(store, request.query, request.top_k)
|
||
elif request.mode == "substring":
|
||
results = await _substring_search(store, request.query, request.top_k)
|
||
elif request.mode == "bm25":
|
||
results = await _bm25_search(store, request.query, request.top_k)
|
||
else:
|
||
# hybrid search
|
||
results = await _hybrid_search(store, request.query, request.top_k)
|
||
|
||
# 格式化响应
|
||
sources = []
|
||
for result in results:
|
||
sources.append(RAGSource(
|
||
doc_id=result.chunk.doc_id,
|
||
content=result.chunk.content[:500],
|
||
score=result.score,
|
||
metadata=result.chunk.metadata
|
||
))
|
||
|
||
# 生成回答（支持LLM总结）
|
||
response_text = await _generate_response_async(request.query, sources, request.use_llm)
|
||
|
||
# 从检索结果动态提取知识图谱
|
||
kg_data = _extract_knowledge_graph(sources, request.query)
|
||
|
||
return RAGQueryResponse(
|
||
response=response_text,
|
||
sources=sources if request.include_sources else [],
|
||
mode=request.mode,
|
||
total_chunks=len(store._chunks),
|
||
knowledge_graph=kg_data
|
||
)
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"搜索失败: {str(e)}")
|
||
|
||
|
||
async def _exact_search(store, query: str, top_k: int):
|
||
"""精确匹配搜索"""
|
||
from src.rag.base import SearchResult
|
||
|
||
query_lower = query.lower().strip()
|
||
results = []
|
||
|
||
for chunk_id, chunk in store._chunks.items():
|
||
content = chunk.content.lower()
|
||
count = content.count(query_lower)
|
||
if count > 0:
|
||
score = 0.8 + 0.19 * min(count, 10) / 10
|
||
if content == query_lower:
|
||
score = 1.0
|
||
results.append(SearchResult(chunk=chunk, score=score))
|
||
|
||
results.sort(key=lambda x: x.score, reverse=True)
|
||
return results[:top_k]
|
||
|
||
|
||
async def _substring_search(store, query: str, top_k: int):
|
||
"""子串搜索"""
|
||
import re
|
||
from src.rag.base import SearchResult
|
||
|
||
query_lower = query.lower().strip()
|
||
|
||
# 提取查询词
|
||
if ' ' in query_lower:
|
||
query_words = [w.strip() for w in query_lower.split() if len(w.strip()) >= 2]
|
||
else:
|
||
query_words = re.findall(r'[\u4e00-\u9fa5]{2,}|[a-zA-Z]+', query_lower)
|
||
if not query_words:
|
||
query_words = [query_lower] if len(query_lower) >= 2 else []
|
||
|
||
if not query_words:
|
||
query_words = [query_lower] if query_lower else []
|
||
|
||
results = []
|
||
|
||
for chunk_id, chunk in store._chunks.items():
|
||
content_lower = chunk.content.lower()
|
||
|
||
total_score = 0
|
||
matched_words = 0
|
||
|
||
for word in query_words:
|
||
if word in content_lower:
|
||
count = content_lower.count(word)
|
||
word_weight = min(len(word) / 10, 1.0) + 0.5
|
||
total_score += 0.1 * count * word_weight
|
||
matched_words += 1
|
||
|
||
if matched_words > 0:
|
||
coverage = matched_words / len(query_words) if query_words else 0
|
||
base_score = 0.5
|
||
coverage_bonus = 0.3 * coverage
|
||
density_score = min(total_score, 0.2)
|
||
score = base_score + coverage_bonus + density_score
|
||
results.append(SearchResult(chunk=chunk, score=min(score, 1.0)))
|
||
|
||
results.sort(key=lambda x: x.score, reverse=True)
|
||
return results[:top_k]
|
||
|
||
|
||
async def _bm25_search(store, query: str, top_k: int):
|
||
"""BM25搜索"""
|
||
from src.rag.base import SearchResult
|
||
|
||
try:
|
||
from rank_bm25 import BM25Okapi
|
||
except ImportError:
|
||
# 如果没有BM25，使用简单关键词匹配
|
||
return await _substring_search(store, query, top_k)
|
||
|
||
# 构建corpus
|
||
corpus = []
|
||
chunk_ids = []
|
||
|
||
for chunk_id, chunk in store._chunks.items():
|
||
corpus.append(chunk.content)
|
||
chunk_ids.append(chunk_id)
|
||
|
||
if not corpus:
|
||
return []
|
||
|
||
bm25 = BM25Okapi(corpus)
|
||
tokenized_query = query.lower().split()
|
||
scores = bm25.get_scores(tokenized_query)
|
||
|
||
# 获取top_k
|
||
top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]
|
||
|
||
results = []
|
||
for idx in top_indices:
|
||
if scores[idx] > 0:
|
||
chunk_id = chunk_ids[idx]
|
||
chunk = store._chunks[chunk_id]
|
||
results.append(SearchResult(chunk=chunk, score=min(scores[idx] / 10, 1.0)))
|
||
|
||
return results
|
||
|
||
|
||
async def _hybrid_search(store, query: str, top_k: int, alpha: float = 0.7):
|
||
"""混合搜索"""
|
||
# 向量搜索
|
||
try:
|
||
query_embedding = await store.embedding_model.aembed(query)
|
||
vector_results = store.search(query_embedding, top_k * 2, threshold=0.0)
|
||
except:
|
||
vector_results = []
|
||
|
||
# BM25搜索
|
||
bm25_results = await _bm25_search(store, query, top_k * 2)
|
||
|
||
# 子串搜索
|
||
substring_results = await _substring_search(store, query, top_k * 2)
|
||
|
||
# 合并结果
|
||
combined = {}
|
||
|
||
for r in vector_results:
|
||
chunk_id = r.chunk.chunk_id
|
||
combined[chunk_id] = (alpha * r.score, r.chunk)
|
||
|
||
for r in bm25_results:
|
||
chunk_id = r.chunk.chunk_id
|
||
if chunk_id in combined:
|
||
old_score, chunk = combined[chunk_id]
|
||
combined[chunk_id] = (old_score + (1 - alpha) * r.score * 0.5, chunk)
|
||
else:
|
||
combined[chunk_id] = ((1 - alpha) * r.score * 0.5, r.chunk)
|
||
|
||
for r in substring_results:
|
||
chunk_id = r.chunk.chunk_id
|
||
if chunk_id in combined:
|
||
old_score, chunk = combined[chunk_id]
|
||
combined[chunk_id] = (old_score + (1 - alpha) * r.score * 0.3, chunk)
|
||
else:
|
||
combined[chunk_id] = ((1 - alpha) * r.score * 0.3, r.chunk)
|
||
|
||
sorted_results = sorted(combined.items(), key=lambda x: x[1][0], reverse=True)
|
||
|
||
from src.rag.base import SearchResult
|
||
return [SearchResult(chunk=chunk, score=score) for _, (score, chunk) in sorted_results[:top_k]]
|
||
|
||
|
||
async def _generate_response_async(query: str, sources: List[RAGSource], use_llm: bool = False) -> str:
|
||
"""生成回答 - 支持LLM总结（异步版本）"""
|
||
if not sources:
|
||
return "未找到相关文献。"
|
||
|
||
# 如果使用LLM生成总结
|
||
if use_llm:
|
||
llm_mgr = get_llm_manager()
|
||
if llm_mgr:
|
||
# 构造上下文
|
||
context = "\n\n".join([
|
||
f"【文献{i+1}】{s.content[:500]}"
|
||
for i, s in enumerate(sources[:5])
|
||
])
|
||
|
||
prompt = f"""你是一个基因家族分析专家。请根据以下检索到的文献内容，回答用户的问题。
|
||
|
||
用户问题: {query}
|
||
|
||
文献内容:
|
||
{context}
|
||
|
||
请生成一个专业、详细的回答，包括：
|
||
1. **核心概念解释** - 清晰解释相关概念
|
||
2. **主要研究发现** - 总结文献中的关键发现
|
||
3. **生物学意义** - 说明其在生物学/育种中的应用价值
|
||
|
||
格式要求：
|
||
- 使用 Markdown 语法（加粗、列表等）
|
||
- 结构清晰，段落分明
|
||
- 关键信息用 **加粗** 标注
|
||
|
||
如果文献内容与问题不完全相关，请提取有价值的信息并说明。"""
|
||
|
||
try:
|
||
from src.llm.base import LLMMessage
|
||
messages = [LLMMessage(role="user", content=prompt)]
|
||
# 使用异步调用
|
||
response = await llm_mgr.acall(messages)
|
||
return response.content if hasattr(response, 'content') else str(response)
|
||
except Exception as e:
|
||
print(f"LLM生成失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
# 默认统计信息
|
||
species_list = set()
|
||
gf_list = set()
|
||
|
||
for s in sources:
|
||
if 'species' in s.metadata:
|
||
species_list.add(s.metadata['species'])
|
||
if 'gene_family' in s.metadata:
|
||
gf_list.add(s.metadata['gene_family'])
|
||
|
||
response = f"找到 {len(sources)} 条相关文献。"
|
||
|
||
if species_list:
|
||
response += f"\n涉及物种: {', '.join(species_list)}。"
|
||
|
||
if gf_list:
|
||
response += f"\n涉及基因家族: {', '.join(gf_list)}。"
|
||
|
||
return response
|
||
|
||
|
||
async def _agentic_search(store, query: str, top_k: int):
|
||
"""
|
||
Agent编排搜索 - 使用新的Orchestrator启用Self-RAG和MultiHopRAG
|
||
"""
|
||
# 使用新的 RAGAgentOrchestrator
|
||
try:
|
||
from src.rag.agentic.orchestrator import RAGAgentOrchestrator
|
||
|
||
# 创建orchestrator实例
|
||
orchestrator = RAGAgentOrchestrator(
|
||
vector_store=store,
|
||
retriever=None, # 使用store直接检索
|
||
max_iterations=3,
|
||
enable_self_rag=True,
|
||
enable_multi_hop=True,
|
||
multi_hop_threshold=0.3
|
||
)
|
||
|
||
# 运行编排
|
||
result = await orchestrator.run(query, store_memory=False)
|
||
|
||
# 转换结果为SearchResult列表
|
||
from src.rag.base import Chunk, SearchResult
|
||
results = []
|
||
for r_data in result.get("results", []):
|
||
chunk = Chunk(
|
||
chunk_id=r_data.get("chunk_id", "unknown"),
|
||
content=r_data.get("content", ""),
|
||
doc_id=r_data.get("doc_id", "unknown"),
|
||
metadata=r_data.get("metadata", {})
|
||
)
|
||
results.append(SearchResult(chunk=chunk, score=r_data.get("score", 0.5)))
|
||
|
||
return results[:top_k]
|
||
|
||
except Exception as e:
|
||
print(f"Orchestrator failed: {e}, falling back to hybrid search")
|
||
# 回退到hybrid search
|
||
return await _hybrid_search(store, query, top_k)
|
||
|
||
|
||
@router.get("/rag/status")
|
||
async def rag_status():
|
||
"""RAG系统状态"""
|
||
try:
|
||
store = get_vector_store()
|
||
return {
|
||
"status": "ready",
|
||
"total_chunks": len(store._chunks),
|
||
"total_documents": len(set(c.doc_id for c in store._chunks.values())),
|
||
"agent_enabled": True
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"status": "error",
|
||
"error": str(e)
|
||
}
|
||
|
||
|
||
@router.post("/rag/analyze")
|
||
async def rag_analyze(request: RAGQueryRequest):
|
||
"""查询分析端点 - 测试Agent编排"""
|
||
try:
|
||
# 查询分析
|
||
analysis = query_analyzer.analyze(request.query)
|
||
|
||
# 确定检索策略
|
||
strategy = strategy_agent.determine_strategy(analysis)
|
||
params = strategy_agent.get_search_params(strategy)
|
||
|
||
return {
|
||
"query": request.query,
|
||
"analysis": {
|
||
"type": analysis.query_type.value,
|
||
"intent": analysis.intent,
|
||
"entities": analysis.entities,
|
||
"keywords": analysis.keywords[:10],
|
||
"required_info": analysis.required_info
|
||
},
|
||
"strategy": {
|
||
"primary_mode": strategy.primary_mode.value,
|
||
"secondary_modes": [m.value for m in strategy.secondary_modes],
|
||
"top_k": strategy.top_k,
|
||
"threshold": strategy.threshold,
|
||
"rerank": strategy.rerank
|
||
},
|
||
"search_params": params
|
||
}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"分析失败: {str(e)}")
|
||
|
||
|
||
# ===== 用户反馈API =====
|
||
|
||
from datetime import datetime
|
||
import json
|
||
import sqlite3
|
||
|
||
# 反馈数据库路径
|
||
FEEDBACK_DB = PROJECT_ROOT / "data" / "feedback.db"
|
||
|
||
def init_feedback_db():
|
||
"""初始化反馈数据库"""
|
||
FEEDBACK_DB.parent.mkdir(parents=True, exist_ok=True)
|
||
conn = sqlite3.connect(FEEDBACK_DB)
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS feedback (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
query TEXT NOT NULL,
|
||
answer TEXT,
|
||
rating INTEGER, -- 1-5 评分
|
||
feedback_type TEXT, -- thumbs_up/thumbs_down/correction
|
||
feedback_text TEXT, -- 用户反馈内容
|
||
session_id TEXT,
|
||
timestamp TEXT NOT NULL
|
||
)
|
||
""")
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS evaluation_history (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
question_id TEXT,
|
||
query TEXT NOT NULL,
|
||
query_type TEXT,
|
||
ndcg REAL,
|
||
answer_quality REAL,
|
||
timestamp TEXT NOT NULL
|
||
)
|
||
""")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
class FeedbackRequest(BaseModel):
|
||
query: str
|
||
answer: str = None
|
||
rating: int = None # 1-5
|
||
feedback_type: str = None # thumbs_up/thumbs_down/correction
|
||
feedback_text: str = None
|
||
session_id: str = None
|
||
|
||
|
||
class EvaluationRecordRequest(BaseModel):
|
||
question_id: str
|
||
query: str
|
||
query_type: str
|
||
ndcg: float
|
||
answer_quality: float
|
||
|
||
|
||
@router.post("/rag/feedback")
|
||
async def submit_feedback(request: FeedbackRequest):
|
||
"""提交用户反馈"""
|
||
try:
|
||
init_feedback_db()
|
||
conn = sqlite3.connect(FEEDBACK_DB)
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("""
|
||
INSERT INTO feedback (query, answer, rating, feedback_type, feedback_text, session_id, timestamp)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
request.query,
|
||
request.answer,
|
||
request.rating,
|
||
request.feedback_type,
|
||
request.feedback_text,
|
||
request.session_id,
|
||
datetime.now().isoformat()
|
||
))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return {"status": "success", "message": "反馈已保存"}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"保存反馈失败: {str(e)}")
|
||
|
||
|
||
@router.post("/rag/evaluation/record")
|
||
async def record_evaluation(request: EvaluationRecordRequest):
|
||
"""记录评估结果"""
|
||
try:
|
||
init_feedback_db()
|
||
conn = sqlite3.connect(FEEDBACK_DB)
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("""
|
||
INSERT INTO evaluation_history (question_id, query, query_type, ndcg, answer_quality, timestamp)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
request.question_id,
|
||
request.query,
|
||
request.query_type,
|
||
request.ndcg,
|
||
request.answer_quality,
|
||
datetime.now().isoformat()
|
||
))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return {"status": "success", "message": "评估记录已保存"}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"保存评估记录失败: {str(e)}")
|
||
|
||
|
||
@router.get("/rag/feedback/stats")
|
||
async def get_feedback_stats():
|
||
"""获取反馈统计"""
|
||
try:
|
||
init_feedback_db()
|
||
conn = sqlite3.connect(FEEDBACK_DB)
|
||
cursor = conn.cursor()
|
||
|
||
# 总反馈数
|
||
cursor.execute("SELECT COUNT(*) FROM feedback")
|
||
total = cursor.fetchone()[0]
|
||
|
||
# 评分分布
|
||
cursor.execute("""
|
||
SELECT rating, COUNT(*)
|
||
FROM feedback
|
||
WHERE rating IS NOT NULL
|
||
GROUP BY rating
|
||
""")
|
||
rating_dist = {row[0]: row[1] for row in cursor.fetchall()}
|
||
|
||
# 反馈类型分布
|
||
cursor.execute("""
|
||
SELECT feedback_type, COUNT(*)
|
||
FROM feedback
|
||
WHERE feedback_type IS NOT NULL
|
||
GROUP BY feedback_type
|
||
""")
|
||
feedback_type_dist = {row[0]: row[1] for row in cursor.fetchall()}
|
||
|
||
# 平均评分
|
||
cursor.execute("SELECT AVG(rating) FROM feedback WHERE rating IS NOT NULL")
|
||
avg_rating = cursor.fetchone()[0] or 0
|
||
|
||
conn.close()
|
||
|
||
return {
|
||
"total_feedback": total,
|
||
"rating_distribution": rating_dist,
|
||
"feedback_type_distribution": feedback_type_dist,
|
||
"average_rating": round(avg_rating, 2)
|
||
}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"获取统计失败: {str(e)}")
|
||
|
||
|
||
@router.get("/rag/evaluation/history")
|
||
async def get_evaluation_history(limit: int = 50):
|
||
"""获取评估历史"""
|
||
try:
|
||
init_feedback_db()
|
||
conn = sqlite3.connect(FEEDBACK_DB)
|
||
cursor = conn.cursor()
|
||
|
||
cursor.execute("""
|
||
SELECT question_id, query, query_type, ndcg, answer_quality, timestamp
|
||
FROM evaluation_history
|
||
ORDER BY timestamp DESC
|
||
LIMIT ?
|
||
""", (limit,))
|
||
|
||
results = []
|
||
for row in cursor.fetchall():
|
||
results.append({
|
||
"question_id": row[0],
|
||
"query": row[1],
|
||
"query_type": row[2],
|
||
"ndcg": row[3],
|
||
"answer_quality": row[4],
|
||
"timestamp": row[5]
|
||
})
|
||
|
||
conn.close()
|
||
|
||
return {"history": results, "count": len(results)}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"获取历史失败: {str(e)}")
|
||
|
||
|
||
# ===== 高级RAG功能 =====
|
||
|
||
from src.rag.advanced import (
|
||
MultiGranularChunker,
|
||
HierarchicalRetriever,
|
||
AdaptiveRetriever,
|
||
RetrievalConfig,
|
||
QueryDecomposer,
|
||
ContextCompressor
|
||
)
|
||
|
||
# 高级组件实例
|
||
_advanced_chunker = None
|
||
_advanced_retriever = None
|
||
_query_decomposer = None
|
||
_context_compressor = None
|
||
|
||
|
||
def get_advanced_components():
|
||
"""获取高级组件"""
|
||
global _advanced_chunker, _advanced_retriever, _query_decomposer, _context_compressor
|
||
|
||
if _advanced_chunker is None:
|
||
_advanced_chunker = MultiGranularChunker()
|
||
_query_decomposer = QueryDecomposer()
|
||
_context_compressor = ContextCompressor()
|
||
|
||
# retriever需要vector store，暂不初始化
|
||
_advanced_retriever = None
|
||
|
||
return {
|
||
"chunker": _advanced_chunker,
|
||
"retriever": _advanced_retriever,
|
||
"decomposer": _query_decomposer,
|
||
"compressor": _context_compressor
|
||
}
|
||
|
||
|
||
class AdvancedSearchRequest(BaseModel):
|
||
query: str
|
||
use_hierarchical: bool = False # 使用分层检索
|
||
use_query_decompose: bool = False # 使用查询分解
|
||
use_compression: bool = True # 使用上下文压缩
|
||
query_type: str = "general" # 查询类型
|
||
# Self-RAG + MultiHopRAG 参数
|
||
use_agentic: bool = False # 使用Agent编排
|
||
enable_self_rag: bool = False # 启用Self-RAG反思
|
||
enable_multi_hop: bool = False # 启用MultiHopRAG多跳检索
|
||
max_iterations: int = 3 # Self-RAG最大迭代次数
|
||
|
||
|
||
class QueryDecomposeRequest(BaseModel):
|
||
query: str
|
||
use_llm: bool = False
|
||
|
||
|
||
# Orchestrator 实例缓存
|
||
_orchestrator = None
|
||
_orchestrator_config = None
|
||
|
||
def get_orchestrator(enable_self_rag: bool = False, enable_multi_hop: bool = False, multi_hop_threshold: float = 0.3):
|
||
"""获取Agent编排器实例"""
|
||
global _orchestrator, _orchestrator_config
|
||
|
||
# 检查配置是否变化
|
||
current_config = (enable_self_rag, enable_multi_hop, multi_hop_threshold)
|
||
|
||
if _orchestrator is None or _orchestrator_config != current_config:
|
||
from src.rag.agentic.orchestrator import RAGAgentOrchestrator
|
||
|
||
llm_mgr = get_llm_manager()
|
||
store = get_vector_store()
|
||
|
||
_orchestrator = RAGAgentOrchestrator(
|
||
retriever=store,
|
||
llm_client=llm_mgr,
|
||
enable_self_rag=enable_self_rag,
|
||
enable_multi_hop=enable_multi_hop,
|
||
multi_hop_threshold=multi_hop_threshold
|
||
)
|
||
_orchestrator_config = current_config
|
||
|
||
return _orchestrator
|
||
|
||
|
||
@router.post("/rag/advanced/search")
|
||
async def advanced_search(request: AdvancedSearchRequest):
|
||
"""高级检索接口 - 支持Self-RAG和MultiHopRAG"""
|
||
try:
|
||
comps = get_advanced_components()
|
||
|
||
results = {
|
||
"query": request.query,
|
||
"features": {
|
||
"hierarchical": request.use_hierarchical,
|
||
"query_decompose": request.use_query_decompose,
|
||
"compression": request.use_compression,
|
||
"agentic": request.use_agentic,
|
||
"self_rag": request.enable_self_rag,
|
||
"multi_hop": request.enable_multi_hop
|
||
}
|
||
}
|
||
|
||
# 1. Agent编排模式 (Self-RAG + MultiHopRAG)
|
||
if request.use_agentic or request.enable_self_rag or request.enable_multi_hop:
|
||
orchestrator = get_orchestrator(
|
||
enable_self_rag=request.enable_self_rag,
|
||
enable_multi_hop=request.enable_multi_hop,
|
||
multi_hop_threshold=0.3 # 降低阈值以更容易触发多跳
|
||
)
|
||
|
||
# 执行Agent编排检索
|
||
# 注意: max_iterations 在 orchestrator 初始化时设置
|
||
agent_result = await orchestrator.run(query=request.query)
|
||
|
||
results["agent_result"] = {
|
||
"final_answer": agent_result.get("answer", agent_result.get("final_answer", "")),
|
||
"iteration": agent_result.get("iterations", agent_result.get("iteration", 0)),
|
||
"query_analysis": agent_result.get("query_analysis", {}),
|
||
"evaluation": agent_result.get("evaluation", {}),
|
||
"reflection": agent_result.get("reflection", {}),
|
||
"multi_hop_used": agent_result.get("multi_hop_used", False),
|
||
"multi_hop_hops": agent_result.get("multi_hop_hops", 0),
|
||
"multi_hop_sources": agent_result.get("multi_hop_sources", [])
|
||
}
|
||
|
||
# 添加检索来源
|
||
sources = agent_result.get("results", [])
|
||
results["sources"] = [
|
||
{
|
||
"doc_id": s.get("doc_id", "unknown"),
|
||
"content": s.get("content", "")[:300],
|
||
"score": s.get("score", 0),
|
||
"metadata": s.get("metadata", {})
|
||
}
|
||
for s in sources[:5]
|
||
]
|
||
|
||
return results
|
||
|
||
# 2. 查询分解
|
||
if request.use_query_decompose:
|
||
sub_queries = comps["decomposer"].decompose(request.query)
|
||
results["sub_queries"] = [
|
||
{
|
||
"query": sq.query,
|
||
"reason": sq.reason,
|
||
"expected_docs": sq.expected_docs
|
||
}
|
||
for sq in sub_queries
|
||
]
|
||
|
||
# 3. 分层检索（需要实际索引）
|
||
if request.use_hierarchical:
|
||
results["message"] = "分层检索需要先构建多粒度索引"
|
||
|
||
# 4. 上下文压缩
|
||
if request.use_compression:
|
||
results["compression"] = "已启用上下文压缩"
|
||
|
||
return results
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
raise HTTPException(status_code=500, detail=f"高级检索失败: {str(e)}")
|
||
|
||
|
||
@router.post("/rag/advanced/decompose")
|
||
async def decompose_query(request: QueryDecomposeRequest):
|
||
"""查询分解接口"""
|
||
try:
|
||
comps = get_advanced_components()
|
||
|
||
if request.use_llm:
|
||
sub_queries = await comps["decomposer"].decompose_with_llm(request.query)
|
||
else:
|
||
sub_queries = comps["decomposer"].decompose(request.query)
|
||
|
||
return {
|
||
"original_query": request.query,
|
||
"sub_queries": [
|
||
{
|
||
"query": sq.query,
|
||
"reason": sq.reason,
|
||
"expected_docs": sq.expected_docs,
|
||
"priority": sq.priority
|
||
}
|
||
for sq in sub_queries
|
||
],
|
||
"count": len(sub_queries)
|
||
}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"查询分解失败: {str(e)}")
|
||
|
||
|
||
def _extract_knowledge_graph(sources: List[RAGSource], query: str) -> KnowledgeGraphData:
|
||
"""
|
||
从RAG检索结果动态提取知识图谱
|
||
提取基因名、物种、表型、过程等实体和关系
|
||
"""
|
||
import re
|
||
|
||
nodes = []
|
||
edges = []
|
||
node_ids = set()
|
||
|
||
# 基因名模式 (支持任意物种)
|
||
gene_pattern = r'\b([A-Z][a-z]?[A-Z]{1,4}\d{1,4}[a-z]?)\b'
|
||
|
||
# 物种模式
|
||
species_pattern = r'\b(Arabidopsis thaliana|Oryza sativa|Zea mays|Glycine max|Solanum lycopersicum|Malus domestica|Carya illinoinensis|pecan|Arabidopsis|rice|maize|soybean|tomato|apple)\b'
|
||
|
||
# 过程/功能关键词
|
||
process_keywords = ['development', 'growth', 'regulation', 'signaling', 'biosynthesis',
|
||
'response', 'metabolism', 'expression', 'flowering', 'fruit development']
|
||
|
||
# 组织/器官关键词
|
||
tissue_keywords = ['root', 'shoot', 'leaf', 'flower', 'fruit', 'seed', 'embryo',
|
||
'endosperm', 'pollen', 'ovule', 'meristem', 'bud']
|
||
|
||
# 处理每个source
|
||
for source in sources:
|
||
content = source.content
|
||
doc_id = source.doc_id
|
||
|
||
# 提取基因
|
||
genes = re.findall(gene_pattern, content)
|
||
for gene in genes:
|
||
if gene not in node_ids:
|
||
nodes.append({
|
||
"id": gene,
|
||
"type": "gene"
|
||
})
|
||
node_ids.add(gene)
|
||
|
||
# 提取物种
|
||
species_list = re.findall(species_pattern, content, re.IGNORECASE)
|
||
for sp in species_list:
|
||
sp_key = sp.lower()
|
||
if sp_key not in node_ids:
|
||
nodes.append({
|
||
"id": sp,
|
||
"type": "species"
|
||
})
|
||
node_ids.add(sp_key)
|
||
|
||
# 提取过程
|
||
for proc in process_keywords:
|
||
if proc.lower() in content.lower():
|
||
proc_key = f"{proc}_{doc_id[:8]}"
|
||
if proc_key not in node_ids:
|
||
nodes.append({
|
||
"id": proc,
|
||
"type": "process"
|
||
})
|
||
node_ids.add(proc_key)
|
||
|
||
# 提取组织
|
||
for tissue in tissue_keywords:
|
||
if tissue.lower() in content.lower():
|
||
tissue_key = f"{tissue}_{doc_id[:8]}"
|
||
if tissue_key not in node_ids:
|
||
nodes.append({
|
||
"id": tissue,
|
||
"type": "tissue"
|
||
})
|
||
node_ids.add(tissue_key)
|
||
|
||
# 构建关系：基因-基因 (共现)
|
||
if len(genes) >= 2:
|
||
for i, g1 in enumerate(genes[:5]): # 限制避免过多边
|
||
for g2 in genes[i+1:6]:
|
||
edges.append({
|
||
"source": g1,
|
||
"target": g2,
|
||
"relation": "co_occurrence"
|
||
})
|
||
|
||
# 构建关系：基因-过程
|
||
for gene in genes[:3]:
|
||
for proc in process_keywords:
|
||
if proc.lower() in content.lower():
|
||
edges.append({
|
||
"source": gene,
|
||
"target": proc,
|
||
"relation": "involved_in"
|
||
})
|
||
|
||
# 构建关系：基因-组织 (表达)
|
||
for gene in genes[:3]:
|
||
for tissue in tissue_keywords:
|
||
if tissue.lower() in content.lower():
|
||
edges.append({
|
||
"source": gene,
|
||
"target": tissue,
|
||
"relation": "expressed_in"
|
||
})
|
||
|
||
# 去重边
|
||
unique_edges = []
|
||
edge_keys = set()
|
||
for edge in edges:
|
||
key = f"{edge['source']}->{edge['target']}"
|
||
if key not in edge_keys:
|
||
unique_edges.append(edge)
|
||
edge_keys.add(key)
|
||
|
||
return KnowledgeGraphData(nodes=nodes, edges=unique_edges)
|
||
|
||
|
||
@router.get("/rag/advanced/status")
|
||
async def advanced_status():
|
||
"""高级功能状态"""
|
||
comps = get_advanced_components()
|
||
|
||
return {
|
||
"status": "ready",
|
||
"features": {
|
||
"multi_granular_chunking": True,
|
||
"hierarchical_retrieval": comps["retriever"] is not None,
|
||
"query_decomposition": True,
|
||
"context_compression": True,
|
||
"citation_extraction": True,
|
||
"phylogeny_extraction": True # v5新增
|
||
},
|
||
"chunk_levels": ["summary", "paragraph", "sentence", "keyword"]
|
||
}
|
||
|
||
|
||
# ========== v5 系统发育提取接口 ==========
|
||
|
||
@router.post("/rag/phylogeny/extract")
|
||
async def extract_phylogeny(request: PhylogenyExtractionRequest):
|
||
"""
|
||
系统发育信息提取接口 (v5.2 - TAIR关键词库增强版)
|
||
|
||
自动从PDF提取系统发育分组信息，集成TAIR权威基因家族数据，自动消歧
|
||
包含4轮审查流程和关键词匹配详情
|
||
"""
|
||
try:
|
||
# 导入v5.2提取器 (TAIR关键词库增强版)
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
from extract_phylogeny_v5_universal_v3 import extract_phylogeny_v52
|
||
|
||
print(f"[Phylogeny Extract v5.2] 开始提取: {request.pdf_path}")
|
||
print(f"[Phylogeny Extract v5.2] DOI: {request.doi}")
|
||
print(f"[Phylogeny Extract v5.2] PMCID: {request.pmcid}")
|
||
|
||
# 读取PDF文本
|
||
import fitz
|
||
doc = fitz.open(request.pdf_path)
|
||
text = ""
|
||
for page in doc:
|
||
text += page.get_text()
|
||
doc.close()
|
||
|
||
# 执行提取（v5.2包含TAIR关键词库）
|
||
result = extract_phylogeny_v52(text)
|
||
|
||
# 返回结果（包含审查详情和TAIR关键词匹配）
|
||
ext = result.get("extraction", {})
|
||
return {
|
||
"status": "success",
|
||
"pdf_name": Path(request.pdf_path).name,
|
||
"gene_family": ext.get("gene_family"),
|
||
"species": ext.get("species"),
|
||
"total_genes": ext.get("total_genes"),
|
||
"total_groups": ext.get("total_groups"),
|
||
"confidence_score": result.get("confidence"),
|
||
"confidence_level": result.get("assessment", {}).get("level"),
|
||
"source": "TAIR+v5.2",
|
||
# v5.2新增：TAIR关键词匹配详情
|
||
"matched_keywords": ext.get("matched_keywords", {}),
|
||
# v5.2新增：审查详情
|
||
"scrutiny": result.get("scrutiny", {}),
|
||
"validation": result.get("validation", {}),
|
||
"assessment": result.get("assessment", {}),
|
||
"version": result.get("version", "v5.2-TAIR")
|
||
}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
print(f"[Phylogeny Extract v5.1] 错误: {e}")
|
||
print(traceback.format_exc())
|
||
raise HTTPException(status_code=500, detail=f"提取失败: {str(e)}")
|
||
|
||
|
||
@router.get("/rag/phylogeny/status/{pdf_name}")
|
||
async def get_phylogeny_status(pdf_name: str):
|
||
"""
|
||
查询系统发育提取状态
|
||
"""
|
||
try:
|
||
result_root = Path(
|
||
os.environ.get("PHYLOGENY_AUTO_RESULTS_DIR", "")
|
||
or PROJECT_ROOT / "evaluation" / "auto_results"
|
||
)
|
||
result_path = result_root / f"{pdf_name}_v5_report.json"
|
||
|
||
if not result_path.exists():
|
||
return {"status": "not_found", "message": "提取结果不存在"}
|
||
|
||
with open(result_path) as f:
|
||
result = json.load(f)
|
||
|
||
return {
|
||
"status": "success",
|
||
"pdf_name": pdf_name,
|
||
"confidence_score": result.get("confidence_score"),
|
||
"confidence_level": result.get("confidence_level"),
|
||
"total_groups": result.get("total_groups"),
|
||
"has_supplementary": result.get("has_supplementary")
|
||
}
|
||
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}")
|