Usage#

This is a hands-on guide to langchain-agensgraph 0.3.1. It assumes you have completed Installation: a running AgensGraph 2.17 or later and the pgvector extension. The examples that embed text or call an LLM need langchain-openai and OPENAI_API_KEY. The others run without either.

Every component is importable from the top-level package:

from langchain_agensgraph import (
    AgensGraph, AgensgraphVector, AgensEngine,
    AgensChatMessageHistory, LLMGraphTransformer,
    AgensVectorRetriever, AgensGraphContextRetriever, AgensText2CypherRetriever,
    AgensCypherQAChain, create_cypher_tool, log_queries,
    GraphDocument, Node, Relationship,
)
from langchain_agensgraph import AgensSaver, AsyncAgensSaver, AgensStore

(AgensSaver, AsyncAgensSaver, and AgensStore are exported only when langgraph can be imported. It is a dependency of the package, so it normally is.)

AgensGraph Cypher tip. AgensGraph folds unquoted identifiers to lower case, so the examples double-quote labels and relationship types to preserve their casing — (m:"Movie"), -[:"ACTED_IN"]->. Most hot paths also have an async counterpart (aquery, asimilarity_search, aadd_texts, aclose, …).

Quickstart#

Connect to a graph, write a little Cypher, and read it back. The conf dict takes psycopg.connect keyword arguments; create=True makes the graph if it doesn’t exist.

from langchain_agensgraph import AgensGraph

conf = {
    "dbname": "mydb",
    "user": "postgres",
    "password": "postgres",
    "host": "localhost",
    "port": 55432,
}

graph = AgensGraph(graph_name="moviegraph", conf=conf, create=True)

graph.query("""
    MERGE (m:"Movie" {name: 'Top Gun'})
    WITH m
    UNWIND ['Tom Cruise', 'Val Kilmer', 'Meg Ryan'] AS actor
    MERGE (a:"Actor" {name: actor})
    MERGE (a)-[:"ACTED_IN"]->(m)
""")

rows = graph.query("""
    MATCH (a:"Actor")-[:"ACTED_IN"]->(:"Movie" {name: 'Top Gun'})
    RETURN a.name AS actor ORDER BY actor
""")
print([r["actor"] for r in rows])    # ['Meg Ryan', 'Tom Cruise', 'Val Kilmer']

Graph store#

AgensGraph.query(query, params=None, timeout=None) runs Cypher and returns a list of dicts, one per row. A vertex comes back as the driver’s agensgraph.Vertex, an edge as an Edge, and a path as a Path, each with its label, id, and properties kept apart. aquery(...) is the async form.

rows = graph.query('MATCH (a:"Actor")-[r:"ACTED_IN"]->(m:"Movie") RETURN a, r LIMIT 1')
actor, edge = rows[0]["a"], rows[0]["r"]
print(actor.label, actor.properties)   # Actor {'name': 'Tom Cruise'}
print(edge.label, edge.start == actor.id)   # ACTED_IN True

# async
import asyncio

async def count():
    rows = await graph.aquery("MATCH (n) RETURN count(n) AS c")
    await graph.aclose()                         # closes the async connection only
    return rows[0]["c"]

print(asyncio.run(count()))

Parameters. Placeholders use psycopg’s %(name)s style. A str parameter is sent as text, so pass the string itself. Wrapping it in json.dumps(...), which 0.2.0 needed, now makes the query match nothing. A dict is sent as JSONB. A bare list is sent as a PostgreSQL array, which breaks on an empty, mixed, or nested list, so wrap a list meant as a Cypher list (for example, the rows of a batched UNWIND ingest) in Jsonb:

from psycopg.types.json import Jsonb

rows = graph.query('MATCH (a:"Actor" {name: %(name)s}) RETURN a.name AS name',
                   {"name": "Meg Ryan"})
print(rows)    # [{'name': 'Meg Ryan'}]

records = [
    {"id": "p1", "title": "A Paper", "year": 2023},
    {"id": "p2", "title": "Another Paper", "year": 2024},
]
papers = AgensGraph("papers", conf, create=True, refresh_schema=False)
papers.query("""
    UNWIND %(rows)s AS row
    MERGE (p:"Paper" {id: row.id})
      SET p.title = row.title, p.year = row.year
""", {"rows": Jsonb(records)})

query_many runs a burst of reads without waiting for each in turn. Each item is a statement or a (statement, params) pair, and you get one result list per item:

counts, meg = graph.query_many([
    'MATCH (a:"Actor") RETURN count(a) AS c',
    ('MATCH (a:"Actor" {name: %(name)s}) RETURN a.name AS name', {"name": "Meg Ryan"}),
])
print(counts, meg)    # [{'c': 3}] [{'name': 'Meg Ryan'}]

Schema introspection. get_schema (a property) returns a text schema for LLM prompts; get_structured_schema returns it as a dict. The schema is read when the store is constructed, so call refresh_schema(force=True) after the graph changes. Construct with enhanced_schema=True to sample more of each label and attach example property values (better Text2Cypher prompting), and with schema_cache_ttl= (seconds) to let refresh_schema skip a refresh that ran recently.

graph = AgensGraph("moviegraph", conf, enhanced_schema=True)
print(graph.get_schema)              # property — no parentheses
print(graph.get_structured_schema)   # dict: node_props / rel_props / relationships / counts / metadata / indexes
graph.refresh_schema(force=True)

Other options. Use it as a context manager, set a per-instance or per-call timeout (seconds), and pass sanitize=True to strip oversized list properties (e.g. raw embeddings) out of results:

with AgensGraph("moviegraph", conf, timeout=15, sanitize=True) as graph:
    rows = graph.query("MATCH (n) RETURN count(n) AS c", timeout=5)
# connection is closed on exit

For large graphs where you only run explicit Cypher, pass refresh_schema=False to skip the introspection scan at construction.

Vector store#

AgensgraphVector stores embeddings in AgensGraph via pgvector and works with any LangChain Embeddings. The examples below use OpenAI’s model:

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")   # 1536-d

No API key? To try the API, use DeterministicFakeEmbedding(size=1536) from langchain_core.embeddings instead. Its vectors are random, so search results are meaningless. Never use it for real data.

Build a store from documents (or texts), then search. Connect with a libpq url.

from langchain_core.documents import Document
from langchain_agensgraph import AgensgraphVector

url = "postgresql://postgres:postgres@localhost:55432/mydb"

docs = [
    Document(page_content="LangChain builds apps by chaining LLM components.",
             metadata={"topic": "frameworks"}),
    Document(page_content="AgensGraph is a multi-model graph database on PostgreSQL.",
             metadata={"topic": "databases"}),
]

store = AgensgraphVector.from_documents(docs, embeddings, url=url,
                                        graph_name="kb", node_label="Chunk",
                                        filter_properties=["topic"])

print(store.similarity_search("What is LangChain?", k=2))
for doc, score in store.similarity_search_with_score("graph database", k=2):
    print(f"[{score:.3f}] {doc.page_content[:50]}")

Adding, deleting, and fetching by id. batch_size bounds each database round-trip; embed_batch_size bounds the embedding API calls. Document.id is populated on retrieval.

store.add_texts(["a third doc", "a fourth doc"], ids=["doc-a", "doc-b"],
                batch_size=500, embed_batch_size=128)
print(store.get_by_ids(["doc-a"]))   # [Document(id='doc-a', ...)]
store.delete(["doc-a", "doc-b"])

Metadata filters and recall. Filters use MongoDB-style operators ($and, $or, $in, $gte, …). Name the metadata properties you filter on in filter_properties when you build the store: they get indexes when the store creates its vector index (not on from_existing_index), so a filtered search reads from an index rather than the whole label. effective_search_ratio over-fetches ANN candidates and trims to k, raising recall. Use from_existing_index to reconnect to an existing index without re-embedding.

vector = AgensgraphVector.from_existing_index(
    embedding=embeddings, index_name="vector",
    node_label="Chunk", graph_name="kb", url=url,
)

flt = {"topic": {"$in": ["databases", "frameworks"]}}
hits = vector.similarity_search_with_score("graph database", k=5, filter=flt,
                                           effective_search_ratio=4.0)

Async. aadd_texts, asimilarity_search, asimilarity_search_with_score, adelete, aget_by_ids, and aclose mirror the sync methods.

Retrievers#

Three BaseRetrievers are provided, sync (invoke) and async (ainvoke) alike. The vector and graph-context retrievers make one server round trip per retrieval; the Text2Cypher retriever adds an EXPLAIN check and any retries.

AgensVectorRetriever runs the store’s vector search (or its hybrid search, for a hybrid store) and puts each hit’s score in metadata["score"]. It also takes filter, effective_search_ratio, and a retrieval_query that shapes what each hit returns.

from langchain_agensgraph import AgensVectorRetriever, AgensgraphVector

store = AgensgraphVector.from_existing_index(
    embedding=embeddings, index_name="vector", node_label="Chunk", graph_name="kb", url=url,
)
retriever = AgensVectorRetriever(store=store, k=2)
docs = retriever.invoke("What is AgensGraph?")
print(docs[0].metadata["score"])

# Every knob is also a per-call override:
docs = retriever.invoke("What is AgensGraph?", k=1)

AgensGraphContextRetriever also fetches each hit’s neighbourhood within expand_by_hops (1 to 3), in the same statement as the seed search. The neighbours land in metadata["_context_nodes_"] and metadata["_context_rels_"]; the ready-made render_graph_context formatter writes them under the text for a prompt. Here a Person vertex is linked to one chunk, using the __id__ property in which the store keeps each document’s id:

from langchain_agensgraph import AgensGraph, AgensGraphContextRetriever
from langchain_agensgraph.retrievers import render_graph_context

store.add_texts(["Ada Lovelace published the first algorithm for a machine."],
                ids=["note-1"])
kb = AgensGraph("kb", conf, refresh_schema=False)
kb.query("""
    MATCH (c:"Chunk") WHERE c.__id__ = %(id)s
    MERGE (p:"Person" {name: %(name)s})
    MERGE (p)-[:"WROTE"]->(c)
""", {"id": "note-1", "name": "Ada Lovelace"})

retriever = AgensGraphContextRetriever(
    store=store, k=4, expand_by_hops=2, document_formatter=render_graph_context,
)
for doc in retriever.invoke("Who wrote the first algorithm?"):
    print(doc.page_content)
# The note-1 chunk prints as:
# Ada Lovelace published the first algorithm for a machine.
#
# Graph context:
# - Person:Ada Lovelace
# - Person:Ada Lovelace -[WROTE]-> Chunk:note-1

AgensText2CypherRetriever has a model write the Cypher, then contains it on the server: writes are refused, the query is checked with EXPLAIN, it runs in a read-only transaction, and one timeout covers the attempt. max_retries feeds an execution error back to the model for a corrected attempt; retry_on_empty=True does the same for a query that runs but matches nothing. The generated query is in each document’s metadata["cypher"].

Connect as an ordinary role. Generated Cypher runs in a read-only transaction, and that is not a boundary for a role that can run a program on the server host. So the chain, the Text2Cypher retriever, and create_cypher_tool refuse to run as a superuser or a member of pg_execute_server_program (ConfigurationError). Connect them as an ordinary role with read access to the graph, or pass allow_server_programs=True if you have decided that role is acceptable.

A graph is a schema of the same name, so read access is two grants. Run the second again after new labels are created, since each label is a table:

CREATE ROLE reader LOGIN PASSWORD 'reader';
GRANT USAGE ON SCHEMA moviegraph TO reader;
GRANT SELECT ON ALL TABLES IN SCHEMA moviegraph TO reader;

from langchain_openai import ChatOpenAI
from langchain_agensgraph import AgensGraph, AgensText2CypherRetriever

reader_conf = {**conf, "user": "reader", "password": "reader"}

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
graph = AgensGraph("moviegraph", reader_conf)

retriever = AgensText2CypherRetriever(graph=graph, llm=llm, max_retries=1,
                                      retry_on_empty=True)
rows = retriever.invoke("Which actors appeared in Top Gun?")
print(rows[0].metadata["cypher"])

Connection pooling#

By default each store opens its own connection. To let concurrent requests share one pool, create an AgensEngine (the driver’s connection pool, sync and async) and pass engine= to every store.

from langchain_agensgraph import AgensEngine, AgensGraph, AgensgraphVector

engine = AgensEngine.from_url(url, min_size=2, max_size=20)   # or .from_conf(conf, ...)

graph = AgensGraph("arxiv", conf, create=True, refresh_schema=False, engine=engine)
store = AgensgraphVector(embeddings, graph_name="arxiv", node_label="Paper", engine=engine)

# ... concurrent requests each borrow their own pooled connection ...
engine.close()        # await engine.aclose() from an event loop

Build a knowledge graph from text#

LLMGraphTransformer extracts a graph from documents using any chat model’s structured output. Call convert_to_graph_documents (or aconvert_to_graph_documents), then load the result with add_graph_documents — it runs in a single transaction, so a partial failure rolls back cleanly. include_source=True adds (:Document)-[:MENTIONS]->(entity) provenance edges.

from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
from langchain_agensgraph import AgensGraph, LLMGraphTransformer

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
transformer = LLMGraphTransformer(
    llm, allowed_nodes=["Person", "Organization", "Location", "Work"],
)

docs = [Document(
    page_content="Ada Lovelace worked with Charles Babbage on the Analytical Engine. "
                 "She is regarded as the first computer programmer.",
    metadata={"source": "wikipedia"},
)]
graph_docs = transformer.convert_to_graph_documents(docs)

graph = AgensGraph("wiki_kg", conf, create=True, refresh_schema=False)
graph.add_graph_documents(graph_docs, include_source=True)

print(graph.query("MATCH (s)-[r]->(t) RETURN s.id AS source, type(r) AS rel, t.id AS target LIMIT 10"))

Ask questions in plain language#

AgensCypherQAChain turns a question into Cypher against the graph’s schema, runs it in a read-only transaction, and answers from the rows. As with the Text2Cypher retriever, writes are refused unless you pass allow_dangerous_requests=True, the query is checked with EXPLAIN before it runs, and execution carries a timeout. The prompt is written for AgensGraph’s Cypher: it asks for quoted labels and rules out pattern expressions such as size((n)--()). On AgensGraph 2.17 it also rules out COUNT { }, EXISTS { }, and CALL { } subqueries and apoc.*; on a server with the GQL clauses (2.18) that rule is left out. The chain then quotes identifiers where your schema needs it. create_cypher_tool wraps the same pipeline as a tool for an agent.

The chain and the tool follow the same role rule as the Text2Cypher retriever; see Retrievers.

from langchain_openai import ChatOpenAI
from langchain_agensgraph import AgensCypherQAChain, AgensGraph, create_cypher_tool

reader_conf = {**conf, "user": "reader", "password": "reader"}

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
graph = AgensGraph("moviegraph", reader_conf)

chain = AgensCypherQAChain.from_llm(llm, graph=graph, return_intermediate_steps=True)
out = chain.invoke({"query": "Who acted in Top Gun?"})
print(out["result"])
print(out["intermediate_steps"])    # [{"query": "MATCH ..."}, {"context": [...]}]

# As an agent tool; answer=False hands the agent rows instead of prose.
tool = create_cypher_tool(graph, llm)
rows_tool = create_cypher_tool(graph, llm, answer=False)

examples= adds your own few-shot (question, Cypher) pairs to the prompt, and DIALECT_EXAMPLES (from langchain_agensgraph.chains) is a curated set of them. The schema the model reads names the graph’s property indexes unless the graph store was built with include_indexes=False.

Chat message history#

AgensChatMessageHistory stores a session’s messages as an ordered chain of graph vertices, isolated per session_id, with an optional window that returns only the most recent N.

from langchain_core.messages import AIMessage, HumanMessage
from langchain_agensgraph import AgensChatMessageHistory, AgensGraph

graph = AgensGraph("chat_log", conf, create=True, refresh_schema=False)
history = AgensChatMessageHistory("session-42", graph=graph, window=20)
history.clear()                  # start the example from an empty session

history.add_message(HumanMessage(content="What is AgensGraph?"))
history.add_message(AIMessage(content="A multi-model graph database on PostgreSQL."))
print(history.messages)     # most recent up to `window`, in order

You can also construct it with conf=... + graph_name=... instead of a shared graph, and there are aadd_messages / aget_messages async forms.

LangGraph checkpointer#

AgensSaver is a LangGraph BaseCheckpointSaver that persists a thread’s state to the graph so it resumes across restarts. (AsyncAgensSaver is an alias of the same class, which implements both sync and async methods.) Pass it as the checkpointer of any compiled LangGraph graph, or of an agent built with create_agent(..., checkpointer=saver). The example below needs no model: each run adds one to a running total, and a second, freshly built graph continues the same thread.

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langchain_agensgraph import AgensGraph, AgensSaver

class State(TypedDict):
    total: Annotated[int, operator.add]

def build_app():
    saver = AgensSaver(graph=AgensGraph("agent_memory", conf, create=True,
                                        refresh_schema=False))
    builder = StateGraph(State)
    builder.add_node("step", lambda state: {"total": 1})
    builder.add_edge(START, "step")
    builder.add_edge("step", END)
    return builder.compile(checkpointer=saver), saver

app, saver = build_app()
cfg = {"configurable": {"thread_id": "t1"}}
saver.delete_thread("t1")                   # start the example from an empty thread
print(app.invoke({"total": 0}, config=cfg))            # {'total': 1}

# A fresh app (e.g. a new process) resumes the same thread:
app2, _ = build_app()
print(app2.invoke({"total": 0}, config=cfg))           # {'total': 2}

The thread lifecycle is covered too: delete_thread(thread_id) purges a thread, copy_thread(source, target) copies one to a new thread id, prune(thread_ids) keeps only each thread’s latest checkpoint, and delete_for_runs(run_ids) removes the checkpoints of the given runs. Each has an a… async form.

LangGraph long-term memory#

AgensSaver persists one thread’s state; AgensStore is cross-thread long-term memory, implementing LangGraph’s BaseStore (get / put / search / delete / list_namespaces, sync and async). Items are ordinary vertices, so a memory can be linked to other memories and to your domain data with ordinary edges. A namespace search also returns the namespace’s descendants, and filters run in the database.

from langchain_agensgraph import AgensGraph, AgensStore

graph = AgensGraph("memories", conf=conf, create=True)
store = AgensStore(graph=graph)

store.put(("users", "alice", "memories"), "m1", {"text": "prefers tea", "topic": "prefs"})
item = store.get(("users", "alice", "memories"), "m1")
print(item.value)    # {'text': 'prefers tea', 'topic': 'prefs'}

hits = store.search(("users", "alice"), filter={"topic": "prefs"}, limit=10)
namespaces = store.list_namespaces(prefix=("users", "*", "memories"))
print(namespaces)    # [('users', 'alice', 'memories')]

Semantic search is opt-in and needs pgvector. Pass an index with the embedding width, an Embeddings, and the fields to embed. Embeddings are kept in a separate table in a companion schema (<graph>_store), not in the item’s properties.

store = AgensStore(
    graph=graph,
    index={"dims": 1536, "embed": embeddings, "fields": ["text"]},
)
store.put(("users", "alice", "memories"), "m2", {"text": "drinks green tea daily"})
hits = store.search(("users", "alice"), query="what do they drink?", limit=5)

See the queries that ran#

log_queries() collects every statement run inside the block, with how long each took, how many rows it returned, and whether it failed. Pass a callable to have each record reported as it finishes as well.

from langchain_agensgraph import log_queries

with log_queries() as statements:
    graph.query("MATCH (n) RETURN count(n) AS c")

for one in statements:
    print(f"{one.elapsed * 1000:6.1f} ms  {one.statement[:80]}")

Run the demos#

The repository ships a runnable demo suite on real datasets, each with a pre-executed notebook.

Demo

What it shows

Scripts

01_arxiv_graphrag

GraphRAG over arXiv (graph + vector, shared engine)

prepare.py → query.py

02_wikipedia_kg

Build a KG from text with LLMGraphTransformer; Text2Cypher QA

build_kg.py → ask.py

03_news_vector_rag

Vector RAG: hybrid search, metadata filters, recall tuning

ingest.py → rag.py

04_chat_memory_agent

Chat agent with message history + checkpointer

agent.py

05_graph_retrievers

The three retrievers side by side, and against the arXiv graph

ingest.py → retrieve.py, arxiv.py

Run each script from the langchain/ directory. The demos read connection settings and OPENAI_API_KEY from examples/demos/.env (copy .env.example); datasets stream from Hugging Face. Start with a small *_LIMIT and scale up once a demo runs end to end, and run 03_news_vector_rag before 04_chat_memory_agent (which reuses its store). evals/text2cypher/ scores generated Cypher by executing it against a dataset of 100 entries.