Frequently Asked Questions

# Via environment variables
export NAM_EMBEDDING__PROVIDER=bedrock
export NAM_EMBEDDING__AWS_REGION=us-east-1
export NAM_EMBEDDING__AWS_PROFILE=my-profile

from neo4j_agent_memory import MemorySettings

settings = MemorySettings(
    embedding={"provider": "bedrock", "aws_region": "us-east-1"}
)

Text → [spaCy] → [GLiNER] → [LLM] → Merge → Filter → Results

from neo4j_agent_memory.extraction import (
    SpacyEntityExtractor,
    GLiNEREntityExtractor,
    LLMEntityExtractor,
)

# Just spaCy
extractor = SpacyEntityExtractor()

# Just GLiNER
extractor = GLiNEREntityExtractor.for_schema("podcast")

# Just LLM
extractor = LLMEntityExtractor(model="gpt-4o-mini")

result = await extractor.extract("Your text here")

extractor = GLiNEREntityExtractor(entity_labels=["person", "company"])

# Schema includes detailed descriptions
extractor = GLiNEREntityExtractor.for_schema("podcast")
# {"person": "A person mentioned in the podcast, including hosts, guests...", ...}

extractor = (
    ExtractorBuilder()
    .with_spacy()                              # Uses spaCy's built-in types
    .with_gliner_schema("podcast", threshold=0.5)  # Uses podcast schema
    .with_llm_fallback()                       # Uses entity type names
    .merge_by_confidence()
    .build()
)

from neo4j_agent_memory.extraction.domain_schemas import DomainSchema

my_schema = DomainSchema(
    name="real_estate",
    entity_types={
        "property": "A real estate property, building, or land parcel",
        "agent": "A real estate agent or broker",
        "price": "A property price, asking price, or sale price",
        "neighborhood": "A neighborhood, district, or area",
    },
)

extractor = GLiNEREntityExtractor(schema=my_schema)

extractor.add_label_mapping("podcast_episode", "EVENT", "PODCAST_EPISODE")

from neo4j_agent_memory.extraction import ExtractionPipeline

pipeline = ExtractionPipeline()

# Extract from multiple texts in parallel
results = await pipeline.extract_batch(
    texts=["Document 1...", "Document 2...", "Document 3..."],
    batch_size=10,           # Process 10 at a time
    on_progress=lambda done, total: print(f"{done}/{total}"),
)

from neo4j_agent_memory.extraction import StreamingExtractor

extractor = StreamingExtractor(
    chunk_size=4000,      # Tokens per chunk
    chunk_overlap=200,    # Overlap to preserve context
)

# Stream results as they're extracted
async for partial_result in extractor.stream(long_document):
    print(f"Found {len(partial_result.entities)} entities in chunk")

# Or get all results with automatic deduplication
result = await extractor.extract(long_document)

from neo4j_agent_memory.extraction import (
    ExtractorBuilder,
    GLiNEREntityExtractor,
)

# Enable GLiREL for relation extraction
extractor = (
    ExtractorBuilder()
    .with_gliner_schema("podcast")
    .with_gliner_relations(threshold=0.3)  # GLiREL for relations
    .build()
)

result = await extractor.extract(text)
# result.relations populated without LLM calls

from neo4j_agent_memory import MemorySettings, DeduplicationConfig

settings = MemorySettings(
    deduplication=DeduplicationConfig(
        enabled=True,
        embedding_threshold=0.92,     # High similarity = auto-merge
        same_as_threshold=0.85,       # Medium similarity = SAME_AS link
        create_same_as=True,
    )
)

// Find potential duplicates
MATCH (e1:Entity)-[r:SAME_AS]->(e2:Entity)
RETURN e1.name, e2.name, r.confidence
ORDER BY r.confidence DESC

// Find all people
MATCH (p:Person) RETURN p

// Find all vehicles
MATCH (v:Vehicle) RETURN v

// Find organizations in a specific location
MATCH (o:Organization)-[:LOCATED_IN]->(l:Location {name: "New York"})
RETURN o

pip install neo4j-agent-memory[cli]

# Extract entities from text
neo4j-memory extract "John works at Acme Corp in New York"

# Extract with specific schema
neo4j-memory extract --schema podcast "..."

# Extract from file
neo4j-memory extract --input document.txt

# Output as table
neo4j-memory extract --format table "..."

# List available schemas
neo4j-memory schemas list

# Get statistics
neo4j-memory stats

# .neo4j-memory.yaml
neo4j:
  uri: bolt://localhost:7687
  username: neo4j
  password: your-password

extraction:
  extractor_type: pipeline
  enable_gliner: true

cli:
  output_format: json

from neo4j_agent_memory import (
    MemorySettings,
    GeocodingConfig,
    GeocodingProvider,
)

settings = MemorySettings(
    geocoding=GeocodingConfig(
        enabled=True,
        provider=GeocodingProvider.NOMINATIM,  # Free, 1 req/sec limit
    )
)

NAM_GEOCODING__ENABLED=true
NAM_GEOCODING__PROVIDER=nominatim

from pydantic import SecretStr

settings = MemorySettings(
    geocoding=GeocodingConfig(
        enabled=True,
        provider=GeocodingProvider.GOOGLE,
        api_key=SecretStr("your-google-api-key"),
    )
)

async with MemoryClient(settings) as memory:
    stats = await memory.long_term.geocode_locations(
        batch_size=50,
        skip_existing=True,  # Don't re-geocode
    )
    print(f"Geocoded {stats['geocoded']} of {stats['processed']} locations")

# Find locations near a point
nearby = await memory.long_term.search_locations_near(
    latitude=40.7128,
    longitude=-74.0060,
    radius_km=10.0,
    limit=20,
)

# Find locations within a bounding box
in_area = await memory.long_term.search_locations_in_bounds(
    min_lat=40.70,
    max_lat=40.80,
    min_lon=-74.02,
    max_lon=-73.95,
)

# Get coordinates for a specific location
coords = await memory.long_term.get_entity_coordinates(entity_id)

entity, _ = await memory.long_term.add_entity(
    name="Custom Location",
    entity_type="LOCATION",
    geocode=False,  # Skip geocoding
    coordinates=(40.7128, -74.0060),  # Provide coordinates directly
)

# For OpenTelemetry
pip install neo4j-agent-memory[opentelemetry]

# For Opik (LLM-focused)
pip install neo4j-agent-memory[opik]

from neo4j_agent_memory import MemorySettings, ObservabilityConfig, TracingProvider

settings = MemorySettings(
    observability=ObservabilityConfig(
        enabled=True,
        provider=TracingProvider.OPIK,  # or OPENTELEMETRY
        project_name="my-agent",
    )
)

import hashlib
from functools import lru_cache

@lru_cache(maxsize=1000)
async def cached_extract(text_hash: str, text: str):
    return await extractor.extract(text)

# Use hash as cache key
text_hash = hashlib.md5(text.encode()).hexdigest()
result = await cached_extract(text_hash, text)

python -m spacy download en_core_web_sm

from neo4j_agent_memory.extraction import is_valid_entity_name, ENTITY_STOPWORDS

is_valid_entity_name("your entity")  # Check if it passes
print("they" in ENTITY_STOPWORDS)    # Check specific word

pip install neo4j-agent-memory[gliner]
# or
uv sync --all-extras

from neo4j_agent_memory.extraction import is_gliner_available

if is_gliner_available():
    extractor = GLiNEREntityExtractor.for_schema("podcast")
else:
    print("GLiNER not available, using spaCy only")
    extractor = SpacyEntityExtractor()