"""Application service for deriving Query candidates from one topic-table snapshot.""" from __future__ import annotations import json import re from copy import deepcopy from dataclasses import dataclass from typing import Any, Callable from core.prompts import load_prompt from query_planning.domain import GenerationRequest, GenerationResult, GeneratorKind from query_planning.service import UnifiedQueryGenerationService PROMPT_NAME = "topic_table_query_generation" PROMPT_VERSION = "topic_table_query_v3" ROUTES: dict[str, dict[str, Any]] = { "unique_hook": {"label": "灵感点和特殊组合", "priority": 100}, "goal_execution": {"label": "目标和想达到的效果", "priority": 90}, "key_execution": {"label": "关键镜头和动作要求", "priority": 80}, "cross_dimension": {"label": "三类元素怎么放在一起", "priority": 70}, "account_fit": {"label": "账号原来的内容风格", "priority": 60}, "evidence_gap": {"label": "参考案例哪些地方不能照搬", "priority": 50}, } class TopicTableGenerationError(ValueError): """Raised when a topic snapshot or LLM result cannot become a Query plan.""" @dataclass(frozen=True) class TopicTablePreview: result: GenerationResult source: dict[str, Any] topic: dict[str, Any] llm_output: dict[str, Any] def topic_table_prompt() -> str: return load_prompt(PROMPT_NAME) def _select_topic(payload: dict[str, Any], topic_id: int | None) -> dict[str, Any]: topics = [topic for topic in payload.get("topics") or [] if isinstance(topic, dict)] if topic_id is not None: for topic in topics: if int(topic.get("id") or 0) == topic_id: return topic raise TopicTableGenerationError(f"选题构建中找不到 topic_id={topic_id}") mature = next((topic for topic in topics if topic.get("status") == "mature"), None) if mature is not None: return mature if topics: return topics[0] raise TopicTableGenerationError("选题构建中没有可用 topic") def _source_summary(source: dict[str, Any]) -> dict[str, Any]: return { "derivation_type": source.get("derivation_type"), "source_type": source.get("source_type"), "source_reference_id": source.get("source_reference_id"), "reason": source.get("reason"), "dataset_from": source.get("dataset_from"), } def _compact_snapshot( payload: dict[str, Any], topic: dict[str, Any], ) -> dict[str, Any]: build = payload.get("build") or {} items = [] for item in topic.get("composition_items") or []: if not isinstance(item, dict) or item.get("is_active") is False: continue items.append( { "id": item.get("id"), "item_level": item.get("item_level"), "dimension": item.get("dimension"), "point_type": item.get("point_type"), "element_name": item.get("element_name"), "category_path": item.get("category_path"), "reason": item.get("reason"), "is_active": item.get("is_active"), "sources": [ _source_summary(source) for source in item.get("sources") or [] if isinstance(source, dict) ], } ) return { "topic_build_id": build.get("id"), "topic_id": topic.get("id"), "build": { "demand": build.get("demand"), "demand_constraints": build.get("demand_constraints"), "strategies_config": build.get("strategies_config"), "status": build.get("status"), }, "topic": { "status": topic.get("status"), "result": topic.get("result"), "topic_direction": topic.get("topic_direction"), "points": [ deepcopy(point) for point in topic.get("points") or [] if isinstance(point, dict) and point.get("is_active") is not False ], "composition_items": items, "item_relations": deepcopy(topic.get("item_relations") or []), }, } def build_topic_table_user_prompt( snapshot: dict[str, Any], field_catalog: list[dict[str, Any]], *, target_query_count: int, ) -> str: return ( "请从 field_catalog 中选择真正能产生创作方法搜索词的具体字段。" "每一项 field_mapping 只能引用一个真实 field_key；不能合并多个字段，也不能自己编字段。" f"所有字段合计生成约 {target_query_count} 条具体搜索词，每个字段生成 1～3 条。" "judgement 必须是一句不超过 45 个汉字的大白话，只说明大模型针对该字段要判断什么。" "不要提前编造答案、参数、比例或效果数据，不要启动搜索。\n\n" + json.dumps( { "topic_build_id": snapshot["topic_build_id"], "topic_id": snapshot["topic_id"], "field_catalog": field_catalog, }, ensure_ascii=False, indent=2, default=str, ) ) def _field_catalog(snapshot: dict[str, Any]) -> list[dict[str, Any]]: topic = snapshot["topic"] topic_id = snapshot["topic_id"] fields: list[dict[str, Any]] = [] def add( *, field_key: str, field_path: str, field_label: str, field_value: Any, route: str, point_id: Any = None, item_ids: list[Any] | None = None, ) -> None: value = str(field_value or "").strip() if not value: return fields.append( { "field_key": field_key, "field_path": field_path, "field_label": field_label, "field_value": value, "route": route, "point_id": point_id, "item_ids": list(item_ids or []), } ) add( field_key="topic.result", field_path=f"topics[id={topic_id}].result", field_label="最终选题描述", field_value=topic.get("result"), route="cross_dimension", ) direction = topic.get("topic_direction") or {} for preference, label in ( ("实质偏好", "账号实质偏好"), ("形式偏好", "账号形式偏好"), ("意图偏好", "账号意图偏好"), ): add( field_key=f"topic_direction.persona_dimensions.{preference}", field_path=f"topics[id={topic_id}].topic_direction.persona_dimensions.{preference}", field_label=label, field_value=(direction.get("persona_dimensions") or {}).get(preference), route="account_fit", ) point_routes = { "灵感点": "unique_hook", "目的点": "goal_execution", "关键点": "key_execution", } for point in topic.get("points") or []: point_id = point.get("id") point_type = str(point.get("point_type") or "选题点") add( field_key=f"point:{point_id}:point_result", field_path=f"topics[id={topic_id}].points[id={point_id}].point_result", field_label=f"{point_type}字段", field_value=point.get("point_result"), route=point_routes.get(point_type, "cross_dimension"), point_id=point_id, item_ids=list(point.get("item_ids") or []), ) for item in topic.get("composition_items") or []: item_id = item.get("id") dimension = str(item.get("dimension") or "元素") point_type = str(item.get("point_type") or "") add( field_key=f"item:{item_id}:element_name", field_path=( f"topics[id={topic_id}].composition_items[id={item_id}].element_name" ), field_label=f"{dimension}字段", field_value=item.get("element_name"), route=point_routes.get(point_type, "cross_dimension"), item_ids=[item_id], ) for source_index, source in enumerate(item.get("sources") or []): add( field_key=f"item:{item_id}:source:{source_index}:reason", field_path=( f"topics[id={topic_id}].composition_items[id={item_id}]" f".sources[{source_index}].reason" ), field_label="参考来源说明", field_value=source.get("reason"), route="evidence_gap", item_ids=[item_id], ) for relation_index, relation in enumerate(topic.get("item_relations") or []): add( field_key=f"relation:{relation_index}:reason", field_path=f"topics[id={topic_id}].item_relations[{relation_index}].reason", field_label="元素关系说明", field_value=relation.get("reason"), route="cross_dimension", item_ids=[ value for value in ( relation.get("source_item_id"), relation.get("target_item_id"), ) if value is not None ], ) return fields def _safe_need_key(value: Any, route: str, index: int) -> str: raw = re.sub(r"[^a-zA-Z0-9_-]+", "_", str(value or "").strip()).strip("_") return raw[:80] or f"{route}_{index + 1}" def _planning_payload( llm_output: dict[str, Any], *, snapshot: dict[str, Any], field_catalog: list[dict[str, Any]], ) -> dict[str, Any]: raw_mappings = llm_output.get("field_mappings") if not isinstance(raw_mappings, list) or not raw_mappings: raise TopicTableGenerationError("大模型没有返回字段与搜索词的对应关系") topic_build_id = snapshot["topic_build_id"] topic_id = snapshot["topic_id"] catalog_by_key = {field["field_key"]: field for field in field_catalog} needs: list[dict[str, Any]] = [] candidates: list[dict[str, Any]] = [] used_field_keys: set[str] = set() position = 0 for index, raw in enumerate(raw_mappings): if not isinstance(raw, dict): continue field_key = str(raw.get("field_key") or "").strip() field = catalog_by_key.get(field_key) if field is None: raise TopicTableGenerationError(f"大模型引用了不存在的选题表字段：{field_key}") if field_key in used_field_keys: raise TopicTableGenerationError(f"大模型重复引用了同一个选题表字段：{field_key}") used_field_keys.add(field_key) judgement = str(raw.get("judgement") or "").strip() if not judgement: continue route = field["route"] priority = ROUTES[route]["priority"] need_key = _safe_need_key(field_key, route, index) source_ref = { "generator_kind": GeneratorKind.TOPIC_TABLE.value, "topic_build_id": topic_build_id, "topic_id": topic_id, "route": route, "field_key": field["field_key"], "field_path": field["field_path"], "field_label": field["field_label"], "field_value": field["field_value"], "point_id": field.get("point_id"), "item_ids": list(field.get("item_ids") or []), } needs.append( { "need_key": need_key, "decision_context": judgement, "unknown_information": judgement, "source_ref": source_ref, "priority": priority, "metadata": { "route": route, "route_label": ROUTES[route]["label"], "source_field": field, }, } ) raw_queries = raw.get("queries") or [] if isinstance(raw_queries, str): raw_queries = [raw_queries] for query in raw_queries: query_text = str(query or "").strip() if not query_text: continue candidates.append( { "query_text": query_text, "axes": { "选题表字段": field["field_path"], "字段值": field["field_value"], "大模型判断": judgement, }, "priority": priority, "original_position": position, "source_refs": [source_ref], "knowledge_need_keys": [need_key], "metadata": { "family_key": GeneratorKind.TOPIC_TABLE.value, "route": route, "route_label": ROUTES[route]["label"], "need_key": need_key, }, } ) position += 1 if not needs or not candidates: raise TopicTableGenerationError("大模型返回内容无法形成字段与搜索词的对应关系") return { "input_snapshot": snapshot, "knowledge_needs": needs, "candidates": candidates, "generator_config": { "prompt_name": PROMPT_NAME, "prompt_version": PROMPT_VERSION, "field_count": len(field_catalog), }, } def generate_topic_table_preview( source_payload: dict[str, Any], *, topic_id: int | None, max_queries: int | None, chat_fn: Callable[[str, str], dict[str, Any]], ) -> TopicTablePreview: topic = _select_topic(source_payload, topic_id) snapshot = _compact_snapshot(source_payload, topic) field_catalog = _field_catalog(snapshot) llm_output = chat_fn( topic_table_prompt(), build_topic_table_user_prompt( snapshot, field_catalog, target_query_count=max_queries or 18, ), ) if not isinstance(llm_output, dict): raise TopicTableGenerationError("大模型返回格式不正确") payload = _planning_payload( llm_output, snapshot=snapshot, field_catalog=field_catalog, ) request = GenerationRequest( generator_kind=GeneratorKind.TOPIC_TABLE, payload=payload, name=f"topic-table-{snapshot['topic_build_id']}-{snapshot['topic_id']}", max_queries=max_queries, metadata={ "source": "pattern_topic_build_api", "prompt_version": PROMPT_VERSION, }, ) result = UnifiedQueryGenerationService().generate(request) return TopicTablePreview( result=result, source={ "topic_build_id": snapshot["topic_build_id"], "topic_id": snapshot["topic_id"], "source_url": source_payload.get("source_url"), }, topic=snapshot["topic"], llm_output=llm_output, ) def preview_to_dict(preview: TopicTablePreview) -> dict[str, Any]: result = preview.result route_counts: dict[str, int] = {route: 0 for route in ROUTES} for need in result.knowledge_needs: route = str(need.metadata.get("route") or "") if route in route_counts: route_counts[route] += 1 dimensions: dict[str, list[str]] = {"实质": [], "形式": [], "意图": []} reference_notes: list[dict[str, str]] = [] for item in preview.topic.get("composition_items") or []: dimension = item.get("dimension") name = item.get("element_name") if dimension in dimensions and name and item.get("is_active"): dimensions[dimension].append(name) for source in item.get("sources") or []: note = str(source.get("reason") or item.get("reason") or "").strip() if name and note: reference_notes.append({"element": str(name), "note": note}) point_groups: dict[str, list[str]] = {"灵感点": [], "目的点": [], "关键点": []} for point in preview.topic.get("points") or []: point_type = str(point.get("point_type") or "") point_result = str(point.get("point_result") or "").strip() if point_type in point_groups and point_result: point_groups[point_type].append(point_result) queries_by_need: dict[str, list[str]] = {} for query in result.selected_queries: need_keys = query.knowledge_need_keys or [str(query.metadata.get("need_key") or "")] for need_key in need_keys: if need_key: queries_by_need.setdefault(need_key, []).append(query.query_text) return { "generator_kind": GeneratorKind.TOPIC_TABLE.value, "source": preview.source, "topic": { "status": preview.topic.get("status"), "result": preview.topic.get("result"), "topic_direction": preview.topic.get("topic_direction"), "dimensions": dimensions, "points": preview.topic.get("points") or [], "input_groups": { "points": point_groups, "reference_notes": reference_notes[:8], }, }, "knowledge_needs": [ { "need_key": need.need_key, "route": need.metadata.get("route"), "route_label": need.metadata.get("route_label"), "decision_context": need.decision_context, "unknown_information": need.unknown_information, "priority": need.priority, "source_ref": need.source_ref, } for need in result.knowledge_needs ], "field_mappings": [ { "field_key": need.source_ref.get("field_key"), "field_path": need.source_ref.get("field_path"), "field_label": need.source_ref.get("field_label"), "field_value": need.source_ref.get("field_value"), "judgement": need.decision_context, "queries": queries_by_need.get(need.need_key, []), } for need in result.knowledge_needs if queries_by_need.get(need.need_key) ], "queries": [ { "query": query.query_text, "priority": query.priority, "route": query.metadata.get("route"), "route_label": query.metadata.get("route_label"), "need_key": query.metadata.get("need_key"), "axes": query.axes, "source_refs": query.source_refs, } for query in result.selected_queries ], "summary": { "generated_count": result.stats.generated_count, "unique_count": result.stats.unique_count, "selected_count": result.stats.selected_count, "dropped_count": result.stats.dropped_count, "route_counts": route_counts, "persisted": False, "search_started": False, }, "prompt": {"name": PROMPT_NAME, "version": PROMPT_VERSION}, }