| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546 |
- """创作 query 筛选器：调用 acquisition/query_filter.txt（LLM 只做 keep/排除），批量判一组 query。
- 命中 A–E 任一即排除：A 语义不通 / B 题材本身 / C 制作工具 / D 应试学术公务政企 / E 作品素材。
- 返回与输入等长的 [{keep: bool, reason: str}]。供 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"
- 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)}
- return [{"keep": bool(by.get(i, {}).get("keep", True)),
- "reason": str(by.get(i, {}).get("reason", ""))[:50]} for i in range(len(queries))]
- except Exception as exc:
- return [{"keep": True, "reason": f"筛选失败:{str(exc)[:30]}"} for _ in queries]
|
