Build the knowledge base you want with Heta.

Heta breaks knowledge-base construction into clear components: models, stores, parsers, steps, search modes, and benchmarks. Start with a simple vector KB, then add keyword retrieval, Heta-style graph knowledge, and evaluation when you need them.

Quick Start What is a Recipe?

Models Stores Parsers Steps Search Benchmarks

How it works

Heta does not ask you to write a complete RAG system all at once. You describe models, stores, parsers, and steps in a Recipe, and Heta builds a KnowledgeBase from that Recipe. After the build, the KB knows which search modes it supports and can be evaluated directly with benchmarks.

01

Recipe

A Recipe is the build plan for a knowledge base. It declares which model to use, where files are stored, which parsers are enabled, and which steps run in order. To reuse or change a KB, change the Recipe.

02

Steps

Steps perform the build: parse, split, embed, index, persist text, or build graph facts. Use only the steps needed for vector retrieval, or continue into the Heta graph procedure when relation-aware retrieval is needed.

03

Search

Search works only with assets the KnowledgeBase has actually built. A vector index unlocks vector search, a text index unlocks full-text search, and graph assets unlock Heta graph search.

04

Benchmark

Benchmarks build KBs from the same Recipe, run query modes, and generate evaluation reports. This lets you compare Recipes with data instead of judging a single query by feel.

Four cases

These cases use a local ObjectStore and in-memory stores. Models use common OpenAI LLM and embedding APIs by default. For Qwen, Milvus, PostgreSQL, or Elasticsearch, replace only the corresponding component.

examples/home_vector_case.en.py

$ OPENAI_API_KEY=... PYTHONPATH=src python docs/examples/home_vector_case.en.py

import asyncio
import os
import shutil
from pathlib import Path

from heta_framework.common.models import EmbeddingModel, LanguageModel
from heta_framework.common.stores import InMemoryVectorStore, LocalObjectStore
from heta_framework.kb import (
    DocumentParserRegistry,
    EmbedChunks,
    IndexVectors,
    KnowledgeBase,
    KnowledgeModels,
    KnowledgeParsers,
    KnowledgeRecipe,
    KnowledgeStores,
    ParseDocuments,
    SplitDocuments,
    SplitDocumentsConfig,
    TextParser,
)


async def main() -> None:
    # 0. Prepare a clean workspace. In production, use your service workspace or object store.
    workspace = Path("heta-demo-vector")
    shutil.rmtree(workspace, ignore_errors=True)

    # 1. Stores: ObjectStore keeps raw files and artifacts; VectorStore keeps the vector index.
    objects = LocalObjectStore(workspace / "objects")
    vectors = InMemoryVectorStore()

    # 2. Models: Heta model clients call common providers through LiteLLM.
    #    Before running: export OPENAI_API_KEY=...
    language = LanguageModel(
        model_name=os.getenv("HETA_LLM_MODEL", "openai/gpt-4o-mini"),
        api_key=os.environ["OPENAI_API_KEY"],
    )
    embedding = EmbeddingModel(
        model_name=os.getenv("HETA_EMBEDDING_MODEL", "openai/text-embedding-3-small"),
        api_key=os.environ["OPENAI_API_KEY"],
    )

    # 3. Input document: this example writes a txt file; PDF/HTML/Office files use other parsers.
    await objects.put(
        "raw/heta.txt",
        (
            "Heta builds KnowledgeBase objects from Recipe definitions. "
            "Vector search retrieves chunks by semantic similarity."
        ).encode("utf-8"),
    )

    # 4. Recipe: declare parser, models, stores, and ordered build steps.
    recipe = KnowledgeRecipe(
        parsers=KnowledgeParsers(documents=DocumentParserRegistry([TextParser()])),
        models=KnowledgeModels(language=language, embedding=embedding),
        stores=KnowledgeStores(objects=objects, vector=vectors),
        steps=(
            ParseDocuments(),
            SplitDocuments(SplitDocumentsConfig(encoding_name="unicode")),
            EmbedChunks(),
            IndexVectors(),
        ),
    )

    # 5. Build + query: build the vector index with a real embedding API.
    #    With generate_answer=True, the query engine also calls the LLM for an answer.
    kb = await KnowledgeBase.create(recipe=recipe, name="home-vector")
    _raise_if_build_failed(kb)
    response = await kb.query(
        "How does Heta build a knowledge base?",
        mode="vector_search",
        top_k=1,
        options={"generate_answer": True},
    )

    print(response.answer)
    print(response.results[0].text)

    await language.aclose()
    await embedding.aclose()
    await vectors.aclose()
    await objects.aclose()


def _raise_if_build_failed(kb: KnowledgeBase) -> None:
    if kb.run_record.status == "succeeded":
        return
    failed_step = next(
        (record for record in reversed(kb.run_record.step_records) if record.status == "failed"),
        None,
    )
    if failed_step is None:
        raise RuntimeError(f"knowledge base build failed: {kb.run_record.status}")
    raise RuntimeError(f"{failed_step.step_name} failed: {failed_step.error}")


asyncio.run(main())

Example output

Heta builds a knowledge base by creating KnowledgeBase objects from Recipe definitions [1].
Heta builds KnowledgeBase objects from Recipe definitions. Vector search retrieves chunks by semantic similarity.

$ OPENAI_API_KEY=... PYTHONPATH=src python docs/examples/home_full_text_case.en.py

import asyncio
import os
import shutil
from pathlib import Path

from heta_framework.common.models import LanguageModel
from heta_framework.common.stores import InMemoryTextIndexStore, LocalObjectStore
from heta_framework.kb import (
    DocumentParserRegistry,
    IndexFullText,
    KnowledgeBase,
    KnowledgeModels,
    KnowledgeParsers,
    KnowledgeRecipe,
    KnowledgeStores,
    ParseDocuments,
    SplitDocuments,
    SplitDocumentsConfig,
    TextParser,
)


async def main() -> None:
    # 0. Prepare a clean workspace.
    workspace = Path("heta-demo-full-text")
    shutil.rmtree(workspace, ignore_errors=True)

    # 1. Stores: ObjectStore keeps document artifacts; TextIndexStore keeps the full-text index.
    objects = LocalObjectStore(workspace / "objects")
    text_index = InMemoryTextIndexStore()

    # 2. Model: full_text_search does not require an LLM; the LLM is only for answer generation.
    #    Before running: export OPENAI_API_KEY=...
    language = LanguageModel(
        model_name=os.getenv("HETA_LLM_MODEL", "openai/gpt-4o-mini"),
        api_key=os.environ["OPENAI_API_KEY"],
    )

    # 3. Input document: full-text indexing is useful for exact terms, IDs, abbreviations, and keywords.
    await objects.put(
        "raw/heta.txt",
        (
            "Heta can add full-text search with IndexFullText. "
            "BM25-style retrieval is useful for exact terms and identifiers."
        ).encode("utf-8"),
    )

    # 4. Recipe: IndexFullText writes chunks into the text index.
    recipe = KnowledgeRecipe(
        parsers=KnowledgeParsers(documents=DocumentParserRegistry([TextParser()])),
        models=KnowledgeModels(language=language),
        stores=KnowledgeStores(objects=objects, text_index=text_index),
        steps=(
            ParseDocuments(),
            SplitDocuments(SplitDocumentsConfig(encoding_name="unicode")),
            IndexFullText(),
        ),
    )

    # 5. Build + query: IndexFullText unlocks full_text_search.
    kb = await KnowledgeBase.create(recipe=recipe, name="home-full-text")
    _raise_if_build_failed(kb)
    response = await kb.query(
        "BM25 exact terms",
        mode="full_text_search",
        top_k=1,
        options={"generate_answer": True},
    )

    print(response.answer)
    print(response.results[0].text)

    await language.aclose()
    await text_index.aclose()
    await objects.aclose()


def _raise_if_build_failed(kb: KnowledgeBase) -> None:
    if kb.run_record.status == "succeeded":
        return
    failed_step = next(
        (record for record in reversed(kb.run_record.step_records) if record.status == "failed"),
        None,
    )
    if failed_step is None:
        raise RuntimeError(f"knowledge base build failed: {kb.run_record.status}")
    raise RuntimeError(f"{failed_step.step_name} failed: {failed_step.error}")


asyncio.run(main())

Example output

BM25-style retrieval is useful for exact terms and identifiers [1].
Heta can add full-text search with IndexFullText. BM25-style retrieval is useful for exact terms and identifiers.

$ OPENAI_API_KEY=... PYTHONPATH=src python docs/examples/home_graph_case.en.py

import asyncio
import os
import shutil
from pathlib import Path

from heta_framework.common.models import EmbeddingModel, LanguageModel
from heta_framework.common.stores import InMemoryVectorStore, LocalObjectStore, SQLStore
from heta_framework.kb import (
    DocumentParserRegistry,
    HetaGraphProcedure,
    KnowledgeBase,
    KnowledgeModels,
    KnowledgeParsers,
    KnowledgeRecipe,
    KnowledgeStores,
    ParseDocuments,
    SplitDocuments,
    SplitDocumentsConfig,
    TextParser,
)


async def main() -> None:
    # 0. Prepare a clean workspace.
    workspace = Path("heta-demo-graph")
    shutil.rmtree(workspace, ignore_errors=True)

    # 1. Stores: ObjectStore keeps artifacts, SQLStore writes graph facts, VectorStore supports graph recall.
    objects = LocalObjectStore(workspace / "objects")
    sql = SQLStore(f"sqlite:///{workspace / 'graph.db'}")
    vectors = InMemoryVectorStore()

    # 2. Models: the LLM extracts entities/relations; the embedding model indexes graph facts.
    #    Before running: export OPENAI_API_KEY=...
    language = LanguageModel(
        model_name=os.getenv("HETA_LLM_MODEL", "openai/gpt-4o-mini"),
        api_key=os.environ["OPENAI_API_KEY"],
    )
    embedding = EmbeddingModel(
        model_name=os.getenv("HETA_EMBEDDING_MODEL", "openai/text-embedding-3-small"),
        api_key=os.environ["OPENAI_API_KEY"],
    )

    # 3. Input document: Heta graph procedure extracts entities and relations from chunks.
    await objects.put(
        "raw/heta.txt",
        (
            "Heta builds knowledge bases from recipes. "
            "A KnowledgeBase is created by running Recipe steps."
        ).encode("utf-8"),
    )

    # 4. Recipe: HetaGraphProcedure expands into entity extraction, relation extraction, and graph build steps.
    recipe = KnowledgeRecipe(
        parsers=KnowledgeParsers(documents=DocumentParserRegistry([TextParser()])),
        models=KnowledgeModels(language=language, embedding=embedding),
        stores=KnowledgeStores(objects=objects, sql=sql, vector=vectors),
        steps=(
            ParseDocuments(),
            SplitDocuments(SplitDocumentsConfig(encoding_name="unicode")),
            *HetaGraphProcedure.build(deduplicate=False).steps(),
        ),
    )

    # 5. Build + query: BuildGraph unlocks heta_graph_search.
    kb = await KnowledgeBase.create(recipe=recipe, name="home-graph")
    _raise_if_build_failed(kb)
    response = await kb.query(
        "How does Heta create a KnowledgeBase?",
        mode="heta_graph_search",
        top_k=3,
        options={"generate_answer": True},
    )

    print(response.answer)
    print(response.results[0].kind, response.results[0].text)

    await language.aclose()
    await embedding.aclose()
    await sql.aclose()
    await vectors.aclose()
    await objects.aclose()


def _raise_if_build_failed(kb: KnowledgeBase) -> None:
    if kb.run_record.status == "succeeded":
        return
    failed_step = next(
        (record for record in reversed(kb.run_record.step_records) if record.status == "failed"),
        None,
    )
    if failed_step is None:
        raise RuntimeError(f"knowledge base build failed: {kb.run_record.status}")
    raise RuntimeError(f"{failed_step.step_name} failed: {failed_step.error}")


asyncio.run(main())

Example output

Heta creates a KnowledgeBase by building it from recipes [1][2][3].
relation Relation: Heta -> KnowledgeBase
Name: builds
Type: creates
Description: Heta builds knowledge bases from recipes.

$ OPENAI_API_KEY=... PYTHONPATH=src python docs/examples/home_benchmark_case.en.py

import asyncio
import json
import os
import shutil
from pathlib import Path

from heta_framework.common.models import EmbeddingModel
from heta_framework.common.stores import InMemoryVectorStore, LocalObjectStore
from heta_framework.evaluation import (
    BenchmarkManifest,
    BenchmarkRunConfig,
    BenchmarkRunner,
    BenchmarkWorkspace,
    EvidenceRecallAtK,
    JsonlBenchmark,
)
from heta_framework.kb import (
    DocumentParserRegistry,
    EmbedChunks,
    IndexVectors,
    KnowledgeModels,
    KnowledgeParsers,
    KnowledgeRecipe,
    KnowledgeStores,
    ParseDocuments,
    SplitDocuments,
    SplitDocumentsConfig,
    TextParser,
)


async def main() -> None:
    # 0. Prepare a minimal benchmark workspace.
    workspace = Path("heta-demo-benchmark")
    shutil.rmtree(workspace, ignore_errors=True)
    workspace.mkdir(parents=True)

    # 1. Benchmark data: JsonlBenchmark needs documents.jsonl and cases.jsonl.
    documents_jsonl = workspace / "documents.jsonl"
    cases_jsonl = workspace / "cases.jsonl"
    documents_jsonl.write_text(
        json.dumps(
            {
                "document_id": "doc_heta",
                "name": "heta.txt",
                "media_type": "text/plain",
                "text": (
                    "Heta builds KnowledgeBase objects from Recipe definitions. "
                    "Vector search retrieves chunks by semantic similarity."
                ),
            }
        )
        + "\n",
        encoding="utf-8",
    )
    cases_jsonl.write_text(
        json.dumps(
            {
                "case_id": "case_vector",
                "query": "What retrieves chunks by semantic similarity?",
                "expected": {
                    "evidence": [
                        {"text": "Vector search retrieves chunks by semantic similarity."}
                    ]
                },
            }
        )
        + "\n",
        encoding="utf-8",
    )

    # 2. Stores + model: BenchmarkRunner builds the KB from the same Recipe, then runs queries.
    objects = LocalObjectStore(workspace / "objects")
    vectors = InMemoryVectorStore()
    embedding = EmbeddingModel(
        model_name=os.getenv("HETA_EMBEDDING_MODEL", "openai/text-embedding-3-small"),
        api_key=os.environ["OPENAI_API_KEY"],
    )

    # 3. Recipe: this benchmark evaluates a minimal vector_search KB.
    recipe = KnowledgeRecipe(
        parsers=KnowledgeParsers(documents=DocumentParserRegistry([TextParser()])),
        models=KnowledgeModels(embedding=embedding),
        stores=KnowledgeStores(objects=objects, vector=vectors),
        steps=(
            ParseDocuments(),
            SplitDocuments(SplitDocumentsConfig(encoding_name="unicode")),
            EmbedChunks(),
            IndexVectors(),
        ),
    )

    # 4. Benchmark: use evidence_recall@1 to check whether top-1 evidence matches.
    benchmark = JsonlBenchmark(
        manifest=BenchmarkManifest(
            name="home_jsonl",
            version="v1",
            split="demo",
            task_type="retrieval",
        ),
        documents_jsonl=documents_jsonl,
        cases_jsonl=cases_jsonl,
        evaluator_list=(EvidenceRecallAtK(k=1),),
    )
    result = await BenchmarkRunner().run(
        benchmark=benchmark,
        recipe=recipe,
        knowledge_base_name="home-benchmark",
        query_modes=("vector_search",),
        workspace=BenchmarkWorkspace(root_dir=workspace, cache_dir=workspace / "cache"),
        config=BenchmarkRunConfig(top_k=1, report_id="home_demo"),
    )
    _raise_if_benchmark_failed(result)

    print(result.report.score_summary)
    print(result.report_key)

    await embedding.aclose()
    await vectors.aclose()
    await objects.aclose()


def _raise_if_benchmark_failed(result) -> None:
    failed_build = next(
        (kb for kb in result.knowledge_bases if kb.run_record.status != "succeeded"),
        None,
    )
    if failed_build is not None:
        failed_step = next(
            (
                record
                for record in reversed(failed_build.run_record.step_records)
                if record.status == "failed"
            ),
            None,
        )
        if failed_step is None:
            raise RuntimeError(f"knowledge base build failed: {failed_build.run_record.status}")
        raise RuntimeError(f"{failed_step.step_name} failed: {failed_step.error}")

    failed_case = next((case for case in result.report.case_results if case.error), None)
    if failed_case is not None:
        raise RuntimeError(f"{failed_case.case_id} failed: {failed_case.error.message}")


asyncio.run(main())

Example output

{'vector_search.evidence_recall@1': 1.0}
_heta/knowledge_bases/home-benchmark/evaluations/home_demo/report.json

Benchmark support

A benchmark adapter prepares data, builds KnowledgeBases, runs query modes, and generates evaluation reports. Use built-in benchmarks or implement the protocol for your own business eval set.

MultiHop-RAG A multi-hop QA benchmark for complex queries, evidence recall, and multi-hop search. Dataset

BEIR A standard information retrieval benchmark. Heta currently supports SciFact, NFCorpus, FiQA, and HotpotQA. Datasets

UDA-Benchmark A real document-analysis benchmark that can build multiple KBs per case to evaluate different Recipes. Source documents