Quick Start

这个页面用一个本地文本文件跑通第一个 Heta KnowledgeBase。

第一遍只构建最小向量知识库：

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

这条路径需要的组件最少，适合先确认安装、模型调用、parser、chunk 和向量检索都能正常工作。后续再按需要加入 full-text search、Heta graph search 或 benchmark。

Install

Heta 已发布到 PyPI。安装时使用包名 heta-framework，代码中使用导入名 heta_framework。

最小向量示例只需要核心包：

python -m pip install heta-framework

如果你的项目需要生产存储或全文索引，可以按需安装 extra：

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

设置模型 API key：

export OPENAI_API_KEY="sk-..."

Heta 的模型层由 LiteLLM 驱动，model_name 使用 LiteLLM 的模型命名方式，例如 openai/gpt-4o-mini、openai/text-embedding-3-small。

Build Your First KnowledgeBase

创建 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())

运行：

python quickstart.py

你会看到类似输出：

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.

第一行是 query engine 用检索结果生成的 answer。第二行是命中的原始 chunk evidence。

这个示例已经完成了三件事：

1. 把 raw/heta.txt 写入 LocalObjectStore。
2. 用 TextParser、SplitDocuments 和 EmbedChunks 生成 chunk 与 embedding。
3. 用 IndexVectors 建立向量索引，并通过 vector_search 查询。

What The Recipe Does

示例中的 recipe 是 Heta 的核心构建单元：

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

KnowledgeBase.create() 会执行这份 recipe。构建完成后，KnowledgeBase 只开放当前 recipe 真正构建出来的 query mode。

在这个最小示例里：

available queries: vector_search

如果继续添加其他 steps，新的 query mode 会随之开放。

Generated Files

示例会生成一个本地 workspace：

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

其中：

• raw/ 保存输入文件。
• parsed/ 保存统一的 ParsedDocument。
• chunks/ 保存切分后的 ParsedChunk。
• embeddings/ 保存 chunk embedding 产物。
• _heta/knowledge_bases/... 保存 KB 的运行记录，供 load()、失败恢复和 delete() 使用。

这个 quickstart 使用 InMemoryVectorStore，所以向量索引只存在于当前进程中。生产环境可以替换为 MilvusVectorStore。

Add More Capabilities

Heta 的构建方式是逐步组合，不需要一开始就选择完整方案。

你想要	添加什么	会得到
语义检索	EmbedChunks + IndexVectors	vector_search
BM25-style 关键词检索	IndexFullText + TextIndexStore	full_text_search
SQL 文本持久化	PersistChunks + SQLStore	sql_text_search
Heta 式图谱检索	HetaGraphProcedure + SQL/vector stores	heta_graph_search
混合检索 / rerank / rewrite / multi-hop	vector、full-text、graph 资产组合	Heta query modes
评估 recipe	BenchmarkRunner + benchmark adapter	EvaluationReport

下一步建议：

• 想知道 Recipe 是什么，看 What Is A Recipe。
• 想选择构建路径，看 Choose A Build Path。
• 想理解查询方式，看 Query A KnowledgeBase。
• 想评估不同构建方案，看 Evaluate A Recipe。

Add Heta Graph Search

如果你想继续体验 Heta 式建图，可以在最小向量知识库之后加入 HetaGraphProcedure。

这条路径会从 chunk 中抽取 entities 和 relations，并把图谱 facts 写入 SQL 与 vector stores，最后开放 heta_graph_search。

它需要 SQL 支持：

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

创建 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())

运行：

python graph_quickstart.py

你会看到类似输出：

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.

这个例子使用：

• LocalObjectStore 保存原始文本和中间产物。
• SQLStore 保存实体、关系和 evidence。
• InMemoryVectorStore 保存图谱 facts 的向量索引。
• HetaGraphProcedure.build(deduplicate=False) 展开 Heta 式建图 steps。

生产环境可以把 SQLite 换成 PostgreSQL，把内存向量库换成 Milvus。Recipe 的整体结构不需要改变。

Replace Local Components

Recipe 不绑定具体存储实现。生产环境通常只替换 components，steps 可以保持不变：

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

同一份 recipe 仍然通过：

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

完成构建。