936 lines
35 KiB
Python
Executable File
936 lines
35 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""Hybrid (Qdrant + Neo4j) QA routes for the V15 SRT knowledge base.
|
||
|
||
This route uses V15HybridRetriever for vector (Qdrant) and graph (Neo4j)
|
||
retrieval. The old SQLite-only path is preserved in the .bak file.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import asyncio
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
from pydantic import BaseModel, Field
|
||
from src.rag.v15_hybrid_retriever import V15HybridRetriever
|
||
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
DEFAULT_KB_DIR = PROJECT_ROOT / "data" / "knowledge_base" / "v15_srt_kb_20260525"
|
||
|
||
router = APIRouter(prefix="/kb-qa", tags=["V15 SRT KB QA"])
|
||
_llm_manager = None
|
||
_result_chunk_cache: Dict[str, List[Dict[str, str]]] = {}
|
||
|
||
|
||
STOPWORDS = {
|
||
"what",
|
||
"which",
|
||
"where",
|
||
"when",
|
||
"who",
|
||
"how",
|
||
"does",
|
||
"did",
|
||
"are",
|
||
"was",
|
||
"were",
|
||
"the",
|
||
"and",
|
||
"for",
|
||
"with",
|
||
"from",
|
||
"into",
|
||
"about",
|
||
"that",
|
||
"this",
|
||
"these",
|
||
"those",
|
||
"gene",
|
||
"genes",
|
||
"protein",
|
||
"proteins",
|
||
"family",
|
||
"tell",
|
||
"me",
|
||
"is",
|
||
"of",
|
||
"in",
|
||
"to",
|
||
"a",
|
||
"an",
|
||
}
|
||
|
||
LAYER_FILTERS = {
|
||
"current_findings": (
|
||
"r.output_layer = 'main_kg' "
|
||
"AND COALESCE(r.claim_scope, '') NOT IN ('background_context', 'cited_prior_work') "
|
||
"AND COALESCE(r.candidate_status, '') != 'candidate'"
|
||
),
|
||
"main_kg": (
|
||
"r.output_layer = 'main_kg' "
|
||
"AND COALESCE(r.claim_scope, '') NOT IN ('background_context', 'cited_prior_work') "
|
||
"AND COALESCE(r.candidate_status, '') != 'candidate'"
|
||
),
|
||
"methods": "r.output_layer = 'method_relations' AND COALESCE(r.candidate_status, '') != 'candidate'",
|
||
"method_relations": "r.output_layer = 'method_relations' AND COALESCE(r.candidate_status, '') != 'candidate'",
|
||
"datasets": "r.output_layer = 'dataset_metadata' AND COALESCE(r.candidate_status, '') != 'candidate'",
|
||
"dataset_metadata": "r.output_layer = 'dataset_metadata' AND COALESCE(r.candidate_status, '') != 'candidate'",
|
||
"prior_work": "r.output_layer = 'background_relations'",
|
||
"background_relations": "r.output_layer = 'background_relations'",
|
||
"candidate": "(r.output_layer = 'candidate_review' OR r.candidate_status = 'candidate')",
|
||
"candidate_review": "(r.output_layer = 'candidate_review' OR r.candidate_status = 'candidate')",
|
||
"all": "1 = 1",
|
||
}
|
||
|
||
RELATION_FAMILY_ALIASES = {
|
||
"baseline_expression": [
|
||
"expressed_in",
|
||
"highly_expressed_during",
|
||
"only_expressed_during",
|
||
"differentially_expressed_under",
|
||
"upregulated_in",
|
||
"downregulated_in",
|
||
"downregulated_during",
|
||
"peak_at",
|
||
"not_induced_under",
|
||
],
|
||
"up_expression": ["upregulated_in", "induced_in", "increased_in"],
|
||
"down_expression": ["downregulated_in", "downregulated_during", "repressed_in", "decreased_in"],
|
||
"differential_expression": ["differentially_expressed_under"],
|
||
"expression_peak": ["peak_at", "reached_maximum_at"],
|
||
"not_induced": ["not_induced_under"],
|
||
"identification": ["identified_in", "identified_from"],
|
||
"member_count": ["has_member_count", "contains_gene_family_members"],
|
||
"classification": ["member_of", "classified_into", "divided_into", "separated_into", "categorized_into"],
|
||
"domain": ["contains_domain", "has_domain"],
|
||
"cis_element": ["contains_cis_element", "has_cis_element", "has_cis_elements"],
|
||
"motif": ["contains_motif"],
|
||
"localization": ["localizes_to"],
|
||
"interaction": ["interacts_with"],
|
||
"binding": ["binds_to"],
|
||
"targeting": ["targets"],
|
||
"homology": ["clustered_with", "homologous_to"],
|
||
"synteny": ["syntenic_with", "has_collinear_pairs_with", "collinear_with"],
|
||
"regulation": ["regulates", "positively_regulates", "negatively_regulates"],
|
||
"effect": ["promotes", "enhances", "overexpression_enhances", "overexpression_increases", "overexpression_improves"],
|
||
"function": ["plays_role_in", "play_roles_in", "involved_in", "participates_in"],
|
||
"identity": ["is_a"],
|
||
}
|
||
|
||
|
||
class KBQARequest(BaseModel):
|
||
question: str = Field(..., description="Question to answer from the V15 SRT KB")
|
||
top_k: int = Field(8, ge=1, le=50)
|
||
layer: str = Field(
|
||
"auto",
|
||
description=(
|
||
"auto/current_findings/methods/datasets/prior_work/"
|
||
"candidate/all"
|
||
),
|
||
)
|
||
include_evidence: bool = True
|
||
include_candidate_fallback: bool = False
|
||
use_llm: bool = True
|
||
llm_provider: Optional[str] = None
|
||
llm_model: Optional[str] = None
|
||
anchor_entity: Optional[str] = Field(
|
||
None,
|
||
description="Optional structured retrieval hint, e.g. GmP450 genes",
|
||
)
|
||
relation_family: Optional[str] = Field(
|
||
None,
|
||
description="Optional question/relation family hint, e.g. baseline_expression",
|
||
)
|
||
preferred_relations: List[str] = Field(
|
||
default_factory=list,
|
||
description="Optional exact relation hints for structured retrieval",
|
||
)
|
||
preferred_papers: List[str] = Field(
|
||
default_factory=list,
|
||
description="Optional paper_id hints for benchmark/debug QA; used as boost, not as a hard filter",
|
||
)
|
||
|
||
|
||
class KBQASource(BaseModel):
|
||
relation_id: str
|
||
paper_id: str = ""
|
||
pubmed_id: str = ""
|
||
p_source: str = ""
|
||
source_result_path: str = ""
|
||
chunk_id: str = ""
|
||
chunk_hint: str = ""
|
||
evidence_key: str = ""
|
||
source: str
|
||
source_type: str = ""
|
||
relation: str
|
||
target: str
|
||
target_type: str = ""
|
||
evidence: str = ""
|
||
section: str = ""
|
||
source_label: str = ""
|
||
claim_scope: str = ""
|
||
output_layer: str = ""
|
||
candidate_status: str = ""
|
||
confidence: str = ""
|
||
score: float = 0.0
|
||
|
||
|
||
class KBQAResponse(BaseModel):
|
||
question: str
|
||
answer: str
|
||
selected_layer: str
|
||
layer_reason: str
|
||
source_count: int
|
||
paper_source_count: int = 0
|
||
sources: List[KBQASource]
|
||
kb_path: str
|
||
answer_mode: str = "extractive"
|
||
llm_used: bool = False
|
||
notes: List[str] = []
|
||
|
||
|
||
def _kb_dir() -> Path:
|
||
configured = os.environ.get("V15_SRT_KB_DIR")
|
||
return Path(configured).expanduser() if configured else DEFAULT_KB_DIR
|
||
|
||
|
||
def get_llm_manager():
|
||
"""Lazy-load the existing project LLM manager for answer synthesis."""
|
||
global _llm_manager
|
||
if _llm_manager is not None:
|
||
return _llm_manager
|
||
try:
|
||
import yaml
|
||
from src.llm.manager import LLMManager
|
||
|
||
config_path = PROJECT_ROOT / "config" / "llm.yaml"
|
||
if config_path.exists():
|
||
config = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
|
||
else:
|
||
config = {"providers": {}, "default": {}}
|
||
|
||
# KB-QA must be able to use DeepSeek even when the shared llm.yaml was
|
||
# created before the DeepSeek provider was added to the project.
|
||
config.setdefault("providers", {}).setdefault(
|
||
"deepseek",
|
||
{
|
||
"name": "DeepSeek",
|
||
"api_base": os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
|
||
"models": [
|
||
{"name": "deepseek-v4-flash", "context_length": 128000},
|
||
{"name": "deepseek-v4-pro", "context_length": 128000},
|
||
{"name": "deepseek-chat", "context_length": 64000},
|
||
],
|
||
"env_key": "DEEPSEEK_API_KEY",
|
||
},
|
||
)
|
||
|
||
provider = os.environ.get("V15_KB_QA_LLM_PROVIDER")
|
||
model = os.environ.get("V15_KB_QA_LLM_MODEL")
|
||
if provider:
|
||
config.setdefault("default", {})["provider"] = provider
|
||
if model:
|
||
config.setdefault("default", {})["model"] = model
|
||
if os.environ.get("V15_KB_QA_LLM_MAX_TOKENS"):
|
||
config.setdefault("default", {})["max_tokens"] = int(os.environ["V15_KB_QA_LLM_MAX_TOKENS"])
|
||
if os.environ.get("V15_KB_QA_LLM_TEMPERATURE"):
|
||
config.setdefault("default", {})["temperature"] = float(os.environ["V15_KB_QA_LLM_TEMPERATURE"])
|
||
|
||
_llm_manager = LLMManager(config)
|
||
return _llm_manager
|
||
except Exception as exc:
|
||
print(f"[KB-QA] LLM manager init failed: {exc}")
|
||
return None
|
||
|
||
|
||
def _classify_layer(question: str, requested_layer: str) -> Tuple[str, str]:
|
||
requested = (requested_layer or "auto").strip().lower()
|
||
if requested != "auto":
|
||
if requested not in LAYER_FILTERS:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"Unsupported layer '{requested_layer}'. "
|
||
f"Use one of: auto, {', '.join(sorted(LAYER_FILTERS))}",
|
||
)
|
||
return requested, "user_selected"
|
||
|
||
q = question.lower()
|
||
if re.search(r"\b(tool|software|method|pipeline|database|using|used|analy[sz]ed by)\b", q) or any(
|
||
word in question for word in ["工具", "软件", "方法", "流程", "用了什么", "使用了什么"]
|
||
):
|
||
return "methods", "auto_detected_method_question"
|
||
if re.search(r"\b(accession|geo|sra|prjna|srp|gse|rna-seq|rnaseq|doi)\b", q) or any(
|
||
word in question for word in ["编号", "登录号", "测序数据", "转录组数据"]
|
||
):
|
||
return "datasets", "auto_detected_dataset_question"
|
||
if re.search(r"\b(prior|previous|background|reported|known|literature|cited)\b", q) or any(
|
||
word in question for word in ["前人", "背景", "已有研究", "文献提到", "报道过"]
|
||
):
|
||
return "prior_work", "auto_detected_prior_work_question"
|
||
if any(word in question for word in ["候选", "不确定", "可能相关"]):
|
||
return "candidate", "auto_detected_candidate_question"
|
||
return "current_findings", "default_current_findings"
|
||
|
||
|
||
def _query_tokens(question: str) -> List[str]:
|
||
raw_tokens = re.findall(r"[A-Za-z][A-Za-z0-9_.:/+-]*|\d+[A-Za-z0-9_.:/+-]*", question)
|
||
tokens: List[str] = []
|
||
for token in raw_tokens:
|
||
cleaned = re.sub(r"[^A-Za-z0-9_]+", " ", token).strip()
|
||
for part in cleaned.split():
|
||
lower = part.lower()
|
||
if len(part) < 2 or lower in STOPWORDS:
|
||
continue
|
||
tokens.append(part)
|
||
# Preserve order while removing duplicates.
|
||
seen = set()
|
||
unique = []
|
||
for token in tokens:
|
||
key = token.lower()
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique.append(token)
|
||
return unique[:12]
|
||
|
||
|
||
def _focus_entity(question: str) -> str:
|
||
"""Extract a focus entity from short definition-style questions."""
|
||
text = question.strip()
|
||
patterns = [
|
||
r"^\s*what\s+(?:is|are)\s+(.+?)[?？。.!]*\s*$",
|
||
r"^\s*define\s+(.+?)[?？。.!]*\s*$",
|
||
r"^\s*(.+?)\s*(?:是什么|是啥|指什么)[?？。.!]*\s*$",
|
||
]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, text, flags=re.IGNORECASE)
|
||
if not match:
|
||
continue
|
||
focus = match.group(1).strip()
|
||
focus = re.sub(r"^(the|a|an)\s+", "", focus, flags=re.IGNORECASE)
|
||
focus = focus.strip(" \t\r\n\"'`.,;:()[]{}")
|
||
if 2 <= len(focus) <= 80:
|
||
return focus
|
||
return ""
|
||
|
||
|
||
def _clean_hint_text(value: str) -> str:
|
||
value = re.sub(r"\s+", " ", value or "").strip()
|
||
value = value.strip(" \t\r\n\"'`.,;:()[]{}?？。")
|
||
return value[:120]
|
||
|
||
|
||
def _infer_anchor_entity(question: str) -> str:
|
||
"""Infer a likely SRT source entity from a natural-language question."""
|
||
patterns = [
|
||
r"问题[:：]\s*(.+?)\s*(?:的|之)\s*表达",
|
||
r"问题[:：]\s*(.+?)\s*(?:的|之)\s*鉴定",
|
||
r"问题[:：]\s*(.+?)\s*(?:报道了多少|有多少)",
|
||
r"expression of\s+(.+?)\s+(?:was|were|reported|under|in|\\?)",
|
||
r"identifying\s+(.+?)(?:\\?|$)",
|
||
r"identification of\s+(.+?)(?:\\?|$)",
|
||
r"reported for\s+(.+?)(?:\\?|$)",
|
||
r"reported in\s+(.+?)(?:\\?|$)",
|
||
r"reported to interact with\s+(.+?)(?:\\?|$)",
|
||
r"was\s+(.+?)\s+upregulated",
|
||
r"was\s+(.+?)\s+downregulated",
|
||
]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, question, flags=re.IGNORECASE)
|
||
if not match:
|
||
continue
|
||
candidate = _clean_hint_text(match.group(1))
|
||
candidate = re.sub(r"^(the|a|an)\s+", "", candidate, flags=re.IGNORECASE)
|
||
candidate = re.split(r"\n|English question:|回答要求[:：]", candidate)[0].strip()
|
||
if 2 <= len(candidate) <= 100 and not re.search(r"\b(where|which|what|how|under)\b", candidate, re.I):
|
||
return candidate
|
||
focus = _focus_entity(question)
|
||
return focus
|
||
|
||
|
||
def _entity_variants(entity: str) -> List[str]:
|
||
entity = _clean_hint_text(entity)
|
||
if not entity:
|
||
return []
|
||
variants = {entity}
|
||
base = re.sub(
|
||
r"\s+(gene|genes|protein|proteins|transcripts?|family|gene family|protein family)$",
|
||
"",
|
||
entity,
|
||
flags=re.IGNORECASE,
|
||
).strip()
|
||
if base and len(base) >= 2:
|
||
variants.add(base)
|
||
if re.search(r"[A-Za-z0-9]$", base):
|
||
variants.add(base + "s")
|
||
variants.add(base + " genes")
|
||
variants.add(base + " proteins")
|
||
if entity.endswith("s") and len(entity) > 3:
|
||
variants.add(entity[:-1])
|
||
return [v for v in variants if len(v) >= 2][:8]
|
||
|
||
|
||
def _infer_relation_family(question: str) -> str:
|
||
q = question.lower()
|
||
checks = [
|
||
("baseline_expression", [r"表达", r"expression", r"expressed", r"tissue", r"condition"]),
|
||
("up_expression", [r"上调", r"upregulated", r"induced"]),
|
||
("down_expression", [r"下调", r"downregulated", r"repressed"]),
|
||
("not_induced", [r"未被诱导", r"not induced"]),
|
||
("identification", [r"鉴定", r"identification", r"identifying", r"identified"]),
|
||
("member_count", [r"多少个成员", r"how many members", r"member count"]),
|
||
("cis_element", [r"顺式", r"cis[- ]?acting", r"promoter", r"启动子"]),
|
||
("domain", [r"结构域", r"domain"]),
|
||
("motif", [r"motif"]),
|
||
("localization", [r"定位", r"localiz"]),
|
||
("interaction", [r"互作", r"interact"]),
|
||
("binding", [r"结合", r"bind"]),
|
||
("synteny", [r"共线", r"synten", r"collinear"]),
|
||
("regulation", [r"调控", r"regulat"]),
|
||
("function", [r"功能", r"function", r"process", r"involved", r"role"]),
|
||
("classification", [r"亚家族", r"类别", r"class", r"group", r"subfamily", r"belong"]),
|
||
]
|
||
for family, patterns in checks:
|
||
if any(re.search(pattern, q) for pattern in patterns):
|
||
return family
|
||
return ""
|
||
|
||
|
||
def _resolve_preferred_relations(relation_family: str, preferred_relations: List[str]) -> List[str]:
|
||
resolved: List[str] = []
|
||
for rel in preferred_relations or []:
|
||
rel = _clean_hint_text(rel)
|
||
if rel and re.fullmatch(r"[A-Za-z0-9_]+", rel):
|
||
resolved.append(rel)
|
||
family_key = (relation_family or "").strip()
|
||
resolved.extend(RELATION_FAMILY_ALIASES.get(family_key, []))
|
||
seen = set()
|
||
unique = []
|
||
for rel in resolved:
|
||
key = rel.lower()
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique.append(rel)
|
||
return unique[:20]
|
||
|
||
|
||
def _build_query_plan(
|
||
question: str,
|
||
anchor_entity: Optional[str] = None,
|
||
relation_family: Optional[str] = None,
|
||
preferred_relations: Optional[List[str]] = None,
|
||
preferred_papers: Optional[List[str]] = None,
|
||
) -> Dict[str, Any]:
|
||
anchor = _clean_hint_text(anchor_entity or "") or _infer_anchor_entity(question)
|
||
family = _clean_hint_text(relation_family or "") or _infer_relation_family(question)
|
||
relations = _resolve_preferred_relations(family, preferred_relations or [])
|
||
papers = [_clean_hint_text(paper) for paper in (preferred_papers or []) if _clean_hint_text(paper)]
|
||
return {
|
||
"anchor_entity": anchor,
|
||
"entity_variants": _entity_variants(anchor),
|
||
"relation_family": family,
|
||
"preferred_relations": relations,
|
||
"preferred_papers": papers[:20],
|
||
}
|
||
|
||
|
||
def _norm_space(text: str) -> str:
|
||
return re.sub(r"\s+", " ", text or "").strip()
|
||
|
||
|
||
def _evidence_key(text: str) -> str:
|
||
text = _norm_space(text).lower()
|
||
text = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", " ", text)
|
||
return _norm_space(text)
|
||
|
||
|
||
def _paper_group_key(row: Dict[str, Any]) -> str:
|
||
return (
|
||
row.get("pubmed_id")
|
||
or row.get("paper_id")
|
||
or row.get("source_result_path")
|
||
or row.get("relation_id")
|
||
or ""
|
||
)
|
||
|
||
|
||
def _resolve_result_path(source_result_path: str) -> Optional[Path]:
|
||
if not source_result_path:
|
||
return None
|
||
path = Path(source_result_path)
|
||
if not path.is_absolute():
|
||
path = PROJECT_ROOT / path
|
||
if path.is_dir():
|
||
path = path / "result.json"
|
||
if path.name != "result.json" and path.suffix != ".json":
|
||
candidate = path / "result.json"
|
||
if candidate.exists():
|
||
path = candidate
|
||
return path if path.exists() else None
|
||
|
||
|
||
def _load_result_chunks(source_result_path: str) -> List[Dict[str, str]]:
|
||
path = _resolve_result_path(source_result_path)
|
||
if path is None:
|
||
return []
|
||
key = str(path)
|
||
if key in _result_chunk_cache:
|
||
return _result_chunk_cache[key]
|
||
chunks: List[Dict[str, str]] = []
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
for idx, chunk in enumerate(data.get("chunks") or []):
|
||
text = _norm_space(str(chunk.get("text") or chunk.get("content") or ""))
|
||
if not text:
|
||
continue
|
||
chunk_id = str(chunk.get("chunk_id") or chunk.get("id") or f"chunk_{idx:04d}")
|
||
chunks.append(
|
||
{
|
||
"chunk_id": chunk_id,
|
||
"section": str(chunk.get("section") or chunk.get("section_title") or ""),
|
||
"text": text,
|
||
"norm_text": _evidence_key(text),
|
||
}
|
||
)
|
||
except Exception:
|
||
chunks = []
|
||
_result_chunk_cache[key] = chunks
|
||
return chunks
|
||
|
||
|
||
def _infer_chunk_id(row: Dict[str, Any]) -> str:
|
||
evidence_norm = _evidence_key(row.get("evidence", ""))
|
||
if not evidence_norm:
|
||
return ""
|
||
chunks = _load_result_chunks(row.get("source_result_path", ""))
|
||
if not chunks:
|
||
return ""
|
||
|
||
anchors = [evidence_norm]
|
||
if len(evidence_norm) > 260:
|
||
anchors.extend([evidence_norm[:260], evidence_norm[:160], evidence_norm[-200:]])
|
||
for anchor in anchors:
|
||
if len(anchor) < 40:
|
||
continue
|
||
for chunk in chunks:
|
||
if anchor in chunk.get("norm_text", ""):
|
||
return chunk.get("chunk_id", "")
|
||
return ""
|
||
|
||
|
||
def _fallback_chunk_hint(row: Dict[str, Any]) -> str:
|
||
parts = []
|
||
if row.get("section"):
|
||
parts.append(f"section={row.get('section')}")
|
||
if row.get("p_source"):
|
||
parts.append(f"p_source={row.get('p_source')}")
|
||
if row.get("source_label"):
|
||
parts.append(f"source_label={row.get('source_label')}")
|
||
if not parts and row.get("relation_id"):
|
||
parts.append(f"relation={row.get('relation_id')}")
|
||
return " | ".join(parts)
|
||
|
||
|
||
def _decorate_source_rows(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
for row in rows:
|
||
row["evidence_key"] = _evidence_key(row.get("evidence", ""))
|
||
chunk_id = _infer_chunk_id(row)
|
||
row["chunk_id"] = chunk_id
|
||
row["chunk_hint"] = chunk_id or _fallback_chunk_hint(row)
|
||
return rows
|
||
|
||
|
||
def _shorten(text: str, limit: int = 420) -> str:
|
||
text = re.sub(r"\s+", " ", text or "").strip()
|
||
if len(text) <= limit:
|
||
return text
|
||
return text[: limit - 3].rstrip() + "..."
|
||
|
||
|
||
def _format_answer(question: str, layer: str, rows: List[Dict[str, Any]], include_evidence: bool) -> str:
|
||
if not rows:
|
||
return (
|
||
"未在当前知识库层中检索到可用证据。可以尝试切换 layer=all、"
|
||
"layer=prior_work、layer=methods 或 layer=datasets。"
|
||
)
|
||
|
||
layer_label = {
|
||
"current_findings": "本文主要发现(main_kg)",
|
||
"main_kg": "本文主要发现(main_kg)",
|
||
"methods": "方法/工具(method_relations)",
|
||
"method_relations": "方法/工具(method_relations)",
|
||
"datasets": "数据集(dataset_metadata)",
|
||
"dataset_metadata": "数据集(dataset_metadata)",
|
||
"prior_work": "前人研究/背景(background_relations)",
|
||
"background_relations": "前人研究/背景(background_relations)",
|
||
"candidate": "候选关系(candidate_review)",
|
||
"candidate_review": "候选关系(candidate_review)",
|
||
"all": "全部非拒绝关系",
|
||
}.get(layer, layer)
|
||
|
||
paper_count = len({_paper_group_key(row) for row in rows})
|
||
lines = [
|
||
f"基于 {layer_label} 检索到 {paper_count} 篇文献中的 {len(rows)} 条去重证据："
|
||
]
|
||
for idx, row in enumerate(rows[:8], start=1):
|
||
pmid = row.get("pubmed_id") or "NA"
|
||
chunk = row.get("chunk_id") or row.get("chunk_hint") or row.get("section") or "NA"
|
||
lines.append(
|
||
f"{idx}. PMID: {pmid} | paper: {row.get('paper_id', '') or 'NA'} | chunk: {chunk}"
|
||
)
|
||
if include_evidence and row.get("evidence"):
|
||
lines.append(f" 证据：{_shorten(row['evidence'])}")
|
||
lines.append("回答只基于当前 V15 SRT 知识库；需要原文级核查时请查看返回的 sources。")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _build_llm_context(rows: List[Dict[str, Any]], max_rows: Optional[int] = None) -> str:
|
||
if max_rows is None:
|
||
max_rows = max(int(os.environ.get("V15_KB_QA_LLM_CONTEXT_EVIDENCE_ROWS", "18")), 1)
|
||
parts = []
|
||
for idx, row in enumerate(rows[:max_rows], start=1):
|
||
evidence = re.sub(r"\s+", " ", row.get("evidence", "") or "").strip()
|
||
if len(evidence) > 900:
|
||
evidence = evidence[:897].rstrip() + "..."
|
||
parts.append(
|
||
"\n".join(
|
||
[
|
||
f"[S{idx}]",
|
||
f"PMID: {row.get('pubmed_id', '')}",
|
||
f"paper_id: {row.get('paper_id', '')}",
|
||
f"chunk: {row.get('chunk_id') or row.get('chunk_hint') or row.get('section', '')}",
|
||
f"layer: {row.get('output_layer', '')}",
|
||
f"claim_scope: {row.get('claim_scope', '')}",
|
||
f"SRT: {row.get('source', '')} --{row.get('relation', '')}--> {row.get('target', '')}",
|
||
f"source_type: {row.get('source_type', '')}; target_type: {row.get('target_type', '')}",
|
||
f"evidence: {evidence}",
|
||
]
|
||
)
|
||
)
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
def _build_answer_prompt(question: str, layer: str, rows: List[Dict[str, Any]]) -> str:
|
||
layer_instruction = {
|
||
"current_findings": "These are current-paper biological findings.",
|
||
"main_kg": "These are current-paper biological findings.",
|
||
"methods": "These are method/tool/software relations. Answer method questions only.",
|
||
"method_relations": "These are method/tool/software relations. Answer method questions only.",
|
||
"datasets": "These are dataset/accession/provenance relations.",
|
||
"dataset_metadata": "These are dataset/accession/provenance relations.",
|
||
"prior_work": "These are prior-work or background relations; explicitly say they are prior/background if used.",
|
||
"background_relations": "These are prior-work or background relations; explicitly say they are prior/background if used.",
|
||
"candidate": "These are candidate/uncertain relations; explicitly warn that they are candidates.",
|
||
"candidate_review": "These are candidate/uncertain relations; explicitly warn that they are candidates.",
|
||
"all": "These relations may come from different layers. Do not mix layer meanings.",
|
||
}.get(layer, "Use the layer metadata to qualify the answer.")
|
||
|
||
return f"""You are answering from an audited plant gene-family SRT knowledge base.
|
||
|
||
Question:
|
||
{question}
|
||
|
||
Selected layer:
|
||
{layer}
|
||
|
||
Layer instruction:
|
||
{layer_instruction}
|
||
|
||
Retrieved evidence records:
|
||
{_build_llm_context(rows)}
|
||
|
||
Rules:
|
||
1. Answer only from the retrieved SRT evidence above. Do not add outside knowledge.
|
||
2. If the retrieved SRTs do not answer the question, say so clearly.
|
||
3. Cite supporting rows with [S1], [S2], etc.
|
||
4. Keep current findings, prior/background work, method/tool facts, datasets, and candidate relations conceptually separate.
|
||
5. If evidence is limited or conflicting, say "the KB evidence is limited" and explain the limitation.
|
||
6. Answer in the same language as the user.s question. If the question contains non-English characters (e.g. Chinese), respond in that language primarily. If the question is purely English, respond in English.
|
||
7. Be concise but conversational; do not just repeat a raw SRT table.
|
||
|
||
Answer:
|
||
"""
|
||
|
||
|
||
async def _generate_llm_answer(
|
||
question: str,
|
||
layer: str,
|
||
rows: List[Dict[str, Any]],
|
||
provider: Optional[str] = None,
|
||
model: Optional[str] = None,
|
||
) -> str:
|
||
if not rows:
|
||
return ""
|
||
|
||
llm_mgr = get_llm_manager()
|
||
if llm_mgr is None:
|
||
raise RuntimeError("LLM manager is not available")
|
||
|
||
prompt = _build_answer_prompt(question, layer, rows)
|
||
|
||
from src.llm.base import LLMMessage
|
||
|
||
provider = provider or os.environ.get("V15_KB_QA_LLM_PROVIDER")
|
||
model = model or os.environ.get("V15_KB_QA_LLM_MODEL")
|
||
max_tokens = int(os.environ.get("V15_KB_QA_LLM_MAX_TOKENS", "1200"))
|
||
temperature = float(os.environ.get("V15_KB_QA_LLM_TEMPERATURE", "0.2"))
|
||
response = await llm_mgr.acall(
|
||
[LLMMessage(role="user", content=prompt)],
|
||
provider=provider,
|
||
model=model,
|
||
max_tokens=max_tokens,
|
||
temperature=temperature,
|
||
)
|
||
return response.content if hasattr(response, "content") else str(response)
|
||
|
||
|
||
def _strip_think_tags(text: str) -> str:
|
||
return re.sub(r"<think>.*?</think>", "", text or "", flags=re.DOTALL).strip()
|
||
|
||
|
||
def _ollama_chat_sync(prompt: str, model: str) -> str:
|
||
raise RuntimeError("Local Ollama is disabled for KB-QA answer generation")
|
||
host = os.environ.get("OLLAMA_HOST", "http://localhost:11434").rstrip("/")
|
||
payload = {
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"stream": False,
|
||
"temperature": float(os.environ.get("V15_KB_QA_LLM_TEMPERATURE", "0.2")),
|
||
"options": {"num_predict": int(os.environ.get("V15_KB_QA_LLM_MAX_TOKENS", "1200"))},
|
||
}
|
||
req = urllib.request.Request(
|
||
f"{host}/api/chat",
|
||
data=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req, timeout=180) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
return _strip_think_tags(data.get("message", {}).get("content", ""))
|
||
|
||
|
||
async def _generate_llm_answer_with_fallback(
|
||
question: str,
|
||
layer: str,
|
||
rows: List[Dict[str, Any]],
|
||
provider: Optional[str] = None,
|
||
model: Optional[str] = None,
|
||
) -> Tuple[str, List[str], str]:
|
||
"""Generate an answer with the configured remote LLM only.
|
||
|
||
Local Ollama fallback is intentionally disabled for QA evaluation. If the
|
||
configured remote/API model fails, the caller will return an extractive
|
||
answer instead of retrying with a local small model.
|
||
"""
|
||
notes: List[str] = []
|
||
provider = provider or os.environ.get("V15_KB_QA_LLM_PROVIDER")
|
||
model = model or os.environ.get("V15_KB_QA_LLM_MODEL")
|
||
|
||
if (provider or "").lower() == "ollama":
|
||
raise RuntimeError("Local Ollama provider is disabled for KB-QA answer generation")
|
||
|
||
try:
|
||
answer = await _generate_llm_answer(question, layer, rows, provider=provider, model=model)
|
||
return answer, notes, f"{provider or 'default'}:{model or 'default'}"
|
||
except Exception as exc:
|
||
notes.append(f"Primary LLM failed: {exc}")
|
||
raise
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hybrid → KBQASource adapter
|
||
# ---------------------------------------------------------------------------
|
||
def _hybrid_row_to_source(row: dict) -> KBQASource:
|
||
"""Map a V15HybridRetriever result row to KBQASource, filling defaults."""
|
||
return KBQASource(
|
||
relation_id=row.get("relation_id", "") or "graph_{}_{}_{}".format(
|
||
row.get("source", ""), row.get("relation", ""), row.get("target", "")
|
||
),
|
||
paper_id=row.get("paper_id", ""),
|
||
pubmed_id=str(row.get("pubmed_id", "") or ""),
|
||
p_source=row.get("p_source", ""),
|
||
source_result_path=row.get("source_result_path", ""),
|
||
chunk_id=row.get("chunk_id", ""),
|
||
chunk_hint=row.get("chunk_hint", ""),
|
||
evidence_key=row.get("evidence_key", ""),
|
||
source=row.get("source", ""),
|
||
source_type=row.get("source_type", ""),
|
||
relation=row.get("relation", ""),
|
||
target=row.get("target", ""),
|
||
target_type=row.get("target_type", ""),
|
||
evidence=row.get("evidence", ""),
|
||
section=row.get("section", ""),
|
||
source_label=row.get("source_label", ""),
|
||
claim_scope=row.get("claim_scope", ""),
|
||
output_layer=row.get("output_layer", ""),
|
||
candidate_status=row.get("candidate_status", ""),
|
||
confidence=str(row.get("confidence", "")) if row.get("confidence") is not None else "",
|
||
score=float(row.get("_vector_score", 0.0) or 0.0),
|
||
)
|
||
|
||
|
||
def _source_model(row: Dict[str, Any]) -> KBQASource:
|
||
return KBQASource(
|
||
relation_id=row.get("relation_id", ""),
|
||
paper_id=row.get("paper_id", ""),
|
||
pubmed_id=row.get("pubmed_id", ""),
|
||
p_source=row.get("p_source", ""),
|
||
source_result_path=row.get("source_result_path", ""),
|
||
chunk_id=row.get("chunk_id", ""),
|
||
chunk_hint=row.get("chunk_hint", ""),
|
||
evidence_key=row.get("evidence_key", ""),
|
||
source=row.get("source", ""),
|
||
source_type=row.get("source_type", ""),
|
||
relation=row.get("relation", ""),
|
||
target=row.get("target", ""),
|
||
target_type=row.get("target_type", ""),
|
||
evidence=row.get("evidence", ""),
|
||
section=row.get("section", ""),
|
||
source_label=row.get("source_label", ""),
|
||
claim_scope=row.get("claim_scope", ""),
|
||
output_layer=row.get("output_layer", ""),
|
||
candidate_status=row.get("candidate_status", ""),
|
||
confidence=row.get("confidence", ""),
|
||
score=float(row.get("score", 0.0) or 0.0),
|
||
)
|
||
|
||
|
||
@router.get("/info")
|
||
async def kb_info() -> Dict[str, Any]:
|
||
"""Return hybrid (Qdrant + Neo4j) KB status."""
|
||
return {
|
||
"backend": "hybrid",
|
||
"qdrant_collection": os.environ.get("V15_QDRANT_COLLECTION", "v15_srt_bge_20260527"),
|
||
"neo4j_uri": os.environ.get("NEO4J_URI", "bolt://localhost:7687"),
|
||
"recommended_default": "Use /api/v1/kb-qa/chat with layer=auto for QA.",
|
||
"note": "SQLite-only path is preserved in kb_qa_routes.py.bak.20260528",
|
||
}
|
||
|
||
|
||
@router.post("/chat", response_model=KBQAResponse)
|
||
async def kb_qa_chat(request: KBQARequest) -> KBQAResponse:
|
||
"""Retrieve SRT evidence via hybrid (Qdrant + Neo4j) retrieval and format an auditable answer."""
|
||
question = request.question.strip()
|
||
if not question:
|
||
raise HTTPException(status_code=400, detail="question is required")
|
||
|
||
selected_layer, reason = _classify_layer(question, request.layer)
|
||
answer_layer = selected_layer
|
||
query_plan = _build_query_plan(
|
||
question,
|
||
anchor_entity=request.anchor_entity,
|
||
relation_family=request.relation_family,
|
||
preferred_relations=request.preferred_relations,
|
||
preferred_papers=request.preferred_papers,
|
||
)
|
||
|
||
# Hybrid retrieval via Qdrant (vector) + Neo4j (graph), no SQLite
|
||
hybrid = V15HybridRetriever()
|
||
anchor = query_plan.get("anchor_entity", "")
|
||
try:
|
||
result = hybrid.retrieve(
|
||
question,
|
||
top_k=request.top_k,
|
||
layer=selected_layer,
|
||
use_sqlite=False,
|
||
use_vector=True,
|
||
use_graph=True,
|
||
include_candidate=request.include_candidate_fallback,
|
||
anchor_entity=anchor or None,
|
||
)
|
||
raw_rows = result["retrieved_evidence"]
|
||
notes = list(result["notes"])
|
||
graph_paths = result.get("graph_paths", [])
|
||
|
||
if not raw_rows and request.layer == "auto" and selected_layer != "all":
|
||
fallback = hybrid.retrieve(
|
||
question,
|
||
top_k=request.top_k,
|
||
layer="all",
|
||
use_sqlite=False,
|
||
use_vector=True,
|
||
use_graph=True,
|
||
include_candidate=False,
|
||
anchor_entity=anchor or None,
|
||
)
|
||
if fallback["retrieved_evidence"]:
|
||
raw_rows = fallback["retrieved_evidence"]
|
||
notes.extend(fallback["notes"])
|
||
answer_layer = "all"
|
||
notes.append("Auto layer returned no result; answered from all layers.")
|
||
graph_paths = fallback.get("graph_paths", [])
|
||
finally:
|
||
hybrid.close()
|
||
|
||
# Decorate rows with evidence_key/chunk_hint (using existing helpers)
|
||
rows = _decorate_source_rows(raw_rows)
|
||
|
||
answer_mode = "extractive"
|
||
llm_used = False
|
||
answer = ""
|
||
if request.use_llm and rows:
|
||
try:
|
||
answer, llm_notes, llm_backend = await _generate_llm_answer_with_fallback(
|
||
question,
|
||
answer_layer,
|
||
rows,
|
||
provider=request.llm_provider,
|
||
model=request.llm_model,
|
||
)
|
||
notes.extend(llm_notes)
|
||
if answer.strip():
|
||
answer_mode = f"llm_synthesis:{llm_backend}"
|
||
llm_used = True
|
||
except Exception as exc:
|
||
notes.append(f"LLM answer synthesis failed; returned extractive answer. Error: {exc}")
|
||
print(f"[KB-QA] LLM answer synthesis failed: {exc}")
|
||
|
||
if not answer.strip():
|
||
answer = _format_answer(question, answer_layer, rows, request.include_evidence)
|
||
|
||
return KBQAResponse(
|
||
question=question,
|
||
answer=answer,
|
||
selected_layer=answer_layer,
|
||
layer_reason=reason,
|
||
source_count=len(rows),
|
||
paper_source_count=len({_paper_group_key(row) for row in rows}),
|
||
sources=[_hybrid_row_to_source(row) for row in rows],
|
||
kb_path="hybrid:qdrant+neo4j",
|
||
answer_mode=answer_mode,
|
||
llm_used=llm_used,
|
||
notes=notes,
|
||
)
|
||
|
||
|
||
@router.get("/entity/{entity_name}/relations")
|
||
async def entity_relations(
|
||
entity_name: str,
|
||
layer: str = Query("all"),
|
||
limit: int = Query(30, ge=1, le=200),
|
||
) -> Dict[str, Any]:
|
||
"""Deprecated: SQLite endpoint removed. Use /chat for hybrid retrieval."""
|
||
return {
|
||
"entity": entity_name,
|
||
"deprecated": True,
|
||
"message": "This endpoint was backed by SQLite and is no longer available. Use /api/v1/kb-qa/chat for hybrid (Qdrant + Neo4j) retrieval.",
|
||
"count": 0,
|
||
"relations": [],
|
||
}
|
||
|
||
|
||
@router.get("/relation/{relation_id}")
|
||
async def relation_detail(relation_id: str) -> Dict[str, Any]:
|
||
"""Deprecated: SQLite endpoint removed."""
|
||
return {
|
||
"deprecated": True,
|
||
"message": "This endpoint was backed by SQLite and is no longer available. Use /api/v1/kb-qa/chat for hybrid (Qdrant + Neo4j) retrieval.",
|
||
}
|