content-find-agent-new/content_agent/integrations/query_prompt_config.py

from __future__ import annotations

import copy
from pathlib import Path
from typing import Any

from content_agent.integrations import config_store

QUERY_PROMPTS_PATH = Path("product_documents/配置/query_prompts.v1.json")

DEFAULT_PROFILE: dict[str, Any] = {
    "prompt_version": "query_variant.v1",
    "system": (
        "You generate one concise Chinese short-video search query. "
        "Return exactly one plain query string. Do not return JSON, "
        "lists, quotes, explanations, or multiple lines."
    ),
    "user": (
        "Seed term:\n{seed_term}\n\n"
        "Evidence context:\n{evidence_context}\n\n"
        "Create one adjacent search phrase that stays faithful to the evidence. "
        "Avoid any phrase listed in existing_search_queries."
    ),
    "temperature": 0.4,
    "max_tokens": 64,
    "evidence_fields": [
        "seed_term",
        "seed_terms",
        "existing_search_queries",
        "source_field",
        "source_index",
        "itemset_items",
        "category_bindings",
        "element_bindings",
        "pattern_source_system",
        "pattern_execution_id",
        "mining_config_id",
        "source_post_id",
        "matched_post_ids",
        "itemset_ids",
        "support",
        "absolute_support",
        "confidence",
    ],
    "variants_per_seed": 1,
    "generic_filter": {
        "queries": [
            "内容",
            "视频",
            "热门",
            "推荐",
            "短视频",
            "热门视频",
            "推荐视频",
            "热门内容",
            "推荐内容",
            "相关视频",
            "相关内容",
            "热点视频",
            "热点内容",
        ],
        "tokens": [
            "短视频",
            "热门",
            "推荐",
            "相关",
            "热点",
            "内容",
            "视频",
            "素材",
            "资料",
            "信息",
            "话题",
        ],
    },
}


def load_profile(
    platform: str,
    strategy_version: str,
    root_dir: Path | str = Path("."),
) -> dict[str, Any]:
    path = Path(root_dir) / QUERY_PROMPTS_PATH
    try:
        config, _raw = config_store.load_json(path)
    except FileNotFoundError:
        return _copy_default_profile()

    profiles = config.get("profiles") if isinstance(config, dict) else {}
    if not isinstance(profiles, dict):
        return _copy_default_profile()

    profile = profiles.get(f"{platform}/{strategy_version}")
    if not isinstance(profile, dict):
        return _copy_default_profile()
    return copy.deepcopy(profile)


def _copy_default_profile() -> dict[str, Any]:
    return copy.deepcopy(DEFAULT_PROFILE)