How Multi-Agent Shared Context Synchronizes Decisions with Agno in Semantica

Semantica synchronizes multi-agent decisions in Agno teams through a single ContextGraph shared by all agents, coordinated via AgnoSharedContext, AgnoContextStore, and AgnoDecisionKit components that enforce thread-safe, role-scoped access.

The semantica-agi/semantica repository provides a production-grade integration layer for Agno multi-agent teams. Instead of isolated agent memories, Semantica implements a unified shared context architecture where every agent operates on the same underlying knowledge graph while maintaining distinct role identities for filtering and auditability.

Core Components of the Shared Context System

Three tightly-coupled components handle multi-agent shared context synchronization in the Agno integration:

AgnoSharedContext: The Central Coordinator

Located at integrations/agno/shared_context.py, this class creates one AgentContext plus ContextGraph and guards access with a thread-safe lock. All agents receive a role-scoped view through bind_agent():

# From integrations/agno/shared_context.py lines 99-106

def bind_agent(self, agent_role: str) -> "AgnoContextStore":
    """Create a store scoped to a specific agent role."""
    return AgnoContextStore(
        shared_context=self,
        agent_role=agent_role,
        session_id=self._session_id
    )

The record_decision() method tags decisions with the calling agent's role (lines 24-33), enabling traceability across the team.

AgnoContextStore: Agent-Level Memory Database

This integrations/agno/context_store.py implementation conforms to Agno's MemoryDb protocol. Its upsert_memory() automatically records lightweight decisions when decision_tracking is enabled:

# From integrations/agno/context_store.py lines 40-48

def upsert_memory(self, memory: Memory) -> None:
    # ... memory extraction logic ...

    if self._decision_tracking:
        self._shared_context.record_decision(
            category=f"memory:{memory.topic}",
            reasoning=f"Stored memory: {memory.summary}"
        )

AgnoDecisionKit: Tool Bundle for Agents

The integrations/agno/decision_kit.py module registers toolbox functions like record_decision and find_precedents. When an agent invokes record_decision, the kit forwards to the shared AgentContext:

# From integrations/agno/decision_kit.py lines 73-82

def record_decision(
    self,
    category: str,
    scenario: str,
    reasoning: str = "",
    outcome: str = "",
    confidence: float = 0.0,
    entities: str = ""
) -> Dict[str, Any]:
    return self._ctx.record_decision(
        category=category,
        scenario=scenario,
        reasoning=reasoning,
        outcome=outcome,
        confidence=confidence,
        entities=entities,
        agent_role=self._agent_role
    )

The Synchronization Flow: Step by Step

Understanding how Agno agents synchronize decisions via shared context requires tracing the exact execution path:

1. Team initialization — AgnoSharedContext instantiates a single AgentContext and ContextGraph under one lock.

2. Agent binding — Each agent calls shared.bind_agent(role), returning a _AgentScopedStore. This store reuses the parent's AgentContext but prefixes conversation IDs with <session_id>::<role> (lines 59-70 in shared_context.py).

3. Decision recording — When AgnoDecisionKit.record_decision is invoked, the call reaches self._ctx.record_decision() on the shared AgentContext. The method tags decisions with category:<agent_role> if supplied (lines 42-44), then writes the node under the same lock.

4. Graph-wide visibility — Because all agents reference the same ContextGraph, decisions from one role appear instantly to others. The find_precedents() function queries the full graph with optional category filtering (lines 53-60).

This architecture ensures synchronization through a single shared graph with role-scoped wrappers that preserve origin identity for audit trails.

Practical Implementation Example

# 1️⃣ Initialise a shared context for a team

from semantica.vector_store import VectorStore
from semantica.context import ContextGraph
from integrations.agno import AgnoSharedContext, AgnoDecisionKit

shared = AgnoSharedContext(
    vector_store=VectorStore(backend="faiss"),
    knowledge_graph=ContextGraph(advanced_analytics=True),
    decision_tracking=True,
)

# 2️⃣ Bind two agents – each gets a role-scoped store

researcher_store = shared.bind_agent("researcher")
analyst_store    = shared.bind_agent("analyst")

# 3️⃣ Attach the decision-tool to the analyst agent

decision_kit = AgnoDecisionKit(context=shared)

# 4️⃣ Analyst records a decision (via the toolkit)

decision_json = decision_kit.record_decision(
    category="risk_assessment",
    scenario="User requests high-value transaction",
    reasoning="Risk score > 0.8",
    outcome="denied",
    confidence=0.92,
    entities="user_id, transaction_id",
)

# 5️⃣ Researcher queries precedents (shared view)

precedents = researcher_store._context.find_precedents_advanced(
    scenario="high-value transaction request",
    category="risk_assessment",
    limit=3,
)

The researcher sees the analyst's decision immediately because both stores reference identical underlying graph structures.

Key Source Files Reference

File	Purpose
integrations/agno/shared_context.py	Core coordinator with binding logic and thread-safe routing
integrations/agno/context_store.py	MemoryDb implementation with automatic decision logging
integrations/agno/decision_kit.py	Agno tool registration (record_decision, find_precedents)
semantica/context.py	Underlying AgentContext with vector and graph storage
semantica/context/graph.py	ContextGraph data structure for decision nodes and causal edges

Summary

• Single source of truth: One ContextGraph shared across all Agno team members prevents synchronization conflicts.
• Role-scoped isolation: bind_agent() creates filtered views while preserving unified storage.
• Automatic decision tracking: AgnoContextStore and AgnoDecisionKit transparently log decisions with origin tags.
• Thread-safe access: AgnoSharedContext serializes graph mutations under a central lock.
• Full auditability: The category:<agent_role> tagging scheme enables precise filtering and precedent retrieval.

Frequently Asked Questions

How does Semantica prevent race conditions when multiple Agno agents write decisions simultaneously?

The AgnoSharedContext class maintains a threading.Lock() that guards all graph mutations. When record_decision() is called, the method acquires this lock before modifying the shared ContextGraph, ensuring atomic updates regardless of which agent initiates the write.

Can agents query only their own decisions, or do they automatically see the entire team's history?

By default, agents have full visibility into the shared graph. However, the find_precedents methods accept a category parameter that can filter by role tag (e.g., category="risk_assessment:analyst"). Applications can implement additional access controls by wrapping these queries in custom logic.

What happens if an Agno agent joins a team mid-conversation?

New agents call bind_agent() to receive their scoped store, which immediately connects to the existing AgentContext and ContextGraph. All prior decisions are visible to the new agent without requiring state synchronization or replay mechanisms.

Does enabling decision_tracking impact performance for high-throughput agent teams?

The AgnoContextStore performs lightweight decision logging via record_decision() with minimal overhead. For intensive workloads, the underlying ContextGraph supports batched operations and configurable persistence backends (vector stores like FAISS) to maintain throughput.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}