Function compile_short_doc

openkb/agent/compiler.py:2176–2254

Compile a short document using a multi-step LLM pipeline with caching. Step 1: Build base context A (schema + doc content), generate summary. Steps 2-4: Delegated to ``_compile_concepts``.

(
    doc_name: str,
    source_path: Path,
    kb_dir: Path,
    model: str,
    max_concurrency: int = DEFAULT_COMPILE_CONCURRENCY,
)

async def compile_short_doc(
    doc_name: str,
    source_path: Path,
    kb_dir: Path,
    model: str,
    max_concurrency: int = DEFAULT_COMPILE_CONCURRENCY,
) -> None:
    """Compile a short document using a multi-step LLM pipeline with caching.

    Step 1: Build base context A (schema + doc content), generate summary.
    Steps 2-4: Delegated to ``_compile_concepts``.
    """
    from openkb.config import load_config

    openkb_dir = kb_dir / ".openkb"
    config = load_config(openkb_dir / "config.yaml")
    language: str = config.get("language", "en")
    entity_types = resolve_entity_types(config)

    wiki_dir = kb_dir / "wiki"
    schema_md = get_agents_md(wiki_dir)
    content = source_path.read_text(encoding="utf-8")

    # Base context A: system + document. cache_control marker on the doc
    # message creates a cache breakpoint that covers (system + doc) for
    # every downstream call (summary, concepts-plan, every concept page).
    system_msg = {
        "role": "system",
        "content": _SYSTEM_TEMPLATE.format(
            schema_md=schema_md,
            language=language,
        ),
    }
    doc_msg = {
        "role": "user",
        "content": _cached_text(
            _SUMMARY_USER.format(
                doc_name=doc_name,
                content=content,
            )
        ),
    }

    # --- Step 1: Generate summary (v1, held in memory) ---
    # The summary is NOT written to disk yet — it's used as cache context
    # for the plan + concept-generation calls, then rewritten into a final
    # v2 (with a whitelist of known wikilink targets) inside
    # _compile_concepts before being written to disk.
    summary_raw = _llm_call(
        model, [system_msg, doc_msg], "summary", response_format=_JSON_RESPONSE_FORMAT
    )
    try:
        summary_parsed = _parse_json(summary_raw)
        doc_brief = summary_parsed.get("description", "")
        summary = summary_parsed.get("content", summary_raw)
    except (json.JSONDecodeError, ValueError):
        doc_brief = ""
        summary = summary_raw