Skip to content

trading_agent.tools.search_knowledge_tool

The search_knowledge tool: searches the user's documents in workspace/shared/knowledge/.

Typical usage example:

tool = SearchKnowledgeTool(knowledge_base)

SearchKnowledgeArguments

Bases: BaseModel

The arguments of search_knowledge.

Attributes:

Name Type Description
query str

A few words to look for.
limit int

The most passages to return.

SearchKnowledgeTool 

SearchKnowledgeTool(knowledge_base)

Bases: Tool

Returns the passages of the knowledge base that best match a query.

Creates the tool.

Parameters:

Name Type Description Default
knowledge_base KnowledgeBase

The index.

 required
Source code in src/trading_agent/tools/search_knowledge_tool.py 
38
39
40
41
42
43
44
def __init__(self, knowledge_base: knowledge_base_module.KnowledgeBase):
    """Creates the tool.

    Args:
        knowledge_base (knowledge_base_module.KnowledgeBase): The index.
    """
    self._knowledge_base = knowledge_base

run async 

run(arguments)

Brings the index up to date, then searches it, in a worker thread.

Parameters:

Name Type Description Default
arguments SearchKnowledgeArguments

The query and the limit.

 required

Returns:

Type Description
ToolResult

The passages, labelled as outside data, or a note that nothing matched.
Source code in src/trading_agent/tools/search_knowledge_tool.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def run(self, arguments: SearchKnowledgeArguments) -> tool.ToolResult:
    """Brings the index up to date, then searches it, in a worker thread.

    Args:
        arguments (SearchKnowledgeArguments): The query and the limit.

    Returns:
        tool.ToolResult: The passages, labelled as outside data, or a note that nothing matched.
    """
    counts = await asyncio.to_thread(self._knowledge_base.refresh)
    if counts.passages == 0:
        return tool.ToolResult.success(
            'The knowledge base is empty: the user has not put any .md, .txt or .csv files in workspace/shared/knowledge/ yet.'
        )
    hits = await asyncio.to_thread(
        self._knowledge_base.search,
        arguments.query,
        arguments.limit,
    )
    if not hits:
        return tool.ToolResult.success(
            f'Nothing in the knowledge base matches {arguments.query!r}.'
        )
    lines = []
    for hit in hits:
        lines.append(f'[{hit.path} · {hit.heading}]')
        lines.append(hit.snippet)
        lines.append('')
    return tool.ToolResult.success(
        untrusted_text.UntrustedText().wrap(
            'the knowledge base',
            '\n'.join(lines).strip(),
        )
    )