Memory & Search

Knowledge graph, search, visualization, and command-history utilities.

knowledge_graph

logger = logging.getLogger(__name__) module-attribute

_get_similar_by_embedding(query, candidates, model='nomic-embed-text', provider='ollama', top_k=20)

Pre-filter candidates by embedding cosine similarity.

Returns top-K candidate strings most similar to query. Falls back to returning all candidates if embedding fails.

Source code in npcpy/memory/knowledge_graph.py
def _get_similar_by_embedding(query, candidates, model='nomic-embed-text',
                              provider='ollama', top_k=20):
    """Pre-filter candidates by embedding cosine similarity.

    Returns top-K candidate strings most similar to query.
    Falls back to returning all candidates if embedding fails.
    """
    if not candidates or len(candidates) <= top_k:
        return list(candidates)

    try:
        from npcpy.gen.embeddings import get_embeddings

        query_emb = np.array(get_embeddings([query], model, provider)[0])
        cand_embs = get_embeddings(list(candidates), model, provider)

        similarities = []
        for i, emb in enumerate(cand_embs):
            emb_arr = np.array(emb)
            norm_product = np.linalg.norm(query_emb) * np.linalg.norm(emb_arr)
            if norm_product > 0:
                sim = float(np.dot(query_emb, emb_arr) / norm_product)
            else:
                sim = 0.0
            similarities.append((sim, i))

        similarities.sort(key=lambda x: -x[0])
        return [candidates[idx] for _, idx in similarities[:top_k]]
    except Exception as e:
        logger.warning(f"Embedding pre-filter failed, using all candidates: {e}")
        return list(candidates)

abstract(groups, model, provider, npc=None, context=None, **kwargs)

Create more abstract terms from groups.

Source code in npcpy/llm_funcs.py
def abstract(groups, 
             model, 
             provider, 
             npc=None,
             context: str = None, 
             **kwargs):
    """
    Create more abstract terms from groups.
    """
    sample_groups = random.sample(groups, min(len(groups), max(3, len(groups) // 2)))

    groups_text_for_prompt = "\n".join([f'- "{g["name"]}"' for g in sample_groups])

    prompt = f"""
        Create more abstract categories from this list of groups.

        Groups:
        {groups_text_for_prompt}

        You will create higher-level concepts that interrelate between the given groups. 

        Create abstract categories that encompass multiple related facts, but do not unnecessarily combine facts with conjunctions. For example, do not try to combine "characters", "settings", and "physical reactions" into a
        compound group like "Characters, Setting, and Physical Reactions". This kind of grouping is not productive and only obfuscates true abstractions. 
        For example, a group that might encompass the three aforermentioned names might be "Literary Themes" or "Video Editing Functionis", depending on the context.
        Your aim is to abstract, not to just arbitrarily generate associations. 

        Group names should never be more than two words. They should not contain gerunds. They should never contain conjunctions like "AND" or "OR".
        Generate no more than 5 new concepts and no fewer than 2. 

        Respond with JSON:
        """ + '{"groups": [{"name": "abstract category name"}]}'

    response = get_llm_response(prompt,
                                model=model,
                                provider=provider,
                                format="json",
                                npc=npc,
                                context=context,
                                **kwargs)

    return response["response"].get("groups", [])

consolidate_facts_llm(new_fact, existing_facts, model, provider, npc=None, context=None, **kwargs)

Uses an LLM to decide if a new fact is novel or redundant.

Source code in npcpy/llm_funcs.py
def consolidate_facts_llm(new_fact, 
                          existing_facts, 
                          model, 
                          provider, 
                          npc=None,
                          context: str =None,
                          **kwargs):
    """
    Uses an LLM to decide if a new fact is novel or redundant.
    """
    prompt = f"""
        Analyze the "New Fact" in the context of the "Existing Facts" list.
        Your task is to determine if the new fact provides genuinely new information or if it is essentially a repeat or minor rephrasing of information already present.

        New Fact:
        "{new_fact['statement']}"

        Existing Facts:
        {json.dumps([f['statement'] for f in existing_facts], indent=2)}

        Possible decisions:
        - 'novel': The fact introduces new, distinct information not covered by the existing facts.
        - 'redundant': The fact repeats information already present in the existing facts.

        Respond with a JSON object:
        """ + '{"decision": "novel or redundant", "reason": "A brief explanation for your decision."}'
    response = get_llm_response(prompt,
                                model=model, 
                                provider=provider, 
                                format="json", 
                                npc=npc,
                                context=context,
                                **kwargs)
    return response['response']

find_similar_facts_chroma(collection, query, query_embedding, n_results=5, metadata_filter=None)

Find facts similar to the query using pre-generated embedding.

Source code in npcpy/memory/knowledge_graph.py
def find_similar_facts_chroma(
    collection,
    query: str,
    query_embedding: List[float],
    n_results: int = 5,
    metadata_filter: Optional[Dict] = None,
) -> List[Dict]:
    """Find facts similar to the query using pre-generated embedding."""
    try:
        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            where=metadata_filter,
        )

        formatted_results = []
        for i, doc in enumerate(results["documents"][0]):
            formatted_results.append(
                {
                    "fact": doc,
                    "metadata": results["metadatas"][0][i],
                    "id": results["ids"][0][i],
                    "distance": (
                        results["distances"][0][i] if "distances" in results else None
                    ),
                }
            )

        return formatted_results
    except Exception as e:
        logger.warning(f"Error searching in Chroma: {e}")
        return []

generate_groups(facts, model=None, provider=None, npc=None, context=None, **kwargs)

Generate conceptual groups for facts

Source code in npcpy/llm_funcs.py
def generate_groups(facts, 
                    model=None,
                    provider=None,
                    npc=None,
                    context: str =None, 
                    **kwargs):
    """Generate conceptual groups for facts"""

    facts_text = "\n".join([f"- {fact['statement']}" for fact in facts])

    prompt = f"""
    Generate conceptual groups for this group off facts:

    {facts_text}

    Create categories that encompass multiple related facts, but do not unnecessarily combine facts with conjunctions. 

    Your aim is to generalize commonly occurring ideas into groups, not to just arbitrarily generate associations. 
    Focus on the key commonly occurring items and expresions.     

    Group names should never be more than two words. They should not contain gerunds. They should never contain conjunctions like "AND" or "OR".
    Respond with JSON:
    """ + '{"groups": [{"name": "group name"}]}'

    response = get_llm_response(prompt,
                                model=model,
                                provider=provider,
                                format="json",
                                context=context,
                                npc=npc,
                                **kwargs)

    return response["response"].get("groups", [])

get_facts(content_text, model=None, provider=None, npc=None, context=None, attempt_number=1, n_attempts=3, **kwargs)

Extract facts from content text

Source code in npcpy/llm_funcs.py
def get_facts(content_text, 
              model= None,
              provider = None,
              npc=None,
              context : str=None, 
              attempt_number=1,
              n_attempts=3,

              **kwargs):
    """Extract facts from content text"""

    prompt = f"""
    Extract facts from this text. A fact is a specific statement that can be sourced from the text.

    Example: if text says "the moon is the earth's only currently known satellite", extract:
    - "The moon is a satellite of earth" 
    - "The moon is the only current satellite of earth"
    - "There may have been other satellites of earth" (inferred from "only currently known")

        A fact is a piece of information that makes a statement about the world.
        A fact is typically a sentence that is true or false.
        Facts may be simple or complex. They can also be conflicting with each other, usually
        because there is some hidden context that is not mentioned in the text.
        In any case, it is simply your job to extract a list of facts that could pertain to
        an individual's personality.

        For example, if a message says:
            "since I am a doctor I am often trying to think up new ways to help people.
            Can you help me set up a new kind of software to help with that?"
        You might extract the following facts:
            - The individual is a doctor
            - They are helpful

        Another example:
            "I am a software engineer who loves to play video games. I am also a huge fan of the
            Star Wars franchise and I am a member of the 501st Legion."
        You might extract the following facts:
            - The individual is a software engineer
            - The individual loves to play video games
            - The individual is a huge fan of the Star Wars franchise
            - The individual is a member of the 501st Legion

        Another example:
            "The quantum tunneling effect allows particles to pass through barriers
            that classical physics says they shouldn't be able to cross. This has
            huge implications for semiconductor design."
        You might extract these facts:
            - Quantum tunneling enables particles to pass through barriers that are
              impassable according to classical physics
            - The behavior of quantum tunneling has significant implications for
              how semiconductors must be designed

        Another example:
            "People used to think the Earth was flat. Now we know it's spherical,
            though technically it's an oblate spheroid due to its rotation."
        You might extract these facts:
            - People historically believed the Earth was flat
            - It is now known that the Earth is an oblate spheroid
            - The Earth's oblate spheroid shape is caused by its rotation

        Another example:
            "My research on black holes suggests they emit radiation, but my professor
            says this conflicts with Einstein's work. After reading more papers, I
            learned this is actually Hawking radiation and doesn't conflict at all."
        You might extract the following facts:
            - Black holes emit radiation
            - The professor believes this radiation conflicts with Einstein's work
            - The radiation from black holes is called Hawking radiation
            - Hawking radiation does not conflict with Einstein's work

        Another example:
            "During the pandemic, many developers switched to remote work. I found
            that I'm actually more productive at home, though my company initially
            thought productivity would drop. Now they're keeping remote work permanent."
        You might extract the following facts:
            - The pandemic caused many developers to switch to remote work
            - The individual discovered higher productivity when working from home
            - The company predicted productivity would decrease with remote work
            - The company decided to make remote work a permanent option

        Thus, it is your mission to reliably extract lists of facts.

    Here is the text:
    Text: "{content_text}"

    Facts should never be more than one or two sentences, and they should not be overly complex or literal. They must be explicitly
    derived or inferred from the source text. Do not simply repeat the source text verbatim when stating the fact. 

    No two facts should share substantially similar claims. They should be conceptually distinct and pertain to distinct ideas, avoiding lengthy convoluted or compound facts .
    Respond with JSON:
    """ + '{"facts": [{"statement": "fact statement that builds on input text to state a specific claim that can be falsified through reference to the source material", "source_text": "text snippets related to the source text", "type": "explicit or inferred"}]}'

    response = get_llm_response(prompt, 
                                model=model,
                                provider=provider, 
                                npc=npc,
                                format="json", 
                                context=context,
                                **kwargs)

    if len(response.get("response", {}).get("facts", [])) == 0 and attempt_number < n_attempts:
        print(f"  Attempt {attempt_number} to extract facts yielded no results. Retrying...")
        return get_facts(content_text, 
                         model=model, 
                         provider=provider, 
                         npc=npc,
                         context=context,
                         attempt_number=attempt_number+1,
                         n_attempts=n_attempts,
                         **kwargs)

    return response["response"].get("facts", [])

get_llm_response(prompt, model=None, provider=None, images=None, npc=None, team=None, messages=None, api_url=None, api_key=None, context=None, stream=False, attachments=None, include_usage=False, n_samples=1, matrix=None, **kwargs)

Generate a response using the specified provider and model.

Source code in npcpy/llm_funcs.py
def get_llm_response(
    prompt: str,
    model: str = None,
    provider: str = None,
    images: List[str] = None,
    npc: Any = None,
    team: Any = None,
    messages: List[Dict[str, str]] = None,
    api_url: str = None,
    api_key: str = None,
    context=None,
    stream: bool = False,
    attachments: List[str] = None,
    include_usage: bool = False,
    n_samples: int = 1,
    matrix: Optional[Dict[str, List[Any]]] = None,
    **kwargs,
):
    """Generate a response using the specified provider and model."""
    logger.debug(
        f"[get_llm_response] {len(messages) if messages else 0} messages, "
        f"prompt_len={len(prompt) if prompt else 0}, "
        f"context={'yes' if context else 'no'}"
    )

    base_model, base_provider, base_api_url, base_api_key = resolve_model_provider(
        npc=npc, team=team, model=model, provider=provider,
        api_url=api_url, api_key=api_key, images=images, attachments=attachments,
    )
    if api_key is None:
        api_key = base_api_key

    use_matrix = matrix is not None and len(matrix) > 0
    multi_sample = n_samples and n_samples > 1

    if not use_matrix and not multi_sample:
        # Simple single-call path
        run_model, run_provider, run_api_url, run_api_key = resolve_model_provider(
            npc=npc, team=team, model=base_model, provider=base_provider,
            api_url=api_url, api_key=api_key, images=images, attachments=attachments,
        )
        tool_capable = bool(kwargs.get("tools"))
        system_message = get_system_message(npc, team, tool_capable=tool_capable) if npc is not None else "You are a helpful assistant."

        # Build messages
        run_messages = copy.deepcopy(messages) if messages else []
        if not run_messages:
            run_messages = [{"role": "system", "content": system_message}]

        # Build full text from prompt and/or context
        full_text = ""
        if prompt and context:
            full_text = f"{prompt}\n\n\nUser Provided Context: {context}"
        elif prompt:
            full_text = prompt
        elif context:
            full_text = f"User Provided Context: {context}"

        if full_text:
            if run_messages[-1]["role"] == "user" and isinstance(run_messages[-1]["content"], str):
                run_messages[-1]["content"] += "\n" + full_text
            else:
                run_messages.append({"role": "user", "content": full_text})

        if not run_model:
            raise ValueError("No model specified. Please set a model in your NPC configuration or team settings.")
        return get_litellm_response(
            full_text or None,
            messages=run_messages,
            model=run_model,
            provider=run_provider,
            api_url=run_api_url,
            api_key=api_key,
            images=images,
            attachments=attachments,
            stream=stream,
            include_usage=include_usage,
            **kwargs,
        )

    # Matrix / multi-sample path
    combos = []
    if use_matrix:
        keys = list(matrix.keys())
        values = []
        for key in keys:
            val = matrix[key]
            if isinstance(val, (list, tuple, set)):
                values.append(list(val))
            else:
                values.append([val])
        for combo_values in itertools.product(*values):
            combos.append(dict(zip(keys, combo_values)))
    else:
        combos.append({})

    runs = []
    for combo in combos:
        run_npc = combo.get("npc", npc)
        run_team = combo.get("team", team)
        run_context = combo.get("context", context)
        extra_kwargs = dict(kwargs)
        for k, v in combo.items():
            if k not in {"model", "provider", "npc", "context", "team"}:
                extra_kwargs[k] = v

        run_model, run_provider, run_api_url, run_api_key = resolve_model_provider(
            npc=run_npc, team=run_team,
            model=combo.get("model", base_model),
            provider=combo.get("provider", base_provider),
            api_url=api_url, api_key=api_key, images=images, attachments=attachments,
        )

        tool_capable = bool(extra_kwargs.get("tools"))
        system_message = get_system_message(run_npc, run_team, tool_capable=tool_capable) if run_npc is not None else "You are a helpful assistant."

        run_messages = copy.deepcopy(messages) if messages else []
        if not run_messages:
            run_messages = [{"role": "system", "content": system_message}]

        full_text = ""
        if prompt and run_context:
            full_text = f"{prompt}\n\n\nUser Provided Context: {run_context}"
        elif prompt:
            full_text = prompt
        elif run_context:
            full_text = f"User Provided Context: {run_context}"

        if full_text:
            if run_messages[-1]["role"] == "user" and isinstance(run_messages[-1]["content"], str):
                run_messages[-1]["content"] += "\n" + full_text
            else:
                run_messages.append({"role": "user", "content": full_text})

        for sample_idx in range(max(1, n_samples or 1)):
            resp = get_litellm_response(
                full_text or None,
                messages=run_messages,
                model=run_model,
                provider=run_provider,
                api_url=run_api_url,
                api_key=api_key,
                images=images,
                attachments=attachments,
                stream=stream,
                include_usage=include_usage,
                **extra_kwargs,
            )
            runs.append({
                "response": resp.get("response") if isinstance(resp, dict) else resp,
                "raw": resp,
                "combo": combo,
                "sample_index": sample_idx,
            })

    aggregated = {
        "response": runs[0]["response"] if runs else None,
        "runs": runs,
    }
    if runs and isinstance(runs[0]["raw"], dict) and "messages" in runs[0]["raw"]:
        aggregated["messages"] = runs[0]["raw"].get("messages")
    return aggregated

Links any node (fact or concept) to ALL relevant concepts in the entire ontology.

Source code in npcpy/llm_funcs.py
def get_related_concepts_multi(node_name: str, 
                               node_type: str, 
                               all_concept_names, 
                               model: str = None,
                               provider: str = None,
                               npc=None,
                               context : str = None, 
                               **kwargs):
    """Links any node (fact or concept) to ALL relevant concepts in the entire ontology."""
    json_format = 'Respond with JSON: {"related_concepts": ["Concept A", "Concept B", ...]}'
    prompt = f"""
    Which of the following concepts from the entire ontology relate to the given {node_type}?
    Select all that apply, from the most specific to the most abstract.

    {node_type.capitalize()}: "{node_name}"

    Available Concepts:
    {json.dumps(all_concept_names, indent=2)}

    {json_format}
    """
    response = get_llm_response(prompt, 
                                model=model, 
                                provider=provider, 
                                format="json", 
                                npc=npc,
                                context=context, 
                                **kwargs)
    return response["response"].get("related_concepts", [])

Identifies which existing facts are causally or thematically related to a new fact.

Source code in npcpy/llm_funcs.py
def get_related_facts_llm(new_fact_statement, 
                          existing_fact_statements, 
                          model = None, 
                          provider = None,
                          npc = None, 
                          attempt_number = 1,
                          n_attempts = 3,
                          context='', 
                          **kwargs):
    """Identifies which existing facts are causally or thematically related to a new fact."""
    prompt = f"""
    A new fact has been learned: "{new_fact_statement}"

    Which of the following existing facts are directly related to it (causally, sequentially, or thematically)?
    Select only the most direct and meaningful connections.

    Existing Facts:
    {json.dumps(existing_fact_statements, indent=2)}

    Respond with JSON:
    """ + '{"related_facts": ["statement of a related fact", ...]}'
    response = get_llm_response(prompt,
                                model=model, 
                                provider=provider, 
                                format="json", 
                                npc=npc,
                                context=context,
                                **kwargs)   
    if attempt_number <= n_attempts:
        if not response["response"].get("related_facts", []):
            print(f"  Attempt {attempt_number} to find related facts yielded no results. Retrying...")
            return get_related_facts_llm(new_fact_statement, 
                                           existing_fact_statements, 
                                           model=model, 
                                           provider=provider, 
                                           npc=npc,
                                           attempt_number=attempt_number+1,
                                           n_attempts=n_attempts,
                                           context=context,
                                           **kwargs)    

    return response["response"].get("related_facts", [])

kg_add_concept(engine, concept_name, concept_description, npc=None, team=None, model=None, provider=None)

Add a new concept to the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_add_concept(
   engine,
   concept_name: str,
   concept_description: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Add a new concept to the knowledge graph"""
   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)

   new_concept = {
       "name": concept_name,
       "description": concept_description,
       "generation": kg_data.get('generation', 0)
   }

   kg_data['concepts'].append(new_concept)
   save_kg_to_db(engine, kg_data, team_name, npc_name, directory_path)

   return f"Added concept: {concept_name}"

kg_add_fact(engine, fact_text, npc=None, team=None, model=None, provider=None)

Add a new fact to the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_add_fact(
   engine,
   fact_text: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Add a new fact to the knowledge graph"""
   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)

   new_fact = {
       "statement": fact_text,
       "source_text": fact_text,
       "type": "manual",
       "generation": kg_data.get('generation', 0),
       "origin": "manual_add"
   }

   kg_data['facts'].append(new_fact)
   save_kg_to_db(engine, kg_data, team_name, npc_name, directory_path)

   return f"Added fact: {fact_text}"

kg_backfill_from_memories(engine, model=None, provider=None, npc=None, get_concepts=True, link_concepts_facts=False, link_concepts_concepts=False, link_facts_facts=False, dry_run=False, context='')

Backfill KG from approved memories that haven't been incorporated yet.

Source code in npcpy/memory/knowledge_graph.py
def kg_backfill_from_memories(
    engine,
    model: str = None,
    provider: str = None,
    npc=None,
    get_concepts: bool = True,
    link_concepts_facts: bool = False,
    link_concepts_concepts: bool = False,
    link_facts_facts: bool = False,
    dry_run: bool = False,
    context: str = ''
):
    """Backfill KG from approved memories that haven't been incorporated yet."""
    from sqlalchemy import text

    stats = {
        'scopes_processed': 0,
        'facts_before': 0,
        'facts_after': 0,
        'concepts_before': 0,
        'concepts_after': 0,
        'scopes': []
    }

    with engine.connect() as conn:
        stats['facts_before'] = conn.execute(text("SELECT COUNT(*) FROM kg_facts")).scalar() or 0
        stats['concepts_before'] = conn.execute(text("SELECT COUNT(*) FROM kg_concepts")).scalar() or 0

    with engine.connect() as conn:
        result = conn.execute(text("""
            SELECT id, npc, team, directory_path, initial_memory, final_memory
            FROM memory_lifecycle
            WHERE status IN ('human-approved', 'human-edited')
            ORDER BY npc, team, directory_path
        """))

        from collections import defaultdict
        memories_by_scope = defaultdict(list)
        for row in result:
            statement = row.final_memory or row.initial_memory
            scope = (row.npc or 'default', row.team or 'global_team', row.directory_path or os.getcwd())
            memories_by_scope[scope].append({
                'statement': statement,
                # Memory IS the fact — keep source_text populated so any existing consumer that
                # only reads source_text still gets the memory text.
                'source_text': statement,
                'memory_id': row.id,
                'type': 'explicit',
                'generation': 0
            })

    if dry_run:
        for scope, facts in memories_by_scope.items():
            stats['scopes'].append({
                'scope': scope,
                'memory_count': len(facts)
            })
        stats['scopes_processed'] = len(memories_by_scope)
        return stats

    for (npc_name, team_name, directory_path), facts in memories_by_scope.items():
        existing_kg = load_kg_from_db(engine, team_name, npc_name, directory_path)

        existing_statements = {f['statement'] for f in existing_kg.get('facts', [])}
        new_facts = [f for f in facts if f['statement'] not in existing_statements]

        if not new_facts:
            continue

        try:
            evolved_kg, _ = kg_evolve_incremental(
                existing_kg=existing_kg,
                new_facts=new_facts,
                model=model or (npc.model if npc else None),
                provider=provider or (npc.provider if npc else None),
                npc=npc,
                context=context,
                get_concepts=get_concepts,
                link_concepts_facts=link_concepts_facts,
                link_concepts_concepts=link_concepts_concepts,
                link_facts_facts=link_facts_facts
            )
            save_kg_to_db(engine, evolved_kg, team_name, npc_name, directory_path)

            stats['scopes'].append({
                'scope': (npc_name, team_name, directory_path),
                'facts_added': len(new_facts),
                'concepts_added': len(evolved_kg.get('concepts', [])) - len(existing_kg.get('concepts', []))
            })
            stats['scopes_processed'] += 1

        except Exception as e:
            logger.warning(f"Error processing scope {npc_name}/{team_name}: {e}")

    with engine.connect() as conn:
        stats['facts_after'] = conn.execute(text("SELECT COUNT(*) FROM kg_facts")).scalar() or 0
        stats['concepts_after'] = conn.execute(text("SELECT COUNT(*) FROM kg_concepts")).scalar() or 0

    return stats

kg_dream_process(existing_kg, model=None, provider=None, npc=None, context='', num_seeds=3)

Source code in npcpy/memory/knowledge_graph.py
def kg_dream_process(existing_kg,
                     model=None,
                     provider=None,
                     npc=None,
                     context='',
                     num_seeds=3):
    current_gen = existing_kg.get('generation', 0)
    next_gen = current_gen + 1
    logger.info(f"DREAMING (Creative Synthesis): Gen {current_gen} -> Gen {next_gen}")
    concepts = existing_kg.get('concepts', [])
    if len(concepts) < num_seeds:
        logger.info(f"Not enough concepts ({len(concepts)}) for dream. Skipping.")
        return existing_kg, {}
    seed_concepts = random.sample(concepts, k=num_seeds)
    seed_names = [c['name'] for c in seed_concepts]
    logger.info(f"Dream seeded with: {seed_names}")
    prompt = f"""
    Write a short, speculative paragraph (a 'dream') that plausibly connects the concepts of {json.dumps(seed_names)}.
    Invent a brief narrative or a hypothetical situation.
    Respond with JSON: {{"dream_text": "A short paragraph..."}}
    """
    response = get_llm_response(prompt,
                                model=model,
                                provider=provider, npc=npc,
                                format="json", context=context)
    dream_text = response['response'].get('dream_text')
    if not dream_text:
        logger.info("Failed to generate a dream narrative. Skipping.")
        return existing_kg, {}
    logger.info(f"Generated Dream: '{dream_text[:150]}...'")

    dream_kg, _ = kg_evolve_incremental(existing_kg, new_content_text=dream_text,
                                         model=model, provider=provider, npc=npc, context=context)

    original_fact_stmts = {f['statement'] for f in existing_kg['facts']}
    for fact in dream_kg['facts']:
        if fact['statement'] not in original_fact_stmts:
            fact['origin'] = 'dream'
    original_concept_names = {c['name'] for c in existing_kg['concepts']}
    for concept in dream_kg['concepts']:
        if concept['name'] not in original_concept_names:
            concept['origin'] = 'dream'
    logger.info("Dream analysis complete. New knowledge integrated.")
    return dream_kg, {}

Semantic search using embeddings via brute-force cosine similarity.

Source code in npcpy/memory/knowledge_graph.py
def kg_embedding_search(
    engine,
    query: str,
    npc=None,
    team=None,
    embedding_model: str = None,
    embedding_provider: str = None,
    similarity_threshold: float = 0.6,
    max_results: int = 20,
    include_concepts: bool = True,
    search_all_scopes: bool = True,
):
    """Semantic search using embeddings via brute-force cosine similarity."""
    from sqlalchemy import text

    try:
        from npcpy.gen.embeddings import get_embeddings
    except ImportError:
        logger.warning("Embeddings not available, falling back to keyword search")
        facts = kg_search_facts(engine, query, npc=npc, team=team,
                               search_all_scopes=search_all_scopes)
        return [{'content': f, 'type': 'fact', 'score': 0.5} for f in facts[:max_results]]

    model = embedding_model or 'nomic-embed-text'
    provider = embedding_provider or 'ollama'

    team_name = getattr(team, 'name', None) if team else None
    npc_name = getattr(npc, 'name', None) if npc else None

    results = []

    query_embedding = np.array(get_embeddings([query], model, provider)[0])

    with engine.connect() as conn:
        if search_all_scopes:
            fact_rows = conn.execute(text(
                "SELECT DISTINCT statement FROM kg_facts"
            )).fetchall()
        else:
            t_name = team_name or 'global_team'
            n_name = npc_name or 'default_npc'
            fact_rows = conn.execute(text("""
                SELECT statement FROM kg_facts
                WHERE team_name = :team AND npc_name = :npc
            """), {"team": t_name, "npc": n_name}).fetchall()

        if fact_rows:
            statements = [r.statement for r in fact_rows]
            embeddings = get_embeddings(statements, model, provider)

            for i, stmt in enumerate(statements):
                emb = np.array(embeddings[i])
                norm_p = np.linalg.norm(query_embedding) * np.linalg.norm(emb)
                if norm_p > 0:
                    sim = float(np.dot(query_embedding, emb) / norm_p)
                    if sim >= similarity_threshold:
                        results.append({'content': stmt, 'type': 'fact', 'score': sim})

        if include_concepts:
            if search_all_scopes:
                concept_rows = conn.execute(text(
                    "SELECT DISTINCT name FROM kg_concepts"
                )).fetchall()
            else:
                concept_rows = conn.execute(text("""
                    SELECT name FROM kg_concepts
                    WHERE team_name = :team AND npc_name = :npc
                """), {"team": t_name, "npc": n_name}).fetchall()

            if concept_rows:
                names = [r.name for r in concept_rows]
                embeddings = get_embeddings(names, model, provider)

                for i, name in enumerate(names):
                    emb = np.array(embeddings[i])
                    norm_p = np.linalg.norm(query_embedding) * np.linalg.norm(emb)
                    if norm_p > 0:
                        sim = float(np.dot(query_embedding, emb) / norm_p)
                        if sim >= similarity_threshold:
                            results.append({'content': name, 'type': 'concept', 'score': sim})

    results.sort(key=lambda x: -x['score'])
    return results[:max_results]

kg_evolve_incremental(existing_kg, new_content_text=None, new_facts=None, model=None, provider=None, npc=None, context='', get_concepts=False, link_concepts_facts=False, link_concepts_concepts=False, link_facts_facts=False, embedding_model=None, embedding_provider=None)

Source code in npcpy/memory/knowledge_graph.py
def kg_evolve_incremental(existing_kg,
                          new_content_text=None,
                          new_facts=None,
                          model=None,
                          provider=None,
                          npc=None,
                          context='',
                          get_concepts=False,
                          link_concepts_facts=False,
                          link_concepts_concepts=False,
                          link_facts_facts=False,
                          embedding_model=None,
                          embedding_provider=None):

    current_gen = existing_kg.get('generation', 0)
    next_gen = current_gen + 1

    newly_added_concepts = []
    concept_links = list(existing_kg.get('concept_links', []))
    fact_to_concept_links = defaultdict(list,
                                        existing_kg.get('fact_to_concept_links', {}))
    fact_to_fact_links = list(existing_kg.get('fact_to_fact_links', []))

    existing_facts = existing_kg.get('facts', [])
    existing_concepts = existing_kg.get('concepts', [])
    existing_concept_names = {c['name'] for c in existing_concepts}
    existing_fact_statements = [f['statement'] for f in existing_facts]
    all_concept_names = list(existing_concept_names)

    all_new_facts = []

    if new_facts:
        all_new_facts = new_facts
        logger.info(f'Using pre-approved facts: {len(all_new_facts)}')
    elif new_content_text:
        logger.info('Extracting facts from content...')
        if len(new_content_text) > 10000:
            for n in range(len(new_content_text) // 10000):
                content_to_sample = new_content_text[n * 10000:(n + 1) * 10000]
                facts = get_facts(content_to_sample,
                                model=model,
                                provider=provider,
                                npc=npc,
                                context=context)
                all_new_facts.extend(facts)
        else:
            all_new_facts = get_facts(new_content_text,
                                model=model,
                                provider=provider,
                                npc=npc,
                                context=context)
    else:
        logger.info("No new content or facts provided")
        return existing_kg, {}

    existing_stmts = {f['statement'] for f in existing_facts}
    for fact in all_new_facts:
        fact['generation'] = next_gen

    final_facts = existing_facts + [f for f in all_new_facts if f['statement'] not in existing_stmts]

    if get_concepts:
        logger.info('Generating groups...')
        candidate_concepts = generate_groups(all_new_facts,
                                            model=model,
                                            provider=provider,
                                            npc=npc,
                                            context=context)
        for cand_concept in candidate_concepts:
            cand_name = cand_concept['name']
            if cand_name in existing_concept_names:
                continue
            cand_concept['generation'] = next_gen
            newly_added_concepts.append(cand_concept)
            if link_concepts_concepts:
                related_concepts = get_related_concepts_multi(cand_name,
                                                            "concept",
                                                            all_concept_names,
                                                            model,
                                                            provider,
                                                            npc,
                                                            context)
                for related_name in related_concepts:
                    if related_name != cand_name:
                        concept_links.append((cand_name, related_name))
            all_concept_names.append(cand_name)

        final_concepts = existing_concepts + newly_added_concepts

        if link_concepts_facts:
            for fact in all_new_facts:
                fact_to_concept_links[fact['statement']] = get_related_concepts_multi(
                    fact['statement'], "fact", all_concept_names,
                    model=model, provider=provider, npc=npc, context=context)
    else:
        final_concepts = existing_concepts

    if link_facts_facts and existing_fact_statements:
        e_model = embedding_model or 'nomic-embed-text'
        e_provider = embedding_provider or 'ollama'

        for new_fact in all_new_facts:
            # Pre-filter existing facts by embedding similarity
            candidates = _get_similar_by_embedding(
                new_fact['statement'], existing_fact_statements,
                e_model, e_provider, top_k=20)
            if candidates:
                related_fact_stmts = get_related_facts_llm(new_fact['statement'],
                                                           candidates,
                                                           model=model,
                                                           provider=provider,
                                                           npc=npc,
                                                           context=context)
                for related_stmt in related_fact_stmts:
                    fact_to_fact_links.append((new_fact['statement'], related_stmt))

    final_kg = {
        "generation": next_gen,
        "facts": final_facts,
        "concepts": final_concepts,
        "concept_links": concept_links,
        "fact_to_concept_links": dict(fact_to_concept_links),
        "fact_to_fact_links": fact_to_fact_links
    }
    return final_kg, {}

kg_evolve_knowledge(engine, content_text, npc=None, team=None, model=None, provider=None)

Evolve the knowledge graph with new content

Source code in npcpy/memory/knowledge_graph.py
def kg_evolve_knowledge(
   engine,
   content_text: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Evolve the knowledge graph with new content"""
   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)

   evolved_kg, _ = kg_evolve_incremental(
       existing_kg=kg_data,
       new_content_text=content_text,
       model=npc.model if npc else model,
       provider=npc.provider if npc else provider,
       npc=npc,
       get_concepts=True,
       link_concepts_facts=False,
       link_concepts_concepts=False,
       link_facts_facts=False
   )

   save_kg_to_db(engine, evolved_kg, team_name, npc_name, directory_path)

   return "Knowledge graph evolved with new content"

kg_explore_concept(engine, concept_name, max_depth=2, breadth_per_step=10, search_all_scopes=True)

Explore all facts and related concepts for a given concept.

Source code in npcpy/memory/knowledge_graph.py
def kg_explore_concept(
    engine,
    concept_name: str,
    max_depth: int = 2,
    breadth_per_step: int = 10,
    search_all_scopes: bool = True
):
    """Explore all facts and related concepts for a given concept."""
    from sqlalchemy import text

    result = {
        'concept': concept_name,
        'direct_facts': [],
        'related_concepts': [],
        'extended_facts': []
    }

    with engine.connect() as conn:
        rows = conn.execute(text("""
            SELECT source FROM kg_links
            WHERE target = :concept AND type = 'fact_to_concept'
        """), {"concept": concept_name})
        result['direct_facts'] = [r.source for r in rows]

        rows = conn.execute(text("""
            SELECT target FROM kg_links
            WHERE source = :concept AND type = 'concept_to_concept'
            UNION
            SELECT source FROM kg_links
            WHERE target = :concept AND type = 'concept_to_concept'
        """), {"concept": concept_name})
        result['related_concepts'] = [r[0] for r in rows]

        if result['related_concepts'] and max_depth > 0:
            placeholders = ','.join([f':c{i}' for i in range(len(result['related_concepts']))])
            params = {f'c{i}': c for i, c in enumerate(result['related_concepts'])}

            rows = conn.execute(text(f"""
                SELECT DISTINCT source FROM kg_links
                WHERE target IN ({placeholders}) AND type = 'fact_to_concept'
            """), params)
            result['extended_facts'] = [r.source for r in rows
                                        if r.source not in result['direct_facts']]

    return result

kg_get_all_facts(engine, npc=None, team=None, model=None, provider=None, search_all_scopes=True)

Get all facts from the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_get_all_facts(
   engine,
   npc=None,
   team=None,
   model=None,
   provider=None,
   search_all_scopes=True
):
   """Get all facts from the knowledge graph"""
   from sqlalchemy import text

   directory_path = os.getcwd()
   team_name = getattr(team, 'name', None) if team else None
   npc_name = getattr(npc, 'name', None) if npc else None

   if search_all_scopes and (not team_name or not npc_name):
       with engine.connect() as conn:
           result = conn.execute(text("SELECT DISTINCT statement FROM kg_facts"))
           return [row.statement for row in result]
   else:
       if not team_name:
           team_name = 'global_team'
       if not npc_name:
           npc_name = 'default_npc'
       kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)
       return [f['statement'] for f in kg_data.get('facts', [])]

kg_get_facts_for_concept(engine, concept_name, npc=None, team=None, model=None, provider=None)

Get all facts linked to a specific concept

Source code in npcpy/memory/knowledge_graph.py
def kg_get_facts_for_concept(
   engine,
   concept_name: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Get all facts linked to a specific concept"""
   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)

   fact_to_concept_links = kg_data.get('fact_to_concept_links', {})
   linked_facts = []

   for fact_statement, linked_concepts in fact_to_concept_links.items():
       if concept_name in linked_concepts:
           linked_facts.append(fact_statement)

   return linked_facts

kg_get_stats(engine, npc=None, team=None, model=None, provider=None)

Get statistics about the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_get_stats(
   engine,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Get statistics about the knowledge graph"""
   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)

   return {
       "total_facts": len(kg_data.get('facts', [])),
       "total_concepts": len(kg_data.get('concepts', [])),
       "total_fact_concept_links": len(kg_data.get('fact_to_concept_links', {})),
       "generation": kg_data.get('generation', 0)
   }

Hybrid search combining multiple methods.

Source code in npcpy/memory/knowledge_graph.py
def kg_hybrid_search(
    engine,
    query: str,
    npc=None,
    team=None,
    mode: str = 'keyword+link',
    max_depth: int = 2,
    breadth_per_step: int = 5,
    max_results: int = 20,
    embedding_model: str = None,
    embedding_provider: str = None,
    similarity_threshold: float = 0.6,
    search_all_scopes: bool = True
):
    """Hybrid search combining multiple methods."""
    all_results = {}

    if 'keyword' in mode or mode == 'link' or mode == 'all':
        keyword_facts = kg_search_facts(engine, query, npc=npc, team=team,
                                        search_all_scopes=search_all_scopes)
        for f in keyword_facts:
            all_results[f] = {'content': f, 'type': 'fact', 'score': 0.7, 'source': 'keyword'}

    if 'embedding' in mode or mode == 'all':
        try:
            emb_results = kg_embedding_search(
                engine, query, npc=npc, team=team,
                embedding_model=embedding_model,
                embedding_provider=embedding_provider,
                similarity_threshold=similarity_threshold,
                max_results=max_results,
                search_all_scopes=search_all_scopes
            )
            for r in emb_results:
                if r['content'] in all_results:
                    all_results[r['content']]['score'] = max(
                        all_results[r['content']]['score'], r['score']
                    ) * 1.1
                    all_results[r['content']]['source'] += '+embedding'
                else:
                    r['source'] = 'embedding'
                    all_results[r['content']] = r
        except Exception as e:
            logger.warning(f"Embedding search failed: {e}")

    if 'link' in mode or mode == 'all':
        link_results = kg_link_search(
            engine, query, npc=npc, team=team,
            max_depth=max_depth,
            breadth_per_step=breadth_per_step,
            max_results=max_results,
            search_all_scopes=search_all_scopes
        )
        for r in link_results:
            if r['content'] in all_results:
                all_results[r['content']]['score'] = max(
                    all_results[r['content']]['score'], r['score']
                ) * 1.05
                all_results[r['content']]['source'] += '+link'
                all_results[r['content']]['depth'] = r.get('depth', 0)
                all_results[r['content']]['path'] = r.get('path', [])
            else:
                r['source'] = 'link'
                all_results[r['content']] = r

    final = sorted(all_results.values(), key=lambda x: -x['score'])
    return final[:max_results]

kg_initial(content, model=None, provider=None, npc=None, context='', facts=None, generation=None, verbose=True, embedding_model=None, embedding_provider=None, zoom_in_enabled=True)

Source code in npcpy/memory/knowledge_graph.py
def kg_initial(content,
               model=None,
               provider=None,
               npc=None,
               context='',
               facts=None,
               generation=None,
               verbose=True,
               embedding_model=None,
               embedding_provider=None,
               zoom_in_enabled=True):

    if generation is None:
        CURRENT_GENERATION = 0
    else:
        CURRENT_GENERATION = generation

    logger.info(f"Running KG Structuring Process (Generation: {CURRENT_GENERATION})")

    if facts is None:
        if not content:
            raise ValueError("kg_initial requires either content_text or a list of facts.")
        logger.info("Mode: Deriving new facts from text content...")
        all_facts = []
        if len(content) > 10000:
            for n in range(len(content) // 10000):
                content_to_sample = content[n * 10000:(n + 1) * 10000]
                extracted = get_facts(content_to_sample,
                                model=model,
                                provider=provider,
                                npc=npc,
                                context=context)
                if verbose:
                    logger.debug(f"Extracted {len(extracted)} facts from segment {n+1}")
                all_facts.extend(extracted)
        else:
            all_facts = get_facts(content,
                                  model=model,
                                  provider=provider,
                                  npc=npc,
                                  context=context)
            if verbose:
                logger.debug(f"Extracted {len(all_facts)} facts from content")
        for fact in all_facts:
            fact['generation'] = CURRENT_GENERATION
    else:
        logger.info(f"Mode: Building structure from {len(facts)} pre-existing facts...")
        all_facts = list(facts)

    all_implied_facts = []
    if zoom_in_enabled:
        logger.info("Inferring implied facts (zooming in)...")
        if len(all_facts) > 20:
            sampled_facts = random.sample(all_facts, k=20)
            for n in range(len(all_facts) // 20):
                implied_facts = zoom_in(sampled_facts,
                                        model=model,
                                        provider=provider,
                                        npc=npc,
                                        context=context)
                all_implied_facts.extend(implied_facts)
                if verbose:
                    logger.debug(f"Inferred {len(implied_facts)} implied facts from sample {n+1}")
        else:
            implied_facts = zoom_in(all_facts,
                                    model=model,
                                    provider=provider,
                                    npc=npc,
                                    context=context)
            all_implied_facts.extend(implied_facts)
            if verbose:
                logger.debug(f"Inferred {len(implied_facts)} implied facts from all facts")

        for fact in all_implied_facts:
            fact['generation'] = CURRENT_GENERATION
    else:
        logger.info("Skipping zoom-in (zoom_in_enabled=False)")

    all_facts = all_facts + all_implied_facts

    logger.info("Generating concepts from all facts...")
    concepts = generate_groups(all_facts,
                               model=model,
                               provider=provider,
                               npc=npc,
                               context=context)
    for concept in concepts:
        concept['generation'] = CURRENT_GENERATION

    if verbose:
        logger.debug(f"Generated {len(concepts)} concepts")

    logger.info("Linking facts to concepts...")
    fact_to_concept_links = defaultdict(list)
    concept_names = [c['name'] for c in concepts if c and 'name' in c]
    for fact in all_facts:
        fact_to_concept_links[fact['statement']] = get_related_concepts_multi(
            fact['statement'], "fact", concept_names, model, provider, npc, context)

    logger.info("Linking facts to other facts...")
    fact_to_fact_links = []
    fact_statements = [f['statement'] for f in all_facts]

    e_model = embedding_model or 'nomic-embed-text'
    e_provider = embedding_provider or 'ollama'

    for i, fact in enumerate(all_facts):
        other_fact_statements = [s for s in fact_statements if s != fact['statement']]
        if not other_fact_statements:
            continue
        try:
            # Pre-filter by embedding similarity to avoid passing too many to LLM
            candidates = _get_similar_by_embedding(
                fact['statement'], other_fact_statements, e_model, e_provider, top_k=20)
            if candidates:
                related_fact_stmts = get_related_facts_llm(fact['statement'],
                                                           candidates,
                                                           model=model,
                                                           provider=provider,
                                                           npc=npc,
                                                           context=context)
                for related_stmt in related_fact_stmts:
                    fact_to_fact_links.append((fact['statement'], related_stmt))
        except Exception as e:
            logger.warning(f"Failed to link fact {i+1}/{len(all_facts)}: {e}")
            continue

    return {
        "generation": CURRENT_GENERATION,
        "facts": all_facts,
        "concepts": concepts,
        "concept_links": [],
        "fact_to_concept_links": dict(fact_to_concept_links),
        "fact_to_fact_links": fact_to_fact_links
    }

Link a fact to a concept in the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_link_fact_to_concept(
   engine,
   fact_text: str,
   concept_name: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Link a fact to a concept in the knowledge graph"""
   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)

   fact_to_concept_links = kg_data.get('fact_to_concept_links', {})

   if fact_text not in fact_to_concept_links:
       fact_to_concept_links[fact_text] = []

   if concept_name not in fact_to_concept_links[fact_text]:
       fact_to_concept_links[fact_text].append(concept_name)
       kg_data['fact_to_concept_links'] = fact_to_concept_links
       save_kg_to_db(engine, kg_data, team_name, npc_name, directory_path)
       return f"Linked fact '{fact_text}' to concept '{concept_name}'"

   return "Fact already linked to concept"

Search KG by traversing links from keyword-matched seeds.

Source code in npcpy/memory/knowledge_graph.py
def kg_link_search(
    engine,
    query: str,
    npc=None,
    team=None,
    max_depth: int = 2,
    breadth_per_step: int = 5,
    max_results: int = 20,
    strategy: str = 'bfs',
    search_all_scopes: bool = True
):
    """Search KG by traversing links from keyword-matched seeds."""
    from sqlalchemy import text
    from collections import deque

    seeds = kg_search_facts(engine, query, npc=npc, team=team,
                           search_all_scopes=search_all_scopes)

    if not seeds:
        return []

    visited = set(seeds[:breadth_per_step])
    results = [{'content': s, 'type': 'fact', 'depth': 0, 'path': [s], 'score': 1.0}
               for s in seeds[:breadth_per_step]]

    if strategy == 'bfs':
        queue = deque()
        for seed in seeds[:breadth_per_step]:
            queue.append((seed, 'fact', 0, [seed], 1.0))
    else:
        queue = []
        for seed in seeds[:breadth_per_step]:
            queue.append((seed, 'fact', 0, [seed], 1.0))

    with engine.connect() as conn:
        while queue and len(results) < max_results:
            if strategy == 'bfs':
                current, curr_type, depth, path, score = queue.popleft()
            else:
                current, curr_type, depth, path, score = queue.pop()

            if depth >= max_depth:
                continue

            linked = []

            result = conn.execute(text("""
                SELECT target, type FROM kg_links WHERE source = :src
            """), {"src": current})
            for row in result:
                target_type = 'concept' if 'concept' in row.type else 'fact'
                linked.append((row.target, target_type, row.type))

            result = conn.execute(text("""
                SELECT source, type FROM kg_links WHERE target = :tgt
            """), {"tgt": current})
            for row in result:
                source_type = 'fact' if 'fact_to' in row.type else 'concept'
                linked.append((row.source, source_type, f"rev_{row.type}"))

            added = 0
            for item_content, item_type, link_type in linked:
                if item_content in visited or added >= breadth_per_step:
                    continue

                visited.add(item_content)
                new_path = path + [item_content]
                new_score = score * 0.8

                results.append({
                    'content': item_content,
                    'type': item_type,
                    'depth': depth + 1,
                    'path': new_path,
                    'score': new_score,
                    'link_type': link_type
                })

                queue.append((item_content, item_type, depth + 1, new_path, new_score))
                added += 1

    results.sort(key=lambda x: (-x['score'], x['depth']))
    return results[:max_results]

kg_list_concepts(engine, npc=None, team=None, model=None, provider=None, search_all_scopes=True)

List all concepts in the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_list_concepts(
   engine,
   npc=None,
   team=None,
   model=None,
   provider=None,
   search_all_scopes=True
):
   """List all concepts in the knowledge graph"""
   from sqlalchemy import text

   directory_path = os.getcwd()
   team_name = getattr(team, 'name', None) if team else None
   npc_name = getattr(npc, 'name', None) if npc else None

   if search_all_scopes and (not team_name or not npc_name):
       with engine.connect() as conn:
           result = conn.execute(text("SELECT DISTINCT name FROM kg_concepts"))
           return [row.name for row in result]
   else:
       if not team_name:
           team_name = 'global_team'
       if not npc_name:
           npc_name = 'default_npc'
       kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)
       return [c['name'] for c in kg_data.get('concepts', [])]

kg_remove_concept(engine, concept_name, npc=None, team=None, model=None, provider=None)

Remove a concept from the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_remove_concept(
   engine,
   concept_name: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Remove a concept from the knowledge graph"""
   from sqlalchemy import text as sa_text

   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   with engine.begin() as conn:
       result = conn.execute(sa_text("""
           DELETE FROM kg_concepts
           WHERE name = :name AND team_name = :team_name
           AND npc_name = :npc_name AND directory_path = :directory_path
       """), {
           "name": concept_name,
           "team_name": team_name,
           "npc_name": npc_name,
           "directory_path": directory_path
       })
       removed_count = result.rowcount

   if removed_count > 0:
       # Also remove any links referencing this concept
       with engine.begin() as conn:
           conn.execute(sa_text("""
               DELETE FROM kg_links
               WHERE (source = :concept OR target = :concept)
               AND team_name = :team_name AND npc_name = :npc_name
               AND directory_path = :directory_path
           """), {
               "concept": concept_name,
               "team_name": team_name,
               "npc_name": npc_name,
               "directory_path": directory_path
           })
       return f"Removed concept: {concept_name}"

   return "Concept not found"

kg_remove_fact(engine, fact_text, npc=None, team=None, model=None, provider=None)

Remove a fact from the knowledge graph

Source code in npcpy/memory/knowledge_graph.py
def kg_remove_fact(
   engine,
   fact_text: str,
   npc=None,
   team=None,
   model=None,
   provider=None
):
   """Remove a fact from the knowledge graph"""
   from sqlalchemy import text as sa_text

   directory_path = os.getcwd()
   team_name = getattr(team, 'name', 'default_team') if team else 'default_team'
   npc_name = npc.name if npc else 'default_npc'

   with engine.begin() as conn:
       result = conn.execute(sa_text("""
           DELETE FROM kg_facts
           WHERE statement = :statement AND team_name = :team_name
           AND npc_name = :npc_name AND directory_path = :directory_path
       """), {
           "statement": fact_text,
           "team_name": team_name,
           "npc_name": npc_name,
           "directory_path": directory_path
       })
       removed_count = result.rowcount

   if removed_count > 0:
       # Also remove any links referencing this fact
       with engine.begin() as conn:
           conn.execute(sa_text("""
               DELETE FROM kg_links
               WHERE (source = :fact OR target = :fact)
               AND team_name = :team_name AND npc_name = :npc_name
               AND directory_path = :directory_path
           """), {
               "fact": fact_text,
               "team_name": team_name,
               "npc_name": npc_name,
               "directory_path": directory_path
           })
       return f"Removed {removed_count} matching fact(s)"

   return "No matching facts found"

kg_search_facts(engine, query, npc=None, team=None, model=None, provider=None, search_all_scopes=True)

Search facts in the knowledge graph by keyword.

Source code in npcpy/memory/knowledge_graph.py
def kg_search_facts(
   engine,
   query: str,
   npc=None,
   team=None,
   model=None,
   provider=None,
   search_all_scopes=True
):
   """Search facts in the knowledge graph by keyword."""
   from sqlalchemy import text

   directory_path = os.getcwd()
   team_name = getattr(team, 'name', None) if team else None
   npc_name = getattr(npc, 'name', None) if npc else None

   matching_facts = []

   if search_all_scopes and (not team_name or not npc_name):
       with engine.connect() as conn:
           result = conn.execute(text("""
               SELECT DISTINCT statement FROM kg_facts
               WHERE LOWER(statement) LIKE LOWER(:query)
           """), {"query": f"%{query}%"})
           matching_facts = [row.statement for row in result]
   else:
       if not team_name:
           team_name = 'global_team'
       if not npc_name:
           npc_name = 'default_npc'
       kg_data = load_kg_from_db(engine, team_name, npc_name, directory_path)
       for fact in kg_data.get('facts', []):
           if query.lower() in fact['statement'].lower():
               matching_facts.append(fact['statement'])

   return matching_facts

kg_sleep_process(existing_kg, model=None, provider=None, npc=None, context='', operations_config=None, embedding_model=None, embedding_provider=None)

Source code in npcpy/memory/knowledge_graph.py
def kg_sleep_process(existing_kg,
                     model=None,
                     provider=None,
                     npc=None,
                     context='',
                     operations_config=None,
                     embedding_model=None,
                     embedding_provider=None):
    current_gen = existing_kg.get('generation', 0)
    next_gen = current_gen + 1
    logger.info(f"SLEEPING (Evolving Knowledge): Gen {current_gen} -> Gen {next_gen}")

    facts_map = {f['statement']: f for f in existing_kg.get('facts', [])}
    concepts_map = {c['name']: c for c in existing_kg.get('concepts', [])}
    fact_links = defaultdict(list, {k: list(v) for k, v in existing_kg.get('fact_to_concept_links', {}).items()})
    concept_links = set(tuple(sorted(link)) for link in existing_kg.get('concept_links', []))
    fact_to_fact_links = set(tuple(sorted(link)) for link in existing_kg.get('fact_to_fact_links', []))

    # Phase 1: Check for unstructured facts
    logger.info("Phase 1: Checking for unstructured facts...")
    facts_with_concepts = set(fact_links.keys())
    orphaned_fact_statements = list(set(facts_map.keys()) - facts_with_concepts)

    if len(orphaned_fact_statements) > 20:
        logger.info(f"Found {len(orphaned_fact_statements)} orphaned facts. Applying full KG structuring process...")
        orphaned_facts_as_dicts = [facts_map[s] for s in orphaned_fact_statements]

        new_structure = kg_initial(
            content=None,
            facts=orphaned_facts_as_dicts,
            model=model,
            provider=provider,
            npc=npc,
            context=context,
            generation=next_gen,
            embedding_model=embedding_model,
            embedding_provider=embedding_provider
        )

        logger.info("Merging new structure into main KG...")
        for concept in new_structure.get("concepts", []):
            if concept['name'] not in concepts_map:
                concepts_map[concept['name']] = concept

        for fact_stmt, new_links in new_structure.get("fact_to_concept_links", {}).items():
            existing_links = set(fact_links.get(fact_stmt, []))
            existing_links.update(new_links)
            fact_links[fact_stmt] = list(existing_links)

        for f1, f2 in new_structure.get("fact_to_fact_links", []):
            fact_to_fact_links.add(tuple(sorted((f1, f2))))
    else:
        logger.info("Knowledge graph is sufficiently structured. Proceeding to refinement.")

    # Phase 2: Refinement operations
    if operations_config is None:
        possible_ops = ['prune', 'deepen']
        ops_to_run = random.sample(possible_ops, k=random.randint(1, 2))
    else:
        ops_to_run = operations_config

    logger.info(f"Phase 2: Executing refinement operations: {ops_to_run}")

    for op in ops_to_run:
        if op == 'prune' and (len(facts_map) > 10 or len(concepts_map) > 5):
            logger.info("Running 'prune' operation...")
            fact_to_check = random.choice(list(facts_map.values()))
            other_facts = [f for f in facts_map.values() if f['statement'] != fact_to_check['statement']]
            consolidation_result = consolidate_facts_llm(fact_to_check, other_facts, model, provider, npc, context)
            if consolidation_result.get('decision') == 'redundant':
                logger.info(f"Pruning redundant fact: '{fact_to_check['statement'][:80]}...'")
                del facts_map[fact_to_check['statement']]

        elif op == 'deepen' and facts_map:
            logger.info("Running 'deepen' operation...")
            fact_to_deepen = random.choice(list(facts_map.values()))
            implied_facts = zoom_in([fact_to_deepen], model, provider, npc, context)
            new_fact_count = 0
            for fact in implied_facts:
                if fact['statement'] not in facts_map:
                    fact.update({'generation': next_gen, 'origin': 'deepen'})
                    facts_map[fact['statement']] = fact
                    new_fact_count += 1
            if new_fact_count > 0:
                logger.info(f"Inferred {new_fact_count} new fact(s).")

        else:
            logger.debug(f"SKIPPED: Operation '{op}' did not run (conditions not met).")

    new_kg = {
        "generation": next_gen,
        "facts": list(facts_map.values()),
        "concepts": list(concepts_map.values()),
        "concept_links": [list(link) for link in concept_links],
        "fact_to_concept_links": dict(fact_links),
        "fact_to_fact_links": [list(link) for link in fact_to_fact_links]
    }
    return new_kg, {}

prune_fact_subset_llm(fact_subset, concept_name, model=None, provider=None, npc=None, context=None, **kwargs)

Identifies redundancies WITHIN a small, topically related subset of facts.

Source code in npcpy/llm_funcs.py
def prune_fact_subset_llm(fact_subset, 
                          concept_name, 
                          model=None,
                          provider=None,
                          npc=None,
                          context : str = None,
                          **kwargs):
    """Identifies redundancies WITHIN a small, topically related subset of facts."""
    print(f"  Step Sleep-A: Pruning fact subset for concept '{concept_name}'...")


    prompt = f"""
    The following facts are all related to the concept "{concept_name}".
    Review ONLY this subset and identify groups of facts that are semantically identical.
    Return only the set of facts that are semantically distinct, and archive the rest.

    Fact Subset: {json.dumps(fact_subset, indent=2)}

    Return a json list of groups
    """ + '{"refined_facts": [fact1, fact2, fact3, ...]}'
    response = get_llm_response(prompt, 
                                model=model, 
                                provider=provider, 
                                npc=None,
                                format="json", 
                                context=context)
    return response['response'].get('refined_facts', [])

remove_idempotent_groups(group_candidates, model=None, provider=None, npc=None, context=None, **kwargs)

Remove groups that are essentially identical in meaning, favoring specificity and direct naming, and avoiding generic structures.

Source code in npcpy/llm_funcs.py
def remove_idempotent_groups(
    group_candidates: List[str],
    model: str = None,
    provider: str =None,
    npc = None, 
    context : str = None,
    **kwargs: Any
) -> List[str]:
    """Remove groups that are essentially identical in meaning, favoring specificity and direct naming, and avoiding generic structures."""

    prompt = f"""Compare these group names. Identify and list ONLY the groups that are conceptually distinct and specific.

    GUIDELINES FOR SELECTING DISTINCT GROUPS:
    1.  **Prioritize Specificity and Direct Naming:** Favor precise nouns or noun phrases that directly name the subject.
    2.  **Prefer Concrete Entities/Actions:** If a name refers to a specific entity or action (e.g., "Earth", "Sun", "Water", "France", "User Authentication Module", "Lamb Shank Braising", "World War I"), keep it if it's distinct.
    3.  **Rephrase Gerunds:** If a name uses a gerund (e.g., "Understanding TDEs"), rephrase it to a noun or noun phrase (e.g., "Tidal Disruption Events").
    4.  **AVOID OVERLY GENERIC TERMS:** Do NOT use very broad or abstract terms that don't add specific meaning. Examples to avoid: "Concepts", "Processes", "Dynamics", "Mechanics", "Analysis", "Understanding", "Interactions", "Relationships", "Properties", "Structures", "Systems", "Frameworks", "Predictions", "Outcomes", "Effects", "Considerations", "Methods", "Techniques", "Data", "Theoretical", "Physical", "Spatial", "Temporal". If a group name seems overly generic or abstract, it should likely be removed or refined.
    5.  **Similarity Check:** If two groups are very similar, keep the one that is more descriptive or specific to the domain.

    EXAMPLE 1:
    Groups: ["Accretion Disk Formation", "Accretion Disk Dynamics", "Formation of Accretion Disks"]
    Distinct Groups: ["Accretion Disk Formation", "Accretion Disk Dynamics"] 

    EXAMPLE 2:
    Groups: ["Causes of Events", "Event Mechanisms", "Event Drivers"]
    Distinct Groups: ["Event Causation", "Event Mechanisms"] 

    EXAMPLE 3:
    Groups: ["Astrophysics Basics", "Fundamental Physics", "General Science Concepts"]
    Distinct Groups: ["Fundamental Physics"] 

    EXAMPLE 4:
    Groups: ["Earth", "The Planet Earth", "Sun", "Our Star"]
    Distinct Groups: ["Earth", "Sun"]

    EXAMPLE 5:
    Groups: ["User Authentication Module", "Authentication System", "Login Process"]
    Distinct Groups: ["User Authentication Module", "Login Process"]

    ---

    Now, analyze the following groups:
    Groups: {json.dumps(group_candidates)}

    Return JSON:
    """ + '{"distinct_groups": ["list of specific, precise, and distinct group names to keep"]}'

    response = get_llm_response(
        prompt,
        model=model,
        provider=provider,
        format="json",
        npc=npc,
        context=context,
        **kwargs
    )

    return response["response"]["distinct_groups"]

save_changelog_to_json(changelog, from_gen, to_gen, path_prefix='changelog')

Source code in npcpy/memory/knowledge_graph.py
def save_changelog_to_json(changelog, from_gen, to_gen, path_prefix="changelog"):
    if not changelog:
        return
    with open(f"{path_prefix}_gen{from_gen}_to_{to_gen}.json", 'w', encoding='utf-8') as f:
        json.dump(changelog, f, indent=4)
    logger.info(f"Saved changelog for Gen {from_gen}->{to_gen}.")

save_kg_with_pandas(kg, path_prefix='kg_state')

Source code in npcpy/memory/knowledge_graph.py
def save_kg_with_pandas(kg, path_prefix="kg_state"):
    generation = kg.get("generation", 0)

    nodes_data = []
    for fact in kg.get('facts', []):
        nodes_data.append({'id': fact['statement'], 'type': 'fact', 'generation': fact.get('generation')})
    for concept in kg.get('concepts', []):
        nodes_data.append({'id': concept['name'], 'type': 'concept', 'generation': concept.get('generation')})
    pd.DataFrame(nodes_data).to_csv(f'{path_prefix}_gen{generation}_nodes.csv', index=False)

    links_data = []
    for fact_stmt, concepts in kg.get("fact_to_concept_links", {}).items():
        for concept_name in concepts:
            links_data.append({'source': fact_stmt, 'target': concept_name, 'type': 'fact_to_concept'})
    for c1, c2 in kg.get("concept_links", []):
        links_data.append({'source': c1, 'target': c2, 'type': 'concept_to_concept'})
    for f1, f2 in kg.get("fact_to_fact_links", []):
        links_data.append({'source': f1, 'target': f2, 'type': 'fact_to_fact'})
    pd.DataFrame(links_data).to_csv(f'{path_prefix}_gen{generation}_links.csv', index=False)
    logger.info(f"Saved KG Generation {generation} to CSV files.")

store_fact_with_embedding(collection, fact, metadata, embedding)

Store a fact with its pre-generated embedding in Chroma DB.

Source code in npcpy/memory/knowledge_graph.py
def store_fact_with_embedding(
    collection, fact: str, metadata: dict, embedding: List[float]
) -> str:
    """Store a fact with its pre-generated embedding in Chroma DB."""
    try:
        fact_id = hashlib.md5(fact.encode()).hexdigest()

        collection.add(
            documents=[fact],
            embeddings=[embedding],
            metadatas=[metadata],
            ids=[fact_id],
        )

        return fact_id
    except Exception as e:
        logger.warning(f"Error storing fact in Chroma: {e}")
        return None

zoom_in(facts, model=None, provider=None, npc=None, context=None, attempt_number=1, n_attempts=3, **kwargs)

Infer new implied facts from existing facts

Source code in npcpy/llm_funcs.py
def zoom_in(facts, 
            model= None,
            provider=None, 
            npc=None,
            context: str = None, 
            attempt_number: int = 1,
            n_attempts=3,            
            **kwargs):
    """Infer new implied facts from existing facts"""
    valid_facts = []
    for fact in facts:
        if isinstance(fact, dict) and 'statement' in fact:
            valid_facts.append(fact)
    if not valid_facts:
        return []     

    fact_lines = []
    for fact in valid_facts:
        fact_lines.append(f"- {fact['statement']}")
    facts_text = "\n".join(fact_lines)

    prompt = f"""
    Look at these facts and infer new implied facts:

    {facts_text}

    What other facts can be reasonably inferred from these?
    """ +"""
    Respond with JSON:
    {
        "implied_facts": [
            {
                "statement": "new implied fact",
                "inferred_from": ["which facts this comes from"]
            }
        ]
    }
    """

    response = get_llm_response(prompt, 
                                model=model, 
                                provider=provider, 
                                format="json", 
                                context=context,
                                npc=npc,
                                **kwargs)

    facts =  response.get("response", {}).get("implied_facts", [])
    if len(facts) == 0:
        return zoom_in(valid_facts, 
                       model=model, 
                       provider=provider, 
                       npc=npc,
                       context=context,
                       attempt_number=attempt_number+1,
                       n_tries=n_attempts,
                       **kwargs)
    return facts

execute_rag_command(command, vector_db_path, embedding_model, embedding_provider, top_k=15, file_contents=None, **kwargs)

Execute the RAG command with support for embedding generation. When file_contents is provided, it searches those instead of the database.

Source code in npcpy/memory/search.py
def execute_rag_command(
    command: str,
    vector_db_path: str, 
    embedding_model: str,
    embedding_provider: str,
    top_k: int = 15,
    file_contents=None,  
    **kwargs
) -> dict:
    """
    Execute the RAG command with support for embedding generation.
    When file_contents is provided, it searches those instead of the database.
    """

    BLUE = "\033[94m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    CYAN = "\033[96m"
    RESET = "\033[0m"
    BOLD = "\033[1m"


    header = f"\n{BOLD}{BLUE}RAG Query: {RESET}{GREEN}{command}{RESET}\n"


    if file_contents and len(file_contents) > 0:
        similar_chunks = search_similar_texts(
            command,
            embedding_model,
            embedding_provider,
            chroma_client=None,  

            docs_to_embed=file_contents,  
            top_k=top_k
        )


        file_info = f"{BOLD}{BLUE}Files Processed: {RESET}{YELLOW}{len(file_contents)}{RESET}\n"
        separator = f"{YELLOW}{'-' * 100}{RESET}\n"


        chunk_results = []
        for i, chunk in enumerate(similar_chunks, 1):
            score = chunk['score']
            text = chunk['text']


            display_text = text[:150] + ("..." if len(text) > 150 else "")
            chunk_results.append(f"{BOLD}{i:2d}{RESET}. {CYAN}[{score:.2f}]{RESET} {display_text}")


        file_results = header + file_info + separator + "\n".join(chunk_results)
        render_markdown(f"FILE SEARCH RESULTS:\n{file_results}")


        plain_chunks = [f"{i+1}. {chunk['text']}" for i, chunk in enumerate(similar_chunks)]
        plain_results = "\n\n".join(plain_chunks)


        prompt = f"""
        The user asked: {command}

        Here are the most relevant sections from the file(s):

        {plain_results}

        Please respond to the user query based on the above information, integrating the information in an additive way, attempting to always find some possible connection
        between the results and the initial input. do not do this haphazardly, be creative yet cautious.
        """


        response = get_llm_response(
            prompt,
            **kwargs
        )
        return response
    else:
        return {"output": "RAG without file_contents requires a vector store backend. Provide file_contents or use the application-layer RAG command.", "messages": kwargs.get('messages', [])}

execute_search_command(command, messages=None, provider=None)

Function Description:

Parameters:
  • command –

    str : Command

  • db_path –

    str : Database path

  • embedding_model –

    None : Embedding model

  • current_npc –

    None : Current NPC

  • text_data –

    None : Text data

  • text_data_embedded –

    None : Embedded text data

  • messages –

    None : Messages

Returns: dict : dict : Dictionary

Source code in npcpy/memory/search.py
def execute_search_command(
    command: str,
    messages=None,
    provider: str = None,
):
    """
    Function Description:

    Args:
        command : str : Command
        db_path : str : Database path

    Keyword Args:
        embedding_model : None : Embedding model
        current_npc : None : Current NPC
        text_data : None : Text data
        text_data_embedded : None : Embedded text data
        messages : None : Messages
    Returns:
        dict : dict : Dictionary

    """

    search_command = command.split()
    if any("-p" in s for s in search_command) or any(
        "--provider" in s for s in search_command
    ):
        provider = (
            search_command[search_command.index("-p") + 1]
            if "-p" in search_command
            else search_command[search_command.index("--provider") + 1]
        )
    else:
        provider = None
    if any("-n" in s for s in search_command) or any(
        "--num_results" in s for s in search_command
    ):
        num_results = (
            search_command[search_command.index("-n") + 1]
            if "-n" in search_command
            else search_command[search_command.index("--num_results") + 1]
        )
    else:
        num_results = 5


    command = command.replace(f"-p {provider}", "").replace(
        f"--provider {provider}", ""
    )
    result = search_web(command, num_results=num_results, provider=provider)
    if messages is None:
        messages = []
        messages.append({"role": "user", "content": command})

    messages.append(
        {"role": "assistant", "content": result[0] + f" \n Citation Links: {result[1]}"}
    )

    return {
        "messages": messages,
        "output": result[0] + f"\n\n\n Citation Links: {result[1]}",
    }

get_llm_response(prompt, model=None, provider=None, images=None, npc=None, team=None, messages=None, api_url=None, api_key=None, context=None, stream=False, attachments=None, include_usage=False, n_samples=1, matrix=None, **kwargs)

Generate a response using the specified provider and model.

Source code in npcpy/llm_funcs.py
def get_llm_response(
    prompt: str,
    model: str = None,
    provider: str = None,
    images: List[str] = None,
    npc: Any = None,
    team: Any = None,
    messages: List[Dict[str, str]] = None,
    api_url: str = None,
    api_key: str = None,
    context=None,
    stream: bool = False,
    attachments: List[str] = None,
    include_usage: bool = False,
    n_samples: int = 1,
    matrix: Optional[Dict[str, List[Any]]] = None,
    **kwargs,
):
    """Generate a response using the specified provider and model."""
    logger.debug(
        f"[get_llm_response] {len(messages) if messages else 0} messages, "
        f"prompt_len={len(prompt) if prompt else 0}, "
        f"context={'yes' if context else 'no'}"
    )

    base_model, base_provider, base_api_url, base_api_key = resolve_model_provider(
        npc=npc, team=team, model=model, provider=provider,
        api_url=api_url, api_key=api_key, images=images, attachments=attachments,
    )
    if api_key is None:
        api_key = base_api_key

    use_matrix = matrix is not None and len(matrix) > 0
    multi_sample = n_samples and n_samples > 1

    if not use_matrix and not multi_sample:
        # Simple single-call path
        run_model, run_provider, run_api_url, run_api_key = resolve_model_provider(
            npc=npc, team=team, model=base_model, provider=base_provider,
            api_url=api_url, api_key=api_key, images=images, attachments=attachments,
        )
        tool_capable = bool(kwargs.get("tools"))
        system_message = get_system_message(npc, team, tool_capable=tool_capable) if npc is not None else "You are a helpful assistant."

        # Build messages
        run_messages = copy.deepcopy(messages) if messages else []
        if not run_messages:
            run_messages = [{"role": "system", "content": system_message}]

        # Build full text from prompt and/or context
        full_text = ""
        if prompt and context:
            full_text = f"{prompt}\n\n\nUser Provided Context: {context}"
        elif prompt:
            full_text = prompt
        elif context:
            full_text = f"User Provided Context: {context}"

        if full_text:
            if run_messages[-1]["role"] == "user" and isinstance(run_messages[-1]["content"], str):
                run_messages[-1]["content"] += "\n" + full_text
            else:
                run_messages.append({"role": "user", "content": full_text})

        if not run_model:
            raise ValueError("No model specified. Please set a model in your NPC configuration or team settings.")
        return get_litellm_response(
            full_text or None,
            messages=run_messages,
            model=run_model,
            provider=run_provider,
            api_url=run_api_url,
            api_key=api_key,
            images=images,
            attachments=attachments,
            stream=stream,
            include_usage=include_usage,
            **kwargs,
        )

    # Matrix / multi-sample path
    combos = []
    if use_matrix:
        keys = list(matrix.keys())
        values = []
        for key in keys:
            val = matrix[key]
            if isinstance(val, (list, tuple, set)):
                values.append(list(val))
            else:
                values.append([val])
        for combo_values in itertools.product(*values):
            combos.append(dict(zip(keys, combo_values)))
    else:
        combos.append({})

    runs = []
    for combo in combos:
        run_npc = combo.get("npc", npc)
        run_team = combo.get("team", team)
        run_context = combo.get("context", context)
        extra_kwargs = dict(kwargs)
        for k, v in combo.items():
            if k not in {"model", "provider", "npc", "context", "team"}:
                extra_kwargs[k] = v

        run_model, run_provider, run_api_url, run_api_key = resolve_model_provider(
            npc=run_npc, team=run_team,
            model=combo.get("model", base_model),
            provider=combo.get("provider", base_provider),
            api_url=api_url, api_key=api_key, images=images, attachments=attachments,
        )

        tool_capable = bool(extra_kwargs.get("tools"))
        system_message = get_system_message(run_npc, run_team, tool_capable=tool_capable) if run_npc is not None else "You are a helpful assistant."

        run_messages = copy.deepcopy(messages) if messages else []
        if not run_messages:
            run_messages = [{"role": "system", "content": system_message}]

        full_text = ""
        if prompt and run_context:
            full_text = f"{prompt}\n\n\nUser Provided Context: {run_context}"
        elif prompt:
            full_text = prompt
        elif run_context:
            full_text = f"User Provided Context: {run_context}"

        if full_text:
            if run_messages[-1]["role"] == "user" and isinstance(run_messages[-1]["content"], str):
                run_messages[-1]["content"] += "\n" + full_text
            else:
                run_messages.append({"role": "user", "content": full_text})

        for sample_idx in range(max(1, n_samples or 1)):
            resp = get_litellm_response(
                full_text or None,
                messages=run_messages,
                model=run_model,
                provider=run_provider,
                api_url=run_api_url,
                api_key=api_key,
                images=images,
                attachments=attachments,
                stream=stream,
                include_usage=include_usage,
                **extra_kwargs,
            )
            runs.append({
                "response": resp.get("response") if isinstance(resp, dict) else resp,
                "raw": resp,
                "combo": combo,
                "sample_index": sample_idx,
            })

    aggregated = {
        "response": runs[0]["response"] if runs else None,
        "runs": runs,
    }
    if runs and isinstance(runs[0]["raw"], dict) and "messages" in runs[0]["raw"]:
        aggregated["messages"] = runs[0]["raw"].get("messages")
    return aggregated

get_ollama_embeddings(texts, model='nomic-embed-text')

Generate embeddings using Ollama.

Source code in npcpy/gen/embeddings.py
def get_ollama_embeddings(
    texts: List[str], model: str = "nomic-embed-text"
) -> List[List[float]]:
    """Generate embeddings using Ollama."""
    import ollama

    embeddings = []
    for text in texts:
        response = ollama.embeddings(model=model, prompt=text)
        embeddings.append(response["embedding"])
    return embeddings

load_file_contents(file_path, chunk_size=None)

Source code in npcpy/data/load.py
def load_file_contents(file_path, chunk_size=None):
    file_ext = os.path.splitext(file_path)[1].upper().lstrip('.')
    full_content = ""
    if not isinstance(chunk_size, int):
        chunk_size=8000
    try:
        if file_ext == 'PDF':
            full_content = load_pdf(file_path)
        elif file_ext == 'DOCX':
            full_content = load_docx(file_path)
        elif file_ext == 'PPTX':
            full_content = load_pptx(file_path)
        elif file_ext in ['HTML', 'HTM']:
            full_content = load_html(file_path)
        elif file_ext == 'CSV':
            df = load_csv(file_path)
            full_content = df.to_string()
        elif file_ext in ['XLS', 'XLSX']:
            df = load_excel(file_path)
            full_content = df.to_string()
        elif file_ext in ['TXT', 'MD', 'PY', 'JSX', 'TSX', 'TS', 'JS', 'JSON', 'SQL', 'NPC', 'JINX', 'LINE', 'YAML', 'DART', 'JAVA']:
            full_content = load_txt(file_path)
        elif file_ext == 'JSON':
            data = load_json(file_path)
            full_content = json.dumps(data, indent=2)
        elif file_ext in ['MP3', 'WAV', 'M4A', 'AAC', 'FLAC', 'OGG']:
            full_content = load_audio(file_path)
        elif file_ext in ['MP4', 'AVI', 'MOV', 'WMV', 'MPG', 'MPEG', 'WEBM', 'MKV']:
            full_content = load_video(file_path)
        else:
            return [f"Unsupported file format for content loading: {file_ext}"]

        if not full_content:
            return []

        return _chunk_text(full_content, chunk_size)

    except Exception as e:
        return [f"Error loading file {file_path}: {str(e)}"]

render_markdown(text)

Renders markdown text, but handles code blocks as plain syntax-highlighted text.

Source code in npcpy/npc_sysenv.py
def render_markdown(text: str) -> None:
    """
    Renders markdown text, but handles code blocks as plain syntax-highlighted text.
    """
    lines = text.split("\n")
    console = Console()

    inside_code_block = False
    code_lines = []
    prose_lines = []
    lang = None

    def _flush_prose():
        if not prose_lines:
            return
        import re
        # Normalize CR/CRLF and collapse 3+ consecutive blank lines to 2.
        block = "\n".join(ln.rstrip('\r') for ln in prose_lines)
        block = re.sub(r'\n{3,}', '\n\n', block)

        _box_re = re.compile(r'[─-╿]')  # U+2500–U+257F box drawing block

        def _is_structured(ln: str) -> bool:
            """Box-drawing art or 4-space-indented code: must be rendered verbatim."""
            s = ln.strip()
            return bool(s) and (bool(_box_re.search(ln)) or ln.startswith('    '))

        lines = block.split('\n')
        segments: list = []   # (is_structured: bool, lines: list[str])
        struct_acc: list = []
        prose_acc: list = []

        def _commit_struct():
            if struct_acc:
                segments.append((True, list(struct_acc)))
                struct_acc.clear()

        def _commit_prose():
            if prose_acc:
                acc = list(prose_acc)
                while acc and not acc[0].strip():
                    acc.pop(0)
                while acc and not acc[-1].strip():
                    acc.pop()
                if acc:
                    segments.append((False, acc))
                prose_acc.clear()

        in_struct = False
        for i, ln in enumerate(lines):
            if _is_structured(ln):
                if not in_struct:
                    _commit_prose()
                    in_struct = True
                struct_acc.append(ln)
            elif not ln.strip():
                if in_struct:
                    # Suppress blank if the next non-blank line is also structured.
                    j = i + 1
                    while j < len(lines) and not lines[j].strip():
                        j += 1
                    if j < len(lines) and _is_structured(lines[j]):
                        pass  # drop blank between consecutive structured lines
                    else:
                        _commit_struct()
                        in_struct = False
                        prose_acc.append(ln)
                else:
                    prose_acc.append(ln)
            else:
                if in_struct:
                    _commit_struct()
                    in_struct = False
                prose_acc.append(ln)

        if in_struct:
            _commit_struct()
        else:
            _commit_prose()

        for is_str, seg_lines in segments:
            seg = '\n'.join(seg_lines)
            if not seg.strip():
                continue
            if is_str:
                # Verbatim output preserves box-drawing structure.
                # Rich Markdown would join consecutive lines as soft-breaks.
                sys.stdout.write(seg + '\n')
                sys.stdout.flush()
            else:
                console.print(Markdown(seg))

        prose_lines.clear()

    for line in lines:
        if line.startswith("```"):
            if inside_code_block:
                code = "\n".join(code_lines)
                if code.strip():
                    syntax = Syntax(
                        code, lang or "python", theme="monokai", line_numbers=False
                    )
                    console.print(syntax)
                code_lines = []
            else:
                _flush_prose()
                lang = line[3:].strip() or None
            inside_code_block = not inside_code_block
        elif inside_code_block:
            code_lines.append(line)
        else:
            prose_lines.append(line)

    _flush_prose()

search_similar_texts(query, embedding_model, embedding_provider, chroma_client=None, docs_to_embed=None, top_k=15)

Search for similar texts using either a Chroma database or direct embedding comparison. With duplicate filtering.

Source code in npcpy/memory/search.py
def search_similar_texts(
    query: str,
    embedding_model: str,
    embedding_provider: str,    
    chroma_client = None,

    docs_to_embed: Optional[List[str]] = None,
    top_k: int = 15,
) -> List[Dict[str, any]]:
    """
    Search for similar texts using either a Chroma database or direct embedding comparison.
    With duplicate filtering.
    """

    print(f"\nQuery to embed: {query}")
    embedded_search_term = get_ollama_embeddings([query], embedding_model)[0]

    if docs_to_embed is None:

        collection_name = f"{embedding_provider}_{embedding_model}_embeddings"
        collection = chroma_client.get_collection(collection_name)
        results = collection.query(
            query_embeddings=[embedded_search_term], n_results=top_k * 2  
        )


        seen_texts = set()
        filtered_results = []

        for idx, (id, distance, document) in enumerate(zip(
            results["ids"][0], results["distances"][0], results["documents"][0]
        )):

            if document not in seen_texts:
                seen_texts.add(document)
                filtered_results.append({
                    "id": id, 
                    "score": float(distance), 
                    "text": document
                })


                if len(filtered_results) >= top_k:
                    break

        return filtered_results

    print(f"\nNumber of documents to embed: {len(docs_to_embed)}")


    unique_docs = list(dict.fromkeys(docs_to_embed))  
    raw_embeddings = get_ollama_embeddings(unique_docs, embedding_model)

    output_embeddings = []
    unique_doc_indices = []

    for idx, emb in enumerate(raw_embeddings):
        if emb:  
            output_embeddings.append(emb)
            unique_doc_indices.append(idx)


    doc_embeddings = np.array(output_embeddings)
    query_embedding = np.array(embedded_search_term)


    if len(doc_embeddings) == 0:
        raise ValueError("No valid document embeddings found")


    doc_norms = np.linalg.norm(doc_embeddings, axis=1, keepdims=True)
    query_norm = np.linalg.norm(query_embedding)


    if query_norm == 0:
        raise ValueError("Query embedding is zero-length")


    cosine_similarities = np.dot(doc_embeddings, query_embedding) / (
        doc_norms.flatten() * query_norm
    )


    top_indices = np.argsort(cosine_similarities)[::-1][:top_k]

    return [
        {
            "id": str(unique_doc_indices[idx]),
            "score": float(cosine_similarities[idx]),
            "text": unique_docs[unique_doc_indices[idx]],
        }
        for idx in top_indices
    ]

search_web(query, num_results=5, provider=None, api_key=None, perplexity_kwargs=None)

Function Description

This function searches the web for information based on a query.

Args: query: The search query. Keyword Args: num_results: The number of search results to retrieve. provider: The search engine provider to use ('perplexity' or 'duckduckgo'). Returns: A list of dictionaries with 'title', 'link', and 'content' keys.

Source code in npcpy/data/web.py
def search_web(
    query: str,
    num_results: int = 5,
    provider: str=None,
    api_key=None,
    perplexity_kwargs: Optional[Dict[str, Any]] = None,
) -> List:
    """
    Function Description:
        This function searches the web for information based on a query.
    Args:
        query: The search query.
    Keyword Args:
        num_results: The number of search results to retrieve.
        provider: The search engine provider to use ('perplexity' or 'duckduckgo').
    Returns:
        A list of dictionaries with 'title', 'link', and 'content' keys.
    """
    if perplexity_kwargs is None:
        perplexity_kwargs = {}
    results = []
    if provider is None:
        provider = 'startpage'

    if provider == "perplexity":
        search_result = search_perplexity(query, api_key=api_key, **perplexity_kwargs)

        return search_result

    if provider == "startpage":
        results = search_startpage(query, num_results=num_results)
        if results:
            return results
        print("Startpage failed, falling back to searxng")
        provider = 'searxng'

    if provider == "searxng":
        results = search_searxng(query, num_results=num_results)
        if results:
            return results
        print("SearXNG failed, falling back to duckduckgo")
        provider = 'duckduckgo'

    if provider == "duckduckgo":
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0"
        }
        try:
            ddgs = DDGS(headers=headers)
            search_results = ddgs.text(query, max_results=num_results)
            results = [
                {"title": r["title"], "link": r["href"], "content": r["body"]}
                for r in search_results
            ]
        except Exception as e:
            print("DuckDuckGo search failed: ", e)
            results = []
    elif provider == 'exa':
        return search_exa(query, api_key=api_key, top_k=num_results)

    elif provider == 'startpage':
        results = search_startpage(query, num_results=num_results)

    elif provider == 'brave':
        results = search_brave(query, num_results=num_results, api_key=api_key)

    elif provider == 'google':
        if _google_search is None:
            print("googlesearch-python not installed, falling back to duckduckgo")
            return search_web(query, num_results=num_results, provider='duckduckgo')
        urls = list(_google_search(query, num_results=num_results))



        for url in urls:
            try:

                headers = {
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
                }
                response = requests.get(url, headers=headers, timeout=5)
                response.raise_for_status()


                soup = BeautifulSoup(response.text, "html.parser")


                title = soup.title.string if soup.title else url


                content = " ".join([p.get_text() for p in soup.find_all("p")])
                content = " ".join(content.split())  

                results.append(
                    {
                        "title": title,
                        "link": url,
                        "content": (
                            content[:500] + "..." if len(content) > 500 else content
                        ),
                    }
                )

            except Exception as e:
                print(f"Error fetching {url}: {str(e)}")
                continue



    content_str = "\n".join(
        [r["content"] + "\n Citation: " + r["link"] + "\n\n\n" for r in results]
    )
    link_str = "\n".join([r["link"] + "\n" for r in results])
    return [content_str, link_str]

kg_vis

_create_networkx_graph(kg)

Helper function to convert our KG dict into a NetworkX graph for analysis.

Source code in npcpy/memory/kg_vis.py
def _create_networkx_graph(kg):
    """Helper function to convert our KG dict into a NetworkX graph for analysis."""
    G = nx.Graph()
    concepts = [c['name'] for c in kg.get('concepts', [])]
    facts = [f['statement'] for f in kg.get('facts', [])]

    G.add_nodes_from(concepts, type='concept')
    G.add_nodes_from(facts, type='fact')

    for fact, linked_concepts in kg.get('fact_to_concept_links', {}).items():
        for concept in linked_concepts:
            if G.has_node(fact) and G.has_node(concept):
                G.add_edge(fact, concept)

    for c1, c2 in kg.get('concept_links', []):
        if G.has_node(c1) and G.has_node(c2):
            G.add_edge(c1, c2)

    return G

_create_networkx_graph_full(kg)

helper to build the complete graph including fact-to-fact links.

Source code in npcpy/memory/kg_vis.py
def _create_networkx_graph_full(kg):
    """helper to build the complete graph including fact-to-fact links."""
    G = nx.Graph()
    concepts = [c['name'] for c in kg.get('concepts', [])]
    facts = [f['statement'] for f in kg.get('facts', [])]
    G.add_nodes_from(concepts, type='concept')
    G.add_nodes_from(facts, type='fact')
    for fact, linked_concepts in kg.get('fact_to_concept_links', {}).items():
        for concept in linked_concepts:
            if G.has_node(fact) and G.has_node(concept): G.add_edge(fact, concept)
    for c1, c2 in kg.get('concept_links', []):
        if G.has_node(c1) and G.has_node(c2): G.add_edge(c1, c2)

    for f1, f2 in kg.get('fact_to_fact_links', []):
        if G.has_node(f1) and G.has_node(f2): G.add_edge(f1, f2)
    return G

load_changelog_from_json(from_gen, to_gen, path_prefix='changelog')

Loads the detailed changelog JSON file created during a 'kg_sleep_process'.

Source code in npcpy/memory/kg_vis.py
def load_changelog_from_json(from_gen, to_gen, path_prefix="changelog"):
    """Loads the detailed changelog JSON file created during a 'kg_sleep_process'."""
    filename = f"{path_prefix}_gen{from_gen}_to_{to_gen}.json"
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            changelog = json.load(f)
            print(f"Successfully loaded changelog from {filename}")
            return changelog
    except FileNotFoundError as e:
        print(f"Error: Could not find changelog file: {e}")
        return None

load_kg_with_pandas(generation, path_prefix='kg_state')

Loads the new graph structure from CSV files.

Source code in npcpy/memory/kg_vis.py
def load_kg_with_pandas(generation, path_prefix="kg_state"):
    """Loads the new graph structure from CSV files."""
    kg = {
        "generation": generation, "facts": [], "concepts": [],
        "concept_links": [], "fact_to_concept_links": {}
    }
    try:
        nodes_df = pd.read_csv(f'{path_prefix}_gen{generation}_nodes.csv')
        links_df = pd.read_csv(f'{path_prefix}_gen{generation}_links.csv')
    except FileNotFoundError as e:
        print(f"Error: Could not find data files for generation {generation}. {e}")
        return None

    for _, row in nodes_df.iterrows():
        if row['type'] == 'fact':
            kg['facts'].append({'statement': row['id'], 'generation': int(row['generation'])})
        elif row['type'] == 'concept':
            kg['concepts'].append({'name': row['id'], 'generation': int(row['generation'])})

    fact_links = defaultdict(list)
    concept_links = []
    for _, row in links_df.iterrows():
        if row['type'] == 'fact_to_concept':
            fact_links[row['source']].append(row['target'])
        elif row['type'] == 'concept_to_concept':
            concept_links.append((row['source'], row['target']))

    kg['fact_to_concept_links'] = dict(fact_links)
    kg['concept_links'] = concept_links

    print(f"Successfully loaded KG Generation {generation} with pandas.")
    return kg

visualize_associative_richness(kg_history, filename='associative_richness.png')

Plots the Associative Richness Index (ARI): Avg. Concepts per Fact.

Source code in npcpy/memory/kg_vis.py
def visualize_associative_richness(kg_history, filename="associative_richness.png"):
    """Plots the Associative Richness Index (ARI): Avg. Concepts per Fact."""
    print(f"Generating Associative Richness chart -> {filename}")
    gens = [kg['generation'] for kg in kg_history]
    ari_scores = []
    for kg in kg_history:
        num_facts = len(kg.get('facts', []))
        total_links = sum(len(links) for links in kg.get('fact_to_concept_links', {}).values())
        ari_scores.append(total_links / num_facts if num_facts > 0 else 0)

    plt.figure(figsize=(12, 8))
    plt.plot(gens, ari_scores, marker='o', linestyle='-', color='#6a0dad', linewidth=2.5, label="System's ARI")
    plt.axhline(y=1, color='gray', linestyle='--', linewidth=2, label='1-to-1 Mapping Baseline (ARI=1.0)')
    plt.xlabel("Generation")
    plt.ylabel("Avg. Concepts per Fact (ARI)")
    plt.xticks(gens)
    plt.legend(loc='lower right')
    plt.ylim(bottom=0)
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_centrality_bubble_chart(kg, node_type='concepts', filename='concept_bubble_chart.png')

Creates a 'bubble chart' where nodes are arranged purely by importance (degree centrality), with the most important nodes in the center.

Parameters:
  • kg –

    Knowledge graph data

  • node_type –

    "concepts", "facts", or "both" - which nodes to visualize

  • filename –

    Output filename

Source code in npcpy/memory/kg_vis.py
def visualize_centrality_bubble_chart(kg, node_type="concepts", filename="concept_bubble_chart.png"):
    """
    Creates a 'bubble chart' where nodes are arranged purely by importance
    (degree centrality), with the most important nodes in the center.

    Args:
        kg: Knowledge graph data
        node_type: "concepts", "facts", or "both" - which nodes to visualize
        filename: Output filename
    """
    print(f"Generating CENTRALITY-BASED Bubble Chart for {node_type} in Gen {kg['generation']} -> {filename}")

    Full_G = _create_networkx_graph(kg)
    if not Full_G.nodes:
        print(f"  - KG {kg['generation']} has no nodes. Skipping.")
        return

    all_nodes = {}
    for node, data in Full_G.nodes(data=True):
        if node_type == "concepts" and data['type'] == 'concept':
            all_nodes[node] = Full_G.degree(node)
        elif node_type == "facts" and data['type'] == 'fact':
            all_nodes[node] = Full_G.degree(node)
        elif node_type == "both":
            all_nodes[node] = {'degree': Full_G.degree(node), 'type': data['type']}

    if not all_nodes:
        print(f"  - KG {kg['generation']} has no {node_type}. Skipping.")
        return

    if node_type == "both":
        sorted_nodes = sorted(all_nodes.items(), key=lambda item: item[1]['degree'], reverse=True)
    else:
        sorted_nodes = sorted(all_nodes.items(), key=lambda item: item[1], reverse=True)

    pos = {}
    if sorted_nodes:
        central_node, _ = sorted_nodes[0]
        pos[central_node] = (0, 0)

        radius = 0.25
        nodes_in_ring = 6
        node_idx = 1

        while node_idx < len(sorted_nodes):
            angle_step = 2 * np.pi / nodes_in_ring
            for i in range(nodes_in_ring):
                if node_idx >= len(sorted_nodes): break
                angle = i * angle_step
                node_name, _ = sorted_nodes[node_idx]
                pos[node_name] = (radius * np.cos(angle), radius * np.sin(angle))
                node_idx += 1

            radius += 0.20
            nodes_in_ring = int(nodes_in_ring * 1.5)

    plt.figure(figsize=(20, 20))

    if node_type == "both":
        concept_nodes = [(name, data) for name, data in sorted_nodes if data['type'] == 'concept']
        fact_nodes = [(name, data) for name, data in sorted_nodes if data['type'] == 'fact']

        if concept_nodes:
            concept_names = [item[0] for item in concept_nodes]
            concept_sizes = [item[1]['degree'] * 200 for item in concept_nodes]
            concept_pos = {name: pos[name] for name in concept_names if name in pos}
            nx.draw_networkx_nodes(None, concept_pos, nodelist=concept_names, 
                                  node_color='#ffb480', node_size=concept_sizes, alpha=0.9)

        if fact_nodes:
            fact_names = [item[0] for item in fact_nodes]
            fact_sizes = [item[1]['degree'] * 100 for item in fact_nodes]
            fact_pos = {name: pos[name] for name in fact_names if name in pos}
            nx.draw_networkx_nodes(None, fact_pos, nodelist=fact_names, 
                                  node_color='#ff6961', node_size=fact_sizes, alpha=0.9)

        from matplotlib.patches import Patch
        legend_elements = [
            Patch(facecolor='#ffb480', label='Concepts'),
            Patch(facecolor='#ff6961', label='Facts')
        ]
        plt.legend(handles=legend_elements, loc='upper right', fontsize=14)

    else:
        node_names = [item[0] for item in sorted_nodes]
        node_sizes = [item[1] * 200 for item in sorted_nodes]

        color = '#ffb480' if node_type == "concepts" else '#ff6961'

        nx.draw_networkx_nodes(None, pos, nodelist=node_names, 
                              node_color=color, node_size=node_sizes, alpha=0.9)


    top_nodes = sorted_nodes[:min(20, len(sorted_nodes))]

    for item in top_nodes:
        node_name = item[0]
        if node_type == "both":
            degree = item[1]['degree']
            node_type_actual = item[1]['type']
        else:
            degree = item[1]
            node_type_actual = node_type.rstrip('s')

        if node_name in pos:
            x, y = pos[node_name]
            font_size = max(8, 8 + 2 * np.log1p(degree))

            if node_type_actual == 'fact':
                label = fill(node_name, 10)
            else:
                label = fill(node_name, 15)

            plt.text(x, y, label, ha='center', va='center', 
                    fontsize=font_size, fontfamily='serif')

    plt.axis('off')

    max_coord = radius * 1.1
    plt.xlim(-max_coord, max_coord)
    plt.ylim(-max_coord, max_coord)
    plt.gca().set_aspect('equal', adjustable='box')

    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_concept_bubble_chart(kg, filename='concept_bubble_chart.png')

Creates a 'bubble chart' of the concept ontology, arranged like a word cloud. - The most important concept (highest degree) is fixed at the center. - All other concepts are arranged around it using a force-directed layout. - Node size is proportional to its total degree. - No edges are drawn, for maximum clarity.

Source code in npcpy/memory/kg_vis.py
def visualize_concept_bubble_chart(kg, filename="concept_bubble_chart.png"):
    """
    Creates a 'bubble chart' of the concept ontology, arranged like a word cloud.
    - The most important concept (highest degree) is fixed at the center.
    - All other concepts are arranged around it using a force-directed layout.
    - Node size is proportional to its total degree.
    - No edges are drawn, for maximum clarity.
    """
    print(f"Generating CENTRALIZED Concept Bubble Chart for Gen {kg['generation']} -> {filename}")

    Full_G = _create_networkx_graph(kg)
    if not Full_G.nodes:
        print(f"  - KG {kg['generation']} has no nodes. Skipping.")
        return

    concepts = {node: Full_G.degree(node) for node, data in Full_G.nodes(data=True) if data['type'] == 'concept'}

    if not concepts:
        print(f"  - KG {kg['generation']} has no concepts. Skipping.")
        return

    Concept_G = nx.Graph()
    Concept_G.add_nodes_from(concepts.keys())
    for c1, c2 in kg.get('concept_links', []):
        if Concept_G.has_node(c1) and Concept_G.has_node(c2):
            Concept_G.add_edge(c1, c2)

    central_node = max(concepts, key=concepts.get)
    fixed_nodes = [central_node]
    pos_initial = {central_node: (0, 0)}

    pos = nx.spring_layout(Concept_G, pos=pos_initial, fixed=fixed_nodes, 
                           k=1.8/math.sqrt(Concept_G.number_of_nodes()),
                           iterations=200, seed=42)

    plt.figure(figsize=(20, 20))


    node_sizes = [concepts[node] * 200 for node in Concept_G.nodes()]

    nx.draw_networkx_nodes(Concept_G, pos, node_color='#ffb480', node_size=node_sizes, alpha=0.9)

    for node, (x, y) in pos.items():
        degree = concepts[node]
        font_size = 8 + 2 * math.log(1 + degree)
        plt.text(x, y, fill(node, 15), ha='center', va='center', fontsize=font_size, fontfamily='serif')

    plt.axis('off')
    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_concept_ontology_graph(kg, filename='concept_ontology.png')

Creates a 'bubble map' of the CONCEPT ontology. - Nodes are concepts only. - Edges are only concept-to-concept links. - Node size is proportional to its total degree (including fact links), representing its overall importance.

Source code in npcpy/memory/kg_vis.py
def visualize_concept_ontology_graph(kg, filename="concept_ontology.png"):
    """
    Creates a 'bubble map' of the CONCEPT ontology.
    - Nodes are concepts only.
    - Edges are only concept-to-concept links.
    - Node size is proportional to its total degree (including fact links),
      representing its overall importance.
    """
    print(f"Generating Concept Ontology Bubble Map for Gen {kg['generation']} -> {filename}")

    Full_G = _create_networkx_graph(kg)
    if not Full_G.nodes:
        print(f"  - KG {kg['generation']} has no nodes. Skipping.")
        return

    Concept_G = nx.Graph()
    concept_names = [c['name'] for c in kg.get('concepts', [])]
    Concept_G.add_nodes_from(concept_names)
    for c1, c2 in kg.get('concept_links', []):
        if Concept_G.has_node(c1) and Concept_G.has_node(c2):
            Concept_G.add_edge(c1, c2)

    node_sizes = [500 + (Full_G.degree(n) * 50) for n in Concept_G.nodes()]

    plt.figure(figsize=(24, 24))

    pos = nx.spring_layout(Concept_G, k=1.5/math.sqrt(Concept_G.number_of_nodes()), iterations=100, seed=42)

    nx.draw_networkx_nodes(Concept_G, pos, node_color='#ffb480', node_size=node_sizes)
    nx.draw_networkx_edges(Concept_G, pos, alpha=0.6, width=1.0, edge_color='gray')
    nx.draw_networkx_labels(Concept_G, pos, font_size=14, font_family='serif')

    plt.axis('off')
    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_concept_trajectories(kg_history, n_pillars=2, n_risers=3, filename='concept_trajectories.png')

To ensure pillars and risers are distinct sets, telling a clearer story about the stable backbone vs. major new themes.

Source code in npcpy/memory/kg_vis.py
def visualize_concept_trajectories(kg_history, n_pillars=2, n_risers=3, filename="concept_trajectories.png"):
    """
    To ensure pillars and risers are distinct sets, telling a clearer story
    about the stable backbone vs. major new themes.
    """
    print(f"Generating Disjoint Concept Trajectories chart -> {filename}")
    centrality_df = pd.DataFrame()

    gens = [kg['generation'] for kg in kg_history]
    for i, kg in enumerate(kg_history):
        G = _create_networkx_graph(kg)
        if not G.nodes: continue
        degree_centrality = nx.degree_centrality(G)
        concept_centrality = {node: cent for node, cent in degree_centrality.items() if G.nodes[node].get('type') == 'concept'}
        s = pd.Series(concept_centrality, name=kg['generation'])
        centrality_df = pd.concat([centrality_df, s.to_frame()], axis=1)
    centrality_df = centrality_df.transpose().sort_index()

    pillars = centrality_df.mean().nlargest(n_pillars).index

    riser_candidates = centrality_df.drop(columns=pillars, errors='ignore')
    centrality_diff = riser_candidates.iloc[-1].fillna(0) - riser_candidates.iloc[0].fillna(0)
    risers = centrality_diff.nlargest(n_risers).index

    concepts_to_plot = pillars.union(risers)

    plt.figure(figsize=(12, 8))
    for concept_name in concepts_to_plot:
        trajectory = centrality_df[concept_name]
        style = '--' if concept_name in pillars else '-'
        linewidth = 1.5 if concept_name in pillars else 2.5
        alpha = 0.8 if concept_name in pillars else 1.0
        plt.plot(trajectory.index, trajectory.values, marker='o', linestyle=style, 
                 label=fill(concept_name, 20), linewidth=linewidth, alpha=alpha)
    plt.xlabel("Generation", fontsize=14)
    plt.ylabel("Degree Centrality", fontsize=14)
    plt.xticks(gens)
    plt.legend(title="Concepts", bbox_to_anchor=(1.05, 1), loc='upper left')
    plt.ylim(bottom=0)
    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_conceptual_support(kg_history, filename='conceptual_support.png')

Plots the Conceptual Support Index (CSI): Avg. Facts per Concept.

Source code in npcpy/memory/kg_vis.py
def visualize_conceptual_support(kg_history, filename="conceptual_support.png"):
    """Plots the Conceptual Support Index (CSI): Avg. Facts per Concept."""
    print(f"Generating Conceptual Support chart -> {filename}")
    gens = [kg['generation'] for kg in kg_history]
    csi_scores = []
    for kg in kg_history:
        num_concepts = len(kg.get('concepts', []))
        total_links = sum(len(links) for links in kg.get('fact_to_concept_links', {}).values())
        csi_scores.append(total_links / num_concepts if num_concepts > 0 else 0)

    plt.figure(figsize=(12, 8))
    plt.plot(gens, csi_scores, marker='o', linestyle='-', color='#17becf', linewidth=2.5, label="System's CSI")
    plt.xlabel("Generation")
    plt.ylabel("Avg. Facts per Concept (CSI)")
    plt.xticks(gens)
    plt.legend(loc='lower right')
    plt.ylim(bottom=0)
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_dual_richness_metrics(kg_history, filename='dual_richness_metrics.png')

Creates a two-panel plot showing ARI and CSI, stacked vertically.

Source code in npcpy/memory/kg_vis.py
def visualize_dual_richness_metrics(kg_history, filename="dual_richness_metrics.png"):
    """
    Creates a two-panel plot showing ARI and CSI, stacked vertically.
    """
    print(f"Generating Dual Richness Metrics chart -> {filename}")

    gens = [kg['generation'] for kg in kg_history]
    ari_scores = []
    csi_scores = []

    for kg in kg_history:
        num_facts = len(kg.get('facts', []))
        num_concepts = len(kg.get('concepts', []))

        total_links = sum(len(links) for links in kg.get('fact_to_concept_links', {}).values())

        ari_scores.append(total_links / num_facts if num_facts > 0 else 0)
        csi_scores.append(total_links / num_concepts if num_concepts > 0 else 0)

    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 16), sharex=True)

    ax1.plot(gens, ari_scores, marker='o', linestyle='-', color='#6a0dad', linewidth=2.5, label="System's ARI")
    ax1.axhline(y=1, color='gray', linestyle='--', linewidth=2, label='1-to-1 Mapping Baseline (ARI=1.0)')
    ax1.set_ylabel("Avg. Concepts per Fact (ARI)", fontsize=14)
    ax1.legend(loc='lower right')
    ax1.set_ylim(bottom=0)

    ax2.plot(gens, csi_scores, marker='o', linestyle='-', color='#17becf', linewidth=2.5, label="System's CSI")
    ax2.set_xlabel("Generation", fontsize=14)
    ax2.set_ylabel("Avg. Facts per Concept (CSI)", fontsize=14)
    ax2.legend(loc='lower right')
    ax2.set_ylim(bottom=0)

    plt.xticks(gens)
    fig.tight_layout(pad=2.0)
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_fact_concept_ratio(kg_pairs, filename='fact_concept_ratio.png')

Updated to work with the new KG structure

Source code in npcpy/memory/kg_vis.py
def visualize_fact_concept_ratio(kg_pairs, filename="fact_concept_ratio.png"):
    """Updated to work with the new KG structure"""
    labels, before_ratios, after_ratios = [], [], []
    for kg_before, kg_after in kg_pairs:
        facts_before = len(kg_before.get('facts', []))
        concepts_before = len(kg_before.get('concepts', []))
        before_ratios.append(facts_before / concepts_before if concepts_before > 0 else 0)

        facts_after = len(kg_after.get('facts', []))
        concepts_after = len(kg_after.get('concepts', []))
        after_ratios.append(facts_after / concepts_after if concepts_after > 0 else 0)

        labels.append(f"Gen {kg_before['generation']}→{kg_after['generation']}")

    x = np.arange(len(labels))
    width = 0.35
    fig, ax = plt.subplots(figsize=(12, 8))
    rects1 = ax.bar(x - width/2, before_ratios, width, label='Before Sleep', color='#95a5a6')
    rects2 = ax.bar(x + width/2, after_ratios, width, label='After Sleep', color='#2ecc71')
    ax.set_ylabel("Fact-to-Concept Ratio",)
    ax.set_xticks(x, labels, fontsize=12, rotation=45, ha="right")
    ax.legend(fontsize=20, frameon=False)
    ax.bar_label(rects1, padding=3, fmt='%.2f', fontsize=10)
    ax.bar_label(rects2, padding=3, fmt='%.2f', fontsize=10)
    ax.set_ylim(bottom=0, top=max(max(before_ratios, default=0), max(after_ratios, default=0)) * 1.5)
    fig.tight_layout()
    plt.savefig(filename, dpi=300)
    plt.close()
    print(f"Saved fact-to-concept ratio chart to {filename}")

visualize_growth(k_graphs, filename='growth_chart.png')

Plots Facts and Concepts as separate lines instead of a stacked area. This allows for independent analysis of each component's growth over time.

Source code in npcpy/memory/kg_vis.py
def visualize_growth(k_graphs, filename="growth_chart.png"):
    """
    Plots Facts and Concepts as separate lines instead of a stacked area.
    This allows for independent analysis of each component's growth over time.
    """
    gens = [kg['generation'] for kg in k_graphs]
    facts_counts = [len(kg.get('facts', [])) for kg in k_graphs]
    concepts_counts = [len(kg.get('concepts', [])) for kg in k_graphs]
    total_nodes = [facts + concepts for facts, concepts in zip(facts_counts, concepts_counts)]

    plt.figure(figsize=(12, 8))


    plt.plot(gens, facts_counts, label='Facts', color='#ff6961', marker='o', linestyle='-', linewidth=2)
    plt.plot(gens, concepts_counts, label='Concepts', color='#ffb480', marker='o', linestyle='-', linewidth=2)
    plt.plot(gens, total_nodes, label='Total Nodes', color='#8ecae6', marker='x', linestyle='--', alpha=0.7)

    plt.xlabel("Generation", fontsize=14)
    plt.ylabel("Number of Nodes", fontsize=14)
    plt.legend(loc='upper left', fontsize=20, frameon=False)
    plt.xticks(gens)
    plt.ylim(bottom=0)

    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()
    print(f"Saved independent growth chart to {filename}")

visualize_key_experiences(kg, filename='key_experiences.png')

Visualizes the full network, highlighting the most central "key experience" facts.

Source code in npcpy/memory/kg_vis.py
def visualize_key_experiences(kg, filename="key_experiences.png"):
    """
    Visualizes the full network, highlighting the most central "key experience" facts.
    """
    print(f"Generating Key Experience network graph for Gen {kg['generation']} -> {filename}")
    G = _create_networkx_graph_full(kg)
    if not G.nodes: return

    facts = {n for n, d in G.nodes(data=True) if d['type'] == 'fact'}
    concepts = {n for n, d in G.nodes(data=True) if d['type'] == 'concept'}


    centrality = nx.degree_centrality(G)


    top_facts = sorted(facts, key=lambda n: centrality[n], reverse=True)[:5]

    node_colors = []
    for node in G:
        if node in top_facts:
            node_colors.append('#ff0000')
        elif G.nodes[node]['type'] == 'fact':
            node_colors.append('#ff6961')
        else:
            node_colors.append('#ffb480')

    plt.figure(figsize=(24, 24))
    pos = nx.spring_layout(G, k=1.5/math.sqrt(G.number_of_nodes()), iterations=100, seed=42)

    nx.draw(G, pos, with_labels=False, node_color=node_colors, 
            node_size=[v * 10000 for v in centrality.values()], 
            width=0.5, edge_color='gray', alpha=0.7)

    labels = {n: fill(n, 15) for n in top_facts + list(concepts)}
    nx.draw_networkx_labels(G, pos, labels=labels, font_size=10)

    plt.axis('off')
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_knowledge_graph_final_interactive(kg, filename='knowledge_graph.html')

Updated to work with the new KG structure

Source code in npcpy/memory/kg_vis.py
def visualize_knowledge_graph_final_interactive(kg, filename="knowledge_graph.html"):
    """Updated to work with the new KG structure"""
    print(f"Generating interactive graph for Gen {kg['generation']} -> {filename}")

    facts = kg.get("facts", [])
    concepts = kg.get("concepts", [])
    fact_to_concept_links = kg.get("fact_to_concept_links", {})
    concept_links = kg.get("concept_links", [])

    node_map = {}
    for fact in facts:
        node_map[fact['statement']] = fact
    for concept in concepts:
        node_map[concept['name']] = concept

    fact_radius = 300
    concept_radius = 600

    node_positions = {}

    if facts:
        for i, fact in enumerate(facts):
            angle = (2 * math.pi * i) / len(facts)
            node_id = fact['statement']
            node_positions[node_id] = {
                'x': fact_radius * math.cos(angle), 
                'y': fact_radius * math.sin(angle)
            }

    if concepts:
        for i, concept in enumerate(concepts):
            angle = (2 * math.pi * i) / len(concepts)
            node_id = concept['name']
            node_positions[node_id] = {
                'x': concept_radius * math.cos(angle), 
                'y': concept_radius * math.sin(angle)
            }

    net = Network(height="100vh", width="100%", bgcolor="#222222", font_color="white", directed=True)

    for fact in facts:
        node_id = fact['statement']
        pos = node_positions.get(node_id, {'x': 0, 'y': 0})
        title_text = f"<strong>Fact (Gen: {fact.get('generation', 'N/A')})</strong><br><em>{fill(node_id, 50)}</em>"
        net.add_node(
            node_id, 
            label=fill(node_id, 25), 
            title=title_text, 
            x=pos['x'], 
            y=pos['y'], 
            color='#ff6961',
            physics=False
        )

    for concept in concepts:
        node_id = concept['name']
        pos = node_positions.get(node_id, {'x': 0, 'y': 0})
        title_text = f"<strong>Concept (Gen: {concept.get('generation', 'N/A')})</strong><br><em>{fill(node_id, 50)}</em>"
        net.add_node(
            node_id, 
            label=fill(node_id, 25), 
            title=title_text, 
            x=pos['x'], 
            y=pos['y'], 
            color='#ffb480',
            physics=False
        )

    for fact_statement, concept_names in fact_to_concept_links.items():
        for concept_name in concept_names:
            if fact_statement in node_map and concept_name in node_map:
                net.add_edge(fact_statement, concept_name, color="#8ecae6", width=1)

visualize_lorenz_curve(kg_history, filename='lorenz_curve.png')

Creates a standalone Lorenz curve plot to compare the degree distribution inequality between the first and final generations.

Source code in npcpy/memory/kg_vis.py
def visualize_lorenz_curve(kg_history, filename="lorenz_curve.png"):
    """
    Creates a standalone Lorenz curve plot to compare the degree distribution
    inequality between the first and final generations.
    """
    print(f"Generating Lorenz Curve comparison -> {filename}")

    fig, ax = plt.subplots(figsize=(10, 10))

    first_gen_kg = next((kg for kg in kg_history if kg.get('facts')), None)
    if first_gen_kg:
        G_first = _create_networkx_graph(first_gen_kg)
        degrees_first = np.array(sorted([d for n, d in G_first.degree()]))
        if degrees_first.size > 0:
            cum_degrees_first = np.cumsum(degrees_first)
            ax.plot(np.linspace(0, 1, len(degrees_first)), cum_degrees_first / cum_degrees_first[-1],
                     label=f"Gen {first_gen_kg['generation']} (Start)", color='#1f77b4', linewidth=2)

    last_gen_kg = kg_history[-1]
    G_last = _create_networkx_graph(last_gen_kg)
    degrees_last = np.array(sorted([d for n, d in G_last.degree()]))
    if degrees_last.size > 0:
        cum_degrees_last = np.cumsum(degrees_last)
        ax.plot(np.linspace(0, 1, len(degrees_last)), cum_degrees_last / cum_degrees_last[-1],
                 label=f"Gen {last_gen_kg['generation']} (End)", color='#ff7f0e', linewidth=2)

    ax.plot([0, 1], [0, 1], linestyle='--', color='black', label='Perfect Equality')

    ax.set_xlabel("Cumulative Share of Nodes", fontsize=14)
    ax.set_ylabel("Cumulative Share of Connections", fontsize=14)
    ax.legend(fontsize=12)
    ax.set_aspect('equal', adjustable='box')
    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_sleep_process(kg_before, kg_after, filename='sleep_process.png')

Simple visualization of before/after states

Source code in npcpy/memory/kg_vis.py
def visualize_sleep_process(kg_before, kg_after, filename="sleep_process.png"):
    """Simple visualization of before/after states"""
    print(f"\n--- Visualizing Sleep Process: Gen {kg_before['generation']} -> Gen {kg_after['generation']} ---")

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 8))

    facts_before = len(kg_before.get('facts', []))
    concepts_before = len(kg_before.get('concepts', []))
    ax1.pie([facts_before, concepts_before], labels=['Facts', 'Concepts'], 
            colors=['#ff6961', '#ffb480'], autopct='%1.1f%%')

    facts_after = len(kg_after.get('facts', []))
    concepts_after = len(kg_after.get('concepts', []))
    ax2.pie([facts_after, concepts_after], labels=['Facts', 'Concepts'], 
            colors=['#ff6961', '#ffb480'], autopct='%1.1f%%')

    plt.savefig(filename, bbox_inches='tight', dpi=300)
    plt.close()
    print(f"Saved sleep process visualization to {filename}")

visualize_specialist_concepts(kg_history, num_to_show=8, filename='specialist_concepts.png')

Plots trajectories of interesting 'middling' concepts by finding those with high variance and peak centrality, while excluding the absolute top global hubs.

Source code in npcpy/memory/kg_vis.py
def visualize_specialist_concepts(kg_history, num_to_show=8, filename="specialist_concepts.png"):
    """
    Plots trajectories of interesting 'middling' concepts by finding those with
    high variance and peak centrality, while excluding the absolute top global hubs.
    """
    print(f"Generating Specialist Concept Trajectories chart -> {filename}")
    centrality_df = pd.DataFrame()
    gens = [kg['generation'] for kg in kg_history]

    for kg in kg_history:
        G = _create_networkx_graph(kg)
        concept_centrality = {n: nx.degree_centrality(G)[n] for n, d in G.nodes(data=True) if d['type'] == 'concept'} if G.nodes else {}
        centrality_df = pd.concat([centrality_df, pd.Series(concept_centrality, name=kg['generation'])], axis=1)
    centrality_df = centrality_df.transpose().sort_index()

    top_hubs = centrality_df.mean().nlargest(5).index
    specialist_candidates = centrality_df.drop(columns=top_hubs, errors='ignore')

    notability_scores = specialist_candidates.max() + specialist_candidates.var().fillna(0)
    concepts_to_plot = notability_scores.nlargest(num_to_show).index

    plt.figure(figsize=(12, 8))
    for name in concepts_to_plot:
        trajectory = centrality_df[name]
        plt.plot(trajectory.index, trajectory.values, marker='o', linestyle='-', label=fill(name, 25))

    plt.xlabel("Generation")
    plt.ylabel("Degree Centrality")

    plt.xticks(gens)

    plt.legend(title="Specialist Concepts",  loc=0, fontsize=17, frameon=False)
    plt.ylim(bottom=0)
    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()

visualize_static_network(kg, top_n_concepts=25, top_n_facts=50, filename='static_network.png')

Creates a clean, ordered bipartite graph showing ONLY the most central concepts and facts, preventing visual clutter.

Source code in npcpy/memory/kg_vis.py
def visualize_static_network(kg, top_n_concepts=25, top_n_facts=50, filename="static_network.png"):
    """
    Creates a clean, ordered bipartite graph showing ONLY the most central concepts
    and facts, preventing visual clutter.
    """
    print(f"Generating ordered static network for Gen {kg['generation']} -> {filename}")
    G = _create_networkx_graph(kg)
    if not G.nodes: return

    concepts = {n for n, d in G.nodes(data=True) if d['type'] == 'concept'}
    facts = {n for n, d in G.nodes(data=True) if d['type'] == 'fact'}

    top_concepts = sorted(concepts, key=G.degree, reverse=True)[:top_n_concepts]
    top_facts = sorted(facts, key=G.degree, reverse=True)[:top_n_facts]

    SubG = G.subgraph(top_concepts + top_facts)

    pos = {}
    for i, node in enumerate(top_concepts): pos[node] = (-1, np.linspace(1, 0, len(top_concepts))[i])
    for i, node in enumerate(top_facts): pos[node] = (1, np.linspace(1, 0, len(top_facts))[i])

    plt.figure(figsize=(16, 24))


    nx.draw_networkx_nodes(SubG, pos, nodelist=top_facts, node_color='#ff6961', node_size=150)
    nx.draw_networkx_nodes(SubG, pos, nodelist=top_concepts, node_color='#ffb480', node_size=1200)

    nx.draw_networkx_edges(SubG, pos, alpha=0.25, width=0.6, edge_color='gray')

    concept_labels = {name: fill(name, 20) for name in top_concepts}
    nx.draw_networkx_labels(SubG, pos, labels=concept_labels, font_size=14, font_family='serif', horizontalalignment='right')

    plt.axis('off')
    plt.tight_layout(pad=0)
    plt.savefig(filename, dpi=300, bbox_inches='tight', pad_inches=0.1)
    plt.close()

visualize_top_concept_centrality(kg_history, top_n=5, filename='concept_centrality.png')

Tracks the degree centrality of the top N most important concepts over time. This shows how a thematic backbone emerges and solidifies within the KG.

Source code in npcpy/memory/kg_vis.py
def visualize_top_concept_centrality(kg_history, top_n=5, filename="concept_centrality.png"):
    """
    Tracks the degree centrality of the top N most important concepts over time.
    This shows how a thematic backbone emerges and solidifies within the KG.
    """
    centrality_data = defaultdict(lambda: [np.nan] * len(kg_history))

    for i, kg in enumerate(kg_history):
        G = _create_networkx_graph(kg)
        if not G.nodes: continue

        degree_centrality = nx.degree_centrality(G)

        concept_centrality = {node: cent for node, cent in degree_centrality.items() if G.nodes[node]['type'] == 'concept'}

        for concept_name, centrality in concept_centrality.items():
            centrality_data[concept_name][i] = centrality

    sorted_concepts = sorted(centrality_data.keys(), key=lambda c: np.nanmax(centrality_data[c]), reverse=True)
    top_concepts = sorted_concepts[:top_n]

    plt.figure(figsize=(12, 8))
    gens = [kg['generation'] for kg in kg_history]

    for concept_name in top_concepts:
        s = pd.Series(centrality_data[concept_name])
        s_interpolated = s.interpolate(method='linear', limit_direction='forward', axis=0)
        plt.plot(gens, s_interpolated, marker='o', linestyle='-', label=fill(concept_name, 20))

    plt.xlabel("Generation", fontsize=14)
    plt.ylabel("Degree Centrality", fontsize=14)
    plt.xticks(gens)
    plt.legend(title="Top Concepts",  loc=0, frameon=False, fontsize=20)
    plt.ylim(bottom=0)
    plt.tight_layout()
    plt.savefig(filename, dpi=300, bbox_inches='tight')
    plt.close()
    print(f"Saved Top Concept Centrality chart to {filename}")

command_history

TABLE_SCHEMAS = {'command_history': ['timestamp', 'command', 'subcommands', 'output', 'location'], 'conversation_history': ['message_id', 'timestamp', 'conversation_id', 'role', 'content', 'directory_path', 'model', 'provider', 'npc', 'team', 'tool_calls', 'tool_results', 'reasoning_content', 'parent_message_id', 'device_id', 'device_name', 'params', 'input_tokens', 'output_tokens', 'cost'], 'jinx_executions': ['message_id', 'jinx_name', 'input', 'timestamp', 'npc', 'team', 'conversation_id', 'output', 'status', 'error_message', 'duration_ms'], 'npc_executions': ['message_id', 'input', 'timestamp', 'npc', 'team', 'conversation_id', 'model', 'provider'], 'message_attachments': ['message_id', 'attachment_name', 'attachment_type', 'attachment_size', 'upload_timestamp', 'file_path'], 'compiled_npcs': ['name', 'source_path', 'compiled_content', 'compiled_at'], 'memory_lifecycle': ['message_id', 'conversation_id', 'npc', 'team', 'directory_path', 'timestamp', 'initial_memory', 'final_memory', 'status', 'model', 'provider', 'created_at'], 'labels': ['entity_type', 'entity_id', 'label', 'metadata', 'created_at'], 'npc_memories': ['npc_name', 'team_name', 'content', 'status', 'created_at', 'updated_at'], 'knowledge_graphs': ['npc_name', 'team_name', 'kg_data', 'generation', 'created_at', 'updated_at']} module-attribute

_HAS_SQLALCHEMY = True module-attribute

CommandHistory

Source code in npcpy/memory/command_history.py
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
class CommandHistory:
    def __init__(self, db: Union[str, Engine] = "~/npcsh_history.db"):
        if isinstance(db, str):
            self.engine = create_engine_from_path(db)
            self.db_path = db
        elif isinstance(db, Engine):
            self.engine = db
            self.db_path = str(db.url)
        else:
            raise TypeError(f"Unsupported type: {type(db)}")

        self._initialize_schema()
        self._setup_execution_triggers()
        self.backfill_execution_tables()
    def backfill_execution_tables(self):
        with self.engine.begin() as conn:
            conn.execute(text("""
                INSERT OR IGNORE INTO jinx_executions 
                (message_id, jinx_name, input, timestamp, npc, team, 
                conversation_id)
                SELECT 
                    message_id,
                    SUBSTR(content, 2, 
                        CASE 
                            WHEN INSTR(SUBSTR(content, 2), ' ') > 0 
                            THEN INSTR(SUBSTR(content, 2), ' ') - 1
                            ELSE LENGTH(content) - 1
                        END
                    ),
                    content,
                    timestamp,
                    npc,
                    team,
                    conversation_id
                FROM conversation_history
                WHERE role = 'user' AND content LIKE '/%'
            """))

            conn.execute(text("""
                INSERT OR IGNORE INTO npc_executions 
                (message_id, input, timestamp, npc, team, conversation_id, 
                model, provider)
                SELECT 
                    message_id,
                    content,
                    timestamp,
                    npc,
                    team,
                    conversation_id,
                    model,
                    provider
                FROM conversation_history
                WHERE role = 'user' AND npc IS NOT NULL
            """))
    def _initialize_schema(self):
        metadata = MetaData()

        Table('command_history', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('timestamp', String(50)),
            Column('command', Text),
            Column('subcommands', Text),
            Column('output', Text),
            Column('location', Text)
        )

        Table('conversation_history', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('message_id', String(50), unique=True, nullable=False),
            Column('timestamp', String(50)),
            Column('role', String(20)),
            Column('content', Text),
            Column('conversation_id', String(100)),
            Column('directory_path', Text),
            Column('model', String(100)),
            Column('provider', String(100)),
            Column('npc', String(100)),
            Column('team', String(100)),
            Column('reasoning_content', Text),
            Column('tool_calls', Text),
            Column('tool_results', Text),
            Column('parent_message_id', String(50)),
            Column('device_id', String(255)),
            Column('device_name', String(255)),
            Column('params', Text),
            Column('input_tokens', Integer),
            Column('output_tokens', Integer),
            Column('cost', String(50))
        )

        Table('message_attachments', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('message_id', String(50), 
                   ForeignKey('conversation_history.message_id', 
                              ondelete='CASCADE'), 
                   nullable=False),
            Column('attachment_name', String(255)),
            Column('attachment_type', String(100)),
            Column('attachment_data', LargeBinary),
            Column('attachment_size', Integer),
            Column('upload_timestamp', String(50)),
            Column('file_path', Text)
        )

        Table('labels', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('entity_type', String(50), nullable=False),
            Column('entity_id', String(100), nullable=False),
            Column('label', String(100), nullable=False),
            Column('metadata', Text),
            Column('created_at', DateTime, default=func.now())
        )

        Table('jinx_executions', metadata,
            Column('message_id', String(50), primary_key=True),
            Column('jinx_name', String(100)),
            Column('input', Text),
            Column('timestamp', String(50)),
            Column('npc', String(100)),
            Column('team', String(100)),
            Column('conversation_id', String(100)),
            Column('output', Text),
            Column('status', String(50)),
            Column('error_message', Text),
            Column('duration_ms', Integer)
        )

        Table('npc_executions', metadata,
            Column('message_id', String(50), primary_key=True),
            Column('input', Text),
            Column('timestamp', String(50)),
            Column('npc', String(100)),
            Column('team', String(100)),
            Column('conversation_id', String(100)),
            Column('model', String(100)),
            Column('provider', String(100))
        )

        Table('memory_lifecycle', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('message_id', String(50), nullable=False),
            Column('conversation_id', String(100), nullable=False),
            Column('npc', String(100), nullable=False),
            Column('team', String(100), nullable=False),
            Column('directory_path', Text, nullable=False),
            Column('timestamp', String(50), nullable=False),
            Column('initial_memory', Text, nullable=False),
            Column('final_memory', Text),
            Column('status', String(50), nullable=False),
            Column('model', String(100)),
            Column('provider', String(100)),
            Column('created_at', DateTime, default=func.now())
        )

        Table('activity_log', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('timestamp', String(50), nullable=False),
            Column('activity_type', String(50), nullable=False),
            Column('activity_data', Text),
            Column('directory_path', Text),
            Column('npc', String(100)),
            Column('device_id', String(255)),
            Column('session_id', String(100)),
        )

        Table('autocomplete_suggestions', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('timestamp', String(50), nullable=False),
            Column('suggestion_type', String(50), nullable=False),
            Column('input_context', Text),
            Column('suggestion', Text, nullable=False),
            Column('accepted', Integer, default=0),
            Column('npc', String(100)),
            Column('model', String(100)),
            Column('provider', String(100)),
            Column('directory_path', Text),
        )

        Table('autocomplete_training', metadata,
            Column('id', Integer, primary_key=True, autoincrement=True),
            Column('created_at', DateTime, default=func.now()),
            Column('suggestion_type', String(50), nullable=False),
            Column('input_text', Text, nullable=False),
            Column('output_text', Text, nullable=False),
            Column('accepted', Integer, nullable=False),
            Column('npc', String(100)),
            Column('model', String(100)),
        )

        metadata.create_all(self.engine, checkfirst=True)
        init_kg_schema(self.engine)

        if 'sqlite' in str(self.engine.url):
            with self.engine.begin() as conn:
                try:
                    conn.execute(text("ALTER TABLE conversation_history ADD COLUMN parent_message_id VARCHAR(50)"))
                except Exception:
                    pass
                try:
                    conn.execute(text("ALTER TABLE conversation_history ADD COLUMN params TEXT"))
                except Exception:
                    pass
                try:
                    conn.execute(text("ALTER TABLE conversation_history ADD COLUMN input_tokens INTEGER"))
                except Exception:
                    pass
                try:
                    conn.execute(text("ALTER TABLE conversation_history ADD COLUMN output_tokens INTEGER"))
                except Exception:
                    pass
                try:
                    conn.execute(text("ALTER TABLE conversation_history ADD COLUMN cost VARCHAR(50)"))
                except Exception:
                    pass
                for col in [
                    "ALTER TABLE jinx_executions ADD COLUMN output TEXT",
                    "ALTER TABLE jinx_executions ADD COLUMN status VARCHAR(50)",
                    "ALTER TABLE jinx_executions ADD COLUMN error_message TEXT",
                    "ALTER TABLE jinx_executions ADD COLUMN duration_ms INTEGER",
                ]:
                    try:
                        conn.execute(text(col))
                    except Exception:
                        pass
                try:
                    conn.execute(text("DROP TABLE IF EXISTS jinx_execution_log"))
                except Exception:
                    pass

    def _setup_execution_triggers(self):
        if 'sqlite' in str(self.engine.url):
            with self.engine.begin() as conn:
                conn.execute(text("""
                    CREATE TRIGGER IF NOT EXISTS populate_jinx_executions
                    AFTER INSERT ON conversation_history
                    WHEN NEW.role = 'user' AND NEW.content LIKE '/%'
                    BEGIN
                        INSERT OR IGNORE INTO jinx_executions 
                        (message_id, jinx_name, input, timestamp, npc, team, 
                         conversation_id)
                        VALUES (
                            NEW.message_id,
                            SUBSTR(NEW.content, 2, 
                                CASE 
                                    WHEN INSTR(SUBSTR(NEW.content, 2), ' ') > 0 
                                    THEN INSTR(SUBSTR(NEW.content, 2), ' ') - 1
                                    ELSE LENGTH(NEW.content) - 1
                                END
                            ),
                            NEW.content,
                            NEW.timestamp,
                            NEW.npc,
                            NEW.team,
                            NEW.conversation_id
                        );
                    END
                """))

                conn.execute(text("""
                    CREATE TRIGGER IF NOT EXISTS populate_npc_executions
                    AFTER INSERT ON conversation_history
                    WHEN NEW.role = 'user' AND NEW.npc IS NOT NULL
                    BEGIN
                        INSERT OR IGNORE INTO npc_executions 
                        (message_id, input, timestamp, npc, team, 
                         conversation_id, model, provider)
                        VALUES (
                            NEW.message_id,
                            NEW.content,
                            NEW.timestamp,
                            NEW.npc,
                            NEW.team,
                            NEW.conversation_id,
                            NEW.model,
                            NEW.provider
                        );
                    END
                """))

    def get_jinx_executions(self, jinx_name: str = None, limit: int = 1000) -> List[Dict]:
        if jinx_name:
            stmt = """
                SELECT je.*, l.label
                FROM jinx_executions je
                LEFT JOIN labels l ON l.entity_type = 'message' 
                    AND l.entity_id = je.message_id
                WHERE je.jinx_name = :jinx_name
                ORDER BY je.timestamp DESC
                LIMIT :limit
            """
            return self._fetch_all(stmt, {"jinx_name": jinx_name, "limit": limit})

        stmt = """
            SELECT je.*, l.label
            FROM jinx_executions je
            LEFT JOIN labels l ON l.entity_type = 'message' 
                AND l.entity_id = je.message_id
            ORDER BY je.timestamp DESC
            LIMIT :limit
        """
        return self._fetch_all(stmt, {"limit": limit})

    def get_npc_executions(self, npc_name: str, limit: int = 1000) -> List[Dict]:
        stmt = """
            SELECT ne.*, l.label
            FROM npc_executions ne
            LEFT JOIN labels l ON l.entity_type = 'message' 
                AND l.entity_id = ne.message_id
            WHERE ne.npc = :npc_name
            ORDER BY ne.timestamp DESC
            LIMIT :limit
        """
        return self._fetch_all(stmt, {"npc_name": npc_name, "limit": limit})

    def label_execution(self, message_id: str, label: str):
        self.add_label('message', message_id, label)

    def add_label(self, entity_type: str, entity_id: str, label: str, metadata: dict = None):
        stmt = """
            INSERT INTO labels (entity_type, entity_id, label, metadata)
            VALUES (:entity_type, :entity_id, :label, :metadata)
        """
        with self.engine.begin() as conn:
            conn.execute(text(stmt), {
                "entity_type": entity_type,
                "entity_id": entity_id,
                "label": label,
                "metadata": json.dumps(metadata) if metadata else None
            })

    def get_labels(self, entity_type: str = None, label: str = None) -> List[Dict]:
        conditions = []
        params = {}

        if entity_type:
            conditions.append("entity_type = :entity_type")
            params["entity_type"] = entity_type
        if label:
            conditions.append("label = :label")
            params["label"] = label

        where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
        stmt = f"SELECT * FROM labels {where} ORDER BY created_at DESC"

        return self._fetch_all(stmt, params)

    def get_training_data_by_label(self, label: str = 'training') -> List[Dict]:
        stmt = """
            SELECT l.entity_type, l.entity_id, l.metadata,
                ch.content, ch.role, ch.npc, ch.conversation_id
            FROM labels l
            LEFT JOIN conversation_history ch ON 
                (l.entity_type = 'message' AND l.entity_id = ch.message_id)
            WHERE l.label = :label
        """
        return self._fetch_all(stmt, {"label": label})
    def _execute_returning_id(self, stmt: str, params: Dict = None) -> Optional[int]:
        """Execute INSERT and return the generated ID"""
        with self.engine.begin() as conn:
            result = conn.execute(text(stmt), params or {})
            return result.lastrowid if hasattr(result, 'lastrowid') else None

    def _fetch_one(self, stmt: str, params: Dict = None) -> Optional[Dict]:
        """Fetch a single row"""
        with self.engine.connect() as conn:
            result = conn.execute(text(stmt), params or {})
            row = result.fetchone()
            return dict(row._mapping) if row else None

    def _fetch_all(self, stmt: str, params: Dict = None) -> List[Dict]:
        """Fetch all rows"""
        with self.engine.connect() as conn:
            result = conn.execute(text(stmt), params or {})
            return [dict(row._mapping) for row in result]

    def add_command(self, command, subcommands, output, location):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        stmt = """
            INSERT INTO command_history (timestamp, command, subcommands, output, location)
            VALUES (:timestamp, :command, :subcommands, :output, :location)
        """
        params = {
            "timestamp": timestamp,
            "command": command,
            "subcommands": str(subcommands),
            "output": str(output),
            "location": location
        }

        with self.engine.begin() as conn:
            conn.execute(text(stmt), params)

    def add_conversation(
        self,
        message_id,
        timestamp,
        role,
        content,
        conversation_id,
        directory_path,
        model=None,
        provider=None,
        npc=None,
        team=None,
        attachments=None,
        reasoning_content=None,
        tool_calls=None,
        tool_results=None,
        parent_message_id=None,
        device_id=None,
        device_name=None,
        gen_params=None,
        input_tokens=None,
        output_tokens=None,
        cost=None,
    ):
        if isinstance(content, (dict, list)):
            content = json.dumps(content, cls=CustomJSONEncoder)

        if tool_calls is not None and not isinstance(tool_calls, str):
            tool_calls = json.dumps(tool_calls, cls=CustomJSONEncoder)
        if tool_results is not None and not isinstance(tool_results, str):
            tool_results = json.dumps(tool_results, cls=CustomJSONEncoder)
        gen_params_json = None
        if gen_params is not None and not isinstance(gen_params, str):
            gen_params_json = json.dumps(gen_params, cls=CustomJSONEncoder)
        elif isinstance(gen_params, str):
            gen_params_json = gen_params

        normalized_directory_path = normalize_path_for_db(directory_path)

        cost_str = str(cost) if cost is not None else None

        stmt = """
            INSERT INTO conversation_history
            (message_id, timestamp, role, content, conversation_id, directory_path, model, provider, npc, team, reasoning_content, tool_calls, tool_results, parent_message_id, device_id, device_name, params, input_tokens, output_tokens, cost)
            VALUES (:message_id, :timestamp, :role, :content, :conversation_id, :directory_path, :model, :provider, :npc, :team, :reasoning_content, :tool_calls, :tool_results, :parent_message_id, :device_id, :device_name, :params, :input_tokens, :output_tokens, :cost)
        """
        params = {
            "message_id": message_id, "timestamp": timestamp, "role": role, "content": content,
            "conversation_id": conversation_id, "directory_path": normalized_directory_path, "model": model,
            "provider": provider, "npc": npc, "team": team, "reasoning_content": reasoning_content,
            "tool_calls": tool_calls, "tool_results": tool_results, "parent_message_id": parent_message_id,
            "device_id": device_id, "device_name": device_name, "params": gen_params_json,
            "input_tokens": input_tokens, "output_tokens": output_tokens, "cost": cost_str
        }
        with self.engine.begin() as conn:
            conn.execute(text(stmt), params)

        if attachments:
            for attachment in attachments:
                self.add_attachment(
                    message_id=message_id,
                    name=attachment.get("name"),
                    attachment_type=attachment.get("type"),
                    data=attachment.get("data"),
                    size=attachment.get("size"),
                    file_path=attachment.get("path")
                )

        return message_id

    def add_memory_to_database(self, message_id: str, conversation_id: str, npc: str, team: str,
                            directory_path: str, initial_memory: str, status: str,
                            model: str = None, provider: str = None, final_memory: str = None):
        """Store a memory entry in the database"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        normalized_directory_path = normalize_path_for_db(directory_path)

        stmt = """
            INSERT INTO memory_lifecycle
            (message_id, conversation_id, npc, team, directory_path, timestamp,
            initial_memory, final_memory, status, model, provider)
            VALUES (:message_id, :conversation_id, :npc, :team, :directory_path,
                    :timestamp, :initial_memory, :final_memory, :status, :model, :provider)
        """

        params = {
            "message_id": message_id, "conversation_id": conversation_id,
            "npc": npc, "team": team, "directory_path": normalized_directory_path,
            "timestamp": timestamp, "initial_memory": initial_memory,
            "final_memory": final_memory, "status": status,
            "model": model, "provider": provider
        }

        return self._execute_returning_id(stmt, params)
    def get_memories_for_scope(
        self,
        npc: str,
        team: str,
        directory_path: str,
        status: Optional[str] = None
    ) -> List[Dict]:

        conditions = []
        params = {}

        if npc:
            conditions.append("npc = :npc")
            params["npc"] = npc
        if team:
            conditions.append("team = :team")
            params["team"] = team
        if directory_path:
            conditions.append("directory_path = :path")
            params["path"] = directory_path
        if status:
            conditions.append("status = :status")
            params["status"] = status

        where = "WHERE " + " AND ".join(conditions) if conditions else ""
        query = f"""
            SELECT id, initial_memory, final_memory,
                status, timestamp, created_at, npc, team, directory_path
            FROM memory_lifecycle
            {where}
            ORDER BY created_at DESC
        """
        return self._fetch_all(query, params)

    def search_memory(self, query: str, npc: str = None, team: str = None, 
                    directory_path: str = None, status_filter: str = None, limit: int = 10):
        """Search memories with hierarchical scope"""
        conditions = ["LOWER(initial_memory) LIKE LOWER(:query) OR LOWER(final_memory) LIKE LOWER(:query)"]
        params = {"query": f"%{query}%"}

        if status_filter:
            conditions.append("status = :status")
            params["status"] = status_filter


        order_parts = []
        if npc:
            order_parts.append(f"CASE WHEN npc = '{npc}' THEN 1 ELSE 2 END")
        if team:
            order_parts.append(f"CASE WHEN team = '{team}' THEN 1 ELSE 2 END")
        if directory_path:
            order_parts.append(f"CASE WHEN directory_path = '{directory_path}' THEN 1 ELSE 2 END")

        order_clause = ", ".join(order_parts) + ", created_at DESC" if order_parts else "created_at DESC"

        stmt = f"""
            SELECT * FROM memory_lifecycle 
            WHERE {' AND '.join(conditions)}
            ORDER BY {order_clause}
            LIMIT :limit
        """
        params["limit"] = limit

        return self._fetch_all(stmt, params)

    def get_memory_examples_for_context(self, npc: str, team: str, directory_path: str,
                                    n_approved: int = 10, n_rejected: int = 10, n_edited: int = 5):
        """Get recent approved, rejected, and edited memories for learning context."""

        scope_order = """
            CASE WHEN npc = :npc AND team = :team AND directory_path = :path THEN 1
                 WHEN npc = :npc AND team = :team THEN 2
                 WHEN team = :team THEN 3
                 ELSE 4 END
        """

        approved_stmt = f"""
            SELECT initial_memory, final_memory, status FROM memory_lifecycle
            WHERE status IN ('human-approved', 'model-approved')
            ORDER BY {scope_order}, created_at DESC
            LIMIT :n_approved
        """

        rejected_stmt = f"""
            SELECT initial_memory, status FROM memory_lifecycle
            WHERE status IN ('human-rejected', 'model-rejected')
            ORDER BY {scope_order}, created_at DESC
            LIMIT :n_rejected
        """

        edited_stmt = f"""
            SELECT initial_memory, final_memory, status FROM memory_lifecycle
            WHERE status = 'human-edited' AND final_memory IS NOT NULL
            ORDER BY {scope_order}, created_at DESC
            LIMIT :n_edited
        """

        params = {"npc": npc, "team": team, "path": directory_path,
                "n_approved": n_approved, "n_rejected": n_rejected, "n_edited": n_edited}

        approved = self._fetch_all(approved_stmt, params)
        rejected = self._fetch_all(rejected_stmt, params)
        edited = self._fetch_all(edited_stmt, params)

        return {"approved": approved, "rejected": rejected, "edited": edited}

    def get_pending_memories(self, limit: int = 50):
        """Get memories pending human approval"""
        stmt = """
            SELECT * FROM memory_lifecycle 
            WHERE status = 'pending_approval'
            ORDER BY created_at ASC
            LIMIT :limit
        """
        return self._fetch_all(stmt, {"limit": limit})

    def update_memory_status(self, memory_id: int, new_status: str, final_memory: str = None):
        """Update memory status and optionally final_memory"""
        stmt = """
            UPDATE memory_lifecycle 
            SET status = :status, final_memory = :final_memory
            WHERE id = :memory_id
        """
        params = {"status": new_status, "final_memory": final_memory, "memory_id": memory_id}

        with self.engine.begin() as conn:
            conn.execute(text(stmt), params)

    def get_approved_memories_by_scope(self):
        """Get all approved/edited memories grouped by (npc, team, path) scope."""
        stmt = """
            SELECT id, npc, team, directory_path, initial_memory, final_memory, status
            FROM memory_lifecycle
            WHERE status IN ('human-approved', 'human-edited')
            ORDER BY npc, team, directory_path, created_at
        """
        rows = self._fetch_all(stmt, {})

        from collections import defaultdict
        memories_by_scope = defaultdict(list)
        for row in rows:
            statement = row.get('final_memory') or row.get('initial_memory')
            scope_key = (
                row.get('npc') or 'default',
                row.get('team') or 'global_team',
                row.get('directory_path') or os.getcwd()
            )
            memories_by_scope[scope_key].append({
                'id': row.get('id'),
                'statement': statement,
                'source_text': '',
                'type': 'explicit',
                'generation': 0
            })
        return dict(memories_by_scope)

    def add_attachment(self, message_id, name, attachment_type, data, size, file_path=None):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        stmt = """
            INSERT INTO message_attachments 
            (message_id, attachment_name, attachment_type, attachment_data, attachment_size, upload_timestamp, file_path)
            VALUES (:message_id, :name, :type, :data, :size, :timestamp, :file_path)
        """
        params = {
            "message_id": message_id,
            "name": name,
            "type": attachment_type,
            "data": data,
            "size": size,
            "timestamp": timestamp,
            "file_path": file_path
        }
        with self.engine.begin() as conn:
            conn.execute(text(stmt), params)

    def save_jinx_execution(
        self,
        triggering_message_id: str,
        conversation_id: str,
        npc_name: Optional[str],
        jinx_name: str,
        jinx_inputs: Dict,
        jinx_output: Any,
        status: str,
        team_name: Optional[str] = None,
        error_message: Optional[str] = None,
        response_message_id: Optional[str] = None,
        duration_ms: Optional[int] = None
    ):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        try:
            inputs_json = json.dumps(jinx_inputs, cls=CustomJSONEncoder)
        except TypeError:
            inputs_json = json.dumps(str(jinx_inputs))

        try:
            if isinstance(jinx_output, (str, int, float, bool, list, dict, type(None))):
                outputs_json = json.dumps(jinx_output, cls=CustomJSONEncoder)
            else:
                outputs_json = json.dumps(str(jinx_output))
        except TypeError:
            outputs_json = json.dumps(f"Non-serializable output: {type(jinx_output)}")

        msg_id = triggering_message_id or f"jinx-{jinx_name}-{timestamp.replace(' ', '-')}"

        stmt = """
            INSERT OR REPLACE INTO jinx_executions
            (message_id, jinx_name, input, timestamp, npc, team,
             conversation_id, output, status, error_message, duration_ms)
            VALUES (:message_id, :jinx_name, :input, :timestamp, :npc, :team,
                    :conversation_id, :output, :status, :error_message, :duration_ms)
        """
        params = {
            "message_id": msg_id,
            "jinx_name": jinx_name,
            "input": inputs_json,
            "timestamp": timestamp,
            "npc": npc_name,
            "team": team_name,
            "conversation_id": conversation_id,
            "output": outputs_json,
            "status": status,
            "error_message": error_message,
            "duration_ms": duration_ms,
        }

        with self.engine.begin() as conn:
            conn.execute(text(stmt), params)

    def get_full_message_content(self, message_id):
        stmt = "SELECT content FROM conversation_history WHERE message_id = :message_id ORDER BY timestamp ASC"
        rows = self._fetch_all(stmt, {"message_id": message_id})
        return "".join(row['content'] for row in rows)

    def update_message_content(self, message_id, full_content):
        stmt = "UPDATE conversation_history SET content = :content WHERE message_id = :message_id"
        with self.engine.begin() as conn:
            conn.execute(text(stmt), {"content": full_content, "message_id": message_id})

    def get_message_attachments(self, message_id) -> List[Dict]:
        stmt = """
            SELECT 
                id, 
                message_id, 
                attachment_name, 
                attachment_type, 
                attachment_size, 
                upload_timestamp
            FROM message_attachments WHERE message_id = :message_id
        """
        return self._fetch_all(stmt, {"message_id": message_id})
    def delete_message(self, conversation_id, message_id):
        """Delete a specific message from a conversation"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        try:
            cursor.execute("""
                DELETE FROM messages 
                WHERE conversation_id = ? AND message_id = ?
            """, (conversation_id, message_id))

            rows_affected = cursor.rowcount
            conn.commit()

            print(f"[DB] Deleted message {message_id} from conversation {conversation_id}. Rows affected: {rows_affected}")

            return rows_affected

        except Exception as e:
            print(f"[DB] Error deleting message: {e}")
            conn.rollback()
            raise
        finally:
            conn.close()

    def get_attachment_data(self, attachment_id) -> Optional[Tuple[bytes, str, str]]:
        stmt = "SELECT attachment_data, attachment_name, attachment_type FROM message_attachments WHERE id = :attachment_id"
        row = self._fetch_one(stmt, {"attachment_id": attachment_id})
        if row:
            return row['attachment_data'], row['attachment_name'], row['attachment_type']
        return None, None, None

    def delete_attachment(self, attachment_id) -> bool:
        stmt = "DELETE FROM message_attachments WHERE id = :attachment_id"
        try:
            with self.engine.begin() as conn:
                conn.execute(text(stmt), {"attachment_id": attachment_id})
            return True
        except Exception as e:
            print(f"Error deleting attachment {attachment_id}: {e}")
            return False

    def get_last_command(self) -> Optional[Dict]:
        stmt = "SELECT * FROM command_history ORDER BY id DESC LIMIT 1"
        return self._fetch_one(stmt)

    def get_most_recent_conversation_id(self) -> Optional[Dict]:
        stmt = "SELECT conversation_id FROM conversation_history ORDER BY id DESC LIMIT 1"
        return self._fetch_one(stmt)

    def get_last_conversation(self, conversation_id) -> Optional[Dict]:
        stmt = """
            SELECT * FROM conversation_history
            WHERE conversation_id = :conversation_id and role = 'user'
            ORDER BY id DESC LIMIT 1
        """
        return self._fetch_one(stmt, {"conversation_id": conversation_id})

    def get_last_message_id(self, conversation_id) -> Optional[str]:
        """Get the message_id of the most recent message in a conversation."""
        stmt = """
            SELECT message_id FROM conversation_history
            WHERE conversation_id = :conversation_id
            ORDER BY id DESC LIMIT 1
        """
        row = self._fetch_one(stmt, {"conversation_id": conversation_id})
        return row["message_id"] if row else None

    def get_messages_by_npc(self, npc, n_last=20) -> List[Dict]:
        stmt = """
            SELECT * FROM conversation_history WHERE npc = :npc
            ORDER BY timestamp DESC LIMIT :n_last
        """
        return self._fetch_all(stmt, {"npc": npc, "n_last": n_last})

    def get_messages_by_team(self, team, n_last=20) -> List[Dict]:
        stmt = """
            SELECT * FROM conversation_history WHERE team = :team
            ORDER BY timestamp DESC LIMIT :n_last
        """
        return self._fetch_all(stmt, {"team": team, "n_last": n_last})

    def get_message_by_id(self, message_id) -> Optional[Dict]:
        stmt = "SELECT * FROM conversation_history WHERE message_id = :message_id"
        return self._fetch_one(stmt, {"message_id": message_id})

    def get_most_recent_conversation_id_by_path(self, path) -> Optional[Dict]:
        stmt = """
            SELECT conversation_id FROM conversation_history WHERE directory_path = :path
            ORDER BY timestamp DESC LIMIT 1
        """
        return self._fetch_one(stmt, {"path": path})

    def get_last_conversation_by_path(self, directory_path) -> Optional[List[Dict]]:
        result_dict = self.get_most_recent_conversation_id_by_path(directory_path)
        if result_dict and result_dict.get('conversation_id'):
            convo_id = result_dict['conversation_id']
            return self.get_conversations_by_id(convo_id)
        return None

    def get_conversations_by_id(self, conversation_id: str) -> List[Dict[str, Any]]:
        stmt = """
            SELECT id, message_id, timestamp, role, content, conversation_id,
                    directory_path, model, provider, npc, team,
                    reasoning_content, tool_calls, tool_results, parent_message_id, params
            FROM conversation_history WHERE conversation_id = :conversation_id
            ORDER BY timestamp ASC
        """
        results = self._fetch_all(stmt, {"conversation_id": conversation_id})

        for message_dict in results:
            attachments = self.get_message_attachments(message_dict["message_id"])
            if attachments:
                message_dict["attachments"] = attachments
            if message_dict.get("tool_calls"):
                try:
                    message_dict["tool_calls"] = json.loads(message_dict["tool_calls"])
                except (json.JSONDecodeError, TypeError):
                    pass
            if message_dict.get("tool_results"):
                try:
                    message_dict["tool_results"] = json.loads(message_dict["tool_results"])
                except (json.JSONDecodeError, TypeError):
                    pass
            if message_dict.get("params"):
                try:
                    message_dict["params"] = json.loads(message_dict["params"])
                except (json.JSONDecodeError, TypeError):
                    pass
        return results

    def get_npc_conversation_stats(self, start_date=None, end_date=None) -> pd.DataFrame:
        date_filter = ""
        params = {}

        if start_date and end_date:
            date_filter = "WHERE timestamp BETWEEN :start_date AND :end_date"
            params = {"start_date": start_date, "end_date": end_date}
        elif start_date:
            date_filter = "WHERE timestamp >= :start_date"
            params = {"start_date": start_date}
        elif end_date:
            date_filter = "WHERE timestamp <= :end_date"
            params = {"end_date": end_date}


        if 'sqlite' in str(self.engine.url):
            group_concat_models = "GROUP_CONCAT(DISTINCT model)"
            group_concat_providers = "GROUP_CONCAT(DISTINCT provider)"
        else:

            group_concat_models = "STRING_AGG(DISTINCT model, ',')"
            group_concat_providers = "STRING_AGG(DISTINCT provider, ',')"

        query = f"""
        SELECT
            npc,
            COUNT(*) as total_messages,
            AVG(LENGTH(content)) as avg_message_length,
            COUNT(DISTINCT conversation_id) as total_conversations,
            COUNT(DISTINCT model) as models_used,
            COUNT(DISTINCT provider) as providers_used,
            {group_concat_models} as model_list,
            {group_concat_providers} as provider_list,
            MIN(timestamp) as first_conversation,
            MAX(timestamp) as last_conversation
        FROM conversation_history
        {date_filter}
        GROUP BY npc
        ORDER BY total_messages DESC
        """

        try:
            df = pd.read_sql(sql=text(query), con=self.engine, params=params)
            return df
        except Exception as e:
            print(f"Error fetching conversation stats with pandas: {e}")
            return pd.DataFrame(columns=[
                'npc', 'total_messages', 'avg_message_length', 'total_conversations',
                'models_used', 'providers_used', 'model_list', 'provider_list',
                'first_conversation', 'last_conversation'
            ])

    def get_command_patterns(self, timeframe='day') -> pd.DataFrame:

        if 'sqlite' in str(self.engine.url):
            time_group_formats = {
                'hour': "strftime('%Y-%m-%d %H', timestamp)",
                'day': "strftime('%Y-%m-%d', timestamp)",
                'week': "strftime('%Y-%W', timestamp)",
                'month': "strftime('%Y-%m', timestamp)"
            }
        else:

            time_group_formats = {
                'hour': "TO_CHAR(timestamp::timestamp, 'YYYY-MM-DD HH24')",
                'day': "TO_CHAR(timestamp::timestamp, 'YYYY-MM-DD')",
                'week': "TO_CHAR(timestamp::timestamp, 'YYYY-WW')",
                'month': "TO_CHAR(timestamp::timestamp, 'YYYY-MM')"
            }

        time_group = time_group_formats.get(timeframe, time_group_formats['day'])


        if 'sqlite' in str(self.engine.url):
            substr_func = "SUBSTR"
            instr_func = "INSTR"
        else:
            substr_func = "SUBSTRING"
            instr_func = "POSITION"

        query = f"""
        WITH parsed_commands AS (
            SELECT
                {time_group} as time_bucket,
                CASE
                    WHEN command LIKE '/%%' THEN {substr_func}(command, 2, {instr_func}({substr_func}(command, 2), ' ') - 1)
                    WHEN command LIKE 'npc %%' THEN {substr_func}(command, 5, {instr_func}({substr_func}(command, 5), ' ') - 1)
                    ELSE command
                END as base_command
            FROM command_history
            WHERE timestamp IS NOT NULL
        )
        SELECT
            time_bucket,
            base_command,
            COUNT(*) as usage_count
        FROM parsed_commands
        WHERE base_command IS NOT NULL AND base_command != ''
        GROUP BY time_bucket, base_command
        ORDER BY time_bucket DESC, usage_count DESC
        """

        try:
            df = pd.read_sql(sql=text(query), con=self.engine)
            return df
        except Exception as e:
            print(f"Error fetching command patterns with pandas: {e}")
            return pd.DataFrame(columns=['time_period', 'command', 'count'])

    def search_commands(self, search_term: str) -> List[Dict]:
        """Searches command history table for a term."""
        stmt = """
            SELECT id, timestamp, command, subcommands, output, location
            FROM command_history
            WHERE LOWER(command) LIKE LOWER(:search_term) OR LOWER(output) LIKE LOWER(:search_term)
            ORDER BY timestamp DESC
            LIMIT 5
        """
        like_term = f"%{search_term}%"
        return self._fetch_all(stmt, {"search_term": like_term})

    def search_conversations(self, search_term: str) -> List[Dict]:
        """Searches conversation history table for a term."""
        stmt = """
            SELECT id, message_id, timestamp, role, content, conversation_id, directory_path, model, provider, npc, team
            FROM conversation_history
            WHERE LOWER(content) LIKE LOWER(:search_term)
            ORDER BY timestamp DESC
            LIMIT 5
        """
        like_term = f"%{search_term}%"
        return self._fetch_all(stmt, {"search_term": like_term})

    def get_all_commands(self, limit: int = 100) -> List[Dict]:
        """Gets the most recent commands."""
        stmt = """
            SELECT id, timestamp, command, subcommands, output, location
            FROM command_history
            ORDER BY id DESC
            LIMIT :limit
        """
        return self._fetch_all(stmt, {"limit": limit})

    def log_activity(self, activity_type: str, activity_data: str = None,
                     directory_path: str = None, npc: str = None,
                     device_id: str = None, session_id: str = None):
        import datetime
        ts = datetime.datetime.now().isoformat()
        with self.engine.begin() as conn:
            conn.execute(text("""
                INSERT INTO activity_log (timestamp, activity_type, activity_data, directory_path, npc, device_id, session_id)
                VALUES (:ts, :atype, :adata, :dpath, :npc, :did, :sid)
            """), {"ts": ts, "atype": activity_type, "adata": activity_data,
                   "dpath": directory_path, "npc": npc, "did": device_id, "sid": session_id})

    def log_autocomplete(self, suggestion_type: str, input_context: str, suggestion: str,
                         accepted: bool, npc: str = None, model: str = None,
                         provider: str = None, directory_path: str = None):
        import datetime
        ts = datetime.datetime.now().isoformat()
        with self.engine.begin() as conn:
            conn.execute(text("""
                INSERT INTO autocomplete_suggestions (timestamp, suggestion_type, input_context, suggestion, accepted, npc, model, provider, directory_path)
                VALUES (:ts, :stype, :ctx, :sug, :acc, :npc, :model, :prov, :dpath)
            """), {"ts": ts, "stype": suggestion_type, "ctx": input_context,
                   "sug": suggestion, "acc": 1 if accepted else 0,
                   "npc": npc, "model": model, "prov": provider, "dpath": directory_path})
            conn.execute(text("""
                INSERT INTO autocomplete_training (suggestion_type, input_text, output_text, accepted, npc, model)
                VALUES (:stype, :inp, :out, :acc, :npc, :model)
            """), {"stype": suggestion_type, "inp": input_context, "out": suggestion,
                   "acc": 1 if accepted else 0, "npc": npc, "model": model})

    def get_activities(self, activity_type: str = None, limit: int = 100,
                       directory_path: str = None, session_id: str = None):
        conditions = []
        params = {"lim": limit}
        if activity_type:
            conditions.append("activity_type = :atype")
            params["atype"] = activity_type
        if directory_path:
            conditions.append("directory_path = :dpath")
            params["dpath"] = directory_path
        if session_id:
            conditions.append("session_id = :sid")
            params["sid"] = session_id
        where = "WHERE " + " AND ".join(conditions) if conditions else ""
        return self._fetch_all(f"SELECT * FROM activity_log {where} ORDER BY timestamp DESC LIMIT :lim", params)

    def get_autocomplete_stats(self, suggestion_type: str = None, npc: str = None):
        conditions = []
        params = {}
        if suggestion_type:
            conditions.append("suggestion_type = :stype")
            params["stype"] = suggestion_type
        if npc:
            conditions.append("npc = :npc")
            params["npc"] = npc
        where = "WHERE " + " AND ".join(conditions) if conditions else ""
        return self._fetch_all(f"""
            SELECT suggestion_type,
                   COUNT(*) as total,
                   SUM(accepted) as accepted,
                   COUNT(*) - SUM(accepted) as rejected
            FROM autocomplete_suggestions {where}
            GROUP BY suggestion_type
        """, params)

    def get_training_data(self, suggestion_type: str = None, accepted_only: bool = False, limit: int = 1000):
        conditions = []
        params = {"lim": limit}
        if suggestion_type:
            conditions.append("suggestion_type = :stype")
            params["stype"] = suggestion_type
        if accepted_only:
            conditions.append("accepted = 1")
        where = "WHERE " + " AND ".join(conditions) if conditions else ""
        return self._fetch_all(f"SELECT * FROM autocomplete_training {where} ORDER BY created_at DESC LIMIT :lim", params)

    def close(self):
        """Dispose of the SQLAlchemy engine."""
        if self.engine:
            try:
                self.engine.dispose()
                logging.info("Disposed SQLAlchemy engine.")
            except Exception as e:
                print(f"Error disposing SQLAlchemy engine: {e}")
        self.engine = None

db_path = db instance-attribute

engine = create_engine_from_path(db) instance-attribute

add_attachment(message_id, name, attachment_type, data, size, file_path=None)

Source code in npcpy/memory/command_history.py
def add_attachment(self, message_id, name, attachment_type, data, size, file_path=None):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    stmt = """
        INSERT INTO message_attachments 
        (message_id, attachment_name, attachment_type, attachment_data, attachment_size, upload_timestamp, file_path)
        VALUES (:message_id, :name, :type, :data, :size, :timestamp, :file_path)
    """
    params = {
        "message_id": message_id,
        "name": name,
        "type": attachment_type,
        "data": data,
        "size": size,
        "timestamp": timestamp,
        "file_path": file_path
    }
    with self.engine.begin() as conn:
        conn.execute(text(stmt), params)

add_command(command, subcommands, output, location)

Source code in npcpy/memory/command_history.py
def add_command(self, command, subcommands, output, location):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    stmt = """
        INSERT INTO command_history (timestamp, command, subcommands, output, location)
        VALUES (:timestamp, :command, :subcommands, :output, :location)
    """
    params = {
        "timestamp": timestamp,
        "command": command,
        "subcommands": str(subcommands),
        "output": str(output),
        "location": location
    }

    with self.engine.begin() as conn:
        conn.execute(text(stmt), params)

add_conversation(message_id, timestamp, role, content, conversation_id, directory_path, model=None, provider=None, npc=None, team=None, attachments=None, reasoning_content=None, tool_calls=None, tool_results=None, parent_message_id=None, device_id=None, device_name=None, gen_params=None, input_tokens=None, output_tokens=None, cost=None)

Source code in npcpy/memory/command_history.py
def add_conversation(
    self,
    message_id,
    timestamp,
    role,
    content,
    conversation_id,
    directory_path,
    model=None,
    provider=None,
    npc=None,
    team=None,
    attachments=None,
    reasoning_content=None,
    tool_calls=None,
    tool_results=None,
    parent_message_id=None,
    device_id=None,
    device_name=None,
    gen_params=None,
    input_tokens=None,
    output_tokens=None,
    cost=None,
):
    if isinstance(content, (dict, list)):
        content = json.dumps(content, cls=CustomJSONEncoder)

    if tool_calls is not None and not isinstance(tool_calls, str):
        tool_calls = json.dumps(tool_calls, cls=CustomJSONEncoder)
    if tool_results is not None and not isinstance(tool_results, str):
        tool_results = json.dumps(tool_results, cls=CustomJSONEncoder)
    gen_params_json = None
    if gen_params is not None and not isinstance(gen_params, str):
        gen_params_json = json.dumps(gen_params, cls=CustomJSONEncoder)
    elif isinstance(gen_params, str):
        gen_params_json = gen_params

    normalized_directory_path = normalize_path_for_db(directory_path)

    cost_str = str(cost) if cost is not None else None

    stmt = """
        INSERT INTO conversation_history
        (message_id, timestamp, role, content, conversation_id, directory_path, model, provider, npc, team, reasoning_content, tool_calls, tool_results, parent_message_id, device_id, device_name, params, input_tokens, output_tokens, cost)
        VALUES (:message_id, :timestamp, :role, :content, :conversation_id, :directory_path, :model, :provider, :npc, :team, :reasoning_content, :tool_calls, :tool_results, :parent_message_id, :device_id, :device_name, :params, :input_tokens, :output_tokens, :cost)
    """
    params = {
        "message_id": message_id, "timestamp": timestamp, "role": role, "content": content,
        "conversation_id": conversation_id, "directory_path": normalized_directory_path, "model": model,
        "provider": provider, "npc": npc, "team": team, "reasoning_content": reasoning_content,
        "tool_calls": tool_calls, "tool_results": tool_results, "parent_message_id": parent_message_id,
        "device_id": device_id, "device_name": device_name, "params": gen_params_json,
        "input_tokens": input_tokens, "output_tokens": output_tokens, "cost": cost_str
    }
    with self.engine.begin() as conn:
        conn.execute(text(stmt), params)

    if attachments:
        for attachment in attachments:
            self.add_attachment(
                message_id=message_id,
                name=attachment.get("name"),
                attachment_type=attachment.get("type"),
                data=attachment.get("data"),
                size=attachment.get("size"),
                file_path=attachment.get("path")
            )

    return message_id

add_label(entity_type, entity_id, label, metadata=None)

Source code in npcpy/memory/command_history.py
def add_label(self, entity_type: str, entity_id: str, label: str, metadata: dict = None):
    stmt = """
        INSERT INTO labels (entity_type, entity_id, label, metadata)
        VALUES (:entity_type, :entity_id, :label, :metadata)
    """
    with self.engine.begin() as conn:
        conn.execute(text(stmt), {
            "entity_type": entity_type,
            "entity_id": entity_id,
            "label": label,
            "metadata": json.dumps(metadata) if metadata else None
        })

add_memory_to_database(message_id, conversation_id, npc, team, directory_path, initial_memory, status, model=None, provider=None, final_memory=None)

Store a memory entry in the database

Source code in npcpy/memory/command_history.py
def add_memory_to_database(self, message_id: str, conversation_id: str, npc: str, team: str,
                        directory_path: str, initial_memory: str, status: str,
                        model: str = None, provider: str = None, final_memory: str = None):
    """Store a memory entry in the database"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    normalized_directory_path = normalize_path_for_db(directory_path)

    stmt = """
        INSERT INTO memory_lifecycle
        (message_id, conversation_id, npc, team, directory_path, timestamp,
        initial_memory, final_memory, status, model, provider)
        VALUES (:message_id, :conversation_id, :npc, :team, :directory_path,
                :timestamp, :initial_memory, :final_memory, :status, :model, :provider)
    """

    params = {
        "message_id": message_id, "conversation_id": conversation_id,
        "npc": npc, "team": team, "directory_path": normalized_directory_path,
        "timestamp": timestamp, "initial_memory": initial_memory,
        "final_memory": final_memory, "status": status,
        "model": model, "provider": provider
    }

    return self._execute_returning_id(stmt, params)

backfill_execution_tables()

Source code in npcpy/memory/command_history.py
def backfill_execution_tables(self):
    with self.engine.begin() as conn:
        conn.execute(text("""
            INSERT OR IGNORE INTO jinx_executions 
            (message_id, jinx_name, input, timestamp, npc, team, 
            conversation_id)
            SELECT 
                message_id,
                SUBSTR(content, 2, 
                    CASE 
                        WHEN INSTR(SUBSTR(content, 2), ' ') > 0 
                        THEN INSTR(SUBSTR(content, 2), ' ') - 1
                        ELSE LENGTH(content) - 1
                    END
                ),
                content,
                timestamp,
                npc,
                team,
                conversation_id
            FROM conversation_history
            WHERE role = 'user' AND content LIKE '/%'
        """))

        conn.execute(text("""
            INSERT OR IGNORE INTO npc_executions 
            (message_id, input, timestamp, npc, team, conversation_id, 
            model, provider)
            SELECT 
                message_id,
                content,
                timestamp,
                npc,
                team,
                conversation_id,
                model,
                provider
            FROM conversation_history
            WHERE role = 'user' AND npc IS NOT NULL
        """))

close()

Dispose of the SQLAlchemy engine.

Source code in npcpy/memory/command_history.py
def close(self):
    """Dispose of the SQLAlchemy engine."""
    if self.engine:
        try:
            self.engine.dispose()
            logging.info("Disposed SQLAlchemy engine.")
        except Exception as e:
            print(f"Error disposing SQLAlchemy engine: {e}")
    self.engine = None

delete_attachment(attachment_id)

Source code in npcpy/memory/command_history.py
def delete_attachment(self, attachment_id) -> bool:
    stmt = "DELETE FROM message_attachments WHERE id = :attachment_id"
    try:
        with self.engine.begin() as conn:
            conn.execute(text(stmt), {"attachment_id": attachment_id})
        return True
    except Exception as e:
        print(f"Error deleting attachment {attachment_id}: {e}")
        return False

delete_message(conversation_id, message_id)

Delete a specific message from a conversation

Source code in npcpy/memory/command_history.py
def delete_message(self, conversation_id, message_id):
    """Delete a specific message from a conversation"""
    conn = sqlite3.connect(self.db_path)
    cursor = conn.cursor()

    try:
        cursor.execute("""
            DELETE FROM messages 
            WHERE conversation_id = ? AND message_id = ?
        """, (conversation_id, message_id))

        rows_affected = cursor.rowcount
        conn.commit()

        print(f"[DB] Deleted message {message_id} from conversation {conversation_id}. Rows affected: {rows_affected}")

        return rows_affected

    except Exception as e:
        print(f"[DB] Error deleting message: {e}")
        conn.rollback()
        raise
    finally:
        conn.close()

get_activities(activity_type=None, limit=100, directory_path=None, session_id=None)

Source code in npcpy/memory/command_history.py
def get_activities(self, activity_type: str = None, limit: int = 100,
                   directory_path: str = None, session_id: str = None):
    conditions = []
    params = {"lim": limit}
    if activity_type:
        conditions.append("activity_type = :atype")
        params["atype"] = activity_type
    if directory_path:
        conditions.append("directory_path = :dpath")
        params["dpath"] = directory_path
    if session_id:
        conditions.append("session_id = :sid")
        params["sid"] = session_id
    where = "WHERE " + " AND ".join(conditions) if conditions else ""
    return self._fetch_all(f"SELECT * FROM activity_log {where} ORDER BY timestamp DESC LIMIT :lim", params)

get_all_commands(limit=100)

Gets the most recent commands.

Source code in npcpy/memory/command_history.py
def get_all_commands(self, limit: int = 100) -> List[Dict]:
    """Gets the most recent commands."""
    stmt = """
        SELECT id, timestamp, command, subcommands, output, location
        FROM command_history
        ORDER BY id DESC
        LIMIT :limit
    """
    return self._fetch_all(stmt, {"limit": limit})

get_approved_memories_by_scope()

Get all approved/edited memories grouped by (npc, team, path) scope.

Source code in npcpy/memory/command_history.py
def get_approved_memories_by_scope(self):
    """Get all approved/edited memories grouped by (npc, team, path) scope."""
    stmt = """
        SELECT id, npc, team, directory_path, initial_memory, final_memory, status
        FROM memory_lifecycle
        WHERE status IN ('human-approved', 'human-edited')
        ORDER BY npc, team, directory_path, created_at
    """
    rows = self._fetch_all(stmt, {})

    from collections import defaultdict
    memories_by_scope = defaultdict(list)
    for row in rows:
        statement = row.get('final_memory') or row.get('initial_memory')
        scope_key = (
            row.get('npc') or 'default',
            row.get('team') or 'global_team',
            row.get('directory_path') or os.getcwd()
        )
        memories_by_scope[scope_key].append({
            'id': row.get('id'),
            'statement': statement,
            'source_text': '',
            'type': 'explicit',
            'generation': 0
        })
    return dict(memories_by_scope)

get_attachment_data(attachment_id)

Source code in npcpy/memory/command_history.py
def get_attachment_data(self, attachment_id) -> Optional[Tuple[bytes, str, str]]:
    stmt = "SELECT attachment_data, attachment_name, attachment_type FROM message_attachments WHERE id = :attachment_id"
    row = self._fetch_one(stmt, {"attachment_id": attachment_id})
    if row:
        return row['attachment_data'], row['attachment_name'], row['attachment_type']
    return None, None, None

get_autocomplete_stats(suggestion_type=None, npc=None)

Source code in npcpy/memory/command_history.py
def get_autocomplete_stats(self, suggestion_type: str = None, npc: str = None):
    conditions = []
    params = {}
    if suggestion_type:
        conditions.append("suggestion_type = :stype")
        params["stype"] = suggestion_type
    if npc:
        conditions.append("npc = :npc")
        params["npc"] = npc
    where = "WHERE " + " AND ".join(conditions) if conditions else ""
    return self._fetch_all(f"""
        SELECT suggestion_type,
               COUNT(*) as total,
               SUM(accepted) as accepted,
               COUNT(*) - SUM(accepted) as rejected
        FROM autocomplete_suggestions {where}
        GROUP BY suggestion_type
    """, params)

get_command_patterns(timeframe='day')

Source code in npcpy/memory/command_history.py
def get_command_patterns(self, timeframe='day') -> pd.DataFrame:

    if 'sqlite' in str(self.engine.url):
        time_group_formats = {
            'hour': "strftime('%Y-%m-%d %H', timestamp)",
            'day': "strftime('%Y-%m-%d', timestamp)",
            'week': "strftime('%Y-%W', timestamp)",
            'month': "strftime('%Y-%m', timestamp)"
        }
    else:

        time_group_formats = {
            'hour': "TO_CHAR(timestamp::timestamp, 'YYYY-MM-DD HH24')",
            'day': "TO_CHAR(timestamp::timestamp, 'YYYY-MM-DD')",
            'week': "TO_CHAR(timestamp::timestamp, 'YYYY-WW')",
            'month': "TO_CHAR(timestamp::timestamp, 'YYYY-MM')"
        }

    time_group = time_group_formats.get(timeframe, time_group_formats['day'])


    if 'sqlite' in str(self.engine.url):
        substr_func = "SUBSTR"
        instr_func = "INSTR"
    else:
        substr_func = "SUBSTRING"
        instr_func = "POSITION"

    query = f"""
    WITH parsed_commands AS (
        SELECT
            {time_group} as time_bucket,
            CASE
                WHEN command LIKE '/%%' THEN {substr_func}(command, 2, {instr_func}({substr_func}(command, 2), ' ') - 1)
                WHEN command LIKE 'npc %%' THEN {substr_func}(command, 5, {instr_func}({substr_func}(command, 5), ' ') - 1)
                ELSE command
            END as base_command
        FROM command_history
        WHERE timestamp IS NOT NULL
    )
    SELECT
        time_bucket,
        base_command,
        COUNT(*) as usage_count
    FROM parsed_commands
    WHERE base_command IS NOT NULL AND base_command != ''
    GROUP BY time_bucket, base_command
    ORDER BY time_bucket DESC, usage_count DESC
    """

    try:
        df = pd.read_sql(sql=text(query), con=self.engine)
        return df
    except Exception as e:
        print(f"Error fetching command patterns with pandas: {e}")
        return pd.DataFrame(columns=['time_period', 'command', 'count'])

get_conversations_by_id(conversation_id)

Source code in npcpy/memory/command_history.py
def get_conversations_by_id(self, conversation_id: str) -> List[Dict[str, Any]]:
    stmt = """
        SELECT id, message_id, timestamp, role, content, conversation_id,
                directory_path, model, provider, npc, team,
                reasoning_content, tool_calls, tool_results, parent_message_id, params
        FROM conversation_history WHERE conversation_id = :conversation_id
        ORDER BY timestamp ASC
    """
    results = self._fetch_all(stmt, {"conversation_id": conversation_id})

    for message_dict in results:
        attachments = self.get_message_attachments(message_dict["message_id"])
        if attachments:
            message_dict["attachments"] = attachments
        if message_dict.get("tool_calls"):
            try:
                message_dict["tool_calls"] = json.loads(message_dict["tool_calls"])
            except (json.JSONDecodeError, TypeError):
                pass
        if message_dict.get("tool_results"):
            try:
                message_dict["tool_results"] = json.loads(message_dict["tool_results"])
            except (json.JSONDecodeError, TypeError):
                pass
        if message_dict.get("params"):
            try:
                message_dict["params"] = json.loads(message_dict["params"])
            except (json.JSONDecodeError, TypeError):
                pass
    return results

get_full_message_content(message_id)

Source code in npcpy/memory/command_history.py
def get_full_message_content(self, message_id):
    stmt = "SELECT content FROM conversation_history WHERE message_id = :message_id ORDER BY timestamp ASC"
    rows = self._fetch_all(stmt, {"message_id": message_id})
    return "".join(row['content'] for row in rows)

get_jinx_executions(jinx_name=None, limit=1000)

Source code in npcpy/memory/command_history.py
def get_jinx_executions(self, jinx_name: str = None, limit: int = 1000) -> List[Dict]:
    if jinx_name:
        stmt = """
            SELECT je.*, l.label
            FROM jinx_executions je
            LEFT JOIN labels l ON l.entity_type = 'message' 
                AND l.entity_id = je.message_id
            WHERE je.jinx_name = :jinx_name
            ORDER BY je.timestamp DESC
            LIMIT :limit
        """
        return self._fetch_all(stmt, {"jinx_name": jinx_name, "limit": limit})

    stmt = """
        SELECT je.*, l.label
        FROM jinx_executions je
        LEFT JOIN labels l ON l.entity_type = 'message' 
            AND l.entity_id = je.message_id
        ORDER BY je.timestamp DESC
        LIMIT :limit
    """
    return self._fetch_all(stmt, {"limit": limit})

get_labels(entity_type=None, label=None)

Source code in npcpy/memory/command_history.py
def get_labels(self, entity_type: str = None, label: str = None) -> List[Dict]:
    conditions = []
    params = {}

    if entity_type:
        conditions.append("entity_type = :entity_type")
        params["entity_type"] = entity_type
    if label:
        conditions.append("label = :label")
        params["label"] = label

    where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
    stmt = f"SELECT * FROM labels {where} ORDER BY created_at DESC"

    return self._fetch_all(stmt, params)

get_last_command()

Source code in npcpy/memory/command_history.py
def get_last_command(self) -> Optional[Dict]:
    stmt = "SELECT * FROM command_history ORDER BY id DESC LIMIT 1"
    return self._fetch_one(stmt)

get_last_conversation(conversation_id)

Source code in npcpy/memory/command_history.py
def get_last_conversation(self, conversation_id) -> Optional[Dict]:
    stmt = """
        SELECT * FROM conversation_history
        WHERE conversation_id = :conversation_id and role = 'user'
        ORDER BY id DESC LIMIT 1
    """
    return self._fetch_one(stmt, {"conversation_id": conversation_id})

get_last_conversation_by_path(directory_path)

Source code in npcpy/memory/command_history.py
def get_last_conversation_by_path(self, directory_path) -> Optional[List[Dict]]:
    result_dict = self.get_most_recent_conversation_id_by_path(directory_path)
    if result_dict and result_dict.get('conversation_id'):
        convo_id = result_dict['conversation_id']
        return self.get_conversations_by_id(convo_id)
    return None

get_last_message_id(conversation_id)

Get the message_id of the most recent message in a conversation.

Source code in npcpy/memory/command_history.py
def get_last_message_id(self, conversation_id) -> Optional[str]:
    """Get the message_id of the most recent message in a conversation."""
    stmt = """
        SELECT message_id FROM conversation_history
        WHERE conversation_id = :conversation_id
        ORDER BY id DESC LIMIT 1
    """
    row = self._fetch_one(stmt, {"conversation_id": conversation_id})
    return row["message_id"] if row else None

get_memories_for_scope(npc, team, directory_path, status=None)

Source code in npcpy/memory/command_history.py
def get_memories_for_scope(
    self,
    npc: str,
    team: str,
    directory_path: str,
    status: Optional[str] = None
) -> List[Dict]:

    conditions = []
    params = {}

    if npc:
        conditions.append("npc = :npc")
        params["npc"] = npc
    if team:
        conditions.append("team = :team")
        params["team"] = team
    if directory_path:
        conditions.append("directory_path = :path")
        params["path"] = directory_path
    if status:
        conditions.append("status = :status")
        params["status"] = status

    where = "WHERE " + " AND ".join(conditions) if conditions else ""
    query = f"""
        SELECT id, initial_memory, final_memory,
            status, timestamp, created_at, npc, team, directory_path
        FROM memory_lifecycle
        {where}
        ORDER BY created_at DESC
    """
    return self._fetch_all(query, params)

get_memory_examples_for_context(npc, team, directory_path, n_approved=10, n_rejected=10, n_edited=5)

Get recent approved, rejected, and edited memories for learning context.

Source code in npcpy/memory/command_history.py
def get_memory_examples_for_context(self, npc: str, team: str, directory_path: str,
                                n_approved: int = 10, n_rejected: int = 10, n_edited: int = 5):
    """Get recent approved, rejected, and edited memories for learning context."""

    scope_order = """
        CASE WHEN npc = :npc AND team = :team AND directory_path = :path THEN 1
             WHEN npc = :npc AND team = :team THEN 2
             WHEN team = :team THEN 3
             ELSE 4 END
    """

    approved_stmt = f"""
        SELECT initial_memory, final_memory, status FROM memory_lifecycle
        WHERE status IN ('human-approved', 'model-approved')
        ORDER BY {scope_order}, created_at DESC
        LIMIT :n_approved
    """

    rejected_stmt = f"""
        SELECT initial_memory, status FROM memory_lifecycle
        WHERE status IN ('human-rejected', 'model-rejected')
        ORDER BY {scope_order}, created_at DESC
        LIMIT :n_rejected
    """

    edited_stmt = f"""
        SELECT initial_memory, final_memory, status FROM memory_lifecycle
        WHERE status = 'human-edited' AND final_memory IS NOT NULL
        ORDER BY {scope_order}, created_at DESC
        LIMIT :n_edited
    """

    params = {"npc": npc, "team": team, "path": directory_path,
            "n_approved": n_approved, "n_rejected": n_rejected, "n_edited": n_edited}

    approved = self._fetch_all(approved_stmt, params)
    rejected = self._fetch_all(rejected_stmt, params)
    edited = self._fetch_all(edited_stmt, params)

    return {"approved": approved, "rejected": rejected, "edited": edited}

get_message_attachments(message_id)

Source code in npcpy/memory/command_history.py
def get_message_attachments(self, message_id) -> List[Dict]:
    stmt = """
        SELECT 
            id, 
            message_id, 
            attachment_name, 
            attachment_type, 
            attachment_size, 
            upload_timestamp
        FROM message_attachments WHERE message_id = :message_id
    """
    return self._fetch_all(stmt, {"message_id": message_id})

get_message_by_id(message_id)

Source code in npcpy/memory/command_history.py
def get_message_by_id(self, message_id) -> Optional[Dict]:
    stmt = "SELECT * FROM conversation_history WHERE message_id = :message_id"
    return self._fetch_one(stmt, {"message_id": message_id})

get_messages_by_npc(npc, n_last=20)

Source code in npcpy/memory/command_history.py
def get_messages_by_npc(self, npc, n_last=20) -> List[Dict]:
    stmt = """
        SELECT * FROM conversation_history WHERE npc = :npc
        ORDER BY timestamp DESC LIMIT :n_last
    """
    return self._fetch_all(stmt, {"npc": npc, "n_last": n_last})

get_messages_by_team(team, n_last=20)

Source code in npcpy/memory/command_history.py
def get_messages_by_team(self, team, n_last=20) -> List[Dict]:
    stmt = """
        SELECT * FROM conversation_history WHERE team = :team
        ORDER BY timestamp DESC LIMIT :n_last
    """
    return self._fetch_all(stmt, {"team": team, "n_last": n_last})

get_most_recent_conversation_id()

Source code in npcpy/memory/command_history.py
def get_most_recent_conversation_id(self) -> Optional[Dict]:
    stmt = "SELECT conversation_id FROM conversation_history ORDER BY id DESC LIMIT 1"
    return self._fetch_one(stmt)

get_most_recent_conversation_id_by_path(path)

Source code in npcpy/memory/command_history.py
def get_most_recent_conversation_id_by_path(self, path) -> Optional[Dict]:
    stmt = """
        SELECT conversation_id FROM conversation_history WHERE directory_path = :path
        ORDER BY timestamp DESC LIMIT 1
    """
    return self._fetch_one(stmt, {"path": path})

get_npc_conversation_stats(start_date=None, end_date=None)

Source code in npcpy/memory/command_history.py
def get_npc_conversation_stats(self, start_date=None, end_date=None) -> pd.DataFrame:
    date_filter = ""
    params = {}

    if start_date and end_date:
        date_filter = "WHERE timestamp BETWEEN :start_date AND :end_date"
        params = {"start_date": start_date, "end_date": end_date}
    elif start_date:
        date_filter = "WHERE timestamp >= :start_date"
        params = {"start_date": start_date}
    elif end_date:
        date_filter = "WHERE timestamp <= :end_date"
        params = {"end_date": end_date}


    if 'sqlite' in str(self.engine.url):
        group_concat_models = "GROUP_CONCAT(DISTINCT model)"
        group_concat_providers = "GROUP_CONCAT(DISTINCT provider)"
    else:

        group_concat_models = "STRING_AGG(DISTINCT model, ',')"
        group_concat_providers = "STRING_AGG(DISTINCT provider, ',')"

    query = f"""
    SELECT
        npc,
        COUNT(*) as total_messages,
        AVG(LENGTH(content)) as avg_message_length,
        COUNT(DISTINCT conversation_id) as total_conversations,
        COUNT(DISTINCT model) as models_used,
        COUNT(DISTINCT provider) as providers_used,
        {group_concat_models} as model_list,
        {group_concat_providers} as provider_list,
        MIN(timestamp) as first_conversation,
        MAX(timestamp) as last_conversation
    FROM conversation_history
    {date_filter}
    GROUP BY npc
    ORDER BY total_messages DESC
    """

    try:
        df = pd.read_sql(sql=text(query), con=self.engine, params=params)
        return df
    except Exception as e:
        print(f"Error fetching conversation stats with pandas: {e}")
        return pd.DataFrame(columns=[
            'npc', 'total_messages', 'avg_message_length', 'total_conversations',
            'models_used', 'providers_used', 'model_list', 'provider_list',
            'first_conversation', 'last_conversation'
        ])

get_npc_executions(npc_name, limit=1000)

Source code in npcpy/memory/command_history.py
def get_npc_executions(self, npc_name: str, limit: int = 1000) -> List[Dict]:
    stmt = """
        SELECT ne.*, l.label
        FROM npc_executions ne
        LEFT JOIN labels l ON l.entity_type = 'message' 
            AND l.entity_id = ne.message_id
        WHERE ne.npc = :npc_name
        ORDER BY ne.timestamp DESC
        LIMIT :limit
    """
    return self._fetch_all(stmt, {"npc_name": npc_name, "limit": limit})

get_pending_memories(limit=50)

Get memories pending human approval

Source code in npcpy/memory/command_history.py
def get_pending_memories(self, limit: int = 50):
    """Get memories pending human approval"""
    stmt = """
        SELECT * FROM memory_lifecycle 
        WHERE status = 'pending_approval'
        ORDER BY created_at ASC
        LIMIT :limit
    """
    return self._fetch_all(stmt, {"limit": limit})

get_training_data(suggestion_type=None, accepted_only=False, limit=1000)

Source code in npcpy/memory/command_history.py
def get_training_data(self, suggestion_type: str = None, accepted_only: bool = False, limit: int = 1000):
    conditions = []
    params = {"lim": limit}
    if suggestion_type:
        conditions.append("suggestion_type = :stype")
        params["stype"] = suggestion_type
    if accepted_only:
        conditions.append("accepted = 1")
    where = "WHERE " + " AND ".join(conditions) if conditions else ""
    return self._fetch_all(f"SELECT * FROM autocomplete_training {where} ORDER BY created_at DESC LIMIT :lim", params)

get_training_data_by_label(label='training')

Source code in npcpy/memory/command_history.py
def get_training_data_by_label(self, label: str = 'training') -> List[Dict]:
    stmt = """
        SELECT l.entity_type, l.entity_id, l.metadata,
            ch.content, ch.role, ch.npc, ch.conversation_id
        FROM labels l
        LEFT JOIN conversation_history ch ON 
            (l.entity_type = 'message' AND l.entity_id = ch.message_id)
        WHERE l.label = :label
    """
    return self._fetch_all(stmt, {"label": label})

label_execution(message_id, label)

Source code in npcpy/memory/command_history.py
def label_execution(self, message_id: str, label: str):
    self.add_label('message', message_id, label)

log_activity(activity_type, activity_data=None, directory_path=None, npc=None, device_id=None, session_id=None)

Source code in npcpy/memory/command_history.py
def log_activity(self, activity_type: str, activity_data: str = None,
                 directory_path: str = None, npc: str = None,
                 device_id: str = None, session_id: str = None):
    import datetime
    ts = datetime.datetime.now().isoformat()
    with self.engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO activity_log (timestamp, activity_type, activity_data, directory_path, npc, device_id, session_id)
            VALUES (:ts, :atype, :adata, :dpath, :npc, :did, :sid)
        """), {"ts": ts, "atype": activity_type, "adata": activity_data,
               "dpath": directory_path, "npc": npc, "did": device_id, "sid": session_id})

log_autocomplete(suggestion_type, input_context, suggestion, accepted, npc=None, model=None, provider=None, directory_path=None)

Source code in npcpy/memory/command_history.py
def log_autocomplete(self, suggestion_type: str, input_context: str, suggestion: str,
                     accepted: bool, npc: str = None, model: str = None,
                     provider: str = None, directory_path: str = None):
    import datetime
    ts = datetime.datetime.now().isoformat()
    with self.engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO autocomplete_suggestions (timestamp, suggestion_type, input_context, suggestion, accepted, npc, model, provider, directory_path)
            VALUES (:ts, :stype, :ctx, :sug, :acc, :npc, :model, :prov, :dpath)
        """), {"ts": ts, "stype": suggestion_type, "ctx": input_context,
               "sug": suggestion, "acc": 1 if accepted else 0,
               "npc": npc, "model": model, "prov": provider, "dpath": directory_path})
        conn.execute(text("""
            INSERT INTO autocomplete_training (suggestion_type, input_text, output_text, accepted, npc, model)
            VALUES (:stype, :inp, :out, :acc, :npc, :model)
        """), {"stype": suggestion_type, "inp": input_context, "out": suggestion,
               "acc": 1 if accepted else 0, "npc": npc, "model": model})

save_jinx_execution(triggering_message_id, conversation_id, npc_name, jinx_name, jinx_inputs, jinx_output, status, team_name=None, error_message=None, response_message_id=None, duration_ms=None)

Source code in npcpy/memory/command_history.py
def save_jinx_execution(
    self,
    triggering_message_id: str,
    conversation_id: str,
    npc_name: Optional[str],
    jinx_name: str,
    jinx_inputs: Dict,
    jinx_output: Any,
    status: str,
    team_name: Optional[str] = None,
    error_message: Optional[str] = None,
    response_message_id: Optional[str] = None,
    duration_ms: Optional[int] = None
):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    try:
        inputs_json = json.dumps(jinx_inputs, cls=CustomJSONEncoder)
    except TypeError:
        inputs_json = json.dumps(str(jinx_inputs))

    try:
        if isinstance(jinx_output, (str, int, float, bool, list, dict, type(None))):
            outputs_json = json.dumps(jinx_output, cls=CustomJSONEncoder)
        else:
            outputs_json = json.dumps(str(jinx_output))
    except TypeError:
        outputs_json = json.dumps(f"Non-serializable output: {type(jinx_output)}")

    msg_id = triggering_message_id or f"jinx-{jinx_name}-{timestamp.replace(' ', '-')}"

    stmt = """
        INSERT OR REPLACE INTO jinx_executions
        (message_id, jinx_name, input, timestamp, npc, team,
         conversation_id, output, status, error_message, duration_ms)
        VALUES (:message_id, :jinx_name, :input, :timestamp, :npc, :team,
                :conversation_id, :output, :status, :error_message, :duration_ms)
    """
    params = {
        "message_id": msg_id,
        "jinx_name": jinx_name,
        "input": inputs_json,
        "timestamp": timestamp,
        "npc": npc_name,
        "team": team_name,
        "conversation_id": conversation_id,
        "output": outputs_json,
        "status": status,
        "error_message": error_message,
        "duration_ms": duration_ms,
    }

    with self.engine.begin() as conn:
        conn.execute(text(stmt), params)

search_commands(search_term)

Searches command history table for a term.

Source code in npcpy/memory/command_history.py
def search_commands(self, search_term: str) -> List[Dict]:
    """Searches command history table for a term."""
    stmt = """
        SELECT id, timestamp, command, subcommands, output, location
        FROM command_history
        WHERE LOWER(command) LIKE LOWER(:search_term) OR LOWER(output) LIKE LOWER(:search_term)
        ORDER BY timestamp DESC
        LIMIT 5
    """
    like_term = f"%{search_term}%"
    return self._fetch_all(stmt, {"search_term": like_term})

search_conversations(search_term)

Searches conversation history table for a term.

Source code in npcpy/memory/command_history.py
def search_conversations(self, search_term: str) -> List[Dict]:
    """Searches conversation history table for a term."""
    stmt = """
        SELECT id, message_id, timestamp, role, content, conversation_id, directory_path, model, provider, npc, team
        FROM conversation_history
        WHERE LOWER(content) LIKE LOWER(:search_term)
        ORDER BY timestamp DESC
        LIMIT 5
    """
    like_term = f"%{search_term}%"
    return self._fetch_all(stmt, {"search_term": like_term})

search_memory(query, npc=None, team=None, directory_path=None, status_filter=None, limit=10)

Search memories with hierarchical scope

Source code in npcpy/memory/command_history.py
def search_memory(self, query: str, npc: str = None, team: str = None, 
                directory_path: str = None, status_filter: str = None, limit: int = 10):
    """Search memories with hierarchical scope"""
    conditions = ["LOWER(initial_memory) LIKE LOWER(:query) OR LOWER(final_memory) LIKE LOWER(:query)"]
    params = {"query": f"%{query}%"}

    if status_filter:
        conditions.append("status = :status")
        params["status"] = status_filter


    order_parts = []
    if npc:
        order_parts.append(f"CASE WHEN npc = '{npc}' THEN 1 ELSE 2 END")
    if team:
        order_parts.append(f"CASE WHEN team = '{team}' THEN 1 ELSE 2 END")
    if directory_path:
        order_parts.append(f"CASE WHEN directory_path = '{directory_path}' THEN 1 ELSE 2 END")

    order_clause = ", ".join(order_parts) + ", created_at DESC" if order_parts else "created_at DESC"

    stmt = f"""
        SELECT * FROM memory_lifecycle 
        WHERE {' AND '.join(conditions)}
        ORDER BY {order_clause}
        LIMIT :limit
    """
    params["limit"] = limit

    return self._fetch_all(stmt, params)

update_memory_status(memory_id, new_status, final_memory=None)

Update memory status and optionally final_memory

Source code in npcpy/memory/command_history.py
def update_memory_status(self, memory_id: int, new_status: str, final_memory: str = None):
    """Update memory status and optionally final_memory"""
    stmt = """
        UPDATE memory_lifecycle 
        SET status = :status, final_memory = :final_memory
        WHERE id = :memory_id
    """
    params = {"status": new_status, "final_memory": final_memory, "memory_id": memory_id}

    with self.engine.begin() as conn:
        conn.execute(text(stmt), params)

update_message_content(message_id, full_content)

Source code in npcpy/memory/command_history.py
def update_message_content(self, message_id, full_content):
    stmt = "UPDATE conversation_history SET content = :content WHERE message_id = :message_id"
    with self.engine.begin() as conn:
        conn.execute(text(stmt), {"content": full_content, "message_id": message_id})

CustomJSONEncoder

Bases: JSONEncoder

Source code in npcpy/memory/command_history.py
class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        try:
            return deep_to_dict(obj)
        except TypeError:
            return super().default(obj)

default(obj)

Source code in npcpy/memory/command_history.py
def default(self, obj):
    try:
        return deep_to_dict(obj)
    except TypeError:
        return super().default(obj)

_resolve_path(base_dir, table, row, ext='csv')

Source code in npcpy/memory/command_history.py
def _resolve_path(base_dir, table, row, ext="csv"):
    import pathlib
    ts = row.get("timestamp", "") or row.get("created_at", "") or row.get("compiled_at", "") or row.get("upload_timestamp", "")
    try:
        dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
    except (ValueError, TypeError):
        try:
            dt = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S")
        except (ValueError, TypeError):
            dt = datetime.now()

    dir_path = row.get("directory_path", "") or row.get("location", "")
    path_part = dir_path.strip("/\\").replace("\\", "/") if dir_path else "_local"

    group_id = (
        row.get("conversation_id")
        or row.get("npc_name")
        or row.get("name")
        or row.get("entity_id")
        or row.get("message_id")
        or "default"
    )

    return (pathlib.Path(base_dir) / table / path_part
            / str(dt.year) / f"{dt.month:02d}" / f"{dt.day:02d}"
            / f"{group_id}.{ext}")

append_row_csv(base_dir, table, row)

Source code in npcpy/memory/command_history.py
def append_row_csv(base_dir, table, row):
    import csv as csv_mod
    columns = TABLE_SCHEMAS.get(table)
    if not columns:
        raise ValueError(f"Unknown table: {table}")
    clean = {col: str(row.get(col, "")) for col in columns}
    path = _resolve_path(base_dir, table, clean, "csv")
    path.parent.mkdir(parents=True, exist_ok=True)
    exists = path.exists()
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv_mod.DictWriter(f, fieldnames=columns)
        if not exists:
            writer.writeheader()
        writer.writerow(clean)
    return str(path)

append_row_parquet(base_dir, table, row)

Source code in npcpy/memory/command_history.py
def append_row_parquet(base_dir, table, row):
    import polars as pl
    columns = TABLE_SCHEMAS.get(table)
    if not columns:
        raise ValueError(f"Unknown table: {table}")
    clean = {col: row.get(col, "") for col in columns}
    for col in columns:
        if clean[col] is None:
            clean[col] = ""
    path = _resolve_path(base_dir, table, clean, "parquet")
    path.parent.mkdir(parents=True, exist_ok=True)
    new_df = pl.DataFrame([{k: str(v) for k, v in clean.items()}])
    if path.exists():
        existing = pl.read_parquet(path)
        combined = pl.concat([existing, new_df])
        combined.write_parquet(path)
    else:
        new_df.write_parquet(path)
    return str(path)

create_engine_from_path(db_path)

Create SQLAlchemy engine from database path, detecting type

Source code in npcpy/memory/command_history.py
def create_engine_from_path(db_path: str) -> Engine:
    """Create SQLAlchemy engine from database path, detecting type"""
    if db_path.startswith('postgresql://') or db_path.startswith('postgres://'):
        return create_engine(db_path)
    else:

        if db_path.startswith('~/'):
            db_path = os.path.expanduser(db_path)
        return create_engine(f'sqlite:///{db_path}')

deep_to_dict(obj)

Recursively convert objects that have a 'to_dict' method to dictionaries, otherwise drop them from the output.

Source code in npcpy/memory/command_history.py
def deep_to_dict(obj):
    """
    Recursively convert objects that have a 'to_dict' method to dictionaries,
    otherwise drop them from the output.
    """
    if isinstance(obj, dict):
        return {key: deep_to_dict(val) for key, val in obj.items()}

    if isinstance(obj, list):
        return [deep_to_dict(item) for item in obj]

    if hasattr(obj, "to_dict") and callable(getattr(obj, "to_dict", None)):
        return deep_to_dict(obj.to_dict())

    if isinstance(obj, (int, float, str, bool, type(None))):
        return obj

    return None

fetch_messages_for_conversation(engine, conversation_id)

Source code in npcpy/memory/command_history.py
def fetch_messages_for_conversation(engine: Engine, conversation_id: str):
    query = text("""
        SELECT role, content, timestamp
        FROM conversation_history
        WHERE conversation_id = :conversation_id
        ORDER BY timestamp ASC
    """)

    with engine.connect() as conn:
        result = conn.execute(query, {"conversation_id": conversation_id})
        return [
            {
                "role": row.role,
                "content": row.content,
                "timestamp": row.timestamp,
            }
            for row in result
        ]

flush_messages(n, messages)

Source code in npcpy/memory/command_history.py
def flush_messages(n: int, messages: list) -> dict:
    if n <= 0:
        return {
            "messages": messages,
            "output": "Error: 'n' must be a positive integer.",
        }

    removed_count = min(n, len(messages))
    del messages[-removed_count:]

    return {
        "messages": messages,
        "output": f"Flushed {removed_count} message(s). Context count is now {len(messages)} messages.",
    }

format_memory_context(memory_examples)

Source code in npcpy/memory/command_history.py
def format_memory_context(memory_examples):
    if not memory_examples:
        return ""

    approved_examples = memory_examples.get("approved", [])
    rejected_examples = memory_examples.get("rejected", [])
    edited_examples = memory_examples.get("edited", [])

    if not approved_examples and not rejected_examples and not edited_examples:
        return ""

    parts = ["MEMORY QUALITY GUIDELINES (based on user feedback):"]

    if approved_examples:
        parts.append("\nAPPROVED — memories like these were kept:")
        for ex in approved_examples[:7]:
            mem = ex.get("final_memory") or ex.get("initial_memory")
            parts.append(f"  + {mem}")

    if edited_examples:
        parts.append("\nCORRECTED — the user fixed these (learn from the corrections):")
        for ex in edited_examples[:5]:
            original = ex.get("initial_memory", "")
            corrected = ex.get("final_memory", "")
            if original and corrected and original != corrected:
                parts.append(f"  BEFORE: {original}")
                parts.append(f"  AFTER:  {corrected}")
                parts.append("")

    if rejected_examples:
        parts.append("\nREJECTED — memories like these were thrown out (do NOT generate similar ones):")
        for ex in rejected_examples[:5]:
            parts.append(f"  x {ex.get('initial_memory')}")

    parts.append("\nRULES derived from this feedback:")
    parts.append("- Match the style and specificity of approved memories.")
    if edited_examples:
        parts.append("- Apply the same corrections the user made in the CORRECTED examples.")
    if rejected_examples:
        parts.append("- Avoid the patterns seen in rejected memories.")
    parts.append("- Each memory must be self-contained: no vague pronouns (this, that, it) without referents.")
    parts.append("- Do not duplicate or closely paraphrase any existing approved memory.")

    return "\n".join(parts)

generate_message_id()

Source code in npcpy/memory/command_history.py
def generate_message_id() -> str:
    return str(uuid.uuid4())

get_available_tables(db_path_or_engine)

Gets the available tables in the database.

Source code in npcpy/memory/command_history.py
def get_available_tables(db_path_or_engine: Union[str, Engine]) -> List[Tuple[str]]:
    """
    Gets the available tables in the database.
    """
    if isinstance(db_path_or_engine, str):
        engine = create_engine_from_path(db_path_or_engine)
    else:
        engine = db_path_or_engine

    try:
        with engine.connect() as conn:
            if 'sqlite' in str(engine.url):
                result = conn.execute(text(
                    "SELECT name FROM sqlite_master WHERE type='table' AND name != 'command_history'"
                ))
            else:

                result = conn.execute(text("""
                    SELECT table_name FROM information_schema.tables 
                    WHERE table_schema = 'public' AND table_name != 'command_history'
                """))

            return [row[0] for row in result]
    except Exception as e:
        print(f"Error getting available tables: {e}")
        return []

get_db_connection(db_path='~/npcsh_history.db')

Get SQLAlchemy engine

Source code in npcpy/memory/command_history.py
def get_db_connection(db_path: str = "~/npcsh_history.db") -> Engine:
    """Get SQLAlchemy engine"""
    return create_engine_from_path(db_path)

get_npc_version_content(engine, npc_name, team_path, version=None)

Get the content of a specific NPC version. If version is None, get latest.

Source code in npcpy/memory/command_history.py
def get_npc_version_content(engine: Engine, npc_name: str, team_path: str, version: int = None) -> Optional[str]:
    """Get the content of a specific NPC version. If version is None, get latest."""
    init_kg_schema(engine)

    with engine.connect() as conn:
        if version is None:
            result = conn.execute(text("""
                SELECT content FROM npc_versions
                WHERE npc_name = :npc_name AND team_path = :team_path
                ORDER BY version DESC LIMIT 1
            """), {"npc_name": npc_name, "team_path": team_path})
        else:
            result = conn.execute(text("""
                SELECT content FROM npc_versions
                WHERE npc_name = :npc_name AND team_path = :team_path AND version = :version
            """), {"npc_name": npc_name, "team_path": team_path, "version": version})

        row = result.fetchone()
        return row[0] if row else None

get_npc_versions(engine, npc_name, team_path)

Get all versions of an NPC config.

Source code in npcpy/memory/command_history.py
def get_npc_versions(engine: Engine, npc_name: str, team_path: str) -> List[Dict[str, Any]]:
    """Get all versions of an NPC config."""
    init_kg_schema(engine)

    with engine.connect() as conn:
        result = conn.execute(text("""
            SELECT id, version, created_at, commit_message
            FROM npc_versions
            WHERE npc_name = :npc_name AND team_path = :team_path
            ORDER BY version DESC
        """), {"npc_name": npc_name, "team_path": team_path})

        return [
            {
                "id": row[0],
                "version": row[1],
                "created_at": row[2].isoformat() if row[2] else None,
                "commit_message": row[3]
            }
            for row in result
        ]

init_kg_schema(engine)

Creates the multi-scoped, path-aware KG tables using SQLAlchemy

Source code in npcpy/memory/command_history.py
def init_kg_schema(engine: Engine):
    """Creates the multi-scoped, path-aware KG tables using SQLAlchemy"""


    metadata = MetaData()

    kg_facts = Table('kg_facts', metadata,
        Column('statement', Text, nullable=False),
        Column('team_name', String(255), nullable=False),
        Column('npc_name', String(255), nullable=False),
        Column('directory_path', Text, nullable=False),
        Column('source_text', Text),
        Column('type', String(100)),
        Column('generation', Integer),
        Column('origin', String(100)),
        # FK to memory_lifecycle(id). Facts produced directly from an approved memory
        # carry the memory's id so consumers can trace back to the source memory and its
        # approval history. Nullable — facts synthesized without a source memory leave it NULL.
        Column('memory_id', Integer),
        UniqueConstraint('statement', 'team_name', 'npc_name', 'directory_path'),
    )

    kg_fact_sources = Table('kg_fact_sources', metadata,
        Column('statement', Text, nullable=False),
        Column('team_name', String(255), nullable=False),
        Column('npc_name', String(255), nullable=False),
        Column('directory_path', Text, nullable=False),
        Column('conversation_id', String(255), nullable=False),
        Column('message_id', String(255), nullable=False),
        UniqueConstraint('statement', 'team_name', 'npc_name', 'directory_path', 'conversation_id', 'message_id'),
        schema=None
    )

    kg_concepts = Table('kg_concepts', metadata,
        Column('name', Text, nullable=False),
        Column('team_name', String(255), nullable=False),
        Column('npc_name', String(255), nullable=False),
        Column('directory_path', Text, nullable=False),
        Column('generation', Integer),
        Column('origin', String(100)),
        UniqueConstraint('name', 'team_name', 'npc_name', 'directory_path'),
        schema=None
    )

    kg_links = Table('kg_links', metadata,
        Column('source', Text, nullable=False),
        Column('target', Text, nullable=False),
        Column('team_name', String(255), nullable=False),
        Column('npc_name', String(255), nullable=False),
        Column('directory_path', Text, nullable=False),
        Column('type', String(100), nullable=False),
        schema=None
    )

    kg_metadata = Table('kg_metadata', metadata,
        Column('key', String(255), nullable=False),
        Column('team_name', String(255), nullable=False),
        Column('npc_name', String(255), nullable=False),
        Column('directory_path', Text, nullable=False),
        Column('value', Text),
        schema=None
    )

    npc_versions = Table('npc_versions', metadata,
        Column('id', Integer, primary_key=True, autoincrement=True),
        Column('npc_name', String(255), nullable=False),
        Column('team_path', Text, nullable=False),
        Column('version', Integer, nullable=False),
        Column('content', Text, nullable=False),
        Column('created_at', DateTime, default=datetime.utcnow),
        Column('commit_message', Text),
        schema=None
    )

    metadata.create_all(engine, checkfirst=True)

    # Idempotent column add for DBs that predate memory_id on kg_facts.
    try:
        with engine.begin() as conn:
            cols = [r[1] for r in conn.execute(text("PRAGMA table_info(kg_facts)")).fetchall()]
            if 'memory_id' not in cols:
                conn.execute(text("ALTER TABLE kg_facts ADD COLUMN memory_id INTEGER"))
    except Exception:
        # Non-SQLite engines use create_all's DDL directly; migration not required there.
        pass

list_files(base_dir, table, ext='csv', limit=100)

Source code in npcpy/memory/command_history.py
def list_files(base_dir, table, ext="csv", limit=100):
    import pathlib
    base = pathlib.Path(base_dir) / table
    if not base.exists():
        return []
    files = sorted(base.rglob(f"*.{ext}"), key=lambda p: str(p), reverse=True)
    return [{"group_id": f.stem, "file_path": str(f)} for f in files[:limit]]

load_file_csv(base_dir, table, group_id)

Source code in npcpy/memory/command_history.py
def load_file_csv(base_dir, table, group_id):
    import csv as csv_mod
    import pathlib
    base = pathlib.Path(base_dir) / table
    if not base.exists():
        return []
    target = f"{group_id}.csv"
    for f in base.rglob(target):
        with open(f, "r", encoding="utf-8") as fh:
            return list(csv_mod.DictReader(fh))
    return []

load_file_parquet(base_dir, table, group_id)

Source code in npcpy/memory/command_history.py
def load_file_parquet(base_dir, table, group_id):
    import polars as pl
    import pathlib
    base = pathlib.Path(base_dir) / table
    if not base.exists():
        return []
    target = f"{group_id}.parquet"
    for f in base.rglob(target):
        return pl.read_parquet(f).to_dicts()
    return []

load_kg_from_db(engine, team_name, npc_name, directory_path)

Loads the KG for a specific scope (team, npc, path) from database.

Source code in npcpy/memory/command_history.py
def load_kg_from_db(engine: Engine, team_name: str, npc_name: str, directory_path: str) -> Dict[str, Any]:
    """Loads the KG for a specific scope (team, npc, path) from database."""
    kg = {
        "generation": 0,
        "facts": [],
        "concepts": [],
        "concept_links": [],
        "fact_to_concept_links": {},
        "fact_to_fact_links": []
    }

    with engine.connect() as conn:
        try:

            result = conn.execute(text("""
                SELECT value FROM kg_metadata 
                WHERE team_name = :team AND npc_name = :npc AND directory_path = :path AND key = 'generation'
            """), {"team": team_name, "npc": npc_name, "path": directory_path})

            row = result.fetchone()
            if row:
                kg['generation'] = int(row.value)


            result = conn.execute(text("""
                SELECT f.statement, f.source_text, f.type, f.generation, f.origin,
                       COUNT(s.conversation_id) as weight
                FROM kg_facts f
                LEFT JOIN kg_fact_sources s
                  ON f.statement = s.statement
                  AND f.team_name = s.team_name
                  AND f.npc_name = s.npc_name
                  AND f.directory_path = s.directory_path
                WHERE f.team_name = :team AND f.npc_name = :npc AND f.directory_path = :path
                GROUP BY f.statement, f.source_text, f.type, f.generation, f.origin
            """), {"team": team_name, "npc": npc_name, "path": directory_path})

            kg['facts'] = [
                {
                    "statement": row.statement,
                    "source_text": row.source_text,
                    "type": row.type,
                    "generation": row.generation,
                    "origin": row.origin,
                    "weight": max(1, row.weight)
                }
                for row in result
            ]


            result = conn.execute(text("""
                SELECT name, generation, origin FROM kg_concepts 
                WHERE team_name = :team AND npc_name = :npc AND directory_path = :path
            """), {"team": team_name, "npc": npc_name, "path": directory_path})

            kg['concepts'] = [
                {"name": row.name, "generation": row.generation, "origin": row.origin}
                for row in result
            ]


            links = {}
            result = conn.execute(text("""
                SELECT source, target, type FROM kg_links 
                WHERE team_name = :team AND npc_name = :npc AND directory_path = :path
            """), {"team": team_name, "npc": npc_name, "path": directory_path})

            for row in result:
                if row.type == 'fact_to_concept':
                    if row.source not in links:
                        links[row.source] = []
                    links[row.source].append(row.target)
                elif row.type == 'concept_to_concept':
                    kg['concept_links'].append((row.source, row.target))
                elif row.type == 'fact_to_fact':
                    kg['fact_to_fact_links'].append((row.source, row.target))

            kg['fact_to_concept_links'] = links

        except SQLAlchemyError:

            init_kg_schema(engine)

    return kg

normalize_path_for_db(path_str)

Normalize a path for consistent database storage. Converts backslashes to forward slashes for cross-platform compatibility.

Source code in npcpy/memory/command_history.py
def normalize_path_for_db(path_str):
    """
    Normalize a path for consistent database storage.
    Converts backslashes to forward slashes for cross-platform compatibility.
    """
    if not path_str:
        return path_str
    normalized = path_str.replace('\\', '/')
    normalized = normalized.rstrip('/')
    return normalized

query_history_for_llm(command_history, query)

Source code in npcpy/memory/command_history.py
def query_history_for_llm(command_history, query):
    results = command_history.search_commands(query)
    formatted_results = [
        f"Command: {r['command']}\nOutput: {r['output']}\nLocation: {r['location']}" for r in results
    ]
    return "\n\n".join(formatted_results)

retrieve_last_conversation(command_history, conversation_id)

Retrieves and formats all messages from the last conversation.

Source code in npcpy/memory/command_history.py
def retrieve_last_conversation(
    command_history: CommandHistory, conversation_id: str
    ) -> str:
    """
    Retrieves and formats all messages from the last conversation.
    """
    last_message = command_history.get_last_conversation(conversation_id)
    if last_message:
        return last_message['content']  
    return "No previous conversation messages found."

rollback_npc_to_version(engine, npc_name, team_path, version)

Rollback an NPC to a specific version. Returns the content if successful.

Source code in npcpy/memory/command_history.py
def rollback_npc_to_version(engine: Engine, npc_name: str, team_path: str, version: int) -> Optional[str]:
    """Rollback an NPC to a specific version. Returns the content if successful."""
    content = get_npc_version_content(engine, npc_name, team_path, version)
    if content is None:
        return None

    save_npc_version(engine, npc_name, team_path, content, f"Rollback to version {version}")

    file_path = os.path.join(team_path, f"{npc_name}.npc")
    with open(file_path, 'w') as f:
        f.write(content)

    return content

save_attachment_to_message(command_history, message_id, file_path, attachment_name=None, attachment_type=None)

Helper function to save a file from disk as an attachment.

Source code in npcpy/memory/command_history.py
def save_attachment_to_message(
    command_history: CommandHistory,
    message_id: str,
    file_path: str,
    attachment_name: str = None,
    attachment_type: str = None,
    ):
    """
    Helper function to save a file from disk as an attachment.
    """
    try:
        if not attachment_name:
            attachment_name = os.path.basename(file_path)

        if not attachment_type:
            _, ext = os.path.splitext(file_path)
            if ext:
                attachment_type = ext.lower()[1:]

        with open(file_path, "rb") as f:
            data = f.read()

        command_history.add_attachment(
            message_id=message_id,
            attachment_name=attachment_name,
            attachment_type=attachment_type,
            attachment_data=data,
            attachment_size=len(data),
        )
        return True
    except Exception as e:
        print(f"Error saving attachment: {str(e)}")
        return False

save_conversation_message(command_history, conversation_id, role, content, wd=None, model=None, provider=None, npc=None, team=None, attachments=None, message_id=None, reasoning_content=None, tool_calls=None, tool_results=None, parent_message_id=None, skip_if_exists=True, device_id=None, device_name=None, gen_params=None, input_tokens=None, output_tokens=None, cost=None)

Saves a conversation message linked to a conversation ID with optional attachments. Now also supports reasoning_content, tool_calls, tool_results, parent_message_id for broadcast grouping, and gen_params for temperature, top_p, top_k, max_tokens, etc. If skip_if_exists is True and message_id already exists, skip saving to prevent duplicates.

Source code in npcpy/memory/command_history.py
def save_conversation_message(
    command_history: CommandHistory,
    conversation_id: str,
    role: str,
    content: str,
    wd: str = None,
    model: str = None,
    provider: str = None,
    npc: str = None,
    team: str = None,
    attachments: List[Dict] = None,
    message_id: str = None,
    reasoning_content: str = None,
    tool_calls: List[Dict] = None,
    tool_results: List[Dict] = None,
    parent_message_id: str = None,
    skip_if_exists: bool = True,
    device_id: str = None,
    device_name: str = None,
    gen_params: Dict = None,
    input_tokens: int = None,
    output_tokens: int = None,
    cost: float = None,
    ):
    """
    Saves a conversation message linked to a conversation ID with optional attachments.
    Now also supports reasoning_content, tool_calls, tool_results, parent_message_id for broadcast grouping,
    and gen_params for temperature, top_p, top_k, max_tokens, etc.
    If skip_if_exists is True and message_id already exists, skip saving to prevent duplicates.
    """
    if wd is None:
        wd = os.getcwd()
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    if message_id is None:
        message_id = generate_message_id()

    if skip_if_exists and message_id:
        existing = command_history.get_message_by_id(message_id)
        if existing:
            print(f"[SAVE_MSG] Skipping save - message_id {message_id} already exists")
            return None

    return command_history.add_conversation(
        message_id,
        timestamp,
        role,
        content,
        conversation_id,
        wd,
        model=model,
        provider=provider,
        npc=npc,
        team=team,
        attachments=attachments,
        reasoning_content=reasoning_content,
        tool_calls=tool_calls,
        tool_results=tool_results,
        parent_message_id=parent_message_id,
        device_id=device_id,
        device_name=device_name,
        gen_params=gen_params,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cost=cost)

save_kg_to_db(engine, kg_data, team_name, npc_name, directory_path, conversation_id=None, message_id=None)

Saves a knowledge graph dictionary to the database. Tracks which conversations/messages produced each fact.

Source code in npcpy/memory/command_history.py
def save_kg_to_db(engine: Engine, kg_data: Dict[str, Any], team_name: str, npc_name: str, directory_path: str,
                  conversation_id: str = None, message_id: str = None):
    """Saves a knowledge graph dictionary to the database. Tracks which conversations/messages produced each fact."""
    try:
        with engine.begin() as conn:

            facts_to_save = [
                {
                    "statement": fact['statement'],
                    "team_name": team_name,
                    "npc_name": npc_name,
                    "directory_path": directory_path,
                    "source_text": fact.get('source_text', ''),
                    "type": fact.get('type', ''),
                    "generation": fact.get('generation', 0),
                    "origin": fact.get('origin', 'organic'),
                    "memory_id": fact.get('memory_id')
                }
                for fact in kg_data.get("facts", [])
            ]

            if facts_to_save:
                if 'sqlite' in str(engine.url):
                    stmt = text("""
                        INSERT OR IGNORE INTO kg_facts
                        (statement, team_name, npc_name, directory_path, source_text, type, generation, origin, memory_id)
                        VALUES (:statement, :team_name, :npc_name, :directory_path, :source_text, :type, :generation, :origin, :memory_id)
                    """)
                else:
                    stmt = text("""
                        INSERT INTO kg_facts
                        (statement, team_name, npc_name, directory_path, source_text, type, generation, origin, memory_id)
                        VALUES (:statement, :team_name, :npc_name, :directory_path, :source_text, :type, :generation, :origin, :memory_id)
                        ON CONFLICT (statement, team_name, npc_name, directory_path) DO UPDATE SET
                            memory_id = COALESCE(EXCLUDED.memory_id, kg_facts.memory_id)
                    """)

                for fact in facts_to_save:
                    conn.execute(stmt, fact)

                # Record which conversation/message produced each fact
                if conversation_id and message_id:
                    src_stmt = text("""
                        INSERT OR IGNORE INTO kg_fact_sources
                        (statement, team_name, npc_name, directory_path, conversation_id, message_id)
                        VALUES (:statement, :team_name, :npc_name, :directory_path, :conversation_id, :message_id)
                    """) if 'sqlite' in str(engine.url) else text("""
                        INSERT INTO kg_fact_sources
                        (statement, team_name, npc_name, directory_path, conversation_id, message_id)
                        VALUES (:statement, :team_name, :npc_name, :directory_path, :conversation_id, :message_id)
                        ON CONFLICT DO NOTHING
                    """)
                    for fact in facts_to_save:
                        conn.execute(src_stmt, {
                            "statement": fact['statement'],
                            "team_name": team_name,
                            "npc_name": npc_name,
                            "directory_path": directory_path,
                            "conversation_id": conversation_id,
                            "message_id": message_id
                        })


            concepts_to_save = [
                {
                    "name": concept['name'],
                    "team_name": team_name,
                    "npc_name": npc_name,
                    "directory_path": directory_path,
                    "generation": concept.get('generation', 0),
                    "origin": concept.get('origin', 'organic')
                }
                for concept in kg_data.get("concepts", [])
            ]

            if concepts_to_save:
                if 'sqlite' in str(engine.url):
                    stmt = text("""
                        INSERT OR IGNORE INTO kg_concepts 
                        (name, team_name, npc_name, directory_path, generation, origin)
                        VALUES (:name, :team_name, :npc_name, :directory_path, :generation, :origin)
                    """)
                else:
                    stmt = text("""
                        INSERT INTO kg_concepts 
                        (name, team_name, npc_name, directory_path, generation, origin)
                        VALUES (:name, :team_name, :npc_name, :directory_path, :generation, :origin)
                        ON CONFLICT (name, team_name, npc_name, directory_path) DO NOTHING
                    """)

                for concept in concepts_to_save:
                    conn.execute(stmt, concept)


            if 'sqlite' in str(engine.url):
                stmt = text("""
                    INSERT OR REPLACE INTO kg_metadata (key, value, team_name, npc_name, directory_path)
                    VALUES ('generation', :generation, :team_name, :npc_name, :directory_path)
                """)
            else:
                stmt = text("""
                    INSERT INTO kg_metadata (key, value, team_name, npc_name, directory_path)
                    VALUES ('generation', :generation, :team_name, :npc_name, :directory_path)
                    ON CONFLICT (key, team_name, npc_name, directory_path) 
                    DO UPDATE SET value = EXCLUDED.value
                """)

            conn.execute(stmt, {
                "generation": str(kg_data.get('generation', 0)),
                "team_name": team_name,
                "npc_name": npc_name,
                "directory_path": directory_path
            })


            conn.execute(text("""
                DELETE FROM kg_links 
                WHERE team_name = :team_name AND npc_name = :npc_name AND directory_path = :directory_path
            """), {"team_name": team_name, "npc_name": npc_name, "directory_path": directory_path})


            for fact, concepts in kg_data.get("fact_to_concept_links", {}).items():
                for concept in concepts:
                    conn.execute(text("""
                        INSERT INTO kg_links (source, target, type, team_name, npc_name, directory_path)
                        VALUES (:source, :target, 'fact_to_concept', :team_name, :npc_name, :directory_path)
                    """), {
                        "source": fact, "target": concept,
                        "team_name": team_name, "npc_name": npc_name, "directory_path": directory_path
                    })

            for c1, c2 in kg_data.get("concept_links", []):
                conn.execute(text("""
                    INSERT INTO kg_links (source, target, type, team_name, npc_name, directory_path)
                    VALUES (:source, :target, 'concept_to_concept', :team_name, :npc_name, :directory_path)
                """), {
                    "source": c1, "target": c2,
                    "team_name": team_name, "npc_name": npc_name, "directory_path": directory_path
                })

            for f1, f2 in kg_data.get("fact_to_fact_links", []):
                conn.execute(text("""
                    INSERT INTO kg_links (source, target, type, team_name, npc_name, directory_path)
                    VALUES (:source, :target, 'fact_to_fact', :team_name, :npc_name, :directory_path)
                """), {
                    "source": f1, "target": f2,
                    "team_name": team_name, "npc_name": npc_name, "directory_path": directory_path
                })

    except Exception as e:
        print(f"Failed to save KG for scope '({team_name}, {npc_name}, {directory_path})': {e}")

save_npc_version(engine, npc_name, team_path, content, commit_message=None)

Save a new version of an NPC config. Returns the new version number.

Source code in npcpy/memory/command_history.py
def save_npc_version(engine: Engine, npc_name: str, team_path: str, content: str, commit_message: str = None) -> int:
    """Save a new version of an NPC config. Returns the new version number."""
    init_kg_schema(engine)

    with engine.begin() as conn:
        result = conn.execute(text("""
            SELECT COALESCE(MAX(version), 0) as max_version
            FROM npc_versions
            WHERE npc_name = :npc_name AND team_path = :team_path
        """), {"npc_name": npc_name, "team_path": team_path})
        row = result.fetchone()
        new_version = (row[0] if row else 0) + 1

        conn.execute(text("""
            INSERT INTO npc_versions (npc_name, team_path, version, content, created_at, commit_message)
            VALUES (:npc_name, :team_path, :version, :content, :created_at, :commit_message)
        """), {
            "npc_name": npc_name,
            "team_path": team_path,
            "version": new_version,
            "content": content,
            "created_at": datetime.utcnow(),
            "commit_message": commit_message
        })

    return new_version

scan_all(base_dir, table, ext='csv')

Source code in npcpy/memory/command_history.py
def scan_all(base_dir, table, ext="csv"):
    import pathlib
    base = pathlib.Path(base_dir) / table
    if not base.exists():
        return None
    files = list(base.rglob(f"*.{ext}"))
    if not files:
        return None
    try:
        import polars as pl
    except ImportError:
        return None
    dfs = []
    for f in files:
        try:
            if ext == "csv":
                dfs.append(pl.read_csv(f))
            else:
                dfs.append(pl.read_parquet(f))
        except Exception:
            pass
    return pl.concat(dfs) if dfs else pl.DataFrame()

search_files(base_dir, table, query, column='content', ext='csv')

Source code in npcpy/memory/command_history.py
def search_files(base_dir, table, query, column="content", ext="csv"):
    import csv as csv_mod
    import pathlib
    base = pathlib.Path(base_dir) / table
    if not base.exists():
        return []
    q = query.lower()
    results = []
    for f in base.rglob(f"*.{ext}"):
        if ext == "csv":
            with open(f, "r", encoding="utf-8") as fh:
                for row in csv_mod.DictReader(fh):
                    if q in (row.get(column) or "").lower():
                        results.append(row)
        elif ext == "parquet":
            try:
                import polars as pl
                df = pl.read_parquet(f)
                if column in df.columns:
                    matches = df.filter(pl.col(column).str.contains(query, literal=True))
                    results.extend(matches.to_dicts())
            except Exception:
                pass
    return results

setup_chroma_db(collection, description='', db_path='')

Initialize Chroma vector database without a default embedding function

Source code in npcpy/memory/command_history.py
def setup_chroma_db(collection, description='', db_path: str = ''):
    """Initialize Chroma vector database without a default embedding function"""
    if db_path == '':
        db_path = os.path.expanduser('~/npcsh_chroma_db')

    try:
        client = chromadb.PersistentClient(path=db_path)

        try:
            collection = client.get_collection(collection)
            print("Connected to existing facts collection")
        except ValueError:
            collection = client.create_collection(
                name=collection,
                metadata={"description": description},
            )
            print("Created new facts collection")

        return client, collection
    except Exception as e:
        print(f"Error setting up Chroma DB: {e}")
        raise

show_history(command_history, args)

Source code in npcpy/memory/command_history.py
def show_history(command_history, args):
    if args:
        search_results = command_history.search_commands(args[0])
        if search_results:
            return "\n".join(
                [f"{item['id']}. [{item['timestamp']}] {item['command']}" for item in search_results]
            )
        else:
            return f"No commands found matching '{args[0]}'"
    else:
        all_history = command_history.get_all_commands()
        return "\n".join([f"{item['id']}. [{item['timestamp']}] {item['command']}" for item in all_history])

start_new_conversation(prepend=None)

Starts a new conversation and returns a unique conversation ID.

Source code in npcpy/memory/command_history.py
def start_new_conversation(prepend: str = None) -> str:
    """
    Starts a new conversation and returns a unique conversation ID.
    """
    if prepend is None:
        prepend = 'npcsh'
    return f"{prepend}_{datetime.now().strftime('%Y%m%d%H%M%S')}"