200 lines
9.4 KiB
Python
200 lines
9.4 KiB
Python
"""Prompt routing checks with mocked model HTTP and an isolated message database."""
|
||
|
||
import json
|
||
|
||
import httpx
|
||
import pytest
|
||
from app.core.config import get_settings
|
||
from app.core.exceptions import AppException
|
||
from app.models.entities import (
|
||
Conversation,
|
||
ConversationMessage,
|
||
IntegrationAssessment,
|
||
IntegrationConversation,
|
||
UserAccount,
|
||
)
|
||
from app.services.conversation_prompts import reply_system_messages, resolve_prompt_selection
|
||
from app.services.conversations import ConversationService
|
||
from app.services.llm_client import LLMClient
|
||
from sqlalchemy import JSON, BigInteger, Integer, MetaData, select
|
||
from sqlalchemy.dialects.postgresql import JSONB
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
|
||
@pytest.mark.parametrize("invalid_key", [None, "../../.env", {"file": "03-成本定价.md"}, ["cost_pricing"]])
|
||
def test_untrusted_prompt_key_cannot_select_a_file(invalid_key) -> None:
|
||
result = resolve_prompt_selection(
|
||
{"prompt_key": invalid_key, "prompt_file": "../../.env"},
|
||
[],
|
||
)
|
||
|
||
assert result["prompt_key"] == "product_manager"
|
||
assert result["prompt_route_source"] == "fallback"
|
||
assert result["prompt_file"] == "开品产品经理-美妆护肤-角色Prompt.md"
|
||
assert "美妆护肤电商开品产品经理" in reply_system_messages(result)[0]["content"]
|
||
|
||
|
||
def test_new_legacy_turn_does_not_revive_an_old_specialist() -> None:
|
||
messages = [
|
||
ConversationMessage(
|
||
role="assistant",
|
||
content="历史定价分析",
|
||
structured_data={"semantic": {"prompt_key": "cost_pricing"}},
|
||
),
|
||
ConversationMessage(role="assistant", content="后来的普通回复", structured_data={}),
|
||
]
|
||
|
||
result = resolve_prompt_selection({"prompt_key": "inherit"}, messages)
|
||
|
||
assert result["prompt_key"] == "product_manager"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_workflow_supplement_does_not_trigger_supplier_lookup() -> None:
|
||
service = object.__new__(ConversationService)
|
||
service.session = object() # Any database lookup would fail this test.
|
||
semantics = {"intent": "general", "prompt_key": "selection_insight", "risk_requested": False}
|
||
|
||
result = await service._normalize_follow_up_semantics("抖音", semantics, Conversation(messages=[]))
|
||
|
||
assert result == semantics
|
||
assert service._response_type("抖音", result) == "general"
|
||
assert service._normalize_semantics("包装成本和供应商报价", result)["risk_requested"] is False
|
||
|
||
|
||
@pytest.fixture
|
||
async def conversation_sessions(tmp_path, monkeypatch):
|
||
get_settings.cache_clear()
|
||
monkeypatch.setenv("RISK_LLM_API_KEY", "test-key")
|
||
monkeypatch.setenv("RISK_LLM_BASE_URL", "https://llm.test/v1")
|
||
monkeypatch.setenv("RISK_LLM_MODEL", "test-model")
|
||
monkeypatch.setenv("RISK_LLM_FALLBACK_API_KEY", "")
|
||
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'conversations.db'}")
|
||
schema = MetaData()
|
||
for model in (UserAccount, Conversation, ConversationMessage, IntegrationAssessment, IntegrationConversation):
|
||
copied = model.__table__.to_metadata(schema)
|
||
for column in copied.columns:
|
||
if isinstance(column.type, JSONB):
|
||
column.type = JSON()
|
||
elif isinstance(column.type, BigInteger):
|
||
column.type = Integer()
|
||
async with engine.begin() as connection:
|
||
await connection.run_sync(schema.create_all)
|
||
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
||
try:
|
||
async with sessions() as session:
|
||
session.add(UserAccount(id=1, emp_no="prompt-user", name="测试用户", role="user"))
|
||
await session.commit()
|
||
await ConversationService(session).create(await session.get(UserAccount, 1))
|
||
await session.commit()
|
||
yield sessions
|
||
finally:
|
||
await engine.dispose()
|
||
get_settings.cache_clear()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("delivery", ["stream", "send"])
|
||
async def test_semantic_routing_inherits_and_switches_across_database_sessions(conversation_sessions, delivery) -> None:
|
||
turns = [
|
||
("帮我判断这款洗发水值不值得做", "selection_insight", "selection_insight", "选品洞察专家"),
|
||
("抖音，89元", "inherit", "selection_insight", "选品洞察专家"),
|
||
("售价和毛利应该怎么定", "cost_pricing", "cost_pricing", "成本定价专家"),
|
||
("帮我审查功效宣称", "compliance_risk", "compliance_risk", "合规风控专家"),
|
||
("继续", "inherit", "compliance_risk", "合规风控专家"),
|
||
("给这款洗发水写主图文案", "listing_selling_points", "listing_selling_points", "上架卖点策划专家"),
|
||
("这是立项书，能过评审吗", "project_review", "project_review", "立项评审专家"),
|
||
]
|
||
experts = {item[3] for item in turns}
|
||
request_count = 0
|
||
|
||
async def handler(request: httpx.Request) -> httpx.Response:
|
||
nonlocal request_count
|
||
request_count += 1
|
||
body = json.loads(request.content)
|
||
current_text = next(message["content"] for message in reversed(body["messages"]) if message["role"] == "user")
|
||
index = next(i for i, item in enumerate(turns) if item[0] == current_text)
|
||
_, requested_key, resolved_key, expert = turns[index]
|
||
if not body["stream"]:
|
||
assert "prompt_key" in body["messages"][0]["content"]
|
||
assert body["response_format"] == {"type": "json_object"}
|
||
previous = turns[index - 1][2] if index else "product_manager"
|
||
assert f"prompt_key={previous}" in body["messages"][1]["content"]
|
||
return httpx.Response(
|
||
200,
|
||
json={
|
||
"choices": [
|
||
{
|
||
"message": {
|
||
"content": json.dumps(
|
||
{
|
||
"intent": "general",
|
||
"risk_requested": False,
|
||
"category": "洗发水",
|
||
"prompt_key": requested_key,
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
}
|
||
},
|
||
],
|
||
},
|
||
)
|
||
systems = "\n".join(message["content"] for message in body["messages"] if message["role"] == "system")
|
||
assert "美妆护肤电商开品产品经理" in systems
|
||
assert {name for name in experts if name in systems} == {expert}
|
||
assert f'"prompt_key": "{resolved_key}"' in systems
|
||
assert "MISSING_MATERIAL" in systems
|
||
assert sum(message["content"] == current_text for message in body["messages"]) == 1
|
||
if index:
|
||
assert any(message["content"] == turns[index - 1][0] for message in body["messages"])
|
||
chunk = {"choices": [{"delta": {"content": f"{expert}：请补充本轮材料。"}}]}
|
||
return httpx.Response(
|
||
200,
|
||
headers={"content-type": "text/event-stream"},
|
||
content=(f"data: {json.dumps(chunk, ensure_ascii=False)}\n\ndata: [DONE]\n\n").encode(),
|
||
)
|
||
|
||
for index, (text, requested, resolved, _) in enumerate(turns):
|
||
# A fresh session for every request proves the route comes from persisted history.
|
||
async with conversation_sessions() as session:
|
||
service = ConversationService(session)
|
||
service.llm = LLMClient(transport=httpx.MockTransport(handler))
|
||
user = await session.get(UserAccount, 1)
|
||
if delivery == "stream":
|
||
events = [event async for event in service.stream(1, text, user)]
|
||
assert [event["type"] for event in events] == ["start", "semantic", "delta", "done"]
|
||
semantic = events[1]["data"]
|
||
else:
|
||
data = await service.send(1, text, user)
|
||
assert data.message_count == 2 * (index + 1)
|
||
semantic = data.messages[-1].semantic
|
||
assert semantic["prompt_key"] == resolved
|
||
assert semantic["prompt_route_source"] == ("inherited" if requested == "inherit" else "semantic")
|
||
async with conversation_sessions() as session:
|
||
stored = await session.scalar(
|
||
select(ConversationMessage)
|
||
.where(ConversationMessage.role == "assistant")
|
||
.order_by(ConversationMessage.id.desc()),
|
||
)
|
||
assert stored.structured_data["semantic"] == semantic
|
||
assert "responseCard" not in stored.structured_data
|
||
# Each turn still uses only semantic extraction + text generation.
|
||
assert request_count == len(turns) * 2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_non_stream_model_failure_keeps_user_message(conversation_sessions) -> None:
|
||
async def unavailable(request: httpx.Request) -> httpx.Response:
|
||
return httpx.Response(503, json={"error": "unavailable"})
|
||
|
||
async with conversation_sessions() as session:
|
||
service = ConversationService(session)
|
||
service.llm = LLMClient(transport=httpx.MockTransport(unavailable))
|
||
with pytest.raises(AppException) as error:
|
||
await service.send(1, "帮我定价", await session.get(UserAccount, 1))
|
||
assert error.value.status_code == 503
|
||
async with conversation_sessions() as session:
|
||
messages = list(await session.scalars(select(ConversationMessage)))
|
||
assert [(message.role, message.content) for message in messages] == [("user", "帮我定价")]
|