Quick Start

This page builds your first Heta KnowledgeBase from a local text file.

The first run builds the smallest vector knowledge base:

raw text
  -> ParseDocuments
  -> SplitDocuments
  -> EmbedChunks
  -> IndexVectors
  -> vector_search

This path uses the fewest components. It is the fastest way to confirm that installation, model calls, parsing, chunking, and vector search are working. After that, add full-text search, Heta graph search, or benchmarks as needed.

Install

Heta is published on PyPI. Install it with the package name heta-framework; import it in Python with heta_framework.

The minimal vector example only needs the core package:

python -m pip install heta-framework

Install extras only when you need production stores or full-text indexing:

python -m pip install "heta-framework[sql]"          # SQLStore and SQLite/PostgreSQL-style flows
python -m pip install "heta-framework[postgres]"     # PostgreSQL driver
python -m pip install "heta-framework[mysql]"        # MySQL driver
python -m pip install "heta-framework[milvus]"       # Milvus VectorStore
python -m pip install "heta-framework[s3]"           # S3-compatible ObjectStore
python -m pip install "heta-framework[text-index]"   # Elasticsearch full-text index

Set your model API key:

export OPENAI_API_KEY="sk-..."

Heta's model layer is powered by LiteLLM. model_name follows LiteLLM naming, for example openai/gpt-4o-mini or openai/text-embedding-3-small.

Build Your First KnowledgeBase

Create quickstart.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. 准备一个干净 workspace；真实服务里通常换成你的业务目录或对象存储。
    workspace = Path("heta-demo-vector")
    shutil.rmtree(workspace, ignore_errors=True)

    # 1. Stores：ObjectStore 管原始文件和中间产物，VectorStore 管向量索引。
    objects = LocalObjectStore(workspace / "objects")
    vectors = InMemoryVectorStore()

    # 2. Models：Heta 的 model client 通过 LiteLLM 调用常见外部模型。
    #    运行前设置：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. 输入文档：这里写入一个 txt；PDF/HTML/Office 可以换成对应 parser。
    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：声明 parser、model、store，以及要执行的 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：用真实 embedding API 建向量索引。
    #    generate_answer=True 时，query engine 会再调用 LLM 生成 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())

Run it:

python quickstart.py

You should see output similar to:

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.

The first line is the answer generated from retrieved evidence. The second line is the source chunk that was retrieved.

This example does three things:

1. Writes raw/heta.txt into a LocalObjectStore.
2. Uses TextParser, SplitDocuments, and EmbedChunks to create chunks and embeddings.
3. Uses IndexVectors to build a vector index and queries it with vector_search.

What The Recipe Does

The recipe is Heta's main build unit:

KnowledgeRecipe
  parsers -> TextParser
  models  -> LanguageModel + EmbeddingModel
  stores  -> LocalObjectStore + InMemoryVectorStore
  steps   -> ParseDocuments -> SplitDocuments -> EmbedChunks -> IndexVectors

KnowledgeBase.create() executes this recipe. After the build, the KnowledgeBase only exposes the query modes that the recipe actually created.

In this minimal example:

available queries: vector_search

If you add more steps, new query modes become available.

Generated Files

The example creates a local workspace:

heta-demo-vector/
  objects/
    raw/
      heta.txt
    parsed/
      ...
    chunks/
      ...
    embeddings/
      ...
    _heta/
      knowledge_bases/
        home-vector/
          manifest.json
          latest_run.json
          runs/
            ...

Where:

• raw/ stores input files.
• parsed/ stores normalized ParsedDocument files.
• chunks/ stores ParsedChunk files.
• embeddings/ stores chunk embedding artifacts.
• _heta/knowledge_bases/... stores KB runtime metadata for load(), recovery, and delete().

This quickstart uses InMemoryVectorStore, so the vector index only lives in the current process. In production, replace it with MilvusVectorStore.

Add More Capabilities

Heta is designed for gradual composition. You do not need to choose the full architecture up front.

You want	Add	You get
Semantic retrieval	EmbedChunks + IndexVectors	vector_search
BM25-style keyword retrieval	IndexFullText + TextIndexStore	full_text_search
SQL text persistence	PersistChunks + SQLStore	sql_text_search
Heta-style graph retrieval	HetaGraphProcedure + SQL/vector stores	heta_graph_search
Hybrid / rerank / rewrite / multi-hop	Vector, full-text, and graph assets	Heta query modes
Recipe evaluation	BenchmarkRunner + benchmark adapter	EvaluationReport

Recommended next steps:

• To understand Recipe, read What Is A Recipe.
• To choose a build path, read Choose A Build Path.
• To query a KB, read Query A KnowledgeBase.
• To evaluate build strategies, read Evaluate A Recipe.

Add Heta Graph Search

To try Heta-style graph building, add HetaGraphProcedure after the minimal vector KB path.

This path extracts entities and relations from chunks, writes graph facts into SQL and vector stores, and then opens heta_graph_search.

It requires SQL support:

python -m pip install "heta-framework[sql]"

Create graph_quickstart.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. 准备一个干净 workspace。
    workspace = Path("heta-demo-graph")
    shutil.rmtree(workspace, ignore_errors=True)

    # 1. Stores：ObjectStore 管文件产物，SQLStore 落图谱 facts，VectorStore 支持图谱召回。
    objects = LocalObjectStore(workspace / "objects")
    sql = SQLStore(f"sqlite:///{workspace / 'graph.db'}")
    vectors = InMemoryVectorStore()

    # 2. Models：LLM 抽取 entity/relation，embedding model 为图谱 facts 建向量索引。
    #    运行前设置：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. 输入文档：Heta graph procedure 会从 chunk 中抽取 entity/relation。
    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 会展开为 extract entities、extract relations、build graph。
    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 会解锁 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())

Run it:

python graph_quickstart.py

You should see output similar to:

Heta creates a KnowledgeBase by building it from recipes. Specifically, the process involves running the steps outlined in the recipes to construct the KnowledgeBase [1][2][3].
relation Relation: Heta -> KnowledgeBase
Name: builds
Type: creates
Description: Heta builds knowledge bases from recipes.

This example uses:

• LocalObjectStore for raw text and intermediate artifacts.
• SQLStore for entities, relations, and evidence.
• InMemoryVectorStore for graph fact vectors.
• HetaGraphProcedure.build(deduplicate=False) to expand Heta-style graph steps.

In production, replace SQLite with PostgreSQL and the in-memory vector store with Milvus. The recipe structure does not need to change.

Replace Local Components

Recipes are not tied to a storage implementation. Production deployment usually replaces components, not steps:

object_store = S3ObjectStore(...)
vector_store = MilvusVectorStore(...)
sql_store = SQLStore("postgresql+psycopg://postgres:postgres@host:5432/postgres")

The same recipe still builds with:

kb = await KnowledgeBase.create(recipe=recipe, name="production-kb")