# Usage This is a hands-on guide to `langchain-agensgraph` 0.3.1. It assumes you have completed [Installation](installation.md): 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: ```python 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. ```python 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. ```python 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`: ```python 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: ```python 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. ```python 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: ```python 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: ```python 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`. ```python 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. ```python 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. ```python 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. ## Hybrid search Hybrid search fuses pgvector ANN similarity with a full-text keyword index using reciprocal rank fusion. Import `SearchType` (and the optional `HybridSearchConfig`) from the vector-store module, build the index with `search_type=SearchType.HYBRID`, and supply the keyword index name when you reconnect. ```python from langchain_agensgraph import AgensgraphVector from langchain_agensgraph.vectorstores.agensgraph_vector import SearchType, HybridSearchConfig # Build: creates both the HNSW vector index and the full-text index. store = AgensgraphVector.from_texts( ["The stock market rallied today.", "AI is transforming business."], embedding=embeddings, url=url, graph_name="headlines", node_label="Article", search_type=SearchType.HYBRID, ) # Query: reconnect with the keyword index, then weight keywords vs. vectors. hybrid = AgensgraphVector.from_existing_index( embedding=embeddings, index_name="vector", keyword_index_name="keyword", search_type=SearchType.HYBRID, node_label="Article", graph_name="headlines", url=url, ) hits = hybrid.similarity_search_with_score( "stock market", k=5, hybrid_config=HybridSearchConfig(keyword_weight=2.0), ) ``` > Hybrid search cannot be combined with a metadata `filter` — doing so raises `ValueError`. ## Retrievers Three `BaseRetriever`s 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. ```python 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: ```python 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: ```sql CREATE ROLE reader LOGIN PASSWORD 'reader'; GRANT USAGE ON SCHEMA moviegraph TO reader; GRANT SELECT ON ALL TABLES IN SCHEMA moviegraph TO reader; ``` ```python 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. ```python 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. ```python 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](#retrievers). ```python 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. ```python 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. ```python 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. ```python 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 (`_store`), not in the item's properties. ```python 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. ```python 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](https://github.com/skaiworldwide-oss/agensgraph-ai/tree/main/langchain/examples/demos) 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.