PMFlow/docs/superpowers/plans/2026-04-15-internal-query-dsl-semantic-facts.md

Internal Query DSL And Semantic Facts Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 为内部只读查询落地 Query DSL Plan + 白名单受控执行器 + semantic facts 优先 + answer verifier 主链，先止住答非所问，再把语义事实读模型和异步抽取链接起来。

Architecture: 第一阶段先收口 rag_query_semantics、快路径和 deterministic query，让内部查询先通过结构化计划和回答合同校验。第二阶段新增 semantic_facts 读模型与异步抽取 run，查询时优先命中 facts，再回退到受控 DB/graph/doc 检索。

Tech Stack: Python 3.11, FastAPI, SQLAlchemy 2.0, Alembic, pytest, AsyncSession, PMFlow AdaptiveRAG

---

Task 1: Query DSL Regression Tests

Files:

• Modify: backend/tests/test_adaptive_rag_mainline_contract.py

• Modify: backend/tests/test_ai_pmo_ingress_mainline.py

• Test: backend/tests/test_adaptive_rag_mainline_contract.py

• Test: backend/tests/test_ai_pmo_ingress_mainline.py

• Step 1: Write the failing tests

def test_query_runtime_planner_keeps_existence_on_fast_path():
    decision = service._plan_query_runtime(
        question="系统里有没有A股项目？",
        effective_question="系统里有没有A股项目？",
        query_type="existence",
        requested_thinking_mode="normal",
        enforce_scene_prompt_only=False,
    )
    assert decision.answer_mode == "fast_path"


@pytest.mark.asyncio
async def test_execute_query_intent_internal_existence_question_does_not_fall_back_to_total_count():
    ...
    assert "项目总数" not in result["answer"]
    assert "有" in result["answer"] or "未检索到" in result["answer"]

• Step 2: Run tests to verify they fail

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py -k "existence or query_runtime_planner" -v

Expected: FAIL because existence is not yet part of the fast-path taxonomy.

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_ai_pmo_ingress_mainline.py -k "total_count or existence" -v

Expected: FAIL because internal existence questions still route through the old count/status behavior.

• Step 3: Implement the minimal code to make the tests pass

QUERY_TYPE_EXISTENCE = "existence"
QUERY_TYPE_DIAGNOSE = "diagnose"
QUERY_TYPE_COMPARE = "compare"

fast_path_candidate = normalized_query_type in {
    QUERY_TYPE_EXISTENCE,
    QUERY_TYPE_COUNT,
    QUERY_TYPE_LIST,
    QUERY_TYPE_STATUS,
}

• Step 4: Re-run the targeted tests

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py -k "existence or query_runtime_planner" -v

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_ai_pmo_ingress_mainline.py -k "total_count or existence" -v

Expected: PASS

Task 2: Query DSL Planning And Answer Verifier

Files:

• Modify: backend/app/config/ai_prompts.py

• Modify: backend/app/services/adaptive_rag_service/shared_part_02.py

• Modify: backend/app/services/adaptive_rag_service/mixin_core_intent.py

• Modify: backend/app/services/adaptive_rag_service/mixin_answer_mainline.py

• Modify: backend/app/services/adaptive_rag_service/mixin_deterministic_query_core.py

• Modify: backend/app/services/adaptive_rag_service/mixin_fast_answer_project_main.py

• Modify: backend/app/services/adaptive_rag_service/mixin_answer_payload.py

• Test: backend/tests/test_adaptive_rag_mainline_contract.py

• Test: backend/tests/test_runtime_structured_semantics_contract.py

• Step 1: Write the failing tests for DSL plan normalization and verifier

def test_normalize_query_semantics_accepts_existence_diagnose_compare():
    normalized = service._normalize_query_semantics({"query_type": "existence"})
    assert normalized["query_type"] == "existence"


def test_answer_verifier_blocks_total_count_substitution():
    verdict = service._verify_internal_answer_contract(
        query_type="existence",
        answer="当前系统里项目总数为42个。",
        sources=[{"source": "db"}],
    )
    assert verdict["ok"] is False

• Step 2: Run tests to verify they fail

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py -k "verifier or normalize_query_semantics" -v

Expected: FAIL because the new query types and verifier do not exist yet.

• Step 3: Implement minimal Query DSL planning and verifier

def _build_internal_query_plan(...):
    return {
        "query_type": query_type,
        "entity_type": entity_type,
        "answer_contract": answer_contract,
        "filters": filters,
        "retrieval_policy": retrieval_policy,
        "verifier_checks": [...],
    }


def _verify_internal_answer_contract(...):
    if query_type == QUERY_TYPE_EXISTENCE and "项目总数" in answer:
        return {"ok": False, "reason": "no_total_count_substitution"}
    return {"ok": True}

• Step 4: Re-run the targeted tests

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py -k "verifier or normalize_query_semantics or existence" -v

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_runtime_structured_semantics_contract.py -k "query_type" -v

Expected: PASS

Task 3: Semantic Facts Schema Tests

Files:

• Create: backend/tests/test_semantic_fact_service.py

• Modify: backend/tests/test_project_knowledge_graph_event_queue_contract.py

• Test: backend/tests/test_semantic_fact_service.py

• Test: backend/tests/test_project_knowledge_graph_event_queue_contract.py

• Step 1: Write the failing tests for semantic facts read/write and event enqueue

@pytest.mark.asyncio
async def test_semantic_fact_service_upserts_normalized_fact():
    ...
    assert row.fact_key == "market_theme"
    assert row.fact_value_norm == "a_share"


@pytest.mark.asyncio
async def test_project_event_queue_emits_semantic_fact_extraction_job_after_commit():
    ...
    assert queued_event["event_type"] == "semantic_fact.extract"

• Step 2: Run tests to verify they fail

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_semantic_fact_service.py -v

Expected: FAIL because the service and models do not exist yet.

• Step 3: Implement minimal schema/service contracts

class SemanticFact(Base):
    __tablename__ = "semantic_facts"
    ...


class SemanticFactExtractionRun(Base):
    __tablename__ = "semantic_fact_extraction_runs"
    ...

• Step 4: Re-run the targeted tests

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_semantic_fact_service.py tests/test_project_knowledge_graph_event_queue_contract.py -k "semantic_fact" -v

Expected: PASS

Task 4: Semantic Facts Migration And Service

Files:

• Create: backend/alembic/versions/<timestamp>_add_semantic_facts.py

• Create: backend/app/models/semantic_fact.py

• Create: backend/app/models/semantic_fact_extraction_run.py

• Modify: backend/app/models/__init__.py

• Create: backend/app/services/semantic_fact_service/service.py

• Create: backend/app/services/semantic_fact_service/mixin_extract.py

• Create: backend/app/services/semantic_fact_service/mixin_query.py

• Test: backend/tests/test_semantic_fact_service.py

• Step 1: Generate the migration and model implementations

op.create_table("semantic_facts", ...)
op.create_index("ix_semantic_facts_tenant_entity_key_norm", "semantic_facts", ...)
op.create_table("semantic_fact_extraction_runs", ...)

• Step 2: Run the semantic fact tests

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_semantic_fact_service.py -v

Expected: PASS

• Step 3: Run syntax verification

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && python -m py_compile app/models/semantic_fact.py app/models/semantic_fact_extraction_run.py app/services/semantic_fact_service/service.py

Expected: PASS

Task 5: Facts-First Retrieval And Async Extraction Wiring

Files:

• Modify: backend/app/services/adaptive_rag_service/mixin_answer_retrieval.py

• Modify: backend/app/services/adaptive_rag_service/mixin_retrieval_router_db.py

• Modify: backend/app/services/adaptive_rag_service/mixin_fast_answer_project_main.py

• Modify: backend/app/services/project_knowledge_graph_service/mixin_event_queue.py

• Modify: backend/app/services/project_knowledge_graph_service/service.py

• Test: backend/tests/test_adaptive_rag_mainline_contract.py

• Test: backend/tests/test_project_knowledge_graph_event_queue_contract.py

• Step 1: Write the failing facts-first retrieval tests

@pytest.mark.asyncio
async def test_internal_query_prefers_semantic_facts_before_db_fallback():
    ...
    assert result["sources"][0]["metadata"]["table"] == "semantic_facts"


@pytest.mark.asyncio
async def test_internal_existence_query_falls_back_to_project_text_when_fact_missing():
    ...
    assert result["sources"][0]["metadata"]["table"] in {"semantic_facts", "projects"}

• Step 2: Run tests to verify they fail

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py -k "semantic_facts or facts_first" -v

Expected: FAIL because retrieval is not facts-first yet.

• Step 3: Implement the facts-first query hook and async extraction enqueue

fact_rows = await semantic_fact_service.query_facts(...)
if fact_rows:
    return build_fact_retrieved_items(fact_rows)
return await self._retrieve_structured_db_fallback(...)

• Step 4: Re-run the targeted tests

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py tests/test_project_knowledge_graph_event_queue_contract.py -k "semantic_facts or facts_first" -v

Expected: PASS

Task 6: End-To-End Verification And Docs Sync

Files:

• Modify: docs/architecture/CODE_REFERENCE_INDEX.md

• Verify: AGENTS.md

• Verify: CLAUDE.md

• Verify: docs/architecture/RUNTIME_GUARDRAILS.md

• Verify: docs/architecture/DATAFLOW.md

• Verify: docs/architecture/ARCHITECTURE_INVENTORY.md

• Step 1: Update reference index for new runtime/model entry points

- `backend/app/services/semantic_fact_service/`
- `backend/app/models/semantic_fact.py`
- `backend/app/models/semantic_fact_extraction_run.py`

• Step 2: Run focused backend verification

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && pytest tests/test_adaptive_rag_mainline_contract.py tests/test_ai_pmo_ingress_mainline.py tests/test_project_knowledge_graph_event_queue_contract.py tests/test_semantic_fact_service.py -v

Expected: PASS

• Step 3: Run compilation verification

Run: cd /Users/yuuuu/Documents/AIProjectManger/backend && python -m py_compile app/config/ai_prompts.py app/services/adaptive_rag_service/*.py app/services/project_knowledge_graph_service/*.py app/services/semantic_fact_service/*.py

Expected: PASS

• Step 4: Attempt runtime reload if environment allows

Run: cd /Users/yuuuu/Documents/AIProjectManger && docker compose restart backend

Expected: backend restarts successfully, or record exact blocker if this machine is not the deploy environment.