Function query

openkb/cli.py:1102–1145

Query the knowledge base with QUESTION.

(ctx, question, save, raw)

)
@click.pass_context
def query(ctx, question, save, raw):
    """Query the knowledge base with QUESTION."""
    kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
    if kb_dir is None:
        click.echo("No knowledge base found. Run `openkb init` first.")
        return

    from openkb.agent.query import run_query

    openkb_dir = kb_dir / ".openkb"
    config = load_config(openkb_dir / "config.yaml")
    _setup_llm_key(kb_dir)
    model: str = config.get("model", DEFAULT_CONFIG["model"])

    stream = _stream_to_tty()
    try:
        answer = asyncio.run(run_query(question, kb_dir, model, stream=stream, raw=raw))
        if not stream and answer:
            click.echo(answer)
    except Exception as exc:
        click.echo(f"[ERROR] Query failed: {exc}")
        return

    append_log(kb_dir / "wiki", "query", question)

    if save and answer:
        import re
        from openkb.lint import list_existing_wiki_targets, strip_ghost_wikilinks

        slug = re.sub(r"[^a-z0-9]+", "-", question.lower()).strip("-")[:60]
        explore_dir = kb_dir / "wiki" / "explorations"
        explore_dir.mkdir(parents=True, exist_ok=True)
        explore_path = explore_dir / f"{slug}.md"
        # Strip ghost wikilinks the agent may have emitted to non-existent
        # concept/summary pages — the schema_md in the agent's instructions
        # encourages [[wikilinks]] but the agent's view of "which pages
        # exist" can drift from disk reality.
        known = list_existing_wiki_targets(kb_dir / "wiki")
        cleaned_answer, _ = strip_ghost_wikilinks(answer, known)
        explore_path.write_text(
            f'---\nquery: "{question}"\n---\n\n{cleaned_answer}\n',
            encoding="utf-8",
        )
        click.echo(f"\nSaved to {explore_path}")


def _cleanup_pageindex(