| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960 |
- """创作 query 筛选器：调用 acquisition/query_filter.txt，批量判一组 query。
- 模型对每条 query 打两项：语义合法性 valid(0-10) + 创作相关性 relevant(bool)。
- 最终保留 keep = valid ≥ VALID_MIN 且 relevant —— 阈值放代码里(好调，不改提示词)。
- · valid：词组合本身说不说得通(机械正交易产生的废话短语在此淘汰)。
- · relevant：搜回来是否和创作知识相关(三把尺=设计决策vs工艺/可迁移/业务阶段；再排 制作/题材本身/应试政企/作品)。
- 返回与输入等长的 [{keep, valid, relevant, reason}]。供 build_creation_demo、filter_multiaxis 等共用。
- """
- from __future__ import annotations
- import json
- from pathlib import Path
- import httpx
- from core.config import Settings
- PROMPT = Path(__file__).resolve().parent / "query_filter.txt"
- VALID_MIN = 6 # 语义合法性阈值：valid ≥ 此值才算说得通；调这里即可松紧
- def filter_queries(queries: list[str], settings: Settings, *, batch: int = 40) -> list[dict]:
- """批量过滤（分批调用，避免一次喂太多）。返回 [{keep, reason}]，与 queries 对齐。"""
- out: list[dict] = []
- for i in range(0, len(queries), batch):
- out.extend(_filter_batch(queries[i:i + batch], settings))
- return out
- def _filter_batch(queries: list[str], settings: Settings) -> list[dict]:
- if not queries:
- return []
- user = json.dumps([{"idx": i, "query": q} for i, q in enumerate(queries)], ensure_ascii=False)
- api = settings.openrouter_base_url.rstrip("/") + "/chat/completions"
- headers = {"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"}
- body = {"model": settings.llm_model, "messages": [
- {"role": "system", "content": PROMPT.read_text("utf-8")},
- {"role": "user", "content": user}], "response_format": {"type": "json_object"}}
- try:
- resp = httpx.post(api, headers=headers, json=body, timeout=120)
- resp.raise_for_status()
- txt = resp.json()["choices"][0]["message"]["content"]
- # query_filter 要求输出数组；有的模型会包一层 {"result":[...]}，都兜住
- data = json.loads(txt)
- arr = data if isinstance(data, list) else next((v for v in data.values() if isinstance(v, list)), [])
- by = {d.get("idx"): d for d in arr if isinstance(d, dict)}
- out = []
- for i in range(len(queries)):
- d = by.get(i, {})
- try:
- valid = int(d.get("valid", 10))
- except (TypeError, ValueError):
- valid = 10
- relevant = bool(d.get("relevant", True))
- out.append({"keep": valid >= VALID_MIN and relevant, # 阈值在代码里判
- "valid": valid, "relevant": relevant,
- "reason": str(d.get("reason", ""))[:50]})
- return out
- except Exception as exc:
- return [{"keep": True, "valid": 10, "relevant": True, "reason": f"筛选失败:{str(exc)[:30]}"} for _ in queries]
|
