Skip to main content

Overview

TypeAgent uses 6 specialized indexes to enable fast knowledge lookup, semantic search, and temporal queries:
  1. Semantic Ref Index - Term to semantic reference mappings
  2. Property Index - Name/value property lookups
  3. Timestamp Index - Date range queries
  4. Message Text Index - Semantic search over messages
  5. Related Terms Index - Synonyms and fuzzy matching
  6. Conversation Threads - Thread-based organization

ITermToSemanticRefIndex

from typeagent.knowpro.interfaces import ITermToSemanticRefIndex
Primary knowledge index mapping terms (words/phrases) to semantic references.

Methods

size

async def size(self) -> int
Get the total number of unique terms in the index.

get_terms

async def get_terms(self) -> list[str]
Retrieve all indexed terms.
terms
list[str]
List of all unique terms in lowercase.

add_term

async def add_term(
    self,
    term: str,
    semantic_ref_ordinal: SemanticRefOrdinal | ScoredSemanticRefOrdinal,
) -> str
Add a term-to-semantic-reference mapping.
term
str
required
The term to index (automatically lowercased).
semantic_ref_ordinal
SemanticRefOrdinal | ScoredSemanticRefOrdinal
required
The semantic reference ordinal, optionally with a relevance score.
normalized_term
str
The normalized (lowercased) term that was indexed.

lookup_term

async def lookup_term(self, term: str) -> list[ScoredSemanticRefOrdinal] | None
Look up semantic references for a term.
term
str
required
The term to search for (case-insensitive).
results
list[ScoredSemanticRefOrdinal] | None
List of semantic reference ordinals with scores, or None if term not found.
Example:
index = await storage.get_semantic_ref_index()

# Look up term
results = await index.lookup_term("machine learning")
if results:
    for scored_ref in results:
        semref = await semantic_refs.get_item(scored_ref.semantic_ref_ordinal)
        print(f"Score {scored_ref.score}: {semref.knowledge}")

remove_term

async def remove_term(
    self,
    term: str,
    semantic_ref_ordinal: SemanticRefOrdinal
) -> None
Remove a specific term-to-semantic-reference mapping.

clear

async def clear(self) -> None
Remove all term mappings from the index.

IPropertyToSemanticRefIndex

from typeagent.knowpro.interfaces import IPropertyToSemanticRefIndex
Index for fast retrieval of name-value properties extracted from entities.

Methods

add_property

async def add_property(
    self,
    property_name: str,
    value: str,
    semantic_ref_ordinal: SemanticRefOrdinal | ScoredSemanticRefOrdinal,
) -> None
Add a property-value pair linked to a semantic reference.
property_name
str
required
Name of the property (e.g., "color", "size", "status").
value
str
required
Value of the property (e.g., "red", "large", "completed").
semantic_ref_ordinal
SemanticRefOrdinal | ScoredSemanticRefOrdinal
required
The semantic reference containing this property.

lookup_property

async def lookup_property(
    self,
    property_name: str,
    value: str
) -> list[ScoredSemanticRefOrdinal] | None
Find semantic references with a specific property value.
property_name
str
required
Name of the property to search.
value
str
required
Value to match.
results
list[ScoredSemanticRefOrdinal] | None
Matching semantic references, or None if not found.
Example:
index = await storage.get_property_index()

# Find all entities with status="completed"
results = await index.lookup_property("status", "completed")
if results:
    print(f"Found {len(results)} completed items")

get_values

async def get_values(self) -> list[str]
Retrieve all unique values across all properties.

size / clear / remove_property / remove_all_for_semref

See protocol definition for additional maintenance methods.

ITimestampToTextRangeIndex

from typeagent.knowpro.interfaces import ITimestampToTextRangeIndex
Index for temporal queries over message ranges.

Contract

  • Timestamps must be ISO-8601 strings sortable lexicographically
  • lookup_range(DateRange) returns items with start <= t < end (end exclusive)
  • If end is None, treats as point query with end = start + epsilon

Methods

add_timestamp

async def add_timestamp(
    self,
    message_ordinal: MessageOrdinal,
    timestamp: str
) -> bool
Add a message timestamp to the index.
message_ordinal
MessageOrdinal
required
The message ordinal.
timestamp
str
required
ISO-8601 timestamp with ‘Z’ suffix (e.g., "2024-03-06T10:00:00Z").
added
bool
True if timestamp was added, False if already existed.

add_timestamps

async def add_timestamps(
    self,
    message_timestamps: list[tuple[MessageOrdinal, str]]
) -> None
Batch add multiple timestamps.
message_timestamps
list[tuple[MessageOrdinal, str]]
required
List of (message_ordinal, timestamp) pairs.

lookup_range

async def lookup_range(
    self,
    date_range: DateRange
) -> list[TimestampedTextRange]
Find messages within a date range.
date_range
DateRange
required
Date range with start and optional end datetime.
ranges
list[TimestampedTextRange]
List of timestamped text ranges (message locations with timestamps).
Example:
from datetime import datetime, timezone
from typeagent.knowpro.interfaces import DateRange

index = await storage.get_timestamp_index()

# Find messages in March 2024
start = datetime(2024, 3, 1, tzinfo=timezone.utc)
end = datetime(2024, 4, 1, tzinfo=timezone.utc)

results = await index.lookup_range(DateRange(start, end))
print(f"Found {len(results)} messages in March 2024")

for ttr in results:
    print(f"  {ttr.timestamp}: message {ttr.range.start.message_ordinal}")

IMessageTextIndex

from typeagent.knowpro.interfaces import IMessageTextIndex
Semantic search index over message content using embeddings.

Type Parameters

TMessage
IMessage
The message type being indexed.

Methods

add_messages

async def add_messages(
    self,
    messages: Iterable[TMessage],
) -> None
Add messages to the text index.
messages
Iterable[TMessage]
required
Messages to index.

lookup_messages

async def lookup_messages(
    self,
    message_text: str,
    max_matches: int | None = None,
    threshold_score: float | None = None,
) -> list[ScoredMessageOrdinal]
Search for messages semantically similar to the query text.
message_text
str
required
Natural language query text.
max_matches
int | None
default:"None"
Maximum number of results to return.
threshold_score
float | None
default:"None"
Minimum similarity score (0.0 to 1.0).
results
list[ScoredMessageOrdinal]
List of message ordinals with relevance scores, sorted by score descending.
Example:
index = await storage.get_message_text_index()

# Semantic search
results = await index.lookup_messages(
    "database performance optimization",
    max_matches=10,
    threshold_score=0.7
)

for scored in results:
    msg = await messages.get_item(scored.message_ordinal)
    print(f"Score {scored.score:.3f}: {msg.text_chunks[0][:100]}...")

lookup_messages_in_subset

async def lookup_messages_in_subset(
    self,
    message_text: str,
    ordinals_to_search: list[MessageOrdinal],
    max_matches: int | None = None,
    threshold_score: float | None = None,
) -> list[ScoredMessageOrdinal]
Search within a specific subset of messages.
List of message ordinals to search within.

ITermToRelatedTermsIndex

from typeagent.knowpro.interfaces import ITermToRelatedTermsIndex
Index for term synonyms and fuzzy term matching.

Properties

aliases
ITermToRelatedTerms
Exact synonym/alias mappings.
fuzzy_index
ITermToRelatedTermsFuzzy | None
Fuzzy matching index using embeddings (may be None).

ITermToRelatedTerms (Aliases)

lookup_term

async def lookup_term(self, text: str) -> list[Term] | None
Look up exact synonyms/aliases for a term.
text
str
required
The term to look up.
List of related terms with optional weights, or None if not found.
async def add_related_term(
    self,
    text: str,
    related_terms: Term | list[Term]
) -> None
Add synonym/alias mappings.
text
str
required
The primary term.
Related term(s) to associate.
Example:
from typeagent.knowpro.interfaces import Term

index = await storage.get_related_terms_index()
aliases = index.aliases

# Add synonyms
await aliases.add_related_term(
    "ML",
    [Term("machine learning"), Term("artificial intelligence", weight=0.8)]
)

# Look up
related = await aliases.lookup_term("ML")
if related:
    print(f"ML is related to: {[t.text for t in related]}")

ITermToRelatedTermsFuzzy (Fuzzy Index)

add_terms

async def add_terms(self, texts: list[str]) -> None
Add terms to the fuzzy index for embedding-based matching.

lookup_term

async def lookup_term(
    self,
    text: str,
    max_hits: int | None = None,
    min_score: float | None = None,
) -> list[Term]
Find terms similar to the query using embeddings.
text
str
required
Query term.
max_hits
int | None
default:"None"
Maximum number of results.
min_score
float | None
default:"None"
Minimum similarity score.
similar_terms
list[Term]
List of similar terms with similarity weights.
Example:
index = await storage.get_related_terms_index()
fuzzy = index.fuzzy_index

if fuzzy:
    # Add terms
    await fuzzy.add_terms(["database", "SQL", "query", "index"])
    
    # Fuzzy lookup
    similar = await fuzzy.lookup_term("databse", max_hits=3)  # Typo
    for term in similar:
        print(f"Did you mean '{term.text}'? (score: {term.weight:.3f})")

IConversationThreads

from typeagent.knowpro.interfaces import IConversationThreads
Index for organizing conversations into thematic threads.

Properties

threads
list[Thread]
List of all conversation threads.

Methods

add_thread

async def add_thread(self, thread: Thread) -> None
Add a new thread to the conversation.
thread
Thread
required
Thread with description and text ranges.

lookup_thread

async def lookup_thread(
    self,
    thread_description: str,
    max_matches: int | None = None,
    threshold_score: float | None = None,
) -> list[ScoredThreadOrdinal] | None
Find threads matching a description.
thread_description
str
required
Natural language description of the thread.
results
list[ScoredThreadOrdinal] | None
Matching thread ordinals with scores, or None if no matches.
Example:
from typeagent.knowpro.interfaces import Thread, TextRange, TextLocation

threads = await storage.get_conversation_threads()

# Add thread
thread = Thread(
    description="Discussion about API design patterns",
    ranges=[
        TextRange(TextLocation(10, 0), TextLocation(15, 0)),
        TextRange(TextLocation(20, 0), TextLocation(22, 0))
    ]
)
await threads.add_thread(thread)

# Search threads
results = await threads.lookup_thread("REST API design")
if results:
    for scored in results:
        thread = threads.threads[scored.thread_ordinal]
        print(f"Score {scored.score:.3f}: {thread.description}")

IConversationSecondaryIndexes

from typeagent.knowpro.interfaces import IConversationSecondaryIndexes
Container holding all 6 secondary indexes for a conversation.

Properties

property_to_semantic_ref_index
IPropertyToSemanticRefIndex | None
Property index.
timestamp_index
ITimestampToTextRangeIndex | None
Timestamp index.
Related terms index.
threads
IConversationThreads | None
Conversation threads.
message_index
IMessageTextIndex[TMessage] | None
Message text index.
Example:
conv = await create_conversation(dbname="chat.db", message_type=ConversationMessage)

indexes = conv.secondary_indexes
if indexes:
    if indexes.property_index:
        print("Property index available")
    if indexes.timestamp_index:
        print("Timestamp index available")
    if indexes.message_index:
        print("Message text search available")

Build docs developers (and LLMs) love