haive.agents.memory.kg_store
============================

.. py:module:: haive.agents.memory.kg_store

.. autoapi-nested-parse::

   Knowledge Graph store integration for MemoryAgent.

   Provides Neo4j-backed KG storage that works alongside the LangGraph store.
   Triples stored in both:
   - LangGraph store (for vector search/retrieval via memory tools)
   - Neo4j (for graph traversal, Cypher queries, APOC algorithms)

   Uses same env vars as graph_db RAG: NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD.



Classes
-------

.. autoapisummary::

   haive.agents.memory.kg_store.Neo4jKGConfig
   haive.agents.memory.kg_store.Neo4jKGStore


Module Contents
---------------

.. py:class:: Neo4jKGConfig(/, **data)

   Bases: :py:obj:`pydantic.BaseModel`


   Configuration for Neo4j knowledge graph connection.

   Uses same env vars as haive.agents.rag.db_rag.graph_db for consistency:
   NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD, NEO4J_DATABASE

   Create a new model by parsing and validating input data from keyword arguments.

   Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be
   validated to form a valid model.

   `self` is explicitly positional-only to allow `self` as a field name.


   .. py:method:: model_post_init(__context)

      Resolve from env vars if not provided.



.. py:class:: Neo4jKGStore(config = None)

   Neo4j-backed knowledge graph store for MemoryAgent.

   Stores KG triples in Neo4j for graph traversal while keeping them
   in the LangGraph store for vector search. Provides Cypher-based
   queries for relationship exploration, path finding, and neighborhood
   analysis.

   Usage:
       kg = Neo4jKGStore(Neo4jKGConfig())
       kg.save_triple("Will", "works_at", "Anthropic", user_id="will")
       kg.save_triple("Will", "uses", "Python", user_id="will")
       triples = kg.query_entity("Will")
       neighbors = kg.query_neighborhood("Will")
       path = kg.query_path("Will", "Python")
       kg.close()


   .. py:method:: close()

      Close the Neo4j driver.



   .. py:method:: create_kg_from_memories(store, user_id = 'default')

      Build KG in Neo4j from existing KG triples in the LangGraph store.

      Reads all KG triples from store namespace ("kg", user_id) and
      creates them in Neo4j for graph traversal.

      :param store: LangGraph BaseStore with existing triples
      :param user_id: User to extract triples for

      :returns: Number of triples synced to Neo4j



   .. py:method:: get_stats()

      Get graph statistics.



   .. py:method:: query_cypher(cypher, params = None)

      Execute a raw Cypher query.

      :param cypher: Cypher query string
      :param params: Query parameters

      :returns: List of result records as dicts



   .. py:method:: query_entity(entity)

      Get all triples involving an entity.

      :param entity: Entity name to query

      :returns: List of {subject, predicate, object} dicts



   .. py:method:: query_neighborhood(entity, limit = 20)

      Query 1-2 hop neighborhood of an entity.

      :param entity: Center entity name
      :param limit: Max results

      :returns: List of neighborhood records



   .. py:method:: query_path(start, end)

      Find shortest path between two entities.

      :param start: Start entity name
      :param end: End entity name

      :returns: List of path records with entities, predicates, hops



   .. py:method:: query_user_triples(user_id, limit = 50)

      Get all triples for a user.

      :param user_id: User to query
      :param limit: Max results

      :returns: List of {subject, predicate, object} dicts



   .. py:method:: save_triple(subject, predicate, obj, user_id = 'default', subject_type = 'Entity', object_type = 'Entity', source = 'memory_agent')

      Save a KG triple to Neo4j using MERGE (upsert).

      :param subject: Source entity name
      :param predicate: Relationship type
      :param obj: Target entity name
      :param user_id: User scope
      :param subject_type: Label for source node
      :param object_type: Label for target node
      :param source: Where this triple came from



   .. py:method:: save_triples_batch(triples, user_id = 'default')

      Save multiple triples efficiently.

      :param triples: List of {subject, predicate, object} dicts
      :param user_id: User scope

      :returns: Number of triples saved