an agent for working with the Carleton research services LLM

example_usage.sh

for f in ./kge_space/*.txt; do                  
  source_id=$(basename "$f" .txt | tr '[:upper:]' '[:lower:]' | tr ' _' '-')
  python skill_agent.py \
    --api-key  key_goes_here \
    --skill    ./skills/kge-extraction.md \
    --task     "For each .txt file in the working directory: Extract the knowledge graph. Then index it." \
    --work-dir ./kge_space \
    --history-turns     10 \
    --max-tool-result   2000 \
    --max-payload-bytes 120000 \
    --verbose

kg_formatter.py

#!/usr/bin/env python3
"""
Use this to turn markdown tables into ampligraph or pykeen formatted tsv
"""

import os
import glob
import pandas as pd
import re
import uuid

def parse_markdown_tables(directory):
    """
    Robustly parses Markdown tables from all .md files in a directory.
    Avoids using pandas.read_html/markdown to gracefully handle LLM formatting quirks.
    """
    rows = []
    
    # Find all markdown files in the quads directory
    for filepath in glob.glob(os.path.join(directory, '*.md')):
        source_id = os.path.basename(filepath).replace('.md', '')
        
        with open(filepath, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                
                # Check if line is a table row
                if line.startswith('|') and line.endswith('|'):
                    # Skip the Markdown separator row (e.g., |---|---|)
                    if re.match(r'^\|[-\s|]+\|$', line):
                        continue
                    
                    # Split by pipe, drop empty first/last elements
                    cols = [c.strip() for c in line.split('|')[1:-1]]
                    
                    # Ensure it matches our 7-column schema
                    if len(cols) == 7:
                        # Skip the header row itself
                        if cols[0].lower() == 'subject':
                            continue
                        
                        # Add the source_id as the 8th column for provenance
                        cols.append(source_id)
                        rows.append(cols)

    columns = ['subject', 'predicate', 'object', 'date', 'prov', 'cited_agent', 'note', 'source']
    df = pd.DataFrame(rows, columns=columns)
    
    # CLEANING: Drop rows where the LLM put '-' as the object (e.g., "George Everest died -")
    # In KGE, an object cannot be null.
    df = df[df['object'] != '-']
    
    return df

def export_temporal_quads(df, output_path):
    """
    STRATEGY 1: (Subject, Predicate, Object, Date)
    Best for TKGE libraries like PyKEEN.
    """
    # Filter out rows with genuinely null dates if the model can't handle them
    quads = df[['subject', 'predicate', 'object', 'date']].copy()
    quads.to_csv(output_path, sep='\t', index=False, header=False)
    print(f"Exported {len(quads)} temporal quads to {output_path}")

def export_ampligraph_timestamped(df, output_path):
    """
    STRATEGY 2: (Subject, Predicate@[Date], Object)
    Quick hack for Ampligraph 2.0. Creates timestamped predicates.
    """
    triples = df[['subject', 'predicate', 'object', 'date']].copy()
    
    # Concatenate predicate and date (e.g., "opposed@[1867]")
    # If date is '-', fallback to just the predicate
    triples['p_time'] = triples.apply(
        lambda row: f"{row['predicate']}@[{row['date']}]" if row['date'] != '-' else row['predicate'], 
        axis=1
    )
    
    export_df = triples[['subject', 'p_time', 'object']]
    export_df.to_csv(output_path, sep='\t', index=False, header=False)
    print(f"Exported {len(export_df)} timestamped triples to {output_path}")

def export_ampligraph_reified(df, output_path):
    """
    STRATEGY 3: Reified Graph for Ampligraph 2.0
    Converts 1 extraction row into 4 structural triples to handle Time properly in 3D.
    """
    reified_triples = []
    
    for idx, row in df.iterrows():
        # Generate a unique event node for this extraction (e.g., Event_8f2a...)
        event_node = f"Event_{uuid.uuid4().hex[:8]}"
        
        # 1. The Subject Triple
        reified_triples.append([event_node, "has_subject", row['subject']])
        # 2. The Predicate Triple
        reified_triples.append([event_node, "has_predicate", row['predicate']])
        # 3. The Object Triple
        reified_triples.append([event_node, "has_object", row['object']])
        
        # 4. The Temporal Triple (only if a date exists)
        if row['date'] != '-':
            reified_triples.append([event_node, "occurred_during", row['date']])
            
        # Optional: Include Provenance!
        if row['prov'] == 'HIST':
            reified_triples.append([event_node, "attributed_to", row['cited_agent']])
            
    export_df = pd.DataFrame(reified_triples, columns=['s', 'p', 'o'])
    export_df.to_csv(output_path, sep='\t', index=False, header=False)
    print(f"Exported {len(export_df)} reified triples to {output_path}")

if __name__ == "__main__":
    # 1. Parse all Markdown files in the /quads folder
    kge_dir = "./kge_space/kge-root/quads"
    if not os.path.exists(kge_dir):
        print(f"Error: Directory {kge_dir} not found.")
    else:
        print("Parsing Markdown tables...")
        df_quads = parse_markdown_tables(kge_dir)
        print(f"Successfully parsed {len(df_quads)} valid extractions.")
        
        # 2. Export to your chosen format(s)
        export_temporal_quads(df_quads, "kge_pykeen_quads.tsv")
        export_ampligraph_timestamped(df_quads, "kge_ampligraph_timestamped.tsv")
        export_ampligraph_reified(df_quads, "kge_ampligraph_reified.tsv")

skill_agent.py

#!/usr/bin/env python3
"""
skill_agent.py — A SKILL.md-guided agentic file editor over an Ollama-compatible API.

Usage:
    python skill_agent.py \\
        --api-key  <key>           \\
        --skill    path/to/SKILL.md \\
        --task     "your task here" \\
        [--work-dir  ./my_files]   \\
        [--host      https://...]  \\
        [--model     qwen3:235b]   \\
        [--chunk-size 100]         \\
        [--history-turns 8]        \\
        [--max-tool-result 4000]   \\
        [--max-payload-bytes 200000] \\
        [--list-models]            \\
        [--verbose]
"""

import json
import re
import sys
from pathlib import Path
from argparse import ArgumentParser, RawDescriptionHelpFormatter
from ollama import Client
import urllib.request
import urllib.parse
from datetime import datetime

# ── Defaults ──────────────────────────────────────────────────────────────────

DEFAULT_HOST              = "https://rcsllm.carleton.ca/rcsapi"
DEFAULT_MODEL             = "qwen3.5:122b"
MAX_TURNS                 = 100
DEFAULT_CHUNK_LINES       = 100
DEFAULT_HISTORY_TURNS     = 20
DEFAULT_MAX_TOOL_RESULT   = 4_000   # chars stored per tool result in history
DEFAULT_MAX_PAYLOAD_BYTES = 200_000 # pre-flight serialized payload cap (bytes)

# ── Tool schema (Ollama/OpenAI function-calling format) ───────────────────────

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_files",
            "description": (
                "List files in a directory relative to the working directory. "
                "Use this to discover what files are available before reading them."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "directory": {
                        "type": "string",
                        "description": "Relative path to directory. Use '.' for the working directory.",
                        "default": "."
                    },
                    "pattern": {
                        "type": "string",
                        "description": "Glob pattern to filter results (e.g. '*.md', '**/*.txt').",
                        "default": "**/*"
                    }
                },
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "grep_file",
            "description": (
                "Search for a regular expression inside a file and return matching lines "
                "with surrounding context. Line numbers are 1-indexed."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Relative path to the file."
                    },
                    "pattern": {
                        "type": "string",
                        "description": "Regular expression to search for."
                    },
                    "context_lines": {
                        "type": "integer",
                        "description": "Lines of context around each match.",
                        "default": 2
                    }
                },
                "required": ["path", "pattern"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": (
                "Read a chunk of a file's text. Use offset/limit to paginate through large files. "
                "For discovery, use grep_file first, then read the relevant chunk."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Relative path to the file."
                    },
                    "offset": {
                        "type": "integer",
                        "description": "1-indexed starting line. Default 1.",
                        "default": 1
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Maximum number of lines to read.",
                        "default": DEFAULT_CHUNK_LINES
                    }
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "str_replace",
            "description": (
                "Replace a unique string snippet in a file with new text. "
                "The old_str must appear exactly once; otherwise an error is returned. "
                "Use this for surgical edits instead of write_file whenever possible."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Relative path to the file."
                    },
                    "old_str": {
                        "type": "string",
                        "description": "Exact existing snippet to replace (must be unique in the file)."
                    },
                    "new_str": {
                        "type": "string",
                        "description": "Text to substitute for old_str."
                    }
                },
                "required": ["path", "old_str", "new_str"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": (
                "Write content to a file, creating it or overwriting it. "
                "Parent directories are created automatically. "
                "IMPORTANT: You must NEVER use this to delete content — "
                "only to create or update files. Prefer str_replace for large files."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Relative path to the file to write."
                    },
                    "content": {
                        "type": "string",
                        "description": "The full content to write to the file."
                    }
                },
                "required": ["path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "append_file",
            "description": "Append text to the end of an existing file (or create it if absent).",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Relative path to the file."
                    },
                    "content": {
                        "type": "string",
                        "description": "Text to append."
                    }
                },
                "required": ["path", "content"]
            }
        }
    }, 
    {
        "type": "function",
        "function": {
            "name": "wikidata_search",
            "description": (
                "Search Wikidata for an entity to find its canonical Q-identifier and description. "
                "Use this to reconcile entities to global knowledge graph identifiers."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The name of the entity to search for (e.g., 'Survey of India')."
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Number of results to return.",
                        "default": 3
                    }
                },
                "required": ["query"]
            }
        }
    }
]

# ── Helpers ───────────────────────────────────────────────────────────────────

def _msg_role(msg) -> str | None:
    """Return the role of a message regardless of whether it's a dict or SDK object."""
    if isinstance(msg, dict):
        return msg.get("role")
    return getattr(msg, "role", None)


def prune_messages(messages, max_turns: int):
    """
    Retain only the initial user task plus the last *max_turns* complete
    assistant/tool cycles.  This prevents the context window from growing
    indefinitely on long tasks.
    """
    if len(messages) <= 1:
        return messages

    # messages[0] is the original user task — always keep it.
    task_msg = messages[0]

    # Group remaining messages into (assistant_msg, [tool_msgs, ...]) tuples.
    turns = []
    i = 1
    while i < len(messages):
        if _msg_role(messages[i]) == "assistant":
            assistant_msg = messages[i]
            tool_msgs = []
            i += 1
            while i < len(messages) and _msg_role(messages[i]) == "tool":
                tool_msgs.append(messages[i])
                i += 1
            turns.append((assistant_msg, tool_msgs))
        else:
            # Unexpected message type; skip it to stay robust.
            i += 1

    kept = [task_msg]
    for assistant_msg, tool_msgs in turns[-max_turns:]:
        kept.append(assistant_msg)
        kept.extend(tool_msgs)
    return kept


def _to_serializable(msg) -> dict:
    """
    Convert a message to a plain dict for json.dumps size estimation.
    Handles both raw dicts (user/tool messages) and Ollama SDK Message
    objects (Pydantic models returned by client.chat).
    """
    if isinstance(msg, dict):
        return msg
    if hasattr(msg, "model_dump"):   # Pydantic v2 (ollama SDK >= 0.2)
        return msg.model_dump()
    if hasattr(msg, "dict"):         # Pydantic v1 fallback
        return msg.dict()
    return vars(msg)                 # last resort


def _payload_bytes(system_prompt: str, history: list) -> int:
    """Return the UTF-8 byte length of the serialized chat payload."""
    serializable = [{"role": "system", "content": system_prompt}] + [
        _to_serializable(m) for m in history
    ]
    return len(json.dumps(serializable).encode("utf-8"))


def enforce_payload_limit(
    system_prompt: str,
    active_history: list,
    history_turns: int,
    max_payload_bytes: int,
) -> list:
    """
    Pre-flight check: if the serialized payload exceeds max_payload_bytes,
    iteratively drop the oldest non-task turns until it fits (or only the
    task message remains).

    The system prompt (which contains the full SKILL) is never modified —
    it is a fixed cost.  If even a bare payload (system prompt + task only)
    exceeds the limit, emit a warning and proceed; the server error is more
    informative than silent truncation of the skill.
    """
    while len(active_history) > 1:
        if _payload_bytes(system_prompt, active_history) <= max_payload_bytes:
            break
        history_turns = max(1, history_turns - 1)
        active_history = prune_messages(active_history, max_turns=history_turns)

    # Warn if the irreducible payload (skill + task only) is already over budget.
    bare_bytes = _payload_bytes(system_prompt, active_history[:1])
    if bare_bytes > max_payload_bytes:
        print(
            f"[WARN] Skill + task alone ({bare_bytes:,} bytes) "
            f"exceeds --max-payload-bytes ({max_payload_bytes:,}). "
            f"Increase --max-payload-bytes to match your server's limit."
        )

    return active_history


def truncate_tool_result(result: str, max_chars: int) -> str:
    """Cap a tool result to max_chars, appending a truncation notice if needed."""
    if len(result) <= max_chars:
        return result
    return result[:max_chars] + f"\n… [truncated — {len(result)} chars total]"


# ── Tool execution ────────────────────────────────────────────────────────────

def execute_tool(name: str, args: dict, work_dir: Path, chunk_size: int) -> str:
    """Dispatch a tool call and return its string result."""

    if name == "list_files":
        directory = work_dir / args.get("directory", ".")
        pattern   = args.get("pattern", "**/*")
        try:
            matches = sorted(p for p in directory.glob(pattern) if p.is_file())
        except Exception as e:
            return f"ERROR: {e}"
        if not matches:
            return "(no files matched)"
        return "\n".join(str(p.relative_to(work_dir)) for p in matches)

    elif name == "grep_file":
        missing = [k for k in ("path", "pattern") if k not in args]
        if missing:
            return f"ERROR: missing required argument(s): {missing}. grep_file requires 'path' and 'pattern'."
        path = work_dir / args["path"]
        pattern = args["pattern"]
        ctx = int(args.get("context_lines", 2))
        try:
            text = path.read_text(encoding="utf-8")
            lines = text.splitlines()
            try:
                regex = re.compile(pattern)
            except re.error as e:
                return f"ERROR: invalid regex — {e}"
            matched = set()
            for i, line in enumerate(lines):
                if regex.search(line):
                    matched.update(range(max(0, i - ctx), min(len(lines), i + ctx + 1)))
            if not matched:
                return "(no matches)"
            out = []
            for i in sorted(matched):
                prefix = ">>> " if regex.search(lines[i]) else "    "
                out.append(f"{prefix}{i + 1:4d}: {lines[i]}")
            return "\n".join(out)
        except FileNotFoundError:
            return f"ERROR: file not found — {path}"
        except Exception as e:
            return f"ERROR: {e}"

    elif name == "read_file":
        if "path" not in args:
            return "ERROR: missing required argument 'path'. read_file requires 'path'."
        path = work_dir / args["path"]
        try:
            text = path.read_text(encoding="utf-8")
            lines = text.splitlines()
            total = len(lines)

            offset = int(args.get("offset", 1))
            limit = int(args.get("limit", chunk_size))
            if offset < 1:
                offset = 1
            start = offset - 1  # 0-indexed
            end = start + limit

            if total == 0:
                return "(empty file)"
            if start >= total:
                return f"[requested lines {offset}+ of {total}]\nERROR: offset past end of file"

            chunk = lines[start:end]
            header = f"[lines {offset}-{min(end, total)} of {total}]"
            return f"{header}\n" + "\n".join(chunk)
        except FileNotFoundError:
            return f"ERROR: file not found — {path}"
        except Exception as e:
            return f"ERROR: {e}"

    elif name == "str_replace":
        missing = [k for k in ("path", "old_str", "new_str") if k not in args]
        if missing:
            return f"ERROR: missing required argument(s): {missing}. str_replace requires 'path', 'old_str', and 'new_str'."
        path = work_dir / args["path"]
        old_str = args["old_str"]
        new_str = args["new_str"]
        try:
            original = path.read_text(encoding="utf-8")
            count = original.count(old_str)
            if count == 0:
                return "ERROR: old_str not found in file"
            if count > 1:
                return f"ERROR: old_str is ambiguous (found {count} times). Provide a larger unique snippet."
            updated = original.replace(old_str, new_str, 1)
            path.write_text(updated, encoding="utf-8")
            return f"OK: replaced {len(old_str)} chars with {len(new_str)} chars → {path.relative_to(work_dir)}"
        except FileNotFoundError:
            return f"ERROR: file not found — {path}"
        except Exception as e:
            return f"ERROR: {e}"

    elif name == "write_file":
        missing = [k for k in ("path", "content") if k not in args]
        if missing:
            return f"ERROR: missing required argument(s): {missing}. write_file requires 'path' and 'content'."
        path = work_dir / args["path"]
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(args["content"], encoding="utf-8")
            return f"OK: wrote {len(args['content'])} chars → {path.relative_to(work_dir)}"
        except Exception as e:
            return f"ERROR: {e}"

    elif name == "append_file":
        missing = [k for k in ("path", "content") if k not in args]
        if missing:
            return f"ERROR: missing required argument(s): {missing}. append_file requires 'path' and 'content'."
        path = work_dir / args["path"]
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            with path.open("a", encoding="utf-8") as f:
                f.write(args["content"])
            return f"OK: appended {len(args['content'])} chars → {path.relative_to(work_dir)}"
        except Exception as e:
            return f"ERROR: {e}"

    elif name == "wikidata_search":
        if "query" not in args:
            return "ERROR: missing required argument 'query'."
        
        query = args["query"]
        limit = int(args.get("limit", 3))
        
        # Wikidata wbsearchentities API
        params = {
            "action": "wbsearchentities",
            "search": query,
            "language": "en",
            "format": "json",
            "limit": limit
        }
        url = "https://www.wikidata.org/w/api.php?" + urllib.parse.urlencode(params)
        
        try:
            # Wikidata requires a user-agent
            req = urllib.request.Request(url, headers={'User-Agent': 'SkillAgent/1.0 (Agentic KG Builder)'})
            with urllib.request.urlopen(req) as response:
                data = json.loads(response.read().decode('utf-8'))
            
            search_results = data.get("search", [])
            if not search_results:
                return f"No Wikidata matches found for '{query}'."
            
            out = [f"Wikidata search results for '{query}':"]
            for r in search_results:
                qid = r.get("id")
                label = r.get("label", "No label")
                desc = r.get("description", "No description provided")
                out.append(f" - {qid}: {label} ({desc})")
            
            return "\n".join(out)
        except Exception as e:
            return f"ERROR querying Wikidata: {e}"

    return f"ERROR: unknown tool '{name}'"

# ── Agent loop ────────────────────────────────────────────────────────────────

def build_system_prompt(skill_text: str, chunk_size: int, history_turns: int) -> str:
    # Grab the actual system time formatted nicely
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    return f"""You are an autonomous file-editing agent operating inside a defined working directory.
Current System Time: {current_time}

You have the following tools:
  • list_files  — discover available files
  • grep_file   — search a file with regex (returns line numbers + context)
  • read_file   — read a chunk of a file using offset/limit (default limit {chunk_size})
  • str_replace — replace a unique snippet inside a file (preferred for edits)
  • write_file  — overwrite or create a file (use for new files or full rewrites only)
  • append_file — append text to the end of a file
  • wikidata_search — search Wikidata to get canonical Q-IDs and descriptions for entities

Rules:
  1. NEVER delete files or remove content without replacement.
  2. For large files, use grep_file to locate content, then read_file in chunks.
  3. Prefer str_replace over write_file so you don't accidentally destroy unchanged content.
  4. Only the most recent {history_turns} tool cycles are remembered. Re-read files if you need earlier context.

Follow the SKILL definition below precisely.
When the task is fully complete, respond with a concise plain-text summary of every file you created or modified. Do not call any further tools.

═══ SKILL ═══
{skill_text}
═══════════════"""


def run_agent(
    client:            Client,
    model:             str,
    skill_text:        str,
    task:              str,
    work_dir:          Path,
    verbose:           bool,
    chunk_size:        int,
    history_turns:     int,
    max_tool_result:   int,
    max_payload_bytes: int,
) -> None:

    system_prompt = build_system_prompt(skill_text, chunk_size, history_turns)
    skill_bytes   = len(skill_text.encode("utf-8"))
    messages      = [{"role": "user", "content": task}]

    bar = "─" * 60
    print(f"\n{bar}")
    print(f"  SKILL AGENT")
    print(f"{bar}")
    print(f"  Model            : {model}")
    print(f"  Work dir         : {work_dir}")
    print(f"  Task             : {task}")
    print(f"  Skill size       : {skill_bytes:,} bytes (never truncated)")
    print(f"  Chunk size       : {chunk_size}")
    print(f"  History window   : {history_turns} turns")
    print(f"  Max tool result  : {max_tool_result} chars")
    print(f"  Max payload      : {max_payload_bytes} bytes")
    print(f"{bar}\n")

    for turn in range(1, MAX_TURNS + 1):
        active_history = prune_messages(messages, max_turns=history_turns)

        # ── Pre-flight payload size check ────────────────────────────────────
        active_history = enforce_payload_limit(
            system_prompt,
            active_history,
            history_turns,
            max_payload_bytes,
        )

        response = client.chat(
            model=model,
            messages=[{"role": "system", "content": system_prompt}] + active_history,
            tools=TOOLS,
        )

        msg        = response.message
        tool_calls = msg.tool_calls or []
        messages.append(msg)

        # ── No tool calls → agent is done ───────────────────────────────────
        if not tool_calls:
            print(f"\n{bar}")
            print("  COMPLETE")
            print(f"{bar}")
            print(msg.content or "(no summary returned)")
            print(f"{bar}\n")
            return

        # ── Execute tool calls ───────────────────────────────────────────────
        for call in tool_calls:
            fn   = call.function
            name = fn.name
            args = fn.arguments

            # Defensive: arguments are normally a dict in recent SDKs,
            # but fall back to json.loads for older servers / edge cases.
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except json.JSONDecodeError:
                    args = {}
            elif not isinstance(args, dict):
                args = {}

            # Pretty-print the call
            arg_str = ", ".join(
                f"{k}={repr(v)[:80]}" for k, v in args.items()
            )
            print(f"  [{turn:02d}] ▶ {name}({arg_str})")

            result = execute_tool(name, args, work_dir, chunk_size)

            # ── Truncate result before storing in history ────────────────────
            stored_result = truncate_tool_result(result, max_tool_result)
            if verbose:
                preview = result[:300].replace("\n", "\n       ")
                print(f"       ← {preview}")
                if len(result) > 300:
                    print(f"       … ({len(result)} chars total)")
                if len(result) > max_tool_result:
                    print(f"       [stored truncated to {max_tool_result} chars]")

            # Feed result back — Ollama expects role "tool"
            messages.append({
                "role":    "tool",
                "content": stored_result,
            })

    print("[WARN] Maximum turn limit reached without completion.")


# ── CLI ───────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = ArgumentParser(
        description=__doc__,
        formatter_class=RawDescriptionHelpFormatter
    )
    parser.add_argument("--host",               default=DEFAULT_HOST,              help="Ollama server URL")
    parser.add_argument("--model",              default=DEFAULT_MODEL,             help="Model name")
    parser.add_argument("--api-key",            required=True,                     help="API key (x-api-key header)")
    parser.add_argument("--skill",              default=None,                      help="Path to SKILL.md")
    parser.add_argument("--work-dir",           default=".",                       help="Working directory for file ops (default: cwd)")
    parser.add_argument("--task",               default=None,                      help="Task description for the agent")
    parser.add_argument("--chunk-size",         type=int, default=DEFAULT_CHUNK_LINES,       help="Default lines per read_file chunk")
    parser.add_argument("--history-turns",      type=int, default=DEFAULT_HISTORY_TURNS,     help="Tool turns to retain in context window")
    parser.add_argument("--max-tool-result",    type=int, default=DEFAULT_MAX_TOOL_RESULT,   help="Max chars stored per tool result in history")
    parser.add_argument("--max-payload-bytes",  type=int, default=DEFAULT_MAX_PAYLOAD_BYTES, help="Pre-flight serialized payload cap in bytes")
    parser.add_argument("--list-models",        action="store_true",               help="List available models and exit")
    parser.add_argument("--verbose",            action="store_true",               help="Print tool results in full")
    args = parser.parse_args()

    client = Client(
        host=args.host,
        headers={"Authorization": f"Bearer {args.api_key}"},
        timeout=2400
    )

    # ── List models mode ─────────────────────────────────────────────────────
    if args.list_models:
        models = client.list()
        print("Available models:")
        for m in models.models:
            print(f"  • {m.model}")
        sys.exit(0)

    # ── Validate required args ───────────────────────────────────────────────
    if not args.skill:
        parser.error("--skill is required unless --list-models is set")
    if not args.task:
        parser.error("--task is required unless --list-models is set")

    skill_path = Path(args.skill)
    if not skill_path.exists():
        sys.exit(f"ERROR: SKILL file not found: {skill_path}")

    skill_text = skill_path.read_text(encoding="utf-8")

    work_dir   = Path(args.work_dir).resolve()

    if not work_dir.is_dir():
        sys.exit(f"ERROR: Working directory does not exist: {work_dir}")

    # Sync the tool schema default so the model knows what to expect.
    for tool in TOOLS:
        if tool["function"]["name"] == "read_file":
            tool["function"]["parameters"]["properties"]["limit"]["default"] = args.chunk_size

    run_agent(
        client            = client,
        model             = args.model,
        skill_text        = skill_text,
        task              = args.task,
        work_dir          = work_dir,
        verbose           = args.verbose,
        chunk_size        = args.chunk_size,
        history_turns     = args.history_turns,
        max_tool_result   = args.max_tool_result,
        max_payload_bytes = args.max_payload_bytes,
    )


if __name__ == "__main__":
    main()

SKILL_kg_extraction.md

name	kge-extraction
description	Extract temporally-grounded knowledge graph quads from sources. Maintains a markdown-based quad store, entity index, and append-only log. Distinguishes between a source's own arguments (ARG) and attributions (HISTOGPHY). Operates in a two-phase workflow: Extract (generate raw quads) and Lint (normalize entities and reconcile graphs).

KGE Extraction Skill

You are a disciplined knowledge graph extraction agent. Your job is to read sources and extract structured subject-predicate-object relationships.

File Architecture

kge-root/
├── SCHEMA.md               ← Predicate vocabulary, date schema, extraction rules
├── index-entities.md       ← Canonical entities: Name | Type | Aliases
├── index-sources.md        ← Source registry: ID | Author | Title | Summary
├── log.md                  ← Append-only chronological operation log
└── quads/                  ← One markdown file per source containing its extracted quads
    └── <source-id>.md

Raw sources are immutable. You read them; never modify them. You manage the kge-root directory.

---

The Extraction Format (Markdown Quads)

Write quads in strict Markdown tables. This prevents syntax errors. Do not worry about IDs; a downstream script will generate them.

Table format for quads/<source-id>.md:

Subject	Predicate	Object	Date	Prov	Cited Agent	Note
Survey of India	opposed	Local Naming Conv.	1856	ARG	-	Survey argued for single name
Weber, Max	argued	Protestant Ethic	1905	HISTOGPHY	Smith, J.	Cited by author as foundational

HARD CONSTRAINTS FOR QUADS:

1. Objects must be Named Entities, NOT clauses.
   • Bad Object: "needed a single authoritative name" (Verb phrase)
   • Good Object: "Uniform Nomenclature Policy" (Reified Concept)
2. Provenance (Prov) is either ARG or HISTOGPHY:
   • ARG: The source author's own claim. (Cited Agent is -)
   • HISTOGPHY: A claim attributed to prior scholarship. (Cited Agent must be named).
3. Predicates must be exact. Use only terms from SCHEMA.md. Do not invent predicates.

---

Operations

Always use grep_file to locate symbols/sections. Use read_file with small limits. Prefer str_replace for small edits.

1. EXTRACT

Command: extract <source-id> <path-to-source>

1. Read the source document.
2. Register: Use the append_file tool to add the source to index-sources.md. NEVER use write_file for this.
3. Extract: Create quads/<source-id>.md. Write a markdown table extracting the structurally significant claims.
   • Ask yourself: Would this look right as a node-link diagram?
   • Convert passive voice: "Edict revoked by King" -> Subject: King | Predicate: reversed | Object: Edict.
   • Reify arguments: If the text says "argued that the survey methodology was flawed", extract Subject: Author | Predicate: opposed | Object: Survey Methodology.
4. Log: Append to log.md using the exact current system time (e.g., ### 2026-05-28 14:30:00 - EXTRACT <source-id>).

2. INDEX

Command: index <source-id>

Run this immediately after EXTRACT.

1. Read the newly created quads/<source-id>.md.
2. For every Subject, Object, and Cited Agent in the new quads, use grep_file on index-entities.md to check if it already exists.
3. If an entity is missing:
   • Run wikidata_search to find its canonical Q-ID.
   • CRITICAL SEARCH RULE: Search ONLY for the bare proper noun (e.g., "Giacomo Medici"). NEVER include descriptive context (like "art trafficker") in the search query, or the API will fail.
   • Use the descriptive context in your own mind to pick the correct result from the returned list (e.g., pick the Q-ID described as "art dealer", not the "Italian general").
   • If no good match is found, use Q-None.
   • Append it to the index in this format: - **Canonical Name:** | Q-ID | Type (Person/Org/Place/Concept/Event) | Aliases
4. If you notice the extraction used a slight variant (e.g., "Survey India" instead of "Survey of India"), use str_replace to fix the quad table to match the canonical index.

3. LINT

Command: lint

Periodic health check to refine the knowledge graph. Run through this checklist:

1. Schema Check: Do all quads in recent files use exact predicates from SCHEMA.md? Fix any invented predicates.
2. Null Date Inference: Scan recent quads for missing dates (-). Can the date be logically inferred from surrounding events?
3. Contradiction Flagging: Scan for identical Subject/Object pairs with different predicates/dates. Add a ## Contradictions section to log.md.
4. Entity Consolidation: Scan index-entities.md for obvious duplicates (e.g., "Charles I" and "King Charles I"). Merge them, update aliases, and fix affected quad files.
5. Q-ID Reconciliation: Scan index-entities.md for entities marked as Q-None. Retry wikidata_search using ONLY the bare entity name (e.g., search "Giacomo Medici", NOT "Giacomo Medici art trafficker"). Read the descriptions of the results to find the match, update the index, and replace Q-None with the correct Q-ID.

Append a summary of changes to log.md.

---

Session Start Protocol

Whenever you receive a new task, you MUST perform these steps before mutating any files:

1. Run list_files on the root directory to see what index and log files already exist.
2. Run list_files on the quads/ directory to see what sources have already been extracted.
3. If SCHEMA.md or log.md exist, read_file to understand the current graph schema and recent history.
4. Only AFTER orienting yourself to the existing files may you begin the requested command.