| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495 |
- """创作 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
- import os
- from pathlib import Path
- import httpx
- from core.config import Settings, load_env_file
- 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)
- messages = [
- {"role": "system", "content": PROMPT.read_text("utf-8")},
- {"role": "user", "content": user},
- ]
- try:
- txt = _chat_content(settings, messages)
- # 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]
- def _chat_content(settings: Settings, messages: list[dict]) -> str:
- """Return chat content. Prefer OpenRouter; fall back to Ark when OpenRouter is region-blocked locally."""
- body = {"model": settings.llm_model, "messages": messages, "response_format": {"type": "json_object"}}
- env = load_env_file(os.getenv("CK_ENV_FILE", ".env"))
- prefer_ark = os.getenv("QUERY_FILTER_PROVIDER") == "ark" or bool(os.getenv("ARK_CHAT_MODEL"))
- openrouter_exc: Exception | None = None
- if settings.openrouter_api_key and not prefer_ark:
- try:
- resp = httpx.post(
- settings.openrouter_base_url.rstrip("/") + "/chat/completions",
- headers={"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"},
- json=body,
- timeout=120,
- )
- resp.raise_for_status()
- return resp.json()["choices"][0]["message"]["content"]
- except Exception as exc:
- openrouter_exc = exc
- ark_key = os.getenv("ARK_API_KEY") or env.get("ARK_API_KEY")
- if ark_key:
- ark_model = os.getenv("ARK_CHAT_MODEL") or env.get("ARK_CHAT_MODEL") or "doubao-seed-1-6-flash-250615"
- ark_url = os.getenv("ARK_CHAT_URL") or env.get("ARK_CHAT_URL") or "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
- resp = httpx.post(
- ark_url,
- headers={"Authorization": f"Bearer {ark_key}", "Content-Type": "application/json"},
- json={"model": ark_model, "messages": messages, "response_format": {"type": "json_object"}},
- timeout=120,
- )
- resp.raise_for_status()
- return resp.json()["choices"][0]["message"]["content"]
- if openrouter_exc:
- raise openrouter_exc
- raise RuntimeError("missing OpenRouter/Ark chat credentials")
|
