langchain-tacnode

Tacnode integration for LangChain and LangGraph — multi-modal retrieval and a drop-in agent stack on top of one hybrid SQL plane.

Docs: this README (overview + quickstarts) · CATALOG.md (full API reference) · ARCHITECTURE.md (complete architecture diagram + how the pieces layer) · TACNODE_CAPABILITIES.md (the platform underneath).

Architecture at a glance

   your application  (config, not agent code)
            │  bind your tables · pick specialists
            ▼
   ┌─────────────────────────────────────────────────────────────┐
   │ langchain-tacnode                                           │
   │   Specialist · Subsystem · coordinate() · ResearchWorkflow  │
   │   six retrievers · TacnodeContextLake                       │
   │   TacnodeEngine · get_chat_client() · get_embeddings()      │
   └───────────┬─────────────────────────────────┬───────────────┘
               │ one SQL plane                   │ chat + embeddings
               ▼                                 ▼
   Tacnode Context Lake              Anthropic · OpenAI-compatible ·
   USING HYBRID tables + MVs         Bedrock
   vector · FTS · JSONB · time travel

The complete diagram — every module, the compiled workflow graph, and the end-to-end data flow — is in ARCHITECTURE.md.

Why Tacnode

Agent context has mostly been engineered at the application layer: retrieval routers, memory libraries, caching tiers, and glue pipelines that decide which store to query and stitch the results back together. That scaffolding exists because the data layer underneath couldn't answer the question natively — the same reason application-layer caching and sharding logic once existed for databases that later absorbed them.

Tacnode approaches context as an infrastructure problem. The Context Lake holds structured rows, JSONB documents, vector embeddings, and full-text indexes in one table inside one transactional boundary, and its query planner co-optimizes across all of them — transactional, vector, keyword, analytical — in a single SQL statement on a single consistent snapshot. Query planning across data modalities is what query planners are for.

With langchain-tacnode, LangChain and LangGraph do what they are built for — orchestration, tool selection, reasoning — while retrieval planning, freshness, and consistency are the database's job. Agents no longer maintain schemas or data pipelines.

The full platform reference is TACNODE_CAPABILITIES.md; the capabilities this package surfaces, each mapped to its LangChain component:

• Single-query hybrid retrieval — vector + full-text + structured filters fused by Reciprocal Rank Fusion in one SQL statement, on one consistent snapshot. Surface: TacnodeHybridRetriever.
• Time travel — every table carries version history: query as of any past moment (FOR SYSTEM_TIME AS OF), cross-temporal joins, row recovery. Surface: TacnodeTimeTravelRetriever.
• Hybrid row+columnar storage, dual query engines — transactional writes and analytical/vector scans run on the same committed data; the planner routes each operator. Surface: TacnodeContextLake creates USING HYBRID tables.
• Distributed indexes at scale — split_hnsw / split_gin partitioned across nodegroups; billion-scale HNSW; fp16/int8/PQ quantization; IVFFlat. Surface: the table DDL and every retriever's queries.
• Multi-modal data in one transactional boundary — structured columns + JSONB + VECTOR(N) + TSVECTOR in one table; one query crosses all of them. Surface: all six retrievers.
• Incremental materialized views — delta-only refresh, nested IMVs, sub-second freshness SLA; MVs take the same indexes as tables. Surface: retrievers point at MVs exactly like tables (e.g. a signature-index MV).
• Generated columns — the FTS TSVECTOR is GENERATED ALWAYS AS ... STORED, maintained by the database on every write. Surface: the table's ts_content column.
• Semantic SQL — inline LLM operators (openai_complete / openai_filter / openai_extract / openai_agg / openai_embed) run inside ordinary SQL. Surface: get_chat_client() is the client-side equivalent; the swap happens at the SQL layer with no agent-code change.
• Agent-loop scale — ~25,000 requests/sec per node on a coroutine scheduler; sub-second failover; zero-downtime upgrades. Surface: ResearchWorkflow's tool loop runs against one context lake.
• Full PostgreSQL ecosystem compatibility — wire-protocol compatible; every driver, ORM, and BI tool works. Surface: TacnodeEngine is plain psycopg3 + SQLAlchemy — no special driver or SDK.

On top of the retrieval plane, the package adds a drop-in agent stack: register your tables and specialists; get a working multi-specialist research workflow with citation-grounded narrative output. Customer writes config, not agent code.

The package is self-contained: TacnodeEngine.from_env(), get_chat_client(), and get_embeddings() build the DB connection, chat model, and embeddings from arguments or environment alone — no external config system required. A host application can still inject its own configured clients; the factories are the batteries-included default, not a requirement.

What's in the box

Connections + providers:

• TacnodeEngine — connection manager (mirrors langchain-postgres PGEngine); from_connection_string / from_engine / from_env.
• get_chat_client / get_embeddings — provider-agnostic chat-model and embeddings factories (OpenAI-compatible / Bedrock / Anthropic).

Storage + retrieval:

• TacnodeContextLake — the multi-modal context-lake table class: creates a USING HYBRID table (text + JSONB + vector + database-generated FTS) with its HNSW and GIN indexes, and manages writes through LangChain's VectorStore interface — add_texts / similarity_search / get_by_ids / delete.
• TacnodeHybridRetriever — vector + FTS + filters fused by RRF in one SQL statement.
• TacnodeVectorRetriever — pure pgvector cosine over any existing table/MV.
• TacnodeVectorByIdRetriever — k-NN by an existing row's stored vector (query string is a row id — no embeddings needed).
• TacnodeFTSRetriever — plainto_tsquery + ts_rank over a split_gin index.
• TacnodeTimeTravelRetriever — FOR SYSTEM_TIME AS OF historical snapshots, with allowlist + offset validation.
• TacnodeStructuredRetriever — parameterized SELECT with whitelisted projection and safe ORDER BY.

Agents:

• Specialist / Subsystem / coordinate — one specialist + its tools, runnable alone or merged into one multi-specialist workflow.
• catalyst_detective / pattern_matcher / risk_analyst / trader_profiler / coordination_analyst — ready-made specialist factories: generic prompts, all schema as arguments.
• ResearchWorkflow — compiled LangGraph StateGraph: LLM tool-selection loop → evidence → root cause → cited narrative; invoke() or live astream(), opt-in SQL/prompt tracing.
• NarrativeComposer — renders accumulated evidence as plain English with [citation:table:id] markers.
• Generic nodes (langchain_tacnode.nodes) — tool_selection_node, analyze_results_node, generate_root_cause_node, narrative_composer_node, route_tool_selection, make_execute_node — for fully custom graphs.

See CATALOG.md for the full API reference and ARCHITECTURE.md for how the pieces layer.

Install

# during development:
pip install -e /path/to/langchain-tacnode

# once published:
pip install langchain-tacnode

Quickstart — hybrid retrieval

from langchain_tacnode import TacnodeEngine, TacnodeHybridRetriever, get_embeddings

engine = TacnodeEngine.from_connection_string("postgresql+psycopg://...")
retriever = TacnodeHybridRetriever(
    engine=engine,
    table_name="docs",
    embedding=get_embeddings(provider="openai"),
    text_column="content",
    vector_column="embedding",
    fts_column="ts_content",
    metadata_filter={"department": "legal"},   # JSONB containment
    top_k=10,
    vector_weight=0.6,
    fts_weight=0.4,
)
docs = retriever.invoke("revenue recognition under ASC 606")

One SQL statement: the top-50 by vector cosine distance and the top-50 by ts_rank are fused per row id via Reciprocal Rank Fusion — SUM(weight / (60 + rank)) — with any filters ANDed into both arms, and the top-k rows returned as Documents.

Three filter mechanisms, all optional and composable:

• metadata_filter — JSONB containment on the metadata column (metadata @> :json).
• column_filter — real SQL-column predicates: {'symbol': 'ACME'} → symbol = :v, {'member_id': [ids]} → member_id IN (…) (keys whitelisted, values bound).
• time_window — (column, lo, hi) anchoring retrieval to a reference time instead of now (either bound may be None).

Quickstart — create a context-lake table and write documents

from langchain_tacnode import TacnodeEngine, TacnodeContextLake, get_embeddings

engine = TacnodeEngine.from_env()
store = TacnodeContextLake.from_texts(
    texts=["...doc one...", "...doc two..."],
    embedding=get_embeddings(provider="bedrock", model="cohere.embed-english-v3"),
    engine=engine,
    table_name="my_events",
    embedding_dimension=1024,        # must match the embedding model's output dim
)
hits = store.similarity_search("share dilution", k=4)

from_texts (and init_context_table) idempotently create a USING HYBRID table with a VECTOR(N) column, a generated TSVECTOR column, a split_hnsw cosine index, and a split_gin FTS index — so TacnodeHybridRetriever has FTS coverage over the same table from day one.

Quickstart — ready-made specialists

from langchain_tacnode import (
    TacnodeEngine, get_chat_client, get_embeddings,
    catalyst_detective, risk_analyst, coordinate,
)

engine = TacnodeEngine.from_env()
emb = get_embeddings()

# Run ONE specialist on a question…
detective = catalyst_detective(engine, emb, table="catalyst_events",
                               column_filter={"symbol": "ACME"})
out = await detective.run("did a filing precede the move for ACME?")
print(out["narrative"])

# …or coordinate several into ONE investigation.
risk = risk_analyst(
    engine, table="exposures",
    projection_columns=["id", "symbol", "exposure", "volatility"],
    sort_columns={"exposure": "exposure", "volatility": "volatility"},
    default_sort="exposure DESC",
)
panel = coordinate([detective, risk], get_chat_client())
result = await panel.invoke({"question": "what happened with ACME yesterday?"})

Each factory returns a Subsystem (one Specialist + its retriever registry). coordinate() merges any number of subsystems into a single LangGraph workflow whose orchestrator routes across all of their tools.

Quickstart — custom workflow

from langchain_tacnode import (
    TacnodeEngine, TacnodeHybridRetriever, TacnodeStructuredRetriever,
    Specialist, ResearchWorkflow, get_chat_client,
)

engine = TacnodeEngine.from_connection_string("postgresql+psycopg://...")
retrievers = {
    "fts_search":    TacnodeHybridRetriever(engine=engine, table_name="posts", ...),
    "entity_lookup": TacnodeStructuredRetriever(engine=engine, table_name="entities", ...),
}
specialists = [
    Specialist(
        name="Detective",
        system_prompt="Find evidence relevant to the question",
        tools=["fts_search"],
        classification_categories=["confirmed", "denied", "unknown"],
    ),
    Specialist(
        name="Profiler",
        system_prompt="Profile the entities involved",
        tools=["entity_lookup"],
        classification_categories=["known", "novel"],
    ),
]
workflow = ResearchWorkflow(
    retrievers=retrievers,
    specialists=specialists,
    llm=get_chat_client(),
    max_iterations=5,
)
result = await workflow.invoke({"question": "what happened with X yesterday?"})
print(result["narrative"])
for citation in result["citations"]:
    print(f"  - {citation}")

ResearchWorkflow compiles a LangGraph StateGraph: an LLM-driven tool-selection loop dispatches to one execute node per registered retriever (accumulating provenance-tagged evidence), then analyze_results → generate_root_cause (structured output with confidence + recommendations) → narrative_composer (plain English with [citation:table:id] markers wired to evidence rows). The compiled graph is drawn in ARCHITECTURE.md.

Live streaming: async for ev in workflow.astream({"question": ...}) yields {'type': 'step', ...} events as each node completes and a final {'type': 'result', ...} — instead of blocking on invoke().

Opt-in tracing: ResearchWorkflow(..., trace=True) (also coordinate(..., trace=True)) records each retriever's parameterized SQL into state['queries'] and each LLM node's prompts into state['prompts'], and astream() additionally emits {'type': 'query', ...} / {'type': 'prompt', ...} events. Off by default — behavior is unchanged unless you ask for it.

Provider-agnostic LLM + embeddings

get_chat_client(...)

get_chat_client(model=None, temperature=0.3, max_tokens=2000, *,
                provider=None, api_key=None, base_url=None, region=None)

Provider precedence: explicit provider= > env flags > Anthropic default.

• USE_OPENAI_LLM=true → ChatOpenAI (any OpenAI-compatible endpoint via OPENAI_BASE_URL)
• USE_BEDROCK=true → ChatBedrock
• default → ChatAnthropic

Models resolve from env (OPENAI_MODEL, AWS_BEDROCK_MODEL, CLAUDE_MODEL) unless the caller passes an override; api_key / base_url / region arguments override env, so the client can be configured entirely in code.

get_embeddings(...)

get_embeddings(provider=None, model=None, *,
               dimensions=None, api_key=None, region=None)

The embeddings mirror of get_chat_client.

• Provider precedence: explicit provider= > EMBEDDING_PROVIDER env > the chat flags.
• Supported: "openai" (OpenAIEmbeddings, default text-embedding-3-small, 1536-dim) and "bedrock" (BedrockEmbeddings, default cohere.embed-english-v3, 1024-dim).
• Anthropic exposes no embedding endpoint — configure OpenAI or Bedrock for embeddings even when chat is Anthropic.
• The same embeddings must write the corpus and embed the queries, and the table's VECTOR(N) dimension must match the model's output.

TacnodeEngine.from_env()

• Resolves the connection string from TACNODE_DATABASE_URL, then DATABASE_URL, then TACNODE_DB_HOST / TACNODE_DB_PORT / TACNODE_DB_USER / TACNODE_DB_PASSWORD / TACNODE_DB_NAME.
• Bare postgres:// / postgresql:// URLs are normalized to the psycopg driver.

Schema ownership

The retrievers and agents run no DDL — they read whatever tables, materialized views, and indexes you point them at; declare those in your own migrations. The one component that issues DDL is TacnodeContextLake: init_context_table() / from_texts(init_table=True) idempotently create the store's own table + indexes. Skip them (construct the store directly, or pass init_table=False) when your schema is owned elsewhere.

License

MIT.