Overview
TypeAgent uses 6 specialized indexes to enable fast knowledge lookup, semantic search, and temporal queries:
- Semantic Ref Index - Term to semantic reference mappings
- Property Index - Name/value property lookups
- Timestamp Index - Date range queries
- Message Text Index - Semantic search over messages
- Related Terms Index - Synonyms and fuzzy matching
- 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.
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.
The term to index (automatically lowercased).
semantic_ref_ordinal
SemanticRefOrdinal | ScoredSemanticRefOrdinal
required
The semantic reference ordinal, optionally with a relevance score.
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.
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.
Name of the property (e.g., "color", "size", "status").
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.
Name of the property to search.
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.
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.
ISO-8601 timestamp with ‘Z’ suffix (e.g., "2024-03-06T10:00:00Z").
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 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
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.
Natural language query text.
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.
ordinals_to_search
list[MessageOrdinal]
required
List of message ordinals to search within.
from typeagent.knowpro.interfaces import ITermToRelatedTermsIndex
Index for term synonyms and fuzzy term matching.
Properties
Exact synonym/alias mappings.
fuzzy_index
ITermToRelatedTermsFuzzy | None
Fuzzy matching index using embeddings (may be None).
lookup_term
async def lookup_term(self, text: str) -> list[Term] | None
Look up exact synonyms/aliases for a term.
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.
related_terms
Term | list[Term]
required
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]}")
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.
Maximum number of results.
min_score
float | None
default:"None"
Minimum similarity score.
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
List of all conversation threads.
Methods
add_thread
async def add_thread(self, thread: Thread) -> None
Add a new thread to the conversation.
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.
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.
term_to_related_terms_index
ITermToRelatedTermsIndex | None
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")