haive.agents.memory.kg_store¶

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¶

Neo4jKGConfig

Configuration for Neo4j knowledge graph connection.

Neo4jKGStore

Neo4j-backed knowledge graph store for MemoryAgent.

Module Contents¶

class haive.agents.memory.kg_store.Neo4jKGConfig(/, **data)¶

Bases: 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.

Parameters:

data (Any)

model_post_init(__context)¶

Resolve from env vars if not provided.

class haive.agents.memory.kg_store.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()

Parameters:

config (Neo4jKGConfig | None)

close()¶

Close the Neo4j driver.

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.

Parameters:

  • store (Any) – LangGraph BaseStore with existing triples

  • user_id (str) – User to extract triples for

Returns:

Number of triples synced to Neo4j

Return type:

int

get_stats()¶

Get graph statistics.

Return type:

dict

query_cypher(cypher, params=None)¶

Execute a raw Cypher query.

Parameters:

  • cypher (str) – Cypher query string

  • params (dict | None) – Query parameters

Returns:

List of result records as dicts

Return type:

list[dict]

query_entity(entity)¶

Get all triples involving an entity.

Parameters:

entity (str) – Entity name to query

Returns:

List of {subject, predicate, object} dicts

Return type:

list[dict]

query_neighborhood(entity, limit=20)¶

Query 1-2 hop neighborhood of an entity.

Parameters:

  • entity (str) – Center entity name

  • limit (int) – Max results

Returns:

List of neighborhood records

Return type:

list[dict]

query_path(start, end)¶

Find shortest path between two entities.

Parameters:

  • start (str) – Start entity name

  • end (str) – End entity name

Returns:

List of path records with entities, predicates, hops

Return type:

list[dict]

query_user_triples(user_id, limit=50)¶

Get all triples for a user.

Parameters:

  • user_id (str) – User to query

  • limit (int) – Max results

Returns:

List of {subject, predicate, object} dicts

Return type:

list[dict]

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).

Parameters:

  • subject (str) – Source entity name

  • predicate (str) – Relationship type

  • obj (str) – Target entity name

  • user_id (str) – User scope

  • subject_type (str) – Label for source node

  • object_type (str) – Label for target node

  • source (str) – Where this triple came from

Return type:

bool

save_triples_batch(triples, user_id='default')¶

Save multiple triples efficiently.

Parameters:

  • triples (list[dict]) – List of {subject, predicate, object} dicts

  • user_id (str) – User scope

Returns:

Number of triples saved

Return type:

int