浅念
188 lines
8.0 KiB
Python
188 lines
8.0 KiB
Python
"""PromptProvider(Task 6): 租户提示词四件套 + $DEFLECTION_POOL 单一事实源注入。
|
|
|
|
- 提示词从租户目录(seed_profiles/<slug>/prompts)读,不再有代码级
|
|
app/engine/prompts 副本;
|
|
- system.md 的迂回话术枚举段替换为 $DEFLECTION_POOL 占位符,组装时注入
|
|
wall.json deflections 池(话术只存 wall.json 一份,改一处两边同步变);
|
|
- extractor/evaluator 的 $existing_info/$messages/$profile/$recent_messages
|
|
替换逻辑从 AIService 原样搬入 Provider(行为等价)。
|
|
|
|
路径纪律: 一律 Path(__file__) 锚定仓库根,禁 cwd 相对。
|
|
"""
|
|
import json
|
|
import pathlib
|
|
|
|
import pytest
|
|
|
|
from app.engine.services.ai_service import DEFAULT_PROVIDER, PromptProvider
|
|
|
|
REPO = pathlib.Path(__file__).resolve().parents[1]
|
|
JC_PROMPTS = REPO / "seed_profiles" / "jiancai" / "prompts"
|
|
AUTO_PROMPTS = REPO / "seed_profiles" / "auto" / "prompts"
|
|
|
|
|
|
def _wall(slug: str) -> dict:
|
|
return json.loads(
|
|
(REPO / "seed_profiles" / slug / "wall.json").read_text(encoding="utf-8"))
|
|
|
|
|
|
# ---- 加载 --------------------------------------------------------------
|
|
|
|
def test_provider_loads_from_tenant_dir():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
assert "HESUPIPE" in p.company_facts # jiancai 事实卡身份标记(审查 Minor-5: 删恒真 or 侧)
|
|
|
|
|
|
def test_provider_reads_all_four_prompts():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
assert "$existing_info" in p.extractor and "$messages" in p.extractor
|
|
assert "$profile" in p.evaluator and "$recent_messages" in p.evaluator
|
|
|
|
|
|
def test_default_provider_is_jiancai_seed():
|
|
# 旧测试/默认路径的代码级默认 = jiancai 出厂(Path(__file__) 锚定,不随 cwd 漂移)
|
|
assert DEFAULT_PROVIDER.system == PromptProvider.from_dir(JC_PROMPTS).system
|
|
|
|
|
|
# ---- $DEFLECTION_POOL 注入(单一事实源) ----------------------------------
|
|
|
|
def test_system_prompt_injects_deflection_pool():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
wall_json = _wall("jiancai")
|
|
sysprompt = p.build_system(wall_deflections=wall_json["deflections"])
|
|
for lang, lines in wall_json["deflections"].items():
|
|
assert lines[0] in sysprompt # 话术池注入 = 单一事实源
|
|
assert "$DEFLECTION_POOL" not in sysprompt
|
|
assert "HESUPIPE" in sysprompt # 事实卡拼接保留
|
|
|
|
|
|
def test_system_md_carries_placeholder():
|
|
# 防漂移锚(审查 I-2): glob 遍历全部租户 system.md——第三租户落地即自动
|
|
# 纳管(约定同步落档 seed_profiles/README.md: 缺占位符 = 话术池静默不注入);
|
|
# 空 glob 恒绿 = 假守护,先断言至少一份
|
|
system_mds = sorted((REPO / "seed_profiles").glob("*/prompts/system.md"))
|
|
assert system_mds
|
|
for md in system_mds:
|
|
assert "$DEFLECTION_POOL" in md.read_text(encoding="utf-8"), md
|
|
|
|
|
|
def test_build_system_appends_facts_after_system():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
out = p.build_system(wall_deflections=_wall("jiancai")["deflections"])
|
|
# 顺序: 行为 prompt(含注入话术池) → 事实卡(与旧 _BASE_CHAT_PROMPT 拼接同构)
|
|
assert out.index("## 语言规则") < out.index("# HESUPIPE 公司事实卡")
|
|
|
|
|
|
def test_build_system_auto_tenant_pool():
|
|
# auto 墙只有 en/zh 池(无 es)——注入按实际池语言枚举,不猜不补
|
|
p = PromptProvider.from_dir(AUTO_PROMPTS)
|
|
wall = _wall("auto")
|
|
sysprompt = p.build_system(wall_deflections=wall["deflections"])
|
|
for lang, lines in wall["deflections"].items():
|
|
assert lines[0] in sysprompt
|
|
assert "$DEFLECTION_POOL" not in sysprompt
|
|
assert "- ES:" not in sysprompt
|
|
|
|
|
|
def test_build_system_injection_is_language_grouped():
|
|
# 注入格式: 按语言分组一行(- EN: … / - ZH: …),紧凑不膨胀
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
out = p.build_system(wall_deflections=_wall("jiancai")["deflections"])
|
|
assert "- EN:" in out and "- ZH:" in out and "- ES:" in out
|
|
# 分组行序: EN 在 ZH 前(确定性输出,防未来语言序随机漂移断言抖动)
|
|
assert out.index("- EN:") < out.index("- ZH:") < out.index("- ES:")
|
|
|
|
|
|
def test_injected_pool_lines_are_wall_clean():
|
|
# 注入让话术池进入 system prompt(AI 会照抄)——池线必须过两墙;
|
|
# 既有 test_spanish_pools_are_wall_clean 只盖 ES,这里补全两租户全语言
|
|
from app.engine.services.boundary_filter import (
|
|
no_contact_value, violates_boundary,
|
|
)
|
|
for slug in ("jiancai", "auto"):
|
|
for lines in _wall(slug)["deflections"].values():
|
|
for line in lines:
|
|
assert not violates_boundary(line), line[:40]
|
|
assert not no_contact_value(line), line[:40]
|
|
|
|
|
|
# ---- extractor/evaluator 组装(自 AIService 原样搬入) ----------------------
|
|
|
|
def test_build_extractor_replaces_placeholders():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
out = p.build_extractor(
|
|
{"目标国家": "Egypt"},
|
|
[{"role": "user", "content": "hi"},
|
|
{"role": "assistant", "content": "hello there"}],
|
|
)
|
|
assert "$existing_info" not in out and "$messages" not in out
|
|
assert '"目标国家": "Egypt"' in out # json.dumps(ensure_ascii=False)
|
|
assert "客户: hi" in out and "顾问: hello there" in out
|
|
|
|
|
|
def test_build_evaluator_replaces_placeholders_and_caps_messages():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
msgs = [{"role": "user", "content": f"m{i}"} for i in range(10)]
|
|
out = p.build_evaluator({"intent_level": "pending"}, msgs)
|
|
assert "$profile" not in out and "$recent_messages" not in out
|
|
assert '"intent_level": "pending"' in out
|
|
assert "客户: m9" in out # 取最近 6 条
|
|
assert "客户: m0" not in out and "客户: m3" not in out
|
|
|
|
|
|
def test_build_extractor_handles_none_info():
|
|
p = PromptProvider.from_dir(JC_PROMPTS)
|
|
out = p.build_extractor(None, [])
|
|
assert "{}" in out # existing_info or {} 语义保留
|
|
|
|
|
|
# ---- from_config(Phase 2 Task 4: 快照 prompts 组构造)----------------------
|
|
|
|
def test_from_config_missing_key_raises():
|
|
# 缺一键 → 显性 ValueError 带键名(不留到 build_system 才炸)
|
|
with pytest.raises(ValueError, match="company_facts"):
|
|
PromptProvider.from_config(
|
|
{"system": "s", "extractor": "e", "evaluator": "v"})
|
|
|
|
|
|
def test_from_config_non_string_key_raises():
|
|
# 值非字符串(JSON null 等)同样显性拒绝
|
|
with pytest.raises(ValueError, match="system"):
|
|
PromptProvider.from_config(
|
|
{"system": None, "company_facts": "c",
|
|
"extractor": "e", "evaluator": "v"})
|
|
|
|
|
|
def test_from_config_non_dict_raises():
|
|
# 入口类型守卫: Task 6 的 PUT prompts 会把请求体直传这里——字符串/
|
|
# 列表进来必须是 ValueError,不能让 .get 的 AttributeError 漏成 500
|
|
for bad in ("system=...", ["system", "company_facts"], 42, None):
|
|
with pytest.raises(ValueError, match="dict"):
|
|
PromptProvider.from_config(bad)
|
|
|
|
|
|
def test_from_config_equals_from_dir():
|
|
d = JC_PROMPTS
|
|
cfg = {name: (d / f"{name}.md").read_text(encoding="utf-8")
|
|
for name in ("system", "company_facts", "extractor", "evaluator")}
|
|
assert PromptProvider.from_config(cfg) == PromptProvider.from_dir(JC_PROMPTS)
|
|
|
|
|
|
# ---- runtime 接线 --------------------------------------------------------
|
|
|
|
def test_registry_runtime_carries_prompts(tmp_path):
|
|
from app.platform.registry import TenantRegistry
|
|
reg = TenantRegistry.load_from_dir(REPO / "seed_profiles", data_dir=tmp_path)
|
|
by_slug = {rt.config.slug: rt for rt in reg.runtimes()}
|
|
assert "管道" in by_slug["jiancai"].prompts.system
|
|
assert "汽车" in by_slug["auto"].prompts.system
|
|
# runtime 的 prompts 即租户目录四件套(与手工 from_dir 等价)
|
|
assert by_slug["jiancai"].prompts.system == PromptProvider.from_dir(JC_PROMPTS).system
|
|
|
|
|
|
def test_conversation_manager_runtime_is_required():
|
|
# Task 5b 质量审查关门建议: runtime 从可选转必填——缺参必须显性炸
|
|
from app.engine.services.conversation import ConversationManager
|
|
with pytest.raises(TypeError):
|
|
ConversationManager(db=None)
|