ljwx-deploy/scripts/knowledge/query_knowledge.py

#!/usr/bin/env python3
"""Interactive query CLI over the knowledge plane.

Thin wrapper over context_pack so dogfooding catches integration bugs.
Mainly for local exploration: 'show me what the knowledge plane has for
<topic>' without going through the full pack-write-to-disk flow.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

from .context_pack import (
    DEFAULT_TOP_K,
    PackInputs,
    PublishRecordStore,
    _build_dify_sections,
    _build_keyword_sections,
    _cap_by_budget,
    _dedupe,
    _qualify_repo,
    _rank,
    _resolve_dify,
    _strip_repo_owner,
)
from .common import utc_now


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Query the knowledge plane (Dify when available, local fallback otherwise)"
    )
    parser.add_argument("--repo", required=True, help="service name or owner__repo")
    parser.add_argument("--query", required=True)
    parser.add_argument("--top-k", type=int, default=DEFAULT_TOP_K)
    parser.add_argument(
        "--deploy-root", type=Path, default=Path.cwd(), help="ljwx-deploy root"
    )
    parser.add_argument("--env", default="dev")
    parser.add_argument(
        "--skip-dify", action="store_true", help="force local fallback"
    )
    parser.add_argument(
        "--format",
        choices=["pretty", "json"],
        default="pretty",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    repo_qualifier = _qualify_repo(args.repo)
    repo_owner_slash = _strip_repo_owner(repo_qualifier)

    inputs = PackInputs(
        repo=repo_owner_slash,
        mode="implement",
        task_text=args.query,
        changed_files=(),
        repo_root=Path.cwd(),
        deploy_root=args.deploy_root,
        output_dir=Path("/dev/null"),
        env=args.env,
        token_budget=20_000,
        top_k=args.top_k,
    )

    store = PublishRecordStore(inputs.deploy_root, inputs.env)
    loaded = store.load_for_repo(repo_qualifier)

    sections = []
    dify_used = False
    if not args.skip_dify:
        client, dataset_ids = _resolve_dify(inputs.deploy_root)
        if client is not None and dataset_ids:
            sections = _build_dify_sections(
                store=store,
                inputs=inputs,
                dify=client,
                dataset_ids=dataset_ids,
                query=args.query,
            )
            dify_used = bool(sections)

    if not sections:
        sections = _build_keyword_sections(store, inputs, args.query)

    sections = _dedupe(sections)
    sections = _rank(sections)
    sections = _cap_by_budget(sections, inputs.token_budget)

    if args.format == "json":
        out = {
            "query": args.query,
            "repo": repo_owner_slash,
            "records_loaded": loaded,
            "dify_used": dify_used,
            "results": [
                {
                    "title": s.title,
                    "tier": s.tier,
                    "source_path": s.source_path,
                    "source_ref": s.source_ref,
                    "document_id": s.document_id,
                    "chunk_id": s.chunk_id,
                    "origin": s.origin,
                    "score": s.score,
                    "snippet": s.text[:280],
                }
                for s in sections
            ],
        }
        print(json.dumps(out, ensure_ascii=False, indent=2))
        return 0

    print(f"# Query: {args.query}")
    print(f"# Repo:  {repo_owner_slash}   records loaded: {loaded}   dify: {dify_used}")
    print(f"# Time:  {utc_now()}")
    print(f"# Results: {len(sections)}\n")
    if not sections:
        print("(no matches)")
        return 0
    for index, s in enumerate(sections, start=1):
        cite = (
            f"[kg:{s.source_repo}:{s.document_id}:{s.chunk_id}]"
            if s.chunk_id
            else f"[kg:{s.source_repo}:{s.source_path}]"
        )
        print(f"## {index}. {s.title}")
        print(f"   tier={s.tier}  origin={s.origin}  score={s.score:.2f}")
        print(f"   path: {s.source_path}")
        print(f"   cite: {cite}")
        snippet = s.text.strip().replace("\n", " ")
        if len(snippet) > 280:
            snippet = snippet[:280] + "…"
        print(f"   {snippet}")
        print()
    return 0


if __name__ == "__main__":
    sys.exit(main())