semantica/cookbook/integrations/agno_multi_agent_shared_context.ipynb

Agno × Semantica: Multi-Agent Shared Context

This notebook shows how an Agno Team of specialist agents can share a single ContextGraph so they:

• Never make contradictory decisions
• Reuse each other's extracted knowledge without coupling implementations
• Maintain a full causal audit trail across all agents

Scenario: A product strategy team with three specialist agents:

Agent	Role	Tools
Researcher	Extracts competitive intelligence from text	AgnoKGToolkit
Analyst	Evaluates opportunities and records decisions	AgnoDecisionKit
Strategist	Synthesises both into a recommendation	both

---

Architecture

AgnoSharedContext (single ContextGraph + VectorStore)
   │
   ├── bind_agent("researcher")  → AgnoContextStore (role-scoped)
   ├── bind_agent("analyst")     → AgnoContextStore (role-scoped)
   └── bind_agent("strategist")  → AgnoContextStore (role-scoped)

Agno Team
   ├── Researcher  memory=researcher_store  tools=[AgnoKGToolkit(context=shared)]
   ├── Analyst     memory=analyst_store     tools=[AgnoDecisionKit(context=shared)]
   └── Strategist  memory=strategist_store  tools=[AgnoKGToolkit, AgnoDecisionKit]

Install

pip install semantica[agno]

1. Imports

In [ ]:

import sys, os, json
sys.path.insert(0, os.path.abspath("../../"))

# ── Semantica core ───────────────────────────────────────────────────────────
from semantica.context import ContextGraph, AgentContext, CausalChainAnalyzer
from semantica.vector_store import VectorStore
from semantica.semantic_extract import NERExtractor, RelationExtractor
from semantica.reasoning import Reasoner
from semantica.kg import GraphBuilder, GraphAnalyzer, CentralityCalculator

# ── Agno integration ─────────────────────────────────────────────────────────
from integrations.agno import (
    AgnoSharedContext,
    AgnoDecisionKit,
    AgnoKGToolkit,
    AGNO_AVAILABLE,
)

print("Semantica imports OK")
print(f"Agno installed: {AGNO_AVAILABLE}")

2. Build the Shared Semantica Backend

A single VectorStore and ContextGraph underpin the entire team. All agents read and write to the same store — role scoping is applied automatically by AgnoSharedContext.

In [ ]:

# ── Single shared backends ───────────────────────────────────────────────────
shared_vector_store = VectorStore(backend="faiss", dimension=768)
shared_graph = ContextGraph(advanced_analytics=True)

print("Shared VectorStore (FAISS) ready")
print("Shared ContextGraph ready")

# ── AgnoSharedContext: the team coordinator ───────────────────────────────────
shared = AgnoSharedContext(
    vector_store=shared_vector_store,
    knowledge_graph=shared_graph,
    decision_tracking=True,
    session_id="product_strategy_team_q1_2026",
)
print(f"\nAgnoSharedContext ready — session: {shared.session_id}")

3. Bind Agent Roles

Each agent gets a role-scoped AgnoContextStore via bind_agent(). All agents share the same underlying graph, but their writes are tagged with their role for filtering.

In [ ]:

# Bind each agent role — idempotent, can be called multiple times safely
researcher_store  = shared.bind_agent("researcher")
analyst_store     = shared.bind_agent("analyst")
strategist_store  = shared.bind_agent("strategist")

print("Agent roles bound:")
for role in shared.bound_roles:
    store = shared.bind_agent(role)
    print(f"  {role:15s} → session={store.session_id}")

# Verify all roles see the same underlying knowledge_graph
assert researcher_store._ctx is analyst_store._ctx
print("\nAll agents share the same AgentContext ✓")

4. Pre-Load Competitive Intelligence

Using native Semantica APIs, we load a competitive landscape into the shared graph. This represents knowledge the team has accumulated from prior research sessions.

In [ ]:

# Competitive intelligence documents
COMPETITIVE_INTEL = [
    {
        "source": "market_research_q4_2025",
        "text": (
            "Competitor Alpha launched a new SaaS analytics platform in Q4 2025. "
            "The product targets mid-market enterprises with annual revenue between "
            "$50M–$500M and has attracted 200 paying customers within 3 months. "
            "Pricing is $2,000/seat/year with volume discounts at 50+ seats. "
            "Alpha raised a $80M Series C led by Sequoia Capital in November 2025."
        ),
    },
    {
        "source": "customer_interviews_q4_2025",
        "text": (
            "Customer interviews reveal strong demand for AI-powered anomaly detection "
            "in financial reporting workflows. 78% of CFOs surveyed cite 'time to insight' "
            "as the top pain point — currently averaging 14 days per reporting cycle. "
            "Competitor Alpha scores poorly on integration depth (NPS: 24) while "
            "our legacy product scores 41. Customers value our data governance features "
            "but want a modern UI and sub-second query times."
        ),
    },
    {
        "source": "technology_scan_q4_2025",
        "text": (
            "Emerging technologies for consideration: LLM-native analytics interfaces "
            "reduce time-to-insight by 60% in pilot studies (Stanford HAI, 2025). "
            "Graph-based anomaly detection outperforms time-series approaches for "
            "multi-entity financial fraud by 34% (ACM SIGMOD 2025). "
            "Vector database adoption in enterprise analytics grew 120% YoY. "
            "Apache Arrow and DuckDB emerging as standards for in-process OLAP."
        ),
    },
]

# Use Semantica NER + RelationExtractor directly for rich extraction
ner = NERExtractor()
rel_extractor = RelationExtractor(confidence_threshold=0.55)
graph_builder = GraphBuilder(merge_entities=True)

for doc in COMPETITIVE_INTEL:
    text = doc['text']
    entities = ner.extract_entities(text) or []
    relations = rel_extractor.extract_relations(text) or []
    print(f"[{doc['source']}]")
    print(f"  Entities: {len(entities)}, Relations: {len(relations)}")
    # Store into shared context for all agents to access
    shared._context.store(text, conversation_id=doc['source'])

print("\nCompetitive intelligence loaded into shared context")

5. Build Agent-Specific Tools

Each toolkit is pointed at the shared context so tool calls across agents modify and read the same graph.

In [ ]:

# Researcher's KG toolkit — builds knowledge from raw text
researcher_kg_kit = AgnoKGToolkit(
    ner_extractor=ner,
    relation_extractor=rel_extractor,
    reasoner=Reasoner(),
    context=shared.knowledge_graph,   # shared graph
)

# Analyst's decision kit — records evaluations and finds precedents
analyst_decision_kit = AgnoDecisionKit(
    context=shared._context,           # shared AgentContext
    max_precedents=5,
    causal_depth=3,
    enable_policy_check=True,
)

# Strategist gets both
strategist_kg_kit = AgnoKGToolkit(
    ner_extractor=ner,
    relation_extractor=rel_extractor,
    reasoner=Reasoner(),
    context=shared.knowledge_graph,
)
strategist_decision_kit = AgnoDecisionKit(
    context=shared._context,
    max_precedents=5,
)

print(f"Researcher toolkit:  {len(researcher_kg_kit._tools)} tools")
print(f"Analyst toolkit:     {len(analyst_decision_kit._tools)} tools")
print(f"Strategist toolkits: {len(strategist_kg_kit._tools)} + {len(strategist_decision_kit._tools)} tools")

6. Simulate Agent Collaboration

We simulate the agents' reasoning steps directly, showing how shared context propagates knowledge between roles.

In [ ]:

print("=" * 65)
print("RESEARCHER AGENT TURN")
print("=" * 65)

# Researcher extracts entities from new competitive intel
new_intel = (
    "Competitor Beta just closed a strategic partnership with Microsoft Azure, "
    "integrating their anomaly detection engine natively into Azure Synapse Analytics. "
    "This gives Beta access to Microsoft's 300,000+ enterprise customer base. "
    "Beta's CEO Sarah Chen announced the deal at Gartner Data & Analytics Summit."
)

# Step 1: Extract entities
entities_result = json.loads(researcher_kg_kit.extract_entities(new_intel))
print(f"\n[researcher] extracted {entities_result['count']} entities:")
for e in entities_result['entities']:
    print(f"  {e['name']:30s}  type={e['type']}")

# Step 2: Extract relations
relations_result = json.loads(researcher_kg_kit.extract_relations(new_intel))
print(f"\n[researcher] extracted {relations_result['count']} relations")

# Step 3: Add to shared graph — now visible to ALL agents
add_result = json.loads(researcher_kg_kit.add_to_graph(
    entities=json.dumps([
        {"name": "Competitor Beta", "type": "COMPANY"},
        {"name": "Microsoft Azure", "type": "COMPANY"},
        {"name": "Azure Synapse Analytics", "type": "PRODUCT"},
        {"name": "Sarah Chen", "type": "PERSON"},
        {"name": "Gartner Data & Analytics Summit", "type": "EVENT"},
    ]),
    relations=json.dumps([
        {"source": "Competitor Beta", "relation": "PARTNERSHIP_WITH", "target": "Microsoft Azure"},
        {"source": "Competitor Beta", "relation": "INTEGRATES_WITH", "target": "Azure Synapse Analytics"},
        {"source": "Sarah Chen", "relation": "CEO_OF", "target": "Competitor Beta"},
    ]),
))
print(f"\n[researcher] added {add_result['nodes_added']} nodes, {add_result['edges_added']} edges to SHARED graph")

In [ ]:

print("=" * 65)
print("ANALYST AGENT TURN (sees researcher's graph additions)")
print("=" * 65)

# Analyst queries the graph the researcher just populated
competitor_query = json.loads(analyst_decision_kit.find_precedents(
    scenario="competitor partnership with cloud hyperscaler threatens market position",
    limit=3,
))
print(f"\n[analyst] find_precedents → {competitor_query['count']} similar past strategic responses found")

# Analyst records a strategic evaluation decision
eval_json = analyst_decision_kit.record_decision(
    category="strategic_response",
    scenario=(
        "Competitor Beta + Microsoft Azure partnership gives Beta access to "
        "300k enterprise customers via Azure Synapse native integration"
    ),
    reasoning=(
        "Threat level: HIGH. Beta's Azure native integration removes our "
        "integration advantage. Existing NPS lead (41 vs 24) remains but "
        "distribution disadvantage is critical. Recommend accelerated cloud-native "
        "partnership evaluation, specifically AWS Marketplace + Snowflake Native App."
    ),
    outcome="escalate_to_strategy",
    confidence=0.85,
    entities="Competitor Beta, Microsoft Azure, AWS Marketplace, Snowflake",
)
eval_result = json.loads(eval_json)
analyst_decision_id = eval_result['decision_id']
print(f"\n[analyst] recorded evaluation → decision_id: {analyst_decision_id}")

In [ ]:

print("=" * 65)
print("STRATEGIST AGENT TURN (sees both researcher + analyst work)")
print("=" * 65)

# Strategist queries the graph for the full competitive picture
related = json.loads(strategist_kg_kit.find_related("Competitor Beta", hops=2))
print(f"\n[strategist] 'Competitor Beta' 2-hop neighbourhood: {related['count']} entity/entities")
for entity in related['related']:
    print(f"  → {entity}")

# Strategist traces what the analyst decided
causal = json.loads(strategist_decision_kit.trace_causal_chain(analyst_decision_id, depth=3))
print(f"\n[strategist] causal chain for analyst decision: {causal}")

# Strategist records the final strategic recommendation
strategy_json = strategist_decision_kit.record_decision(
    category="product_strategy",
    scenario="Q1 2026 product strategy: respond to Beta+Azure threat",
    reasoning=(
        "Based on researcher's KG (Beta+Azure integration, 300k customer reach) "
        "and analyst's evaluation (threat level HIGH, escalated decision). "
        "Strategy: (1) Accelerate AWS Marketplace listing by Q2 2026. "
        "(2) Launch Snowflake Native App by Q3 2026. "
        "(3) Invest $2M in UI modernisation to widen NPS lead. "
        "(4) Fast-track LLM-native analytics interface (60% time-to-insight improvement per HAI study). "
        "Existing NPS advantage (41 vs 24) provides 18-month window before Beta catches up."
    ),
    outcome="approved",
    confidence=0.88,
    entities="AWS Marketplace, Snowflake, LLM Analytics, Q2 2026, Q3 2026",
)
strategy_result = json.loads(strategy_json)
print(f"\n[strategist] final recommendation recorded → {strategy_result['decision_id']}")

7. Verify Shared Memory Pool

Memories written by one agent are readable by all others.

In [ ]:

from integrations.agno.context_store import _MemoryRow as MemoryRow

# Researcher writes a memory
researcher_row = MemoryRow(
    memory="Beta + Azure partnership announced at Gartner Summit — threat level HIGH",
    user_id="researcher",
)
researcher_store.upsert_memory(researcher_row)

# Analyst writes a memory
analyst_row = MemoryRow(
    memory="NPS advantage (41 vs 24) gives 18-month window — accelerate cloud partnerships",
    user_id="analyst",
)
analyst_store.upsert_memory(analyst_row)

# Strategist reads ALL memories from both agents
strategist_memories = strategist_store.read_memories()

print(f"Strategist sees {len(strategist_memories)} shared memory item(s):")
for m in strategist_memories:
    uid = getattr(m, 'user_id', '?')
    text = getattr(m, 'memory', str(m))
    print(f"  [{uid:12s}] {text[:80]}")

8. Wire into Agno Team (requires API key)

In [ ]:

if AGNO_AVAILABLE:
    from agno.agent import Agent
    from agno.team import Team
    from agno.memory import AgentMemory
    from agno.models.openai import OpenAIChat

    researcher_agent = Agent(
        name="Researcher",
        model=OpenAIChat(id="gpt-4o"),
        memory=AgentMemory(db=researcher_store),
        tools=[researcher_kg_kit],
        show_tool_calls=True,
        description=(
            "You are a competitive intelligence researcher. "
            "Use extract_entities, extract_relations, and add_to_graph "
            "to build a structured knowledge graph from market intelligence. "
            "Always add discoveries to the shared graph."
        ),
    )

    analyst_agent = Agent(
        name="Analyst",
        model=OpenAIChat(id="gpt-4o"),
        memory=AgentMemory(db=analyst_store),
        tools=[analyst_decision_kit],
        show_tool_calls=True,
        description=(
            "You are a strategic analyst. Use find_precedents to check historical "
            "responses to similar threats, then record_decision with your evaluation. "
            "Always check if a similar situation was handled before acting."
        ),
    )

    strategist_agent = Agent(
        name="Strategist",
        model=OpenAIChat(id="gpt-4o"),
        memory=AgentMemory(db=strategist_store),
        tools=[strategist_kg_kit, strategist_decision_kit],
        show_tool_calls=True,
        description=(
            "You are the Chief Strategy Officer. Synthesise the researcher's knowledge "
            "graph and the analyst's decision record into a concrete product strategy. "
            "Use find_related to explore the competitive graph, then record_decision "
            "with the final approved strategy."
        ),
    )

    strategy_team = Team(
        name="Product Strategy Team",
        agents=[researcher_agent, analyst_agent, strategist_agent],
        mode="coordinate",
    )

    strategy_team.print_response(
        "Competitor Beta just announced a native Azure integration. "
        "Analyse the competitive landscape and recommend our Q1 2026 product strategy."
    )
else:
    print("[Agno not installed — skipping live team run]")
    print()
    print("Expected team coordination flow:")
    print("  1. Researcher: extract_entities + add_to_graph (Beta+Azure)")
    print("  2. Analyst: find_precedents + record_decision (threat=HIGH, escalate)")
    print("  3. Strategist: find_related + trace_causal_chain + record_decision (final strategy)")

9. Post-Session Analysis with Semantica

After the team session, use native Semantica APIs for cross-agent audit, analytics, and causal chain review.

In [ ]:

# Team-level insights from AgnoSharedContext
insights = shared.get_shared_insights()
print("Team session insights:")
if isinstance(insights, dict):
    for k, v in insights.items():
        print(f"  {k}: {v}")
else:
    print(f"  {insights}")

print(f"\nBound agent roles: {shared.bound_roles}")

In [ ]:

# Find all cross-agent strategic decisions
all_strategic = shared.find_precedents(
    scenario="cloud partnership competitive response",
    category="strategic_response",
)
print(f"Cross-agent strategic precedents: {len(all_strategic or [])}")

In [ ]:

# Graph analytics on the shared knowledge graph (Semantica native)
try:
    analyzer = GraphAnalyzer()
    analysis = analyzer.analyze_graph(shared.knowledge_graph)
    print("Shared knowledge graph analysis:")
    if isinstance(analysis, dict):
        for k, v in list(analysis.items())[:6]:
            print(f"  {k}: {v}")
    else:
        print(f"  {analysis}")
except Exception as e:
    print(f"GraphAnalyzer: {e}")

In [ ]:

# Which entities are most central in the competitive intelligence graph?
try:
    centrality = CentralityCalculator()
    scores = centrality.calculate_degree_centrality(shared.knowledge_graph)
    print("Most central entities in shared graph:")
    if isinstance(scores, dict):
        top = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:5]
        for entity, score in top:
            print(f"  {entity:35s} centrality={score:.4f}")
    else:
        print(f"  {scores}")
except Exception as e:
    print(f"CentralityCalculator: {e}")

In [ ]:

# Direct Semantica causal chain analysis (no Agno needed)
try:
    causal_analyzer = CausalChainAnalyzer(graph_store=shared.knowledge_graph)
    # Query all decisions made during this session
    decisions = shared.knowledge_graph.find_precedents(category="product_strategy", limit=10)
    print(f"Product strategy decisions in shared graph: {len(decisions or [])}")
    for d in (decisions or [])[:3]:
        scenario = d.get('scenario', '') if isinstance(d, dict) else str(d)
        outcome = d.get('outcome', '') if isinstance(d, dict) else ''
        print(f"  [{outcome:20s}] {scenario[:70]}")
except Exception as e:
    print(f"CausalChainAnalyzer: {e}")

Summary

Pattern	Implementation
Single shared knowledge graph	AgnoSharedContext(vector_store, knowledge_graph)
Role-scoped memory	shared.bind_agent("researcher") → _AgentScopedStore
Cross-agent memory visibility	All stores read from shared._shared_memories
KG tool sharing	AgnoKGToolkit(context=shared.knowledge_graph)
Decision tool sharing	AgnoDecisionKit(context=shared._context)
Thread-safe binding	AgnoSharedContext._lock (RLock)
Post-session analytics	GraphAnalyzer, CentralityCalculator, CausalChainAnalyzer — all Semantica native

Key design rule: Every agent writes to the same underlying graph via different role-scoped stores. The Agno integration is a thin routing layer — Semantica's full power is available at any point directly.