mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: enhance context module with persistence and FastEmbed

Browse Source 
- Updated AgentContext, AgentMemory, and ContextGraph to support save/load persistence
- Integrated FastEmbed into VectorStore for high-performance local embeddings
- Replaced DemoVectorStore with production VectorStore in docs and examples
- Rebuilt 19_Context_Module.ipynb as a deep dive into context engineering
- Updated documentation and README to reflect new capabilities
This commit is contained in:
KaifAhmad1
2025-12-18 00:56:56 +05:30
parent 1620de371f
commit e45875f924
7 changed files with 485 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -178,7 +178,7 @@ flowchart TD
| **Knowledge Graphs** | Entity resolution, temporal support, graph analytics, query interface | Production-ready, queryable knowledge structures |
| **Ontology Generation** | 6-stage LLM pipeline, OWL generation, HermiT/Pellet validation | Automated ontology creation from documents |
| **GraphRAG** | Hybrid vector + graph retrieval, multi-hop reasoning | 91% accuracy, 30% improvement over vector-only |
| **Agent Memory** | Persistent memory, RAG integration, MCP-compatible tools | Context-aware agents with semantic understanding |
| **Agent Memory** | Persistent memory (Save/Load), Hybrid Retrieval (Vector+Graph), FastEmbed support | Context-aware agents with semantic understanding |
| **Pipeline Orchestration** | Parallel execution, custom steps, orchestrator-worker pattern | Scalable, flexible data processing |
| **Quality Assurance** | Conflict detection, deduplication, quality scoring, provenance | Trusted knowledge graphs ready for production |
cookbook/introduction/19_Context_Module.ipynb
+186 -166
View File
@@ -6,29 +6,27 @@
"source": [
"[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/Hawksight-AI/semantica/blob/main/cookbook/introduction/19_Context_Module.ipynb)\n",
"\n",
"# Context Engineering Module\n",
"# Deep Dive: Semantica Context Module (Architecture 2.0)\n",
"\n",
"## Overview\n",
"\n",
"This notebook provides a comprehensive guide to Semantica's **Context Engineering Module** - a powerful system for building context graphs, managing agent memory, retrieving context, and linking entities. You'll learn how to use the new synchronous Architecture 2.0 features, including hierarchical memory with token management.\n",
"The **Context Module** is the core state management system of Semantica. It allows agents to maintain coherent, persistent, and structured memory across long interactions. Unlike simple RAG systems that only use vector similarity, Semantica's Context Module combines:\n",
"\n",
"1. **FastEmbed Integration**: High-performance, local embedding generation.\n",
"2. **Context Graph**: A structured knowledge graph for reasoning about relationships.\n",
"3. **Hierarchical Memory**: A tiered system with short-term (token-limited) and long-term (vector-backed) storage.\n",
"4. **Hybrid Retrieval**: Combining vector search, graph traversal (GraphRAG), and keyword matching.\n",
"5. **Persistence**: Full state serialization.\n",
"\n",
"This notebook provides a technical deep dive into these components.\n",
"\n",
"**Documentation**: [API Reference](https://semantica.readthedocs.io/reference/context/)\n",
"\n",
"### Learning Objectives\n",
"\n",
"- **Hierarchical Memory**: Manage short-term (token-buffered) and long-term (vector-stored) memory\n",
"- **Context Graph**: Build and query dynamic knowledge graphs\n",
"- **Hybrid Retrieval**: Combine vector search, graph traversal, and keyword matching\n",
"- **Entity Linking**: Resolve entities across conversations\n",
"- **Configuration**: Customize behavior via YAML or environment variables\n",
"\n",
"---\n",
"\n",
"## Installation\n",
"## 1. Setup and Configuration\n",
"\n",
"```bash\n",
"pip install semantica\n",
"```"
"We need `semantica` and `fastembed` for local, high-speed embedding generation."
]
},
{
@@ -37,7 +35,7 @@
"metadata": {},
"outputs": [],
"source": [
"!pip install semantica\n"
"!pip install semantica fastembed"
]
},
{
@@ -46,52 +44,23 @@
"metadata": {},
"outputs": [],
"source": [
"# Setup: Create a mock vector store for demonstration\n",
"from typing import List, Dict, Any, Optional\n",
"from semantica.context import VectorStore\n",
"import os\n",
"import shutil\n",
"import numpy as np\n",
"\n",
"class MockVectorStore(VectorStore):\n",
" def __init__(self):\n",
" self.items = {}\n",
" self.counter = 0\n",
" \n",
" def add(self, texts: List[str], metadata: Optional[List[Dict[str, Any]]] = None, **kwargs) -> List[str]:\n",
" ids = []\n",
" for i, text in enumerate(texts):\n",
" id_ = f\"id_{self.counter}\"\n",
" self.items[id_] = {\"text\": text, \"metadata\": metadata[i] if metadata else {}}\n",
" ids.append(id_)\n",
" self.counter += 1\n",
" return ids\n",
" \n",
" def search(self, query: str, limit: int = 5, **kwargs) -> List[Dict[str, Any]]:\n",
" # Simple keyword match for mock\n",
" results = []\n",
" for id_, item in self.items.items():\n",
" if any(w.lower() in item[\"text\"].lower() for w in query.split()):\n",
" results.append({\n",
" \"id\": id_,\n",
" \"content\": item[\"text\"],\n",
" \"score\": 0.9,\n",
" \"metadata\": item[\"metadata\"]\n",
" })\n",
" return results[:limit]\n",
" \n",
" def delete(self, ids: List[str], **kwargs) -> bool:\n",
" for id_ in ids:\n",
" self.items.pop(id_, None)\n",
" return True\n",
"\n",
"vs = MockVectorStore()"
"from semantica.context import AgentContext, ContextGraph, AgentMemory, EntityLinker\n",
"from semantica.vector_store import VectorStore"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 1. High-Level Interface: AgentContext\n",
"## 2. Vector Store with FastEmbed\n",
"\n",
"The `AgentContext` class is the easiest way to get started. It unifies vector storage, knowledge graphs, and memory management."
"The `VectorStore` manages long-term memory. We will configure it to use **FastEmbed**, which runs efficient, quantized embedding models locally on the CPU.\n",
"\n",
"We use the `inmemory` backend for this demo, but Semantica supports Qdrant, Weaviate, and FAISS for production."
]
},
{
@@ -100,124 +69,133 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.context import AgentContext, ContextGraph\n",
"# 1. Initialize Vector Store\n",
"vs = VectorStore(backend=\"inmemory\", dimension=384)\n",
"\n",
"# Initialize with vector store and a new in-memory knowledge graph\n",
"# 2. Configure FastEmbed\n",
"# We explicitly set the method to 'fastembed' and choose a lightweight, high-performance model.\n",
"if hasattr(vs, \"embedder\") and vs.embedder:\n",
" print(\"Configuring VectorStore to use FastEmbed...\")\n",
" vs.embedder.set_text_model(\n",
" method=\"fastembed\", \n",
" model_name=\"BAAI/bge-small-en-v1.5\"\n",
" )\n",
"\n",
"# 3. Verify Embedding Generation\n",
"text = \"Semantica enables complex agent behaviors.\"\n",
"vector = vs.embed(text)\n",
"print(f\"Generated embedding shape: {vector.shape}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 3. Context Graph Construction\n",
"\n",
"The `ContextGraph` stores structured data. While vectors capture *similarity*, graphs capture *relationships*.\n",
"\n",
"We will manually build a small graph to understand the API:\n",
"- `add_node(node_id, node_type, content, **properties)`\n",
"- `add_edge(source_id, target_id, edge_type, **properties)`"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"kg = ContextGraph()\n",
"\n",
"# Add Nodes\n",
"# Note: 'content' is what is indexed for keyword search.\n",
"kg.add_node(\n",
" node_id=\"user_alice\", \n",
" node_type=\"Person\", \n",
" content=\"Alice\", \n",
" role=\"Lead Engineer\"\n",
")\n",
"kg.add_node(\n",
" node_id=\"tech_python\", \n",
" node_type=\"Technology\", \n",
" content=\"Python\", \n",
" version=\"3.11\"\n",
")\n",
"kg.add_node(\n",
" node_id=\"project_semantica\", \n",
" node_type=\"Project\", \n",
" content=\"Semantica Framework\"\n",
")\n",
"\n",
"# Add Edges (Relationships)\n",
"kg.add_edge(source_id=\"user_alice\", target_id=\"tech_python\", edge_type=\"USES\")\n",
"kg.add_edge(source_id=\"tech_python\", target_id=\"project_semantica\", edge_type=\"POWERS\")\n",
"\n",
"# Traverse the Graph\n",
"print(\"Neighbors of Python:\")\n",
"neighbors = kg.get_neighbors(\"tech_python\")\n",
"for n in neighbors:\n",
" # The neighbor dict contains the connected node info and the relationship that led to it\n",
" print(f\" - [Relationship: {n['relationship']}] -> {n['content']} ({n['type']})\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 4. AgentContext: The Unified Interface\n",
"\n",
"`AgentContext` combines the `VectorStore` and `ContextGraph` into a single system. It handles:\n",
"1. **Memory Management**: Routing inputs to short-term or long-term memory.\n",
"2. **Hybrid Retrieval**: Querying both vectors and the graph simultaneously.\n",
"\n",
"We will initialize it with limits to demonstrate the hierarchy."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"context = AgentContext(\n",
" vector_store=vs,\n",
" knowledge_graph=kg,\n",
" token_limit=2000, # Max tokens in short-term memory\n",
" short_term_limit=10 # Max items in short-term memory\n",
" token_limit=500, # Max tokens in Short-Term Memory (STM)\n",
" short_term_limit=5 # Max items in STM\n",
")\n",
"\n",
"# Store a memory (automatically goes to short-term and long-term)\n",
"# Store a new memory\n",
"# This is automatically embedded (FastEmbed) and indexed.\n",
"context.store(\n",
" \"The user, Alice, is a data scientist interested in Python.\",\n",
" conversation_id=\"conv_1\",\n",
" user_id=\"alice_01\"\n",
" content=\"Alice is optimizing the graph traversal algorithms in Semantica.\",\n",
" conversation_id=\"dev_sync_1\",\n",
" user_id=\"alice\"\n",
")\n",
"\n",
"# Retrieve context (automatically uses hybrid retrieval)\n",
"results = context.retrieve(\"What does Alice do?\")\n",
"\n",
"for res in results:\n",
" print(f\"Found: {res['content']} (Score: {res['score']})\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 2. Hierarchical Memory Management\n",
"\n",
"Semantica uses a two-tier memory system:\n",
"1. **Short-Term Memory**: A fast, in-memory buffer limited by tokens (to fit in LLM context windows) and item count.\n",
"2. **Long-Term Memory**: Persistent storage backed by the vector store.\n",
"\n",
"Let's observe how the token limit works."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from semantica.context import AgentMemory\n",
"\n",
"# Initialize memory with strict limits for demonstration\n",
"memory = AgentMemory(\n",
" vector_store=vs,\n",
" token_limit=50, # Very small token limit\n",
" short_term_limit=5 # Max 5 items\n",
")\n",
"\n",
"# Add memories\n",
"for i in range(10):\n",
" memory.store(f\"Memory item {i}: This is a sentence with some tokens.\")\n",
" print(f\"Added item {i}. Short-term size: {len(memory.short_term_memory)}\")\n",
"\n",
"print(\"\\nFinal short-term memory content:\")\n",
"for item in memory.short_term_memory:\n",
" print(f\"- {item.content}\")\n",
" \n",
"# Notice that older items are pruned to respect the token limit and item count."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 3. Context Graph & GraphRAG\n",
"\n",
"The `ContextGraph` allows you to structure information as nodes and edges, enabling \"GraphRAG\" - retrieving information based on relationships rather than just semantic similarity."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from semantica.context import ContextGraph\n",
"\n",
"graph = ContextGraph()\n",
"\n",
"# Manually building a graph\n",
"graph.add_node(\"n1\", \"person\", \"Alice\")\n",
"graph.add_node(\"n2\", \"language\", \"Python\")\n",
"graph.add_node(\"n3\", \"library\", \"Semantica\")\n",
"\n",
"graph.add_edge(\"n1\", \"n2\", \"uses\")\n",
"graph.add_edge(\"n2\", \"n3\", \"powers\")\n",
"\n",
"# Query the graph\n",
"neighbors = graph.get_neighbors(\"n2\", hops=1)\n",
"print(\"Neighbors of Python:\", neighbors)\n",
"\n",
"# Using the graph in AgentContext\n",
"context = AgentContext(vector_store=vs, knowledge_graph=graph)\n",
"\n",
"# Retrieve with graph expansion\n",
"# Retrieve Context\n",
"# 'use_graph=True' enables GraphRAG: it finds entities in the query ('Alice') \n",
"# and expands to their neighbors in the graph.\n",
"results = context.retrieve(\n",
" \"Alice\",\n",
" query=\"What is Alice working on?\",\n",
" use_graph=True,\n",
" expand_graph=True # Will pull in 'Python' because Alice uses it\n",
" expand_graph=True\n",
")\n",
"\n",
"print(\"\\nGraph-enhanced Retrieval:\")\n",
"print(\"\\n--- Hybrid Retrieval Results ---\")\n",
"for res in results:\n",
" print(f\"- {res['content']}\")"
" print(f\"[{res['score']:.2f}] {res['content']}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 4. Entity Linking\n",
"## 5. Hierarchical Memory Management\n",
"\n",
"The `EntityLinker` helps ensure that \"Alice\", \"Alice Smith\", and \"she\" (in context) refer to the same entity ID."
"Watch how the `AgentContext` manages memory pressure. We defined `short_term_limit=5`.\n",
"As we add more items, the oldest ones are flushed from the active buffer but remain safe in the Vector Store."
]
},
{
@@ -226,26 +204,29 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.context import EntityLinker\n",
"print(f\"STM Count (Start): {len(context.memory.short_term_memory)}\")\n",
"\n",
"linker = EntityLinker()\n",
"# Fill up memory\n",
"for i in range(1, 10):\n",
" context.store(f\"Log entry {i}: System status check.\")\n",
"\n",
"# Generate a canonical URI\n",
"uri = linker.generate_uri(\"Python Programming Language\")\n",
"print(f\"Canonical URI: {uri}\")\n",
"print(f\"STM Count (End): {len(context.memory.short_term_memory)}\")\n",
"print(\"\\nCurrent Short-Term Memory Items:\")\n",
"for item in context.memory.short_term_memory:\n",
" print(f\" - {item.content}\")\n",
"\n",
"# Check similarity\n",
"score = linker._calculate_text_similarity(\"Python\", \"Python Lang\")\n",
"print(f\"Similarity Score: {score}\")"
"# Notice that earlier log entries are gone from this list, \n",
"# but they are still retrievable via search."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 5. Configuration\n",
"## 6. Persistence\n",
"\n",
"You can configure the context module using the `config` object or environment variables."
"To build stateful agents, we must save and load the context. \n",
"**Crucial Step**: When loading, we must ensure the new `VectorStore` instance is configured with the same embedding model (`FastEmbed`) so vectors match."
]
},
{
@@ -254,13 +235,52 @@
"metadata": {},
"outputs": [],
"source": [
"from semantica.context import config\n",
"SAVE_PATH = \"./semantica_context_state\"\n",
"\n",
"# Set global configuration\n",
"config.context_config.set(\"token_limit\", 4096)\n",
"config.context_config.set(\"retention_days\", 30)\n",
"# 1. Save state\n",
"print(f\"Saving context to {SAVE_PATH}...\")\n",
"context.save(SAVE_PATH)\n",
"\n",
"print(f\"Current Token Limit: {config.context_config.get('token_limit')}\")"
"# 2. Initialize a fresh AgentContext\n",
"print(\"Initializing fresh agent...\")\n",
"new_kg = ContextGraph()\n",
"new_vs = VectorStore(backend=\"inmemory\", dimension=384)\n",
"\n",
"# !!! IMPORTANT: Re-configure FastEmbed before loading !!!\n",
"if hasattr(new_vs, \"embedder\") and new_vs.embedder:\n",
" new_vs.embedder.set_text_model(\n",
" method=\"fastembed\", \n",
" model_name=\"BAAI/bge-small-en-v1.5\"\n",
" )\n",
"\n",
"restored_context = AgentContext(vector_store=new_vs, knowledge_graph=new_kg)\n",
"\n",
"# 3. Load state\n",
"restored_context.load(SAVE_PATH)\n",
"\n",
"# 4. Verify restoration\n",
"print(f\"Restored STM items: {len(restored_context.memory.short_term_memory)}\")\n",
"print(f\"Restored Graph nodes: {len(new_kg.nodes)}\")\n",
"\n",
"# Cleanup\n",
"if os.path.exists(SAVE_PATH):\n",
" shutil.rmtree(SAVE_PATH)\n",
" print(\"Cleanup complete.\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"You have successfully built a persistent, graph-aware, memory-managed context system using Semantica 2.0 components.\n",
"\n",
"**Key Takeaways:**\n",
"- Use `VectorStore` with `FastEmbed` for efficient local vectors.\n",
"- Use `ContextGraph` to map relationships (`add_node`, `add_edge`).\n",
"- Use `AgentContext` to manage the lifecycle of memories and retrieval.\n",
"- Always configure your embedder on the fresh instance before calling `load()`."
]
}
],
@@ -280,9 +300,9 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.8.10"
"version": "3.11.9"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
}
semantica/context/agent_context.py
+67
View File
@@ -156,6 +156,73 @@ class AgentContext:
"""Access underlying ContextGraph instance for building."""
return self._graph_builder
def save(self, path: str) -> None:
"""
Save context state (memory, vector store, graph) to disk.
Args:
path: Directory path to save to
"""
import os
os.makedirs(path, exist_ok=True)
# 1. Save AgentMemory state
if hasattr(self._memory, "save"):
self._memory.save(path)
# 2. Save VectorStore state
if hasattr(self.vector_store, "save"):
vs_path = os.path.join(path, "vector_store")
self.vector_store.save(vs_path)
# 3. Save KnowledgeGraph state
if self.knowledge_graph:
if hasattr(self.knowledge_graph, "save_to_file"):
# JSON export for ContextGraph
kg_path = os.path.join(path, "knowledge_graph.json")
self.knowledge_graph.save_to_file(kg_path)
elif hasattr(self.knowledge_graph, "save"):
# Generic save
kg_path = os.path.join(path, "knowledge_graph")
self.knowledge_graph.save(kg_path)
self.logger.info(f"Saved agent context to {path}")
def load(self, path: str) -> None:
"""
Load context state from disk.
Args:
path: Directory path to load from
"""
import os
if not os.path.exists(path):
self.logger.warning(f"Context path not found: {path}")
return
# 1. Load AgentMemory state
if hasattr(self._memory, "load"):
self._memory.load(path)
# 2. Load VectorStore state
if hasattr(self.vector_store, "load"):
vs_path = os.path.join(path, "vector_store")
if os.path.exists(vs_path):
self.vector_store.load(vs_path)
# 3. Load KnowledgeGraph state
if self.knowledge_graph:
kg_json_path = os.path.join(path, "knowledge_graph.json")
kg_dir_path = os.path.join(path, "knowledge_graph")
if hasattr(self.knowledge_graph, "load_from_file") and os.path.exists(kg_json_path):
self.knowledge_graph.load_from_file(kg_json_path)
elif hasattr(self.knowledge_graph, "load") and os.path.exists(kg_dir_path):
self.knowledge_graph.load(kg_dir_path)
self.logger.info(f"Loaded agent context from {path}")
def store(
self,
content: Union[str, List[str], List[Dict[str, Any]]],
semantica/context/agent_memory.py
+49
View File
@@ -129,6 +129,55 @@ class AgentMemory:
# Statistics
self.stats = {"total_items": 0, "items_by_type": {}, "last_accessed": None}
def save(self, path: str) -> None:
"""
Save memory state to disk.
Args:
path: Directory path to save to
"""
import os
import pickle
os.makedirs(path, exist_ok=True)
data = {
"memory_items": self.memory_items,
"memory_index": self.memory_index,
"short_term_memory": self.short_term_memory,
"stats": self.stats
}
with open(os.path.join(path, "agent_memory.pkl"), "wb") as f:
pickle.dump(data, f)
self.logger.info(f"Saved agent memory to {path}")
def load(self, path: str) -> None:
"""
Load memory state from disk.
Args:
path: Directory path to load from
"""
import os
import pickle
file_path = os.path.join(path, "agent_memory.pkl")
if not os.path.exists(file_path):
self.logger.warning(f"Memory file not found: {file_path}")
return
with open(file_path, "rb") as f:
data = pickle.load(f)
self.memory_items = data.get("memory_items", {})
self.memory_index = data.get("memory_index", deque(maxlen=self.max_memory_size))
self.short_term_memory = data.get("short_term_memory", [])
self.stats = data.get("stats", {"total_items": 0, "items_by_type": {}, "last_accessed": None})
self.logger.info(f"Loaded agent memory from {path}")
def store(
self,
content: str,
semantica/context/context_graph.py
+54
View File
@@ -266,6 +266,59 @@ class ContextGraph:
metadata=properties
))
def save_to_file(self, path: str) -> None:
"""
Save context graph to file (JSON format).
Args:
path: File path to save to
"""
import json
data = {
"nodes": [node.to_dict() for node in self.nodes.values()],
"edges": [edge.to_dict() for edge in self.edges]
}
with open(path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
self.logger.info(f"Saved context graph to {path}")
def load_from_file(self, path: str) -> None:
"""
Load context graph from file (JSON format).
Args:
path: File path to load from
"""
import json
import os
if not os.path.exists(path):
self.logger.warning(f"File not found: {path}")
return
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
# Clear existing
self.nodes.clear()
self.edges.clear()
self._adjacency.clear()
self.node_type_index.clear()
self.edge_type_index.clear()
# Load nodes
nodes = data.get("nodes", [])
self.add_nodes(nodes)
# Load edges
edges = data.get("edges", [])
self.add_edges(edges)
self.logger.info(f"Loaded context graph from {path}")
def find_node(self, node_id: str) -> Optional[Dict[str, Any]]:
"""Find a node by ID."""
node = self.nodes.get(node_id)
@@ -278,6 +331,7 @@ class ContextGraph:
}
return None
def find_nodes(self, node_type: Optional[str] = None) -> List[Dict[str, Any]]:
"""Find nodes, optionally filtered by type."""
if node_type:
semantica/context/context_usage.md
+27 -19
View File
@@ -115,31 +115,39 @@ deleted_count = context.forget(days_old=90)
print(f"Deleted {deleted_count} old memories")
```
## Basic Usage
### Persistence (Save/Load)
### Using Main Classes Directly
You can save the entire state of the agent (Memory, Graph, and Vector Index) to disk and reload it later.
```python
from semantica.context import ContextGraph, AgentMemory, ContextRetriever
# Save state
context.save("./my_agent_state")
# Step 1: Build context graph
graph = ContextGraph()
# Add entities and relationships manually or via build methods
entities = [{"id": "e1", "text": "Python", "type": "Language"}]
relationships = [] # ...
graph.build_from_entities_and_relationships(entities, relationships)
# Load state
new_context = AgentContext(vector_store=VectorStore(), knowledge_graph=ContextGraph())
new_context.load("./my_agent_state")
```
# Step 2: Initialize agent memory with token limits
memory = AgentMemory(
vector_store=vs,
knowledge_graph=graph,
token_limit=1000 # Custom token limit
)
memory_id = memory.store("User asked about Python", metadata={"type": "conversation"})
## Basic Usage
# Step 3: Retrieve context
retriever = ContextRetriever(memory_store=memory, knowledge_graph=graph, vector_store=vs)
results = retriever.retrieve("Python programming", max_results=5)
### Initialization with Backends
You can configure the `VectorStore` with different backends (`inmemory`, `faiss`, `chroma`, `qdrant`, `weaviate`, `milvus`) and embedding models (including FastEmbed).
```python
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
# Initialize Vector Store with FastEmbed
vs = VectorStore(backend="inmemory", dimension=384)
if hasattr(vs, "embedder") and vs.embedder:
vs.embedder.set_text_model(method="fastembed", model_name="BAAI/bge-small-en-v1.5")
# Initialize Context Graph
kg = ContextGraph()
# Initialize Agent Context
context = AgentContext(vector_store=vs, knowledge_graph=kg)
```
## Context Graph Construction
semantica/vector_store/vector_store.py
+101
View File
@@ -44,6 +44,7 @@ import numpy as np
from ..utils.exceptions import ProcessingError, ValidationError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
from ..embeddings import EmbeddingGenerator
class VectorStore:
@@ -89,6 +90,40 @@ class VectorStore:
)
self.retriever = VectorRetriever(backend=backend, **self.config)
# Initialize embedding generator
try:
self.embedder = EmbeddingGenerator()
# Set default model if not configured, or respect global config
# For now, we try to ensure a model is loaded if possible
if hasattr(self.embedder, "set_text_model"):
# Use a lightweight default if none specified, or let EmbeddingGenerator handle defaults
pass
except Exception as e:
self.logger.warning(f"Could not initialize embedding generator: {e}")
self.embedder = None
def embed(self, text: str) -> np.ndarray:
"""
Generate embedding for text using the internal embedder.
Args:
text: Text to embed
Returns:
Numpy array of embedding
"""
if self.embedder:
try:
return self.embedder.generate_embeddings(text)
except Exception as e:
self.logger.warning(f"Embedding generation failed: {e}")
# Fallback or raise? AgentMemory expects None or valid embedding.
# Returning random vector as fallback for now (matches DemoVectorStore behavior)
# to prevent crashes, but logging warning.
self.logger.warning("Using random fallback embedding")
return np.random.rand(self.dimension).astype(np.float32)
def store(
self,
vectors: List[np.ndarray],
@@ -189,6 +224,72 @@ class VectorStore:
)
raise
def save(self, path: str) -> None:
"""
Save vector store to disk.
Args:
path: Directory path to save to
"""
import os
import pickle
os.makedirs(path, exist_ok=True)
# Save metadata and vectors (generic fallback)
# Ideally, backends like FAISS have their own save methods
if hasattr(self.indexer, "save_index"):
self.indexer.save_index(os.path.join(path, "index.bin"))
# Save Python-level data
data = {
"vectors": self.vectors,
"metadata": self.metadata,
"config": self.config,
"backend": self.backend,
"dimension": self.dimension
}
with open(os.path.join(path, "store_data.pkl"), "wb") as f:
pickle.dump(data, f)
self.logger.info(f"Saved vector store to {path}")
def load(self, path: str) -> None:
"""
Load vector store from disk.
Args:
path: Directory path to load from
"""
import os
import pickle
data_path = os.path.join(path, "store_data.pkl")
if not os.path.exists(data_path):
self.logger.warning(f"Store data not found: {data_path}")
return
with open(data_path, "rb") as f:
data = pickle.load(f)
self.vectors = data.get("vectors", {})
self.metadata = data.get("metadata", {})
self.config = data.get("config", {})
self.backend = data.get("backend", "faiss")
self.dimension = data.get("dimension", 768)
# Restore backend-specific index
if hasattr(self.indexer, "load_index"):
index_path = os.path.join(path, "index.bin")
if os.path.exists(index_path):
self.indexer.load_index(index_path)
else:
# Rebuild if index file missing but vectors present
self.indexer.create_index(list(self.vectors.values()), list(self.vectors.keys()))
self.logger.info(f"Loaded vector store from {path}")
def search_vectors(
self, query_vector: np.ndarray, k: int = 10, **options
) -> List[Dict[str, Any]]: