Memory¶
aigise.memory.search.search_controller ¶
Search controller for orchestrating memory searches.
STRATEGY_REGISTRY = {'embedding_search': EmbeddingSearchStrategy, 'keyword_search': KeywordSearchStrategy, 'title_browse': TitleBrowseStrategy} module-attribute ¶
logger = logging.getLogger(__name__) module-attribute ¶
DomainConfig dataclass ¶
Configuration for a knowledge domain in the memory system.
A domain defines a coherent set of node types, relationships, and search strategies for a particular area of knowledge (e.g., code, Q&A pairs, documentation).
Multiple domains can be combined to create a rich knowledge graph.
default_strategy: str = 'embedding_search' class-attribute instance-attribute ¶
Default search strategy if LLM doesn't select one.
description: str = '' class-attribute instance-attribute ¶
Human-readable description of the domain.
embedding_dimension: int = 3072 class-attribute instance-attribute ¶
Default embedding dimension for this domain (Gemini default).
name: str instance-attribute ¶
Unique identifier for this domain.
node_types: Dict[str, NodeTypeConfig] = field(default_factory=dict) class-attribute instance-attribute ¶
Node type configurations keyed by label.
relationships: Dict[str, RelationshipConfig] = field(default_factory=dict) class-attribute instance-attribute ¶
Relationship configurations keyed by type name.
search_strategies: List[str] = field(default_factory=list) class-attribute instance-attribute ¶
Ordered list of search strategy names to try.
__post_init__() ¶
Validate and register the domain configuration.
get_node_labels() -> List[str] ¶
Get all node labels in this domain.
get_node_type(label: str) -> Optional[NodeTypeConfig] ¶
Get a node type configuration by label.
get_relationship(type_name: str) -> Optional[RelationshipConfig] ¶
Get a relationship configuration by type name.
get_relationship_types() -> List[str] ¶
Get all relationship types in this domain.
get_similarity_searchable_types() -> List[str] ¶
Get node types that support similarity search.
merge_with(other: 'DomainConfig') -> 'DomainConfig' ¶
Merge this domain with another, creating a combined configuration.
The other domain's configurations take precedence on conflicts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other | 'DomainConfig' | Another domain configuration to merge with. | required |
Returns:
| Type | Description |
|---|---|
'DomainConfig' | New merged DomainConfig. |
validate() -> List[str] ¶
Validate the domain configuration.
Returns:
| Type | Description |
|---|---|
List[str] | List of validation error messages (empty if valid). |
MemorySearchController ¶
Controller for orchestrating memory search operations.
This controller: 1. Selects appropriate search strategies (LLM-driven or heuristic) 2. Executes searches with multi-round refinement 3. Evaluates result sufficiency 4. Combines results from multiple strategies if needed
__init__(domain_config: Optional['DomainConfig'] = None, max_iterations: int = 3, sufficiency_threshold: int = 1, use_llm_selection: bool = True) ¶
Initialize the search controller.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
domain_config | Optional['DomainConfig'] | Domain configuration defining available strategies. | None |
max_iterations | int | Maximum number of refinement iterations. | 3 |
sufficiency_threshold | int | Minimum results needed to consider sufficient. | 1 |
use_llm_selection | bool | Whether to use LLM for strategy selection. | True |
add_strategy(name: str, strategy: SearchStrategy) -> None ¶
Add a custom search strategy.
get_available_strategies() -> List[str] ¶
Get names of available strategies.
search(query: str, node_types: Optional[List[str]] = None, client: Any = None, max_results: int = 10, min_score: float = 0.0, metadata: Optional[Dict[str, Any]] = None) -> SearchResult async ¶
Execute a memory search.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query | str | The search query. | required |
node_types | Optional[List[str]] | Optional list of node types to search. | None |
client | Any | Neo4j client for executing queries. | None |
max_results | int | Maximum results to return. | 10 |
min_score | float | Minimum score threshold. | 0.0 |
metadata | Optional[Dict[str, Any]] | Additional search context. | None |
Returns:
| Type | Description |
|---|---|
SearchResult | SearchResult with found items and metadata. |
SearchContext dataclass ¶
Context for a search operation.
domain_config: Optional['DomainConfig'] = None class-attribute instance-attribute ¶
Domain configuration to use.
max_results: int = 10 class-attribute instance-attribute ¶
Maximum number of results to return.
metadata: Dict[str, Any] = field(default_factory=dict) class-attribute instance-attribute ¶
Additional context metadata.
min_score: float = 0.0 class-attribute instance-attribute ¶
Minimum score threshold.
node_types: Optional[List[str]] = None class-attribute instance-attribute ¶
Restrict search to specific node types (None = all).
query: str instance-attribute ¶
The search query.
SearchResult dataclass ¶
Result of a memory search operation.
has_results: bool property ¶
Check if any results were found.
items: List[SearchResultItem] = field(default_factory=list) class-attribute instance-attribute ¶
Search result items.
iterations: int = 1 class-attribute instance-attribute ¶
Number of search iterations performed.
metadata: Dict[str, Any] = field(default_factory=dict) class-attribute instance-attribute ¶
Additional search metadata.
strategy_used: str = '' class-attribute instance-attribute ¶
Name of the strategy that found results.
sufficient: bool = False class-attribute instance-attribute ¶
Whether the results are considered sufficient.
total_found: int = 0 class-attribute instance-attribute ¶
Total number of results found (before limiting).
get_best_result() -> Optional[SearchResultItem] ¶
Get the highest-scoring result.
SearchResultItem dataclass ¶
A single search result item.
highlight: Optional[str] = None class-attribute instance-attribute ¶
Highlighted match context (for keyword search).
match_type: str = 'exact' class-attribute instance-attribute ¶
Type of match: 'exact', 'similarity', 'keyword', 'browse'.
node_id: str instance-attribute ¶
The Neo4j element ID of the matched node.
node_label: str instance-attribute ¶
The Neo4j label of the matched node.
properties: Dict[str, Any] instance-attribute ¶
Properties of the matched node.
score: float = 1.0 class-attribute instance-attribute ¶
Relevance score (0.0 to 1.0).
get_display_text() -> str ¶
Get a human-readable display text for this result.
SearchStrategy ¶
Bases: ABC
Abstract base class for search strategies.
A search strategy defines how to search for relevant information in the memory graph. Different strategies may use vector similarity, keyword matching, graph traversal, etc.
can_handle_query(query: str, context: SearchContext) -> float async ¶
Estimate how well this strategy can handle a query.
This method helps the search controller select the best strategy. Override to provide better estimates based on query characteristics.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query | str | The search query. | required |
context | SearchContext | Search context. | required |
Returns:
| Type | Description |
|---|---|
float | Confidence score from 0.0 (cannot handle) to 1.0 (ideal match). |
get_description() -> str abstractmethod ¶
Get a human-readable description of this strategy.
get_name() -> str abstractmethod ¶
Get the unique name of this strategy.
get_supported_node_types() -> Optional[List[str]] ¶
Get the list of node types this strategy supports.
Returns:
| Type | Description |
|---|---|
Optional[List[str]] | List of supported node types, or None for all types. |
search(context: SearchContext, client: Any) -> List[SearchResultItem] abstractmethod async ¶
Execute the search strategy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context | SearchContext | Search context with query and parameters. | required |
client | Any | Neo4j client for executing queries. | required |
Returns:
| Type | Description |
|---|---|
List[SearchResultItem] | List of search result items, ordered by relevance. |
supports_node_type(node_type: str) -> bool ¶
Check if this strategy supports a particular node type.
Override in subclasses to restrict to specific node types.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_type | str | Node label to check. | required |
Returns:
| Type | Description |
|---|---|
bool | True if this strategy can search this node type. |
get_memory_settings() -> MemorySettings ¶
Get the global memory settings instance.
Returns:
| Name | Type | Description |
|---|---|---|
MemorySettings | MemorySettings | The singleton settings instance. |
aigise.memory.tools.memory_search_tools ¶
Memory search tools for agents.
_search_controller: Optional[MemorySearchController] = None module-attribute ¶
logger = logging.getLogger(__name__) module-attribute ¶
MemorySearchController ¶
Controller for orchestrating memory search operations.
This controller: 1. Selects appropriate search strategies (LLM-driven or heuristic) 2. Executes searches with multi-round refinement 3. Evaluates result sufficiency 4. Combines results from multiple strategies if needed
__init__(domain_config: Optional['DomainConfig'] = None, max_iterations: int = 3, sufficiency_threshold: int = 1, use_llm_selection: bool = True) ¶
Initialize the search controller.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
domain_config | Optional['DomainConfig'] | Domain configuration defining available strategies. | None |
max_iterations | int | Maximum number of refinement iterations. | 3 |
sufficiency_threshold | int | Minimum results needed to consider sufficient. | 1 |
use_llm_selection | bool | Whether to use LLM for strategy selection. | True |
add_strategy(name: str, strategy: SearchStrategy) -> None ¶
Add a custom search strategy.
get_available_strategies() -> List[str] ¶
Get names of available strategies.
search(query: str, node_types: Optional[List[str]] = None, client: Any = None, max_results: int = 10, min_score: float = 0.0, metadata: Optional[Dict[str, Any]] = None) -> SearchResult async ¶
Execute a memory search.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query | str | The search query. | required |
node_types | Optional[List[str]] | Optional list of node types to search. | None |
client | Any | Neo4j client for executing queries. | None |
max_results | int | Maximum results to return. | 10 |
min_score | float | Minimum score threshold. | 0.0 |
metadata | Optional[Dict[str, Any]] | Additional search context. | None |
Returns:
| Type | Description |
|---|---|
SearchResult | SearchResult with found items and metadata. |
_get_search_controller() -> MemorySearchController ¶
Get or create the search controller singleton.
get_entity_by_id(node_label: str, node_key: Dict[str, Any], *, include_relationships: bool = False, tool_context: ToolContext) -> Dict[str, Any] async ¶
Get a specific entity from memory by its identifier.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_label | str | Label of the node (e.g., "Question", "Answer"). | required |
node_key | Dict[str, Any] | Properties to identify the node (e.g., {"answer_id": "..."}). | required |
include_relationships | bool | Whether to include connected entities. | False |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
get_merged_domain(*domain_names: str) -> DomainConfig ¶
Get a merged domain from multiple registered domains.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*domain_names | str | Names of domains to merge. | () |
Returns:
| Type | Description |
|---|---|
DomainConfig | Merged domain configuration. |
Raises:
| Type | Description |
|---|---|
ValueError | If any domain name is not found. |
get_related_entities(node_label: str, node_key: Dict[str, Any], *, relationship_types: Optional[List[str]] = None, direction: str = 'both', max_results: int = 10, tool_context: ToolContext) -> Dict[str, Any] async ¶
Get entities related to a specific node in the memory graph.
Use this to explore connections between cached knowledge, such as finding all topics related to a question or all content that references a specific topic.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_label | str | Label of the node to find relations for (e.g., "Topic"). | required |
node_key | Dict[str, Any] | Properties to identify the node (e.g., {"name": "authentication"}). | required |
relationship_types | Optional[List[str]] | Optional list of relationship types to follow. Options: "ABOUT", "HAS_ANSWER", "HAS_TOPIC", "RELATED_TO", "MENTIONS". | None |
direction | str | Direction to traverse relationships: - "outgoing": source -> target (what this node points to) - "incoming": source <- target (what points to this node) - "both": both directions (default) | 'both' |
max_results | int | Maximum number of results. Default is 10. | 10 |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Example
Find all Text/Question nodes that have the "authentication" topic¶
get_related_entities( node_label="Topic", node_key={"name": "authentication"}, relationship_types=["HAS_TOPIC"], direction="incoming", # Text --[HAS_TOPIC]--> Topic )
list_memory_contents(*, node_types: Optional[List[str]] = None, limit: int = 20, offset: int = 0, order_by: str = 'created_at', tool_context: ToolContext) -> Dict[str, Any] async ¶
List contents of the memory graph by node type.
Use this to browse what's in memory without a specific search query. Useful for exploring available cached knowledge.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_types | Optional[List[str]] | Node types to list. Defaults to ["Question", "Topic"]. | None |
limit | int | Maximum items per type. Default is 20. | 20 |
offset | int | Number of items to skip (for pagination). Default is 0. | 0 |
order_by | str | Property to order by. Default is "created_at". | 'created_at' |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
requires_sandbox(*sandbox_types: str) -> Callable[[F], F] ¶
Universal decorator for declaring sandbox dependencies.
This decorator works for both: - Tool functions: Only marks the function with metadata - Toolset factories: Marks function AND injects metadata into returned instance
The decorator is purely declarative - it does not create or fetch sandboxes. It only adds __sandbox_requirements__ metadata for static analysis via collect_sandbox_dependencies().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*sandbox_types | str | Variable number of sandbox type names that the tool or toolset depends on (e.g., "main", "gdb_mcp", "neo4j"). | () |
Returns:
| Type | Description |
|---|---|
Callable[[F], F] | A decorator function that adds sandbox_requirements metadata. |
safe_tool_execution(func: F) -> F ¶
Decorator to wrap tool functions with error handling.
Catches all exceptions and returns a formatted error message with backtrace. Works for both sync and async functions.
Returns:
| Type | Description |
|---|---|
F | dict with "error" key containing failure message and backtrace |
search_memory(query: str, *, node_types: Optional[List[str]] = None, max_results: int = 5, min_score: float = 0.5, tool_context: ToolContext) -> Dict[str, Any] async ¶
Search the memory graph for relevant cached knowledge.
Use this tool to find cached Q&A pairs, topics, or code entities that are semantically similar to your query.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query | str | The search query - describe what you're looking for. | required |
node_types | Optional[List[str]] | Optional list of node types to search. Defaults to ["Question", "Topic"]. Other options: "Answer", "Function", "Class", "File". | None |
max_results | int | Maximum number of results to return. Default is 5. | 5 |
min_score | float | Minimum similarity score (0-1). Default is 0.5. | 0.5 |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
aigise.memory.tools.memory_update_tools ¶
Memory update tools for agents.
_update_controller: Optional[MemoryUpdateController] = None module-attribute ¶
logger = logging.getLogger(__name__) module-attribute ¶
GraphOperations ¶
Executes graph operations for adding, updating, and deleting nodes/relationships.
This class handles: - Creating new nodes with proper MERGE handling - Updating existing nodes - Creating relationships between nodes - Ensuring indexes exist
__init__(domain_config: Optional['DomainConfig'] = None) ¶
Initialize graph operations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
domain_config | Optional['DomainConfig'] | Domain configuration for node/relationship schemas. | None |
add_entities_batch(entities: List[ExtractedEntity], client: Any, aigise_session_id: Optional[str] = None) -> List[OperationResult] async ¶
Add multiple entities to the graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
entities | List[ExtractedEntity] | Entities to add. | required |
client | Any | Neo4j client. | required |
aigise_session_id | Optional[str] | Optional session ID. | None |
Returns:
| Type | Description |
|---|---|
List[OperationResult] | List of operation results. |
add_entity(entity: ExtractedEntity, client: Any, aigise_session_id: Optional[str] = None) -> OperationResult async ¶
Add an entity to the graph.
Uses MERGE to avoid duplicates based on unique keys.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
entity | ExtractedEntity | Entity to add. | required |
client | Any | Neo4j client. | required |
aigise_session_id | Optional[str] | Optional session ID for tracking. | None |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult with operation details. |
add_relationship(relationship: DiscoveredRelationship, client: Any) -> OperationResult async ¶
Add a relationship to the graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
relationship | DiscoveredRelationship | Relationship to add. | required |
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult with operation details. |
add_relationships_batch(relationships: List[DiscoveredRelationship], client: Any) -> List[OperationResult] async ¶
Add multiple relationships to the graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
relationships | List[DiscoveredRelationship] | Relationships to add. | required |
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
List[OperationResult] | List of operation results. |
delete_entity(label: str, match_key: Dict[str, Any], client: Any) -> OperationResult async ¶
Delete a node from the graph.
Uses DETACH DELETE to remove the node and all its relationships.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
label | str | Node label (e.g., "Question", "Topic"). | required |
match_key | Dict[str, Any] | Properties to identify the node to delete. | required |
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult with operation details. |
delete_relationship(rel_type: str, source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], client: Any) -> OperationResult async ¶
Delete a relationship from the graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rel_type | str | Type of relationship to delete. | required |
source_label | str | Label of source node. | required |
source_key | Dict[str, Any] | Properties to identify source node. | required |
target_label | str | Label of target node. | required |
target_key | Dict[str, Any] | Properties to identify target node. | required |
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult with operation details. |
ensure_indexes(client: Any) -> bool async ¶
Ensure required indexes exist.
Creates both regular and vector indexes for memory nodes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
bool | True if indexes were created/verified successfully. |
MemoryUpdateController ¶
Controller for orchestrating memory update operations.
This controller: 1. Extracts entities from content (using EntityExtractor) 2. Discovers relationships between entities (using RelationshipDiscoverer) 3. Decides operation type for each entity (using LLMOperationDecider) 4. Executes graph operations (using GraphOperations)
__init__(domain_config: Optional['DomainConfig'] = None, use_llm_extraction: bool = True, generate_embeddings: bool = True, similarity_threshold: float = 0.7, use_llm_decision: bool = False) ¶
Initialize the update controller.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
domain_config | Optional['DomainConfig'] | Domain configuration defining entity types. | None |
use_llm_extraction | bool | Whether to use LLM for semantic extraction. | True |
generate_embeddings | bool | Whether to generate embeddings for entities. | True |
similarity_threshold | float | Threshold for similarity-based relationships. | 0.7 |
use_llm_decision | bool | Whether to use LLM for operation type decisions. | False |
delete_entity(label: str, match_key: Dict[str, Any], client: Any) -> OperationResult async ¶
Delete an entity from the memory graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
label | str | Node label (e.g., "Question", "Topic"). | required |
match_key | Dict[str, Any] | Properties to identify the node. | required |
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult with operation details. |
delete_relationship(rel_type: str, source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], client: Any) -> OperationResult async ¶
Delete a relationship from the memory graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rel_type | str | Relationship type. | required |
source_label | str | Label of source node. | required |
source_key | Dict[str, Any] | Properties to identify source node. | required |
target_label | str | Label of target node. | required |
target_key | Dict[str, Any] | Properties to identify target node. | required |
client | Any | Neo4j client. | required |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult with operation details. |
link_entities(source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], relationship_type: str, client: Any, properties: Optional[Dict[str, Any]] = None) -> OperationResult async ¶
Create a relationship between two existing entities.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
source_label | str | Label of source node. | required |
source_key | Dict[str, Any] | Properties to identify source node. | required |
target_label | str | Label of target node. | required |
target_key | Dict[str, Any] | Properties to identify target node. | required |
relationship_type | str | Type of relationship to create. | required |
client | Any | Neo4j client. | required |
properties | Optional[Dict[str, Any]] | Optional relationship properties. | None |
Returns:
| Type | Description |
|---|---|
OperationResult | OperationResult for the relationship creation. |
store_knowledge(content: str, content_type: str = 'text', client: Any = None, aigise_session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> UpdateResult async ¶
Store knowledge in the memory graph.
Generic method for storing any type of content.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content | str | Content to store. | required |
content_type | str | Type of content ('text', 'code', 'question', 'answer'). | 'text' |
client | Any | Neo4j client. | None |
aigise_session_id | Optional[str] | Optional session ID. | None |
metadata | Optional[Dict[str, Any]] | Additional metadata. | None |
Returns:
| Type | Description |
|---|---|
UpdateResult | UpdateResult with operation details. |
store_knowledge_with_decision(content: str, content_type: str = 'text', client: Any = None, aigise_session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> UpdateResult async ¶
Store knowledge using LLM to decide operation type for each entity.
This method uses the LLMOperationDecider to intelligently decide whether to ADD, UPDATE, DELETE, or skip each extracted entity based on what already exists in the graph.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content | str | Content to store. | required |
content_type | str | Type of content. | 'text' |
client | Any | Neo4j client. | None |
aigise_session_id | Optional[str] | Optional session ID. | None |
metadata | Optional[Dict[str, Any]] | Additional metadata. | None |
Returns:
| Type | Description |
|---|---|
UpdateResult | UpdateResult with operation details. |
store_qa_pair(question: str, answer: str, answering_agent: str, answering_model: str, client: Any, aigise_session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> UpdateResult async ¶
Store a question-answer pair in the memory graph.
This is the main entry point for storing Q&A knowledge. It extracts entities, discovers relationships, and persists to Neo4j.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
question | str | The question text. | required |
answer | str | The answer text. | required |
answering_agent | str | Name of the agent that generated the answer. | required |
answering_model | str | Model used to generate the answer. | required |
client | Any | Neo4j client. | required |
aigise_session_id | Optional[str] | Optional session ID for tracking. | None |
metadata | Optional[Dict[str, Any]] | Additional metadata to store. | None |
Returns:
| Type | Description |
|---|---|
UpdateResult | UpdateResult with operation details. |
_get_update_controller() -> MemoryUpdateController ¶
Get or create the update controller singleton.
cache_qa_pair(question: str, answer: str, answering_agent: str, answering_model: str, *, metadata: Optional[Dict[str, Any]] = None, tool_context: ToolContext) -> Dict[str, Any] async ¶
Cache a question-answer pair in the memory graph.
Use this tool AFTER successfully answering a question to store it for future retrieval. This enables the agent to quickly answer similar questions in the future.
The system automatically: - Generates embeddings for similarity search - Extracts topics and code references - Creates relationships to related entities - Finds connections to existing cached questions
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
question | str | The question text that was asked. | required |
answer | str | The answer text that was generated. | required |
answering_agent | str | Name of the agent that generated the answer. | required |
answering_model | str | Model identifier used to generate the answer. | required |
metadata | Optional[Dict[str, Any]] | Optional additional metadata to store. | None |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
delete_from_memory(node_label: str, node_key: Dict[str, Any], *, tool_context: ToolContext) -> Dict[str, Any] async ¶
Delete an entity from the memory graph.
Use this to remove outdated or incorrect cached knowledge. This will also remove all relationships connected to the node.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_label | str | Label of the node to delete. Options: - "Question": A cached question - "Answer": A cached answer - "Topic": A topic/concept - "Function": A function entity - "Class": A class entity - "File": A file entity | required |
node_key | Dict[str, Any] | Properties to identify the node. Examples: - {"question_hash": "..."} for Question - {"name": "..."} for Topic - {"name": "...", "file_path": "..."} for Function/Class - {"path": "..."} for File | required |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
delete_relationship_from_memory(relationship_type: str, source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], *, tool_context: ToolContext) -> Dict[str, Any] async ¶
Delete a relationship from the memory graph.
Use this to remove incorrect or outdated connections between entities.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
relationship_type | str | Type of relationship to delete. Options: - "ABOUT": What something is about - "HAS_ANSWER": Question to answer link - "HAS_TOPIC": Content to topic link - "RELATED_TO": General relation - "MENTIONS": Content mentions code entity | required |
source_label | str | Label of the source node. | required |
source_key | Dict[str, Any] | Properties to identify the source node. | required |
target_label | str | Label of the target node. | required |
target_key | Dict[str, Any] | Properties to identify the target node. | required |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
ensure_memory_indexes(*, tool_context: ToolContext) -> Dict[str, Any] async ¶
Ensure memory graph indexes are created.
This tool creates the necessary indexes for efficient querying: - Regular indexes for exact match lookups - Vector indexes for similarity search
You typically don't need to call this manually as it's called automatically during the first cache operation.
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
get_merged_domain(*domain_names: str) -> DomainConfig ¶
Get a merged domain from multiple registered domains.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*domain_names | str | Names of domains to merge. | () |
Returns:
| Type | Description |
|---|---|
DomainConfig | Merged domain configuration. |
Raises:
| Type | Description |
|---|---|
ValueError | If any domain name is not found. |
link_entities(source_label: str, source_key: Dict[str, Any], target_label: str, target_key: Dict[str, Any], relationship_type: str, *, properties: Optional[Dict[str, Any]] = None, tool_context: ToolContext) -> Dict[str, Any] async ¶
Create a relationship between two entities in the memory graph.
Use this to manually connect related knowledge that the automatic extraction might have missed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
source_label | str | Label of the source node (e.g., "Question"). | required |
source_key | Dict[str, Any] | Properties to identify source (e.g., {"question_hash": "..."}). | required |
target_label | str | Label of the target node (e.g., "Topic"). | required |
target_key | Dict[str, Any] | Properties to identify target (e.g., {"name": "..."}). | required |
relationship_type | str | Type of relationship. Options: - "ABOUT": What something is about - "HAS_ANSWER": Question to answer link - "HAS_TOPIC": Content to topic link - "RELATED_TO": General relation - "MENTIONS": Content mentions code entity | required |
properties | Optional[Dict[str, Any]] | Optional properties for the relationship. | None |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
requires_sandbox(*sandbox_types: str) -> Callable[[F], F] ¶
Universal decorator for declaring sandbox dependencies.
This decorator works for both: - Tool functions: Only marks the function with metadata - Toolset factories: Marks function AND injects metadata into returned instance
The decorator is purely declarative - it does not create or fetch sandboxes. It only adds __sandbox_requirements__ metadata for static analysis via collect_sandbox_dependencies().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*sandbox_types | str | Variable number of sandbox type names that the tool or toolset depends on (e.g., "main", "gdb_mcp", "neo4j"). | () |
Returns:
| Type | Description |
|---|---|
Callable[[F], F] | A decorator function that adds sandbox_requirements metadata. |
safe_tool_execution(func: F) -> F ¶
Decorator to wrap tool functions with error handling.
Catches all exceptions and returns a formatted error message with backtrace. Works for both sync and async functions.
Returns:
| Type | Description |
|---|---|
F | dict with "error" key containing failure message and backtrace |
store_knowledge(content: str, content_type: str = 'text', *, metadata: Optional[Dict[str, Any]] = None, tool_context: ToolContext) -> Dict[str, Any] async ¶
Store knowledge in the memory graph.
Use this tool to persist any kind of knowledge for future retrieval. The system will automatically extract entities, topics, and relationships from the content.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
content | str | The content to store. | required |
content_type | str | Type of content. Options: - "text": Generic text content - "code": Code content (will extract functions/classes) - "question": A question to cache - "answer": An answer to cache | 'text' |
metadata | Optional[Dict[str, Any]] | Optional additional metadata to store. | None |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with: |
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|
Dict[str, Any] |
|