Document knowledge

Store text, search it by keywords and return passages with citations. This chapter also covers revisions, paging, scoped corpora and semantic retrieval. The SDK and a running server are enough for features/knowledge: it builds answers from stored text without a model provider.

What the knowledge store is

Use app.knowledge to access documents in your namespace's global scope. Its Knowledge wrapper calls the server's seven POST /api/v2/knowledge/{put,put-batch,get,list,delete,search,stats} operations. The server splits text into chunks, indexes them and ranks keyword matches with BM25. The SDK checks bounds first, so invalid IDs and oversized text raise ValueError before a network request.

Knowledge follows memory ownership and scopes. Authentication selects the namespace. Inside a traversal, requests carry the task lease so the server can resolve session and workflow ownership.

Keep these retrieval properties in mind:

• BM25 matches query terms, rather than their meaning. A search for "luggage" will not find a passage containing only "baggage".
• Each match has a doc_id#ordinal citation. It remains valid until the document is rewritten, letting a reader locate the quoted passage.

Ingest documents

Call put for one document, or put_many for an atomic batch of up to 100. Supply text for server chunking or your own chunks. The knowledge example exposes batch ingestion through a traversal:

from __future__ import annotations

from typing import Any, Dict, List

from atfgraph import Agent
from atfgraph.knowledge import Knowledge

app = Agent(node_id="knowledge_desk", version="1.0.0", description="Document knowledge with cited BM25 retrieval")

knowledge = Knowledge(lambda: app.client)


@app.traversal()
async def ingest(documents: List[Dict[str, Any]]) -> dict:
    """Write documents atomically. Each item: ``doc_id``, ``title``, ``metadata`` and
    ``text`` or ``chunks``."""
    result = await knowledge.put_many(documents)
    return {
        "documents": result.documents,
        "chunks": result.chunks,
        "items": [{"doc_id": item.doc_id, "revision": item.revision, "chunks": item.chunks, "created": item.created} for item in result.items],
    }

The explicitly constructed Knowledge(lambda: app.client) provides the same interface as app.knowledge. Its fixture documents are in examples/python/features/knowledge/fixtures.py. One supplies plain text; another supplies chunks with page numbers and metadata:

from __future__ import annotations

from typing import Any, Dict, List

DOCUMENTS: List[Dict[str, Any]] = [
    {
        "doc_id": "policy:refunds",
        "title": "Refund policy",
        "metadata": {"kind": "policy", "owner": "finance"},
        "text": (
            "Refunds are issued to the original form of payment within 14 days of approval. "
            "Refund fees apply to non-flexible fares; flexible fares are refunded in full."
        ),
    },
    {
        "doc_id": "policy:baggage",
        "title": "Baggage allowance",
        "metadata": {"kind": "policy", "owner": "operations"},
        "chunks": [
            {"text": "Checked baggage allowance is 23 kg per passenger on domestic sectors and 30 kg on international sectors.", "page": 1},
            {"text": "Excess baggage fees apply above the allowance and are charged per kilogram at the airport.", "page": 2, "metadata": {"section": "fees"}},
        ],
    },
]

Each batch item accepts doc_id, either text or chunks, title, metadata, chunk_size, chunk_overlap and expected_revision. Other fields, or repeated document IDs within one batch, raise ValueError before submission.

Limits and chunking

Use chunk_size and chunk_overlap only with text; supplying them alongside explicit chunks raises an error. atfgraph.chunk_text(text, size, overlap) previews the server's chunking algorithm, as checked by test_chunk_text_mirrors_the_server_rules. The server's final split determines citations.

By default, text and queries are normalized to Unicode NFC so equivalent composed and decomposed characters match. Set Knowledge(lambda: app.client, normalize=False) to send the original text unchanged.

Replace and compare-and-set

Writing an existing doc_id replaces it and increments its revision. Use expected_revision=0 to create only if absent, or supply the current revision to avoid overwriting another update. A mismatch raises ConflictError. Conditional deletion uses the same check and returns False when the document is already absent. The real-daemon knowledge test covers these cases.

Search with citations

search(query, top_k=, offset=, filters=, doc_id=, min_score=) returns KnowledgeSearchResult. Its KnowledgeHit items are sorted by descending score, then document ID and chunk ordinal. Metadata filters use exact matches; doc_id restricts the search to one document. This is the example's ask traversal:

from __future__ import annotations

from typing import List, Optional

from pydantic import BaseModel, Field

from atfgraph import Agent
from atfgraph.knowledge import Knowledge

app = Agent(node_id="knowledge_desk", version="1.0.0")
knowledge = Knowledge(lambda: app.client)


class Passage(BaseModel):
    citation: str
    title: str
    text: str
    score: float
    page: Optional[int] = None
    matched_terms: List[str] = Field(default_factory=list)


class Answer(BaseModel):
    question: str
    terms: List[str]
    passages: List[Passage]
    total: int
    scanned: int


@app.traversal()
async def ask(question: str, top_k: int = 5, kind: Optional[str] = None) -> Answer:
    result = await knowledge.search(question, top_k=top_k, filters={"kind": kind} if kind else None)
    return Answer(
        question=question,
        terms=result.terms,
        passages=[
            Passage(citation=hit.citation, title=hit.title, text=hit.text, score=hit.score, page=hit.page, matched_terms=hit.matched_terms)
            for hit in result.items
        ],
        total=result.total,
        scanned=result.scanned,
    )

In the daemon test, search("refund fees", top_k=3) finds policy:refunds#0 first, with query terms ["refund", "fees"] and matched terms ["fees", "refund"]. total counts matching chunks after filtering, scanned counts them before filtering, and corpus_chunks is the scope's size. next_offset indicates another page.

Use these helpers when assembling an answer or prompt:

• result.citations() returns the citation IDs.
• hit.reference() formats a title and locator, such as Refund policy (policy:refunds#0), including a page when known, for example (b#2, p. 4).
• result.context(max_chars=6000) creates numbered passages and trims at passage boundaries so retained citations still identify complete passages.

The example's extractive_answer selects sentences covering the most query terms and labels them with passage numbers. For model-written answers, follow examples/python/docs_assistant/main.py: reject citation IDs that were not present in the retrieved evidence.

Paging through a ranking

Iterate search_all(query, page_size=100, max_results=10000) to follow offset pages automatically in the same deterministic ranking. The contract-double test checks that the citations match a single search covering the same results.

Scopes and layered knowledge

Choose another scope with app.knowledge.scope("session"), scope("actor") or scope("workflow"). During execution, the server resolves session and workflow IDs from the task. Outside execution, provide session_id= or root_id=.

Knowledge.search_across([session_kb, kb], query, top_k=5) merges results from multiple corpora by score, then scope order, document ID and ordinal. Scores from separate corpora are only approximately comparable. Each hit retains its scope label. For example:

from __future__ import annotations

import asyncio
import os

from atfgraph import Client
from atfgraph.knowledge import Knowledge


async def main() -> None:
    async with Client(os.getenv("ATFGRAPH_SERVER")) as client:
        kb = Knowledge.for_client(client)
        session_kb = kb.scope("session", session_id="se_demo")
        await session_kb.put("note", "Traveller prefers refund to voucher.", title="Session note")
        merged = await Knowledge.search_across([session_kb, kb], "refund voucher lounge", top_k=5)
        for hit in merged.items:
            print(hit.scope, hit.reference(), round(hit.score, 3))


if __name__ == "__main__":
    asyncio.run(main())

Use Knowledge.for_client(client) in a script or test. Inside an agent, use app.knowledge.

Catalog, statistics and deletion

stats() reports document, chunk, token, term and posting counts, average chunk size, strategy == "bm25" and server limits. list(prefix=, cursor=, limit=) returns document summaries in ID order and a next_cursor; documents(prefix=, page_size=) follows those pages. get(doc_id, include_chunks=True) returns the document and citations, or found == False. The example combines these in catalog and forget:

from __future__ import annotations

from atfgraph import Agent
from atfgraph.knowledge import Knowledge

app = Agent(node_id="knowledge_desk", version="1.0.0")
knowledge = Knowledge(lambda: app.client)


@app.traversal()
async def catalog() -> dict:
    stats = await knowledge.stats()
    documents = [
        {"doc_id": item.doc_id, "title": item.title, "chunks": item.chunks, "revision": item.revision, "metadata": item.metadata}
        async for item in knowledge.documents(page_size=100)
    ]
    return {"documents": stats.documents, "chunks": stats.chunks, "terms": stats.terms, "strategy": stats.strategy, "items": documents}


@app.traversal()
async def forget(doc_id: str) -> dict:
    return {"doc_id": doc_id, "deleted": await knowledge.delete(doc_id)}

For bulk deletion, list the matching documents and delete them individually. The docs assistant uses documents(prefix="docs:<namespace>:") for this pattern. There is no delete-by-prefix operation.

Run the example

Run the worker and driver from the repository root, with ATFGRAPH_SERVER pointing to your server. This fixture example needs no model settings.

python -m atfgraph.worker --module examples.python.features.knowledge.agent:app
python -m examples.python.features.knowledge.call

The real-daemon knowledge test uses the fixture described in Testing. The same test module includes an in-memory KnowledgeFakeServer for fast SDK tests. That double tests client behavior; the daemon test establishes integration with the actual server.

When to use vector memory instead

Use knowledge storage when you need text passages and citations. Use vector memory to store embeddings you provide, along with metadata, and rank them by cosine similarity. Its set_vector, get_vector, delete_vector, similarity_search and search methods do not themselves create a text corpus or citation locators.

To combine them, call kb.semantic(EmbeddingConfig(provider=..., model=...)). The returned SemanticKnowledge writes the document to knowledge storage, then embeds its server-produced chunks into scoped vector keys shaped as kb:<index>:<doc_id>#<ordinal>. Search supports hybrid, semantic and lexical modes. Hybrid search combines BM25 and cosine rankings using reciprocal rank. Semantic matches are checked against the live document revision so stale vectors are dropped.

Supply an embedding callable from texts to vectors, or use http_embedding_provider(url, model, api_key=...) with an OpenAI-compatible endpoint. The retrieval example reads ATFGRAPH_EMBEDDING_URL and ATFGRAPH_EMBEDDING_MODEL; without them it uses a deterministic local lexicon embedder for offline examples. retrieval_basics and docs_assistant use the same layer, covered by tests/test_semantic_knowledge.py.

Not verified: actor-scoped search against a real daemon by the tests cited here. test_knowledge.py covers global and session scopes end to end but checks the actor request only through mock transport. Confirm actor behavior with kb.scope("actor").put(...) and .search(...) against a configured server.

Next

• Review Memory scopes and vectors.
• Build cited answers with Model calls.
• Use the real-daemon test fixture.