import copy import pytest from content_agent.business_modules import search_intent from content_agent.errors import ContentAgentError from content_agent.integrations.query_prompt_config import DEFAULT_PROFILE from content_agent.run_service import RunService from content_agent.schemas import RunStartRequest from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE FORBIDDEN_FIXED_BUSINESS_TERMS = [ "\u8d2a\u8150", "\u57fa\u5c42\u516c\u804c\u4eba\u5458", "\u6848\u4f8b", "\u89e3\u8bfb", "\u8b66\u793a", ] class _Runtime: def __init__(self): self.rows = {} def append_jsonl(self, _run_id, filename, rows): self.rows[filename] = rows def _seed_pack(): return { "seed_terms": ["中医养生"], "query_seed_points": [ { "point_text": "分享气血食疗", "point_type": "目的点", "rank": 1, "post_id": "p1", "id": "qp1", "coverage_post_count": 5, "category_id": "c1", }, { "point_text": "办公室八段锦", "point_type": "灵感点", "rank": 2, "post_id": "p2", "id": "qp2", "coverage_post_count": 3, "category_id": "c1", }, ], "itemset_items": ["补气血"], "category_bindings": [{"category_id": "c1"}], "element_bindings": [ { "element_id": "e1", "category_id": "c1", "sample_elements": [ { "id": "point_1", "name": "食疗补气血", "point_type": "目的点", "point_text": "分享气血食疗方法", "post_id": "p1", }, { "id": "point_2", "name": "八段锦", "point_type": "灵感点", "point_text": "办公室八段锦", "post_id": "p2", }, { "id": "point_3", "name": "日常保健", "point_type": "关键点", "point_text": "不应该进入搜索词", "post_id": "p3", }, ], } ], "pattern_source_system": "pg_pattern_v2", "pattern_execution_id": 1987, "mining_config_id": 58, "source_post_id": "60219550", "matched_post_ids": ["60219550"], "itemset_ids": [1607977], "support": 0.2, "absolute_support": 31, "confidence": 0.8, } def test_search_seed_and_queries_do_not_inject_fixed_business_terms(tmp_path): service = RunService( runtime_root=tmp_path / "runtime" / "v1", query_variant_client=FakeQueryVariantClient( { "爱国情感": "家国叙事素材", "人物故事": "榜样人物素材", } ), ) state = service.start_run( RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1") ) run_id = state["run_id"] pattern_seed_pack = service.read_json(run_id, "pattern_seed_pack.json") queries = service.read_jsonl(run_id, "search_queries.jsonl") p2_queries = [ row for row in queries if row["search_query_generation_method"] in {"item_single", "llm_variant"} ] assert pattern_seed_pack["seed_terms"] == ["爱国情感", "人物故事"] assert [row["search_query_id"] for row in p2_queries] == ["q_001", "q_002", "q_003", "q_004"] assert [row["search_query"] for row in p2_queries] == [ "爱国情感", "家国叙事素材", "人物故事", "榜样人物素材", ] assert [row["search_query_generation_method"] for row in p2_queries] == [ "item_single", "llm_variant", "item_single", "llm_variant", ] assert p2_queries[1]["llm_variant_of"] == "q_001" assert p2_queries[3]["llm_variant_of"] == "q_003" for value in [ *pattern_seed_pack["seed_terms"], *(row["search_query"] for row in p2_queries), ]: assert not any(term in value for term in FORBIDDEN_FIXED_BUSINESS_TERMS) def test_search_queries_preserve_source_terms_for_replay(tmp_path): service = RunService( runtime_root=tmp_path / "runtime" / "v1", query_variant_client=FakeQueryVariantClient( { "爱国情感": "家国叙事素材", "人物故事": "榜样人物素材", } ), ) state = service.start_run( RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1") ) queries = service.read_jsonl(state["run_id"], "search_queries.jsonl") p2_queries = [ query for query in queries if query["search_query_generation_method"] in {"item_single", "llm_variant"} ] expected_source_terms = [["爱国情感"], ["爱国情感"], ["人物故事"], ["人物故事"]] for query, source_terms in zip(p2_queries, expected_source_terms, strict=True): assert query["query_source_terms"] == source_terms assert query["query_source_fields"] == ["seed_terms"] assert query["raw_payload"]["query_source_terms"] == source_terms assert query["pattern_seed_ref"]["source_field"] == "seed_terms" assert query["pattern_seed_ref"]["seed_term"] == source_terms[0] assert query["raw_payload"]["pattern_seed_ref"]["seed_term"] == source_terms[0] llm_queries = [ query for query in p2_queries if query["search_query_generation_method"] == "llm_variant" ] assert len(llm_queries) == 2 for query in llm_queries: assert query["raw_payload"]["llm_prompt_version"] == "fake-query-prompt-v1" assert query["raw_payload"]["llm_generation_model"] == "fake-query-model" assert query["raw_payload"]["llm_input_evidence"]["source_field"] == "seed_terms" assert query["raw_payload"]["llm_input_evidence"]["itemset_items"] def test_search_intent_custom_evidence_fields_whitelist(): client = FakeQueryVariantClient({"中医养生": "气血食疗"}) client.profile = copy.deepcopy(DEFAULT_PROFILE) client.profile["evidence_fields"] = ["seed_term", "support"] runtime = _Runtime() queries = search_intent.run("run_1", "policy_1", _seed_pack(), runtime, client) llm_query = [row for row in queries if row["search_query_generation_method"] == "llm_variant"][0] assert list(llm_query["llm_input_evidence"].keys()) == ["seed_term", "support"] assert list(llm_query["raw_payload"]["llm_input_evidence"].keys()) == ["seed_term", "support"] assert llm_query["query_source_fields"] == ["seed_terms"] def test_v4_search_intent_uses_three_direct_sources_without_llm_variant(): client = FakeQueryVariantClient({"中医养生": "气血食疗"}) runtime = _Runtime() queries = search_intent.run( "run_1", "policy_1", _seed_pack(), runtime, client, strategy_version="V4", ) assert client.calls == [] # 三路首轮搜索池:seed_terms、query_seed_points、分类叶子元素;不再调 LLM 变体。 assert [row["search_query"] for row in queries] == [ "中医养生", "气血食疗", "办公室八段锦", "食疗补气血", "八段锦", "日常保健", ] assert [row["search_query_generation_method"] for row in queries] == [ "seed_term", "query_seed_point", "query_seed_point", "category_leaf_element", "category_leaf_element", "category_leaf_element", ] assert all("llm_variant_of" not in row for row in queries) assert runtime.rows["search_queries.jsonl"] == queries first = next(row for row in queries if row["search_query"] == "气血食疗") assert first["pattern_seed_ref"]["query_source_type"] == "query_seed_point" assert first["pattern_seed_ref"]["query_source_text"] == "气血食疗" assert first["pattern_seed_ref"]["query_source_rank"] == 1 source_ref = first["query_source_refs"][0]["source_ref"] assert source_ref["point_type"] == "目的点" assert source_ref["point_text"] == "分享气血食疗" # 原文不被覆盖 assert source_ref["cleaned_text"] == "气血食疗" assert source_ref["qsp_rank"] == 1 assert first["raw_payload"]["query_source_refs"][0]["query_source_text"] == "气血食疗" def test_v4_falls_back_to_seed_terms_when_query_seed_points_empty(): runtime = _Runtime() seed_pack = _seed_pack() seed_pack["query_seed_points"] = [] seed_pack["element_bindings"] = [] queries = search_intent.run( "run_1", "policy_1", seed_pack, runtime, FakeQueryVariantClient(), strategy_version="V4", ) assert [row["search_query"] for row in queries] == ["中医养生"] assert all(row["search_query_generation_method"] == "seed_term" for row in queries) def test_v4_search_intent_dedupes_query_seed_points(): runtime = _Runtime() seed_pack = _seed_pack() seed_pack["query_seed_points"] = [ {"point_text": "气血食疗", "point_type": "灵感点", "rank": 1, "post_id": "p1", "id": "x1"}, {"point_text": "气血食疗", "point_type": "灵感点", "rank": 2, "post_id": "p2", "id": "x2"}, ] queries = search_intent.run( "run_1", "policy_1", seed_pack, runtime, FakeQueryVariantClient(), strategy_version="V4", ) matching = [row for row in queries if row["search_query"] == "气血食疗"] assert len(matching) == 1 assert matching[0]["search_query_generation_method"] == "query_seed_point" def test_v4_search_intent_limits_query_seed_points_to_top_two(): runtime = _Runtime() seed_pack = _seed_pack() seed_pack["element_bindings"] = [] seed_pack["query_seed_points"] = [ {"point_text": "第一点", "point_type": "灵感点", "rank": 1, "post_id": "p1", "id": "x1"}, {"point_text": "第二点", "point_type": "灵感点", "rank": 2, "post_id": "p2", "id": "x2"}, {"point_text": "第三点", "point_type": "灵感点", "rank": 3, "post_id": "p3", "id": "x3"}, ] queries = search_intent.run( "run_1", "policy_1", seed_pack, runtime, FakeQueryVariantClient(), strategy_version="V4", ) assert [row["search_query"] for row in queries] == ["中医养生", "第一点", "第二点"] def test_search_intent_custom_generic_filter_blocks_query(): client = FakeQueryVariantClient({"中医养生": "禁用泛词"}) client.profile = copy.deepcopy(DEFAULT_PROFILE) client.profile["generic_filter"] = {"queries": ["禁用泛词"], "tokens": []} with pytest.raises(ContentAgentError) as exc: search_intent.run("run_1", "policy_1", _seed_pack(), _Runtime(), client) assert exc.value.error_code == "QUERY_GENERATION_FAILED" assert exc.value.detail["reason"] == "llm_variant_generic" def test_search_intent_rejects_unsupported_variants_per_seed(): client = FakeQueryVariantClient({"中医养生": "气血食疗"}) client.profile = copy.deepcopy(DEFAULT_PROFILE) client.profile["variants_per_seed"] = 2 with pytest.raises(ContentAgentError) as exc: search_intent.run("run_1", "policy_1", _seed_pack(), _Runtime(), client) assert exc.value.error_code == "QUERY_GENERATION_FAILED" assert exc.value.detail == {"reason": "variants_per_seed_unsupported", "variants_per_seed": 2}