kx_guo
b0b6ef014f
- 保留 pgvector 与混合检索能力，返回 TOP 5 相似案例\n- 将相似案例上下文传入 AI 预防建议生成\n- 修复风险评估调用缺少 similar_cases 参数导致的运行报错\n- 保留 Embedding 配置、数据库迁移、回填脚本及 RAG 测试\n- 恢复基础字符串数组格式的 AI 预防建议协议
88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
from datetime import date
|
|
|
|
import pytest
|
|
from app.services.rag import AnomalyCaseRAG
|
|
|
|
|
|
def _row(**values):
|
|
return {
|
|
"id": 1,
|
|
"date": date(2026, 1, 10),
|
|
"product_name": "洗发水 500ml",
|
|
"product_category": "洗护类",
|
|
"material_type": "软管",
|
|
"supplier": "祥印",
|
|
"factory": "晶品",
|
|
"stage_new": "产品开发",
|
|
"anomaly_type": "密封性异常",
|
|
"description": "软管漏液",
|
|
"solution": "增加密封性测试",
|
|
"root_cause": "封口参数不稳定",
|
|
"defect_level": "B",
|
|
**values,
|
|
}
|
|
|
|
|
|
def test_rag_terms_and_structured_text_score() -> None:
|
|
terms = AnomalyCaseRAG._terms("软管密封性异常", "洗发水", "软管")
|
|
score = AnomalyCaseRAG._score(
|
|
_row(),
|
|
query="洗发水软管密封性异常",
|
|
category="洗发水",
|
|
material="软管",
|
|
supplier="祥印",
|
|
factory=None,
|
|
stage="产品开发",
|
|
anomaly_type="密封性异常",
|
|
terms=terms,
|
|
)
|
|
|
|
assert "密封性" in terms
|
|
assert score > 20
|
|
case = AnomalyCaseRAG._to_case(_row())
|
|
assert case.product == "洗发水 500ml"
|
|
assert case.problem == "密封性异常"
|
|
assert case.solution == "增加密封性测试"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rag_does_not_call_embedding_until_vectors_are_stored() -> None:
|
|
class FakeMappings:
|
|
def all(self):
|
|
return [_row()]
|
|
|
|
class FakeResult:
|
|
def mappings(self):
|
|
return FakeMappings()
|
|
|
|
class FakeSession:
|
|
async def execute(self, _statement):
|
|
return FakeResult()
|
|
|
|
rag = AnomalyCaseRAG(FakeSession(), top_k=1)
|
|
|
|
async def vector_column_ready():
|
|
return True
|
|
|
|
async def has_stored_vectors():
|
|
return False
|
|
|
|
async def embedding_must_not_run(_query):
|
|
raise AssertionError("没有已回填向量时不应调用 Embedding")
|
|
|
|
rag._vector_column_ready = vector_column_ready
|
|
rag._has_stored_vectors = has_stored_vectors
|
|
rag._query_embedding = embedding_must_not_run
|
|
|
|
rows = await rag._load_rows(
|
|
"洗发水软管密封性异常",
|
|
category="洗发水",
|
|
material="软管",
|
|
supplier=None,
|
|
factory=None,
|
|
stage=None,
|
|
anomaly_type=None,
|
|
)
|
|
|
|
assert len(rows) == 1
|