OKF Generators

dedup_okf_generator.py

import os
import yaml
from pathlib import Path
from typing import Dict, Any
from pypdf import PdfReader
from google import genai
from google.genai import types

class DeduplicatedOKFGenerator:
    def __init__(self, target_dir: str, batch_size_pages: int = 10):
        self.target_dir = Path(target_dir)
        self.target_dir.mkdir(parents=True, exist_ok=True)
        self.batch_size = batch_size_pages
        self.client = genai.Client()
        
        # Central memory registry tracking concepts across the entire document
        # Format: { "concept_filename": { "metadata": {...}, "content": "..." } }
        self.registry: Dict[str, Dict[str, Any]] = {}

    def process_large_pdf(self, pdf_path: str):
        """Processes massive PDFs page-by-page using an in-memory dedup registry."""
        reader = PdfReader(pdf_path)
        total_pages = len(reader.pages)
        print(f"📚 Scaling Pipeline: processing {total_pages} pages with semantic deduplication.")

        for start_page in range(0, total_pages, self.batch_size):
            end_page = min(start_page + self.batch_size, total_pages)
            batch_text = []
            
            for page_num in range(start_page, end_page):
                text = reader.pages[page_num].extract_text()
                if text:
                    batch_text.append(text)
            
            if batch_text:
                print(f"🔍 Parsing chunk: Pages {start_page + 1} to {end_page}")
                self.extract_concepts_from_chunk("\n".join(batch_text))

        # Persist memory to disk and generate index layout map
        self.flush_registry_to_disk()
        self.generate_master_index()

    def extract_concepts_from_chunk(self, chunk_text: str):
        """Asks the model to isolate concepts using clean data wrappers."""
        system_instruction = """
        You are a Data Harmonization Engine. Extract distinct concepts from this text.
        Output each concept separated by '=== NEW_CONCEPT: filename ==='.
        Every concept MUST contain valid OKF YAML frontmatter with 'type', 'title', and 'tags'.
        Keep filenames lowercase, alphanumeric, and snake_case (e.g. core_user_metrics).
        """
        
        response = self.client.models.generate_content(
            model='gemini-2.5-flash',
            contents=chunk_text,
            config=types.GenerateContentConfig(
                system_instruction=system_instruction,
                temperature=0.1
            )
        )
        
        self._process_model_output(response.text)

    def _process_model_output(self, model_output: str):
        """Parses individual blocks and registers or merges them in memory."""
        blocks = model_output.split("=== NEW_CONCEPT: ")
        for block in blocks:
            if not block.strip():
                continue
            try:
                header_line, file_content = block.split("===\n", 1)
                filename = header_line.strip().replace(".md", "").strip()
                clean_raw = file_content.strip()

                if not clean_raw.startswith("---"):
                    continue

                # Isolate frontmatter block vs markdown text payload
                parts = clean_raw.split("---", 2)
                if len(parts) >= 3:
                    metadata = yaml.safe_load(parts[1]) or {}
                    content = parts[2].strip()
                    
                    # Deduplication routing check
                    if filename in self.registry:
                        print(f"🔄 Semantic Collision Detected for [{filename}]. Merging concepts...")
                        self.registry[filename] = self._merge_concepts_via_llm(
                            existing=self.registry[filename], 
                            incoming={"metadata": metadata, "content": content}
                        )
                    else:
                        # Register new concept
                        self.registry[filename] = {"metadata": metadata, "content": content}
            except Exception as e:
                pass

    def _merge_concepts_via_llm(self, existing: dict, incoming: dict) -> dict:
        """Uses Gemini to unify metadata and resolve data duplication cleanly."""
        merge_prompt = f"""
        You are an advanced Git Conflict Resolution Engine. Merge two versions of an OKF concept file cleanly.
        Ensure tags are combined without duplicates, descriptions are synthesized, and data formatting doesn't repeat.

        --- VERSION A (EXISTING) ---
        YAML: {existing['metadata']}
        Markdown: {existing['content']}

        --- VERSION B (INCOMING NEW INSIGHTS) ---
        YAML: {incoming['metadata']}
        Markdown: {incoming['content']}

        Output format MUST be valid OKF Markdown with YAML frontmatter wrapped in '---'. Do not output any chatter.
        """
        
        response = self.client.models.generate_content(
            model='gemini-2.5-flash',
            contents=merge_prompt,
            config=types.GenerateContentConfig(temperature=0.1)
        )
        
        # Unpack consolidated outputs
        text = response.text.strip()
        parts = text.split("---", 2)
        if len(parts) >= 3:
            try:
                metadata = yaml.safe_load(parts[1]) or {}
                content = parts[2].strip()
                return {"metadata": metadata, "content": content}
            except Exception:
                pass
        
        # Fallback structural merge if model output drops tokens
        existing["content"] += f"\n\n## Appended Context\n{incoming['content']}"
        return existing

    def flush_registry_to_disk(self):
        """Writes the clean, deduplicated memory state out to static files."""
        print(f"💾 Writing {len(self.registry)} unique, deduplicated OKF assets to storage tier...")
        for filename, data in self.registry.items():
            file_path = self.target_dir / f"{filename}.md"
            with open(file_path, "w", encoding="utf-8") as f:
                f.write("---\n")
                yaml.dump(data["metadata"], f, default_flow_style=False, sort_keys=False)
                f.write("---\n\n")
                f.write(data["content"] + "\n")

    def generate_master_index(self):
        """Creates a clean navigation roadmap indexing file."""
        index_content = [
            "---",
            "type: navigation_index",
            "title: Enterprise Knowledge Index Map",
            "description: Compact root architecture map tracking system-level nodes.",
            "---",
            "\n# System Architecture Index\n"
        ]
        for filename in sorted(self.registry.keys()):
            title = self.registry[filename]["metadata"].get("title", filename)
            ctype = self.registry[filename]["metadata"].get("type", "concept")
            index_content.append(f"- [{title}]({filename}.md) — `type: {ctype}`")

        with open(self.target_dir / "index.md", "w", encoding="utf-8") as f:
            f.write("\n".join(index_content))
        print("🎉 Deduplicated master compilation completely successfully.")

if __name__ == "__main__":
    if "GEMINI_API_KEY" not in os.environ:
        print("Please export GEMINI_API_KEY path variables.")
        exit(1)
        
    generator = DeduplicatedOKFGenerator(target_dir="./knowledge_vault", batch_size_pages=12)
    generator.process_large_pdf("massive_enterprise_manual.pdf")

okf_mcp_server.py

import os
import yaml
from pathlib import Path
from typing import Dict, Any, List
from mcp.server.fastmcp import FastMCP

# Define the local or mounted path to your hosted OKF files
OKF_BUNDLE_PATH = os.environ.get("OKF_BUNDLE_PATH", "./knowledge_vault")

# Initialize FastMCP Server
mcp = FastMCP("Enterprise-OKF-Knowledge-Server")

def parse_okf_file(file_path: Path) -> Dict[str, Any]:
    """Helper to cleanly extract YAML frontmatter and content from an OKF file."""
    with open(file_path, "r", encoding="utf-8") as f:
        text = f.read()
        
    if not text.startswith("---"):
        return {"metadata": {}, "content": text}
        
    parts = text.split("---", 2)
    if len(parts) >= 3:
        try:
            metadata = yaml.safe_load(parts[1]) or {}
            content = parts[2].strip()
            return {"metadata": metadata, "content": content}
        except yaml.YAMLError:
            pass
            
    return {"metadata": {}, "content": text}

# --- MCP RESOURCES ---
# This exposes every individual OKF markdown file to the LLM agent via universal URIs

@mcp.resource("okf://concepts")
def list_all_concepts() -> str:
    """Lists every available corporate concept tracking ID in the OKF bundle."""
    base_dir = Path(OKF_BUNDLE_PATH)
    if not base_dir.exists():
        return "OKF knowledge bundle path not found."
        
    files = list(base_dir.glob("*.md"))
    if not files:
        return "No OKF concepts found in bundle directory."
        
    output = []
    for file in files:
        data = parse_okf_file(file)
        title = data["metadata"].get("title", file.stem)
        concept_type = data["metadata"].get("type", "unknown")
        output.append(f"- **{file.stem}** | Type: `{concept_type}` | Title: {title}")
        
    return "\n".join(output)

@mcp.resource("okf://concept/{concept_id}")
def get_concept(concept_id: str) -> str:
    """Retrieves the full structural content and metadata of a single OKF concept by ID."""
    file_path = Path(OKF_BUNDLE_PATH) / f"{concept_id}.md"
    if not file_path.exists():
        return f"Error: Concept '{concept_id}' does not exist inside this OKF catalog instance."
        
    data = parse_okf_file(file_path)
    metadata_str = yaml.dump(data["metadata"], default_flow_style=False)
    
    return f"--- OKF METADATA ---\n{metadata_str}---\n\n--- CONCEPT CONTENT ---\n{data['content']}"

# --- MCP TOOLS ---
# This gives the agent structural search functions to systematically traverse the knowledge graph

@mcp.tool()
def search_okf_by_type(concept_type: str) -> List[str]:
    """
    Search the enterprise OKF repository for files matching a specific core schema type 
    (e.g., 'metric_definition', 'table_schema', 'runbook').
    """
    base_dir = Path(OKF_BUNDLE_PATH)
    matching_concepts = []
    
    for file in base_dir.glob("*.md"):
        data = parse_okf_file(file)
        if data["metadata"].get("type") == concept_type:
            matching_concepts.append(file.stem)
            
    return matching_concepts

@mcp.tool()
def search_okf_by_tag(tag: str) -> List[str]:
    """Search the enterprise OKF catalog for concepts associated with a specific organizational tag."""
    base_dir = Path(OKF_BUNDLE_PATH)
    matching_concepts = []
    
    for file in base_dir.glob("*.md"):
        data = parse_okf_file(file)
        tags = data["metadata"].get("tags", [])
        if isinstance(tags, list) and tag in tags:
            matching_concepts.append(file.stem)
            
    return matching_concepts

if __name__ == "__main__":
    # Start the standard MCP server utilizing stdio transport mode (ideal for deployment sidecars)
    print(f"Initializing OKF MCP Server listening over path: {OKF_BUNDLE_PATH}")
    mcp.run(transport="stdio")

scaled_okf_generator.py

import os
import yaml
from pathlib import Path
from pypdf import PdfReader
from google import genai
from google.genai import types

class LargeScaleOKFGenerator:
    def __init__(self, target_dir: str, batch_size_pages: int = 10):
        self.target_dir = Path(target_dir)
        self.target_dir.mkdir(parents=True, exist_ok=True)
        self.batch_size = batch_size_pages
        self.client = genai.Client()

    def process_large_pdf(self, pdf_path: str):
        """Processes massive PDFs by slicing them into batches to avoid token execution ceilings."""
        reader = PdfReader(pdf_path)
        total_pages = len(reader.pages)
        print(f"📚 Processing large document: {total_pages} total pages found.")

        # Iteratively process chunks of pages
        for start_page in range(0, total_pages, self.batch_size):
            end_page = min(start_page + self.batch_size, total_pages)
            print(f"Processing batch: Pages {start_page + 1} to {end_page}...")
            
            batch_text = []
            for page_num in range(start_page, end_page):
                text = reader.pages[page_num].extract_text()
                if text:
                    batch_text.append(text)
            
            if batch_text:
                self.extract_concepts_from_chunk("\n".join(batch_text))

        # After compiling all individual modular files, build the structural index
        self.generate_master_index()

    def extract_concepts_from_chunk(self, chunk_text: str):
        """Calls the LLM over the text fragment to build atomized concepts."""
        system_instruction = """
        You are a Document Partitioning Engine. Extract distinct concepts from this text.
        Output each concept separated by '=== NEW_CONCEPT: filename ==='.
        Every output block must include valid OKF YAML frontmatter containing 'type', 'title', and 'tags'.
        If a file already exists or is a continuation of a topic, provide its clean atomic components.
        """
        
        response = self.client.models.generate_content(
            model='gemini-2.5-flash',
            contents=chunk_text,
            config=types.GenerateContentConfig(
                system_instruction=system_instruction,
                temperature=0.1
            )
        )
        
        self._write_or_append_concepts(response.text)

    def _write_or_append_concepts(self, model_output: str):
        """Writes new concepts, or merges content safely if the concept was found in a prior batch."""
        blocks = model_output.split("=== NEW_CONCEPT: ")
        for block in blocks:
            if not block.strip():
                continue
            try:
                header_line, file_content = block.split("===\n", 1)
                filename = header_line.strip().replace(".md", "").strip()
                clean_content = file_content.strip()

                file_path = self.target_dir / f"{filename}.md"
                
                # Check if concept file already exists from a previous page batch
                if file_path.exists():
                    # Append new insights to the end of the file body
                    with open(file_path, "a", encoding="utf-8") as f:
                        f.write(f"\n\n## Additional Context (Document Continuation)\n{clean_content}")
                    print(f"🔄 Appended continuing data to existing concept: {filename}")
                else:
                    # Write as a brand new standalone file
                    with open(file_path, "w", encoding="utf-8") as f:
                        f.write(clean_content + "\n")
                    print(f"✅ Generated brand new atomic concept: {filename}")
            except Exception as e:
                pass

    def generate_master_index(self):
        """Creates an index.md file to act as the primary navigation layout node for the OKF bundle."""
        print("🗺️ Building master index.md layout structure...")
        markdown_files = list(self.target_dir.glob("*.md"))
        
        index_content = [
            "---",
            "type: navigation_index",
            "title: Global Document Layout Map",
            "description: Structural map auto-generated from comprehensive source manual.",
            "---",
            "\n# System Architecture Index\n",
            "This master index references all modular technical concepts extracted from the root asset.\n"
        ]
        
        for file in markdown_files:
            if file.name == "index.md":
                continue
            # Basic catalog indexing entry point link
            index_content.append(f"- [{file.stem}]({file.name})")

        with open(self.target_dir / "index.md", "w", encoding="utf-8") as f:
            f.write("\n".join(index_content))
        print("🎉 Master index compilation successfully complete.")

if __name__ == "__main__":
    # Assumes GEMINI_API_KEY is available in your shell environment variables
    generator = LargeScaleOKFGenerator(target_dir="./knowledge_vault", batch_size_pages=15)
    
    # Put your massive 200-page operational manual here
    generator.process_large_pdf("massive_enterprise_manual.pdf")

simple_okf_generator.py

# pip install pypdf google-genai pyyaml

import os
import re
from pathlib import Path
import yaml
from pypdf import PdfReader
from google import genai
from google.genai import types

class UniversalOKFGenerator:
    def __init__(self, target_dir: str):
        self.target_dir = Path(target_dir)
        self.target_dir.mkdir(parents=True, exist_ok=True)
        # Initialize Google GenAI client (requires GEMINI_API_KEY env variable)
        self.client = genai.Client()

    def extract_text_from_pdf(self, pdf_path: str) -> str:
        """Reads a PDF file and extracts all raw text content."""
        print(f"📄 Extracting text from: {pdf_path}")
        reader = PdfReader(pdf_path)
        full_text = []
        for page in reader.pages:
            text = page.extract_text()
            if text:
                full_text.append(text)
        return "\n".join(full_text)

    def generate_bundle_from_text(self, raw_text: str):
        """Uses Gemini to split raw text into separate, cross-linked OKF markdown files."""
        print("🤖 Analyzing document structure with Gemini...")
        
        system_instruction = """
        You are an expert Enterprise Knowledge Architect. Your job is to take raw text from a document and break it down into a modular bundle following the Open Knowledge Format (OKF) specification.
        
        Analyze the input text and identify distinct, re-usable concepts. Examples of concepts: metric definitions, table schemas, business rules, or step-by-step runbooks.
        
        For each distinct concept found, output a block that can be easily parsed. Use the exact delimiter '=== NEW_CONCEPT: filename ===' before each file.
        
        Each concept file MUST match this exact format:
        1. YAML Frontmatter wrapped in '---'. It must include a 'type' (e.g., metric_definition, table_schema, runbook, corporate_policy), a 'title', a descriptive 'description', and a list of 'tags'.
        2. Deep structural Markdown content.
        3. Standard markdown links to other generated filenames if they are related (e.g., [See Subscriptions Schema](db_subscriptions.md)).
        
        Make sure your markdown handles technical formatting, tables, and mathematical formulas cleanly.
        """

        prompt = f"""
        Analyze the following corporate document text and extract all standalone OKF concepts. 
        Create cross-links between the concepts where applicable using their markdown filenames.

        --- START DOCUMENT TEXT ---
        {raw_text}
        --- END DOCUMENT TEXT ---
        """

        # Execute structured model generation
        response = self.client.models.generate_content(
            model='gemini-2.5-flash',
            contents=prompt,
            config=types.GenerateContentConfig(
                system_instruction=system_instruction,
                temperature=0.2
            )
        )

        self._parse_and_write_files(response.text)

    def _parse_and_write_files(self, model_output: str):
        """Internal helper to split model text and write valid OKF markdown files."""
        # Split output by our custom token delimiter
        concept_blocks = model_output.split("=== NEW_CONCEPT: ")
        
        for block in concept_blocks:
            if not block.strip():
                continue
                
            try:
                # Extract target filename and file contents
                header_line, file_content = block.split("===\n", 1)
                filename = header_line.strip().replace(".md", "")
                
                # Sanitize block content to keep only the markdown text
                clean_content = file_content.strip()
                if clean_content.startswith("```markdown"):
                    clean_content = clean_content.split("```markdown", 1)[1].rsplit("```", 1)[0].strip()
                elif clean_content.startswith("```"):
                    clean_content = clean_content.split("```", 1)[1].rsplit("```", 1)[0].strip()

                # Basic validation: ensure frontmatter exists
                if not clean_content.startswith("---"):
                    print(f"⚠️ Skipping {filename}: Missing frontmatter marker.")
                    continue

                file_path = self.target_dir / f"{filename}.md"
                with open(file_path, "w", encoding="utf-8") as f:
                    f.write(clean_content + "\n")
                    
                print(f"✅ Generated OKF concept: {file_path}")
                
            except Exception as e:
                print(f"❌ Failed to parse a concept block due to error: {e}")

if __name__ == "__main__":
    # Quick sanity validation for environment setup
    if "GEMINI_API_KEY" not in os.environ:
        print("❌ Error: Please set your GEMINI_API_KEY environment variable first.")
        exit(1)

    # Initialize the generator targeting your production vault folder
    generator = UniversalOKFGenerator(target_dir="./knowledge_vault")
    
    # Path to your source enterprise asset
    sample_pdf = "corporate_analytics_spec.pdf"
    
    # Execution routine if the sample file exists
    if os.path.exists(sample_pdf):
        extracted_text = generator.extract_text_from_pdf(sample_pdf)
        generator.generate_bundle_from_text(extracted_text)
        print("\n🎉 OKF Bundle construction complete.")
    else:
        print(f"\n💡 Place a file named '{sample_pdf}' in this folder to run the universal parser pipeline.")