MCPcopy Index your code
hub / github.com/VectifyAI/OpenKB / _write_concept

Function _write_concept

openkb/agent/compiler.py:953–1021  ·  view source on GitHub ↗

Write or update a concept page, managing the sources frontmatter.

(
    wiki_dir: Path, name: str, content: str, source_file: str, is_update: bool, brief: str = ""
)

Source from the content-addressed store, hash-verified

951
952
953def _write_concept(
954 wiki_dir: Path, name: str, content: str, source_file: str, is_update: bool, brief: str = ""
955) -> None:
956 """Write or update a concept page, managing the sources frontmatter."""
957 concepts_dir = wiki_dir / "concepts"
958 concepts_dir.mkdir(parents=True, exist_ok=True)
959 safe_name = _sanitize_concept_name(name)
960 path = (concepts_dir / f"{safe_name}.md").resolve()
961 if not path.is_relative_to(concepts_dir.resolve()):
962 logger.warning("Concept name escapes concepts dir: %s", name)
963 return
964
965 if is_update and path.exists():
966 existing = path.read_text(encoding="utf-8")
967 if source_file not in existing:
968 existing = _prepend_source_to_frontmatter(existing, source_file)
969 # Strip frontmatter from LLM content to avoid duplicate blocks
970 clean_parts = frontmatter.split(content)
971 clean = clean_parts[1].lstrip("\n") if clean_parts is not None else content
972 # Replace body with LLM rewrite (prompt asks for full rewrite, not delta)
973 ex_parts = frontmatter.split(existing)
974 if ex_parts is not None:
975 fm_block, _ = ex_parts
976 existing = fm_block + "\n" + clean
977 else:
978 # Malformed/absent frontmatter (opening ``---`` with no closing
979 # delimiter, or no frontmatter at all): rebuild valid frontmatter
980 # rather than writing a bare body. Recover any sources already
981 # listed in the broken block first.
982 recovered: list[str] = []
983 for ln in existing.split("\n"):
984 if ln.lstrip().startswith("sources:"):
985 parsed = _parse_yaml_list_value(ln)
986 if parsed:
987 recovered = parsed
988 break
989 merged = [source_file] + [s for s in recovered if s != source_file]
990 fm_lines = [
991 _yaml_kv_line("type", "Concept"),
992 _yaml_list_line("sources", merged),
993 ]
994 if brief:
995 fm_lines.append(_yaml_kv_line("description", brief))
996 existing = frontmatter.block(fm_lines) + clean
997 atomic_write_text(path, existing)
998 return
999 # Guarantee type + refresh description on update; remove legacy brief:.
1000 ex_parts2 = frontmatter.split(existing)
1001 if ex_parts2 is not None:
1002 fm_block, body = ex_parts2
1003 fm_block = _set_fm_line(fm_block, "type", "Concept")
1004 if brief:
1005 fm_block = _set_fm_line(fm_block, "description", brief)
1006 # Drop legacy brief: lines (migrated to description:).
1007 fm_block = frontmatter.drop_line(fm_block, "brief")
1008 existing = fm_block + body
1009 atomic_write_text(path, existing)
1010 else:

Calls 3

atomic_write_textFunction · 0.90
_sanitize_concept_nameFunction · 0.85