from __future__ import annotations import re from uuid import uuid4 from typing import Any from ...config.ai_prompts import ( PROMPT_EXTERNAL_COMPARE_WITH_SOURCES_QA, PROMPT_EXTERNAL_DIRECT_FACT_QA, PROMPT_EXTERNAL_SIGNAL_SOURCE_RECOMMENDATION_PACKET, PROMPT_EXTERNAL_SIGNAL_SOURCE_RECOMMENDATION_QA, PROMPT_EXTERNAL_SUMMARY_WITH_SOURCES_QA, ) from ..prompt_runtime_service import PromptRuntimeRenderResult, PromptRuntimeService from .schemas import ( ExternalContextPacket, SignalSourceRecommendationCandidate, SignalSourceRecommendationPacket, ) class ExternalAnswerPresenter: _PROMPT_CONFIG = { "direct_fact": ( "PROMPT_EXTERNAL_DIRECT_FACT_QA", PROMPT_EXTERNAL_DIRECT_FACT_QA, "external_direct_fact", ), "summary_with_sources": ( "PROMPT_EXTERNAL_SUMMARY_WITH_SOURCES_QA", PROMPT_EXTERNAL_SUMMARY_WITH_SOURCES_QA, "external_summary_with_sources", ), "compare_with_sources": ( "PROMPT_EXTERNAL_COMPARE_WITH_SOURCES_QA", PROMPT_EXTERNAL_COMPARE_WITH_SOURCES_QA, "external_compare_with_sources", ), } def __init__( self, *, llm_service: Any, db: Any, tenant_id: int, ) -> None: self.llm_service = llm_service self.db = db self.tenant_id = int(tenant_id or 1) @classmethod def _normalize_contract(cls, value: Any) -> str: normalized = str(value or "").strip().lower() if normalized in cls._PROMPT_CONFIG: return normalized return "summary_with_sources" @staticmethod def _build_evidence_blocks(packet: ExternalContextPacket) -> str: blocks: list[str] = [] for idx, item in enumerate(list(packet.verified_evidence or [])[:5], start=1): source_label = str(item.source_label or item.title or item.target_url or "").strip() title = str(item.title or "").strip() snippet = str(item.snippet or "").strip() target_url = str(item.target_url or "").strip() block_lines = [f"[{idx}] 来源: {source_label or '外部来源'}"] if title and title != source_label: block_lines.append(f"标题: {title}") if snippet: block_lines.append(f"证据: {snippet}") if target_url: block_lines.append(f"链接: {target_url}") blocks.append("\n".join(block_lines)) return "\n\n".join(blocks).strip() or "无" @staticmethod def _build_source_summaries(packet: ExternalContextPacket) -> str: labels: list[str] = [] for item in list(packet.verified_evidence or [])[:5]: label = str(item.source_label or item.title or item.target_url or "").strip() if label and label not in labels: labels.append(label) return "；".join(labels[:5]) if labels else "无" @staticmethod def _build_plan_summary(packet: ExternalContextPacket) -> str: plan = packet.plan rows = [ f"主题: {str(plan.topic or '').strip() or '未指定'}", f"回答合同: {str(plan.answer_contract or '').strip() or 'summary_with_sources'}", ] if plan.time_range_hint: rows.append(f"时间范围: {plan.time_range_hint}") if plan.location_hint: rows.append(f"地点: {plan.location_hint}") if plan.subject_entities: rows.append(f"对象: {'、'.join([str(item).strip() for item in plan.subject_entities[:5] if str(item).strip()])}") if plan.signal_source_objectives: rows.append( "任务目标: " + "；".join([str(item).strip() for item in plan.signal_source_objectives[:3] if str(item).strip()]) ) return "\n".join(rows) @classmethod def _is_signal_source_recommendation_request( cls, *, packet: ExternalContextPacket, ) -> bool: interaction_level = str(packet.plan.interaction_level or "").strip().lower() if interaction_level and interaction_level != "recommendation": return False if interaction_level != "recommendation": return False recommendation_goal = str(packet.plan.recommendation_goal or "").strip() return bool(recommendation_goal or packet.plan.project_object_anchors or packet.plan.signal_source_objectives) @staticmethod def _build_recommendation_uncertainty_text(packet: ExternalContextPacket) -> str: retrieval_status = str(packet.retrieval_status or "").strip().lower() if retrieval_status == "verifier_blocked": return "当前外部公开资料还在校验，这里先给推荐，不把它当成已验证事实。" if retrieval_status == "no_relevant_results": return "当前外部公开资料还没找到直接证据，这里先按项目目标给推荐。" if retrieval_status == "retrieval_failed": return "这次外部检索失败，这里先按项目目标给推荐，后面还需要补核验。" return "" @staticmethod def _collect_signal_source_objectives(packet: ExternalContextPacket) -> list[str]: objectives: list[str] = [] for item in list(packet.plan.signal_source_objectives or []): normalized = str(item or "").strip() if normalized and normalized not in objectives: objectives.append(normalized) if objectives: return objectives[:4] for item in list(packet.plan.project_object_anchors or []): normalized = str(item or "").strip() if normalized and normalized not in objectives: objectives.append(normalized) return objectives[:4] @staticmethod def _objective_label(value: Any) -> str: text = str(value or "").strip() if not text: return "" for separator in ("：", ":"): if separator in text: return str(text.split(separator, 1)[0] or "").strip() or text return text @classmethod def _extract_objective_parts(cls, value: Any) -> tuple[str, str]: text = str(value or "").strip() if not text: return "", "" for separator in ("：", ":"): if separator in text: label, detail = text.split(separator, 1) return str(label or "").strip(), str(detail or "").strip() return text, "" @staticmethod def _sanitize_recommendation_candidate_text(value: Any) -> str: text = re.sub(r"\s+", " ", str(value or "")).strip(" ，,。；;：:") text = re.sub(r"^(?:当前|建议|推荐)\s*", "", text) return text[:36] @classmethod def _objective_candidate_name(cls, *, label: str, detail: str) -> str: candidate = detail or label candidate = re.sub(r"^(?:编写脚本|收集|跟踪|监控|处理|实现|支持|自动|实时|用于|把|对|完成|生成并推送)\s*", "", candidate) candidate = re.sub(r"^(?:收到|基于|围绕|用于)\s*", "", candidate) candidate = re.sub(r"(?:后自动|并在.+)$", "", candidate).strip() candidate = cls._sanitize_recommendation_candidate_text(candidate) if candidate and candidate not in {"数据与阈值", "结果发给客户", "话题波动"}: return candidate return cls._sanitize_recommendation_candidate_text(label) or "外部变化线索" @classmethod def _objective_candidate_reason(cls, *, label: str, detail: str) -> str: detail_text = cls._sanitize_recommendation_candidate_text(detail) label_text = cls._sanitize_recommendation_candidate_text(label) if detail_text and label_text: return f"它最早支撑“{label_text}”，也更容易先看出这件事有没有开始偏。" if label_text: return f"它能更早暴露“{label_text}”是不是开始出问题。" return "它更容易先把问题从项目现状里提前暴露出来。" @classmethod def _build_signal_source_recommendation_fallback_packet( cls, *, packet: ExternalContextPacket, ) -> SignalSourceRecommendationPacket: candidates: list[SignalSourceRecommendationCandidate] = [] for raw in cls._collect_signal_source_objectives(packet)[:3]: label, detail = cls._extract_objective_parts(raw) name = cls._objective_candidate_name(label=label, detail=detail) reason = cls._objective_candidate_reason(label=label, detail=detail) if name and all(existing.name != name for existing in candidates): candidates.append( SignalSourceRecommendationCandidate( candidate_ref=f"cand_{uuid4().hex[:12]}", name=name, display_name=name, reason=reason, business_value=reason, why_now=reason, source_origin="recommendation", materialization_payload={ "display_name": name, "source_origin": "recommendation", }, ) ) fallback_seed = [ ("最近异常反馈", "它能看出问题是不是已经开始影响到真实结果，而不是只停在内部波动。"), ("上游变化记录", "它更容易解释这次波动到底是内部原因还是外部变化带出来的。"), ("结果失败回执", "它能最早把真正失败的位置暴露出来，避免只看到表面症状。"), ] for name, reason in fallback_seed: if len(candidates) >= 3: break if all(existing.name != name for existing in candidates): candidates.append( SignalSourceRecommendationCandidate( candidate_ref=f"cand_{uuid4().hex[:12]}", name=name, display_name=name, reason=reason, business_value=reason, why_now=reason, source_origin="recommendation", materialization_payload={ "display_name": name, "source_origin": "recommendation", }, ) ) return SignalSourceRecommendationPacket( candidate_set_id=f"recset_{uuid4().hex[:12]}", resource_role="signal_source", source_kind="external", evidence_mode="verified" if str(packet.retrieval_status or "").strip().lower() == "evidence_found" else "advisory", candidates=candidates[:3], caveat=cls._build_recommendation_uncertainty_text(packet) or "这些先算建议，不代表已经接入或验证完成。", ) @staticmethod def _ensure_recommendation_packet_defaults( packet: SignalSourceRecommendationPacket, *, fallback_packet: SignalSourceRecommendationPacket, ) -> SignalSourceRecommendationPacket: if not str(packet.candidate_set_id or "").strip(): packet.candidate_set_id = str(fallback_packet.candidate_set_id or "").strip() or f"recset_{uuid4().hex[:12]}" packet.resource_role = "signal_source" packet.source_kind = str(packet.source_kind or fallback_packet.source_kind or "external").strip() or "external" packet.evidence_mode = ( str(packet.evidence_mode or fallback_packet.evidence_mode or "advisory").strip() or "advisory" ) for item in list(packet.candidates or []): if not str(item.candidate_ref or "").strip(): item.candidate_ref = f"cand_{uuid4().hex[:12]}" payload = dict(item.materialization_payload or {}) payload.setdefault("display_name", str(item.display_name or item.name or "").strip()) payload.setdefault("candidate_ref", str(item.candidate_ref or "").strip()) payload.setdefault("source_origin", str(item.source_origin or "recommendation").strip() or "recommendation") item.materialization_payload = payload return packet @classmethod def _recommendation_packet_has_risky_or_weak_content( cls, *, recommendation_packet: SignalSourceRecommendationPacket, packet: ExternalContextPacket, ) -> bool: if len(recommendation_packet.candidates) < 2: return True risky_markers = ( "%", "已确认", "无需额外对接验证", "相关性达", "普遍可采集", "近一年", "今年", "本月", "这一层", "主链", "packet", "route", "verifier", ) for item in list(recommendation_packet.candidates or []): text = f"{str(item.name or '')} {str(item.reason or '')}".strip() if not text: return True if any(marker in text for marker in risky_markers): return True if str(packet.impact_status or "").strip().lower() == "direct_match": return False return False @classmethod def _merge_recommendation_packets( cls, *, primary: SignalSourceRecommendationPacket, fallback: SignalSourceRecommendationPacket, ) -> SignalSourceRecommendationPacket: merged: list[SignalSourceRecommendationCandidate] = [] for item in list(primary.candidates or []) + list(fallback.candidates or []): name = cls._sanitize_recommendation_candidate_text(item.display_name or item.name) reason = str(item.reason or item.business_value or item.why_now or "").strip() if not name or not reason: continue if any(existing.display_name == name for existing in merged): continue merged.append( SignalSourceRecommendationCandidate( candidate_ref=str(item.candidate_ref or "").strip() or f"cand_{uuid4().hex[:12]}", name=name, display_name=name, reason=reason, business_value=str(item.business_value or reason).strip(), why_now=str(item.why_now or reason).strip(), source_origin=str(item.source_origin or "recommendation").strip() or "recommendation", materialization_payload=dict(item.materialization_payload or {}), ) ) caveat = str(primary.caveat or "").strip() or str(fallback.caveat or "").strip() packet = SignalSourceRecommendationPacket( candidate_set_id=str(primary.candidate_set_id or fallback.candidate_set_id or "").strip() or f"recset_{uuid4().hex[:12]}", project_binding=primary.project_binding or fallback.project_binding, resource_role="signal_source", source_kind=str(primary.source_kind or fallback.source_kind or "external").strip() or "external", evidence_mode=str(primary.evidence_mode or fallback.evidence_mode or "advisory").strip() or "advisory", candidates=merged[:3], caveat=caveat, ) return cls._ensure_recommendation_packet_defaults(packet, fallback_packet=fallback) @classmethod def _render_signal_source_recommendation_text( cls, *, recommendation_packet: SignalSourceRecommendationPacket, packet: ExternalContextPacket, ) -> str: candidates = list(recommendation_packet.candidates or [])[:3] if not candidates: recommendation_packet = cls._build_signal_source_recommendation_fallback_packet(packet=packet) candidates = list(recommendation_packet.candidates or [])[:3] lines = ["如果只是先给建议，我会先看这 3 类外部信号："] for index, item in enumerate(candidates, start=1): name = cls._sanitize_recommendation_candidate_text(item.name) reason = str(item.reason or "").strip().rstrip("。") if reason.startswith("因为"): lines.append(f"{index}. {name}，{reason}") else: lines.append(f"{index}. {name}，因为{reason}") caveat = str(recommendation_packet.caveat or "").strip() or cls._build_recommendation_uncertainty_text(packet) if caveat: lines.append(caveat.rstrip("。") + "。") return "\n".join(lines) async def _build_signal_source_recommendation_packet( self, *, question: str, packet: ExternalContextPacket, ) -> SignalSourceRecommendationPacket: fallback_packet = self._build_signal_source_recommendation_fallback_packet(packet=packet) generator = getattr(self.llm_service, "generate_structured_object", None) if not callable(generator): return fallback_packet plan_summary = self._build_plan_summary(packet) fit_brief_summary = str(packet.plan.fit_brief_summary or "").strip() or "未提供" recommendation_goal = str(packet.plan.recommendation_goal or "").strip() or "给当前项目推荐最值得先关注的外部信号源。" project_object_anchors = "、".join( [str(item).strip() for item in list(packet.plan.project_object_anchors or []) if str(item).strip()][:4] ) or "无" signal_source_objectives = "；".join( [str(item).strip() for item in list(packet.plan.signal_source_objectives or []) if str(item).strip()][:4] ) or "无" prompt_result = await self._render_prompt_result( prompt_key="PROMPT_EXTERNAL_SIGNAL_SOURCE_RECOMMENDATION_PACKET", fallback_template=PROMPT_EXTERNAL_SIGNAL_SOURCE_RECOMMENDATION_PACKET, question=str(question or "").strip(), plan_summary=plan_summary, fit_brief_summary=fit_brief_summary, recommendation_goal=recommendation_goal, project_object_anchors=project_object_anchors, signal_source_objectives=signal_source_objectives, retrieval_status=str(packet.retrieval_status or "").strip() or "unknown", uncertainties="；".join([str(item).strip() for item in list(packet.uncertainties or []) if str(item).strip()]) or "无", source_summaries=self._build_source_summaries(packet), ) prompt = prompt_result.prompt try: structured = await generator( prompt=prompt, db=self.db, tenant_id=self.tenant_id, function_name="external_signal_source_recommendation_packet", use_prompt_profile=False, temperature=0.1, response_mode="fast", routing_meta={ "tier_alias": "low", "response_contract": "structured", "fallback_policy": "fast_local", "execution_mode": "query", **prompt_result.routing_meta(), }, ) payload = getattr(structured, "payload", None) if payload is None and hasattr(structured, "model_dump"): payload = structured.model_dump() recommendation_packet = SignalSourceRecommendationPacket.model_validate(payload or {}) recommendation_packet = self._merge_recommendation_packets( primary=recommendation_packet, fallback=fallback_packet, ) recommendation_packet = self._ensure_recommendation_packet_defaults( recommendation_packet, fallback_packet=fallback_packet, ) if self._recommendation_packet_has_risky_or_weak_content( recommendation_packet=recommendation_packet, packet=packet, ): return fallback_packet return recommendation_packet except Exception: return fallback_packet async def _build_signal_source_recommendation_response( self, *, question: str, packet: ExternalContextPacket, thinking_mode: str, answer_max_tokens: int | None = None, ) -> dict[str, str]: _ = (thinking_mode, answer_max_tokens) recommendation_packet = await self._build_signal_source_recommendation_packet( question=question, packet=packet, ) text = self._render_signal_source_recommendation_text( recommendation_packet=recommendation_packet, packet=packet, ) return { "answer_text": text, "prompt_variant_id": "external_signal_source_recommendation", "recommendation_packet": recommendation_packet.model_dump(), } @staticmethod def _build_failure_response(*, question: str, retrieval_status: str) -> dict[str, str]: short_question = str(question or "").strip()[:80] if retrieval_status == "verifier_blocked": return { "answer_text": ( f"关于“{short_question}”，我拿到过外部候选结果，但它们没有通过核验，" "所以先不给你直接结论。你如果愿意，我可以继续按更明确的对象或范围再检一次。" ), "prompt_variant_id": "external_verifier_blocked", } if retrieval_status == "no_relevant_results": return { "answer_text": ( f"关于“{short_question}”，外部检索已经执行完成，但没有找到足够相关的外部公开证据。" "现阶段我不建议把它直接当成当前项目的有效外部变化。" ), "prompt_variant_id": "external_no_relevant_results", } return { "answer_text": ( f"关于“{short_question}”，外部检索这次执行失败，没有产出可用的外部公开证据。" "你可以稍后重试，或把要核验的公开对象说得更具体一些。" ), "prompt_variant_id": "external_retrieval_failed", } @classmethod def _build_indirect_only_response(cls, *, question: str, packet: ExternalContextPacket) -> dict[str, str]: short_question = str(question or "").strip()[:80] source_summaries = cls._build_source_summaries(packet) return { "answer_text": ( f"关于“{short_question}”，当前外部证据只能说明它和当前项目有背景相关性，" "不能直接说明它已经命中当前项目对象。" f"现有来源：{source_summaries}。如果你要，我可以继续把它整理成一份映射验证清单。" ), "prompt_variant_id": "external_indirect_only", } async def _render_prompt( self, *, prompt_key: str, fallback_template: str, **kwargs: Any, ) -> str: return ( await self._render_prompt_result( prompt_key=prompt_key, fallback_template=fallback_template, **kwargs, ) ).prompt async def _render_prompt_result( self, *, prompt_key: str, fallback_template: str, **kwargs: Any, ) -> PromptRuntimeRenderResult: if not hasattr(self.db, "execute"): return PromptRuntimeRenderResult( prompt=str(fallback_template).format(**kwargs).strip(), prompt_key=prompt_key, source="default_fallback", ) return await PromptRuntimeService(self.db).render( tenant_id=self.tenant_id, prompt_key=prompt_key, fallback_template=fallback_template, **kwargs, ) async def present( self, *, question: str, context_packet: dict[str, Any] | ExternalContextPacket, thinking_mode: str, answer_max_tokens: int | None = None, ) -> dict[str, str]: packet = ( context_packet if isinstance(context_packet, ExternalContextPacket) else ExternalContextPacket.model_validate(context_packet) ) if self._is_signal_source_recommendation_request( packet=packet, ): return await self._build_signal_source_recommendation_response( question=question, packet=packet, thinking_mode=thinking_mode, answer_max_tokens=answer_max_tokens, ) retrieval_status = str(packet.retrieval_status or "").strip().lower() or "retrieval_failed" impact_status = str(packet.impact_status or "").strip().lower() or "no_direct_impact" if retrieval_status != "evidence_found": return self._build_failure_response( question=question, retrieval_status=retrieval_status, ) if impact_status == "indirect_only": return self._build_indirect_only_response( question=question, packet=packet, ) verifier = dict(packet.verifier or {}) if not packet.verified_evidence or not bool(verifier.get("ok")): return { "answer_text": ( f"当前关于“{str(question or '').strip()[:80]}”的外部证据不足。" "请补充更明确的对象、范围或继续提供可核验来源后，我再继续分析。" ), "prompt_variant_id": "external_clarify", } contract = self._normalize_contract(packet.plan.answer_contract) prompt_key, fallback_template, prompt_variant_id = self._PROMPT_CONFIG[contract] prompt_result = await self._render_prompt_result( prompt_key=prompt_key, fallback_template=fallback_template, question=str(question or "").strip(), plan_summary=self._build_plan_summary(packet), evidence_blocks=self._build_evidence_blocks(packet), source_summaries=self._build_source_summaries(packet), time_range=str(packet.plan.time_range_hint or "").strip() or "未指定", uncertainties="；".join([str(item).strip() for item in list(packet.uncertainties or []) if str(item).strip()]) or "无", ) prompt = prompt_result.prompt deep_thinking = str(thinking_mode or "").strip().lower() == "deep" if deep_thinking: prompt = ( f"{prompt}\n\n" "补充要求（深入思考模式）：\n" "1. 先给结论，再给证据链；\n" "2. 明确区分已验证事实与推断；\n" "3. 若存在不确定项，给出待确认清单。" ) effective_max_tokens = max(256, min(int(answer_max_tokens or (1400 if deep_thinking else 900)), 3000)) routing_meta = None function_name = "adaptive_rag_answer_deep" if deep_thinking else "adaptive_rag_answer_normal" if deep_thinking: routing_meta = { "tier_alias": "low", "planned_tier": "low", "executed_tier": "low", "response_contract": "complete", "fallback_policy": "low_to_high", "execution_mode": "query", "routing_decision": "force_low_first", **prompt_result.routing_meta(), } else: routing_meta = prompt_result.routing_meta() answer = await self.llm_service.generate( prompt, db=self.db, tenant_id=self.tenant_id, function_name=function_name, temperature=0.15 if deep_thinking else 0.2, max_tokens=effective_max_tokens, use_prompt_profile=False, routing_meta=routing_meta, ) text = str(answer or "").strip() if not text: raise ValueError("external_answer_presenter_empty_response") return { "answer_text": text, "prompt_variant_id": prompt_variant_id, }