1484 lines
70 KiB
Python
1484 lines
70 KiB
Python
"""Adaptive RAG mixin module."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from .shared import *
|
|
from .shared_utils import _env_bool
|
|
from .internal_query_context import internal_context_target_resolution, update_internal_query_context_packet
|
|
|
|
|
|
class AdaptiveRAGAnswerRetrievalMixin:
|
|
_ENV_ENABLE_FIT_BRIEF_EXTERNAL_RECOMMENDATION = "PMFLOW_ENABLE_FIT_BRIEF_EXTERNAL_RECOMMENDATION"
|
|
_EXTERNAL_QUERY_SLOT_INTEGRATION_TYPE = "ai_capability_slots"
|
|
_EXTERNAL_QUERY_PLATFORM_TENANT_ID = int(os.getenv("PMFLOW_PLATFORM_RUNTIME_TENANT_ID", "1") or "1")
|
|
|
|
@staticmethod
|
|
def _external_query_slot_applies(query_semantics: Optional[dict[str, Any]]) -> bool:
|
|
if not isinstance(query_semantics, dict):
|
|
return False
|
|
route_mode = str(query_semantics.get("route_mode") or "").strip().lower()
|
|
source_policy = str(query_semantics.get("effective_source_policy") or query_semantics.get("source_policy_hint") or "").strip().lower()
|
|
route_decision = query_semantics.get("route_decision") if isinstance(query_semantics.get("route_decision"), dict) else {}
|
|
decision_route_mode = str(route_decision.get("route_mode") or "").strip().lower()
|
|
decision_source_policy = str(route_decision.get("effective_source_policy") or "").strip().lower()
|
|
return route_mode == "external" or decision_route_mode == "external" or source_policy == "external" or decision_source_policy == "external"
|
|
|
|
async def _load_active_external_query_slot_binding(self, *, query_semantics: Optional[dict[str, Any]]) -> dict[str, Any] | None:
|
|
if not self._external_query_slot_applies(query_semantics):
|
|
return None
|
|
db = getattr(self, "db", None)
|
|
if db is None:
|
|
return None
|
|
try:
|
|
from sqlalchemy import and_, select
|
|
|
|
from ...models.integration import Integration
|
|
except Exception:
|
|
return None
|
|
try:
|
|
row = (
|
|
await db.execute(
|
|
select(Integration).where(
|
|
and_(
|
|
Integration.tenant_id == self._EXTERNAL_QUERY_PLATFORM_TENANT_ID,
|
|
Integration.type == self._EXTERNAL_QUERY_SLOT_INTEGRATION_TYPE,
|
|
)
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
except Exception:
|
|
return None
|
|
config = row.config if row is not None and isinstance(row.config, dict) else {}
|
|
active = config.get("active") if isinstance(config.get("active"), dict) else {}
|
|
slots = active.get("slots") if isinstance(active.get("slots"), dict) else {}
|
|
saved = slots.get("external_query") if isinstance(slots.get("external_query"), dict) else {}
|
|
primary_provider_id = str(saved.get("primary_provider_id") or saved.get("primary_engine_id") or "").strip()
|
|
if not primary_provider_id:
|
|
return None
|
|
fallback_provider_ids = [
|
|
str(item or "").strip()
|
|
for item in list(saved.get("fallback_provider_ids") or saved.get("fallback_engine_ids") or [])
|
|
if str(item or "").strip()
|
|
]
|
|
return {
|
|
"slot_key": "external_query",
|
|
"slot_version": str(config.get("active_version") or config.get("version") or "unpublished").strip(),
|
|
"primary_provider_id": primary_provider_id,
|
|
"fallback_provider_ids": fallback_provider_ids,
|
|
"verifier_required": bool(saved.get("verifier_required", True)),
|
|
"source_filtering_required": bool(saved.get("source_filtering_required", True)),
|
|
"failure_policy": str(saved.get("failure_policy") or "explicit_fail").strip(),
|
|
}
|
|
|
|
@staticmethod
|
|
def _order_external_query_adapters(*, adapters: dict[str, Any], binding: dict[str, Any] | None) -> list[Any]:
|
|
if not isinstance(binding, dict):
|
|
return [adapters["mcp"], adapters["live_search"]]
|
|
ordered_ids = [
|
|
str(binding.get("primary_provider_id") or "").strip(),
|
|
*[
|
|
str(item or "").strip()
|
|
for item in list(binding.get("fallback_provider_ids") or [])
|
|
if str(item or "").strip()
|
|
],
|
|
]
|
|
ordered: list[Any] = []
|
|
for provider_id in ordered_ids:
|
|
adapter = adapters.get(provider_id)
|
|
if adapter is not None and adapter not in ordered:
|
|
ordered.append(adapter)
|
|
for provider_id in ("mcp", "live_search", "dashscope_responses"):
|
|
adapter = adapters.get(provider_id)
|
|
if adapter is not None and adapter not in ordered:
|
|
ordered.append(adapter)
|
|
return ordered
|
|
|
|
async def _execute_external_mcp_search(self, plan) -> dict[str, Any]:
|
|
from ..external_query_service.schemas import ExternalAdapterResult, ExternalQueryPlan
|
|
from ..mcp_config import McpConfigService
|
|
|
|
try:
|
|
normalized_plan = plan if isinstance(plan, ExternalQueryPlan) else ExternalQueryPlan.model_validate(plan)
|
|
except Exception as exc:
|
|
return ExternalAdapterResult(
|
|
adapter_name="mcp",
|
|
status="error",
|
|
evidence=[],
|
|
meta={"reason": "plan_invalid", "error": str(exc)},
|
|
).model_dump(mode="json")
|
|
|
|
raw_queries = [str(item or "").strip() for item in list(normalized_plan.search_queries or []) if str(item or "").strip()]
|
|
queries = raw_queries or [str(normalized_plan.topic or "").strip()]
|
|
if not any(queries):
|
|
return ExternalAdapterResult(
|
|
adapter_name="mcp",
|
|
status="empty",
|
|
evidence=[],
|
|
meta={"reason": "empty_query"},
|
|
).model_dump(mode="json")
|
|
|
|
db = getattr(self, "db", None)
|
|
if db is None:
|
|
return ExternalAdapterResult(
|
|
adapter_name="mcp",
|
|
status="unavailable",
|
|
evidence=[],
|
|
meta={"reason": "db_unavailable"},
|
|
).model_dump(mode="json")
|
|
|
|
try:
|
|
service = McpConfigService(db, tenant_id=int(getattr(self, "tenant_id", 1) or 1))
|
|
evidence = []
|
|
seen_urls: set[str] = set()
|
|
query_limit = min(len(queries), 5) if bool(normalized_plan.requires_multi_source) else 1
|
|
per_query_limit = max(1, min(int(normalized_plan.result_limit or 3), 2 if bool(normalized_plan.requires_multi_source) else int(normalized_plan.result_limit or 3)))
|
|
searched_queries: list[str] = []
|
|
for query in queries[:query_limit]:
|
|
if not query:
|
|
continue
|
|
searched_queries.append(query)
|
|
rows = await service.search_external(
|
|
query=query,
|
|
result_limit=per_query_limit,
|
|
query_kind=normalized_plan.query_kind,
|
|
)
|
|
for item in list(rows or []):
|
|
url = str(getattr(item, "target_url", "") or (item.get("target_url") if isinstance(item, dict) else "") or "").strip()
|
|
if url and url in seen_urls:
|
|
continue
|
|
if url:
|
|
seen_urls.add(url)
|
|
evidence.append(item)
|
|
if len(evidence) >= int(normalized_plan.result_limit or 3):
|
|
break
|
|
if len(evidence) >= int(normalized_plan.result_limit or 3):
|
|
break
|
|
except Exception as exc:
|
|
return ExternalAdapterResult(
|
|
adapter_name="mcp",
|
|
status="error",
|
|
evidence=[],
|
|
meta={"reason": "backend_direct_runtime_error", "error": str(exc), "query": queries[0] if queries else ""},
|
|
).model_dump(mode="json")
|
|
|
|
if not evidence:
|
|
return ExternalAdapterResult(
|
|
adapter_name="mcp",
|
|
status="empty",
|
|
evidence=[],
|
|
meta={"reason": "no_backend_direct_result", "query": queries[0] if queries else ""},
|
|
).model_dump(mode="json")
|
|
|
|
return ExternalAdapterResult(
|
|
adapter_name="mcp",
|
|
status="success",
|
|
evidence=evidence,
|
|
meta={
|
|
"query": searched_queries[0] if searched_queries else (queries[0] if queries else ""),
|
|
"searched_queries": searched_queries,
|
|
"provider": "mcp_backend_direct",
|
|
},
|
|
).model_dump(mode="json")
|
|
|
|
async def _execute_external_query_gateway(
|
|
self,
|
|
*,
|
|
question: str,
|
|
query_semantics: Optional[dict[str, Any]] = None,
|
|
fit_brief: Optional[dict[str, Any]] = None,
|
|
result_limit: int = 3,
|
|
max_pages: int = 3,
|
|
) -> tuple[list[RetrievedItem], dict[str, Any]]:
|
|
active_external_slot = await self._load_active_external_query_slot_binding(query_semantics=query_semantics)
|
|
live_search_ready = all(
|
|
callable(item)
|
|
for item in (
|
|
getattr(self, "_search_live_external_results", None),
|
|
getattr(self, "_fetch_url_text", None),
|
|
getattr(self, "_chunk_text", None),
|
|
getattr(self, "_score_live_external_search_result", None),
|
|
getattr(self, "_score_live_external_chunk", None),
|
|
)
|
|
)
|
|
mcp_ready = callable(getattr(self, "_execute_external_mcp_search", None)) and getattr(self, "db", None) is not None
|
|
dashscope_ready = bool(str(os.getenv("DASHSCOPE_API_KEY") or "").strip()) or getattr(self, "db", None) is not None
|
|
if not live_search_ready and not mcp_ready and not dashscope_ready:
|
|
if active_external_slot:
|
|
return [], {
|
|
"mode": "external_query_slot_unavailable",
|
|
"slot_key": "external_query",
|
|
"slot_version": active_external_slot.get("slot_version"),
|
|
"slot_runtime_source": "active_slot",
|
|
"failure_policy": "explicit_fail",
|
|
"reason": "external_query_active_slot_has_no_available_adapter",
|
|
"domains_used": ["url"],
|
|
"db_query_ms": 0,
|
|
"vector_search_ms": 0,
|
|
"graph_query_ms": 0,
|
|
"db_timeout": False,
|
|
"vector_timeout": False,
|
|
"graph_timeout": False,
|
|
"tool_trace": [
|
|
{
|
|
"step": 1,
|
|
"tool": "external_query_gateway",
|
|
"slot_key": "external_query",
|
|
"verifier_ok": False,
|
|
"reason": "active_external_query_slot_unavailable",
|
|
}
|
|
],
|
|
}
|
|
legacy_retrieval = getattr(self, "_retrieve_live_external_context", None)
|
|
if callable(legacy_retrieval):
|
|
rows, meta = await legacy_retrieval(
|
|
question=question,
|
|
max_results=result_limit,
|
|
max_pages=max_pages,
|
|
)
|
|
normalized_meta = dict(meta or {})
|
|
normalized_meta.setdefault("mode", "live_external_context" if rows else "live_external_context_empty")
|
|
normalized_meta.setdefault("domains_used", ["url"])
|
|
normalized_meta.setdefault("db_query_ms", 0)
|
|
normalized_meta.setdefault("vector_search_ms", 0)
|
|
normalized_meta.setdefault("graph_query_ms", 0)
|
|
normalized_meta.setdefault("db_timeout", False)
|
|
normalized_meta.setdefault("vector_timeout", False)
|
|
normalized_meta.setdefault("graph_timeout", False)
|
|
normalized_meta.setdefault("tool_trace", [])
|
|
return rows, normalized_meta
|
|
|
|
from ..external_query_service import ExternalQueryPlanner, ExternalProviderRegistry, ExternalResearchRuntime
|
|
from ..external_query_service.adapters import (
|
|
LiveSearchFallbackAdapter,
|
|
MCPExternalAdapter,
|
|
build_dashscope_responses_adapter_from_runtime,
|
|
)
|
|
|
|
planner = ExternalQueryPlanner(
|
|
llm_service=None
|
|
if _env_bool("AI_EXTERNAL_QUERY_FAST_PLANNER_FIRST", True)
|
|
else getattr(self, "llm", None),
|
|
db=getattr(self, "db", None),
|
|
tenant_id=getattr(self, "tenant_id", 1),
|
|
query_builder=None,
|
|
)
|
|
adapters = {
|
|
"dashscope_responses": await build_dashscope_responses_adapter_from_runtime(
|
|
db=getattr(self, "db", None),
|
|
tenant_id=getattr(self, "_EXTERNAL_QUERY_PLATFORM_TENANT_ID", 1),
|
|
),
|
|
"mcp": MCPExternalAdapter(
|
|
search_callable=getattr(self, "_execute_external_mcp_search", None),
|
|
fetch_text_callable=getattr(self, "_fetch_url_text", None),
|
|
chunk_text_callable=getattr(self, "_chunk_text", None),
|
|
score_chunk_callable=getattr(self, "_score_live_external_chunk", None),
|
|
max_pages=max_pages,
|
|
),
|
|
"live_search": LiveSearchFallbackAdapter(
|
|
build_queries_callable=None,
|
|
search_results_callable=getattr(self, "_search_live_external_results", None),
|
|
fetch_text_callable=getattr(self, "_fetch_url_text", None),
|
|
chunk_text_callable=getattr(self, "_chunk_text", None),
|
|
score_search_result_callable=getattr(self, "_score_live_external_search_result", None),
|
|
score_chunk_callable=getattr(self, "_score_live_external_chunk", None),
|
|
max_pages=max_pages,
|
|
),
|
|
}
|
|
registry = ExternalProviderRegistry(
|
|
adapters=self._order_external_query_adapters(adapters=adapters, binding=active_external_slot),
|
|
provider_statuses=ExternalProviderRegistry.default_provider_statuses(
|
|
mcp_available=mcp_ready,
|
|
live_search_available=live_search_ready,
|
|
dashscope_responses_available=bool(adapters["dashscope_responses"].available),
|
|
),
|
|
)
|
|
execution = await ExternalResearchRuntime(
|
|
planner=planner,
|
|
registry=registry,
|
|
adapter_timeout_seconds=float(os.getenv("AI_EXTERNAL_RESEARCH_ADAPTER_TIMEOUT_SECONDS", "14") or "14"),
|
|
max_parallel_adapters=3,
|
|
).execute(
|
|
question=question,
|
|
query_semantics=query_semantics,
|
|
fit_brief=fit_brief,
|
|
)
|
|
context_packet = execution.to_context_packet(question=question).model_dump(mode="json")
|
|
verifier = dict(execution.verifier or {})
|
|
if not bool(verifier.get("ok")):
|
|
blocked_meta = {
|
|
"mode": "external_query_gateway_verifier_blocked",
|
|
"adapter": execution.adapter_name,
|
|
"verifier": verifier,
|
|
"planner": dict(execution.planner_meta or {}),
|
|
"fallback_used": bool((execution.meta or {}).get("fallback_used")),
|
|
"adapter_attempts": list((execution.meta or {}).get("adapter_attempts") or []),
|
|
"query": str(((execution.meta or {}).get("query") or (execution.plan.search_queries[:1] or [""])[0])).strip(),
|
|
"search_result_count": int((execution.meta or {}).get("search_result_count") or 0),
|
|
"fetched_pages": int((execution.meta or {}).get("fetched_pages") or 0),
|
|
"tool_trace": [
|
|
{
|
|
"step": 1,
|
|
"tool": "external_query_gateway",
|
|
"adapter": execution.adapter_name,
|
|
"planner": str((execution.planner_meta or {}).get("planner") or "unknown"),
|
|
"verifier_ok": False,
|
|
}
|
|
],
|
|
"domains_used": ["url"],
|
|
"db_query_ms": 0,
|
|
"vector_search_ms": 0,
|
|
"graph_query_ms": 0,
|
|
"db_timeout": False,
|
|
"vector_timeout": False,
|
|
"graph_timeout": False,
|
|
"external_context_packet": context_packet,
|
|
}
|
|
if active_external_slot:
|
|
blocked_meta.update(
|
|
{
|
|
"slot_key": "external_query",
|
|
"slot_version": active_external_slot.get("slot_version"),
|
|
"slot_runtime_source": "active_slot",
|
|
"slot_provider_id": execution.adapter_name,
|
|
"failure_policy": "explicit_fail",
|
|
}
|
|
)
|
|
return [], blocked_meta
|
|
|
|
rows: list[RetrievedItem] = []
|
|
for item in list(execution.evidence or [])[: max(1, int(result_limit))]:
|
|
source_type = str(getattr(item, "source_type", "url") or "url").strip().lower() or "url"
|
|
normalized_source = source_type if source_type in {"url", "mcp_search"} else "url"
|
|
title = str(getattr(item, "title", "") or "").strip()
|
|
source_label = str(getattr(item, "source_label", "") or "").strip() or title
|
|
target_url = str(getattr(item, "target_url", "") or "").strip()
|
|
snippet = str(getattr(item, "snippet", "") or "").strip()
|
|
freshness_score = float(getattr(item, "freshness_score", 0.0) or 0.0)
|
|
authority_score = float(getattr(item, "authority_score", 0.0) or 0.0)
|
|
rows.append(
|
|
RetrievedItem(
|
|
source=normalized_source,
|
|
content=(snippet or title or target_url)[:1800],
|
|
score=round(max(freshness_score, authority_score, 0.55), 4),
|
|
source_label=(
|
|
f"MCP[{source_label[:120]}]"
|
|
if normalized_source == "mcp_search"
|
|
else f"URL[{source_label[:120]}]"
|
|
),
|
|
metadata={
|
|
"title": title[:255],
|
|
"url": target_url,
|
|
"file_url": target_url,
|
|
"target_url": target_url,
|
|
"retrieved_via": "external_query_gateway",
|
|
"external_source_type": source_type,
|
|
"external_adapter": execution.adapter_name,
|
|
"freshness_score": round(freshness_score, 4),
|
|
"authority_score": round(authority_score, 4),
|
|
"grounded_fields": list(getattr(item, "grounded_fields", []) or []),
|
|
**(dict(getattr(item, "raw_metadata", {}) or {})),
|
|
},
|
|
)
|
|
)
|
|
|
|
meta = dict(execution.meta or {})
|
|
meta.update(
|
|
{
|
|
"mode": "external_query_gateway" if rows else "external_query_gateway_empty",
|
|
"adapter": execution.adapter_name,
|
|
"verifier": verifier,
|
|
"planner": dict(execution.planner_meta or {}),
|
|
"query": str(meta.get("query") or (execution.plan.search_queries[:1] or [""])[0]).strip(),
|
|
"search_result_count": int(meta.get("search_result_count") or 0),
|
|
"fetched_pages": int(meta.get("fetched_pages") or 0),
|
|
"domains_used": ["url"],
|
|
"db_query_ms": 0,
|
|
"vector_search_ms": 0,
|
|
"graph_query_ms": 0,
|
|
"db_timeout": False,
|
|
"vector_timeout": False,
|
|
"graph_timeout": False,
|
|
"external_context_packet": context_packet,
|
|
"tool_trace": [
|
|
{
|
|
"step": 1,
|
|
"tool": "external_query_gateway",
|
|
"adapter": execution.adapter_name,
|
|
"planner": str((execution.planner_meta or {}).get("planner") or "unknown"),
|
|
"verifier_ok": True,
|
|
"result_count": len(rows),
|
|
"fallback_used": bool(meta.get("fallback_used")),
|
|
}
|
|
],
|
|
}
|
|
)
|
|
if active_external_slot:
|
|
meta.update(
|
|
{
|
|
"slot_key": "external_query",
|
|
"slot_version": active_external_slot.get("slot_version"),
|
|
"slot_runtime_source": "active_slot",
|
|
"slot_provider_id": execution.adapter_name,
|
|
"failure_policy": "explicit_fail",
|
|
}
|
|
)
|
|
return rows, meta
|
|
|
|
async def _maybe_fetch_live_external_freshness_context(
|
|
self,
|
|
*,
|
|
question: str,
|
|
effective_question: str,
|
|
query_semantics: Optional[dict[str, Any]],
|
|
scene_semantics: Optional[dict[str, Any]],
|
|
) -> tuple[list[RetrievedItem], dict[str, Any]]:
|
|
semantics = scene_semantics if isinstance(scene_semantics, dict) else {}
|
|
if not self._requires_external_freshness(question=question, scene_semantics=semantics):
|
|
return [], {"mode": "skip_non_freshness"}
|
|
search_question = str(effective_question or question or "").strip()
|
|
if not search_question:
|
|
return [], {"mode": "skip_empty_question"}
|
|
execute_gateway = getattr(self, "_execute_external_query_gateway", None)
|
|
try:
|
|
if callable(execute_gateway):
|
|
rows, live_meta = await execute_gateway(
|
|
question=search_question,
|
|
query_semantics=query_semantics,
|
|
result_limit=3,
|
|
max_pages=3,
|
|
)
|
|
else:
|
|
rows, live_meta = await self._retrieve_live_external_context(
|
|
question=search_question,
|
|
max_results=3,
|
|
max_pages=3,
|
|
)
|
|
except Exception as exc:
|
|
mode = "external_query_gateway_error" if callable(execute_gateway) else "live_external_context_error"
|
|
return [], {"mode": mode, "error": str(exc)}
|
|
meta = dict(live_meta or {})
|
|
if not str(meta.get("mode") or "").strip():
|
|
if callable(execute_gateway):
|
|
meta["mode"] = "external_query_gateway" if rows else "external_query_gateway_empty"
|
|
else:
|
|
meta["mode"] = "live_external_context" if rows else "live_external_context_empty"
|
|
meta["external_freshness_only"] = bool(
|
|
isinstance(query_semantics, dict) and query_semantics.get("external_freshness_only")
|
|
)
|
|
return rows, meta
|
|
|
|
async def _build_retrieval_response(
|
|
self,
|
|
*,
|
|
clean_q: str,
|
|
effective_q: str,
|
|
asked_by: Optional[int],
|
|
asked_via: str,
|
|
project_id: Optional[int],
|
|
source_chat_id: Optional[str],
|
|
session_id: Optional[int],
|
|
turn_index: int,
|
|
context_session_id: Optional[int],
|
|
conversation: list[dict[str, Any]],
|
|
state_row: Optional[QASessionState],
|
|
rewrite_meta: dict[str, Any],
|
|
query_type: str,
|
|
query_semantics: dict[str, Any],
|
|
retrieval_domains: list[str],
|
|
scene_semantics: dict[str, Any],
|
|
thinking_mode: str,
|
|
active_thinking_mode: str,
|
|
deep_thinking: bool,
|
|
forced_deep_reason: str,
|
|
deep_retrieval_steps: int,
|
|
deep_vector_top_k: int,
|
|
system_snapshot: dict[str, Any],
|
|
user_snapshot: Optional[dict[str, Any]],
|
|
started: float,
|
|
message_intent: Optional[str] = None,
|
|
mixed_signal_part: Optional[str] = None,
|
|
answer_mode: str = ANSWER_MODE_RETRIEVAL_NORMAL,
|
|
consultant_mode_preference: Optional[str] = None,
|
|
route_decision: Optional[dict[str, Any]] = None,
|
|
internal_context_packet: Any = None,
|
|
) -> dict[str, Any]:
|
|
project_operating_context = (
|
|
query_semantics.get("project_operating_context")
|
|
if isinstance(query_semantics.get("project_operating_context"), dict)
|
|
else {}
|
|
)
|
|
execution_recipe = (
|
|
query_semantics.get("execution_recipe")
|
|
if isinstance(query_semantics.get("execution_recipe"), dict)
|
|
else {}
|
|
)
|
|
domains = [str(item).strip().lower() for item in retrieval_domains if str(item or "").strip()]
|
|
if not domains:
|
|
domains = ["db", "graph", "doc", "feishu", "pmflow", "url"]
|
|
if str(asked_via or "").strip().lower() == "feishu" and source_chat_id and "feishu" not in domains:
|
|
domains.append("feishu")
|
|
normalize_route_decision = getattr(self, "_normalize_query_route_decision", None)
|
|
normalized_route_decision = (
|
|
normalize_route_decision(route_decision)
|
|
if callable(normalize_route_decision)
|
|
else (dict(route_decision or {}) if isinstance(route_decision, dict) else {})
|
|
)
|
|
execution_primary_lane = str(execution_recipe.get("primary_lane") or "").strip().lower()
|
|
external_route_only = str(
|
|
normalized_route_decision.get("effective_source_policy")
|
|
or query_semantics.get("effective_source_policy")
|
|
or query_semantics.get("source_policy")
|
|
or ""
|
|
).strip().lower() == "external" or execution_primary_lane == "external"
|
|
fit_brief_external_recommendation_enabled = _env_bool(
|
|
self._ENV_ENABLE_FIT_BRIEF_EXTERNAL_RECOMMENDATION,
|
|
True,
|
|
)
|
|
fit_brief: dict[str, Any] = {}
|
|
build_project_fit_brief = getattr(self, "_build_project_fit_brief", None)
|
|
if (
|
|
fit_brief_external_recommendation_enabled
|
|
and external_route_only
|
|
and bool(execution_recipe.get("fit_brief_required"))
|
|
and callable(build_project_fit_brief)
|
|
):
|
|
try:
|
|
fit_brief = (
|
|
await build_project_fit_brief(
|
|
question=effective_q or clean_q,
|
|
project_operating_context=project_operating_context,
|
|
system_snapshot=system_snapshot,
|
|
user_snapshot=user_snapshot,
|
|
)
|
|
) or {}
|
|
except Exception:
|
|
fit_brief = {}
|
|
if external_route_only:
|
|
retrieved, search_meta = await self._execute_external_query_gateway(
|
|
question=effective_q,
|
|
query_semantics=query_semantics,
|
|
fit_brief=fit_brief,
|
|
result_limit=3,
|
|
max_pages=3,
|
|
)
|
|
else:
|
|
retrieved, search_meta = await self._multi_step_search(
|
|
question=effective_q,
|
|
domains=domains,
|
|
project_id=project_id,
|
|
source_chat_id=source_chat_id,
|
|
query_semantics=query_semantics,
|
|
internal_context_packet=internal_context_packet,
|
|
session_state=state_row,
|
|
max_steps_override=deep_retrieval_steps if deep_thinking else None,
|
|
vector_top_k=deep_vector_top_k if deep_thinking else None,
|
|
)
|
|
db_query_ms = int(search_meta.get("db_query_ms") or 0)
|
|
vector_search_ms = int(search_meta.get("vector_search_ms") or 0)
|
|
graph_query_ms = int(search_meta.get("graph_query_ms") or 0)
|
|
db_timeout = bool(search_meta.get("db_timeout"))
|
|
vector_timeout = bool(search_meta.get("vector_timeout"))
|
|
graph_timeout = bool(search_meta.get("graph_timeout"))
|
|
tool_trace = list(search_meta.get("tool_trace") or [])
|
|
tool_trace.insert(
|
|
0,
|
|
{
|
|
"step": 0,
|
|
"tool": "thinking_mode",
|
|
"mode": active_thinking_mode,
|
|
"fast_path_bypassed": bool(deep_thinking),
|
|
"reason": forced_deep_reason or (
|
|
"manual_deep" if self._normalize_thinking_mode(thinking_mode) == THINKING_MODE_DEEP else "normal"
|
|
),
|
|
},
|
|
)
|
|
tool_trace.insert(
|
|
0,
|
|
{
|
|
"step": 0,
|
|
"tool": "execution_recipe",
|
|
"primary_lane": execution_recipe.get("primary_lane"),
|
|
"plan_handoff_mode": execution_recipe.get("plan_handoff_mode"),
|
|
"fit_brief_required": bool(execution_recipe.get("fit_brief_required")),
|
|
},
|
|
)
|
|
tool_trace.insert(
|
|
0,
|
|
{
|
|
"step": 0,
|
|
"tool": "query_route_gateway",
|
|
"route_mode": normalized_route_decision.get("route_mode"),
|
|
"effective_source_policy": normalized_route_decision.get("effective_source_policy"),
|
|
"decision_origin": normalized_route_decision.get("decision_origin"),
|
|
},
|
|
)
|
|
final_domains = [str(x) for x in (search_meta.get("domains_used") or domains)]
|
|
prompt_variant_id = self._peek_answer_prompt_variant()
|
|
external_context_packet = (
|
|
dict(search_meta.get("external_context_packet") or {})
|
|
if external_route_only and isinstance(search_meta.get("external_context_packet"), dict)
|
|
else None
|
|
)
|
|
build_external_recommendation = getattr(self, "_build_external_recommendation", None)
|
|
external_recommendation = (
|
|
build_external_recommendation(
|
|
external_context_packet=external_context_packet,
|
|
fit_brief=fit_brief,
|
|
)
|
|
if (
|
|
fit_brief_external_recommendation_enabled
|
|
and external_route_only
|
|
and isinstance(external_context_packet, dict)
|
|
and callable(build_external_recommendation)
|
|
)
|
|
else {}
|
|
)
|
|
if external_route_only and isinstance(external_context_packet, dict) and isinstance(external_recommendation, dict):
|
|
external_context_packet = {
|
|
**external_context_packet,
|
|
"retrieval_status": external_recommendation.get("retrieval_status"),
|
|
"impact_status": external_recommendation.get("impact_status"),
|
|
"affected_signal_sources": list(external_recommendation.get("affected_signal_sources") or []),
|
|
}
|
|
build_plan_handoff_fields = getattr(self, "_build_plan_handoff_fields", None)
|
|
plan_handoff_fields = (
|
|
build_plan_handoff_fields(
|
|
execution_recipe=execution_recipe,
|
|
project_operating_context=project_operating_context,
|
|
)
|
|
if callable(build_plan_handoff_fields)
|
|
else {}
|
|
)
|
|
load_analysis_project_snapshot = getattr(self, "_load_analysis_project_snapshot_for_packet", None)
|
|
analysis_project_snapshot = (
|
|
await load_analysis_project_snapshot(
|
|
project_operating_context=project_operating_context,
|
|
fallback_project_id=project_id,
|
|
)
|
|
if callable(load_analysis_project_snapshot)
|
|
else {}
|
|
)
|
|
build_analysis_packet_payload = getattr(self, "_build_analysis_packet_payload", None)
|
|
analysis_packet = (
|
|
build_analysis_packet_payload(
|
|
route_decision=normalized_route_decision,
|
|
project_operating_context=project_operating_context,
|
|
execution_recipe=execution_recipe,
|
|
fit_brief=fit_brief,
|
|
external_recommendation=external_recommendation,
|
|
recommended_next_step=plan_handoff_fields.get("recommended_next_step"),
|
|
answer="",
|
|
evidence_pack={},
|
|
query_type=query_type if query_type != QUERY_TYPE_UNKNOWN else QUERY_TYPE_EXPLAIN,
|
|
clarification_code=None,
|
|
anchor_kind=None,
|
|
meta_followup_kind=None,
|
|
internal_context_packet=internal_context_packet,
|
|
project_snapshot=analysis_project_snapshot,
|
|
)
|
|
if callable(build_analysis_packet_payload)
|
|
else {}
|
|
)
|
|
response_payload_snapshot = {
|
|
"project_operating_context": project_operating_context,
|
|
"execution_recipe": execution_recipe,
|
|
"fit_brief": fit_brief,
|
|
"external_recommendation": external_recommendation,
|
|
"analysis_packet": analysis_packet,
|
|
"handoff_mode": plan_handoff_fields.get("handoff_mode"),
|
|
"can_enter_plan": plan_handoff_fields.get("can_enter_plan"),
|
|
"recommended_next_step": plan_handoff_fields.get("recommended_next_step"),
|
|
}
|
|
external_verifier = (
|
|
dict(external_context_packet.get("verifier") or {})
|
|
if isinstance(external_context_packet, dict)
|
|
else {}
|
|
)
|
|
external_missing_fields = [
|
|
str(item).strip()
|
|
for item in list(external_verifier.get("missing_fields") or [])
|
|
if str(item).strip()
|
|
]
|
|
external_insufficient_evidence = bool(
|
|
external_route_only
|
|
and isinstance(external_context_packet, dict)
|
|
and (
|
|
not bool(external_verifier.get("ok"))
|
|
or not list(external_context_packet.get("verified_evidence") or [])
|
|
)
|
|
)
|
|
|
|
max_score = retrieved[0].score if retrieved else 0.0
|
|
active_retrieved = list(retrieved)
|
|
answer_text = await self._generate_answer(
|
|
effective_q,
|
|
active_retrieved,
|
|
conversation,
|
|
project_id,
|
|
system_snapshot=system_snapshot,
|
|
user_snapshot=user_snapshot,
|
|
relevance_max=max_score,
|
|
thinking_mode=active_thinking_mode,
|
|
context_block_limit=DEEP_CONTEXT_BLOCKS if deep_thinking else None,
|
|
answer_max_tokens=DEEP_ANSWER_MAX_TOKENS if deep_thinking else None,
|
|
asked_via=asked_via,
|
|
message_intent=message_intent,
|
|
mixed_signal_part=mixed_signal_part,
|
|
query_type=query_type,
|
|
consultant_mode_preference=consultant_mode_preference,
|
|
scene_semantics=scene_semantics,
|
|
query_semantics=query_semantics,
|
|
external_context_packet=external_context_packet,
|
|
)
|
|
external_presentation = (
|
|
self._peek_external_answer_presentation()
|
|
if hasattr(self, "_peek_external_answer_presentation")
|
|
else {}
|
|
)
|
|
signal_source_recommendation = (
|
|
self._normalize_signal_source_recommendation(
|
|
external_presentation.get("recommendation_packet"),
|
|
project_id=project_id,
|
|
)
|
|
if hasattr(self, "_normalize_signal_source_recommendation")
|
|
else {}
|
|
)
|
|
research_workspace = (
|
|
dict(external_presentation.get("research_workspace") or {})
|
|
if isinstance(external_presentation.get("research_workspace"), dict)
|
|
else {}
|
|
)
|
|
if signal_source_recommendation:
|
|
response_payload_snapshot["signal_source_recommendation"] = signal_source_recommendation
|
|
if research_workspace:
|
|
response_payload_snapshot["research_workspace"] = research_workspace
|
|
prompt_variant_id = self._peek_answer_prompt_variant()
|
|
routing_snapshot = self._peek_last_answer_routing_snapshot()
|
|
used_sources = self._dedupe_sources(active_retrieved)
|
|
if external_insufficient_evidence:
|
|
answer_mode = ANSWER_MODE_CLARIFY
|
|
max_score = min(float(max_score or 0.0), 0.4)
|
|
if not used_sources and not external_insufficient_evidence:
|
|
used_sources = [
|
|
{
|
|
"source": "system",
|
|
"label": "System Snapshot",
|
|
"score": round(max(max_score, 0.6), 4),
|
|
"metadata": {"mode": "snapshot_only", "relevance_max": round(max_score, 4)},
|
|
}
|
|
]
|
|
if user_snapshot:
|
|
used_sources.append(
|
|
{
|
|
"source": "system",
|
|
"label": "User Snapshot",
|
|
"score": round(max(max_score, 0.58), 4),
|
|
"metadata": {"mode": "snapshot_only", "relevance_max": round(max_score, 4)},
|
|
}
|
|
)
|
|
|
|
pre_retry_guard_semantics = self._evaluate_post_retrieval_guard(
|
|
deep_intent=str((scene_semantics or {}).get("deep_intent") or "none"),
|
|
sources=used_sources,
|
|
relevance_max=max_score,
|
|
)
|
|
auto_retry_reason = ""
|
|
if not deep_thinking:
|
|
auto_retry_reason = str(pre_retry_guard_semantics.get("auto_retry_reason") or "").strip()
|
|
if auto_retry_reason and not external_route_only:
|
|
deep_retrieved, deep_meta = await self._multi_step_search(
|
|
question=effective_q,
|
|
domains=domains,
|
|
project_id=project_id,
|
|
source_chat_id=source_chat_id,
|
|
query_semantics=query_semantics,
|
|
internal_context_packet=internal_context_packet,
|
|
session_state=state_row,
|
|
max_steps_override=deep_retrieval_steps,
|
|
vector_top_k=deep_vector_top_k,
|
|
)
|
|
deep_max_score = deep_retrieved[0].score if deep_retrieved else 0.0
|
|
deep_answer_text = await self._generate_answer(
|
|
effective_q,
|
|
deep_retrieved,
|
|
conversation,
|
|
project_id,
|
|
system_snapshot=system_snapshot,
|
|
user_snapshot=user_snapshot,
|
|
relevance_max=deep_max_score,
|
|
thinking_mode=THINKING_MODE_DEEP,
|
|
context_block_limit=DEEP_CONTEXT_BLOCKS,
|
|
answer_max_tokens=DEEP_ANSWER_MAX_TOKENS,
|
|
asked_via=asked_via,
|
|
message_intent=message_intent,
|
|
mixed_signal_part=mixed_signal_part,
|
|
query_type=query_type,
|
|
consultant_mode_preference=consultant_mode_preference,
|
|
scene_semantics=scene_semantics,
|
|
query_semantics=query_semantics,
|
|
)
|
|
deep_sources = self._dedupe_sources(deep_retrieved)
|
|
if not deep_sources:
|
|
deep_sources = [
|
|
{
|
|
"source": "system",
|
|
"label": "System Snapshot",
|
|
"score": round(max(deep_max_score, 0.6), 4),
|
|
"metadata": {"mode": "snapshot_only", "relevance_max": round(deep_max_score, 4)},
|
|
}
|
|
]
|
|
if user_snapshot:
|
|
deep_sources.append(
|
|
{
|
|
"source": "system",
|
|
"label": "User Snapshot",
|
|
"score": round(max(deep_max_score, 0.58), 4),
|
|
"metadata": {"mode": "snapshot_only", "relevance_max": round(deep_max_score, 4)},
|
|
}
|
|
)
|
|
|
|
guard_semantics = self._evaluate_post_retrieval_guard(
|
|
deep_intent=str((scene_semantics or {}).get("deep_intent") or "none"),
|
|
sources=deep_sources,
|
|
relevance_max=deep_max_score,
|
|
)
|
|
merged_metrics = self._merge_search_metrics(
|
|
{
|
|
"db_query_ms": db_query_ms,
|
|
"vector_search_ms": vector_search_ms,
|
|
"graph_query_ms": graph_query_ms,
|
|
"db_timeout": db_timeout,
|
|
"vector_timeout": vector_timeout,
|
|
"graph_timeout": graph_timeout,
|
|
},
|
|
deep_meta,
|
|
)
|
|
db_query_ms = int(merged_metrics.get("db_query_ms") or 0)
|
|
vector_search_ms = int(merged_metrics.get("vector_search_ms") or 0)
|
|
graph_query_ms = int(merged_metrics.get("graph_query_ms") or 0)
|
|
db_timeout = bool(merged_metrics.get("db_timeout"))
|
|
vector_timeout = bool(merged_metrics.get("vector_timeout"))
|
|
graph_timeout = bool(merged_metrics.get("graph_timeout"))
|
|
|
|
deep_trace = list(deep_meta.get("tool_trace") or [])
|
|
tool_trace.append(
|
|
{
|
|
"step": 98,
|
|
"tool": "auto_deep_retry",
|
|
"reason": auto_retry_reason,
|
|
"prev_relevance_max": round(float(max_score or 0.0), 4),
|
|
"prev_snapshot_only": bool(pre_retry_guard_semantics.get("snapshot_only_sources")),
|
|
}
|
|
)
|
|
for row in deep_trace:
|
|
tool_trace.append(row)
|
|
|
|
final_domains = [str(x) for x in (deep_meta.get("domains_used") or final_domains or domains)]
|
|
max_score = deep_max_score
|
|
answer_text = deep_answer_text
|
|
used_sources = deep_sources
|
|
active_retrieved = list(deep_retrieved)
|
|
deep_thinking = True
|
|
active_thinking_mode = THINKING_MODE_DEEP
|
|
answer_mode = ANSWER_MODE_RETRIEVAL_DEEP
|
|
prompt_variant_id = self._peek_answer_prompt_variant()
|
|
routing_snapshot = self._peek_last_answer_routing_snapshot()
|
|
|
|
evidence_guard_triggered = False
|
|
clarification_due_to_evidence = bool(external_insufficient_evidence)
|
|
freshness_guard_triggered = False
|
|
if self._requires_external_freshness(question=clean_q, scene_semantics=scene_semantics) and not self._has_external_grounding_sources(used_sources):
|
|
live_rows, live_meta = await self._maybe_fetch_live_external_freshness_context(
|
|
question=clean_q,
|
|
effective_question=effective_q,
|
|
query_semantics=query_semantics,
|
|
scene_semantics=scene_semantics,
|
|
)
|
|
if live_rows:
|
|
active_retrieved = self._merge_retrieved_items(
|
|
active_retrieved,
|
|
live_rows,
|
|
limit=max(DEFAULT_TOP_K * 2, 8),
|
|
)
|
|
max_score = active_retrieved[0].score if active_retrieved else max_score
|
|
answer_text = await self._generate_answer(
|
|
effective_q,
|
|
active_retrieved,
|
|
conversation,
|
|
project_id,
|
|
system_snapshot=system_snapshot,
|
|
user_snapshot=user_snapshot,
|
|
relevance_max=max_score,
|
|
thinking_mode=active_thinking_mode,
|
|
context_block_limit=DEEP_CONTEXT_BLOCKS if deep_thinking else None,
|
|
answer_max_tokens=DEEP_ANSWER_MAX_TOKENS if deep_thinking else None,
|
|
asked_via=asked_via,
|
|
message_intent=message_intent,
|
|
mixed_signal_part=mixed_signal_part,
|
|
query_type=query_type,
|
|
consultant_mode_preference=consultant_mode_preference,
|
|
scene_semantics=scene_semantics,
|
|
query_semantics=query_semantics,
|
|
external_context_packet=(
|
|
dict(live_meta.get("external_context_packet") or {})
|
|
if isinstance(live_meta.get("external_context_packet"), dict)
|
|
else external_context_packet
|
|
),
|
|
)
|
|
if isinstance(live_meta.get("external_context_packet"), dict):
|
|
external_context_packet = dict(live_meta.get("external_context_packet") or {})
|
|
external_recommendation = (
|
|
build_external_recommendation(
|
|
external_context_packet=external_context_packet,
|
|
fit_brief=fit_brief,
|
|
)
|
|
if fit_brief_external_recommendation_enabled and callable(build_external_recommendation)
|
|
else {}
|
|
)
|
|
if isinstance(external_recommendation, dict):
|
|
external_context_packet = {
|
|
**external_context_packet,
|
|
"retrieval_status": external_recommendation.get("retrieval_status"),
|
|
"impact_status": external_recommendation.get("impact_status"),
|
|
"affected_signal_sources": list(external_recommendation.get("affected_signal_sources") or []),
|
|
}
|
|
external_verifier = dict(external_context_packet.get("verifier") or {})
|
|
external_missing_fields = [
|
|
str(item).strip()
|
|
for item in list(external_verifier.get("missing_fields") or [])
|
|
if str(item).strip()
|
|
]
|
|
external_insufficient_evidence = bool(
|
|
external_route_only
|
|
and (
|
|
not bool(external_verifier.get("ok"))
|
|
or not list(external_context_packet.get("verified_evidence") or [])
|
|
)
|
|
)
|
|
used_sources = self._dedupe_sources(active_retrieved)
|
|
if "url" not in final_domains:
|
|
final_domains.append("url")
|
|
tool_trace.append(
|
|
{
|
|
"step": 95,
|
|
"tool": "live_external_context",
|
|
"result_count": len(live_rows),
|
|
"search_result_count": int(live_meta.get("search_result_count") or 0),
|
|
"fetched_pages": int(live_meta.get("fetched_pages") or 0),
|
|
"query": str(live_meta.get("query") or "").strip()[:240],
|
|
}
|
|
)
|
|
prompt_variant_id = self._peek_answer_prompt_variant()
|
|
routing_snapshot = self._peek_last_answer_routing_snapshot()
|
|
elif str(live_meta.get("mode") or "").strip():
|
|
if isinstance(live_meta.get("external_context_packet"), dict):
|
|
external_context_packet = dict(live_meta.get("external_context_packet") or {})
|
|
external_recommendation = (
|
|
build_external_recommendation(
|
|
external_context_packet=external_context_packet,
|
|
fit_brief=fit_brief,
|
|
)
|
|
if fit_brief_external_recommendation_enabled and callable(build_external_recommendation)
|
|
else {}
|
|
)
|
|
if isinstance(external_recommendation, dict):
|
|
external_context_packet = {
|
|
**external_context_packet,
|
|
"retrieval_status": external_recommendation.get("retrieval_status"),
|
|
"impact_status": external_recommendation.get("impact_status"),
|
|
"affected_signal_sources": list(external_recommendation.get("affected_signal_sources") or []),
|
|
}
|
|
external_verifier = dict(external_context_packet.get("verifier") or {})
|
|
external_missing_fields = [
|
|
str(item).strip()
|
|
for item in list(external_verifier.get("missing_fields") or [])
|
|
if str(item).strip()
|
|
]
|
|
external_insufficient_evidence = bool(
|
|
external_route_only
|
|
and (
|
|
not bool(external_verifier.get("ok"))
|
|
or not list(external_context_packet.get("verified_evidence") or [])
|
|
)
|
|
)
|
|
if external_insufficient_evidence:
|
|
clarification_due_to_evidence = True
|
|
answer_mode = ANSWER_MODE_CLARIFY
|
|
max_score = min(float(max_score or 0.0), 0.4)
|
|
active_retrieved = []
|
|
used_sources = []
|
|
answer_text = await self._generate_answer(
|
|
effective_q,
|
|
active_retrieved,
|
|
conversation,
|
|
project_id,
|
|
system_snapshot=system_snapshot,
|
|
user_snapshot=user_snapshot,
|
|
relevance_max=max_score,
|
|
thinking_mode=active_thinking_mode,
|
|
context_block_limit=DEEP_CONTEXT_BLOCKS if deep_thinking else None,
|
|
answer_max_tokens=DEEP_ANSWER_MAX_TOKENS if deep_thinking else None,
|
|
asked_via=asked_via,
|
|
message_intent=message_intent,
|
|
mixed_signal_part=mixed_signal_part,
|
|
query_type=query_type,
|
|
consultant_mode_preference=consultant_mode_preference,
|
|
scene_semantics=scene_semantics,
|
|
query_semantics=query_semantics,
|
|
external_context_packet=external_context_packet,
|
|
)
|
|
external_presentation = (
|
|
self._peek_external_answer_presentation()
|
|
if hasattr(self, "_peek_external_answer_presentation")
|
|
else {}
|
|
)
|
|
signal_source_recommendation = (
|
|
self._normalize_signal_source_recommendation(
|
|
external_presentation.get("recommendation_packet"),
|
|
project_id=project_id,
|
|
)
|
|
if hasattr(self, "_normalize_signal_source_recommendation")
|
|
else {}
|
|
)
|
|
research_workspace = (
|
|
dict(external_presentation.get("research_workspace") or {})
|
|
if isinstance(external_presentation.get("research_workspace"), dict)
|
|
else {}
|
|
)
|
|
response_payload_snapshot["external_recommendation"] = external_recommendation
|
|
if signal_source_recommendation:
|
|
response_payload_snapshot["signal_source_recommendation"] = signal_source_recommendation
|
|
if research_workspace:
|
|
response_payload_snapshot["research_workspace"] = research_workspace
|
|
prompt_variant_id = self._peek_answer_prompt_variant()
|
|
routing_snapshot = self._peek_last_answer_routing_snapshot()
|
|
tool_trace.append(
|
|
{
|
|
"step": 95,
|
|
"tool": "live_external_context",
|
|
"mode": str(live_meta.get("mode") or "").strip(),
|
|
"query": str(live_meta.get("query") or "").strip()[:240],
|
|
}
|
|
)
|
|
if (
|
|
self._requires_external_freshness(question=clean_q, scene_semantics=scene_semantics)
|
|
and not self._has_external_grounding_sources(used_sources)
|
|
and not (external_route_only and isinstance(external_context_packet, dict) and bool(external_context_packet))
|
|
):
|
|
freshness_guard_triggered = True
|
|
clarification_due_to_evidence = True
|
|
answer_text = self._build_external_freshness_guard_answer(clean_q)
|
|
used_sources = [
|
|
{
|
|
"source": "system",
|
|
"label": "External Freshness Guard",
|
|
"score": 0.93,
|
|
"metadata": {"mode": "clarification", "reason": "missing_external_freshness_evidence"},
|
|
}
|
|
]
|
|
max_score = min(float(max_score or 0.0), 0.4)
|
|
if "system_context" not in final_domains:
|
|
final_domains.append("system_context")
|
|
tool_trace.append(
|
|
{
|
|
"step": 96,
|
|
"tool": "freshness_guard",
|
|
"reason": "missing_external_freshness_evidence",
|
|
}
|
|
)
|
|
answer_mode = ANSWER_MODE_CLARIFY
|
|
if (
|
|
not external_insufficient_evidence
|
|
and not freshness_guard_triggered
|
|
and not evidence_guard_triggered
|
|
and self._is_snapshot_only_sources(used_sources)
|
|
and float(max_score or 0.0) < 0.45
|
|
):
|
|
clarification_due_to_evidence = True
|
|
answer_text = (
|
|
f"当前关于“{clean_q[:80]}”的证据不足，现阶段只命中系统快照。"
|
|
"请补充更明确的对象、范围或文档入口，我再给你可核验的结论。"
|
|
)
|
|
used_sources = [
|
|
{
|
|
"source": "system",
|
|
"label": "Evidence Guard",
|
|
"score": 0.91,
|
|
"metadata": {"mode": "clarification", "reason": "insufficient_evidence"},
|
|
}
|
|
]
|
|
max_score = min(float(max_score or 0.0), 0.4)
|
|
if "system_context" not in final_domains:
|
|
final_domains.append("system_context")
|
|
tool_trace.append(
|
|
{
|
|
"step": 97,
|
|
"tool": "clarification_guard",
|
|
"reason": "snapshot_only_low_relevance",
|
|
}
|
|
)
|
|
answer_mode = ANSWER_MODE_CLARIFY
|
|
|
|
guard_semantics = self._evaluate_post_retrieval_guard(
|
|
deep_intent=str((scene_semantics or {}).get("deep_intent") or "none"),
|
|
sources=used_sources,
|
|
relevance_max=max_score,
|
|
)
|
|
evidence_guard_triggered = bool(guard_semantics.get("should_block_with_guard"))
|
|
if evidence_guard_triggered:
|
|
answer_text = self._build_evidence_guard_answer(
|
|
clean_q,
|
|
guard_focus=str(guard_semantics.get("guard_focus") or "call_chain"),
|
|
)
|
|
used_sources = [
|
|
{
|
|
"source": "system",
|
|
"label": "Evidence Guard",
|
|
"score": 0.99,
|
|
"metadata": {"mode": "guard", "reason": "snapshot_only_process_query"},
|
|
}
|
|
]
|
|
max_score = min(float(max_score or 0.0), 0.35)
|
|
if "system_context" not in final_domains:
|
|
final_domains.append("system_context")
|
|
tool_trace.append(
|
|
{
|
|
"step": 99,
|
|
"tool": "answer_guard",
|
|
"reason": "insufficient_call_chain_evidence",
|
|
}
|
|
)
|
|
answer_mode = ANSWER_MODE_BLOCKED
|
|
|
|
elapsed = int((time.perf_counter() - started) * 1000)
|
|
grounding_verdict = GroundingVerdict(
|
|
status=GROUNDING_STATUS_GROUNDED,
|
|
missing_fields=[],
|
|
reason_code=None,
|
|
)
|
|
if evidence_guard_triggered:
|
|
grounding_verdict = GroundingVerdict(
|
|
status=GROUNDING_STATUS_BLOCKED,
|
|
missing_fields=["call_chain_evidence", "target_scope"],
|
|
reason_code="insufficient_evidence",
|
|
)
|
|
elif clarification_due_to_evidence:
|
|
grounding_verdict = GroundingVerdict(
|
|
status=GROUNDING_STATUS_INSUFFICIENT_EVIDENCE,
|
|
missing_fields=(
|
|
["external_freshness_evidence"]
|
|
if freshness_guard_triggered
|
|
else (external_missing_fields or ["high_relevance_evidence"])
|
|
),
|
|
reason_code="insufficient_evidence",
|
|
)
|
|
elif self._is_snapshot_only_sources(used_sources) or float(max_score or 0.0) < 0.45:
|
|
grounding_verdict = GroundingVerdict(
|
|
status=GROUNDING_STATUS_PARTIALLY_GROUNDED,
|
|
missing_fields=["high_relevance_evidence"] if float(max_score or 0.0) < 0.45 else [],
|
|
reason_code=None,
|
|
)
|
|
|
|
retrieval_metrics = {
|
|
"db_query_ms": int(db_query_ms or 0),
|
|
"vector_search_ms": int(vector_search_ms or 0),
|
|
"graph_query_ms": int(graph_query_ms or 0),
|
|
"db_timeout": bool(db_timeout),
|
|
"vector_timeout": bool(vector_timeout),
|
|
"graph_timeout": bool(graph_timeout),
|
|
}
|
|
customer_presentation = (
|
|
getattr(self, "_last_customer_answer_presentation", {})
|
|
if isinstance(getattr(self, "_last_customer_answer_presentation", {}), dict)
|
|
else {}
|
|
)
|
|
customer_anchor_sources = [
|
|
item for item in list(customer_presentation.get("anchor_sources") or []) if isinstance(item, dict)
|
|
]
|
|
if customer_anchor_sources and not evidence_guard_triggered and not clarification_due_to_evidence:
|
|
seen_anchor_ids = {
|
|
str((item.get("metadata") if isinstance(item.get("metadata"), dict) else {}).get("project_id") or "")
|
|
for item in used_sources
|
|
if isinstance(item, dict)
|
|
}
|
|
for item in customer_anchor_sources:
|
|
meta = item.get("metadata") if isinstance(item.get("metadata"), dict) else {}
|
|
anchor_id = str(meta.get("project_id") or "").strip()
|
|
if anchor_id and anchor_id not in seen_anchor_ids:
|
|
used_sources.append(item)
|
|
seen_anchor_ids.add(anchor_id)
|
|
if isinstance(customer_presentation.get("packet"), dict):
|
|
response_payload_snapshot["customer_evidence_packet"] = dict(customer_presentation.get("packet") or {})
|
|
if isinstance(customer_presentation.get("interaction_context"), dict):
|
|
response_payload_snapshot["customer_interaction_context_packet"] = dict(customer_presentation.get("interaction_context") or {})
|
|
if isinstance(customer_presentation.get("answer_process"), dict):
|
|
response_payload_snapshot["answer_process_packet"] = dict(customer_presentation.get("answer_process") or {})
|
|
filter_sources_for_route = getattr(self, "_filter_sources_for_route", None)
|
|
if callable(filter_sources_for_route):
|
|
used_sources = filter_sources_for_route(
|
|
sources=used_sources,
|
|
route_decision=normalized_route_decision,
|
|
)
|
|
render_with_contract = getattr(self, "_render_answer_with_route_contract", None)
|
|
if callable(render_with_contract):
|
|
answer_text = render_with_contract(
|
|
answer=answer_text,
|
|
sources=used_sources,
|
|
route_decision=normalized_route_decision,
|
|
)
|
|
effective_source_policy = str(normalized_route_decision.get("effective_source_policy") or "").strip().lower()
|
|
contract_verdict = {"ok": True, "reason_code": None, "missing_fields": []}
|
|
guard_payload: Optional[dict[str, Any]] = None
|
|
if effective_source_policy == "external" and answer_mode != ANSWER_MODE_CLARIFY:
|
|
contract_verdict = self._verify_external_answer_contract(
|
|
answer=answer_text,
|
|
sources=used_sources,
|
|
route_decision=normalized_route_decision,
|
|
)
|
|
if not bool(contract_verdict.get("ok")):
|
|
guard_payload = self._build_external_answer_contract_guard_payload(
|
|
reason_code=str(contract_verdict.get("reason_code") or ""),
|
|
missing_fields=list(contract_verdict.get("missing_fields") or []),
|
|
)
|
|
else:
|
|
contract_verdict = self._verify_internal_answer_contract(
|
|
query_type=query_type,
|
|
answer=answer_text,
|
|
sources=used_sources,
|
|
route_decision=normalized_route_decision,
|
|
internal_context_packet=internal_context_packet,
|
|
)
|
|
if not bool(contract_verdict.get("ok")):
|
|
guard_payload = self._build_internal_answer_contract_guard_payload(
|
|
query_type=query_type,
|
|
reason_code=str(contract_verdict.get("reason_code") or ""),
|
|
missing_fields=list(contract_verdict.get("missing_fields") or []),
|
|
target_resolution=internal_context_target_resolution(
|
|
packet=internal_context_packet,
|
|
fallback=(query_semantics or {}).get("target_resolution"),
|
|
),
|
|
)
|
|
if guard_payload:
|
|
answer_text = str(guard_payload.get("answer") or answer_text)
|
|
used_sources = list(guard_payload.get("sources") or [])
|
|
answer_mode = str(guard_payload.get("answer_mode") or answer_mode or ANSWER_MODE_BLOCKED)
|
|
max_score = min(float(max_score or 0.0), 0.35)
|
|
if "system_context" not in final_domains:
|
|
final_domains.append("system_context")
|
|
tool_trace.append(
|
|
{
|
|
"step": 100,
|
|
"tool": (
|
|
"external_answer_contract_guard"
|
|
if effective_source_policy == "external"
|
|
else "internal_answer_contract_guard"
|
|
),
|
|
"reason": str(
|
|
guard_payload.get("reason_code")
|
|
or (
|
|
"external_answer_contract_violation"
|
|
if effective_source_policy == "external"
|
|
else "internal_answer_contract_violation"
|
|
)
|
|
),
|
|
}
|
|
)
|
|
grounding_verdict = GroundingVerdict(
|
|
status=str(
|
|
guard_payload.get("grounding_status")
|
|
or (GROUNDING_STATUS_INSUFFICIENT_EVIDENCE if effective_source_policy == "external" else GROUNDING_STATUS_BLOCKED)
|
|
),
|
|
missing_fields=list(guard_payload.get("missing_fields") or []),
|
|
reason_code=str(
|
|
guard_payload.get("reason_code")
|
|
or (
|
|
"external_answer_contract_violation"
|
|
if effective_source_policy == "external"
|
|
else "internal_answer_contract_violation"
|
|
)
|
|
),
|
|
)
|
|
internal_context_packet = update_internal_query_context_packet(
|
|
internal_context_packet,
|
|
executor_summary={
|
|
"answer_mode": str(answer_mode or "").strip() or ANSWER_MODE_RETRIEVAL_NORMAL,
|
|
"retrieval_domains": list(final_domains or []),
|
|
"db_query_ms": int(db_query_ms or 0),
|
|
"vector_search_ms": int(vector_search_ms or 0),
|
|
"graph_query_ms": int(graph_query_ms or 0),
|
|
},
|
|
verifier_summary={
|
|
"ok": bool(contract_verdict.get("ok")),
|
|
"reason_code": str(contract_verdict.get("reason_code") or "").strip() or None,
|
|
"missing_fields": [
|
|
str(item).strip()
|
|
for item in list(contract_verdict.get("missing_fields") or [])
|
|
if str(item).strip()
|
|
],
|
|
},
|
|
uncertainties=[
|
|
str(grounding_verdict.reason_code or "").strip()
|
|
if str(grounding_verdict.reason_code or "").strip()
|
|
else "",
|
|
*[
|
|
str(item).strip()
|
|
for item in list(grounding_verdict.missing_fields or [])
|
|
if str(item).strip()
|
|
],
|
|
],
|
|
)
|
|
inject_runtime_summaries = getattr(self, "_inject_internal_query_runtime_summaries", None)
|
|
if callable(inject_runtime_summaries):
|
|
retrieval_metrics = inject_runtime_summaries(
|
|
retrieval_metrics=retrieval_metrics,
|
|
query_semantics=query_semantics,
|
|
internal_context_packet=internal_context_packet,
|
|
)
|
|
runtime_fields = self._build_qa_record_runtime_fields(
|
|
answer=answer_text,
|
|
sources=used_sources,
|
|
missing_fields=grounding_verdict.missing_fields,
|
|
answer_mode=answer_mode,
|
|
grounding_status=grounding_verdict.status,
|
|
prompt_variant_id=prompt_variant_id,
|
|
reason_code=grounding_verdict.reason_code,
|
|
routing_snapshot=routing_snapshot,
|
|
tool_trace=tool_trace,
|
|
retrieval_metrics=retrieval_metrics,
|
|
)
|
|
load_analysis_project_snapshot = getattr(self, "_load_analysis_project_snapshot_for_packet", None)
|
|
analysis_project_snapshot = (
|
|
await load_analysis_project_snapshot(
|
|
project_operating_context=project_operating_context,
|
|
fallback_project_id=project_id,
|
|
)
|
|
if callable(load_analysis_project_snapshot)
|
|
else {}
|
|
)
|
|
if callable(build_analysis_packet_payload):
|
|
response_payload_snapshot["analysis_packet"] = build_analysis_packet_payload(
|
|
route_decision=normalized_route_decision,
|
|
project_operating_context=project_operating_context,
|
|
execution_recipe=execution_recipe,
|
|
fit_brief=fit_brief,
|
|
external_recommendation=external_recommendation,
|
|
recommended_next_step=plan_handoff_fields.get("recommended_next_step"),
|
|
answer=answer_text,
|
|
evidence_pack=runtime_fields.get("evidence_pack") if isinstance(runtime_fields, dict) else {},
|
|
query_type=query_type if query_type != QUERY_TYPE_UNKNOWN else QUERY_TYPE_EXPLAIN,
|
|
clarification_code=None,
|
|
anchor_kind=None,
|
|
meta_followup_kind=None,
|
|
internal_context_packet=internal_context_packet,
|
|
project_snapshot=analysis_project_snapshot,
|
|
)
|
|
qa = QARecord(
|
|
tenant_id=self.tenant_id,
|
|
question=clean_q,
|
|
answer=answer_text,
|
|
sources=used_sources,
|
|
intent="question",
|
|
domains_used=final_domains,
|
|
response_time_ms=elapsed,
|
|
asked_by=asked_by,
|
|
asked_via=(asked_via or "pmflow")[:20],
|
|
project_id=project_id,
|
|
session_id=session_id,
|
|
turn_index=turn_index,
|
|
**self._build_qa_record_external_fields(
|
|
anchor_kind=None,
|
|
anchor_ref=None,
|
|
routing_snapshot=routing_snapshot,
|
|
route_decision=normalized_route_decision,
|
|
response_payload=response_payload_snapshot,
|
|
),
|
|
**runtime_fields,
|
|
)
|
|
self.db.add(qa)
|
|
confidence = round(max(0.2, min(1.0, max_score if max_score > 0 else 0.55)), 4)
|
|
if evidence_guard_triggered or answer_mode == ANSWER_MODE_CLARIFY:
|
|
confidence = min(confidence, 0.35)
|
|
missing_fields = list(grounding_verdict.missing_fields)
|
|
await self._update_session_state_after_turn(
|
|
session_state=state_row,
|
|
question=clean_q,
|
|
effective_question=effective_q,
|
|
answer=answer_text,
|
|
query_type=query_type if query_type != QUERY_TYPE_UNKNOWN else QUERY_TYPE_EXPLAIN,
|
|
ambiguity="clear" if not missing_fields else "low_evidence",
|
|
rewrite_meta=rewrite_meta,
|
|
sources=used_sources,
|
|
response_payload=response_payload_snapshot,
|
|
route_decision=normalized_route_decision,
|
|
)
|
|
await self.db.flush()
|
|
persist_route_memory = getattr(self, "_persist_query_route_memory", None)
|
|
if callable(persist_route_memory):
|
|
await persist_route_memory(
|
|
qa_record_id=int(qa.id) if getattr(qa, "id", None) is not None else None,
|
|
question_text=clean_q,
|
|
project_id=project_id,
|
|
route_decision=normalized_route_decision,
|
|
)
|
|
await self.db.commit()
|
|
await self.db.refresh(qa)
|
|
await self.context_cache.append_session_messages(
|
|
tenant_id=self.tenant_id,
|
|
session_id=context_session_id,
|
|
question=clean_q,
|
|
answer=answer_text,
|
|
)
|
|
|
|
return self._build_response_payload(
|
|
answer=answer_text,
|
|
sources=used_sources,
|
|
domains_used=final_domains,
|
|
response_time_ms=elapsed,
|
|
db_query_ms=db_query_ms,
|
|
vector_search_ms=vector_search_ms,
|
|
graph_query_ms=graph_query_ms,
|
|
db_timeout=db_timeout,
|
|
vector_timeout=vector_timeout,
|
|
graph_timeout=graph_timeout,
|
|
tool_trace=runtime_fields["tool_trace"],
|
|
qa_id=int(qa.id),
|
|
session_id=int(session_id) if session_id is not None else None,
|
|
turn_index=int(turn_index),
|
|
relevance_max=round(max_score, 4),
|
|
query_type=query_type if query_type != QUERY_TYPE_UNKNOWN else QUERY_TYPE_EXPLAIN,
|
|
confidence=confidence,
|
|
missing_fields=missing_fields,
|
|
thinking_mode=active_thinking_mode,
|
|
reason_code=grounding_verdict.reason_code or ("answer_generated" if answer_mode not in {ANSWER_MODE_CLARIFY, ANSWER_MODE_BLOCKED} else "insufficient_evidence"),
|
|
answer_mode=answer_mode,
|
|
grounding_status=grounding_verdict.status,
|
|
policy_version=runtime_fields["policy_version"],
|
|
citations=runtime_fields["citations"],
|
|
sources_used=runtime_fields["sources_used"],
|
|
evidence_pack=runtime_fields["evidence_pack"],
|
|
prompt_variant_id=prompt_variant_id,
|
|
route_decision=normalized_route_decision,
|
|
project_operating_context=project_operating_context,
|
|
execution_recipe=execution_recipe,
|
|
fit_brief=fit_brief,
|
|
external_recommendation=external_recommendation,
|
|
signal_source_recommendation=response_payload_snapshot.get("signal_source_recommendation")
|
|
if isinstance(response_payload_snapshot, dict)
|
|
else None,
|
|
research_workspace=response_payload_snapshot.get("research_workspace")
|
|
if isinstance(response_payload_snapshot, dict)
|
|
else None,
|
|
analysis_packet=analysis_packet,
|
|
internal_context_packet=internal_context_packet,
|
|
customer_evidence_packet=response_payload_snapshot.get("customer_evidence_packet")
|
|
if isinstance(response_payload_snapshot, dict)
|
|
else None,
|
|
customer_interaction_context_packet=response_payload_snapshot.get("customer_interaction_context_packet")
|
|
if isinstance(response_payload_snapshot, dict)
|
|
else None,
|
|
answer_process_packet=response_payload_snapshot.get("answer_process_packet")
|
|
if isinstance(response_payload_snapshot, dict)
|
|
else None,
|
|
)
|