Implementing Multi-Agent Shared Context with Agno Integration: A Complete Guide
The Agno integration in Semantica provides a thread-safe, shared knowledge graph that allows multiple Agno agents to read from and write to a unified memory store while maintaining role-scoped views.
Implementing multi-agent shared context with Agno integration enables coordinated AI teams to share institutional knowledge without conflicts. The semantica-agi/semantica repository provides a production-ready implementation that combines vector storage with graph-based analytics. This architecture leverages AgnoSharedContext as the central coordinator, ensuring every agent operates on consistent, real-time data.
Architecture of the Agno Integration
The integration centers on four core components that transform Semantica’s hybrid memory into Agno-compatible storage.
AgnoSharedContext Coordinator
The AgnoSharedContext class acts as the team-wide coordinator for shared memory operations. Located in integrations/agno/shared_context.py, this component holds a single AgentContext instance that manages both the vector store and knowledge graph. It implements a re-entrant lock (threading.RLock) to guarantee thread-safety when multiple agents bind simultaneously or record decisions concurrently.
The coordinator exposes the bind_agent(role) method, which returns a role-scoped store instance. According to the source code (lines [117‑122]), the bind_agent method wraps all operations with self._lock, ensuring that agents binding the same role receive identical _AgentScopedStore instances—guaranteeing idempotence validated in TestBindAgent.test_bind_idempotent.
Role-Scoped Storage with _AgentScopedStore
The _AgentScopedStore class provides per-role views of the shared context while maintaining data consistency. Every instance delegates operations to the parent AgentContext but tags memories and decisions with the agent’s specific role. This architecture stores data in two locations: a local cache (self._memories) for fast access and the shared pool (self._shared._shared_memories) defined at lines [84‑86] of shared_context.py.
When any agent calls upsert_memory, the data immediately becomes visible to all other agents through this shared pool, as confirmed by TestSharedMemoryPool in the test suite.
AgnoContextStore and AgentContext
The AgnoContextStore implements Agno’s MemoryDb protocol using Semantica’s underlying AgentContext. Located in integrations/agno/context_store.py, this class handles upsert_memory, read_memories, and delete_memory operations while optionally recording lightweight decisions. When decision_tracking=True, each memory insertion triggers self._context.record_decision (lines [88‑96]), creating an auditable trail of agent actions.
The AgentContext (Semantica’s core component) provides the hybrid vector and graph capabilities, exposing methods like store, retrieve, record_decision, and find_precedents_advanced for analytics-driven retrieval.
Setting Up the Shared Context
Initialize the shared context by providing vector store and knowledge graph configurations. The following example creates a FAISS-backed store with advanced analytics enabled:
from semantica.context import ContextGraph
from semantica.vector_store import VectorStore
from integrations.agno import AgnoSharedContext
# Create coordinator with automatic decision tracking
shared = AgnoSharedContext(
vector_store=VectorStore(backend="faiss"),
knowledge_graph=ContextGraph(advanced_analytics=True),
decision_tracking=True, # Auto-record decisions on memory ops
graph_expansion=True, # Enrich reads with graph neighbors
)
# Bind role-scoped stores for different agent types
researcher_store = shared.bind_agent("researcher")
analyst_store = shared.bind_agent("analyst")
The decision_tracking parameter globally enables audit trails, while graph_expansion activates contextual enrichment during retrieval operations.
Binding Agents and Configuring Memory
Connect the scoped stores to Agno agents using the standard memory parameter and specialized toolkits:
from agno.agent import Agent
from agno.team import Team
from integrations.agno import AgnoKGToolkit, AgnoDecisionKit
researcher = Agent(
name="Researcher",
memory=researcher_store, # Role-scoped memory DB
tools=[AgnoKGToolkit(context=shared)], # Shared graph access
)
analyst = Agent(
name="Analyst",
memory=analyst_store,
tools=[AgnoDecisionKit(context=shared)], # Decision-tracking toolkit
)
team = Team(agents=[researcher, analyst], mode="coordinate")
The AgnoKGToolkit provides agents with direct access to the shared knowledge graph, while AgnoDecisionKit enables explicit decision recording capabilities.
Shared Memory Operations
Agents interact with memory through standard Agno methods, with automatic synchronization across the team.
Storing and Retrieving Shared Memories
When one agent writes a memory, others can read it immediately through the shared pool:
# Researcher writes to shared memory
researcher.memory.upsert_memory(
{"memory": "Regulation XYZ released", "user_id": "u123"}
)
# Analyst reads the same memory instantly
memories = analyst.memory.read_memories()
print([m.memory for m in memories])
# Output: ["Regulation XYZ released"]
Recording Explicit Decisions
For strategic choices requiring audit trails, use the coordinator’s direct method:
shared.record_decision(
category="strategy",
scenario="Enter European market",
reasoning="High demand forecast",
outcome="approved",
confidence=0.92,
agent_role="analyst" # Tags as "strategy:analyst"
)
This creates a structured decision entry in the knowledge graph, separate from raw memory storage.
Retrieving Context for Prompt Engineering
Leverage the shared graph for advanced prompt grounding and analytics:
# Get analytics over the entire team's graph
prompt_context = shared.get_shared_insights()
# Find relevant precedents for current scenario
precedents = shared._context.find_precedents_advanced(
scenario="market expansion",
limit=3
)
# Prepend precedents to LLM prompts for better grounding
The find_precedents_advanced method traverses the knowledge graph to locate semantically similar historical decisions, enabling few-shot learning across agent interactions.
Thread Safety and Concurrency
The implementation guarantees thread-safety through threading.RLock in AgnoSharedContext. When multiple agents bind simultaneously or perform concurrent write operations, the lock ensures data consistency without deadlocks. The re-entrant design allows the same thread to acquire the lock multiple times, supporting complex nested operations in agent toolkits.
All shared memory writes are atomic—either the local cache and shared pool update together, or neither updates, preventing partial state visibility across the team.
Summary
- AgnoSharedContext coordinates multi-agent access through a thread-safe singleton pattern, ensuring all agents share one
AgentContextinstance. - _AgentScopedStore provides role-tagged views that write to both local caches and the global
self._shared._shared_memoriespool for immediate visibility. - Decision tracking can be enabled globally via
decision_tracking=Trueor used explicitly viarecord_decision()with custom agent roles. - Source files implementing this functionality reside in
integrations/agno/shared_context.pyandintegrations/agno/context_store.py, with comprehensive tests intests/integrations/agno/test_shared_context.py. - Thread safety is enforced by
threading.RLockwrapped around thebind_agent()method, making the system safe for concurrent multi-agent deployments.
Frequently Asked Questions
How does AgnoSharedContext ensure data consistency across multiple agents?
AgnoSharedContext uses a threading.RLock to synchronize access to shared resources. When agents bind to the context or perform write operations, the lock prevents race conditions while allowing the same thread to re-enter safely. Additionally, the _AgentScopedStore writes all data to both a local cache and the shared pool (self._shared._shared_memories), ensuring every agent reads from the same consistent state.
Can different agent roles access each other's memories?
Yes, all agents share the same underlying memory pool by design. While each role receives a scoped store that tags entries with its identifier (e.g., "researcher" or "analyst"), the read_memories() method accesses the global shared pool. This architecture enables cross-functional collaboration while maintaining attribution of which role created specific memories or decisions.
What is the performance impact of enabling graph_expansion?
Graph expansion adds minimal latency to read operations while significantly improving retrieval relevance. When enabled, the AgnoContextStore performs neighborhood expansion on the knowledge graph during read_memories() calls, fetching related entities and decisions. For production deployments with large graphs, consider caching frequent queries or limiting expansion depth via the underlying AgentContext configuration.
How do I disable decision tracking for specific memory operations?
Decision tracking can be controlled globally but not per-operation through the standard API. Set decision_tracking=False when initializing AgnoSharedContext to disable automatic decision recording entirely. For granular control, use the base AgentContext methods directly instead of the Agno-compatible store, or filter decisions post-hoc using the agent_role tags stored with each entry.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →