通过 Heta 构建你想要的 知识库。
Heta 将知识库构建拆成清晰的组件：models、stores、parsers、 steps、search modes 和 benchmarks。你可以先搭建一个简单的向量知识库， 再按需要加入关键词检索、Heta 式图谱知识和评测能力。
工作方式
Heta 不要求你一次写完一整套复杂 RAG 流程。你先用 Recipe 选好 models、stores、parsers 和 steps，Heta 再按这个 Recipe 构建 KnowledgeBase。 建好以后，系统会知道它支持哪些 Search 方式，也可以直接跑 Benchmark。
01
Recipe
Recipe 就是知识库的配置清单。你在这里写清楚要用哪个 model、把文件存到哪里、 用哪些 parser、按什么 steps 构建。以后要复用或调整知识库，只需要改这份 Recipe。
02
Steps
Steps 是真正执行构建的步骤，比如 parse、split、embed、index 或 build graph。 你可以只用向量检索需要的 steps，也可以继续加入 Heta graph 相关 steps。
03
Search
Search 会根据 KnowledgeBase 已经建好的内容工作。建了 vector index 就能用 vector search，建了 text index 就能用 full-text search，建了 graph 就能用 Heta graph search。
04
Benchmark
Benchmark 用同一个 Recipe 自动建库、发起 query，并生成 evaluation report。 这样你可以比较不同 Recipe 的效果，而不是只凭感觉判断哪套方案更好。
四个 case
下面四个 case 使用本地 ObjectStore 和内存 store，模型默认走常见的 OpenAI LLM 与 embedding API。换成 Qwen、Milvus、 PostgreSQL 或 Elasticsearch 时，只需要替换对应 component。
$
OPENAI_API_KEY=... PYTHONPATH=src python docs/examples/home_vector_case.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())
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.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. 准备一个干净 workspace。
workspace = Path("heta-demo-full-text")
shutil.rmtree(workspace, ignore_errors=True)
# 1. Stores：ObjectStore 保存文档产物，TextIndexStore 保存全文索引。
objects = LocalObjectStore(workspace / "objects")
text_index = InMemoryTextIndexStore()
# 2. Model：full_text_search 不依赖 LLM；这里提供 LLM 是为了生成 answer。
# 运行前设置：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. 输入文档：全文索引适合精确术语、编号、缩写和关键词。
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 会把 chunk 写入 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 会解锁 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.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())
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.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. 准备一个最小 benchmark workspace。
workspace = Path("heta-demo-benchmark")
shutil.rmtree(workspace, ignore_errors=True)
workspace.mkdir(parents=True)
# 1. Benchmark data：JsonlBenchmark 需要 documents.jsonl 和 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 会用同一个 Recipe 先建库，再发起 query。
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：这里评测一个最小 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：用 evidence_recall@1 评估 top-1 evidence 是否命中。
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.jsonBenchmark 支持
Benchmark adapter 负责准备数据、构建 KnowledgeBase、运行 query modes， 并生成 evaluation report。你可以使用内置 benchmark，也可以按协议接入自己的业务评测集。
MultiHop-RAG
多跳问答 benchmark，适合验证复杂 query、证据召回和 multi-hop search。
Dataset
UDA-Benchmark
真实文档分析 benchmark，支持按 case 构建多个 KB 来评测不同 Recipe。
Source documents