このページはまだお使いの言語に翻訳されていないか、翻訳が技術レビュー待ちです。以下には英語の原文を表示しています。 英語のページを開く
Semantic Search with Policy Guardrails
This guide shows how to add Control Zero policy enforcement to a semantic search system, controlling who can search what data and which embedding models are used.
What You Will Build
A semantic search service that:
- Generates embeddings for queries
- Searches a vector database
- Enforces access control on data collections
- Restricts embedding model usage
Implementation
Setup
pip install controlzero openai chromadb
from controlzero import Client
import openai
import chromadb
cz = Client(api_key="cz_live_your_api_key_here")
openai_client = openai.OpenAI()
chroma = chromadb.Client()
Search with Policy Enforcement
def search(
query: str,
collection_name: str,
agent_id: str,
n_results: int = 10,
) -> list[dict]:
"""Search a collection with policy enforcement."""
# Enforce: can this agent access this collection?
cz.guard(f"vectorstore/{collection_name}", method="read", args={"agent_id": agent_id})
# Enforce: can this agent use embeddings?
cz.guard("embedding", method="generate", args={"model": "text-embedding-3-small", "agent_id": agent_id})
# Generate query embedding
response = openai_client.embeddings.create(
model="text-embedding-3-small",
input=query,
)
query_embedding = response.data[0].embedding
# Search the collection
collection = chroma.get_collection(collection_name)
results = collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
)
return [
{"document": doc, "metadata": meta, "distance": dist}
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0],
)
]
Multi-Collection Search
def search_across_collections(
query: str,
collections: list[str],
agent_id: str,
) -> dict[str, list[dict]]:
"""Search multiple collections, skipping those the agent cannot access."""
results = {}
for collection_name in collections:
decision = cz.guard(f"vectorstore/{collection_name}", method="read", args={"agent_id": agent_id})
if decision.effect == "allow":
results[collection_name] = search(
query, collection_name, agent_id
)
else:
# Log that access was denied, but continue with other collections
results[collection_name] = []
return results
Example Policy
{
"name": "semantic-search-policy",
"rules": [
{
"effect": "allow",
"action": "data:read",
"resource": "vectorstore/public-docs"
},
{
"effect": "allow",
"action": "data:read",
"resource": "vectorstore/product-catalog"
},
{
"effect": "deny",
"action": "data:read",
"resource": "vectorstore/financial-reports"
},
{
"effect": "allow",
"action": "embedding:generate",
"resource": "model/text-embedding-3-small"
}
]
}