PMFlow/backend/tests/test_adaptive_rag_external_...

490 lines
17 KiB
Python

from __future__ import annotations
from datetime import timedelta
import pytest
import pytest_asyncio
from sqlalchemy import JSON
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.models.base import Base
from app.models.feishu_reply_outbox import FeishuReplyOutbox
from app.models.ingest_event import IngestEvent
from app.models.qa_record import QARecord
from app.models.qa_session import QASession
from app.models.qa_session_state import QASessionState
from app.services.adaptive_rag_service.mixin_core_intent import AdaptiveRAGCoreIntentMixin
from app.services.adaptive_rag_service.mixin_followup_anchor import (
AdaptiveRAGFollowupAnchorMixin,
FollowupAnchorResolution,
)
from app.services.adaptive_rag_service.mixin_qa_session_core import AdaptiveRAGQASessionCoreMixin
from app.services.adaptive_rag_service.mixin_session_state import AdaptiveRAGSessionStateMixin
from app.utils.timezone import now_cn_naive
def _use_sqlite_json_types() -> None:
for table in (QARecord.__table__, QASessionState.__table__, IngestEvent.__table__):
for column in table.columns:
if column.name in {
"sources",
"domains_used",
"citations",
"sources_used",
"evidence_pack",
"tool_trace",
"retrieval_metrics",
"last_entity_ids",
"rewrite_history",
"pending_clarification",
"routing_snapshot",
}:
column.type = JSON()
if table.name == "ingest_events" and column.name == "payload":
column.type = JSON()
@pytest_asyncio.fixture
async def db_session():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
_use_sqlite_json_types()
async with engine.begin() as conn:
await conn.run_sync(
lambda sync_conn: Base.metadata.create_all(
sync_conn,
tables=[
QASession.__table__,
QASessionState.__table__,
QARecord.__table__,
FeishuReplyOutbox.__table__,
IngestEvent.__table__,
],
)
)
maker = async_sessionmaker(engine, expire_on_commit=False)
async with maker() as session:
yield session
await engine.dispose()
class _FakeContextCache:
async def get_session_messages(self, *, tenant_id: int, session_id: int | None):
_ = (tenant_id, session_id)
return []
async def append_session_messages(self, *, tenant_id: int, session_id: int | None, question: str, answer: str):
_ = (tenant_id, session_id, question, answer)
return None
@staticmethod
def messages_to_turns(messages):
_ = messages
return []
class _Harness(
AdaptiveRAGCoreIntentMixin,
AdaptiveRAGQASessionCoreMixin,
AdaptiveRAGSessionStateMixin,
AdaptiveRAGFollowupAnchorMixin,
):
def __init__(self, db) -> None:
super().__init__(db, tenant_id=1)
self.context_cache = _FakeContextCache()
@pytest.mark.asyncio
async def test_external_conversation_key_creates_and_reuses_real_qa_session(db_session) -> None:
service = _Harness(db_session)
session_id, turn_index = await service._touch_session_for_new_record(
asked_by=None,
asked_via="feishu",
project_id=None,
question="帮我解释下系统架构",
external_conversation_key="feishu:chat-1:root-1",
)
assert session_id is not None
assert turn_index == 1
reused_session_id, reused_turn_index = await service._touch_session_for_new_record(
asked_by=None,
asked_via="feishu",
project_id=None,
question="系统有什么模型?",
external_conversation_key="feishu:chat-1:root-1",
)
assert reused_session_id == session_id
assert reused_turn_index == 2
row = await db_session.get(QASession, int(session_id))
assert row is not None
assert row.user_id is None
assert row.external_conversation_key == "feishu:chat-1:root-1"
@pytest.mark.asyncio
async def test_meta_followup_prefers_single_recent_reply_candidate(db_session) -> None:
service = _Harness(db_session)
session = QASession(
tenant_id=1,
user_id=None,
title="系统架构",
external_conversation_key="feishu:chat-1:root-1",
started_at=now_cn_naive(),
last_message_at=now_cn_naive(),
message_count=1,
)
db_session.add(session)
await db_session.flush()
db_session.add(
QARecord(
tenant_id=1,
question="帮我解释下系统的架构",
answer="PMFlow 系统采用前后端分离架构。",
sources=[],
intent="question",
domains_used=[],
response_time_ms=10,
asked_by=None,
asked_via="feishu",
project_id=None,
session_id=int(session.id),
turn_index=1,
)
)
await db_session.commit()
state = await service._ensure_session_state(session_id=int(session.id))
resolution = await service._resolve_followup_anchor(
question="我问你这个是哪个模型执行回复的?",
session_id=int(session.id),
context_session_id=int(session.id),
session_state=state,
)
assert resolution.anchor_mode == "implicit_recent_reply"
assert resolution.anchor_reply_id is not None
assert resolution.is_meta_followup is True
assert resolution.candidate_replies == []
@pytest.mark.asyncio
async def test_ambiguous_recent_reply_returns_numbered_candidates(db_session) -> None:
service = _Harness(db_session)
now = now_cn_naive()
session = QASession(
tenant_id=1,
user_id=None,
title="新会话",
external_conversation_key="feishu:chat-1:root-1",
started_at=now,
last_message_at=now,
message_count=2,
)
db_session.add(session)
await db_session.flush()
db_session.add_all(
[
QARecord(
tenant_id=1,
question="帮我解释下系统架构",
answer="PMFlow 系统采用前后端分离架构。",
sources=[],
intent="question",
domains_used=[],
response_time_ms=10,
asked_by=None,
asked_via="feishu",
project_id=None,
session_id=int(session.id),
turn_index=1,
created_at=now - timedelta(minutes=1),
),
QARecord(
tenant_id=1,
question="系统有什么模型?",
answer="当前有 low/high 两层模型路由。",
sources=[],
intent="question",
domains_used=[],
response_time_ms=12,
asked_by=None,
asked_via="feishu",
project_id=None,
session_id=int(session.id),
turn_index=2,
created_at=now,
),
]
)
await db_session.commit()
state = await service._ensure_session_state(session_id=int(session.id))
resolution = await service._resolve_followup_anchor(
question="那你刚才那条是哪个模型回复的?",
session_id=int(session.id),
context_session_id=int(session.id),
session_state=state,
)
assert resolution.anchor_mode == "ambiguous_candidates"
assert resolution.anchor_reply_id is None
assert len(resolution.candidate_replies) == 2
assert resolution.candidate_replies[0]["index"] == 1
assert resolution.candidate_replies[1]["index"] == 2
@pytest.mark.asyncio
async def test_pending_clarification_selection_restores_original_question(db_session) -> None:
service = _Harness(db_session)
session = QASession(
tenant_id=1,
user_id=None,
title="新会话",
external_conversation_key="feishu:chat-1:root-1",
started_at=now_cn_naive(),
last_message_at=now_cn_naive(),
message_count=1,
)
db_session.add(session)
await db_session.flush()
state = await service._ensure_session_state(session_id=int(session.id))
state.pending_clarification = {
"kind": "followup_anchor_selection",
"original_question": "我问你这个是哪个模型执行回复的?",
"candidates": [
{"index": 1, "qa_id": 101, "label": "系统架构那条"},
{"index": 2, "qa_id": 102, "label": "系统模型那条"},
],
}
await db_session.commit()
resolved = service._consume_pending_clarification_selection(
question="第二条",
session_state=state,
)
assert resolved is not None
assert resolved["selected_qa_id"] == 102
assert resolved["effective_question"] == "我问你这个是哪个模型执行回复的?"
@pytest.mark.asyncio
async def test_explicit_reply_to_message_id_resolves_anchor_without_session_history(db_session) -> None:
service = _Harness(db_session)
now = now_cn_naive()
db_session.add(
IngestEvent(
tenant_id=1,
channel="feishu_message",
event_key="om_user_message_1",
status="done",
payload={"message_text": "帮我解释下系统的架构"},
)
)
db_session.add(
FeishuReplyOutbox(
tenant_id=1,
trace_id="trace-1",
source_message_id="om_user_message_1",
sent_message_id="om_bot_reply_1",
chat_id="oc-chat-1",
message_text='<at user_id="ou-1">林宇航</at> PMFlow 系统采用前后端分离架构。',
status="sent",
created_at=now - timedelta(minutes=1),
sent_at=now - timedelta(minutes=1),
)
)
await db_session.commit()
resolution = await service._resolve_followup_anchor(
question="这个是哪个模型去回复我问题的?",
session_id=None,
context_session_id=None,
session_state=None,
thread_context={"reply_to_message_id": "om_bot_reply_1"},
)
assert resolution.anchor_mode == "explicit"
assert resolution.is_meta_followup is True
assert resolution.anchor_reply is not None
assert resolution.anchor_reply["question"] == "帮我解释下系统的架构"
assert "前后端分离架构" in resolution.anchor_reply["anchor_reply_preview"]
@pytest.mark.asyncio
async def test_explicit_reply_falls_back_to_current_ingest_closure_turns_when_outbox_mapping_missing(db_session) -> None:
service = _Harness(db_session)
db_session.add(
IngestEvent(
tenant_id=1,
channel="feishu_message",
event_key="om_current_message_1",
status="done",
payload={
"message_text": "这个是哪个模型去回复我问题的?",
"closure_turns": [
{"role": "user", "text": "帮我解释下系统的架构"},
{"role": "assistant", "text": "PMFlow 系统采用前后端分离架构。"},
{"role": "user", "text": "这个是哪个模型去回复我问题的?"},
],
},
)
)
await db_session.commit()
resolution = await service._resolve_followup_anchor(
question="这个是哪个模型去回复我问题的?",
session_id=None,
context_session_id=None,
session_state=None,
source_message_id="om_current_message_1",
thread_context={"reply_to_message_id": "om_unmapped_reply"},
)
assert resolution.anchor_mode == "explicit"
assert resolution.is_meta_followup is True
assert resolution.anchor_reply is not None
assert resolution.anchor_reply["question"] == "帮我解释下系统的架构"
assert "前后端分离架构" in resolution.anchor_reply["anchor_reply_preview"]
@pytest.mark.asyncio
async def test_recent_followup_candidates_can_cross_sessions_by_conversation_scope_key(db_session) -> None:
service = _Harness(db_session)
now = now_cn_naive()
previous_session = QASession(
tenant_id=1,
user_id=None,
title="旧会话",
external_conversation_key="feishu:msg-old-1",
started_at=now - timedelta(minutes=2),
last_message_at=now - timedelta(minutes=1),
message_count=1,
)
current_session = QASession(
tenant_id=1,
user_id=None,
title="新会话",
external_conversation_key="feishu:msg-new-1",
started_at=now,
last_message_at=now,
message_count=1,
)
db_session.add_all([previous_session, current_session])
await db_session.flush()
db_session.add(
QARecord(
tenant_id=1,
question="帮我解释下系统架构",
answer="PMFlow 系统采用前后端分离架构。",
sources=[],
intent="question",
domains_used=[],
response_time_ms=10,
asked_by=None,
asked_via="feishu",
project_id=None,
session_id=int(previous_session.id),
turn_index=1,
conversation_scope_key="feishu:chat-1:ou-1",
source_chat_id="chat-1",
source_sender_open_id="ou-1",
source_message_id="om_user_1",
created_at=now - timedelta(minutes=1),
)
)
await db_session.commit()
state = await service._ensure_session_state(session_id=int(current_session.id))
resolution = await service._resolve_followup_anchor(
question="这个是哪个模型回复的?",
session_id=int(current_session.id),
context_session_id=int(current_session.id),
session_state=state,
conversation_scope_key="feishu:chat-1:ou-1",
source_chat_id="chat-1",
source_sender_open_id="ou-1",
)
assert resolution.anchor_mode == "implicit_recent_reply"
assert resolution.anchor_reply_id is not None
assert resolution.anchor_kind == "meta_recent_reply"
assert resolution.anchor_reply is not None
assert resolution.anchor_reply["question"] == "帮我解释下系统架构"
@pytest.mark.asyncio
async def test_meta_followup_model_identity_reads_routing_snapshot_and_hides_real_model(db_session) -> None:
service = _Harness(db_session)
now = now_cn_naive()
session = QASession(
tenant_id=1,
user_id=None,
title="系统架构",
external_conversation_key="feishu:chat-1:ou-1",
started_at=now,
last_message_at=now,
message_count=1,
)
db_session.add(session)
await db_session.flush()
qa = QARecord(
tenant_id=1,
question="帮我解释下系统架构",
answer="PMFlow 系统采用前后端分离架构。",
sources=[],
intent="question",
domains_used=[],
response_time_ms=12,
asked_by=None,
asked_via="feishu",
project_id=None,
session_id=int(session.id),
turn_index=1,
conversation_scope_key="feishu:chat-1:ou-1",
source_chat_id="chat-1",
source_sender_open_id="ou-1",
source_message_id="om_user_1",
routing_snapshot={
"planned_tier": "low",
"executed_tier": "high",
"route_source": "default",
"route_rule_name": "engine_high_reasoning_glm5",
"routing_decision": "low_to_high_escalated",
"provider": "openai",
"model": "GLM-5",
},
)
db_session.add(qa)
await db_session.commit()
payload = await service._build_meta_followup_payload(
question="这个是哪个模型回复的?",
anchor_resolution=FollowupAnchorResolution(
anchor_mode="implicit_recent_reply",
anchor_reply_id=int(qa.id),
anchor_session_id=int(session.id),
anchor_kind="meta_recent_reply",
is_meta_followup=True,
anchor_reply={
"qa_id": int(qa.id),
"question": "帮我解释下系统架构",
"anchor_reply_preview": "PMFlow 系统采用前后端分离架构。",
"source_message_id": "om_user_1",
},
),
conversation_scope_key="feishu:chat-1:ou-1",
)
assert payload is not None
assert payload["meta_followup_kind"] == "reply_model_identity"
assert payload["reason_code"] == "meta_followup_answered"
assert "planned_tier=low" in payload["answer"]
assert "executed_tier=high" in payload["answer"]
assert "engine_high_reasoning_glm5" in payload["answer"]
assert "glm-5" not in payload["answer"].lower()