"""Adaptive RAG internal target resolution helpers.""" from __future__ import annotations from .shared import * from .internal_query_context import ( internal_context_operation_contract, internal_context_query_plan, internal_context_target_resolution, ) from .internal_domain_catalog import config_target_catalog, internal_domain_target class AdaptiveRAGTargetResolutionMixin: _TARGET_ENTITY_TYPES = {"project", "task", "feature", "config", "doc", "person", "user", "signal", "unknown"} _TARGET_RESOLUTION_MODES = {"deterministic", "llm_rerank", "semantic_filter", "clarify", "none"} _BUSINESS_FRAME_FIELDS = { "goal_kind", "target_scope", "resource_role", "source_kind", "expected_answer_shape", "conversation_posture", "project_binding", "subject_binding", "object_binding", "candidate_set_binding", "evidence_mode", } @classmethod def _normalize_target_entity_type_name(cls, value: Any) -> str: _ = cls normalized = str(value or "").strip().lower() if normalized == "person": return "user" if normalized == "document": return "doc" return normalized @classmethod def _normalize_object_mentions( cls, raw: Any, *, question: str = "", ) -> list[str]: _ = (cls, question) mentions: list[str] = [] rows = raw if isinstance(raw, list) else [] for item in rows: text = normalize_lookup_fragment(item, limit=32) if len(text) >= 2 and text not in mentions: mentions.append(text) return mentions[:4] def _build_operation_contract( self, *, question: str, query_semantics: Optional[dict[str, Any]], ) -> dict[str, Any]: _ = question semantics = query_semantics if isinstance(query_semantics, dict) else {} query_type = str(semantics.get("query_type") or QUERY_TYPE_UNKNOWN).strip().lower() or QUERY_TYPE_UNKNOWN business_frame = self._coalesce_business_frame(query_semantics=semantics) query_profile = dict(semantics.get("query_profile") or {}) if isinstance(semantics.get("query_profile"), dict) else {} config_bound = ( str(business_frame.get("resource_role") or "").strip().lower() == "config" or str(business_frame.get("object_binding") or "").strip().lower().startswith("config:") ) if query_type in {QUERY_TYPE_EXISTENCE, QUERY_TYPE_UNKNOWN} and config_bound: query_type = QUERY_TYPE_STATUS object_mentions = self._normalize_object_mentions( semantics.get("object_mentions"), question=question, ) contract = { "query_type": query_type, "answer_contract": ( self._internal_answer_contract_for_query_type(query_type) if hasattr(self, "_internal_answer_contract_for_query_type") else "clarify_or_fail" ), "query_profile": query_profile, "object_mentions": object_mentions, "clarify_policy": "ask_when_target_ambiguous", } contract.update({key: value for key, value in business_frame.items() if key in self._BUSINESS_FRAME_FIELDS}) return contract @classmethod def _coalesce_business_frame( cls, *, query_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: semantics = query_semantics if isinstance(query_semantics, dict) else {} contract = operation_contract if isinstance(operation_contract, dict) else {} rows: list[dict[str, Any]] = [] if isinstance(semantics.get("business_frame"), dict): rows.append(dict(semantics.get("business_frame") or {})) rows.append(dict(semantics)) if isinstance(contract.get("business_frame"), dict): rows.append(dict(contract.get("business_frame") or {})) rows.append(dict(contract)) frame: dict[str, Any] = {} for row in rows: if not isinstance(row, dict): continue for key in cls._BUSINESS_FRAME_FIELDS: value = row.get(key) if value in (None, "", []): continue if key == "project_binding": try: value = int(value) except Exception: value = str(value).strip() or None if value is not None: frame[key] = value return frame def _build_session_followup_target( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: dict[str, Any], ) -> Optional[dict[str, Any]]: semantics = query_semantics if isinstance(query_semantics, dict) else {} followup_context = ( semantics.get("session_followup_context") if isinstance(semantics.get("session_followup_context"), dict) else {} ) rewrite_mode = str(followup_context.get("rewrite_mode") or "").strip().lower() if rewrite_mode not in {"state_subject", "state_entities"}: return None entity_ids: list[int] = [] for value in list(followup_context.get("last_entity_ids") or []): try: entity_ids.append(int(value)) except Exception: continue if len(entity_ids) != 1: return None business_frame = self._coalesce_business_frame( query_semantics=semantics, operation_contract=operation_contract, ) conversation_posture = str(business_frame.get("conversation_posture") or "").strip().lower() target_scope = str(business_frame.get("target_scope") or "").strip().lower() resource_role = str(business_frame.get("resource_role") or "").strip().lower() project_binding = business_frame.get("project_binding") object_binding = str(business_frame.get("object_binding") or "").strip().lower() requested_entity_types = self._requested_internal_entity_types( query_semantics=semantics, operation_contract=operation_contract, ) is_project_followup = bool(requested_entity_types) or ( target_scope in {"current_project", "explicit_project"} and project_binding not in (None, "", []) ) or object_binding.startswith("project:") if conversation_posture == "ambient_chat" or not is_project_followup: return None mentions = list(operation_contract.get("object_mentions") or []) resolved_label = ( str(business_frame.get("subject_binding") or "").strip() or str(followup_context.get("last_subject") or "").strip() or (str(mentions[0]).strip() if mentions else "") or None ) resolved_project_id = entity_ids[0] if requested_entity_types == {"project"}: return { "target_entity_type": "project", "target_ref": f"project:{resolved_project_id}", "resolved_label": resolved_label, "resolution_mode": "deterministic", "binding_source": "session_followup_context", "confidence": 0.93 if rewrite_mode == "state_subject" else 0.88, "candidate_summary": [], } return self._build_generic_domain_target_from_business_frame( query_semantics=semantics, operation_contract=operation_contract, binding_source="session_followup_context", confidence=0.9 if rewrite_mode == "state_subject" else 0.86, resolved_label_override=resolved_label, ) def _config_target_ref_from_business_frame( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], ) -> str | None: semantics = query_semantics if isinstance(query_semantics, dict) else {} business_frame = self._coalesce_business_frame( query_semantics=semantics, operation_contract=operation_contract, ) object_binding = str(business_frame.get("object_binding") or "").strip().lower() if object_binding.startswith("config:"): return object_binding return None def _config_domain_requested( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], ) -> bool: semantics = query_semantics if isinstance(query_semantics, dict) else {} business_frame = self._coalesce_business_frame( query_semantics=semantics, operation_contract=operation_contract, ) return str(business_frame.get("resource_role") or "").strip().lower() == "config" def _requested_internal_entity_types( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], ) -> set[str]: business_frame = self._coalesce_business_frame( query_semantics=query_semantics, operation_contract=operation_contract, ) object_binding = str(business_frame.get("object_binding") or "").strip().lower() if object_binding.startswith("config:"): return {"config"} if object_binding.startswith("feature:"): return {"feature"} if object_binding.startswith("task:"): return {"task"} if object_binding.startswith("signal:"): return {"signal"} if object_binding.startswith(("doc:", "document:")): return {"doc"} if object_binding.startswith(("user:", "person:")): return {"user"} if object_binding.startswith("project:"): return {"project"} resource_role = str(business_frame.get("resource_role") or "").strip().lower() if resource_role == "config": return {"config"} if resource_role == "feature": return {"feature"} if resource_role == "task": return {"task"} if resource_role in {"signal", "signal_source", "external_input"}: return {"signal"} if resource_role == "people": return {"user"} if resource_role == "doc": return {"doc"} target_scope = str(business_frame.get("target_scope") or "").strip().lower() if resource_role == "project_object" or ( target_scope in {"current_project", "explicit_project"} and business_frame.get("project_binding") not in (None, "", []) ): return {"project"} return set() @staticmethod def _config_target_label(target_ref: str, mention: str | None = None) -> str: normalized = str(mention or "").strip() if normalized: return normalized if str(target_ref or "").endswith("mcp_template_center"): return "MCP 配置" return "模型配置" if str(target_ref or "").endswith("model_routing") else "系统配置" def _build_project_target_from_business_frame( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], ) -> Optional[dict[str, Any]]: business_frame = self._coalesce_business_frame( query_semantics=query_semantics, operation_contract=operation_contract, ) try: project_binding = int(business_frame.get("project_binding")) except Exception: project_binding = None if project_binding is None: return None target_scope = str(business_frame.get("target_scope") or "").strip().lower() resource_role = str(business_frame.get("resource_role") or "").strip().lower() requested_entity_types = self._requested_internal_entity_types( query_semantics=query_semantics, operation_contract=operation_contract, ) if requested_entity_types and requested_entity_types != {"project"}: return None if target_scope not in {"current_project", "explicit_project"} and resource_role != "project_object": return None return { "target_entity_type": "project", "target_ref": f"project:{project_binding}", "resolved_label": str(business_frame.get("subject_binding") or "").strip() or None, "resolution_mode": "semantic_filter", "confidence": 0.84, "candidate_summary": [], } def _build_generic_domain_target_from_business_frame( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], binding_source: str = "semantic_filter", confidence: float = 0.74, resolved_label_override: str | None = None, ) -> Optional[dict[str, Any]]: business_frame = self._coalesce_business_frame( query_semantics=query_semantics, operation_contract=operation_contract, ) requested_entity_types = self._requested_internal_entity_types( query_semantics=query_semantics, operation_contract=operation_contract, ) if len(requested_entity_types) != 1: return None entity_type = next(iter(requested_entity_types)) if entity_type in {"project", "config"}: return None catalog_row = internal_domain_target(entity_type) if not isinstance(catalog_row, dict): return None object_binding = str(business_frame.get("object_binding") or "").strip() mentions = list((operation_contract or {}).get("object_mentions") or []) resolved_label = ( str(resolved_label_override or "").strip() or str(business_frame.get("subject_binding") or "").strip() or (str(mentions[0]).strip() if mentions else "") or str(catalog_row.get("label") or "").strip() or None ) normalized_object_binding = object_binding.lower() target_ref = str(catalog_row.get("target_ref") or "").strip() if entity_type == "feature" and normalized_object_binding.startswith("feature:"): target_ref = object_binding elif entity_type == "task" and normalized_object_binding.startswith("task:"): target_ref = object_binding elif entity_type == "signal" and normalized_object_binding.startswith("signal:"): target_ref = object_binding elif entity_type == "doc" and normalized_object_binding.startswith(("doc:", "document:")): target_ref = object_binding elif entity_type == "user" and normalized_object_binding.startswith(("user:", "person:")): target_ref = object_binding return { "target_entity_type": entity_type, "target_ref": target_ref or None, "resolved_label": resolved_label, "resolution_mode": "semantic_filter", "binding_source": binding_source, "confidence": round(max(0.0, min(1.0, float(confidence))), 4), "candidate_summary": [], } @classmethod def _normalize_target_candidate(cls, payload: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]: row = payload if isinstance(payload, dict) else {} entity_type = cls._normalize_target_entity_type_name( row.get("target_entity_type") or row.get("entity_type") ) if entity_type not in cls._TARGET_ENTITY_TYPES: return None score = max(0.0, min(1.0, float(row.get("score") or 0.0))) return { "target_entity_type": entity_type, "target_ref": str(row.get("target_ref") or "").strip() or None, "resolved_label": str(row.get("resolved_label") or row.get("display_name") or "").strip() or None, "score": round(score, 4), "source_kind": str(row.get("source_kind") or "unknown").strip().lower() or "unknown", "match_label": str(row.get("match_label") or "").strip() or None, "source_confidence": round(max(0.0, min(1.0, float(row.get("source_confidence") or score or 0.0))), 4), } @staticmethod def _target_candidate_identity(row: dict[str, Any]) -> str: return "|".join( [ str(row.get("target_entity_type") or ""), str(row.get("target_ref") or ""), str(row.get("resolved_label") or ""), ] ) async def _collect_semantic_fact_target_candidates( self, *, mentions: list[str], project_id: Optional[int], operation_contract: dict[str, Any], ) -> list[dict[str, Any]]: if not mentions: return [] candidates: list[dict[str, Any]] = [] fact_query = getattr(self, "_query_project_semantic_facts_by_lookup_terms", None) if not callable(fact_query): return [] for mention in mentions[:3]: fact_rows, _used_keyword = await fact_query( query_terms=[mention], project_id=project_id, limit=10, ) project_ids: list[int] = [] fact_confidence_by_project: dict[int, float] = {} for row in fact_rows: try: entity_id = int(row.entity_id) except Exception: continue if entity_id not in project_ids: project_ids.append(entity_id) fact_confidence_by_project[entity_id] = max( float(fact_confidence_by_project.get(entity_id) or 0.0), float(getattr(row, "confidence", 0.0) or 0.0), ) if not project_ids: continue project_rows = ( await self.db.execute( select(Project).where( and_( Project.tenant_id == self.tenant_id, Project.id.in_(project_ids), ) ) ) ).scalars().all() for project in project_rows: confidence = max(0.88, min(0.99, float(fact_confidence_by_project.get(int(project.id)) or 0.0))) candidates.append( { "target_entity_type": "project", "target_ref": f"project:{int(project.id)}", "resolved_label": str(project.title or project.name or f"项目#{int(project.id)}"), "score": round(min(0.99, confidence + 0.03), 4), "source_kind": "semantic_facts", "match_label": mention, "source_confidence": round(confidence, 4), } ) return candidates async def _collect_project_target_candidates( self, *, mentions: list[str], project_id: Optional[int], ) -> list[dict[str, Any]]: if not mentions: return [] candidates: list[dict[str, Any]] = [] similarity_scorer = getattr(self, "_lookup_similarity_score", None) for mention in mentions[:3]: matched_rows: list[Project] = [] searcher = getattr(self, "_search_projects_by_keyword", None) if callable(searcher): matched_rows = await searcher(keyword=mention, project_id=project_id, limit=6) else: stmt = select(Project).where(Project.tenant_id == self.tenant_id) if project_id is not None: stmt = stmt.where(Project.id == int(project_id)) like = f"%{mention}%" stmt = stmt.where( or_( Project.title.ilike(like), Project.name.ilike(like), Project.description.ilike(like), ) ) matched_rows = (await self.db.execute(stmt.order_by(Project.updated_at.desc()).limit(6))).scalars().all() if not matched_rows: stmt = select(Project).where(Project.tenant_id == self.tenant_id) if project_id is not None: stmt = stmt.where(Project.id == int(project_id)) project_rows = ( await self.db.execute( stmt.order_by(Project.updated_at.desc(), Project.id.desc()).limit(120) ) ).scalars().all() scored_rows: list[tuple[float, Project]] = [] for project in project_rows: title = str(project.title or "").strip() name = str(project.name or "").strip() description = str(project.description or "").strip() if callable(similarity_scorer): score = max( similarity_scorer(mention, title), similarity_scorer(mention, name), similarity_scorer(mention, description) * 0.8, ) else: lowered = mention.lower() title_l = title.lower() name_l = name.lower() description_l = description.lower() score = 0.0 if lowered and (lowered in title_l or lowered in name_l): score = 0.9 elif lowered and lowered in description_l: score = 0.76 if score >= 0.62: scored_rows.append((score, project)) scored_rows.sort(key=lambda item: item[0], reverse=True) matched_rows = [row for _, row in scored_rows[:6]] for project in matched_rows: title = str(project.title or "").strip().lower() name = str(project.name or "").strip().lower() description = str(project.description or "").strip().lower() lowered = mention.lower() score = 0.82 if callable(similarity_scorer): score = max( 0.74, similarity_scorer(mention, title) if title else 0.0, similarity_scorer(mention, name) if name else 0.0, (similarity_scorer(mention, description) * 0.8) if description else 0.0, ) elif lowered and (lowered in title or lowered in name): score = 0.91 elif lowered and lowered in description: score = 0.86 candidates.append( { "target_entity_type": "project", "target_ref": f"project:{int(project.id)}", "resolved_label": str(project.title or project.name or f"项目#{int(project.id)}"), "score": round(score, 4), "source_kind": "project_db", "match_label": mention, "source_confidence": round(score, 4), } ) return candidates def _collect_feature_target_candidates( self, *, question: str, mentions: list[str], system_snapshot: Optional[dict[str, Any]], ) -> list[dict[str, Any]]: matcher = getattr(self, "_match_feature_catalog_entries", None) if not callable(matcher): return [] search_text = " ".join(mentions[:2]) if mentions else question rows = matcher( question=search_text, system_snapshot=system_snapshot, limit=4, ) candidates: list[dict[str, Any]] = [] for row in rows: name = str(row.get("name") or "").strip() route = str(row.get("route") or "").strip() match_score = max(0.0, min(1.0, float(row.get("match_score") or 0.0))) if not name and not route: continue score = min(0.92, max(0.72, 0.62 + match_score * 0.25)) target_key = route or name candidates.append( { "target_entity_type": "feature", "target_ref": f"feature:{target_key}", "resolved_label": name or target_key, "score": round(score, 4), "source_kind": "feature_catalog", "match_label": name or target_key, "source_confidence": round(match_score or score, 4), } ) return candidates def _collect_config_target_candidates( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: dict[str, Any], ) -> list[dict[str, Any]]: target_ref = self._config_target_ref_from_business_frame( query_semantics=query_semantics, operation_contract=operation_contract, ) requested_config_domain = self._config_domain_requested( query_semantics=query_semantics, operation_contract=operation_contract, ) if not target_ref and not requested_config_domain: return [] mentions = list(operation_contract.get("object_mentions") or []) rows: list[dict[str, Any]] = [] if target_ref: rows.append( { "target_entity_type": "config", "target_ref": target_ref, "resolved_label": self._config_target_label(target_ref, mentions[0] if mentions else None), "score": 0.9 if target_ref.endswith("model_routing") else 0.88 if target_ref.endswith("mcp_template_center") else 0.86, "source_kind": "system_config", "match_label": mentions[0] if mentions else self._config_target_label(target_ref), "source_confidence": 0.9 if target_ref.endswith("model_routing") else 0.88 if target_ref.endswith("mcp_template_center") else 0.86, } ) return rows for item in config_target_catalog(): candidate_ref = str(item.get("target_ref") or "").strip() if not candidate_ref: continue label = str(item.get("label") or "").strip() or self._config_target_label(candidate_ref) rows.append( { "target_entity_type": "config", "target_ref": candidate_ref, "resolved_label": label, "score": 0.88 if candidate_ref.endswith("model_routing") else 0.87, "source_kind": "system_config", "match_label": mentions[0] if mentions else label, "source_confidence": 0.88 if candidate_ref.endswith("model_routing") else 0.87, } ) return rows async def _collect_kg_target_candidates( self, *, question: str, mentions: list[str], project_id: Optional[int], ) -> list[dict[str, Any]]: resolver = getattr(self, "entity_resolution", None) if not callable(resolver): return [] resolution = await resolver( question=" ".join(mentions[:2]) if mentions else question, project_id=project_id, limit=6, ) matched_entities = resolution.get("matched_entities") if isinstance(resolution, dict) else [] if not isinstance(matched_entities, list) or not matched_entities: return [] ids: list[int] = [] matched_score: dict[int, float] = {} for row in matched_entities: if not isinstance(row, dict): continue try: entity_id = int(row.get("entity_id")) except Exception: continue ids.append(entity_id) matched_score[entity_id] = max( float(matched_score.get(entity_id) or 0.0), float(row.get("score") or 0.0), ) if not ids: return [] rows = ( await self.db.execute( select(KGEntity).where( and_( KGEntity.tenant_id == self.tenant_id, KGEntity.id.in_(ids), ) ) ) ).scalars().all() candidates: list[dict[str, Any]] = [] for row in rows: payload = self._safe_dict(row.payload) if hasattr(self, "_safe_dict") else (row.payload if isinstance(row.payload, dict) else {}) entity_score = max(0.0, min(1.0, float(matched_score.get(int(row.id)) or 0.0))) project_ref = payload.get("project_id") if project_ref is not None: try: project_ref = int(project_ref) except Exception: project_ref = None if project_ref is not None: candidates.append( { "target_entity_type": "project", "target_ref": f"project:{int(project_ref)}", "resolved_label": str(row.display_name or row.entity_key or f"项目#{int(project_ref)}"), "score": round(min(0.89, max(0.68, 0.58 + entity_score * 0.25)), 4), "source_kind": "knowledge_graph", "match_label": str(row.display_name or row.entity_key or "").strip() or None, "source_confidence": round(entity_score or 0.68, 4), } ) return candidates async def _collect_internal_target_candidates( self, *, question: str, project_id: Optional[int], query_semantics: Optional[dict[str, Any]], system_snapshot: Optional[dict[str, Any]], operation_contract: dict[str, Any], ) -> list[dict[str, Any]]: _ = query_semantics mentions = list(operation_contract.get("object_mentions") or []) providers = await asyncio.gather( self._collect_semantic_fact_target_candidates( mentions=mentions, project_id=project_id, operation_contract=operation_contract, ), self._collect_project_target_candidates( mentions=mentions, project_id=project_id, ), self._collect_kg_target_candidates( question=question, mentions=mentions, project_id=project_id, ), ) rows: list[dict[str, Any]] = [] for bucket in providers: for item in bucket: normalized = self._normalize_target_candidate(item) if normalized is not None: rows.append(normalized) for item in self._collect_feature_target_candidates( question=question, mentions=mentions, system_snapshot=system_snapshot, ): normalized = self._normalize_target_candidate(item) if normalized is not None: rows.append(normalized) for item in self._collect_config_target_candidates( query_semantics=query_semantics, operation_contract=operation_contract, ): normalized = self._normalize_target_candidate(item) if normalized is not None: rows.append(normalized) requested_entity_types = self._requested_internal_entity_types( query_semantics=query_semantics, operation_contract=operation_contract, ) deduped: dict[str, dict[str, Any]] = {} for row in rows: key = self._target_candidate_identity(row) existing = deduped.get(key) if existing is None or float(row.get("score") or 0.0) > float(existing.get("score") or 0.0): deduped[key] = row filtered = list(deduped.values()) if requested_entity_types: filtered = [ row for row in filtered if self._normalize_target_entity_type_name(row.get("target_entity_type")) in requested_entity_types ] return sorted(filtered, key=lambda item: float(item.get("score") or 0.0), reverse=True)[:8] async def _rerank_internal_target_candidates( self, *, question: str, operation_contract: dict[str, Any], candidates: list[dict[str, Any]], ) -> Optional[dict[str, Any]]: if len(candidates) < 2 or not hasattr(self, "llm"): return None prompt = PROMPT_INTERNAL_TARGET_RESOLUTION.format( question=str(question or "")[:600], operation_contract=json.dumps(operation_contract, ensure_ascii=False)[:1000], candidates=json.dumps( [ { "index": idx, "target_entity_type": str(item.get("target_entity_type") or ""), "target_ref": str(item.get("target_ref") or ""), "resolved_label": str(item.get("resolved_label") or ""), "score": float(item.get("score") or 0.0), "source_kind": str(item.get("source_kind") or ""), } for idx, item in enumerate(candidates[:4]) ], ensure_ascii=False, ), ).strip() try: raw = await self.llm.generate( prompt, db=self.db, tenant_id=self.tenant_id, function_name="internal_target_resolution", temperature=0.0, max_tokens=220, use_prompt_profile=False, ) except Exception: return None parser = getattr(self, "_parse_json_object_safe", None) parsed = parser(raw) if callable(parser) else {} if not isinstance(parsed, dict): return None if bool(parsed.get("clarify")): return { "target_entity_type": "unknown", "target_ref": None, "resolved_label": None, "resolution_mode": "clarify", "confidence": round(max(0.0, min(1.0, float(parsed.get("confidence") or 0.0))), 4), "candidate_summary": candidates[:4], } try: selected_index = int(parsed.get("selected_index")) except Exception: return None if selected_index < 0 or selected_index >= len(candidates[:4]): return None selected = dict(candidates[selected_index]) return { "target_entity_type": str(selected.get("target_entity_type") or "unknown"), "target_ref": selected.get("target_ref"), "resolved_label": selected.get("resolved_label"), "resolution_mode": "llm_rerank", "confidence": round(max(0.0, min(1.0, float(parsed.get("confidence") or selected.get("score") or 0.0))), 4), "candidate_summary": candidates[:4], } def _build_semantic_filter_target( self, *, query_semantics: Optional[dict[str, Any]], operation_contract: dict[str, Any], ) -> Optional[dict[str, Any]]: semantics = query_semantics if isinstance(query_semantics, dict) else {} business_frame = self._coalesce_business_frame( query_semantics=semantics, operation_contract=operation_contract, ) project_bound = self._build_project_target_from_business_frame( query_semantics=semantics, operation_contract=operation_contract, ) if project_bound is not None: return project_bound config_ref = self._config_target_ref_from_business_frame( query_semantics=semantics, operation_contract=operation_contract, ) if config_ref: mentions = list(operation_contract.get("object_mentions") or []) return { "target_entity_type": "config", "target_ref": config_ref, "resolved_label": self._config_target_label(config_ref, mentions[0] if mentions else None), "resolution_mode": "deterministic", "confidence": 0.84, "candidate_summary": [], } mentions = list(operation_contract.get("object_mentions") or []) resolved_label = ( str(business_frame.get("subject_binding") or "").strip() or (mentions[0] if mentions else None) ) generic_domain = self._build_generic_domain_target_from_business_frame( query_semantics=semantics, operation_contract=operation_contract, confidence=0.74 if resolved_label else 0.62, resolved_label_override=resolved_label, ) if generic_domain is not None: return generic_domain target_scope = str(business_frame.get("target_scope") or "").strip().lower() resource_role = str(business_frame.get("resource_role") or "").strip().lower() if target_scope in {"current_project", "explicit_project"} or resource_role == "project_object": return { "target_entity_type": "project", "target_ref": None, "resolved_label": resolved_label, "resolution_mode": "semantic_filter", "confidence": 0.78 if resolved_label else 0.64, "candidate_summary": [], } return None async def _resolve_internal_target( self, *, question: str, project_id: Optional[int], query_semantics: Optional[dict[str, Any]], system_snapshot: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], ) -> dict[str, Any]: semantics = query_semantics if isinstance(query_semantics, dict) else {} operation = operation_contract if isinstance(operation_contract, dict) else {} source_policy = str( semantics.get("effective_source_policy") or semantics.get("source_policy") or "" ).strip().lower() if source_policy != "internal": return { "target_entity_type": "unknown", "target_ref": None, "resolved_label": None, "resolution_mode": "none", "confidence": 0.0, "candidate_summary": [], } session_bound_target = self._build_session_followup_target( query_semantics=semantics, operation_contract=operation, ) if session_bound_target is not None: return session_bound_target candidates = await self._collect_internal_target_candidates( question=question, project_id=project_id, query_semantics=semantics, system_snapshot=system_snapshot, operation_contract=operation, ) if not candidates: fallback = self._build_semantic_filter_target( query_semantics=semantics, operation_contract=operation, ) if fallback is not None: return fallback return { "target_entity_type": "unknown", "target_ref": None, "resolved_label": None, "resolution_mode": "clarify", "confidence": 0.0, "candidate_summary": [], } top = candidates[0] second = candidates[1] if len(candidates) > 1 else None margin = float(top.get("score") or 0.0) - float(second.get("score") or 0.0) if second else 1.0 if second and str(top.get("target_entity_type") or "") != str(second.get("target_entity_type") or "") and margin < 0.1: reranked = await self._rerank_internal_target_candidates( question=question, operation_contract=operation, candidates=candidates, ) if reranked is not None: return reranked if float(top.get("score") or 0.0) >= 0.88 or margin >= 0.12 or second is None: if ( second is not None and str(top.get("target_entity_type") or "").strip().lower() == "config" and not self._config_target_ref_from_business_frame( query_semantics=semantics, operation_contract=operation, ) ): return { "target_entity_type": "unknown", "target_ref": None, "resolved_label": None, "resolution_mode": "clarify", "confidence": round(max(0.0, min(1.0, float(top.get("score") or 0.0))), 4), "candidate_summary": candidates[:4], } return { "target_entity_type": str(top.get("target_entity_type") or "unknown"), "target_ref": top.get("target_ref"), "resolved_label": top.get("resolved_label"), "resolution_mode": "deterministic", "confidence": round(max(0.0, min(1.0, float(top.get("score") or 0.0))), 4), "candidate_summary": candidates[:4], } if str(top.get("target_entity_type") or "") == str(second.get("target_entity_type") or ""): return { "target_entity_type": str(top.get("target_entity_type") or "unknown"), "target_ref": top.get("target_ref"), "resolved_label": top.get("resolved_label"), "resolution_mode": "deterministic", "confidence": round(max(0.0, min(1.0, float(top.get("score") or 0.0))), 4), "candidate_summary": candidates[:4], } return { "target_entity_type": "unknown", "target_ref": None, "resolved_label": None, "resolution_mode": "clarify", "confidence": round(max(0.0, min(1.0, float(top.get("score") or 0.0))), 4), "candidate_summary": candidates[:4], } def _patch_scene_semantics_with_target_resolution( self, *, scene_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], target_resolution: Optional[dict[str, Any]], ) -> dict[str, Any]: normalized = dict(scene_semantics or {}) query_type = str((operation_contract or {}).get("query_type") or QUERY_TYPE_UNKNOWN).strip().lower() target_entity_type = str((target_resolution or {}).get("target_entity_type") or "unknown").strip().lower() snapshot_focus = str(normalized.get("snapshot_focus") or "none").strip().lower() if query_type == QUERY_TYPE_EXISTENCE and target_entity_type == "project": normalized["snapshot_focus"] = "none" return normalized if target_entity_type not in {"feature", "config"} and snapshot_focus == "feature": normalized["snapshot_focus"] = "none" return normalized return normalized def _should_allow_snapshot_direct_answer( self, *, scene_semantics: Optional[dict[str, Any]], operation_contract: Optional[dict[str, Any]], target_resolution: Optional[dict[str, Any]], ) -> bool: snapshot_focus = str((scene_semantics or {}).get("snapshot_focus") or "none").strip().lower() query_type = str((operation_contract or {}).get("query_type") or QUERY_TYPE_UNKNOWN).strip().lower() target_entity_type = str((target_resolution or {}).get("target_entity_type") or "unknown").strip().lower() if snapshot_focus in {"identity", "architecture"}: return query_type in {QUERY_TYPE_EXPLAIN, QUERY_TYPE_STATUS} if snapshot_focus == "feature": return query_type in {QUERY_TYPE_EXPLAIN, QUERY_TYPE_STATUS, QUERY_TYPE_LIST} and target_entity_type in { "feature", "config", "unknown", } return False @staticmethod def _build_operation_contract_summary(operation_contract: Optional[dict[str, Any]]) -> dict[str, Any]: payload = operation_contract if isinstance(operation_contract, dict) else {} mentions = [ str(item or "").strip() for item in list(payload.get("object_mentions") or []) if str(item or "").strip() ] return { "query_type": str(payload.get("query_type") or QUERY_TYPE_UNKNOWN).strip().lower() or QUERY_TYPE_UNKNOWN, "answer_contract": str(payload.get("answer_contract") or "clarify_or_fail").strip() or "clarify_or_fail", "object_mentions": mentions[:4], "clarify_policy": str(payload.get("clarify_policy") or "").strip() or None, } @staticmethod def _build_target_resolution_summary(target_resolution: Optional[dict[str, Any]]) -> dict[str, Any]: payload = target_resolution if isinstance(target_resolution, dict) else {} candidates: list[dict[str, Any]] = [] for item in list(payload.get("candidate_summary") or [])[:4]: if not isinstance(item, dict): continue candidates.append( { "target_entity_type": str(item.get("target_entity_type") or "unknown").strip().lower() or "unknown", "target_ref": str(item.get("target_ref") or "").strip() or None, "resolved_label": str(item.get("resolved_label") or "").strip() or None, "score": round(max(0.0, min(1.0, float(item.get("score") or 0.0))), 4), "source_kind": str(item.get("source_kind") or "unknown").strip().lower() or "unknown", } ) return { "target_entity_type": str(payload.get("target_entity_type") or "unknown").strip().lower() or "unknown", "target_ref": str(payload.get("target_ref") or "").strip() or None, "resolved_label": str(payload.get("resolved_label") or "").strip() or None, "resolution_mode": str(payload.get("resolution_mode") or "none").strip().lower() or "none", "confidence": round(max(0.0, min(1.0, float(payload.get("confidence") or 0.0))), 4), "candidate_summary": candidates, } @staticmethod def _build_query_plan_summary(query_plan: Optional[dict[str, Any]]) -> dict[str, Any]: payload = query_plan if isinstance(query_plan, dict) else {} filters: list[dict[str, Any]] = [] for item in list(payload.get("filters") or [])[:4]: if not isinstance(item, dict): continue filters.append( { "field": str(item.get("field") or "").strip() or None, "operator": str(item.get("operator") or "").strip() or None, "value": item.get("value"), "value_source": str(item.get("value_source") or "").strip() or None, } ) return { "plan_version": str(payload.get("plan_version") or "unknown").strip() or "unknown", "query_type": str(payload.get("query_type") or QUERY_TYPE_UNKNOWN).strip().lower() or QUERY_TYPE_UNKNOWN, "entity_type": str(payload.get("entity_type") or "unknown").strip().lower() or "unknown", "target_ref": str(payload.get("target_ref") or "").strip() or None, "answer_contract": str(payload.get("answer_contract") or "").strip() or None, "filters": filters, } def _inject_internal_query_runtime_summaries( self, *, retrieval_metrics: Optional[dict[str, Any]], query_semantics: Optional[dict[str, Any]], internal_context_packet: Any = None, ) -> dict[str, Any]: metrics = dict(retrieval_metrics or {}) semantics = query_semantics if isinstance(query_semantics, dict) else {} metrics["operation_contract_summary"] = self._build_operation_contract_summary( internal_context_operation_contract( packet=internal_context_packet, fallback=semantics.get("operation_contract"), ) ) metrics["target_resolution_summary"] = self._build_target_resolution_summary( internal_context_target_resolution( packet=internal_context_packet, fallback=semantics.get("target_resolution"), ) ) metrics["query_plan_summary"] = self._build_query_plan_summary( internal_context_query_plan( packet=internal_context_packet, fallback=semantics.get("query_plan"), ) ) return metrics __all__ = ["AdaptiveRAGTargetResolutionMixin"]