Merge pull request #24 from Hawksight-AI/staging

Reorganize context module with registry, methods, and config
This commit is contained in:
Mohd Kaif
2025-11-15 21:53:04 +05:30
committed by GitHub
9 changed files with 1804 additions and 14 deletions
+180 -2
View File
@@ -6,23 +6,79 @@ formalizing context as a graph of connections to enable meaningful agent
understanding and memory. It integrates RAG with knowledge graphs to provide
persistent context for intelligent agents.
Algorithms Used:
Context Graph Construction:
- Graph Building: Node and edge construction from entities and relationships
- Entity Extraction: Entity extraction from conversations and text
- Relationship Extraction: Relationship extraction from conversations
- Intent Extraction: Intent classification from conversations
- Sentiment Analysis: Sentiment extraction from conversations
- Graph Traversal: BFS/DFS for neighbor discovery and multi-hop traversal
- Graph Indexing: Type-based indexing for efficient node/edge lookup
Agent Memory Management:
- Vector Embedding: Embedding generation for memory items
- Vector Search: Similarity search in vector space
- Keyword Search: Fallback keyword-based search
- Retention Policy: Time-based memory retention and cleanup
- Memory Indexing: Deque-based memory index for efficient access
- Knowledge Graph Integration: Entity and relationship updates to knowledge graph
Context Retrieval:
- Vector Similarity Search: Cosine similarity in vector space
- Graph Traversal: Multi-hop graph expansion for related entities
- Memory Search: Vector and keyword search in memory store
- Result Ranking: Score-based ranking and merging
- Deduplication: Content-based result deduplication
- Hybrid Scoring: Weighted combination of multiple retrieval sources
Entity Linking:
- URI Generation: Hash-based and text-based URI assignment
- Text Similarity: Word overlap-based similarity calculation
- Knowledge Graph Lookup: Entity matching in knowledge graph
- Cross-Document Linking: Entity linking across multiple documents
- Bidirectional Linking: Symmetric relationship creation
- Entity Web Construction: Graph-based entity connection web
Key Features:
- Context graph construction from entities and relationships
- Context graph construction from entities, relationships, and conversations
- Agent memory management with RAG integration
- Entity linking across sources with URI assignment
- Hybrid context retrieval (vector + graph + memory)
- Conversation history management
- Context accumulation and synthesis
- Graph-based context traversal and querying
- Method registry for custom context methods
- Configuration management with environment variables and config files
Main Classes:
- ContextGraphBuilder: Builds context graphs from various sources
- ContextNode: Context graph node data structure
- ContextEdge: Context graph edge data structure
- AgentMemory: Manages persistent agent memory with RAG
- MemoryItem: Memory item data structure
- EntityLinker: Links entities across sources with URIs
- EntityLink: Entity link data structure
- LinkedEntity: Linked entity with context
- ContextRetriever: Retrieves relevant context from multiple sources
- RetrievedContext: Retrieved context item data structure
- MethodRegistry: Registry for custom context methods
- ContextConfig: Configuration manager for context module
Convenience Functions:
- build_context: Build context graph and manage memory in one call
Example Usage:
>>> from semantica.context import ContextGraphBuilder, AgentMemory
>>> from semantica.context import build_context, ContextGraphBuilder, AgentMemory
>>> # Using convenience function
>>> result = build_context(
... entities=entities,
... relationships=relationships,
... vector_store=vs,
... knowledge_graph=kg
... )
>>> # Using classes directly
>>> builder = ContextGraphBuilder()
>>> graph = builder.build_from_entities_and_relationships(entities, relationships)
>>> memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
@@ -33,12 +89,26 @@ Author: Semantica Contributors
License: MIT
"""
from typing import Any, Dict, List, Optional, Union
from pathlib import Path
from .context_graph import ContextGraphBuilder, ContextNode, ContextEdge
from .entity_linker import EntityLinker, EntityLink, LinkedEntity
from .agent_memory import AgentMemory, MemoryItem
from .context_retriever import ContextRetriever, RetrievedContext
from .registry import MethodRegistry, method_registry
from .methods import (
build_context_graph,
store_memory,
retrieve_context,
link_entities,
get_context_method,
list_available_methods,
)
from .config import ContextConfig, context_config
__all__ = [
# Main classes
"ContextGraphBuilder",
"ContextNode",
"ContextEdge",
@@ -49,4 +119,112 @@ __all__ = [
"MemoryItem",
"ContextRetriever",
"RetrievedContext",
# Registry
"MethodRegistry",
"method_registry",
# Methods
"build_context_graph",
"store_memory",
"retrieve_context",
"link_entities",
"get_context_method",
"list_available_methods",
# Config
"ContextConfig",
"context_config",
# Convenience
"build_context",
]
def build_context(
entities: Optional[List[Dict[str, Any]]] = None,
relationships: Optional[List[Dict[str, Any]]] = None,
conversations: Optional[List[Union[str, Dict[str, Any]]]] = None,
vector_store: Optional[Any] = None,
knowledge_graph: Optional[Any] = None,
graph_method: str = "entities_relationships",
store_initial_memories: bool = False,
**options
) -> Dict[str, Any]:
"""
Build context graph and optionally manage memory (convenience function).
This is a user-friendly wrapper that builds context graphs and optionally
stores initial memories in one call.
Args:
entities: List of entity dictionaries
relationships: List of relationship dictionaries
conversations: List of conversation files or dictionaries
vector_store: Vector store instance for memory
knowledge_graph: Knowledge graph instance
graph_method: Graph construction method (default: "entities_relationships")
store_initial_memories: Whether to store initial memories from entities
**options: Additional options passed to builders
Returns:
Dictionary containing:
- graph: Context graph dictionary
- memory_ids: List of stored memory IDs (if store_initial_memories=True)
- statistics: Graph and memory statistics
Examples:
>>> from semantica.context import build_context
>>> entities = [{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"}]
>>> relationships = [{"source_id": "e1", "target_id": "e2", "type": "related_to"}]
>>> result = build_context(
... entities=entities,
... relationships=relationships,
... vector_store=vs,
... knowledge_graph=kg
... )
>>> print(f"Graph has {result['graph']['statistics']['node_count']} nodes")
"""
from ..utils.logging import get_logger
logger = get_logger("context")
# Build context graph
graph = build_context_graph(
entities=entities,
relationships=relationships,
conversations=conversations,
method=graph_method,
**options
)
memory_ids = []
# Optionally store initial memories
if store_initial_memories and vector_store:
memory = AgentMemory(
vector_store=vector_store,
knowledge_graph=knowledge_graph,
**options
)
# Store entity-based memories
if entities:
for entity in entities[:10]: # Limit to first 10
entity_text = entity.get("text") or entity.get("label") or entity.get("name", "")
if entity_text:
memory_id = memory.store(
f"Entity: {entity_text}",
metadata={
"type": "entity",
"entity_id": entity.get("id"),
"entity_type": entity.get("type")
},
entities=[entity] if entity.get("id") else None
)
memory_ids.append(memory_id)
return {
"graph": graph,
"memory_ids": memory_ids,
"statistics": {
"graph": graph.get("statistics", {}),
"memories_stored": len(memory_ids)
}
}
+28 -3
View File
@@ -5,18 +5,42 @@ This module provides comprehensive agent memory management and context retrieval
integrating RAG (Retrieval-Augmented Generation) with knowledge graphs to give
agents persistent context across conversations and interactions.
Algorithms Used:
Memory Storage:
- Vector Embedding: Embedding generation for memory items using embedding models
- Vector Indexing: Vector store indexing for efficient similarity search
- Memory Indexing: Deque-based memory index for efficient temporal access
- Knowledge Graph Integration: Entity and relationship updates to knowledge graph
- Metadata Storage: Dictionary-based metadata storage and retrieval
Memory Retrieval:
- Vector Similarity Search: Cosine similarity search in vector space
- Keyword Search: Fallback keyword-based search using word overlap
- Score Ranking: Relevance score-based result ranking
- Filter Matching: Metadata-based filtering (type, date range, etc.)
- Result Deduplication: Content-based deduplication of results
Memory Management:
- Retention Policy: Time-based memory retention and cleanup
- Memory Statistics: Counter-based statistics tracking
- Conversation History: Temporal-based conversation history retrieval
- Memory Deletion: Cascading deletion from vector store and memory index
Key Features:
- Persistent memory storage for agents
- Vector-based context retrieval
- Vector-based context retrieval with embedding support
- Knowledge graph context integration
- Conversation history management
- Context accumulation over time
- Memory retrieval for agent decision-making
- Retention policy management
- Retention policy management (time-based cleanup)
- Memory statistics and analytics
- Metadata-based filtering and search
- Fallback keyword search when vector store unavailable
Main Classes:
- MemoryItem: Memory item data structure
- MemoryItem: Memory item data structure with content, timestamp, metadata, entities, relationships
- AgentMemory: Agent memory manager with RAG integration
Example Usage:
@@ -25,6 +49,7 @@ Example Usage:
>>> memory_id = memory.store("User asked about Python", metadata={"type": "conversation"})
>>> results = memory.retrieve("Python", max_results=5)
>>> history = memory.get_conversation_history(conversation_id="conv_123")
>>> stats = memory.get_statistics()
Author: Semantica Contributors
License: MIT
+131
View File
@@ -0,0 +1,131 @@
"""
Configuration Management Module for Context Engineering
This module provides centralized configuration management for context engineering operations,
supporting multiple configuration sources including environment variables, config files,
and programmatic configuration.
Supported Configuration Sources:
- Environment variables: CONTEXT_RETENTION_POLICY, CONTEXT_MAX_MEMORY_SIZE, etc.
- Config files: YAML, JSON, TOML formats
- Programmatic: Python API for setting context configurations
Algorithms Used:
- Environment Variable Parsing: OS-level environment variable access
- YAML Parsing: YAML parser for configuration file loading
- JSON Parsing: JSON parser for configuration file loading
- TOML Parsing: TOML parser for configuration file loading
- Fallback Chain: Priority-based configuration resolution
- Dictionary Merging: Deep merge algorithms for configuration updates
Key Features:
- Environment variable support for context parameters
- Config file support (YAML, JSON, TOML formats)
- Programmatic configuration via Python API
- Method-specific configuration management
- Automatic fallback chain (config file -> environment -> defaults)
- Global config instance for easy access
Main Classes:
- ContextConfig: Main configuration manager class for context module
Example Usage:
>>> from semantica.context.config import context_config
>>> retention = context_config.get("retention_policy", default="unlimited")
>>> context_config.set("retention_policy", "30_days")
>>> method_config = context_config.get_method_config("graph")
Author: Semantica Contributors
License: MIT
"""
import os
from typing import Optional, Dict, Any
from pathlib import Path
from ..utils.logging import get_logger
class ContextConfig:
"""Configuration manager for context module - supports .env files, environment variables, and programmatic config."""
def __init__(self, config_file: Optional[str] = None):
"""Initialize configuration manager."""
self.logger = get_logger("context_config")
self._configs: Dict[str, Any] = {}
self._method_configs: Dict[str, Dict] = {}
self._load_config_file(config_file)
self._load_env_vars()
def _load_config_file(self, config_file: Optional[str]):
"""Load configuration from file."""
if config_file and Path(config_file).exists():
try:
# Support YAML, JSON, TOML
if config_file.endswith('.yaml') or config_file.endswith('.yml'):
import yaml
with open(config_file, 'r') as f:
data = yaml.safe_load(f) or {}
self._configs.update(data.get("context", {}))
self._method_configs.update(data.get("context_methods", {}))
elif config_file.endswith('.json'):
import json
with open(config_file, 'r') as f:
data = json.load(f) or {}
self._configs.update(data.get("context", {}))
self._method_configs.update(data.get("context_methods", {}))
elif config_file.endswith('.toml'):
import toml
with open(config_file, 'r') as f:
data = toml.load(f) or {}
self._configs.update(data.get("context", {}))
self._method_configs.update(data.get("context_methods", {}))
self.logger.info(f"Loaded context config from {config_file}")
except Exception as e:
self.logger.warning(f"Failed to load config file {config_file}: {e}")
def _load_env_vars(self):
"""Load configuration from environment variables."""
# Context-specific environment variables with CONTEXT_ prefix
env_prefix = "CONTEXT_"
for key, value in os.environ.items():
if key.startswith(env_prefix):
config_key = key[len(env_prefix):].lower()
# Try to convert to appropriate type
if value.lower() in ('true', 'false'):
self._configs[config_key] = value.lower() == 'true'
elif value.isdigit():
self._configs[config_key] = int(value)
else:
try:
self._configs[config_key] = float(value)
except ValueError:
self._configs[config_key] = value
def set(self, key: str, value: Any):
"""Set a configuration value."""
self._configs[key] = value
def get(self, key: str, default: Any = None) -> Any:
"""Get a configuration value."""
return self._configs.get(key, default)
def set_method_config(self, method_name: str, config: Dict[str, Any]):
"""Set method-specific configuration."""
self._method_configs[method_name] = config
def get_method_config(self, method_name: str) -> Dict[str, Any]:
"""Get method-specific configuration."""
return self._method_configs.get(method_name, {})
def get_all(self) -> Dict[str, Any]:
"""Get all configurations."""
return {
"configs": self._configs.copy(),
"method_configs": self._method_configs.copy()
}
context_config = ContextConfig()
+29 -3
View File
@@ -6,8 +6,30 @@ formalizing context as a graph of connections. It turns context from intuition
into infrastructure, enabling meaningful connections between concepts, entities,
and conversations.
Algorithms Used:
Graph Construction:
- Node Creation: Dictionary-based node storage with type indexing
- Edge Creation: List-based edge storage with type indexing
- Entity Extraction: Entity extraction from conversations and text
- Relationship Extraction: Relationship extraction from conversations
- Intent Extraction: Intent classification from conversations
- Sentiment Analysis: Sentiment extraction from conversations
- URI Assignment: Entity linker-based URI assignment for entities
Graph Traversal:
- BFS (Breadth-First Search): Multi-hop neighbor discovery
- Graph Indexing: Type-based indexing for efficient node/edge lookup
- Neighbor Discovery: Outgoing and incoming edge traversal
- Multi-hop Expansion: Iterative expansion for related entities
Graph Querying:
- Type Filtering: Node type-based filtering
- Metadata Filtering: Dictionary-based metadata matching
- Graph Statistics: Node and edge type counting
Key Features:
- Builds context graphs from entities and relationships
- Builds context graphs from entities, relationships, and conversations
- Creates meaningful connections between concepts
- Assigns URLs/URIs to entities for web-like context
- Formalizes context into graph structure
@@ -15,10 +37,12 @@ Key Features:
- Enables context traversal and querying
- Conversation-based graph construction
- Intent and sentiment extraction
- Multi-hop relationship traversal
- Graph statistics and analytics
Main Classes:
- ContextNode: Context graph node data structure
- ContextEdge: Context graph edge data structure
- ContextNode: Context graph node data structure with node_id, node_type, content, metadata, properties
- ContextEdge: Context graph edge data structure with source_id, target_id, edge_type, weight, metadata
- ContextGraphBuilder: Context graph builder for formalizing context
Example Usage:
@@ -28,6 +52,7 @@ Example Usage:
>>> builder.add_node("node1", "entity", "Python programming")
>>> builder.add_edge("node1", "node2", "related_to", weight=0.9)
>>> neighbors = builder.get_neighbors("node1", max_hops=2)
>>> results = builder.query(node_type="entity", confidence=0.8)
Author: Semantica Contributors
License: MIT
@@ -40,6 +65,7 @@ from collections import defaultdict
from .entity_linker import EntityLinker
from ..utils.exceptions import ValidationError, ProcessingError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
from ..utils.types import EntityDict, RelationshipDict
+31 -4
View File
@@ -6,18 +6,44 @@ retrieving relevant context from memory, knowledge graphs, and vector stores
to inform decision-making. It supports hybrid retrieval combining multiple
sources for optimal context relevance.
Algorithms Used:
Vector Retrieval:
- Vector Similarity Search: Cosine similarity search in vector space
- Query Embedding: Embedding generation for search queries
- Top-K Retrieval: Top-K result selection based on similarity scores
Graph Retrieval:
- Keyword Matching: Word overlap-based node matching
- Graph Traversal: Multi-hop graph expansion for related entities
- Relevance Scoring: Word overlap-based relevance calculation
- Entity Expansion: BFS-based entity relationship traversal
Memory Retrieval:
- Memory Search: Vector and keyword search in memory store
- Conversation History: Temporal-based memory retrieval
Result Processing:
- Result Ranking: Score-based ranking and merging
- Deduplication: Content-based result deduplication
- Score Aggregation: Maximum score selection for duplicate results
- Metadata Merging: Dictionary-based metadata merging
- Entity Merging: Set-based entity deduplication
Key Features:
- Retrieve context from multiple sources (memory, graph, vector)
- Hybrid retrieval (vector + graph + memory)
- Context relevance ranking
- Hybrid retrieval (vector + graph + memory) with weighted combination
- Context relevance ranking and scoring
- Context aggregation and synthesis
- Ontology-aware context retrieval
- Real-time context updates
- Graph expansion for related entities
- Multi-hop relationship traversal
- Result deduplication and merging
- Configurable retrieval strategies
Main Classes:
- RetrievedContext: Retrieved context item data structure
- RetrievedContext: Retrieved context item data structure with content, score, source, metadata, related_entities, related_relationships
- ContextRetriever: Context retriever for hybrid retrieval
Example Usage:
@@ -25,7 +51,8 @@ Example Usage:
>>> retriever = ContextRetriever(memory_store=mem, knowledge_graph=kg, vector_store=vs)
>>> results = retriever.retrieve("Python programming", max_results=5)
>>> for result in results:
... print(result.content, result.score)
... print(f"{result.content}: {result.score:.2f}")
... print(f"Related entities: {len(result.related_entities)}")
Author: Semantica Contributors
License: MIT
+818
View File
@@ -0,0 +1,818 @@
# Context Module Usage Guide
This guide demonstrates how to use the context module for building context graphs, managing agent memory, retrieving context, and linking entities for intelligent agents.
## Table of Contents
1. [Basic Usage](#basic-usage)
2. [Context Graph Construction](#context-graph-construction)
3. [Agent Memory Management](#agent-memory-management)
4. [Context Retrieval](#context-retrieval)
5. [Entity Linking](#entity-linking)
6. [Using Methods](#using-methods)
7. [Using Registry](#using-registry)
8. [Configuration](#configuration)
9. [Advanced Examples](#advanced-examples)
## Basic Usage
### Using the Convenience Function
```python
from semantica.context import build_context
# Sample entities and relationships
entities = [
{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"},
{"id": "e2", "text": "Machine Learning", "type": "CONCEPT"},
]
relationships = [
{"source_id": "e1", "target_id": "e2", "type": "used_for"},
]
# Build context graph and optionally store memories
result = build_context(
entities=entities,
relationships=relationships,
vector_store=vs, # Optional vector store
knowledge_graph=kg, # Optional knowledge graph
graph_method="entities_relationships",
store_initial_memories=False
)
print(f"Graph has {result['graph']['statistics']['node_count']} nodes")
print(f"Graph has {result['graph']['statistics']['edge_count']} edges")
```
### Using Main Classes
```python
from semantica.context import ContextGraphBuilder, AgentMemory, ContextRetriever
# Step 1: Build context graph
builder = ContextGraphBuilder()
graph = builder.build_from_entities_and_relationships(entities, relationships)
# Step 2: Initialize agent memory
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
memory_id = memory.store("User asked about Python", metadata={"type": "conversation"})
# Step 3: Retrieve context
retriever = ContextRetriever(memory_store=memory, knowledge_graph=kg, vector_store=vs)
results = retriever.retrieve("Python programming", max_results=5)
```
## Context Graph Construction
### Building from Entities and Relationships
```python
from semantica.context import ContextGraphBuilder
builder = ContextGraphBuilder()
entities = [
{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"},
{"id": "e2", "text": "Machine Learning", "type": "CONCEPT"},
{"id": "e3", "text": "TensorFlow", "type": "FRAMEWORK"},
]
relationships = [
{"source_id": "e1", "target_id": "e2", "type": "used_for", "confidence": 0.9},
{"source_id": "e3", "target_id": "e2", "type": "implements", "confidence": 0.95},
]
graph = builder.build_from_entities_and_relationships(entities, relationships)
print(f"Nodes: {graph['statistics']['node_count']}")
print(f"Edges: {graph['statistics']['edge_count']}")
```
### Building from Conversations
```python
from semantica.context import ContextGraphBuilder
builder = ContextGraphBuilder()
conversations = [
{
"id": "conv1",
"content": "User asked about Python programming",
"entities": [
{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"}
],
"relationships": []
},
{
"id": "conv2",
"content": "User asked about machine learning",
"entities": [
{"id": "e2", "text": "Machine Learning", "type": "CONCEPT"}
],
"relationships": []
}
]
graph = builder.build_from_conversations(
conversations,
link_entities=True,
extract_intents=True,
extract_sentiments=True
)
```
### Adding Nodes and Edges Manually
```python
from semantica.context import ContextGraphBuilder
builder = ContextGraphBuilder()
# Add nodes
builder.add_node("node1", "entity", "Python programming", confidence=0.9)
builder.add_node("node2", "concept", "Machine Learning", confidence=0.95)
# Add edges
builder.add_edge("node1", "node2", "related_to", weight=0.9)
# Get neighbors
neighbors = builder.get_neighbors("node1", max_hops=2)
print(f"Neighbors: {neighbors}")
# Query graph
results = builder.query(node_type="entity", confidence=0.8)
```
### Using Graph Construction Methods
```python
from semantica.context.methods import build_context_graph
# Build from entities and relationships
graph = build_context_graph(
entities=entities,
relationships=relationships,
method="entities_relationships"
)
# Build from conversations
graph = build_context_graph(
conversations=conversations,
method="conversations"
)
# Hybrid construction
graph = build_context_graph(
entities=entities,
relationships=relationships,
conversations=conversations,
method="hybrid"
)
```
## Agent Memory Management
### Storing Memories
```python
from semantica.context import AgentMemory
memory = AgentMemory(
vector_store=vs,
knowledge_graph=kg,
retention_policy="30_days",
max_memory_size=10000
)
# Store a memory
memory_id = memory.store(
"User asked about Python programming",
metadata={
"type": "conversation",
"conversation_id": "conv_123",
"user_id": "user_456"
},
entities=[
{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"}
]
)
print(f"Stored memory: {memory_id}")
```
### Retrieving Memories
```python
from semantica.context import AgentMemory
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
# Retrieve memories
results = memory.retrieve(
"Python programming",
max_results=5,
min_score=0.5,
type="conversation"
)
for result in results:
print(f"Content: {result['content']}")
print(f"Score: {result['score']:.2f}")
print(f"Timestamp: {result['timestamp']}")
```
### Getting Specific Memory
```python
from semantica.context import AgentMemory
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
# Get specific memory
memory_item = memory.get_memory("mem_abc123")
if memory_item:
print(f"Content: {memory_item['content']}")
print(f"Metadata: {memory_item['metadata']}")
```
### Conversation History
```python
from semantica.context import AgentMemory
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
# Get conversation history
history = memory.get_conversation_history(
conversation_id="conv_123",
max_items=100
)
for item in history:
print(f"{item['timestamp']}: {item['content']}")
```
### Memory Management
```python
from semantica.context import AgentMemory
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
# Delete specific memory
memory.delete_memory("mem_abc123")
# Clear memories by filters
deleted_count = memory.clear_memory(
type="conversation",
start_date="2024-01-01",
end_date="2024-12-31"
)
print(f"Deleted {deleted_count} memories")
# Get statistics
stats = memory.get_statistics()
print(f"Total items: {stats['total_items']}")
print(f"Items by type: {stats['items_by_type']}")
```
### Using Memory Methods
```python
from semantica.context.methods import store_memory
# Store memory using method
memory_id = store_memory(
"User asked about Python",
vector_store=vs,
knowledge_graph=kg,
method="store",
metadata={"type": "conversation"}
)
# Store conversation memory
memory_id = store_memory(
"User asked about machine learning",
vector_store=vs,
knowledge_graph=kg,
method="conversation",
metadata={"conversation_id": "conv_123"}
)
```
## Context Retrieval
### Basic Retrieval
```python
from semantica.context import ContextRetriever
retriever = ContextRetriever(
memory_store=memory,
knowledge_graph=kg,
vector_store=vs,
use_graph_expansion=True,
max_expansion_hops=2
)
# Retrieve context
results = retriever.retrieve(
"Python programming",
max_results=5,
min_relevance_score=0.5
)
for result in results:
print(f"Content: {result.content}")
print(f"Score: {result.score:.2f}")
print(f"Source: {result.source}")
print(f"Related entities: {len(result.related_entities)}")
```
### Retrieval Methods
```python
from semantica.context.methods import retrieve_context
# Vector-based retrieval only
results = retrieve_context(
"Python programming",
vector_store=vs,
method="vector",
max_results=5
)
# Graph-based retrieval only
results = retrieve_context(
"Python programming",
knowledge_graph=kg,
method="graph",
max_results=5
)
# Memory-based retrieval only
results = retrieve_context(
"Python programming",
memory_store=memory,
method="memory",
max_results=5
)
# Hybrid retrieval (all sources)
results = retrieve_context(
"Python programming",
memory_store=memory,
knowledge_graph=kg,
vector_store=vs,
method="hybrid",
max_results=5
)
```
### Graph Expansion
```python
from semantica.context import ContextRetriever
retriever = ContextRetriever(
knowledge_graph=kg,
use_graph_expansion=True,
max_expansion_hops=3
)
# Retrieve with graph expansion
results = retriever.retrieve(
"Python",
max_results=10,
max_hops=3
)
for result in results:
print(f"Content: {result.content}")
print(f"Related entities: {len(result.related_entities)}")
for entity in result.related_entities[:3]:
print(f" - {entity['content']} (hop: {entity['hop']})")
```
## Entity Linking
### Basic Entity Linking
```python
from semantica.context import EntityLinker
linker = EntityLinker(
knowledge_graph=kg,
similarity_threshold=0.8,
base_uri="https://semantica.dev/entity/"
)
# Assign URI to entity
uri = linker.assign_uri(
"entity_1",
"Python",
"PROGRAMMING_LANGUAGE"
)
print(f"URI: {uri}")
# Link entities in text
entities = [
{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"},
{"id": "e2", "text": "Machine Learning", "type": "CONCEPT"},
]
linked_entities = linker.link(
"Python is used for machine learning",
entities=entities
)
for entity in linked_entities:
print(f"{entity.text}: {entity.uri}")
print(f"Linked to {len(entity.linked_entities)} entities")
```
### Explicit Entity Linking
```python
from semantica.context import EntityLinker
linker = EntityLinker(knowledge_graph=kg)
# Create explicit link
linker.link_entities(
"entity_1",
"entity_2",
link_type="related_to",
confidence=0.9,
source="manual"
)
# Get entity links
links = linker.get_entity_links("entity_1")
for link in links:
print(f"{link.source_entity_id} --{link.link_type}--> {link.target_entity_id}")
```
### Finding Similar Entities
```python
from semantica.context import EntityLinker
linker = EntityLinker(knowledge_graph=kg)
# Find similar entities
similar = linker.find_similar_entities(
"Python",
entity_type="PROGRAMMING_LANGUAGE",
threshold=0.8
)
for entity_id, similarity in similar:
print(f"{entity_id}: {similarity:.2f}")
```
### Building Entity Web
```python
from semantica.context import EntityLinker
linker = EntityLinker(knowledge_graph=kg)
# Link multiple entities
linker.link_entities("e1", "e2", "related_to", confidence=0.9)
linker.link_entities("e2", "e3", "related_to", confidence=0.85)
# Build entity web
web = linker.build_entity_web()
print(f"Total entities: {web['statistics']['total_entities']}")
print(f"Total links: {web['statistics']['total_links']}")
for entity_id, info in web['entities'].items():
print(f"{entity_id}: {info['uri']} ({info['links']} links)")
```
### Using Linking Methods
```python
from semantica.context.methods import link_entities
# URI assignment only
linked = link_entities(
entities,
method="uri"
)
# Similarity-based linking
linked = link_entities(
entities,
knowledge_graph=kg,
method="similarity"
)
# Knowledge graph-based linking
linked = link_entities(
entities,
knowledge_graph=kg,
method="knowledge_graph"
)
# Cross-document linking
linked = link_entities(
entities,
knowledge_graph=kg,
method="cross_document",
context=[{"source": "doc1"}, {"source": "doc2"}]
)
```
## Using Methods
### Available Methods
```python
from semantica.context.methods import (
build_context_graph,
store_memory,
retrieve_context,
link_entities,
get_context_method,
list_available_methods
)
# List all available methods
all_methods = list_available_methods()
print(all_methods)
# List methods for specific task
graph_methods = list_available_methods("graph")
print(f"Graph methods: {graph_methods}")
# Get custom method
custom_method = get_context_method("graph", "custom_method")
if custom_method:
result = custom_method(entities, relationships)
```
## Using Registry
### Registering Custom Methods
```python
from semantica.context.registry import method_registry
def custom_graph_builder(entities, relationships, **kwargs):
"""Custom graph building method."""
# Custom implementation
return {"nodes": [], "edges": [], "statistics": {}}
# Register custom method
method_registry.register("graph", "custom_builder", custom_graph_builder)
# Use custom method
from semantica.context.methods import build_context_graph
graph = build_context_graph(entities, relationships, method="custom_builder")
# List registered methods
methods = method_registry.list_all("graph")
print(f"Registered graph methods: {methods}")
# Unregister method
method_registry.unregister("graph", "custom_builder")
```
## Configuration
### Using Configuration
```python
from semantica.context.config import context_config
# Get configuration
retention = context_config.get("retention_policy", default="unlimited")
max_size = context_config.get("max_memory_size", default=10000)
# Set configuration
context_config.set("retention_policy", "30_days")
context_config.set("max_memory_size", 5000)
# Method-specific configuration
context_config.set_method_config("graph", {
"extract_entities": True,
"extract_relationships": True
})
method_config = context_config.get_method_config("graph")
# Get all configurations
all_configs = context_config.get_all()
```
### Environment Variables
```bash
# Set environment variables
export CONTEXT_RETENTION_POLICY=30_days
export CONTEXT_MAX_MEMORY_SIZE=5000
export CONTEXT_SIMILARITY_THRESHOLD=0.8
```
### Config Files
```yaml
# context_config.yaml
context:
retention_policy: 30_days
max_memory_size: 5000
similarity_threshold: 0.8
context_methods:
graph:
extract_entities: true
extract_relationships: true
memory:
retention_policy: unlimited
```
```python
from semantica.context.config import ContextConfig
# Load from config file
config = ContextConfig(config_file="context_config.yaml")
```
## Advanced Examples
### Complete Agent Context Workflow
```python
from semantica.context import (
build_context,
AgentMemory,
ContextRetriever,
EntityLinker
)
# Step 1: Build context graph
result = build_context(
entities=entities,
relationships=relationships,
vector_store=vs,
knowledge_graph=kg,
store_initial_memories=True
)
graph = result['graph']
memory_ids = result['memory_ids']
# Step 2: Initialize memory
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
# Step 3: Store conversation
conversation_id = "conv_123"
memory.store(
"User asked about Python programming",
metadata={
"type": "conversation",
"conversation_id": conversation_id
}
)
# Step 4: Link entities
linker = EntityLinker(knowledge_graph=kg)
linked = linker.link("Python is used for machine learning", entities=entities)
# Step 5: Retrieve context
retriever = ContextRetriever(
memory_store=memory,
knowledge_graph=kg,
vector_store=vs
)
results = retriever.retrieve("Python programming", max_results=5)
# Step 6: Use retrieved context
for result in results:
print(f"Context: {result.content}")
print(f"Relevance: {result.score:.2f}")
if result.related_entities:
print(f"Related: {[e['content'] for e in result.related_entities]}")
```
### Multi-Source Context Integration
```python
from semantica.context import ContextRetriever
# Initialize retriever with multiple sources
retriever = ContextRetriever(
memory_store=memory,
knowledge_graph=kg,
vector_store=vs,
hybrid_alpha=0.5 # Balance between vector and graph
)
# Retrieve with hybrid approach
results = retriever.retrieve(
"Python machine learning frameworks",
max_results=10,
use_graph_expansion=True,
max_hops=2
)
# Process results
for result in results:
print(f"Source: {result.source}")
print(f"Content: {result.content[:100]}...")
print(f"Score: {result.score:.2f}")
print(f"Related entities: {len(result.related_entities)}")
print("---")
```
### Conversation-Based Context Building
```python
from semantica.context import ContextGraphBuilder, AgentMemory
builder = ContextGraphBuilder()
memory = AgentMemory(vector_store=vs, knowledge_graph=kg)
# Process conversations
conversations = [
{
"id": "conv1",
"content": "User asked about Python",
"timestamp": "2024-01-01T10:00:00",
"entities": [{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"}]
},
{
"id": "conv2",
"content": "User asked about machine learning",
"timestamp": "2024-01-01T11:00:00",
"entities": [{"id": "e2", "text": "Machine Learning", "type": "CONCEPT"}]
}
]
# Build graph from conversations
graph = builder.build_from_conversations(
conversations,
link_entities=True,
extract_intents=True
)
# Store conversations in memory
for conv in conversations:
memory.store(
conv["content"],
metadata={
"type": "conversation",
"conversation_id": conv["id"],
"timestamp": conv["timestamp"]
},
entities=conv.get("entities", [])
)
# Retrieve conversation history
history = memory.get_conversation_history(conversation_id="conv1")
```
### Entity Web Construction
```python
from semantica.context import EntityLinker
linker = EntityLinker(
knowledge_graph=kg,
similarity_threshold=0.8
)
# Link multiple entities
entities = [
{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"},
{"id": "e2", "text": "Machine Learning", "type": "CONCEPT"},
{"id": "e3", "text": "TensorFlow", "type": "FRAMEWORK"},
{"id": "e4", "text": "PyTorch", "type": "FRAMEWORK"},
]
# Link entities
linked = linker.link("", entities=entities)
# Create explicit links
linker.link_entities("e1", "e2", "used_for", confidence=0.9)
linker.link_entities("e3", "e2", "implements", confidence=0.95)
linker.link_entities("e4", "e2", "implements", confidence=0.95)
linker.link_entities("e3", "e4", "related_to", confidence=0.8)
# Build entity web
web = linker.build_entity_web()
print(f"Entity Web Statistics:")
print(f" Total entities: {web['statistics']['total_entities']}")
print(f" Total links: {web['statistics']['total_links']}")
for entity_id, info in web['entities'].items():
print(f" {entity_id}: {info['uri']} ({info['links']} links)")
```
+26 -2
View File
@@ -6,6 +6,27 @@ engineering, linking entities across different sources to build the web of
context. It assigns each entity a unique URL/URI and connects them meaningfully
to enable semantic understanding.
Algorithms Used:
URI Generation:
- Hash-based URI: MD5 hash-based URI generation for entities without text
- Text-based URI: URL-safe text encoding for entity text
- Type-based URI: Entity type inclusion in URI
- URI Registry: Dictionary-based URI-to-entity mapping
Entity Linking:
- Text Similarity: Word overlap-based similarity calculation (Jaccard-like)
- Knowledge Graph Lookup: Entity matching in knowledge graph
- Similarity Threshold: Threshold-based entity matching
- Cross-Document Linking: Entity linking across multiple documents
- Bidirectional Linking: Symmetric relationship creation
Entity Web Construction:
- Graph Building: Graph construction from entity links
- Link Aggregation: Link counting and statistics
- Entity Registry: Entity-to-URI mapping
- Link Storage: Dictionary-based link storage (entity_id -> List[EntityLink])
Key Features:
- Links entities across different sources
- Assigns unique identifiers (URLs/URIs) to entities
@@ -15,10 +36,12 @@ Key Features:
- Enables entity disambiguation and resolution
- Similarity-based entity matching
- Bidirectional entity linking
- Entity web construction and statistics
- Configurable similarity thresholds
Main Classes:
- EntityLink: Entity link data structure
- LinkedEntity: Linked entity with context
- EntityLink: Entity link data structure with source_entity_id, target_entity_id, link_type, confidence, source, metadata
- LinkedEntity: Linked entity with context including entity_id, uri, text, type, linked_entities, context, confidence
- EntityLinker: Entity linker for context engineering
Example Usage:
@@ -27,6 +50,7 @@ Example Usage:
>>> uri = linker.assign_uri("entity_1", "Python", "PROGRAMMING_LANGUAGE")
>>> linked_entities = linker.link("Python is a programming language", entities=entities)
>>> linker.link_entities("entity_1", "entity_2", "related_to", confidence=0.9)
>>> web = linker.build_entity_web()
Author: Semantica Contributors
License: MIT
+448
View File
@@ -0,0 +1,448 @@
"""
Context Methods Module
This module provides all context engineering methods as simple, reusable functions for
context graph construction, agent memory management, context retrieval, and entity linking.
It supports multiple context engineering approaches and integrates with the method registry
for extensibility.
Supported Methods:
Context Graph Construction:
- "entities_relationships": Build graph from entities and relationships
- "conversations": Build graph from conversations
- "hybrid": Hybrid graph construction combining multiple sources
Agent Memory Management:
- "store": Store memory items with RAG integration
- "retrieve": Retrieve memories using vector search
- "conversation": Conversation history management
- "hybrid": Hybrid memory retrieval (vector + graph)
Context Retrieval:
- "vector": Vector-based context retrieval
- "graph": Graph-based context retrieval
- "memory": Memory-based context retrieval
- "hybrid": Hybrid retrieval combining all sources
Entity Linking:
- "uri": URI assignment for entities
- "similarity": Similarity-based entity linking
- "knowledge_graph": Knowledge graph-based linking
- "cross_document": Cross-document entity linking
Algorithms Used:
Context Graph Construction:
- Graph Building: Node and edge construction from entities and relationships
- Entity Extraction: Entity extraction from conversations and text
- Relationship Extraction: Relationship extraction from conversations
- Intent Extraction: Intent classification from conversations
- Sentiment Analysis: Sentiment extraction from conversations
- Graph Traversal: BFS/DFS for neighbor discovery and multi-hop traversal
- Graph Indexing: Type-based indexing for efficient node/edge lookup
Agent Memory Management:
- Vector Embedding: Embedding generation for memory items
- Vector Search: Similarity search in vector space
- Keyword Search: Fallback keyword-based search
- Retention Policy: Time-based memory retention and cleanup
- Memory Indexing: Deque-based memory index for efficient access
- Knowledge Graph Integration: Entity and relationship updates to knowledge graph
Context Retrieval:
- Vector Similarity Search: Cosine similarity in vector space
- Graph Traversal: Multi-hop graph expansion for related entities
- Memory Search: Vector and keyword search in memory store
- Result Ranking: Score-based ranking and merging
- Deduplication: Content-based result deduplication
- Hybrid Scoring: Weighted combination of multiple retrieval sources
Entity Linking:
- URI Generation: Hash-based and text-based URI assignment
- Text Similarity: Word overlap-based similarity calculation
- Knowledge Graph Lookup: Entity matching in knowledge graph
- Cross-Document Linking: Entity linking across multiple documents
- Bidirectional Linking: Symmetric relationship creation
- Entity Web Construction: Graph-based entity connection web
Key Features:
- Multiple context graph construction methods
- Multiple agent memory management methods
- Multiple context retrieval methods
- Multiple entity linking methods
- Method dispatchers with registry support
- Custom method registration capability
- Consistent interface across all methods
Main Functions:
- build_context_graph: Context graph construction wrapper
- store_memory: Memory storage wrapper
- retrieve_context: Context retrieval wrapper
- link_entities: Entity linking wrapper
- get_context_method: Get context method by name
Example Usage:
>>> from semantica.context.methods import build_context_graph, retrieve_context
>>> graph = build_context_graph(entities, relationships, method="entities_relationships")
>>> results = retrieve_context("Python programming", method="hybrid", max_results=5)
>>> from semantica.context.methods import get_context_method
>>> method = get_context_method("graph", "custom_method")
Author: Semantica Contributors
License: MIT
"""
from typing import Any, Dict, List, Optional, Callable, Union
from pathlib import Path
from ..utils.logging import get_logger
from ..utils.exceptions import ProcessingError, ConfigurationError
from .context_graph import ContextGraphBuilder, ContextNode, ContextEdge
from .agent_memory import AgentMemory, MemoryItem
from .context_retriever import ContextRetriever, RetrievedContext
from .entity_linker import EntityLinker, EntityLink, LinkedEntity
from .registry import method_registry
logger = get_logger("context_methods")
def build_context_graph(
entities: Optional[List[Dict[str, Any]]] = None,
relationships: Optional[List[Dict[str, Any]]] = None,
conversations: Optional[List[Union[str, Dict[str, Any]]]] = None,
method: str = "entities_relationships",
**kwargs
) -> Dict[str, Any]:
"""
Build context graph from various sources (convenience function).
This is a user-friendly wrapper that builds context graphs using the specified method.
Args:
entities: List of entity dictionaries
relationships: List of relationship dictionaries
conversations: List of conversation files or dictionaries
method: Graph construction method (default: "entities_relationships")
- "entities_relationships": Build from entities and relationships
- "conversations": Build from conversations
- "hybrid": Hybrid construction combining multiple sources
**kwargs: Additional options passed to ContextGraphBuilder
Returns:
Context graph dictionary containing:
- nodes: List of context nodes
- edges: List of context edges
- statistics: Graph statistics
Examples:
>>> from semantica.context.methods import build_context_graph
>>> entities = [{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"}]
>>> relationships = [{"source_id": "e1", "target_id": "e2", "type": "related_to"}]
>>> graph = build_context_graph(entities, relationships, method="entities_relationships")
>>> print(f"Graph has {graph['statistics']['node_count']} nodes")
"""
# Check for custom method in registry
custom_method = method_registry.get("graph", method)
if custom_method:
try:
return custom_method(entities, relationships, conversations, **kwargs)
except Exception as e:
logger.warning(f"Custom method {method} failed: {e}, falling back to default")
try:
builder = ContextGraphBuilder(**kwargs)
if method == "entities_relationships":
if not entities or not relationships:
raise ProcessingError("entities and relationships required for entities_relationships method")
return builder.build_from_entities_and_relationships(entities, relationships, **kwargs)
elif method == "conversations":
if not conversations:
raise ProcessingError("conversations required for conversations method")
return builder.build_from_conversations(conversations, **kwargs)
elif method == "hybrid":
graph = {}
if entities and relationships:
graph1 = builder.build_from_entities_and_relationships(entities, relationships, **kwargs)
graph = graph1
if conversations:
graph2 = builder.build_from_conversations(conversations, **kwargs)
# Merge graphs
if graph:
graph["nodes"].extend(graph2.get("nodes", []))
graph["edges"].extend(graph2.get("edges", []))
else:
graph = graph2
return graph
else:
raise ProcessingError(f"Unknown graph construction method: {method}")
except Exception as e:
logger.error(f"Failed to build context graph: {e}")
raise
def store_memory(
content: str,
vector_store: Optional[Any] = None,
knowledge_graph: Optional[Any] = None,
method: str = "store",
**kwargs
) -> str:
"""
Store memory item (convenience function).
This is a user-friendly wrapper that stores memory using the specified method.
Args:
content: Memory content
vector_store: Vector store instance
knowledge_graph: Knowledge graph instance
method: Memory storage method (default: "store")
- "store": Standard memory storage with RAG
- "conversation": Conversation memory storage
**kwargs: Additional options passed to AgentMemory
Returns:
Memory ID
Examples:
>>> from semantica.context.methods import store_memory
>>> memory_id = store_memory("User asked about Python", vector_store=vs, method="store")
>>> print(f"Stored memory: {memory_id}")
"""
# Check for custom method in registry
custom_method = method_registry.get("memory", method)
if custom_method:
try:
return custom_method(content, vector_store, knowledge_graph, **kwargs)
except Exception as e:
logger.warning(f"Custom method {method} failed: {e}, falling back to default")
try:
memory = AgentMemory(
vector_store=vector_store,
knowledge_graph=knowledge_graph,
**kwargs
)
metadata = kwargs.get("metadata", {})
if method == "conversation":
metadata["type"] = "conversation"
return memory.store(
content,
metadata=metadata,
entities=kwargs.get("entities"),
relationships=kwargs.get("relationships"),
**{k: v for k, v in kwargs.items() if k not in ["metadata", "entities", "relationships"]}
)
except Exception as e:
logger.error(f"Failed to store memory: {e}")
raise
def retrieve_context(
query: str,
memory_store: Optional[Any] = None,
knowledge_graph: Optional[Any] = None,
vector_store: Optional[Any] = None,
method: str = "hybrid",
max_results: int = 5,
**kwargs
) -> List[RetrievedContext]:
"""
Retrieve relevant context (convenience function).
This is a user-friendly wrapper that retrieves context using the specified method.
Args:
query: Search query
memory_store: Memory store instance
knowledge_graph: Knowledge graph instance
vector_store: Vector store instance
method: Retrieval method (default: "hybrid")
- "vector": Vector-based retrieval only
- "graph": Graph-based retrieval only
- "memory": Memory-based retrieval only
- "hybrid": Hybrid retrieval combining all sources
max_results: Maximum number of results
**kwargs: Additional options passed to ContextRetriever
Returns:
List of RetrievedContext objects
Examples:
>>> from semantica.context.methods import retrieve_context
>>> results = retrieve_context("Python programming", vector_store=vs, method="hybrid")
>>> for result in results:
... print(f"{result.content}: {result.score:.2f}")
"""
# Check for custom method in registry
custom_method = method_registry.get("retrieval", method)
if custom_method:
try:
return custom_method(query, memory_store, knowledge_graph, vector_store, max_results, **kwargs)
except Exception as e:
logger.warning(f"Custom method {method} failed: {e}, falling back to default")
try:
retriever = ContextRetriever(
memory_store=memory_store,
knowledge_graph=knowledge_graph,
vector_store=vector_store,
**kwargs
)
if method == "vector":
retriever.use_graph_expansion = False
retriever.memory_store = None
elif method == "graph":
retriever.vector_store = None
retriever.memory_store = None
elif method == "memory":
retriever.vector_store = None
retriever.use_graph_expansion = False
return retriever.retrieve(query, max_results=max_results, **kwargs)
except Exception as e:
logger.error(f"Failed to retrieve context: {e}")
raise
def link_entities(
entities: List[Dict[str, Any]],
knowledge_graph: Optional[Any] = None,
method: str = "similarity",
**kwargs
) -> List[LinkedEntity]:
"""
Link entities across sources (convenience function).
This is a user-friendly wrapper that links entities using the specified method.
Args:
entities: List of entity dictionaries
knowledge_graph: Knowledge graph instance
method: Linking method (default: "similarity")
- "uri": URI assignment only
- "similarity": Similarity-based linking
- "knowledge_graph": Knowledge graph-based linking
- "cross_document": Cross-document linking
**kwargs: Additional options passed to EntityLinker
Returns:
List of LinkedEntity objects
Examples:
>>> from semantica.context.methods import link_entities
>>> entities = [{"id": "e1", "text": "Python", "type": "PROGRAMMING_LANGUAGE"}]
>>> linked = link_entities(entities, knowledge_graph=kg, method="similarity")
>>> for entity in linked:
... print(f"{entity.text}: {entity.uri}")
"""
# Check for custom method in registry
custom_method = method_registry.get("linking", method)
if custom_method:
try:
return custom_method(entities, knowledge_graph, **kwargs)
except Exception as e:
logger.warning(f"Custom method {method} failed: {e}, falling back to default")
try:
linker = EntityLinker(
knowledge_graph=knowledge_graph,
**kwargs
)
if method == "uri":
# Just assign URIs
linked = []
for entity in entities:
entity_id = entity.get("id") or entity.get("entity_id")
entity_text = entity.get("text") or entity.get("label") or entity.get("name", "")
entity_type = entity.get("type") or entity.get("entity_type")
uri = linker.assign_uri(entity_id, entity_text, entity_type)
linked.append(LinkedEntity(
entity_id=entity_id,
uri=uri,
text=entity_text,
type=entity_type or "UNKNOWN",
linked_entities=[],
context=entity.get("metadata", {}),
confidence=entity.get("confidence", 1.0)
))
return linked
else:
# Use full linking
return linker.link(
text="", # Not used for entity list
entities=entities,
context=kwargs.get("context")
)
except Exception as e:
logger.error(f"Failed to link entities: {e}")
raise
def get_context_method(task: str, name: str) -> Optional[Callable]:
"""
Get a registered context method.
Args:
task: Task type ("graph", "memory", "retrieval", "linking")
name: Method name
Returns:
Registered method or None if not found
Examples:
>>> from semantica.context.methods import get_context_method
>>> method = get_context_method("graph", "custom_method")
>>> if method:
... result = method(entities, relationships)
"""
return method_registry.get(task, name)
def list_available_methods(task: Optional[str] = None) -> Dict[str, List[str]]:
"""
List all available context methods.
Args:
task: Optional task type filter
Returns:
Dictionary mapping task types to method names
Examples:
>>> from semantica.context.methods import list_available_methods
>>> all_methods = list_available_methods()
>>> graph_methods = list_available_methods("graph")
"""
return method_registry.list_all(task)
# Register default methods
method_registry.register("graph", "entities_relationships", build_context_graph)
method_registry.register("graph", "conversations", build_context_graph)
method_registry.register("graph", "hybrid", build_context_graph)
method_registry.register("memory", "store", store_memory)
method_registry.register("memory", "conversation", store_memory)
method_registry.register("retrieval", "vector", retrieve_context)
method_registry.register("retrieval", "graph", retrieve_context)
method_registry.register("retrieval", "memory", retrieve_context)
method_registry.register("retrieval", "hybrid", retrieve_context)
method_registry.register("linking", "uri", link_entities)
method_registry.register("linking", "similarity", link_entities)
method_registry.register("linking", "knowledge_graph", link_entities)
method_registry.register("linking", "cross_document", link_entities)
+113
View File
@@ -0,0 +1,113 @@
"""
Method Registry Module for Context Engineering
This module provides a method registry system for registering custom context engineering methods,
enabling extensibility and community contributions to the context toolkit.
Supported Registration Types:
- Method Registry: Register custom context methods for:
* "graph": Context graph construction methods
* "memory": Agent memory management methods
* "retrieval": Context retrieval methods
* "linking": Entity linking methods
Algorithms Used:
- Registry Pattern: Dictionary-based registration and lookup
- Dynamic Registration: Runtime function registration
- Type Checking: Type validation for registered components
- Lookup Algorithms: Hash-based O(1) lookup for methods
- Task-based Organization: Hierarchical organization by task type
Key Features:
- Method registry for custom context methods
- Task-based method organization (graph, memory, retrieval, linking)
- Dynamic registration and unregistration
- Easy discovery of available methods
- Support for community-contributed extensions
Main Classes:
- MethodRegistry: Registry for custom context methods
Global Instances:
- method_registry: Global method registry instance
Example Usage:
>>> from semantica.context.registry import method_registry
>>> method_registry.register("graph", "custom_method", custom_graph_function)
>>> available = method_registry.list_all("graph")
Author: Semantica Contributors
License: MIT
"""
from typing import Dict, Callable, Any, List, Optional
class MethodRegistry:
"""Registry for custom context methods."""
_methods: Dict[str, Dict[str, Callable]] = {
"graph": {},
"memory": {},
"retrieval": {},
"linking": {},
}
@classmethod
def register(cls, task: str, name: str, method_func: Callable):
"""
Register a method for a specific task.
Args:
task: Task type ("graph", "memory", "retrieval", "linking")
name: Method name
method_func: Method function or callable
"""
if task not in cls._methods:
cls._methods[task] = {}
cls._methods[task][name] = method_func
@classmethod
def get(cls, task: str, name: str) -> Optional[Callable]:
"""
Get a registered method.
Args:
task: Task type
name: Method name
Returns:
Registered method or None if not found
"""
return cls._methods.get(task, {}).get(name)
@classmethod
def list_all(cls, task: Optional[str] = None) -> Dict[str, List[str]]:
"""
List all registered methods.
Args:
task: Optional task type filter
Returns:
Dictionary mapping task types to method names
"""
if task:
return {task: list(cls._methods.get(task, {}).keys())}
return {t: list(m.keys()) for t, m in cls._methods.items()}
@classmethod
def unregister(cls, task: str, name: str):
"""
Unregister a method.
Args:
task: Task type
name: Method name
"""
if task in cls._methods and name in cls._methods[task]:
del cls._methods[task][name]
method_registry = MethodRegistry()