Pipeline, modes & LLMs

RAGPipeline wires the whole flow together:

parse → retrieve → reason → prove → assemble context → generate (LLM) → firewall → RAGResult

import hypermesh as hm
from hypermesh.rag import RAGPipeline, LLMConfig

db = hm.connect("data/")
pipe = RAGPipeline(db, table="DRONES", mode="hybrid", require_proof=True,
                   rules=[rule], llm_config=LLMConfig.mock())
result = await pipe.query("which edges are coordinated threats?")

Constructor

RAGPipeline(
    db, table,
    entity_map=None, openai_api_key=None, llm_config=None,
    top_k=40, token_budget=3000, db_dir="",
    # neuro-symbolic
    mode="neuro", require_proof=False, rules=None, rule_store=None,
    max_reason_iterations=64, reason_deadline_ms=None, provenance_resolver=None,
)

Parameter	Default	Meaning
db	—	An open hm.Connection.
table	—	Hyperedge table to query (e.g. "DRONES").
entity_map	auto	{node_id: {type, display, short}}; auto-loaded from {db_dir}/{TABLE}_entity_map.json if omitted.
openai_api_key	env	Falls back to OPENAI_API_KEY.
llm_config	OpenAI	Full LLMConfig override (see below).
top_k	40	Max hyperedges retrieved per query.
token_budget	3000	Max context tokens for the hyperedge block.
db_dir	HMDB_DIR/data	DB directory (locates entity maps + rule store).
mode	"neuro"	neuro | symbolic | hybrid. Invalid → ValueError.
require_proof	False	Abstain unless a symbolic fact is proved.
rules	None	Ad-hoc rule dicts; overrides the rule store for this pipeline.
rule_store	auto	A RuleStore; defaults to RuleStore(db_dir).
max_reason_iterations	64	Per-stratum fixpoint cap.
reason_deadline_ms	None	Wall-clock budget for reasoning.
provenance_resolver	None	Hook to enrich proof-leaf provenance.

Properties: pipe.table, pipe.model.

Modes

Mode	Reasoning	LLM	Use case
neuro	off	yes	Classic RAG; exploratory Q&A, no guarantees.
symbolic	on	no	Audit/compliance: derived facts + proofs, deterministic answer text, zero generation.
hybrid	on	yes	Product answers: reasoning feeds the LLM, firewall gates the output.

In symbolic mode no model is called at all — the answer is rendered deterministically from the proven facts, and require_proof makes it abstain when nothing is proved.

query()

result: RAGResult = await pipe.query("…")

Runs the full flow and returns a RAGResult. Raises ValueError if the table isn’t found.

stream()

parsed, edges, token_stream = await pipe.stream("…")
async for token in token_stream:
    print(token, end="")

Returns (ParsedQuery, list[RetrievedEdge], AsyncIterator[str]) for token-by-token generation. Streaming is generation-only (the firewall gates whole answers), so use query() when you need the verifiability guarantees.

RAGResult

Field	Meaning
query / parsed	Input query and parsed form.
answer	Final (firewalled) answer text.
cited_edges / all_edges	RetrievedEdges cited / retrieved.
confidence / low_confidence	0.0 when abstained; low_confidence if abstained or < 0.5.
mode	The mode that ran.
abstained	True if the firewall abstained.
derived_facts	List of {step_tag, fact, predicate, args, confidence, rule_id, rule_tag, hedge_tags}.
proofs	List of proof DAGs (ProofTree.as_dict() + step_tag/rule_tag).
firewall	{outcome, abstained, coverage, supported_count, stripped_count, stripped}.
rules_fired	Rule ids that produced facts.
model	Model name ("symbolic", "mock", "gpt-4o", …).
retrieval_ms / reasoning_ms / generation_ms / total_ms	Stage timings.
prompt_tokens / completion_tokens / context_tokens	Token accounting.
interaction_id	Harvested interaction id (for feedback/training).

result.as_dict() returns the JSON form (used by the REST API).

Plug in any LLM

The LLM is fully pluggable via LLMConfig. The reason → prove → firewall guarantees apply no matter which model writes the text, so a weaker/cheaper local model is still safe to deploy.

from hypermesh.rag import LLMConfig

Backend	Constructor	Notes
OpenAI	LLMConfig(model="gpt-4o", api_key=...)	Default backend.
OpenAI-compatible	LLMConfig(model=..., base_url=...)	vLLM, Groq, Together, Azure, llama.cpp server.
Ollama	LLMConfig.ollama("phi3:mini")	base_url=http://localhost:11434/v1.
LM Studio	LLMConfig.lmstudio()	base_url=http://localhost:1234/v1.
Local GGUF	LLMConfig.llama_cpp("model.gguf")	In-process; pip install llama-cpp-python.
Mock	LLMConfig.mock()	Deterministic, offline; for tests/CI/air-gapped demos.
Custom (any protocol)	LLMConfig.custom(fn)	Bring-your-own — Anthropic, Gemini, a fine-tune.

Custom backend — LLMConfig.custom(fn)

chat_fn(messages, cfg) may be sync or async (sync runs off the event loop) and receives OpenAI-style [{role, content}, ...]. It must return one of:

• a str — the answer text;
• a dict {"text", "prompt_tokens", "completion_tokens", "model"};
• a tuple (text, prompt_tokens, completion_tokens[, model]).

import anthropic
client = anthropic.Anthropic()

def claude(messages, cfg):
    sys  = next((m["content"] for m in messages if m["role"] == "system"), "")
    user = [m for m in messages if m["role"] != "system"]
    resp = client.messages.create(
        model="claude-3-5-sonnet-latest", system=sys,
        messages=[{"role": m["role"], "content": m["content"]} for m in user],
        max_tokens=cfg.max_tokens,
    )
    return {"text": resp.content[0].text,
            "prompt_tokens": resp.usage.input_tokens,
            "completion_tokens": resp.usage.output_tokens,
            "model": "claude-3-5-sonnet"}

pipe = RAGPipeline(db, table="DRONES", mode="hybrid", rules=[rule],
                   llm_config=LLMConfig.custom(claude))

Whatever the custom model emits, the firewall still strips any sentence that doesn’t cite a proven tag — the integration point cannot weaken the guarantee.

LLMConfig fields

Field	Default	Meaning
model	"gpt-4o-mini"	Model name / GGUF path.
base_url	OpenAI	API base URL.
api_key	""	API key.
temperature	0.1	Low — factual, not creative.
max_tokens	800	Generation cap.
timeout_s	60	Request timeout.
backend	"openai"	openai | llama_cpp | mock | custom.
gguf_path	""	GGUF path when backend="llama_cpp".
n_gpu_layers	-1	-1 = all layers on GPU.
n_ctx	4096	Context window for llama.cpp.
chat_fn	None	Callable for backend="custom".

Next: Python API reference or the REST API.