trading_agent.knowledge.knowledge_base

A searchable index of the user's documents in workspace/shared/knowledge/.

Put Markdown, text or CSV files in the folder, such as strategy write-ups, broker rules or exchange circulars, and every desk can search them with search_knowledge. Each file is cut into passages of about 1,500 characters, kept under the nearest Markdown heading, and indexed with SQLite's FTS5 full-text search. Words are matched by their stem, so "gapped" finds "gap", and results are ranked by BM25, the usual relevance score of search engines.

The index lives in workspace/shared/knowledge.db. Before each search, files that were added, changed or removed since the last search are indexed again, so there is no separate import step.

Typical usage example:

knowledge = KnowledgeBase(Path('workspace/shared/knowledge'), Path('workspace/shared/knowledge.db')) knowledge.refresh() hits = knowledge.search('opening range breakout', 5)

INDEXED_SUFFIXES module-attribute

INDEXED_SUFFIXES = ('.md', '.markdown', '.txt', '.csv')

KnowledgeHit dataclass

KnowledgeHit(path, heading, snippet)

One passage that matched.

Attributes:

Name	Type	Description
path	str	The file, relative to the knowledge folder.
heading	str	The Markdown heading the passage sits under, or the file name.
snippet	str	The matching part, with matched words in [brackets].

RefreshCounts dataclass

RefreshCounts(indexed, removed, passages)

What one refresh did.

Attributes:

Name	Type	Description
indexed	int	Files indexed because they were new or changed.
removed	int	Files dropped because they were deleted.
passages	int	Passages in the index afterwards.

KnowledgeBase

KnowledgeBase(knowledge_directory, index_file)

The index over one knowledge folder.

Describes the knowledge base without touching the disk.

Parameters:

Name	Type	Description	Default
knowledge_directory	Path	The folder of documents.	required
index_file	Path	The SQLite file holding the index.	required

Source code in src/trading_agent/knowledge/knowledge_base.py

def __init__(self, knowledge_directory: Path, index_file: Path):
    """Describes the knowledge base without touching the disk.

    Args:
        knowledge_directory (Path): The folder of documents.
        index_file (Path): The SQLite file holding the index.
    """
    self._knowledge_directory = knowledge_directory
    self._index_file = index_file

refresh

refresh()

Indexes new and changed files and forgets deleted ones.

Returns:

Type	Description
RefreshCounts	What changed.

Source code in src/trading_agent/knowledge/knowledge_base.py

def refresh(self) -> RefreshCounts:
    """Indexes new and changed files and forgets deleted ones.

    Returns:
        RefreshCounts: What changed.
    """
    current = {}
    for path in self._documents():
        status = path.stat()
        relative = str(path.relative_to(self._knowledge_directory))
        current[relative] = (status.st_mtime, status.st_size)
    with self._connection() as connection:
        known = {}
        for row in connection.execute(
            'select path, modified, size from files'
        ):
            known[row[0]] = (row[1], row[2])
        indexed = 0
        for relative, signature in current.items():
            if known.get(relative) == signature:
                continue
            self._index_file_passages(connection, relative, signature)
            indexed += 1
        removed = 0
        for relative in known:
            if relative not in current:
                connection.execute(
                    'delete from passages where path = ?', (relative,)
                )
                connection.execute(
                    'delete from files where path = ?', (relative,)
                )
                removed += 1
        passages = connection.execute(
            'select count(*) from passages'
        ).fetchone()[0]
    return RefreshCounts(
        indexed=indexed, removed=removed, passages=passages
    )

search

search(query, limit)

Finds the passages that best match the query.

Passages holding every word come first; when there are none, passages holding any of the words are returned.

Parameters:

Name	Type	Description	Default
query	str	Words to look for.	required
limit	int	The most passages to return.	required

Returns:

Type	Description
list[KnowledgeHit]	The best passages, most relevant first.

Source code in src/trading_agent/knowledge/knowledge_base.py

def search(self, query: str, limit: int) -> list[KnowledgeHit]:
    """Finds the passages that best match the query.

    Passages holding every word come first; when there are none, passages holding any of the words are returned.

    Args:
        query (str): Words to look for.
        limit (int): The most passages to return.

    Returns:
        list[KnowledgeHit]: The best passages, most relevant first.
    """
    words = _WORD_PATTERN.findall(query.lower())
    if not words:
        return []
    quoted = []
    for word in words:
        quoted.append(f'"{word}"')
    with self._connection() as connection:
        hits = self._match(connection, ' AND '.join(quoted), limit)
        if not hits:
            hits = self._match(connection, ' OR '.join(quoted), limit)
    return hits