from __future__ import annotations from typing import Any from ...config.ai_prompts import ( PROMPT_EXTERNAL_COMPARE_WITH_SOURCES_QA, PROMPT_EXTERNAL_DIRECT_FACT_QA, PROMPT_EXTERNAL_SUMMARY_WITH_SOURCES_QA, ) from ..prompt_release import PromptReleaseService from .schemas import ExternalContextPacket 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()])}") return "\n".join(rows) async def _render_prompt( self, *, prompt_key: str, fallback_template: str, **kwargs: Any, ) -> str: if not hasattr(self.db, "execute"): return str(fallback_template).format(**kwargs).strip() return await PromptReleaseService(self.db).render_prompt( 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) ) 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 = await self._render_prompt( 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 "无", ) 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", } 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, }