Agentic_Rag/scripts/smoke_test.py

from app.config import get_settings
from app.schemas import QueryRequest
from app.service import AgenticRAGService


def main() -> None:
    service = AgenticRAGService(get_settings())
    response = service.invoke(
        QueryRequest(
            query=(
                "统计 U1001 最近 30 天有效订单数、实付总额和低于 99 元的订单数，"
                "并结合内部运费规则说明是否存在需支付基础运费的订单。"
            ),
            debug=True,
        )
    )
    print(response.model_dump_json(indent=2))
    if response.termination_reason != "evidence_accepted":
        raise SystemExit("Smoke Test 未通过：证据未被接受。")
    if {item.source_type for item in response.citations} != {"milvus", "sql"}:
        raise SystemExit("Smoke Test 未通过：未同时获得 Milvus 与 SQL 证据。")
    if {item["tool"] for item in response.executed_queries} != {
        "milvus_search",
        "sql_query",
    }:
        raise SystemExit("Smoke Test 未通过：实际工具 Query 记录不完整。")


if __name__ == "__main__":
    main()