refactor(pipeline/llm): 文本大模型工厂与 OpenAI 兼容适配器

新增 providers 包:协议、Crawler 网关适配器、output 去围栏、token 启发式与工厂;llm_client 与 keyword_suggest 走统一入口。未改各 generate_* 提示词。补充 MA_LLM_TEXT_PROVIDER 说明至 .env.example。

Made-with: Cursor
This commit is contained in:
hub-gif 2026-04-27 09:56:21 +08:00
parent 37870901ee
commit 52c9dc1697
9 changed files with 200 additions and 185 deletions

View File

@ -29,6 +29,12 @@ CSRF_TRUSTED_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
# OPENAI_TEXT_MODEL=
# 别名LLM_API_KEY、LLM_BASE_URL、LLM_MODEL
# 报告/策略等「纯文本」任务的后端选择(默认经 crawler 副本内 AI_crawler 调 OpenAI 兼容网关;其它取值待扩展)
# MA_LLM_TEXT_PROVIDER=crawler_openai_compatible
# 独立策略稿 / 报告内「策略与机会」LLM 块采样温度(默认 0.1,低于 chat_completion_text 的 0.2,减轻同提示多轮漂移)。设 0 更稳、设 0.2 与旧默认接近。
# MA_STRATEGY_LLM_TEMPERATURE=0.1
# --- 可选:流水线侧 LLM 开关 ---
# MA_SKIP_LLM_KEYWORD_SUGGEST=1
# MA_ENABLE_LLM_COMMENT_SENTIMENT=1

View File

@ -1,21 +1,17 @@
"""在报告生成前:基于评价正文调用大模型,联想补充**关注词**与**使用场景触发组**(写入 effective_report_config)。"""
"""在报告生成前:基于评价正文调用大模型,联想**短语候选**(写入 keyword_suggest_llm.json不再合并进预设词表)。"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
from typing import Any
from django.conf import settings
from .llm_client import call_llm
MAX_CHUNK_CHARS = 24_000
MAX_CHUNKS = 12
# 场景联想单次送入模型的评价摘录上限(字符);过大易顶上下文
SCENARIO_CORPUS_MAX_CHARS = 18_000
_CHUNK_SYSTEM = """你是电商评价挖掘助手。输入 JSON 含 keyword、excerpt_index、excerpts一段用户评价正文合集
任务 excerpts 中抽取值得纳入关注词/卖点监测**中文短语**212 字为主可为词组
任务 excerpts 中抽取**可人工选用的监测短语**卖点体验规格等****系统预设词表212 字为主可为词组
硬性规则
- 仅输出一段 JSON{"phrases": ["短语1", ...]}短语共 620
@ -24,25 +20,6 @@ _CHUNK_SYSTEM = """你是电商评价挖掘助手。输入 JSON 含 keyword、ex
- 不要输出 JSON 以外的文字"""
def _ensure_ai_crawler_path() -> None:
root = Path(settings.CRAWLER_JD_ROOT).resolve()
if not root.is_dir():
raise FileNotFoundError(f"爬虫副本目录不存在: {root}")
rs = str(root)
if rs not in sys.path:
sys.path.insert(0, rs)
def _call_llm(system_prompt: str, user_prompt: str) -> str:
_ensure_ai_crawler_path()
import AI_crawler as ac # noqa: WPS433
return ac.chat_completion_text(
system_prompt=system_prompt,
user_prompt=user_prompt,
)
def _chunk_comment_texts(texts: list[str]) -> list[str]:
"""将全量评价划为若干段,控制单段字符量与最大段数。"""
parts: list[str] = []
@ -92,127 +69,6 @@ def _parse_phrases_object(raw: str) -> list[str]:
return []
_SCENARIO_SYSTEM = """你是电商用户研究助手。输入 JSON 含:
- ``keyword``监测词
- ``existing_scenarios``数组每项为 ``{"label": "展示名", "triggers": ["子串1", ...]}``统计时若评价正文**包含任一 trigger 子串**则计入该 label与宿主系统规则一致
- ``excerpts``多条用户评价正文摘录已截断拼接
任务**不重复** ``existing_scenarios`` 中已有 ``label``逐字比较勿改写字的前提下 excerpts 归纳 **412 **新的用途/场景监测组覆盖评论里**明显出现但未被现有组覆盖**的消费情境下午茶露营宿舍须确有文本依据
硬性规则
- **仅输出**一段 JSON``{"scenarios": [{"label": "展示名", "triggers": ["子串1", "子串2", ...]}, ...]}``
- 每条 ``label`` 216 每组 ``triggers`` 310 每条 trigger **212 字中文**子串用于**子串命中**计数
- 不要医疗功效治愈降血糖承诺不要与 existing label 同名或仅差空格
- 不要输出 JSON 以外的文字"""
def _sample_corpus_for_scenarios(texts: list[str], *, max_chars: int) -> str:
"""取评价正文前部拼接至 max_chars供单次场景联想。"""
parts: list[str] = []
n = 0
for t in texts:
s = (t or "").strip()
if not s:
continue
extra = len(s) + 1
if n + extra > max_chars:
remain = max_chars - n - 1
if remain > 40:
parts.append(s[:remain])
break
parts.append(s)
n += extra
return "\n".join(parts)
def _parse_scenarios_object(raw: str) -> list[dict[str, Any]]:
t = (raw or "").strip()
t = re.sub(r"^```(?:json)?\s*", "", t, flags=re.IGNORECASE)
t = re.sub(r"\s*```$", "", t)
try:
obj = json.loads(t)
except json.JSONDecodeError:
obj = None
if not isinstance(obj, dict):
m = re.search(r"\{[\s\S]*\}", t)
if not m:
return []
try:
obj = json.loads(m.group(0))
except json.JSONDecodeError:
return []
arr = obj.get("scenarios")
if not isinstance(arr, list):
return []
out: list[dict[str, Any]] = []
for item in arr:
if not isinstance(item, dict):
continue
label = str(item.get("label") or "").strip()[:80]
tr_raw = item.get("triggers")
triggers: list[str] = []
if isinstance(tr_raw, list):
seen_t: set[str] = set()
for x in tr_raw[:24]:
s = str(x).strip()
if len(s) < 2 or len(s) > 24:
continue
if s in seen_t:
continue
seen_t.add(s)
triggers.append(s)
if label and len(triggers) >= 1:
out.append({"label": label, "triggers": triggers[:12]})
return out
def suggest_scenario_groups_llm(
*,
keyword: str,
existing_groups: list[dict[str, Any]],
all_comment_texts: list[str],
) -> dict[str, Any]:
"""
单次调用模型基于评价摘录扩展 ``comment_scenario_groups`` 形态的新组label + triggers
"""
if not all_comment_texts:
return {
"suggested_scenario_groups": [],
"scenario_rationale": "无评价正文可分析。",
}
existing_compact: list[dict[str, Any]] = []
for g in (existing_groups or [])[:36]:
if not isinstance(g, dict):
continue
lab = str(g.get("label") or "").strip()
tr = g.get("triggers")
ts: list[str] = []
if isinstance(tr, list):
for x in tr[:16]:
s = str(x).strip()
if s:
ts.append(s[:48])
if lab and ts:
existing_compact.append({"label": lab[:80], "triggers": ts})
excerpts = _sample_corpus_for_scenarios(
all_comment_texts, max_chars=SCENARIO_CORPUS_MAX_CHARS
)
payload = {
"keyword": keyword,
"existing_scenarios": existing_compact,
"excerpts": excerpts,
}
raw = _call_llm(_SCENARIO_SYSTEM, json.dumps(payload, ensure_ascii=False))
scenarios = _parse_scenarios_object(raw)
return {
"suggested_scenario_groups": scenarios[:14],
"scenario_rationale": (
f"基于约 {len(excerpts)} 字评价摘录单次调用模型;"
f"{len(existing_compact)} 组既有场景之外补充 {len(scenarios[:14])} 组候选。"
),
}
def suggest_focus_keywords_from_all_comments(
*,
keyword: str,
@ -245,7 +101,7 @@ def suggest_focus_keywords_from_all_comments(
"excerpt_index": i + 1,
"excerpts": ch,
}
raw = _call_llm(_CHUNK_SYSTEM, json.dumps(payload, ensure_ascii=False))
raw = call_llm(_CHUNK_SYSTEM, json.dumps(payload, ensure_ascii=False))
collected.extend(_parse_phrases_object(raw))
seen: set[str] = set()
@ -264,7 +120,7 @@ def suggest_focus_keywords_from_all_comments(
"suggested_focus_keywords": out_kw,
"rationale": (
f"基于全量 {len(all_comment_texts)} 条评价文本,分 {len(chunks)} 段调用模型抽取短语并去重;"
f"已排除与当前关注词统计表完全相同的词"
"报告主文不以子串词表统计为主指标,本结果仅供业务人工参考"
),
"chunks_processed": len(chunks),
"total_comment_texts": len(all_comment_texts),

View File

@ -1,47 +1,30 @@
"""竞品报告 LLM 调用:路径注入与网关 ``chat_completion_text`` 封装"""
"""竞品报告 LLM 调用:经 `providers` 工厂选择后端,并统一对输出做去围栏等归一化"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from django.conf import settings
from .providers.factory import get_text_llm
from .providers.output_normalize import strip_outer_markdown_fence
def ensure_ai_crawler_path() -> None:
root = Path(settings.CRAWLER_JD_ROOT).resolve()
if not root.is_dir():
raise FileNotFoundError(f"爬虫副本目录不存在: {root}")
rs = str(root)
if rs not in sys.path:
sys.path.insert(0, rs)
def call_llm(system_prompt: str, user_prompt: str) -> str:
ensure_ai_crawler_path()
import AI_crawler as ac # noqa: WPS433
raw = ac.chat_completion_text(
system_prompt=system_prompt,
user_prompt=user_prompt,
def call_llm(
system_prompt: str,
user_prompt: str,
*,
temperature: float | None = None,
) -> str:
raw = get_text_llm().complete_text(
system_prompt,
user_prompt,
temperature=temperature,
)
return ac.strip_outer_markdown_fence(raw)
return strip_outer_markdown_fence(raw)
def estimate_chat_input_tokens(system_prompt: str, user_prompt: str) -> int:
"""与 ``AI_crawler._estimate_chat_input_tokens`` 一致,用于在调用前预判上下文。"""
total_chars = len(system_prompt or "") + len(user_prompt or "")
return int(total_chars * 0.55) + 512
"""与当前所选文本后端的预检一致;默认与 ``AI_crawler`` 的保守估算同口径。"""
return get_text_llm().estimate_input_tokens(system_prompt, user_prompt)
def llm_context_window_size() -> int:
"""与 ``AI_crawler.chat_completion_text`` 使用的上下文上限一致。"""
raw = (
os.environ.get("LLM_CONTEXT_WINDOW")
or os.environ.get("OPENAI_CONTEXT_WINDOW")
or "32768"
).strip()
try:
return max(4096, int(raw))
except ValueError:
return 32768
"""与当前所选后端的上下文上限一致;默认与 ``AI_crawler.chat_completion_text`` 使用的环境变量一致。"""
return get_text_llm().context_window_tokens()

View File

@ -0,0 +1,11 @@
"""文本大模型调用的协议、适配器与工厂(与具体提示词/业务生成逻辑解耦)。"""
from __future__ import annotations
from .factory import get_text_llm, reset_text_llm_client_for_tests
from .protocol import TextLlmClient
__all__ = [
"TextLlmClient",
"get_text_llm",
"reset_text_llm_client_for_tests",
]

View File

@ -0,0 +1,64 @@
"""
`crawler_copy/jd_pc_search/AI_crawler.chat_completion_text` 访问 OpenAI 兼容网关与配料识别等共用凭据与配置
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from django.conf import settings
from .token_heuristics import estimate_crawler_style_input_tokens
def ensure_ai_crawler_path() -> None:
root = Path(settings.CRAWLER_JD_ROOT).resolve()
if not root.is_dir():
raise FileNotFoundError(f"爬虫副本目录不存在: {root}")
rs = str(root)
if rs not in sys.path:
sys.path.insert(0, rs)
def _llm_context_window_size_from_env() -> int:
raw = (
os.environ.get("LLM_CONTEXT_WINDOW")
or os.environ.get("OPENAI_CONTEXT_WINDOW")
or "32768"
).strip()
try:
return max(4096, int(raw))
except ValueError:
return 32768
class CrawlerOpenAiCompatibleTextLlm:
"""
文本任务默认后端复用 `AI_crawler` `chat_completion_text` 与上下文预检逻辑
更换为其它云厂商时应新增独立适配器并在 `factory` 中注册而非修改本类
"""
def complete_text(
self,
system_prompt: str,
user_prompt: str,
*,
temperature: float | None = None,
) -> str:
ensure_ai_crawler_path()
import AI_crawler as ac # noqa: WPS433
kwargs: dict[str, object] = {
"system_prompt": system_prompt,
"user_prompt": user_prompt,
}
if temperature is not None:
kwargs["temperature"] = float(temperature)
return ac.chat_completion_text(**kwargs)
def estimate_input_tokens(self, system_prompt: str, user_prompt: str) -> int:
return estimate_crawler_style_input_tokens(system_prompt, user_prompt)
def context_window_tokens(self) -> int:
return _llm_context_window_size_from_env()

View File

@ -0,0 +1,48 @@
"""
根据 `MA_LLM_TEXT_PROVIDER` 选择文本大模型实现未设置时与历史行为一致 AI_crawler OpenAI 兼容网关
"""
from __future__ import annotations
import os
from .crawler_openai_compatible import CrawlerOpenAiCompatibleTextLlm
from .protocol import TextLlmClient
# 模块级单例:避免重复构造;测试可用 `reset_text_llm_client_for_tests` 切换实现。
_client: TextLlmClient | None = None
# 与历史默认行为一致
_DEFAULT_ID = "crawler_openai_compatible"
_ENV_KEY = "MA_LLM_TEXT_PROVIDER"
def _provider_id() -> str:
raw = (os.environ.get(_ENV_KEY) or _DEFAULT_ID).strip().lower()
return raw or _DEFAULT_ID
def _build_client(pid: str) -> TextLlmClient:
if pid in (
"crawler_openai_compatible",
"crawler",
"default",
"openai_compatible",
):
return CrawlerOpenAiCompatibleTextLlm()
raise ValueError(
f"不支持的 {_ENV_KEY}={pid!r}"
f"当前仅实现 {_DEFAULT_ID}(经 AI_crawler 的 OpenAI 兼容 `chat/completions`)。"
)
def get_text_llm() -> TextLlmClient:
global _client
if _client is None:
_client = _build_client(_provider_id())
return _client
def reset_text_llm_client_for_tests() -> None:
"""供 pytest/集成测试在修改环境变量后清空缓存的客户端。"""
global _client
_client = None

View File

@ -0,0 +1,15 @@
"""对模型原始输出做与业务无关的轻量归一化(不修改提示词)。"""
from __future__ import annotations
def strip_outer_markdown_fence(text: str) -> str:
"""若模型用 ``` / ```markdown 包裹全文,去掉最外层围栏。与 `AI_crawler.strip_outer_markdown_fence` 行为一致。"""
t = (text or "").strip()
if not t.startswith("```"):
return t
lines = t.split("\n")
if lines and lines[0].strip().startswith("```"):
lines = lines[1:]
while lines and lines[-1].strip() == "```":
lines = lines[:-1]
return "\n".join(lines).strip()

View File

@ -0,0 +1,22 @@
"""文本 LLM 客户端协议:业务侧只依赖本接口,具体网关由适配器 + 工厂选择。"""
from __future__ import annotations
from typing import Protocol, runtime_checkable
@runtime_checkable
class TextLlmClient(Protocol):
def complete_text(
self,
system_prompt: str,
user_prompt: str,
*,
temperature: float | None = None,
) -> str:
"""一次 system + user 的纯文本补全,返回助理正文(无通用后处理,由 `llm_client.call_llm` 统一去围栏等)。"""
def estimate_input_tokens(self, system_prompt: str, user_prompt: str) -> int:
"""与当次后端的 `max_tokens` 预检/截断策略一致的输入侧 token 保守估算。"""
def context_window_tokens(self) -> int:
"""当前配置下的上下文 token 上限(与预检、策略模块档位一致)。"""

View File

@ -0,0 +1,10 @@
"""
`crawler_copy/.../AI_crawler` `_estimate_chat_input_tokens` 同口径的保守估算
供预检策略档位与 OpenAI 兼容适配器共用 tiktoken 时避免 max_tokens 400
"""
from __future__ import annotations
def estimate_crawler_style_input_tokens(system_prompt: str, user_prompt: str) -> int:
total_chars = len(system_prompt or "") + len(user_prompt or "")
return int(total_chars * 0.55) + 512