11 KiB

Raw Blame History

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.

11 KiB Raw Blame History Unescape Escape

Internal Query DSL And Semantic Facts Implementation Plan

Task 1: Query DSL Regression Tests

Task 2: Query DSL Planning And Answer Verifier

Task 3: Semantic Facts Schema Tests

Task 4: Semantic Facts Migration And Service

Task 5: Facts-First Retrieval And Async Extraction Wiring

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

11 KiB

Raw Blame History