이 페이지는 아직 사용자의 언어로 번역되지 않았거나 번역이 기술 검토를 기다리고 있습니다. 아래에는 영어 원문이 표시됩니다. 영어 페이지 열기
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"
}
]
}