195 lines
11 KiB
Python
195 lines
11 KiB
Python
"""Independent demand/role/expression axes and deterministic business suggestions."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from apps.server.analysis import PROMPT_VERSION, RESULT_VERSION, SYSTEM_PROMPT, checked_results
|
|
from apps.server.db import digest, encoded, identifier, now
|
|
from apps.server.model_client import ModelCallError
|
|
from apps.server.tests.test_analysis import analysis_client, answer, attach, client, pg_client, seed, wire
|
|
|
|
|
|
CONDITIONS = {"service": "装修服务", "target_customer": "有装修需求的业主", "region": None,
|
|
"region_scope": "unrestricted", "exclude_service_promotions": True}
|
|
|
|
|
|
def source(text, ref="note", media=False):
|
|
return {"source_ref": ref, "text": text, "media_uninterpreted": media}
|
|
|
|
|
|
def result(text, *, demand="explicit", role="suspected_demand_side", expressions=("self_need",), relevance="relevant", region=None):
|
|
return {"source_ref": "note", "demand": demand, "evidence": [text], "role": role,
|
|
"role_evidence": [] if role == "unknown" else [text],
|
|
"expressions": [{"type": expression, "evidence": [] if expression == "unknown" else [text]} for expression in expressions],
|
|
"business_relevance": {"value": relevance, "evidence": [] if relevance == "unclear" else [text]},
|
|
"stage": None, "region": {"value": region, "evidence": text} if region else None, "budget": None}
|
|
|
|
|
|
def check(item, text, conditions=None):
|
|
return checked_results(encoded({"results": [item]}), [source(text)], "completed", conditions or CONDITIONS)[0]
|
|
|
|
|
|
def test_axes_self_need_coexists_with_question_advice_and_service_provider():
|
|
text = "我家准备装修，问大家家装顺序。"
|
|
current = check(result(text, expressions=("self_need", "question", "seek_advice")), text)
|
|
assert current["demand"] == "explicit" and current["match"] == "matched" and current["next_step"] == "review_author_need"
|
|
assert {entry["type"] for entry in current["expressions"]} == {"self_need", "question", "seek_advice"}
|
|
text = "我是设计师可以提供装修服务，同时我家也准备装修。"
|
|
provider = check(result(text, role="suspected_service_provider", expressions=("service_offer", "self_need")), text)
|
|
assert provider["match"] == "matched" and provider["next_step"] == "review_author_need"
|
|
assert "service_promotion_excluded" not in provider["match_reasons"]
|
|
|
|
|
|
@pytest.mark.parametrize("text,role,expressions,next_step", [
|
|
("你是哪里的嘛？", "unknown", ("question",), "skip"),
|
|
("开始装修了吗？", "unknown", ("question",), "skip"),
|
|
("我有设计师朋友，可以推荐给你", "suspected_service_provider", ("service_offer",), "inspect_comments"),
|
|
("展示一下我们装修公司的案例", "suspected_service_provider", ("service_offer", "case_share"), "inspect_comments"),
|
|
("建议先确定需求，再考虑半包", "participant", ("advice",), "skip"),
|
|
])
|
|
def test_axes_questions_offers_advice_do_not_inherit_authors_need(text, role, expressions, next_step):
|
|
current = check(result(text, demand="discussion", role=role, expressions=expressions), text)
|
|
assert current["match"] == "not_matched" and current["next_step"] == next_step
|
|
assert "no_self_need_signal" in current["match_reasons"]
|
|
if "service_offer" in expressions:
|
|
assert "service_promotion_excluded" in current["match_reasons"]
|
|
|
|
|
|
@pytest.mark.parametrize("scope,condition_region,own_region,match,reason", [
|
|
("unrestricted", None, None, "matched", "region_unrestricted"),
|
|
("unspecified", None, None, "insufficient", "region_unspecified"),
|
|
("specified", "杭州", None, "insufficient", "region_missing"),
|
|
("specified", "杭州", "杭州", "matched", "region_exact"),
|
|
("specified", "杭州", "余杭", "insufficient", "region_potential_mismatch"),
|
|
("specified", "杭州", "上海", "insufficient", "region_potential_mismatch"),
|
|
])
|
|
def test_axes_region_missing_and_different_words_are_not_guessed(scope, condition_region, own_region, match, reason):
|
|
text = "我家准备装修" + ("，地点" + own_region if own_region else "")
|
|
current = check(result(text, region=own_region), text, {**CONDITIONS, "region_scope": scope, "region": condition_region})
|
|
assert current["match"] == match and reason in current["match_reasons"]
|
|
assert current["next_step"] == ("review_author_need" if match == "matched" else "need_more_context")
|
|
|
|
|
|
def test_axes_possible_need_comment_binding_irrelevant_unclear_and_uninterpreted_media():
|
|
text = "我也考虑装修，预算还没定"
|
|
item = result(text, demand="possible")
|
|
item["source_ref"] = "comment_2"
|
|
current = checked_results(encoded({"results": [item]}), [source(text, ref="comment_2")], "partial", CONDITIONS)[0]
|
|
assert current["match"] == "possible" and current["next_step"] == "review_comment_need"
|
|
assert "reading_partial" in current["match_reasons"]
|
|
assert check(result(text, relevance="irrelevant"), text)["match"] == "not_matched"
|
|
assert check(result(text, relevance="unclear"), text)["match"] == "insufficient"
|
|
media = checked_results('{"results":[]}', [source("", media=True)], "completed", CONDITIONS)[0]
|
|
assert media["role"] == "unknown" and media["demand"] == media["match"] == "insufficient"
|
|
assert media["business_relevance"] == {"value": "unclear", "evidence": []} and media["next_step"] == "need_more_context"
|
|
assert "media_uninterpreted" in media["missing_information"]
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", [
|
|
lambda item: item.update(role_evidence=["其他来源的服务"]),
|
|
lambda item: item["expressions"][0].update(evidence=["其他来源的需求"]),
|
|
lambda item: item["business_relevance"].update(evidence=["商家条件中的装修服务"]),
|
|
lambda item: item["business_relevance"].update(evidence=[]),
|
|
lambda item: item.update(role="confirmed_owner"),
|
|
lambda item: item["expressions"].append(copy.deepcopy(item["expressions"][0])),
|
|
lambda item: item["expressions"].append({"type": "unknown", "evidence": []}),
|
|
lambda item: item.update(expressions=[{"type": "question", "evidence": ["我家准备装修"]}]),
|
|
lambda item: item.update(demand="discussion"),
|
|
lambda item: item.update(next_step="send_message"),
|
|
lambda item: item.update(match="matched"),
|
|
])
|
|
def test_axes_reject_fabricated_cross_source_dimensions_and_model_actions(mutation):
|
|
text = "我家准备装修"
|
|
item = result(text)
|
|
mutation(item)
|
|
with pytest.raises(ModelCallError) as error:
|
|
check(item, text)
|
|
assert error.value.code == "MODEL_OUTPUT_INVALID"
|
|
|
|
|
|
def test_axes_no_own_need_rejects_service_area_as_own_project_region():
|
|
text = "我在杭州提供装修服务"
|
|
item = result(text, demand="discussion", role="suspected_service_provider", expressions=("service_offer",), region="杭州")
|
|
with pytest.raises(ModelCallError):
|
|
check(item, text)
|
|
# A short question cannot support a service-provider role without an offer.
|
|
item = result("哪里", demand="discussion", role="suspected_service_provider", expressions=("question",))
|
|
with pytest.raises(ModelCallError):
|
|
check(item, "哪里")
|
|
assert "设计师所在地、服务覆盖地" in SYSTEM_PROMPT and "role应unknown" in SYSTEM_PROMPT
|
|
|
|
|
|
class Briefs:
|
|
def __init__(self, owner):
|
|
self.owner = owner
|
|
self.records = {}
|
|
self.reads = []
|
|
|
|
def add(self, key, **conditions):
|
|
value = {**CONDITIONS, **conditions}
|
|
self.records[key] = {"id": key, "digest": digest(value), "conditions": value}
|
|
return key
|
|
|
|
def get_confirmed(self, key, owner):
|
|
self.reads.append((key, owner))
|
|
if owner != self.owner or key not in self.records:
|
|
raise HTTPException(404, "不存在")
|
|
return self.records[key]
|
|
|
|
|
|
def test_axes_context_digest_reuse_old_records_and_sources_stay_immutable(analysis_client):
|
|
client = analysis_client
|
|
reading_id, owner = seed(client)
|
|
calls = []
|
|
|
|
def handler(request):
|
|
import json
|
|
payload = json.loads(json.loads(request.content)["messages"][1]["content"])
|
|
calls.append(payload)
|
|
assert set(payload) == {"business_context", "sources"}
|
|
assert not any("private-" in encoded(value) for value in payload.values())
|
|
return wire(answer(payload))
|
|
|
|
service = attach(client, handler)
|
|
briefs = service.business_briefs = Briefs(owner)
|
|
first_id = briefs.add("brief-first")
|
|
other_id = briefs.add("brief-other", region="杭州", region_scope="specified")
|
|
store = service.store
|
|
original_reading = store.one("SELECT data_json FROM reading_runs WHERE id=$1", (reading_id,))
|
|
old_id = identifier("analysis")
|
|
old = {"status": "completed", "reading_digest": "old", "prompt_version": "renovation-demand-v2", "result_version": "demand-result-v1", "results": [{"demand": "discussion"}]}
|
|
store.execute("INSERT INTO analysis_runs(id,owner_id,reading_run_id,data_json,created_at) VALUES($1,$2,$3,$4,$5)", (old_id, owner, reading_id, encoded(old), now()))
|
|
original_old = service.get(old_id, owner)
|
|
first = asyncio.run(service.create(reading_id, owner, first_id))
|
|
assert first["status"] == "completed" and first["prompt_version"] == PROMPT_VERSION and first["result_version"] == RESULT_VERSION
|
|
assert first["context"]["kind"] == "confirmed_business_brief" and first["context"]["business_brief_digest"] == briefs.records[first_id]["digest"]
|
|
assert asyncio.run(service.create(reading_id, owner, first_id)) == first and len(calls) == 1
|
|
second = asyncio.run(service.create(reading_id, owner, other_id))
|
|
assert second["id"] != first["id"] and len(calls) == 2
|
|
assert second["context"]["context_digest"] != first["context"]["context_digest"]
|
|
assert second["results"][0]["match"] == "insufficient" and "region_missing" in second["results"][0]["match_reasons"]
|
|
assert store.one("SELECT data_json FROM reading_runs WHERE id=$1", (reading_id,)) == original_reading
|
|
assert service.get(old_id, owner) == original_old and "context" not in original_old
|
|
assert service.get(first["id"], owner) == first
|
|
with pytest.raises(HTTPException):
|
|
asyncio.run(service.create(reading_id, owner, "draft-or-other-owner"))
|
|
assert len(calls) == 2
|
|
|
|
|
|
def test_axes_implicit_reading_context_and_explicit_override(client):
|
|
reading_id, owner = seed(client)
|
|
service = attach(client, lambda _: wire({"results": []}))
|
|
briefs = service.business_briefs = Briefs(owner)
|
|
first = briefs.add("brief-first")
|
|
other = briefs.add("brief-other", region_scope="unspecified")
|
|
reading = service.reading.get(reading_id, owner)
|
|
reading["business_brief_id"] = first
|
|
assert service.context(reading, owner)["business_brief_id"] == first
|
|
assert service.context(reading, owner, other)["business_brief_id"] == other
|
|
legacy = service.context(service.reading.get(reading_id, owner), owner)
|
|
assert legacy["kind"] == "discovery_conditions" and legacy["conditions"]["region_scope"] == "unspecified"
|