ORG_KNOWLEDGE_ID = "org-knowledge-base"
def store_with_promotion(agent_id: str, content: str, promote: bool = False):
"""Store in agent's private memory, optionally promote to org-wide."""
# Always store in agent's private space
client.store(
agent_id=agent_id,
content=content,
memory_type="semantic",
)
# Optionally promote to shared org knowledge
if promote:
client.store(
agent_id=ORG_KNOWLEDGE_ID,
content=content,
memory_type="semantic",
metadata={"promoted_by": agent_id},
)
def recall_with_org(agent_id: str, query: str, top_k: int = 10):
"""Recall from both private memory and org knowledge base."""
private = client.recall(agent_id=agent_id, query=query, top_k=top_k)
org = client.recall(agent_id=ORG_KNOWLEDGE_ID, query=query, top_k=top_k)
# Merge and deduplicate
all_results = private.results + org.results
seen = set()
unique = []
for r in sorted(all_results, key=lambda x: x.relevance_score, reverse=True):
if r.content not in seen:
seen.add(r.content)
unique.append(r)
return unique[:top_k]
# Agent discovers something important
store_with_promotion(
"researcher-agent",
"Z3rno's main competitor Mem0 just launched enterprise features.",
promote=True, # This is org-wide knowledge
)
# Any agent can access org knowledge
results = recall_with_org("writer-agent", "competitor updates")