weaviate_db.py
"""
Per-User Isolation: Weaviate
============================
Each user gets a private view of one shared knowledge base. Documents inserted
with a user_id are visible only to that user, documents inserted without one
are shared with everyone.
Weaviate keeps the owner in a user_id text property, so a scoped search matches
the caller's id or an unset property:
- Alice sees her own chunks plus shared content, never Bob's
- Bob sees his own chunks plus shared content, never Alice's
- user_id=None is the admin view and sees everything
Requirements:
- ./cookbook/scripts/run_weaviate.sh
- uv pip install weaviate-client
- OPENAI_API_KEY
"""
import asyncio
from typing import List
from agno.agent import Agent
from agno.knowledge.document import Document
from agno.knowledge.knowledge import Knowledge
from agno.models.openai import OpenAIResponses
from agno.vectordb.weaviate import Distance, VectorIndex, Weaviate
# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
ALICE_SALARY = "Alice's salary is $180,000. Reviewed annually in March."
BOB_SALARY = "Bob's salary is $215,000. Reviewed annually in June."
HOLIDAYS = "The company is closed on January 1, July 4, and December 25."
COLLECTION_NAME = "per_user_isolation_demo"
def show(label: str, results: List[Document]) -> None:
"""Print one search result set."""
print(f"{label} -> {len(results)} results")
for d in results:
print(f" - {d.content[:80]}")
print()
# ---------------------------------------------------------------------------
# Create Knowledge Base
# ---------------------------------------------------------------------------
vector_db = Weaviate(
collection=COLLECTION_NAME,
vector_index=VectorIndex.HNSW,
distance=Distance.COSINE,
local=True,
)
# Start clean: the owner property is created with the collection and cannot be added in place.
if vector_db.exists():
vector_db.drop()
vector_db.create()
knowledge = Knowledge(
name="per_user_demo",
description="Per-user RAG isolation demo (Weaviate)",
vector_db=vector_db,
)
# ---------------------------------------------------------------------------
# Run Demo
# ---------------------------------------------------------------------------
if __name__ == "__main__":
async def main() -> None:
await knowledge.ainsert(
name="alice_salary",
text_content=ALICE_SALARY,
user_id="alice",
)
await knowledge.ainsert(
name="bob_salary",
text_content=BOB_SALARY,
user_id="bob",
)
# The last insert has no user_id, which makes it shared with everyone.
await knowledge.ainsert(
name="company_holidays",
text_content=HOLIDAYS,
)
print("\n" + "=" * 60)
print("SCOPED SEARCH: three callers, one corpus")
print("=" * 60 + "\n")
alice_view = await knowledge.asearch(query="salary", user_id="alice")
show("Alice (user_id='alice')", alice_view)
alice_text = " ".join(d.content for d in alice_view)
assert "180,000" in alice_text, "Alice cannot retrieve her own document"
assert "January 1" in alice_text, (
"Shared content is unreachable from Alice's scoped view"
)
assert "215,000" not in alice_text, (
"Isolation broken: Alice's scoped view leaked Bob's salary"
)
bob_view = await knowledge.asearch(query="salary", user_id="bob")
show("Bob (user_id='bob')", bob_view)
bob_text = " ".join(d.content for d in bob_view)
assert "215,000" in bob_text, "Bob cannot retrieve his own document"
assert "January 1" in bob_text, (
"Shared content is unreachable from Bob's scoped view"
)
assert "180,000" not in bob_text, (
"Isolation broken: Bob's scoped view leaked Alice's salary"
)
admin_view = await knowledge.asearch(query="salary", user_id=None)
show("Admin (user_id=None)", admin_view)
admin_text = " ".join(d.content for d in admin_view)
for expected in ("180,000", "215,000", "January 1"):
assert expected in admin_text, (
f"Admin view is missing {expected}, it has to see every owner"
)
assert all(d.content in admin_text for d in alice_view), (
"Admin view has to be a superset of a scoped user's view"
)
print("Alice and Bob each see their own chunk plus the shared one.")
print("Admin sees the whole corpus.")
print("\n" + "=" * 60)
print("AGENT-MEDIATED RETRIEVAL: the owner has to survive the handoff")
print("=" * 60 + "\n")
alice_agent = Agent(
name="Alice's Assistant",
model=OpenAIResponses(id="gpt-5.5"),
knowledge=knowledge,
search_knowledge=True,
user_id="alice",
instructions=[
"Answer questions using ONLY the knowledge you can retrieve.",
"If you don't know, say so - do not invent salary figures.",
],
markdown=True,
)
response = await alice_agent.arun("What is Bob's salary?")
print("Alice's agent on 'What is Bob's salary?':")
print(response.content)
# Assert on what retrieval returned, not on the model's prose.
retrieved = " ".join(
item["content"]
for ref in (response.references or [])
for item in (ref.references or [])
if isinstance(item, dict) and item.get("content")
)
# Empty references would make the isolation check below pass on nothing
assert retrieved, (
"Retrieval returned no documents, so the isolation check below would pass on nothing"
)
assert "215,000" not in retrieved, (
"Isolation broken: Alice's agent retrieved Bob's salary. The owner was "
"dropped between the run context and the vector DB, so retrieval ran "
"unscoped (user_id=None, the admin view)."
)
print("\nisolation holds: Bob's salary never reached Alice's agent")
print("\nDone.")
asyncio.run(main())
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U agno openai weaviate weaviate-client
3
Export your API keys
export OPENAI_API_KEY="your_openai_api_key_here"
export WCD_API_KEY="your_wcd_api_key_here"
$Env:OPENAI_API_KEY="your_openai_api_key_here"
$Env:WCD_API_KEY="your_wcd_api_key_here"
4
Run the example
Save the code above as
weaviate_db.py, then run:python weaviate_db.py