322 lines
14 KiB
Python
322 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import queue
|
|
import threading
|
|
from functools import lru_cache
|
|
import os
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from . import __version__
|
|
from .config import QAConfig
|
|
from .pipeline import QAAnswerError, QAV1Pipeline
|
|
|
|
|
|
class QARequest(BaseModel):
|
|
question: str = Field(min_length=1)
|
|
debug: bool = False
|
|
use_provisional_draft: Optional[bool] = None
|
|
max_rounds: Optional[int] = Field(default=None, ge=1, le=2)
|
|
ablation: Optional[Dict[str, Any]] = None
|
|
use_llm: Optional[bool] = None
|
|
chain_top_k: Optional[int] = Field(default=3, ge=0, le=20)
|
|
chain_hops: Optional[int] = Field(default=2, ge=1, le=2)
|
|
chain_max_edges: Optional[int] = Field(default=12, ge=0, le=120)
|
|
chain_max_nodes: Optional[int] = Field(default=10, ge=0, le=80)
|
|
chain_mode: Optional[str] = Field(default="display")
|
|
chain_min_score: Optional[float] = Field(default=0.45, ge=0.0, le=1.0)
|
|
native_chain_enabled: bool = True
|
|
native_chain_max_hops: int = Field(default=3, ge=1, le=3)
|
|
native_chain_top_k: int = Field(default=5, ge=0, le=20)
|
|
native_chain_min_hops: int = Field(default=1, ge=1, le=3)
|
|
native_chain_beam_width: int = Field(default=60, ge=1, le=300)
|
|
native_chain_per_node_expand: int = Field(default=12, ge=1, le=50)
|
|
endpoint_path_enabled: bool = False
|
|
endpoint_path_max_hops: int = Field(default=3, ge=1, le=3)
|
|
endpoint_path_top_k: int = Field(default=5, ge=0, le=10)
|
|
endpoint_path_max_endpoint_candidates: int = Field(default=3, ge=1, le=5)
|
|
endpoint_path_per_node_limit: int = Field(default=16, ge=2, le=50)
|
|
endpoint_path_max_expansions: int = Field(default=24, ge=10, le=1000)
|
|
endpoint_path_max_frontier_states: int = Field(default=120, ge=20, le=2000)
|
|
endpoint_path_time_budget_seconds: float = Field(default=5.0, ge=1.0, le=30.0)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_pipeline() -> QAV1Pipeline:
|
|
return QAV1Pipeline()
|
|
|
|
|
|
router = APIRouter(prefix="/api/v1/qa-v1-1-73", tags=["QA V1.1.73"] )
|
|
|
|
|
|
@router.get("/health")
|
|
def health() -> Dict[str, Any]:
|
|
return {"status": "ok", "service": "qa_v1_1_73", "version": __version__}
|
|
|
|
|
|
@router.get("/info")
|
|
def info() -> Dict[str, Any]:
|
|
config = QAConfig.from_env()
|
|
return {
|
|
"name": "QA V1.1.73 Assertion-Anchored Full Graph",
|
|
"version": __version__,
|
|
"pipeline": [
|
|
"question",
|
|
"provisional answer",
|
|
"structured planner",
|
|
"direct Qdrant recall over V15/V16 canonical assertions",
|
|
"canonical seed extraction",
|
|
"Neo4j exact seed retrieval",
|
|
"Neo4j bounded graph expansion",
|
|
"BGE rerank",
|
|
"graph-balanced top30 evidence chain, with at least 15 graph-derived items when available",
|
|
"BGE top10 Qdrant source seed selection for cascaded native Neo4j two-hop chain retrieval",
|
|
"BGE top10 complete assertions locked as hop 1 and expanded from each target in Neo4j",
|
|
"merge assertion-anchored chains with the unchanged formal native graph lane",
|
|
"compose a validated final LLM answer for chat; retrieval-only search returns evidence without an answer",
|
|
],
|
|
"safety_boundary": (
|
|
"The provisional answer is never a final evidence source. "
|
|
"The original-question retrieval lane is always retained."
|
|
),
|
|
"config": {
|
|
"alias_path": str(config.alias_path),
|
|
"taxonomy_path": str(config.taxonomy_path),
|
|
"reranker_model": str(config.reranker_model),
|
|
"max_rounds": config.max_rounds,
|
|
"max_evidence": config.max_evidence,
|
|
"provisional_draft": config.enable_provisional_draft,
|
|
"second_round": config.enable_second_round,
|
|
"bge_rerank": config.enable_bge_rerank,
|
|
"graph_schema": config.graph_schema,
|
|
"backend": {
|
|
"release_id": os.environ.get("NORMALIZED_RELEASE_ID", ""),
|
|
"qdrant_collection": os.environ.get(
|
|
"V15_QDRANT_COLLECTION", ""
|
|
),
|
|
"graph_schema": config.graph_schema,
|
|
},
|
|
"knowledge_chain_defaults": {
|
|
"chain_top_k": 3,
|
|
"chain_hops": 2,
|
|
"chain_max_edges": 12,
|
|
"chain_max_nodes": 10,
|
|
"chain_mode": "display",
|
|
"used_for_answer_generation": True,
|
|
"graph_min_evidence_in_top30": 15,
|
|
"native_chain_max_hops": 3,
|
|
"native_chain_top_k": 5,
|
|
"native_chain_min_hops": 1,
|
|
"endpoint_path_enabled": False,
|
|
"endpoint_path_max_hops": 3,
|
|
"endpoint_path_top_k": 5,
|
|
},
|
|
"request_ablation_supported": [
|
|
"use_provisional_draft",
|
|
"max_rounds",
|
|
"ablation.retrieval.use_qdrant",
|
|
"ablation.retrieval.use_neo4j",
|
|
"ablation.retrieval.use_graph_expansion",
|
|
"ablation.retrieval.use_v15",
|
|
"ablation.use_bge_rerank",
|
|
"ablation.native_seed_policy",
|
|
"ablation.native_seed_top_k",
|
|
"ablation.native_target_seed_policy",
|
|
"endpoint_path_enabled",
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
@router.post("/chat")
|
|
async def chat(request: QARequest) -> Dict[str, Any]:
|
|
if request.use_llm is False:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail={"error_code": "llm_required", "answer": "", "message": "Final-answer LLM is required"},
|
|
)
|
|
pipeline = get_pipeline()
|
|
if not pipeline.llm.available:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail={"error_code": "llm_unavailable", "answer": "", "message": "Final-answer LLM is not configured"},
|
|
)
|
|
try:
|
|
ablation = dict(request.ablation or {})
|
|
ablation["use_final_llm"] = True
|
|
return await asyncio.to_thread(
|
|
pipeline.answer,
|
|
request.question,
|
|
debug=request.debug,
|
|
use_provisional_draft=request.use_provisional_draft,
|
|
max_rounds=request.max_rounds,
|
|
ablation=ablation,
|
|
chain_top_k=request.chain_top_k,
|
|
chain_hops=request.chain_hops,
|
|
chain_max_edges=request.chain_max_edges,
|
|
chain_max_nodes=request.chain_max_nodes,
|
|
chain_mode=request.chain_mode,
|
|
chain_min_score=request.chain_min_score,
|
|
native_chain_enabled=request.native_chain_enabled,
|
|
native_chain_max_hops=request.native_chain_max_hops,
|
|
native_chain_top_k=request.native_chain_top_k,
|
|
native_chain_min_hops=request.native_chain_min_hops,
|
|
native_chain_beam_width=request.native_chain_beam_width,
|
|
native_chain_per_node_expand=request.native_chain_per_node_expand,
|
|
endpoint_path_enabled=request.endpoint_path_enabled,
|
|
endpoint_path_max_hops=request.endpoint_path_max_hops,
|
|
endpoint_path_top_k=request.endpoint_path_top_k,
|
|
endpoint_path_max_endpoint_candidates=request.endpoint_path_max_endpoint_candidates,
|
|
endpoint_path_per_node_limit=request.endpoint_path_per_node_limit,
|
|
endpoint_path_max_expansions=request.endpoint_path_max_expansions,
|
|
endpoint_path_max_frontier_states=request.endpoint_path_max_frontier_states,
|
|
endpoint_path_time_budget_seconds=request.endpoint_path_time_budget_seconds,
|
|
)
|
|
except QAAnswerError as exc:
|
|
raise HTTPException(
|
|
status_code=exc.status_code,
|
|
detail={"error_code": exc.error_code, "answer": "", "message": str(exc)},
|
|
) from exc
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error_code": "qa_failure", "answer": "", "message": str(exc)},
|
|
) from exc
|
|
|
|
|
|
@router.post("/chat/stream")
|
|
async def chat_stream(request: QARequest) -> StreamingResponse:
|
|
"""Stream truthful pipeline milestones, then the unchanged chat payload."""
|
|
if request.use_llm is False:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail={"error_code": "llm_required", "answer": "", "message": "Final-answer LLM is required"},
|
|
)
|
|
pipeline = get_pipeline()
|
|
if not pipeline.llm.available:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail={"error_code": "llm_unavailable", "answer": "", "message": "Final-answer LLM is not configured"},
|
|
)
|
|
|
|
events: queue.Queue[tuple[str, Dict[str, Any]] | None] = queue.Queue()
|
|
|
|
def emit(payload: Dict[str, Any]) -> None:
|
|
events.put(("progress", payload))
|
|
|
|
def run() -> None:
|
|
try:
|
|
ablation = dict(request.ablation or {})
|
|
ablation["use_final_llm"] = True
|
|
result = pipeline.answer(
|
|
request.question,
|
|
debug=request.debug,
|
|
use_provisional_draft=request.use_provisional_draft,
|
|
max_rounds=request.max_rounds,
|
|
ablation=ablation,
|
|
chain_top_k=request.chain_top_k,
|
|
chain_hops=request.chain_hops,
|
|
chain_max_edges=request.chain_max_edges,
|
|
chain_max_nodes=request.chain_max_nodes,
|
|
chain_mode=request.chain_mode,
|
|
chain_min_score=request.chain_min_score,
|
|
native_chain_enabled=request.native_chain_enabled,
|
|
native_chain_max_hops=request.native_chain_max_hops,
|
|
native_chain_top_k=request.native_chain_top_k,
|
|
native_chain_min_hops=request.native_chain_min_hops,
|
|
native_chain_beam_width=request.native_chain_beam_width,
|
|
native_chain_per_node_expand=request.native_chain_per_node_expand,
|
|
endpoint_path_enabled=request.endpoint_path_enabled,
|
|
endpoint_path_max_hops=request.endpoint_path_max_hops,
|
|
endpoint_path_top_k=request.endpoint_path_top_k,
|
|
endpoint_path_max_endpoint_candidates=request.endpoint_path_max_endpoint_candidates,
|
|
endpoint_path_per_node_limit=request.endpoint_path_per_node_limit,
|
|
endpoint_path_max_expansions=request.endpoint_path_max_expansions,
|
|
endpoint_path_max_frontier_states=request.endpoint_path_max_frontier_states,
|
|
endpoint_path_time_budget_seconds=request.endpoint_path_time_budget_seconds,
|
|
progress_callback=emit,
|
|
)
|
|
events.put(("complete", result))
|
|
except QAAnswerError as exc:
|
|
events.put(("error", {
|
|
"success": False,
|
|
"status": exc.status_code,
|
|
"error_code": exc.error_code,
|
|
"message": str(exc),
|
|
"answer": "",
|
|
}))
|
|
except Exception as exc:
|
|
events.put(("error", {
|
|
"success": False,
|
|
"status": 500,
|
|
"error_code": "qa_failure",
|
|
"message": str(exc),
|
|
"answer": "",
|
|
}))
|
|
finally:
|
|
events.put(None)
|
|
|
|
threading.Thread(target=run, name="qa-v1-1-73-stream", daemon=True).start()
|
|
|
|
def stream():
|
|
while True:
|
|
try:
|
|
item = events.get(timeout=15)
|
|
except queue.Empty:
|
|
yield ": keepalive\n\n"
|
|
continue
|
|
if item is None:
|
|
break
|
|
event_name, payload = item
|
|
yield (
|
|
f"event: {event_name}\n"
|
|
f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
|
)
|
|
|
|
return StreamingResponse(
|
|
stream(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|
|
|
|
|
|
@router.post("/search")
|
|
async def search(request: QARequest) -> Dict[str, Any]:
|
|
response = await asyncio.to_thread(
|
|
get_pipeline().answer,
|
|
request.question,
|
|
debug=True,
|
|
use_provisional_draft=request.use_provisional_draft,
|
|
max_rounds=request.max_rounds,
|
|
ablation=dict(request.ablation or {}),
|
|
chain_top_k=request.chain_top_k,
|
|
chain_hops=request.chain_hops,
|
|
chain_max_edges=request.chain_max_edges,
|
|
chain_max_nodes=request.chain_max_nodes,
|
|
chain_mode=request.chain_mode,
|
|
chain_min_score=request.chain_min_score,
|
|
native_chain_enabled=request.native_chain_enabled,
|
|
native_chain_max_hops=request.native_chain_max_hops,
|
|
native_chain_top_k=request.native_chain_top_k,
|
|
native_chain_min_hops=request.native_chain_min_hops,
|
|
native_chain_beam_width=request.native_chain_beam_width,
|
|
native_chain_per_node_expand=request.native_chain_per_node_expand,
|
|
endpoint_path_enabled=request.endpoint_path_enabled,
|
|
endpoint_path_max_hops=request.endpoint_path_max_hops,
|
|
endpoint_path_top_k=request.endpoint_path_top_k,
|
|
endpoint_path_max_endpoint_candidates=request.endpoint_path_max_endpoint_candidates,
|
|
endpoint_path_per_node_limit=request.endpoint_path_per_node_limit,
|
|
endpoint_path_max_expansions=request.endpoint_path_max_expansions,
|
|
endpoint_path_max_frontier_states=request.endpoint_path_max_frontier_states,
|
|
endpoint_path_time_budget_seconds=request.endpoint_path_time_budget_seconds,
|
|
retrieval_only=True,
|
|
)
|
|
if isinstance(response, dict):
|
|
response["answer"] = ""
|
|
return response
|